diff --git a/crates/graphforge-api/src/bulk_construction.rs b/crates/graphforge-api/src/bulk_construction.rs index ebb4d9924..74f380ad9 100644 --- a/crates/graphforge-api/src/bulk_construction.rs +++ b/crates/graphforge-api/src/bulk_construction.rs @@ -18,10 +18,10 @@ use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::sync::Arc; use arrow::array::{ - Array, ArrayRef, BooleanArray, FixedSizeBinaryArray, Float32Array, Float64Array, Int8Array, - Int16Array, Int32Array, Int64Array, LargeListArray, LargeStringArray, ListArray, StringArray, - StructArray, Time64NanosecondArray, TimestampMicrosecondArray, UInt8Array, UInt16Array, - UInt32Array, UInt64Array, + Array, ArrayRef, BooleanArray, FixedSizeBinaryArray, FixedSizeBinaryBuilder, Float32Array, + Float64Array, Int8Array, Int16Array, Int32Array, Int64Array, LargeListArray, LargeStringArray, + ListArray, StringArray, StructArray, Time64NanosecondArray, TimestampMicrosecondArray, + UInt8Array, UInt16Array, UInt32Array, UInt64Array, }; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use arrow::record_batch::RecordBatch; @@ -302,6 +302,48 @@ pub fn bulk_edge_input_schema(properties: Vec) -> Result, +) -> Result { + let required = match kind { + BulkInputKind::Node => &NODE_REQUIRED[..], + BulkInputKind::Edge => &EDGE_REQUIRED[..], + }; + let property_fields = source.schema().fields()[required.len()..] + .iter() + .map(|field| field.as_ref().clone()) + .collect::>(); + let mut fields = match kind { + BulkInputKind::Node => graphforge_storage::CONSTRUCTION_NODE_SCHEMA + .fields() + .to_vec(), + BulkInputKind::Edge => graphforge_storage::CONSTRUCTION_EDGE_SCHEMA + .fields() + .to_vec(), + }; + fields.extend(property_fields.into_iter().map(Arc::new)); + let schema = Arc::new(Schema::new(fields)); + required_columns.extend(source.columns()[required.len()..].iter().cloned()); + RecordBatch::try_new(schema, required_columns).map_err(|error| { + contract_error(kind, BulkValidationReason::ProjectState, &error.to_string()) + }) +} + +fn canonical_import_uuid_array( + kind: BulkInputKind, + values: impl ExactSizeIterator, +) -> Result { + let mut builder = FixedSizeBinaryBuilder::with_capacity(values.len(), 16); + for value in values { + builder.append_value(value.as_bytes()).map_err(|error| { + contract_error(kind, BulkValidationReason::ProjectState, &error.to_string()) + })?; + } + Ok(Arc::new(builder.finish())) +} + /// Canonical receipt schema used by the later publication slices. /// /// Receipts retain input order and identify the created object, its node label @@ -331,28 +373,6 @@ pub fn bulk_receipt_schema() -> SchemaRef { } impl GraphForge { - pub(crate) fn import_base_membership( - &self, - candidates: &[Uuid], - kind: graphforge_storage::UuidIndexKind, - input_kind: BulkInputKind, - ) -> Result, BulkValidationError> { - let mut index = open_membership_index(self, input_kind)?; - let Some(index) = index.as_mut() else { - return Ok(vec![false; candidates.len()]); - }; - index - .probe(kind, candidates) - .map(|(found, _)| found) - .map_err(|error| { - contract_error( - input_kind, - BulkValidationReason::ProjectState, - &error.to_string(), - ) - }) - } - pub(crate) fn normalize_import_nodes( &self, operation_uuid: OperationId, @@ -361,26 +381,23 @@ impl GraphForge { self.normalize_bulk_nodes(operation_uuid, batches, false) } - pub(crate) fn normalize_import_edges( + pub(crate) fn normalize_import_node_chunk( &self, operation_uuid: OperationId, - batches: &[RecordBatch], - imported_endpoints: &BTreeSet, - ) -> Result { - let empty_nodes = ValidatedBulkNodes { - rows: Vec::new(), - operation_uuid, - source_generation_uuid: *self - .current_generation_uuid - .lock() - .expect("generation UUID lock poisoned"), - }; - self.normalize_bulk_edges( - operation_uuid, - batches, - &empty_nodes, - false, - Some(imported_endpoints), + batch: &RecordBatch, + ) -> Result { + let normalized = + self.normalize_import_nodes(operation_uuid, std::slice::from_ref(batch))?; + let identities = canonical_import_uuid_array( + BulkInputKind::Node, + normalized.rows().iter().map(|row| row.node_uuid), + )?; + let labels = + StringArray::from_iter_values(normalized.rows().iter().map(|row| row.label.as_str())); + canonical_import_chunk( + BulkInputKind::Node, + batch, + vec![identities, Arc::new(labels)], ) } @@ -786,6 +803,62 @@ impl GraphForge { }) } + pub(crate) fn normalize_import_edge_chunk( + &self, + operation_uuid: OperationId, + batch: &RecordBatch, + ) -> Result { + // Construction sealing owns the global endpoint proof across all node + // chunks. This prevalidation still checks the complete edge schema, + // identities, properties, and within-chunk duplicates without retaining + // every imported node UUID in memory. + let assumed_endpoints = ValidatedBulkNodes { + rows: candidate_endpoint_uuids(std::slice::from_ref(batch))? + .into_iter() + .enumerate() + .map(|(ordinal, node_uuid)| BulkNodeRow { + row_ordinal: ordinal as u64, + node_uuid, + label: String::new(), + properties: BTreeMap::new(), + }) + .collect(), + operation_uuid, + source_generation_uuid: *self + .current_generation_uuid + .lock() + .expect("generation UUID lock poisoned"), + }; + let normalized = self.normalize_bulk_edges( + operation_uuid, + std::slice::from_ref(batch), + &assumed_endpoints, + true, + None, + )?; + canonical_import_chunk( + BulkInputKind::Edge, + batch, + vec![ + canonical_import_uuid_array( + BulkInputKind::Edge, + normalized.rows().iter().map(|row| row.edge_uuid), + )?, + Arc::new(StringArray::from_iter_values( + normalized.rows().iter().map(|row| row.rel_type.as_str()), + )), + canonical_import_uuid_array( + BulkInputKind::Edge, + normalized.rows().iter().map(|row| row.source_uuid), + )?, + canonical_import_uuid_array( + BulkInputKind::Edge, + normalized.rows().iter().map(|row| row.target_uuid), + )?, + ], + ) + } + /// Validate and atomically publish a logical Arrow edge batch. /// /// Exact retries return the original ordered receipt without publishing a diff --git a/crates/graphforge-api/src/import_session.rs b/crates/graphforge-api/src/import_session.rs index 3a80b8a70..b3c9a4c90 100644 --- a/crates/graphforge-api/src/import_session.rs +++ b/crates/graphforge-api/src/import_session.rs @@ -1,12 +1,10 @@ //! Durable, bounded staged graph-import sessions (#738). -use std::collections::{BTreeSet, HashMap}; use std::fs::{self, File}; -use std::io::{BufRead, BufReader, BufWriter, Read, Seek, SeekFrom, Write}; +use std::io::{BufReader, BufWriter, Write}; use std::path::{Component, Path, PathBuf}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; -use arrow::array::{Array, FixedSizeBinaryArray}; use arrow::ipc::reader::FileReader as ArrowFileReader; use arrow::ipc::writer::FileWriter as ArrowFileWriter; use arrow::record_batch::RecordBatch; @@ -16,7 +14,7 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use uuid::Uuid; -use crate::{BulkInputKind, CancellationToken, GraphForge, OperationId}; +use crate::{BulkInputKind, CancellationToken, GraphConstructionBudgets, GraphForge, OperationId}; const FORMAT_VERSION: u32 = 1; const SESSION_DIR: &str = "import-sessions"; @@ -40,7 +38,7 @@ pub struct ImportSessionLimits { impl Default for ImportSessionLimits { fn default() -> Self { Self { - batch_rows: 8_192, + batch_rows: GraphConstructionBudgets::default().max_batch_rows, max_source_bytes: 1 << 40, max_files: 100_000, max_rejected_rows: 1_000, @@ -124,6 +122,40 @@ pub struct ImportProgress { pub peak_batch_rows: u64, /// Configured source-reader concurrency bound. pub io_concurrency_limit: u64, + /// Durable, content-free evidence from the ordinary graph-construction path. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub construction: Option, +} + +/// Sanitized durable construction evidence for an ordinary staged import. +#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)] +pub struct ImportConstructionEvidence { + /// Configured authoritative construction chunk budget. + pub configured_batch_rows: u64, + /// Number of durably accepted construction chunks. + pub accepted_chunks: u64, + /// Whether the sole generation publication was committed. + pub publication_committed: bool, + /// Exact application-I/O attribution across the closed construction phases. + pub application_io: graphforge_storage::ConstructionPhaseAttribution, + /// Exact accepted input rows. + pub input_rows: u64, + /// Exact non-replay input batches. + pub input_batches: u64, + /// Immutable construction artifacts accepted from authenticated receipts. + pub immutable_artifacts: u64, + /// Application payload bytes submitted by construction artifact writers. + pub write_bytes: u64, + /// Application write submissions by construction artifact writers. + pub write_operations: u64, + /// File and directory durability barriers completed by construction. + pub fsync_operations: u64, + /// Largest retained Arrow row window. + pub peak_batch_rows: u64, + /// Largest retained Arrow byte window. + pub peak_batch_bytes: u64, + /// Exact transient allocation high-water retained across resume. + pub transient_peak_allocated_bytes: u64, } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -151,6 +183,8 @@ struct SessionManifest { progress: ImportProgress, sources: Vec, #[serde(default)] + construction_session_uuid: Option, + #[serde(default)] updated_unix_millis: u64, } @@ -177,15 +211,6 @@ impl GraphForge { )); } fs::create_dir_all(root.join("sources")).map_err(storage)?; - copy_tree(&self.dir, &root.join("graph"))?; - if !graphforge_storage::uuid_membership_index_is_fresh(&root.join("graph"))? { - graphforge_storage::rebuild_uuid_membership_indexes( - &root.join("graph"), - graphforge_storage::UuidIndexBuildLimits::default(), - )?; - } - File::create(root.join("nodes.uuidx")).map_err(storage)?; - File::create(root.join("edges.uuidx")).map_err(storage)?; let manifest = SessionManifest { format_version: FORMAT_VERSION, session_uuid, @@ -201,6 +226,7 @@ impl GraphForge { ..ImportProgress::default() }, sources: Vec::new(), + construction_session_uuid: None, updated_unix_millis: unix_millis()?, }; write_manifest(&root, &manifest)?; @@ -231,6 +257,19 @@ impl GraphForge { }) } + /// Reopen the durable, content-free status of any import session, including terminal ones. + pub fn import_session_status( + &self, + session_uuid: Uuid, + ) -> Result<(ImportPhase, ImportProgress), GfError> { + let root = import_root(self, session_uuid)?; + let manifest = read_manifest(&root)?; + if manifest.format_version != FORMAT_VERSION || manifest.session_uuid != session_uuid { + return Err(validation("incompatible or mismatched import manifest")); + } + Ok((manifest.phase, manifest.progress)) + } + /// Abort and remove durable staging for non-terminal sessions older than `max_age`. pub fn cleanup_stale_import_sessions(&self, max_age: Duration) -> Result { let sessions = self.resolved_generation.container_root().join(SESSION_DIR); @@ -250,7 +289,7 @@ impl GraphForge { continue; } let root = entry.path(); - let mut manifest = read_manifest(&root)?; + let manifest = read_manifest(&root)?; if matches!( manifest.phase, ImportPhase::Committed | ImportPhase::Aborted @@ -258,111 +297,16 @@ impl GraphForge { { continue; } - manifest.phase = ImportPhase::Aborted; - manifest.updated_unix_millis = now; - write_manifest(&root, &manifest)?; - for path in [root.join("sources"), root.join("graph")] { - if path.exists() { - fs::remove_dir_all(path).map_err(storage)?; - } + GraphImportSession { + root, + manifest, + observed: Instant::now(), } + .abort(self)?; cleaned = cleaned.saturating_add(1); } Ok(cleaned) } - - fn publish_import_tree( - &self, - staged_graph: &Path, - operation_uuid: Uuid, - expected_parent: Uuid, - cancellation: Option<&CancellationToken>, - ) -> Result { - use graphforge_storage::{ - ProjectCapability, ProjectGenerationRequest, ProjectStageOutcome, - }; - - let _visibility = self.graph_visibility.acquire(cancellation)?; - cancellation.map_or(Ok(()), CancellationToken::checkpoint)?; - let container = self.resolved_generation.container_root(); - let parent = graphforge_storage::resolve_project_generation(container)?; - parent.validate_complete_participant_inventory()?; - if parent.generation_uuid() != expected_parent { - return Err(validation( - "project generation changed before import commit", - )); - } - if !graphforge_storage::uuid_membership_index_is_fresh(staged_graph)? { - graphforge_storage::rebuild_uuid_membership_indexes( - staged_graph, - graphforge_storage::UuidIndexBuildLimits::default(), - )?; - } - let graph_files = graphforge_storage::capture_graph_files(staged_graph)?.1; - let recorded_at = (self.clock.lock().expect("clock lock poisoned"))()?; - let receipt = graphforge_exec::MutationReceipt::default(); - let participants = super::graph_publication_participants( - &parent, - graph_files, - self.semantic_storage_bindings - .lock() - .expect("semantic storage binding lock poisoned") - .as_ref(), - parent.capability("provenance")?.is_some(), - &receipt, - operation_uuid, - None, - recorded_at, - )?; - let generation_uuid = super::mutation_generation_uuid(operation_uuid, &participants); - let request = ProjectGenerationRequest { - transaction_uuid: operation_uuid, - generation_uuid, - capabilities: parent - .capabilities() - .into_iter() - .map(|capability| ProjectCapability { - capability_id: capability.capability_id, - capability_version: capability.capability_version, - }) - .collect(), - participants, - }; - cancellation.map_or(Ok(()), CancellationToken::checkpoint)?; - let publication = match graphforge_storage::stage_project_generation_with_graph_tree_mode( - container, - &request, - Some(staged_graph), - self.lifecycle_mode, - )? { - ProjectStageOutcome::AlreadyPublished(receipt) => receipt, - ProjectStageOutcome::Staged(staged) => staged - .validate( - |_| Ok(()), - |actual_parent, _| { - if actual_parent.generation_uuid() != expected_parent { - return Err(validation( - "project generation changed before import publication", - )); - } - Ok(()) - }, - )? - .publish()?, - }; - *self - .current_generation_uuid - .lock() - .expect("generation UUID lock poisoned") = publication.generation_uuid; - let resolved = graphforge_storage::resolve_project_generation(container)?; - super::rematerialize_graph_workspace(&resolved, &self.dir)?; - *self - .runtime_catalog - .lock() - .expect("runtime catalog poisoned") = super::load_runtime_catalog(&self.dir)?; - self.adjacency_provider.invalidate(); - Ok(publication.generation_uuid) - } } impl GraphImportSession { @@ -487,6 +431,9 @@ impl GraphImportSession { ) -> Result { self.ensure_open()?; self.ensure_base(graph)?; + let mut construction = self.open_construction(graph)?; + let session_root = self.root.clone(); + let batch_rows = self.manifest.limits.batch_rows; for input_kind in [BulkInputKind::Node, BulkInputKind::Edge] { for source_index in 0..self.manifest.sources.len() { let source = self.manifest.sources[source_index].clone(); @@ -494,101 +441,118 @@ impl GraphImportSession { continue; } let mut batch_index = 0_u64; - for_each_source_batch( - &self.root, - &source, - self.manifest.limits.batch_rows, - |batch| { - if batch_index < source.batches_staged { - batch_index += 1; - return Ok(()); - } - if cancellation.is_some_and(CancellationToken::is_cancelled) { - return Err(cancelled()); - } - let operation = import_batch_operation( - self.manifest.operation_uuid, - source.sequence, - batch_index, - ); - let recovering = source.inflight_batch == Some(batch_index); - if !recovering { - self.manifest.sources[source_index].inflight_batch = Some(batch_index); - write_manifest(&self.root, &self.manifest)?; - } - let staged = match input_kind { - BulkInputKind::Node => { - stage_node_batch(graph, &self.root, operation, &batch, recovering) - } - BulkInputKind::Edge => { - stage_edge_batch(graph, &self.root, operation, &batch, recovering) - } - }; - if let Err(error) = staged { - self.manifest.sources[source_index].inflight_batch = None; - let remaining = self - .manifest - .limits - .max_rejected_rows - .saturating_sub(self.manifest.progress.rows_rejected); - self.manifest.progress.rows_rejected = self - .manifest - .progress - .rows_rejected - .saturating_add((batch.num_rows() as u64).min(remaining)); - write_manifest(&self.root, &self.manifest)?; - return Err(error); - } + for_each_source_batch(&session_root, &source, batch_rows, |batch| { + if batch_index < source.batches_staged { + batch_index += 1; + return Ok(()); + } + if cancellation.is_some_and(CancellationToken::is_cancelled) { + return Err(cancelled()); + } + let operation = import_batch_operation( + self.manifest.operation_uuid, + source.sequence, + batch_index, + ); + let batch = match input_kind { + BulkInputKind::Node => graph.normalize_import_node_chunk(operation, &batch), + BulkInputKind::Edge => graph.normalize_import_edge_chunk(operation, &batch), + } + .map_err(|error| validation(error.to_string()))?; + if batch.num_rows() == 0 { batch_index += 1; self.manifest.sources[source_index].batches_staged = batch_index; self.manifest.sources[source_index].inflight_batch = None; - self.manifest.progress.rows_accepted = self + write_manifest(&self.root, &self.manifest)?; + return Ok(()); + } + let recovering = source.inflight_batch == Some(batch_index); + if !recovering { + self.manifest.sources[source_index].inflight_batch = Some(batch_index); + write_manifest(&self.root, &self.manifest)?; + } + let chunk_id = format!("import-{:020}-{:020}", source.sequence, batch_index); + let staged = match (input_kind, cancellation) { + (BulkInputKind::Node, Some(token)) => { + construction.append_nodes_with_cancellation(&chunk_id, &batch, token) + } + (BulkInputKind::Node, None) => construction.append_nodes(&chunk_id, &batch), + (BulkInputKind::Edge, Some(token)) => { + construction.append_edges_with_cancellation(&chunk_id, &batch, token) + } + (BulkInputKind::Edge, None) => construction.append_edges(&chunk_id, &batch), + }; + if let Err(error) = staged { + self.manifest.sources[source_index].inflight_batch = None; + let remaining = self .manifest - .progress - .rows_accepted - .saturating_add(batch.num_rows() as u64); - self.manifest.progress.peak_batch_rows = self + .limits + .max_rejected_rows + .saturating_sub(self.manifest.progress.rows_rejected); + self.manifest.progress.rows_rejected = self .manifest .progress - .peak_batch_rows - .max(batch.num_rows() as u64); + .rows_rejected + .saturating_add((batch.num_rows() as u64).min(remaining)); write_manifest(&self.root, &self.manifest)?; - Ok(()) - }, - )?; + return Err(error); + } + batch_index += 1; + self.manifest.sources[source_index].batches_staged = batch_index; + self.manifest.sources[source_index].inflight_batch = None; + self.manifest.progress.rows_accepted = self + .manifest + .progress + .rows_accepted + .saturating_add(batch.num_rows() as u64); + self.manifest.progress.peak_batch_rows = self + .manifest + .progress + .peak_batch_rows + .max(batch.num_rows() as u64); + self.update_construction_progress(&construction.progress())?; + write_manifest(&self.root, &self.manifest)?; + Ok(()) + })?; self.manifest.sources[source_index].staged = true; self.manifest.progress.files_pending = self.manifest.progress.files_pending.saturating_sub(1); write_manifest(&self.root, &self.manifest)?; } } + construction.validate_and_seal(cancellation)?; + self.update_construction_progress(&construction.progress())?; self.manifest.phase = ImportPhase::Validated; self.checkpoint() } /// Abort without changing CURRENT; removes staged sources or quarantines on cleanup failure. - pub fn abort(mut self) -> Result { + pub fn abort(mut self, graph: &GraphForge) -> Result { if self.manifest.phase == ImportPhase::Committed { return Err(validation("committed import cannot be aborted")); } - self.manifest.phase = ImportPhase::Aborted; - self.checkpoint()?; - let progress = self.manifest.progress.clone(); - let cleanup = [self.root.join("sources"), self.root.join("graph")] - .into_iter() - .try_for_each(|path| { - if path.exists() { - fs::remove_dir_all(path) - } else { - Ok(()) - } - }); + let cleanup = (|| { + if let Some(session_uuid) = self.manifest.construction_session_uuid { + graph + .resume_graph_construction(session_uuid, self.construction_budgets())? + .discard()?; + self.manifest.construction_session_uuid = None; + } + let sources = self.root.join("sources"); + if sources.exists() { + fs::remove_dir_all(sources).map_err(storage)?; + } + Ok::<(), GfError>(()) + })(); match cleanup { - Ok(()) => Ok(progress), + Ok(()) => { + self.manifest.phase = ImportPhase::Aborted; + self.checkpoint() + } Err(error) => { self.manifest.phase = ImportPhase::Quarantined; let _ = write_manifest(&self.root, &self.manifest); - Err(storage(error)) + Err(error) } } } @@ -599,29 +563,21 @@ impl GraphImportSession { graph: &GraphForge, cancellation: Option<&CancellationToken>, ) -> Result { - if let Some(published) = graphforge_storage::published_project_transaction( - graph.resolved_generation.container_root(), - self.manifest.operation_uuid, - )? { - self.manifest.phase = ImportPhase::Committed; - self.checkpoint()?; - return Ok(published.generation_uuid); - } - self.ensure_base(graph)?; if self.manifest.phase != ImportPhase::Validated || self.manifest.progress.files_pending != 0 { return Err(validation("import must be fully validated before commit")); } - let generation = graph.publish_import_tree( - &self.root.join("graph"), - self.manifest.operation_uuid, - self.manifest.base_generation_uuid, - cancellation, - )?; + self.ensure_base(graph)?; + let mut construction = self.open_construction(graph)?; + let publication = match cancellation { + Some(token) => construction.seal_and_publish_with_cancellation(token)?, + None => construction.seal_and_publish()?, + }; + self.update_construction_progress(&construction.progress())?; self.manifest.phase = ImportPhase::Committed; self.checkpoint()?; - Ok(generation) + Ok(publication.generation_uuid) } fn ensure_open(&self) -> Result<(), GfError> { @@ -646,6 +602,57 @@ impl GraphImportSession { Ok(()) } + fn construction_budgets(&self) -> GraphConstructionBudgets { + let mut budgets = GraphConstructionBudgets::default(); + budgets.max_batch_rows = self.manifest.limits.batch_rows; + budgets.max_run_records = budgets + .max_run_records + .max(self.manifest.limits.batch_rows.saturating_mul(4)); + budgets + } + + fn open_construction<'a>( + &mut self, + graph: &'a GraphForge, + ) -> Result, GfError> { + let budgets = self.construction_budgets(); + if let Some(session_uuid) = self.manifest.construction_session_uuid { + return graph.resume_graph_construction(session_uuid, budgets); + } + let session = graph.begin_graph_construction(budgets)?; + self.manifest.construction_session_uuid = Some(session.session_uuid()); + write_manifest(&self.root, &self.manifest)?; + Ok(session) + } + + fn update_construction_progress( + &mut self, + progress: &crate::GraphConstructionProgress, + ) -> Result<(), GfError> { + let application_io = + graphforge_storage::ConstructionPhaseAttribution::from_construction(&progress.evidence); + application_io.validate_for_qualification()?; + self.manifest.progress.construction = Some(ImportConstructionEvidence { + configured_batch_rows: u64::try_from(self.manifest.limits.batch_rows) + .unwrap_or(u64::MAX), + accepted_chunks: progress.accepted_chunks, + publication_committed: progress.publication_committed, + application_io, + input_rows: progress.evidence.input_rows, + input_batches: progress.evidence.input_batches, + immutable_artifacts: progress.evidence.immutable_artifacts, + write_bytes: progress.evidence.write_bytes, + write_operations: progress.evidence.write_operations, + fsync_operations: progress.evidence.fsync_operations, + peak_batch_rows: progress.evidence.peak_batch_rows, + peak_batch_bytes: progress.evidence.peak_batch_bytes, + transient_peak_allocated_bytes: progress + .evidence + .storage_transient_peak_total_allocated_bytes, + }); + Ok(()) + } + fn next_sequence(&self) -> Result { if self.manifest.sources.len() as u64 >= self.manifest.limits.max_files { return Err(limit("import max_files exceeded")); @@ -692,286 +699,6 @@ fn unix_millis() -> Result { .unwrap_or(u64::MAX)) } -fn stage_node_batch( - graph: &GraphForge, - root: &Path, - operation: OperationId, - batch: &RecordBatch, - allow_replay: bool, -) -> Result<(), GfError> { - let graph_dir = root.join("graph"); - let normalized = graph - .normalize_import_nodes(operation, std::slice::from_ref(batch)) - .map_err(|error| validation(error.to_string()))?; - let candidates = normalized - .rows() - .iter() - .map(|row| row.node_uuid) - .collect::>(); - let base_nodes = graph - .import_base_membership( - &candidates, - graphforge_storage::UuidIndexKind::Node, - BulkInputKind::Node, - ) - .map_err(|error| validation(error.to_string()))?; - let base_edges = graph - .import_base_membership( - &candidates, - graphforge_storage::UuidIndexKind::Edge, - BulkInputKind::Node, - ) - .map_err(|error| validation(error.to_string()))?; - let session_nodes = probe_session_index(&root.join("nodes.uuidx"), &candidates)?; - let session_edges = probe_session_index(&root.join("edges.uuidx"), &candidates)?; - if allow_replay - && session_nodes.iter().all(|found| *found) - && base_nodes.iter().all(|found| !*found) - && base_edges.iter().all(|found| !*found) - && session_edges.iter().all(|found| !*found) - { - return Ok(()); - } - if base_nodes - .iter() - .chain(&base_edges) - .chain(&session_nodes) - .chain(&session_edges) - .any(|found| *found) - { - return Err(validation( - "import node UUID conflicts with staged graph identity", - )); - } - let mut catalog = super::load_runtime_catalog(&graph_dir)?; - let mut writer = graphforge_storage::GraphWriter::open(&graph_dir, graph.ontology_mode)?; - for row in normalized.rows() { - let type_id = graph - .ontology - .as_ref() - .and_then(|ontology| ontology.entity_type_id(&row.label)) - .unwrap_or_else(|| { - graphforge_ir::runtime_entity_type_id(catalog.intern_label(&row.label)) - }); - writer.create_node(row.node_uuid, type_id)?; - let properties = row - .properties - .iter() - .map(|(name, value)| { - catalog.intern_property(name, Some(&row.label)); - Ok((name.clone(), crate::construction::prop_literal(value)?)) - }) - .collect::, GfError>>()?; - if !properties.is_empty() { - writer.set_properties(&row.node_uuid, Some(&row.label), properties)?; - } - } - writer.flush()?; - super::persist_runtime_catalog(&graph_dir, &catalog)?; - merge_session_index(&root.join("nodes.uuidx"), &candidates)?; - Ok(()) -} - -fn stage_edge_batch( - graph: &GraphForge, - root: &Path, - operation: OperationId, - batch: &RecordBatch, - allow_replay: bool, -) -> Result<(), GfError> { - let graph_dir = root.join("graph"); - let endpoints = batch_endpoint_uuids(batch)?; - let imported_found = probe_session_index(&root.join("nodes.uuidx"), &endpoints)?; - let imported_endpoints = endpoints - .iter() - .zip(imported_found) - .filter_map(|(candidate, found)| found.then_some(*candidate)) - .collect::>(); - let normalized = graph - .normalize_import_edges(operation, std::slice::from_ref(batch), &imported_endpoints) - .map_err(|error| validation(error.to_string()))?; - let candidates = normalized - .rows() - .iter() - .map(|row| row.edge_uuid) - .collect::>(); - let base_edges = graph - .import_base_membership( - &candidates, - graphforge_storage::UuidIndexKind::Edge, - BulkInputKind::Edge, - ) - .map_err(|error| validation(error.to_string()))?; - let base_nodes = graph - .import_base_membership( - &candidates, - graphforge_storage::UuidIndexKind::Node, - BulkInputKind::Edge, - ) - .map_err(|error| validation(error.to_string()))?; - let session_edges = probe_session_index(&root.join("edges.uuidx"), &candidates)?; - let session_nodes = probe_session_index(&root.join("nodes.uuidx"), &candidates)?; - if allow_replay - && session_edges.iter().all(|found| *found) - && base_edges.iter().all(|found| !*found) - && base_nodes.iter().all(|found| !*found) - && session_nodes.iter().all(|found| !*found) - { - return Ok(()); - } - if base_edges - .iter() - .chain(&base_nodes) - .chain(&session_edges) - .chain(&session_nodes) - .any(|found| *found) - { - return Err(validation( - "import edge UUID conflicts with staged graph identity", - )); - } - let mut catalog = super::load_runtime_catalog(&graph_dir)?; - let mut writer = graphforge_storage::GraphWriter::open(&graph_dir, graph.ontology_mode)?; - let endpoints = normalized - .rows() - .iter() - .flat_map(|row| [row.source_uuid, row.target_uuid]) - .collect::>(); - crate::bulk_construction::register_existing_endpoints(&mut writer, &graph_dir, &endpoints)?; - for row in normalized.rows() { - catalog.intern_relation_type(&row.rel_type); - writer.create_edge( - row.edge_uuid, - &row.rel_type, - &row.source_uuid, - &row.target_uuid, - )?; - let properties = row - .properties - .iter() - .map(|(name, value)| { - catalog.intern_property(name, Some(&row.rel_type)); - Ok((name.clone(), crate::construction::prop_literal(value)?)) - }) - .collect::, GfError>>()?; - if !properties.is_empty() { - writer.set_edge_properties(&row.edge_uuid, Some(&row.rel_type), properties)?; - } - } - writer.flush()?; - super::persist_runtime_catalog(&graph_dir, &catalog)?; - merge_session_index(&root.join("edges.uuidx"), &candidates)?; - Ok(()) -} - -fn batch_endpoint_uuids(batch: &RecordBatch) -> Result, GfError> { - let mut endpoints = Vec::with_capacity(batch.num_rows().saturating_mul(2)); - for name in ["source_uuid", "target_uuid"] { - let column = batch - .column_by_name(name) - .ok_or_else(|| validation(format!("missing {name}")))?; - let values = column - .as_any() - .downcast_ref::() - .ok_or_else(|| validation(format!("{name} must be fixed-size binary")))?; - for row in 0..values.len() { - endpoints.push(Uuid::from_slice(values.value(row)).map_err(storage)?); - } - } - endpoints.sort_unstable(); - endpoints.dedup(); - Ok(endpoints) -} - -fn probe_session_index(path: &Path, candidates: &[Uuid]) -> Result, GfError> { - let mut file = File::open(path).map_err(storage)?; - let bytes = file.metadata().map_err(storage)?.len(); - if bytes % 16 != 0 { - return Err(validation("corrupt import membership index")); - } - let count = bytes / 16; - candidates - .iter() - .map(|candidate| { - let mut low = 0_u64; - let mut high = count; - let needle = candidate.as_bytes(); - let mut slot = [0_u8; 16]; - while low < high { - let mid = low + (high - low) / 2; - file.seek(SeekFrom::Start(mid * 16)).map_err(storage)?; - file.read_exact(&mut slot).map_err(storage)?; - match slot.cmp(needle) { - std::cmp::Ordering::Less => low = mid + 1, - std::cmp::Ordering::Greater => high = mid, - std::cmp::Ordering::Equal => return Ok(true), - } - } - Ok(false) - }) - .collect() -} - -fn merge_session_index(path: &Path, candidates: &[Uuid]) -> Result<(), GfError> { - let mut additions = candidates.to_vec(); - additions.sort_unstable(); - additions.dedup(); - let old_file = File::open(path).map_err(storage)?; - if old_file.metadata().map_err(storage)?.len() % 16 != 0 { - return Err(validation("corrupt import membership index")); - } - let temp = path.with_extension("uuidx.tmp"); - let mut output = BufWriter::new(File::create(&temp).map_err(storage)?); - let mut old = BufReader::new(old_file); - let mut old_value = read_index_uuid(&mut old)?; - let mut new_values = additions.iter().peekable(); - while old_value.is_some() || new_values.peek().is_some() { - match (old_value, new_values.peek()) { - (Some(current), Some(new_value)) if current <= *new_value.as_bytes() => { - output.write_all(¤t).map_err(storage)?; - old_value = read_index_uuid(&mut old)?; - } - (_, Some(new_value)) => { - output.write_all(new_value.as_bytes()).map_err(storage)?; - new_values.next(); - } - (Some(current), None) => { - output.write_all(¤t).map_err(storage)?; - old_value = read_index_uuid(&mut old)?; - } - (None, None) => break, - } - } - output.flush().map_err(storage)?; - output.get_ref().sync_all().map_err(storage)?; - drop(output); - fs::rename(temp, path).map_err(storage) -} - -fn read_index_uuid(reader: &mut BufReader) -> Result, GfError> { - if reader.fill_buf().map_err(storage)?.is_empty() { - return Ok(None); - } - let mut value = [0_u8; 16]; - reader.read_exact(&mut value).map_err(storage)?; - Ok(Some(value)) -} - -fn import_batch_operation(base: Uuid, source: u64, batch: u64) -> OperationId { - let mut digest = Sha256::new(); - digest.update(b"graphforge.import.batch.v1\0"); - digest.update(base.as_bytes()); - digest.update(source.to_be_bytes()); - digest.update(batch.to_be_bytes()); - let digest = digest.finalize(); - let mut bytes = [0_u8; 16]; - bytes[..6].copy_from_slice(&base.as_bytes()[..6]); - bytes[6..].copy_from_slice(&digest[..10]); - bytes[6] = (bytes[6] & 0x0f) | 0x70; - bytes[8] = (bytes[8] & 0x3f) | 0x80; - OperationId(Uuid::from_bytes(bytes)) -} - fn for_each_source_batch( root: &Path, source: &SourceRecord, @@ -1063,24 +790,19 @@ fn reject_unsafe_path(path: &Path) -> Result<(), GfError> { Ok(()) } -fn copy_tree(source: &Path, destination: &Path) -> Result<(), GfError> { - fs::create_dir_all(destination).map_err(storage)?; - for entry in fs::read_dir(source).map_err(storage)? { - let entry = entry.map_err(storage)?; - let file_type = entry.file_type().map_err(storage)?; - if file_type.is_symlink() { - return Err(validation("staged graph copy refuses symlink entries")); - } - let target = destination.join(entry.file_name()); - if file_type.is_dir() { - copy_tree(&entry.path(), &target)?; - } else if file_type.is_file() { - fs::copy(entry.path(), &target).map_err(storage)?; - } else { - return Err(validation("staged graph copy refuses special files")); - } - } - Ok(()) +fn import_batch_operation(base: Uuid, source: u64, batch: u64) -> OperationId { + let mut digest = Sha256::new(); + digest.update(b"graphforge.import.batch.v1\0"); + digest.update(base.as_bytes()); + digest.update(source.to_be_bytes()); + digest.update(batch.to_be_bytes()); + let digest = digest.finalize(); + let mut bytes = [0_u8; 16]; + bytes[..6].copy_from_slice(&base.as_bytes()[..6]); + bytes[6..].copy_from_slice(&digest[..10]); + bytes[6] = (bytes[6] & 0x0f) | 0x70; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + OperationId(Uuid::from_bytes(bytes)) } fn sync_file(path: &Path) -> Result<(), GfError> { @@ -1113,9 +835,11 @@ fn cancelled() -> GfError { #[cfg(test)] mod tests { + use std::collections::HashMap; use std::sync::Arc; use arrow::array::{FixedSizeBinaryArray, StringArray}; + use arrow::datatypes::DataType; use parquet::arrow::ArrowWriter; use super::*; @@ -1162,6 +886,14 @@ mod tests { (directory, project, graph) } + fn construction_root(graph: &GraphForge, session_uuid: Uuid) -> PathBuf { + graph + .resolved_generation + .container_root() + .join(".graphforge-construction") + .join(session_uuid.simple().to_string()) + } + #[test] fn arrow_session_resumes_stages_and_publishes_one_generation() { let (_directory, project, graph) = fixture(); @@ -1226,8 +958,16 @@ mod tests { .register_parquet(BulkInputKind::Node, &parquet) .unwrap(); session.validate(&graph).unwrap(); - let progress = session.abort().unwrap(); + let construction_uuid = session.manifest.construction_session_uuid.unwrap(); + assert!(construction_root(&graph, construction_uuid).exists()); + let progress = session.abort(&graph).unwrap(); assert_eq!(progress.rows_accepted, 1); + assert!(!construction_root(&graph, construction_uuid).exists()); + assert!( + graph + .resume_graph_construction(construction_uuid, GraphConstructionBudgets::default()) + .is_err() + ); assert_eq!(*graph.current_generation_uuid.lock().unwrap(), before); drop(graph); GraphForge::new(project.to_str()).unwrap(); @@ -1323,14 +1063,29 @@ mod tests { .append_arrow(BulkInputKind::Node, std::slice::from_ref(&batch)) .unwrap(); let batch_operation = import_batch_operation(operation.0, 0, 0); - stage_node_batch(&graph, &session.root, batch_operation, &batch, false).unwrap(); + let normalized = graph + .normalize_import_node_chunk(batch_operation, &batch) + .unwrap(); + let mut construction = session.open_construction(&graph).unwrap(); + construction + .append_nodes( + "import-00000000000000000000-00000000000000000000", + &normalized, + ) + .unwrap(); + drop(construction); session.manifest.sources[0].inflight_batch = Some(0); write_manifest(&session.root, &session.manifest).unwrap(); let session_uuid = session.session_uuid(); drop(session); let mut resumed = graph.resume_import_session(session_uuid).unwrap(); - assert_eq!(resumed.validate(&graph).unwrap().rows_accepted, 1); + let progress = resumed.validate(&graph).unwrap(); + assert_eq!(progress.rows_accepted, 1); + let construction = progress.construction.unwrap(); + assert_eq!(construction.accepted_chunks, 1); + assert_eq!(construction.input_batches, 1); + let construction_uuid = resumed.manifest.construction_session_uuid.unwrap(); drop(resumed); let mut manifest = read_manifest(&import_root(&graph, session_uuid).unwrap()).unwrap(); @@ -1343,8 +1098,185 @@ mod tests { 1 ); let root = import_root(&graph, session_uuid).unwrap(); - assert!(!root.join("graph").exists()); assert!(!root.join("sources").exists()); + assert!(!construction_root(&graph, construction_uuid).exists()); assert_eq!(read_manifest(&root).unwrap().phase, ImportPhase::Aborted); } + + #[test] + fn zero_row_node_and_edge_sources_are_canonical_and_publishable() { + let (_directory, project, graph) = fixture(); + let empty_nodes = RecordBatch::new_empty(bulk_node_input_schema(Vec::new()).unwrap()); + let empty_edges = RecordBatch::new_empty(bulk_edge_input_schema(Vec::new()).unwrap()); + let normalized_nodes = graph + .normalize_import_node_chunk(OperationId(Uuid::now_v7()), &empty_nodes) + .unwrap(); + let normalized_edges = graph + .normalize_import_edge_chunk(OperationId(Uuid::now_v7()), &empty_edges) + .unwrap(); + assert_eq!(normalized_nodes.num_rows(), 0); + assert_eq!(normalized_edges.num_rows(), 0); + assert_eq!( + normalized_nodes.column(0).data_type(), + &DataType::FixedSizeBinary(16) + ); + assert_eq!( + normalized_edges.column(0).data_type(), + &DataType::FixedSizeBinary(16) + ); + + let retained_node = Uuid::now_v7(); + let mut session = graph + .begin_import_session(OperationId(Uuid::now_v7()), ImportSessionLimits::default()) + .unwrap(); + session + .append_arrow(BulkInputKind::Node, &[empty_nodes, nodes(&[retained_node])]) + .unwrap(); + session + .append_arrow(BulkInputKind::Edge, &[empty_edges]) + .unwrap(); + let progress = session.validate(&graph).unwrap(); + assert_eq!(progress.rows_accepted, 1); + assert_eq!(progress.files_pending, 0); + assert_eq!(progress.construction.as_ref().unwrap().accepted_chunks, 1); + let generation = session.commit(&graph, None).unwrap(); + + drop(graph); + let reopened = GraphForge::new(project.to_str()).unwrap(); + assert_eq!( + *reopened.current_generation_uuid.lock().unwrap(), + generation + ); + assert_eq!(reopened.node_count("Person").unwrap(), 1); + } + + #[test] + fn commit_rechecks_the_import_base_generation() { + let (_directory, _project, graph) = fixture(); + let mut session = graph + .begin_import_session(OperationId(Uuid::now_v7()), ImportSessionLimits::default()) + .unwrap(); + session + .append_arrow(BulkInputKind::Node, &[nodes(&[Uuid::now_v7()])]) + .unwrap(); + session.validate(&graph).unwrap(); + graph.add_node("Other", &HashMap::new()).unwrap(); + let independent = *graph.current_generation_uuid.lock().unwrap(); + + let error = session.commit(&graph, None).unwrap_err(); + assert!( + matches!(error, GfError::Validation(message) if message == "project generation changed since import began") + ); + assert_eq!(*graph.current_generation_uuid.lock().unwrap(), independent); + } + + #[test] + fn stale_cleanup_quarantines_when_construction_authority_changed() { + let (_directory, _project, graph) = fixture(); + let mut session = graph + .begin_import_session(OperationId(Uuid::now_v7()), ImportSessionLimits::default()) + .unwrap(); + session + .append_arrow(BulkInputKind::Node, &[nodes(&[Uuid::now_v7()])]) + .unwrap(); + session.validate(&graph).unwrap(); + let session_uuid = session.session_uuid(); + let construction_uuid = session.manifest.construction_session_uuid.unwrap(); + drop(session); + + graph.add_node("Other", &HashMap::new()).unwrap(); + let independent = *graph.current_generation_uuid.lock().unwrap(); + let root = import_root(&graph, session_uuid).unwrap(); + let mut manifest = read_manifest(&root).unwrap(); + manifest.updated_unix_millis = 0; + write_manifest(&root, &manifest).unwrap(); + + let error = graph + .cleanup_stale_import_sessions(Duration::from_secs(1)) + .unwrap_err(); + assert!(matches!( + error, + GfError::Validation(_) | GfError::Storage(_) + )); + assert_eq!( + read_manifest(&root).unwrap().phase, + ImportPhase::Quarantined + ); + assert!(construction_root(&graph, construction_uuid).exists()); + assert_eq!(*graph.current_generation_uuid.lock().unwrap(), independent); + } + + #[test] + fn parquet_construction_receipts_scale_linearly_and_survive_reopen() { + assert_eq!(ImportSessionLimits::default().batch_rows, 65_536); + + fn run(multiplier: usize) -> ImportConstructionEvidence { + let (_directory, project, graph) = fixture(); + let source_dir = tempfile::tempdir().unwrap(); + let parquet = source_dir.path().join("nodes.parquet"); + let ids = (0..(4 * multiplier)) + .map(|_| Uuid::now_v7()) + .collect::>(); + let batch = nodes(&ids); + let mut writer = + ArrowWriter::try_new(File::create(&parquet).unwrap(), batch.schema(), None) + .unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + let limits = ImportSessionLimits { + batch_rows: 4, + ..ImportSessionLimits::default() + }; + let mut session = graph + .begin_import_session(OperationId(Uuid::now_v7()), limits) + .unwrap(); + session + .register_parquet(BulkInputKind::Node, &parquet) + .unwrap(); + let session_uuid = session.session_uuid(); + let validated = session.validate(&graph).unwrap(); + assert_eq!( + validated.construction.as_ref().unwrap().accepted_chunks, + multiplier as u64 + ); + session.commit(&graph, None).unwrap(); + let (_, durable) = graph.import_session_status(session_uuid).unwrap(); + let receipt = durable.construction.unwrap(); + assert!(receipt.publication_committed); + assert_eq!(receipt.input_rows, (4 * multiplier) as u64); + assert_eq!(receipt.input_batches, multiplier as u64); + assert_eq!(receipt.peak_batch_rows, 4); + + drop(graph); + let reopened = GraphForge::new(project.to_str()).unwrap(); + let (phase, reopened_progress) = reopened.import_session_status(session_uuid).unwrap(); + assert_eq!(phase, ImportPhase::Committed); + assert_eq!(reopened_progress.construction.as_ref(), Some(&receipt)); + receipt + } + + let receipts = [run(1), run(2), run(4)]; + for (previous, next) in receipts.iter().zip(receipts.iter().skip(1)) { + assert_eq!(next.accepted_chunks, previous.accepted_chunks * 2); + assert_eq!(next.input_rows, previous.input_rows * 2); + assert_eq!(next.input_batches, previous.input_batches * 2); + for (smaller, larger) in [ + (previous.write_bytes, next.write_bytes), + (previous.write_operations, next.write_operations), + (previous.immutable_artifacts, next.immutable_artifacts), + (previous.fsync_operations, next.fsync_operations), + ( + previous.application_io.totals.write_calls, + next.application_io.totals.write_calls, + ), + ] { + assert!(larger >= smaller, "durable work must be monotonic"); + assert!( + larger <= smaller.saturating_mul(3), + "doubling rows exceeded the bounded linear work envelope" + ); + } + } + } } diff --git a/crates/graphforge-api/src/lib.rs b/crates/graphforge-api/src/lib.rs index b3ef258f7..28c85ba8d 100644 --- a/crates/graphforge-api/src/lib.rs +++ b/crates/graphforge-api/src/lib.rs @@ -204,7 +204,8 @@ pub use graphforge_core::{ SpatialType, SpatialValue, TemporalValue, }; pub use import_session::{ - GraphImportSession, ImportPhase, ImportProgress, ImportSessionLimits, ImportSourceKind, + GraphImportSession, ImportConstructionEvidence, ImportPhase, ImportProgress, + ImportSessionLimits, ImportSourceKind, }; // The Arrow-backed result of [`GraphForge::execute`]. pub use generation_diff::{ diff --git a/crates/graphforge-api/src/resumable_construction.rs b/crates/graphforge-api/src/resumable_construction.rs index e14860146..c037cb7e4 100644 --- a/crates/graphforge-api/src/resumable_construction.rs +++ b/crates/graphforge-api/src/resumable_construction.rs @@ -214,6 +214,15 @@ impl GraphConstructionSession<'_> { } } + /// Complete global shape and endpoint validation without publishing CURRENT. + pub(crate) fn validate_and_seal( + &mut self, + cancellation: Option<&crate::CancellationToken>, + ) -> Result<(), GfError> { + self.prepare_encoding(cancellation)?; + Ok(()) + } + /// Seal, shape, encode, and atomically publish exactly one generation. pub fn seal_and_publish(&mut self) -> Result { self.seal_and_publish_inner(None) @@ -235,19 +244,7 @@ impl GraphConstructionSession<'_> { token.checkpoint()?; } let _visibility = self.graph.graph_visibility.lock()?; - let topology_generation = self.inner.parent_topology_generation().saturating_add(1); - let encoding = if self.inner.state() == GraphConstructionState::Staging { - self.inner - .seal_and_prepare_canonical_encoding_with_cancellation( - topology_generation, - || cancellation.is_some_and(crate::CancellationToken::is_cancelled), - )? - } else { - self.inner - .prepare_canonical_encoding_with_cancellation(topology_generation, || { - cancellation.is_some_and(crate::CancellationToken::is_cancelled) - })? - }; + let encoding = self.prepare_encoding(cancellation)?; if let Some(token) = cancellation { token.checkpoint()?; } @@ -323,10 +320,33 @@ impl GraphConstructionSession<'_> { }) } + fn prepare_encoding( + &mut self, + cancellation: Option<&crate::CancellationToken>, + ) -> Result { + let topology_generation = self.inner.parent_topology_generation().saturating_add(1); + if self.inner.state() == GraphConstructionState::Staging { + self.inner + .seal_and_prepare_canonical_encoding_with_cancellation(topology_generation, || { + cancellation.is_some_and(crate::CancellationToken::is_cancelled) + }) + } else { + self.inner + .prepare_canonical_encoding_with_cancellation(topology_generation, || { + cancellation.is_some_and(crate::CancellationToken::is_cancelled) + }) + } + } + /// Abort an unsealed session without changing project authority. pub fn abort(&mut self) -> Result<(), GfError> { self.inner.abort() } + + /// Reclaim every authenticated private artifact of an unpublished session. + pub(crate) fn discard(self) -> Result<(), GfError> { + self.inner.discard() + } } fn replace_workspace(prepared: &Path, target: &Path) -> Result<(), GfError> { diff --git a/crates/graphforge-bindings-node/src/import_session.rs b/crates/graphforge-bindings-node/src/import_session.rs index ffe659f06..0d64d2996 100644 --- a/crates/graphforge-bindings-node/src/import_session.rs +++ b/crates/graphforge-bindings-node/src/import_session.rs @@ -296,7 +296,11 @@ impl GraphImportSession { "import session handle is closed".into(), )) })?; - let progress = session.abort().map_err(|error| to_napi_err(&error))?; + let graph = self + .engine + .read() + .map_err(|_| to_napi_err(&GfError::Execution("GraphForge lock poisoned".into())))?; + let progress = session.abort(&graph).map_err(|error| to_napi_err(&error))?; Ok(progress_output(progress)) } } diff --git a/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json b/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json index 9a25b6f65..679be16cd 100644 --- a/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json +++ b/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json @@ -1,8 +1,8 @@ { "contractVersion": 1, "rustManifest": "../../../tests/contracts/non-cypher-rust-surface.json", - "releaseSurfaceCount": 258, - "releaseSurfaceDigest": "6963ebeb6999197aff6de3dbb996c72cbc41561dd2d7400ae81e6d296f8165a7", + "releaseSurfaceCount": 259, + "releaseSurfaceDigest": "5b0c6f3b4995545f6527b8be3b3fea7cb7aca0d070ad280ae6ec80e24ac9e938", "requiredEquivalent": [ "GraphForge.adopt_ontology", "GraphForge.clear_ontology", @@ -205,6 +205,12 @@ "GraphImportSession.validate" ], "languageSpecific": { + "GraphForge.import_session_status": { + "nodeMembers": [ + "GraphImportSession.status" + ], + "reason": "Node reads live session status from its owned handle; terminal receipt reopening is a Rust and CLI lifecycle operation." + }, "GraphForge.execute_stream": { "nodeMembers": [ "GraphForge.plan", diff --git a/crates/graphforge-bindings-py/src/import_session.rs b/crates/graphforge-bindings-py/src/import_session.rs index c003dd50b..3d0a2eb77 100644 --- a/crates/graphforge-bindings-py/src/import_session.rs +++ b/crates/graphforge-bindings-py/src/import_session.rs @@ -59,6 +59,17 @@ impl GraphForge { }) .map_err(|error| to_pyerr(py, &error)) } + + /// Abort and reclaim a staged import while releasing the GIL on `&self`. + pub(crate) fn run_import_abort( + &self, + py: Python<'_>, + session: GraphImportSession, + ) -> PyResult { + self.ensure_open()?; + py.detach(|| session.abort(&self.inner)) + .map_err(|error| to_pyerr(py, &error)) + } } fn phase_name(phase: ImportPhase) -> &'static str { @@ -237,9 +248,11 @@ impl PyGraphImportSession { /// Abort without changing CURRENT. fn abort(&self, py: Python<'_>) -> PyResult> { let session = self.take_inner(py)?; - let progress = py - .detach(|| session.abort()) - .map_err(|error| to_pyerr(py, &error))?; + let progress = self + .parent + .bind(py) + .borrow() + .run_import_abort(py, session)?; progress_dict(py, &progress) } } diff --git a/crates/graphforge-bindings-py/tests/non_cypher_release.py b/crates/graphforge-bindings-py/tests/non_cypher_release.py index 9a9f8c8c1..25c63b7aa 100644 --- a/crates/graphforge-bindings-py/tests/non_cypher_release.py +++ b/crates/graphforge-bindings-py/tests/non_cypher_release.py @@ -23,8 +23,8 @@ RUST_MANIFEST = ROOT / "tests/contracts/non-cypher-rust-surface.json" RUST_GATE = ROOT / "scripts/ci/non-cypher-surface-gate.py" PYO3_SOURCE = ROOT / "crates/graphforge-bindings-py/src/lib.rs" -EXPECTED_RUST_DIGEST = "8e3a0711619a5e50231bf510a76328ea44b7706dfd28c524b760c6564b805bc3" -EXPECTED_RELEASE_DIGEST = "6963ebeb6999197aff6de3dbb996c72cbc41561dd2d7400ae81e6d296f8165a7" +EXPECTED_RUST_DIGEST = "499f902e4713931d7b3591fbaf978e6694c3029fde08c5c5476547563e5776e5" +EXPECTED_RELEASE_DIGEST = "5b0c6f3b4995545f6527b8be3b3fea7cb7aca0d070ad280ae6ec80e24ac9e938" PYTHON_ONLY_METHODS = frozenset( { @@ -257,7 +257,7 @@ def _classification_report() -> dict[str, object]: for group in manifest["method_evidence_groups"].values() for method_id in group["ids"] } - assert len(release_methods) == 258 + assert len(release_methods) == 259 assert _digest(release_methods) == EXPECTED_RELEASE_DIGEST assert set(EVIDENCE) == set(manifest["method_evidence_groups"]) diff --git a/crates/graphforge-cli/src/portable_cli.rs b/crates/graphforge-cli/src/portable_cli.rs index 453888298..36c3fdb5b 100644 --- a/crates/graphforge-cli/src/portable_cli.rs +++ b/crates/graphforge-cli/src/portable_cli.rs @@ -539,6 +539,8 @@ pub(crate) enum ImportSessionCommand { Begin(ImportSessionBeginArgs), /// Resume an existing session by UUID. Resume(ImportSessionResumeArgs), + /// Read durable progress, including a terminal construction receipt. + Status(ImportSessionIdArgs), /// Register a Parquet source path into the session. RegisterParquet(ImportSessionRegisterArgs), /// Checkpoint session progress. @@ -611,6 +613,17 @@ pub(crate) fn run_import_session( let session = graph.resume_import_session(canonical_uuid(&args.session_uuid)?)?; write_session_receipt(session.session_uuid(), "resumed", json, output) } + ImportSessionCommand::Status(args) => { + let session_uuid = canonical_uuid(&args.session_uuid)?; + let (phase, progress) = graph.import_session_status(session_uuid)?; + write_progress( + session_uuid, + &format!("{phase:?}").to_ascii_lowercase(), + &progress, + json, + output, + ) + } ImportSessionCommand::RegisterParquet(args) => { let mut session = graph.resume_import_session(canonical_uuid(&args.session_uuid)?)?; let kind = match args.kind { @@ -639,6 +652,7 @@ pub(crate) fn run_import_session( ImportSessionCommand::Commit(args) => { let mut session = graph.resume_import_session(canonical_uuid(&args.session_uuid)?)?; let generation = session.commit(graph, None)?; + let (_, progress) = session.status(); if json { write_json( &serde_json::json!({ @@ -646,6 +660,7 @@ pub(crate) fn run_import_session( "outcome": "committed", "session_uuid": session.session_uuid(), "generation_uuid": generation, + "construction": progress.construction, }), output, ) @@ -661,7 +676,7 @@ pub(crate) fn run_import_session( ImportSessionCommand::Abort(args) => { let session = graph.resume_import_session(canonical_uuid(&args.session_uuid)?)?; let session_uuid = session.session_uuid(); - let progress = session.abort()?; + let progress = session.abort(graph)?; write_progress(session_uuid, "aborted", &progress, json, output) } ImportSessionCommand::Cleanup(args) => { @@ -720,6 +735,7 @@ fn write_progress( "rows_accepted": progress.rows_accepted, "rows_rejected": progress.rows_rejected, "bytes_accepted": progress.bytes_accepted, + "construction": progress.construction, }), output, ) diff --git a/crates/graphforge-storage/src/graph_construction.rs b/crates/graphforge-storage/src/graph_construction.rs index 3b62851d7..b4d933ba8 100644 --- a/crates/graphforge-storage/src/graph_construction.rs +++ b/crates/graphforge-storage/src/graph_construction.rs @@ -2705,6 +2705,51 @@ impl GraphConstructionSession { replace_checkpoint_control(&self.root, &self.checkpoint) } + /// Authentically reclaim an unpublished session and every file it owns. + /// + /// Published or publishing sessions belong to generation recovery and can + /// never be discarded through the private staging lifecycle. + pub fn discard(mut self) -> Result<(), GfError> { + self.revalidate_authority()?; + self.recover_intent()?; + if matches!( + self.checkpoint.publication_state, + Some( + ConstructionPublicationState::Publishing | ConstructionPublicationState::Published + ) + ) { + return Err(storage( + "published construction belongs to generation recovery", + )); + } + self.checkpoint.state = GraphConstructionState::Aborted; + replace_checkpoint_control(&self.root, &self.checkpoint)?; + + let private = self + .project + .open_child_directory(OsStr::new(PRIVATE_ROOT)) + .map_err(storage)?; + let operation_name = self.checkpoint.operation_uuid.simple().to_string(); + let session_identity = self.root.identity(); + let mut remaining = self + .checkpoint + .budgets + .max_chunks + .saturating_mul(16) + .saturating_add( + u64::try_from(self.checkpoint.budgets.max_schema_groups).unwrap_or(u64::MAX), + ) + .saturating_add(4_096); + crate::file_lock::unlock(&self.session_lock).map_err(storage)?; + drop(self); + remove_owned_directory_tree( + &private, + OsStr::new(&operation_name), + session_identity, + &mut remaining, + ) + } + #[allow(clippy::too_many_lines)] fn recover_intent(&mut self) -> Result<(), GfError> { let mut file = match self.root.open_child_file(OsStr::new(INTENT)) { @@ -2973,6 +3018,61 @@ impl GraphConstructionSession { } } +fn remove_owned_directory_tree( + parent: &StableDirectory, + name: &OsStr, + expected: FileIdentity, + remaining: &mut u64, +) -> Result<(), GfError> { + if *remaining == 0 { + return Err(storage( + "construction discard exceeded authenticated entry bound", + )); + } + *remaining -= 1; + let directory = parent.open_child_directory(name).map_err(storage)?; + if directory.identity() != expected { + return Err(storage("construction discard directory identity changed")); + } + let child_limit = usize::try_from(*remaining).unwrap_or(usize::MAX); + let names = directory + .child_names_bounded(child_limit) + .map_err(storage)?; + for child_name in names { + if *remaining == 0 { + return Err(storage( + "construction discard exceeded authenticated entry bound", + )); + } + *remaining -= 1; + match directory.open_child_file(&child_name) { + Ok(file) => { + let identity = file_identity(&file).map_err(storage)?; + drop(file); + directory + .unlink_child_if_identity(&child_name, identity) + .map_err(storage)?; + } + Err(file_error) => { + let child = directory + .open_child_directory(&child_name) + .map_err(|directory_error| { + storage(format!( + "construction discard child is neither an authenticated file nor directory: file={file_error}; directory={directory_error}" + )) + })?; + let identity = child.identity(); + drop(child); + remove_owned_directory_tree(&directory, &child_name, identity, remaining)?; + } + } + } + drop(directory); + parent + .remove_child_directory_if_identity(name, expected) + .map_err(storage) +} + #[derive(Clone, Debug, Serialize, Deserialize)] struct ReceiptPointer { operation_uuid: Uuid, @@ -10262,6 +10362,33 @@ mod tests { session } + fn construction_session_root(root: &TempDir, operation: Uuid) -> PathBuf { + root.path() + .join(PRIVATE_ROOT) + .join(operation.simple().to_string()) + } + + #[test] + fn discard_reclaims_open_and_sealed_session_trees() { + for (index, seal) in [false, true].into_iter().enumerate() { + let root = TempDir::new().unwrap(); + let operation = Uuid::from_u128(9_350 + index as u128); + let mut session = open(&root, operation.as_u128()); + session + .append(ConstructionChunkKind::Node, "nodes", &node_batch(1, 2)) + .unwrap(); + if seal { + session.seal().unwrap(); + } + let private_root = construction_session_root(&root, operation); + assert!(private_root.exists()); + + session.discard().unwrap(); + + assert!(!private_root.exists()); + } + } + fn publish_empty_generation( root: &TempDir, target: Uuid, @@ -10329,6 +10456,14 @@ mod tests { .to_string() .contains("result changed") ); + let private_root = construction_session_root(&root, operation); + let error = session.discard().unwrap_err(); + assert!( + error + .to_string() + .contains("published construction belongs to generation recovery") + ); + assert!(private_root.exists()); } #[test] diff --git a/scripts/ci/test-non-cypher-surface-gate.py b/scripts/ci/test-non-cypher-surface-gate.py index 9c3fb8062..d1155b813 100644 --- a/scripts/ci/test-non-cypher-surface-gate.py +++ b/scripts/ci/test-non-cypher-surface-gate.py @@ -29,7 +29,7 @@ def validate(self, manifest: dict) -> list[str]: def test_checked_in_inventory_is_complete(self) -> None: self.assertEqual(GATE.validate(), []) - self.assertEqual(len(GATE.public_methods()), 373) + self.assertEqual(len(GATE.public_methods()), 374) self.assertEqual(len(GATE.algorithm_registry()), 94) def test_new_or_removed_public_method_fails_frozen_digest(self) -> None: diff --git a/tests/contracts/non-cypher-rust-surface.json b/tests/contracts/non-cypher-rust-surface.json index 5aa8a51f0..8dc8ee04e 100644 --- a/tests/contracts/non-cypher-rust-surface.json +++ b/tests/contracts/non-cypher-rust-surface.json @@ -1,7 +1,7 @@ { "contract_version": 1, "scope": "Rust non-Cypher public release surface", - "public_method_digest": "8e3a0711619a5e50231bf510a76328ea44b7706dfd28c524b760c6564b805bc3", + "public_method_digest": "499f902e4713931d7b3591fbaf978e6694c3029fde08c5c5476547563e5776e5", "method_policy": { "receiver_defaults": { "GraphForge": "release-tested", @@ -880,6 +880,7 @@ "ids": [ "GraphForge.begin_import_session", "GraphForge.cleanup_stale_import_sessions", + "GraphForge.import_session_status", "GraphForge.resume_import_session", "GraphImportSession.abort", "GraphImportSession.append_arrow", @@ -899,6 +900,10 @@ "path": "crates/graphforge-api/src/import_session.rs", "symbol": "parquet_abort_and_missing_endpoint_preserve_prior_generation" }, + { + "path": "crates/graphforge-api/src/import_session.rs", + "symbol": "parquet_construction_receipts_scale_linearly_and_survive_reopen" + }, { "path": "crates/graphforge-api/src/import_session.rs", "symbol": "cancellation_and_missing_endpoint_are_durable_fail_closed"