diff --git a/crates/graphforge-api/src/resumable_construction.rs b/crates/graphforge-api/src/resumable_construction.rs index c194461e..2f2fe9ee 100644 --- a/crates/graphforge-api/src/resumable_construction.rs +++ b/crates/graphforge-api/src/resumable_construction.rs @@ -1,5 +1,8 @@ //! Public Rust facade for resumable, bounded, disk-owned graph construction. +#[cfg(test)] +mod codec_tests; + use arrow::record_batch::RecordBatch; use graphforge_core::uuid::Uuid; use sha2::{Digest, Sha256}; @@ -777,7 +780,10 @@ mod tests { #[test] fn construction_application_reads_reconcile_and_scale_at_one_two_four() { let mut observations = Vec::new(); - for scale in [1_024_usize, 2_048, 4_096] { + // Each node retains 16 identity bytes and at least 18 compact detail bytes. + // 4,096 rows therefore exceed 100,000 payload bytes before Parquet/control + // overhead; retain the same dominance threshold and every phase ceiling. + for scale in [4_096_usize, 8_192, 16_384] { let graph = GraphForge::new(None).unwrap(); let mut session = graph.begin_graph_construction(Default::default()).unwrap(); let ids = (0..scale).map(|_| Uuid::now_v7()).collect::>(); diff --git a/crates/graphforge-api/src/resumable_construction/codec_tests.rs b/crates/graphforge-api/src/resumable_construction/codec_tests.rs new file mode 100644 index 00000000..883b47a2 --- /dev/null +++ b/crates/graphforge-api/src/resumable_construction/codec_tests.rs @@ -0,0 +1,373 @@ +//! Public construction and interchange preserve compact-detail graph identity. + +use std::sync::Arc; + +use arrow::array::{Array, FixedSizeBinaryArray, Int64Array, ListArray, StringArray}; +use arrow::datatypes::{DataType, Field, Schema}; +use arrow::record_batch::RecordBatch; +use graphforge_core::portable::{ + PortableV2Limits, PortableV2Mode, PortableV2Output, PortableV2SelectionProfile, +}; +use graphforge_storage::UuidMembershipIndex; +use uuid::Uuid; + +use crate::{ + CONSTRUCTION_EDGE_SCHEMA, CONSTRUCTION_NODE_SCHEMA, GraphConstructionBudgets, GraphForge, + OperationId, PortableSelection, PortableV2ExportRequest, PortableV2ImportRequest, + PortableVerifyRequest, verify_portable_v2, +}; + +type NodeRow = (Uuid, Vec); +type EdgeRow = (Uuid, String, Uuid, Uuid); + +fn uuid_array(ids: &[Uuid]) -> FixedSizeBinaryArray { + FixedSizeBinaryArray::try_from_iter(ids.iter().map(Uuid::as_bytes)).unwrap() +} + +fn uuid_at(batch: &RecordBatch, column: usize, row: usize) -> Uuid { + assert_eq!(batch.column(column).null_count(), 0); + Uuid::from_slice( + batch + .column(column) + .as_any() + .downcast_ref::() + .unwrap() + .value(row), + ) + .unwrap() +} + +fn graph_rows(graph: &GraphForge) -> (Vec, Vec) { + let nodes = graph + .execute("MATCH (n) RETURN n.node_uuid, labels(n) ORDER BY n.node_uuid") + .unwrap(); + let node_rows = nodes + .batches + .iter() + .flat_map(|batch| { + let labels = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(labels.null_count(), 0); + (0..batch.num_rows()) + .map(|row| { + let values = labels.value(row); + let names = values.as_any().downcast_ref::().unwrap(); + assert_eq!(names.null_count(), 0); + ( + uuid_at(batch, 0, row), + names.iter().map(|name| name.unwrap().to_owned()).collect(), + ) + }) + .collect::>() + }) + .collect(); + let edges = graph.execute("MATCH (a)-[r]->(b) RETURN r.edge_uuid, type(r), a.node_uuid, b.node_uuid ORDER BY r.edge_uuid").unwrap(); + let edge_rows = edges + .batches + .iter() + .flat_map(|batch| { + let routes = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(routes.null_count(), 0); + (0..batch.num_rows()) + .map(|row| { + ( + uuid_at(batch, 0, row), + routes.value(row).to_owned(), + uuid_at(batch, 2, row), + uuid_at(batch, 3, row), + ) + }) + .collect::>() + }) + .collect(); + (node_rows, edge_rows) +} + +fn ordinals(graph: &GraphForge, ids: &[Uuid]) -> Vec> { + UuidMembershipIndex::open(&graph.dir) + .unwrap() + .lookup_node_surrogates(ids) + .unwrap() + .0 +} + +fn assert_properties(graph: &GraphForge, nodes: &[Uuid], edges: &[Uuid]) { + for (query, expected) in [ + ( + "MATCH (n) RETURN n.node_uuid, n.score ORDER BY n.node_uuid", + nodes + .iter() + .enumerate() + .map(|(i, &id)| { + ( + id, + if i == 1 { + None + } else { + Some(41 + i64::try_from(i).unwrap()) + }, + ) + }) + .collect::>(), + ), + ( + "MATCH ()-[r]->() RETURN r.edge_uuid, r.weight ORDER BY r.edge_uuid", + edges + .iter() + .enumerate() + .map(|(i, &id)| { + ( + id, + if i == 1 { + None + } else { + Some(73 + i64::try_from(i).unwrap()) + }, + ) + }) + .collect::>(), + ), + ] { + let result = graph.execute(query).unwrap(); + let actual = result + .batches + .iter() + .flat_map(|batch| { + let values = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + values + .iter() + .enumerate() + .map(|(row, value)| (uuid_at(batch, 0, row), value)) + .collect::>() + }) + .collect::>(); + assert_eq!(actual, expected); + } +} + +fn property_schema(base: &Schema, name: &str) -> Arc { + let mut fields = base.fields().to_vec(); + fields.push(Arc::new(Field::new(name, DataType::Int64, true))); + Arc::new(Schema::new(fields)) +} + +#[test] +fn compact_public_construction_reopens_and_round_trips_exact_graph() { + let root = tempfile::tempdir().unwrap(); + let source = root.path().join("source"); + let graph = GraphForge::new(source.to_str()).unwrap(); + let ids = [Uuid::from_u128(1), Uuid::from_u128(2), Uuid::from_u128(3)]; + let edge_ids = [ + Uuid::from_u128(11), + Uuid::from_u128(12), + Uuid::from_u128(13), + Uuid::from_u128(14), + ]; + let budgets = GraphConstructionBudgets { + max_batch_rows: 2, + max_run_records: 8, + ..Default::default() + }; + let mut session = graph.begin_graph_construction(budgets).unwrap(); + let session_id = session.session_uuid(); + let node_chunk = |indexes: &[usize], names: Vec<&str>| { + RecordBatch::try_new( + property_schema(&CONSTRUCTION_NODE_SCHEMA, "score"), + vec![ + Arc::new(uuid_array( + &indexes.iter().map(|&i| ids[i]).collect::>(), + )), + Arc::new(StringArray::from(names)), + Arc::new(Int64Array::from( + indexes + .iter() + .map(|&i| { + if i == 1 { + None + } else { + Some(41 + i64::try_from(i).unwrap()) + } + }) + .collect::>(), + )), + ], + ) + .unwrap() + }; + session + .append_nodes("first", &node_chunk(&[2, 0], vec!["Équipe", "Person"])) + .unwrap(); + drop(session); + let checkpoint = source + .join(".graphforge-construction") + .join(session_id.simple().to_string()) + .join("checkpoint.json"); + let control: serde_json::Value = + serde_json::from_slice(&std::fs::read(&checkpoint).unwrap()).unwrap(); + assert_eq!( + control["format_version"], 7, + "public new sessions use compact details" + ); + let mut session = graph + .resume_graph_construction(session_id, budgets) + .unwrap(); + session + .append_nodes("second", &node_chunk(&[1], vec!["Person"])) + .unwrap(); + let endpoints = [(0, 1), (0, 1), (2, 2), (1, 2)]; + for (chunk, indexes) in [[0, 2], [1, 3]].iter().enumerate() { + let batch = RecordBatch::try_new( + property_schema(&CONSTRUCTION_EDGE_SCHEMA, "weight"), + vec![ + Arc::new(uuid_array( + &indexes.iter().map(|&i| edge_ids[i]).collect::>(), + )), + Arc::new(StringArray::from( + indexes + .iter() + .map(|&i| if i < 2 { "KNOWS" } else { "LIÉ" }) + .collect::>(), + )), + Arc::new(uuid_array( + &indexes + .iter() + .map(|&i| ids[endpoints[i].0]) + .collect::>(), + )), + Arc::new(uuid_array( + &indexes + .iter() + .map(|&i| ids[endpoints[i].1]) + .collect::>(), + )), + Arc::new(Int64Array::from( + indexes + .iter() + .map(|&i| { + if i == 1 { + None + } else { + Some(73 + i64::try_from(i).unwrap()) + } + }) + .collect::>(), + )), + ], + ) + .unwrap(); + session + .append_edges(&format!("edges-{chunk}"), &batch) + .unwrap(); + } + let published = session.seal_and_publish().unwrap(); + assert!(!published.idempotent_replay); + drop(session); + let mut replay = graph + .resume_graph_construction(session_id, budgets) + .unwrap(); + let receipt = replay.seal_and_publish().unwrap(); + assert!(receipt.idempotent_replay); + assert_eq!(receipt.generation_uuid, published.generation_uuid); + assert_eq!( + graphforge_storage::resolve_project_generation(&source) + .unwrap() + .generation_uuid(), + published.generation_uuid + ); + drop(replay); + drop(graph); + + let graph = GraphForge::new(source.to_str()).unwrap(); + let expected_nodes = vec![ + (ids[0], vec!["Person".into()]), + (ids[1], vec!["Person".into()]), + (ids[2], vec!["Équipe".into()]), + ]; + let expected_edges = edge_ids + .iter() + .enumerate() + .map(|(i, &id)| { + ( + id, + if i < 2 { "KNOWS" } else { "LIÉ" }.into(), + ids[endpoints[i].0], + ids[endpoints[i].1], + ) + }) + .collect::>(); + assert_eq!( + graph_rows(&graph), + (expected_nodes.clone(), expected_edges.clone()) + ); + let expected_ordinals = vec![Some(1), Some(2), Some(3)]; + assert_eq!(ordinals(&graph, &ids), expected_ordinals); + assert_properties(&graph, &ids, &edge_ids); + drop(graph); + let graph = GraphForge::new(source.to_str()).unwrap(); + assert_properties(&graph, &ids, &edge_ids); + assert_eq!( + graph_rows(&graph), + (expected_nodes.clone(), expected_edges.clone()) + ); + assert_eq!(ordinals(&graph, &ids), expected_ordinals); + let limits = PortableV2Limits::default(); + let package = root.path().join("graph.gfpb"); + let exported = graph + .export_portable_v2( + &PortableV2ExportRequest { + selection: PortableSelection::Current, + output_path: package.clone(), + representation: PortableV2Output::Bundle, + profile: PortableV2SelectionProfile::Complete, + subset: None, + limits, + }, + None, + |_| {}, + ) + .unwrap(); + let verified = verify_portable_v2( + &PortableVerifyRequest { + input: package.clone(), + mode: PortableV2Mode::Full, + limits, + }, + None, + ) + .unwrap(); + assert_eq!(verified.package_digest, exported.package_digest); + drop(graph); + let target = root.path().join("imported"); + assert!(!target.exists()); + let imported = GraphForge::import_portable_v2( + &target, + &PortableV2ImportRequest { + input: package, + operation_id: OperationId(Uuid::from_u128(100)), + limits, + }, + None, + ) + .unwrap(); + assert!(!imported.idempotent_replay); + let graph = GraphForge::new(target.to_str()).unwrap(); + assert_eq!( + graph.resolved_generation.generation_uuid(), + imported.generation_uuid + ); + assert_eq!(imported.package_digest, exported.package_digest); + assert_eq!(graph_rows(&graph), (expected_nodes, expected_edges)); + assert_eq!(ordinals(&graph, &ids), expected_ordinals); + assert_properties(&graph, &ids, &edge_ids); +} diff --git a/crates/graphforge-api/tests/scale_g500_ladder.rs b/crates/graphforge-api/tests/scale_g500_ladder.rs index eb0b943c..246fc303 100644 --- a/crates/graphforge-api/tests/scale_g500_ladder.rs +++ b/crates/graphforge-api/tests/scale_g500_ladder.rs @@ -7628,17 +7628,12 @@ fn tiny_construction_ladder_resumes_and_scales_bounded_work_linearly() { progress.evidence.input_batches ); assert_eq!(progress.evidence.immutable_artifacts, 7 * factor + 4); - // Each full node-detail run (272 * 65,536 bytes = 17 MiB) and - // edge-detail run (304 * 65,536 bytes = 19 MiB) crosses the 16 MiB - // per-stream cache window once. The final edge chunk is two rows - // short and still crosses once; the separately acknowledged one-edge - // chunk does not. These synchronized rollovers add two real barriers - // per factor to the original artifact publication protocol on Linux. - let rollover_fsyncs = u64::from(cfg!(target_os = "linux")) * 2 * factor; - assert_eq!( - progress.evidence.fsync_operations, - 23 * factor + 13 + rollover_fsyncs - ); + // Compact Node details use 16 + 1 + 4 = 21 bytes per row, and + // LINK details use 48 + 1 + 4 = 53. At 65,536 rows their streams + // remain below the 16 MiB cache window, as do the fixed identity + // and endpoint streams. No cache-window rollover adds a barrier + // to the artifact publication protocol. + assert_eq!(progress.evidence.fsync_operations, 23 * factor + 13); assert!(progress.evidence.peak_batch_rows <= CONSTRUCTION_BATCH_ROWS as u64); assert!(progress.evidence.peak_accounted_live_bytes <= 64 * 1024 * 1024); assert!(progress.evidence.peak_run_records <= budgets.max_run_records as u64); @@ -7675,11 +7670,11 @@ fn tiny_construction_ladder_resumes_and_scales_bounded_work_linearly() { // One bounded edge merge window may overlap its immutable // identity, endpoint, and detail inputs with the unified // identity output. These are the construction format's - // fixed record widths, so this is a derived byte bound, + // wire widths for this fixture, so this is a derived bound, // not general-purpose disk slack. const IDENTITY_RECORD_BYTES: u64 = 16; const ENDPOINT_RECORD_BYTES: u64 = 48; - const EDGE_DETAIL_RECORD_BYTES: u64 = 304; + const EDGE_DETAIL_RECORD_BYTES: u64 = 48 + 1 + REL_TYPE.len() as u64; const UNIFIED_IDENTITY_RECORD_BYTES: u64 = 32; let fixed_edge_merge_window_bytes = CONSTRUCTION_BATCH_ROWS as u64 * (IDENTITY_RECORD_BYTES diff --git a/crates/graphforge-storage/src/construction_detail_codec.rs b/crates/graphforge-storage/src/construction_detail_codec.rs new file mode 100644 index 00000000..27c53f4f --- /dev/null +++ b/crates/graphforge-storage/src/construction_detail_codec.rs @@ -0,0 +1,281 @@ +//! Bounded private detail records. Version six retains its padded wire layout. +use std::io::{self, Read}; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum DetailCodec { + Legacy, + Compact, +} + +impl DetailCodec { + pub(crate) fn from_version(version: u32) -> io::Result { + match version { + 6 => Ok(Self::Legacy), + 7 => Ok(Self::Compact), + _ => Err(invalid("unsupported construction detail version")), + } + } + + pub(crate) fn validate_size(self, width: usize, rows: u64, bytes: u64) -> io::Result<()> { + if !matches!(width, 272 | 304) { + return Err(invalid("invalid construction detail record domain")); + } + let minimum_width = match self { + Self::Legacy => width, + Self::Compact => width - 254, + }; + let minimum = rows + .checked_mul(minimum_width as u64) + .ok_or_else(|| invalid("detail byte bound overflow"))?; + let maximum = rows + .checked_mul(width as u64) + .ok_or_else(|| invalid("detail byte bound overflow"))?; + if bytes < minimum || bytes > maximum { + return Err(invalid("detail bytes disagree with bounded row count")); + } + Ok(()) + } + + pub(crate) fn bytes(self, record: &[u8; N]) -> io::Result<&[u8]> { + let prefix = prefix::()?; + let length = usize::from(record[prefix]); + if length == 0 || record[prefix + 1 + length..].iter().any(|byte| *byte != 0) { + return Err(invalid("invalid construction detail name or padding")); + } + std::str::from_utf8(&record[prefix + 1..prefix + 1 + length]) + .map_err(|_| invalid("construction detail name is not UTF-8"))?; + Ok(match self { + Self::Legacy => record, + Self::Compact => &record[..prefix + 1 + length], + }) + } + + /// Return a bounded padded in-memory record, preserving existing consumers. + pub(crate) fn read( + self, + reader: &mut impl Read, + ) -> io::Result> { + let prefix = prefix::()?; + let mut record = [0; N]; + loop { + match reader.read(&mut record[..1]) { + Ok(0) => return Ok(None), + Ok(_) => break, + Err(error) if error.kind() == io::ErrorKind::Interrupted => {} + Err(error) => return Err(error), + } + } + reader.read_exact(&mut record[1..=prefix])?; + let length = usize::from(record[prefix]); + if length == 0 { + return Err(invalid("empty construction detail name")); + } + let end = match self { + Self::Legacy => N, + Self::Compact => prefix + 1 + length, + }; + reader.read_exact(&mut record[prefix + 1..end])?; + self.bytes(&record)?; + Ok(Some(record)) + } +} + +fn prefix() -> io::Result { + match N { + 272 => Ok(16), + 304 => Ok(48), + _ => Err(invalid("invalid construction detail record domain")), + } +} + +fn invalid(message: &'static str) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, message) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn roundtrip(name: &str) { + let prefix = prefix::().unwrap(); + let mut record = [0; N]; + for (index, byte) in record[..prefix].iter_mut().enumerate() { + *byte = u8::try_from(index + 1).unwrap(); + } + record[prefix] = u8::try_from(name.len()).unwrap(); + record[prefix + 1..prefix + 1 + name.len()].copy_from_slice(name.as_bytes()); + let mut golden = record[..prefix + 1].to_vec(); + golden.extend_from_slice(name.as_bytes()); + assert_eq!(DetailCodec::Compact.bytes(&record).unwrap(), golden); + assert_eq!(DetailCodec::Legacy.bytes(&record).unwrap(), record); + for codec in [DetailCodec::Legacy, DetailCodec::Compact] { + let bytes = codec.bytes(&record).unwrap(); + assert_eq!(codec.read::(&mut &*bytes).unwrap(), Some(record)); + for length in 1..bytes.len() { + assert!(codec.read::(&mut &bytes[..length]).is_err()); + } + assert_eq!(codec.read::(&mut &[][..]).unwrap(), None); + } + } + + #[test] + fn detail_codec_golden_roundtrips_and_truncation() { + for name in [ + "a".to_owned(), + "EDGE".to_owned(), + "x".repeat(255), + format!("{}a", "é".repeat(127)), + ] { + roundtrip::<272>(&name); + roundtrip::<304>(&name); + } + } + + #[test] + fn detail_codec_streaming_partitions_and_invalid_order() { + for width in [272, 304] { + for codec in [DetailCodec::Legacy, DetailCodec::Compact] { + let prefix = width - 256; + let mut records = Vec::new(); + for id in 1..=3_u8 { + let mut record = vec![0_u8; width]; + record[15] = id; + record[prefix] = 4; + record[prefix + 1..prefix + 5].copy_from_slice("éé".as_bytes()); + if codec == DetailCodec::Compact { + record.truncate(prefix + 5); + } + records.push(record); + } + let wire = records.concat(); + for split in 0..=wire.len() { + let mut validator = DetailValidator::new(codec, width).unwrap(); + validator.consume(&wire[..split]).unwrap(); + validator.consume(&wire[split..]).unwrap(); + assert_eq!(validator.finish().unwrap(), 3); + } + let mut validator = DetailValidator::new(codec, width).unwrap(); + for byte in &wire { + validator.consume(std::slice::from_ref(byte)).unwrap(); + } + assert_eq!(validator.finish().unwrap(), 3); + let mut invalid_utf8 = wire.clone(); + invalid_utf8[prefix + 1] = 255; + let cases = [ + [records[0].clone(), records[0].clone()].concat(), + [records[1].clone(), records[0].clone()].concat(), + invalid_utf8, + wire[..wire.len() - 1].to_vec(), + ]; + for malformed in cases { + for split in 0..=malformed.len() { + let mut validator = DetailValidator::new(codec, width).unwrap(); + let result = validator + .consume(&malformed[..split]) + .and_then(|()| validator.consume(&malformed[split..])) + .and_then(|()| validator.finish()); + assert!( + result.is_err(), + "codec={codec:?} width={width} split={split}" + ); + } + } + } + } + } + + #[test] + fn detail_codec_rejects_invalid_names_padding_and_versions() { + let mut record = [0; 304]; + assert!(DetailCodec::Compact.bytes(&record).is_err()); + record[48] = 1; + record[49] = 255; + assert!(DetailCodec::Compact.bytes(&record).is_err()); + record[49] = b'a'; + record[303] = 1; + assert!(DetailCodec::Legacy.bytes(&record).is_err()); + assert!(DetailCodec::Compact.bytes(&record).is_err()); + assert_eq!(DetailCodec::from_version(6).unwrap(), DetailCodec::Legacy); + assert_eq!(DetailCodec::from_version(7).unwrap(), DetailCodec::Compact); + assert!(DetailCodec::from_version(5).is_err()); + assert!(DetailCodec::from_version(8).is_err()); + } +} + +/// Incremental validation retains one bounded record and its preceding UUID. +/// Input block sizes do not change parsing or memory bounds. +pub(crate) struct DetailValidator { + codec: DetailCodec, + width: usize, + record: [u8; 304], + filled: usize, + previous: Option<[u8; 16]>, + records: u64, +} + +impl DetailValidator { + pub(crate) fn new(codec: DetailCodec, width: usize) -> io::Result { + if !matches!(width, 272 | 304) { + return Err(invalid("invalid construction detail record domain")); + } + Ok(Self { + codec, + width, + record: [0; 304], + filled: 0, + previous: None, + records: 0, + }) + } + + pub(crate) fn consume(&mut self, mut bytes: &[u8]) -> io::Result<()> { + let prefix = self.width - 256; + while !bytes.is_empty() { + let target = if self.filled <= prefix { + prefix + 1 + } else { + let length = usize::from(self.record[prefix]); + if length == 0 { + return Err(invalid("empty construction detail name")); + } + match self.codec { + DetailCodec::Legacy => self.width, + DetailCodec::Compact => prefix + 1 + length, + } + }; + let count = (target - self.filled).min(bytes.len()); + self.record[self.filled..self.filled + count].copy_from_slice(&bytes[..count]); + self.filled += count; + bytes = &bytes[count..]; + if self.filled == target && target > prefix + 1 { + if self.width == 272 { + let record: &[u8; 272] = self.record[..272].try_into().expect("fixed domain"); + self.codec.bytes(record)?; + } else { + self.codec.bytes(&self.record)?; + } + let uuid: [u8; 16] = self.record[..16].try_into().expect("UUID prefix"); + if self.previous.is_some_and(|previous| previous >= uuid) { + return Err(invalid( + "construction detail UUIDs are not strictly ordered", + )); + } + self.previous = Some(uuid); + self.records = self + .records + .checked_add(1) + .ok_or_else(|| invalid("construction detail row count overflow"))?; + self.record.fill(0); + self.filled = 0; + } + } + Ok(()) + } + + pub(crate) fn finish(&self) -> io::Result { + if self.filled != 0 { + return Err(invalid("truncated construction detail record")); + } + Ok(self.records) + } +} diff --git a/crates/graphforge-storage/src/construction_detail_tests.rs b/crates/graphforge-storage/src/construction_detail_tests.rs new file mode 100644 index 00000000..fa2ca4e4 --- /dev/null +++ b/crates/graphforge-storage/src/construction_detail_tests.rs @@ -0,0 +1,545 @@ +// Included within the construction tests to reuse the real writer fixtures. +mod compact_details { + use super::*; + + fn create( + root: &TempDir, + operation: Uuid, + version: u32, + budgets: GraphConstructionBudgets, + allocation: Option<&crate::StorageAllocationOperation>, + ) -> GraphConstructionSession { + GraphConstructionSession::open_internal_with_format( + root.path(), + root.path(), + operation, + 0, + graphforge_core::OntologyMode::Exploratory, + None, + budgets, + crate::filesystem_admission::ProjectLifecycleMode::Durable, + allocation, + version, + ) + .unwrap() + } + + fn raw_inventory(paths: &[PathBuf]) -> (BTreeMap, u64) { + let mut pending = paths.to_vec(); + let mut identities = BTreeMap::new(); + let mut references = 0_u64; + let mut visited = 0; + while let Some(path) = pending.pop() { + visited += 1; + assert!(visited < 100_000); + let metadata = std::fs::symlink_metadata(&path).unwrap(); + assert!(!metadata.file_type().is_symlink()); + if metadata.is_dir() { + pending.extend( + std::fs::read_dir(path) + .unwrap() + .map(|entry| entry.unwrap().path()), + ); + } else { + assert!(metadata.is_file()); + let file = File::open(&path).unwrap(); + let identity = file_identity(&file).unwrap(); + let key = crate::storage_attribution::native_identity_key( + identity.volume_serial, + &identity.file_id, + ); + let allocated = graphforge_filesystem::file_space_usage(&file) + .unwrap() + .allocated_bytes; + if let Some(previous) = identities.insert(key, allocated) { + assert_eq!(previous, allocated); + } + references += 1; + } + } + (identities, references) + } + + #[test] + fn detail_codec_legacy_and_compact_resume_cross_multiple_merge_levels() { + let mut measurements = Vec::new(); + for version in [6, 7] { + let root = TempDir::new().unwrap(); + crate::open_or_initialize_project(root.path()).unwrap(); + let operation = Uuid::new_v4(); + let budgets = GraphConstructionBudgets { + merge_fan_in: 2, + max_batch_rows: 128, + max_run_records: 512, + ..GraphConstructionBudgets::default() + }; + // Match first-party pre-operation setup: persistent lock files are + // real baseline owners even though they allocate zero data blocks. + drop(crate::begin_graph_object_publication(root.path()).unwrap()); + drop(crate::project_publication::wait_for_writer_lock(root.path()).unwrap()); + let paths = crate::StorageAllocationOperation::project_paths(root.path()).unwrap(); + let allocation = crate::StorageAllocationOperation::from_paths(&paths).unwrap(); + let mut session = create(&root, operation, version, budgets, Some(&allocation)); + // Genuine legacy output: selected before the initial checkpoint exists. + for chunk in 0..8 { + session + .append( + ConstructionChunkKind::Node, + &format!("nodes-{chunk}"), + &node_batch(1 + chunk * 128, 128), + ) + .unwrap(); + } + drop(session); + let mut session = GraphConstructionSession::open_with_allocation( + root.path(), + root.path(), + operation, + 0, + graphforge_core::OntologyMode::Exploratory, + None, + budgets, + crate::filesystem_admission::ProjectLifecycleMode::Durable, + true, + &allocation, + ) + .unwrap(); + assert_eq!(session.checkpoint.format_version, version); + for chunk in 0..8 { + session + .append( + ConstructionChunkKind::Edge, + &format!("edges-{chunk}"), + &edge_batch(10_000 + chunk * 128, 128), + ) + .unwrap(); + } + let mut detail_bytes = 0; + let mut detail_allocated = 0; + for sequence in 0..16 { + let mut file = session + .root + .open_child_file(OsStr::new(&receipt_name(sequence))) + .unwrap(); + let receipt: ConstructionChunkReceipt = decode_bounded(&mut file).unwrap(); + let expected_width = match (version, receipt.kind) { + (6, ConstructionChunkKind::Node) => NODE_DETAIL_WIDTH, + (6, ConstructionChunkKind::Edge) => EDGE_DETAIL_WIDTH, + (7, ConstructionChunkKind::Node) => 16 + 1 + "Person".len(), + (7, ConstructionChunkKind::Edge) => 48 + 1 + "R".len(), + _ => unreachable!(), + }; + assert_eq!(receipt.details.bytes, receipt.rows * expected_width as u64); + let physical = session + .root + .open_child_file(OsStr::new(&receipt.details.name)) + .unwrap(); + let usage = graphforge_filesystem::file_space_usage(&physical).unwrap(); + assert_eq!(physical.metadata().unwrap().len(), receipt.details.bytes); + assert_eq!(usage.allocated_bytes, receipt.details.allocated_bytes); + detail_bytes += receipt.details.bytes; + detail_allocated += usage.allocated_bytes; + } + session.seal().unwrap(); + let shape = session.shape_canonical_with_cancellation(|| false).unwrap(); + assert!(session.evidence().merge_passes >= 3); + assert_eq!((shape.node_count, shape.edge_count), (1024, 1024)); + let outputs = [&shape.node_details, &shape.edge_details]; + let mut canonical = Vec::new(); + for (index, name) in outputs.into_iter().enumerate() { + let mut file = session + .root + .open_child_file(OsStr::new(name.as_ref().unwrap())) + .unwrap(); + let codec = DetailCodec::from_version(version).unwrap(); + if index == 0 { + while let Some(record) = codec.read::(&mut file).unwrap() { + canonical.push(record.to_vec()); + } + } else { + while let Some(record) = codec.read::(&mut file).unwrap() { + canonical.push(record.to_vec()); + } + } + } + let peak = session + .evidence() + .storage_transient_peak_total_allocated_bytes; + assert!(peak > detail_allocated); + drop(session); + let mut session = GraphConstructionSession::open_with_allocation( + root.path(), + root.path(), + operation, + 0, + graphforge_core::OntologyMode::Exploratory, + None, + budgets, + crate::filesystem_admission::ProjectLifecycleMode::Durable, + true, + &allocation, + ) + .unwrap(); + assert_eq!(session.checkpoint.format_version, version); + let replay_shape = session.shape_canonical_with_cancellation(|| false).unwrap(); + assert_eq!(replay_shape, shape); + let encoded = session.encode_canonical(&shape, 1).unwrap(); + let publication = session + .publish_canonical(&encoded, Uuid::new_v4(), Uuid::new_v4()) + .unwrap(); + let peak = session + .evidence() + .storage_transient_peak_total_allocated_bytes; + let (raw, references) = raw_inventory(&paths); + assert!( + allocation + .snapshot() + .unwrap() + .matches_file_inventory(&raw, references) + ); + assert_eq!(allocation.totals().unwrap().0, raw.values().sum::()); + drop(session); + let mut replay = GraphConstructionSession::resume_with_mode_and_lifecycle( + root.path(), + operation, + graphforge_core::OntologyMode::Exploratory, + budgets, + crate::filesystem_admission::ProjectLifecycleMode::Durable, + ) + .unwrap(); + assert_eq!(replay.checkpoint.format_version, version); + let repeated = replay + .publish_canonical( + &encoded, + publication.generation_uuid, + publication.transaction_uuid, + ) + .unwrap(); + assert_eq!(repeated.generation_uuid, publication.generation_uuid); + measurements.push((detail_bytes, detail_allocated, peak, canonical)); + } + assert_eq!(measurements[0].3, measurements[1].3); + assert!(measurements[1].0 < measurements[0].0); + assert!(measurements[1].1 < measurements[0].1); + assert!(measurements[1].2 < measurements[0].2); + eprintln!( + "legacy/compact detail EOF, allocation, construction peak: {:?} {:?}", + (measurements[0].0, measurements[0].1, measurements[0].2), + (measurements[1].0, measurements[1].1, measurements[1].2) + ); + } + #[test] + fn detail_codec_partial_control_temp_preserved_and_mixed_version_refused() { + for version in [6, 7] { + let root = TempDir::new().unwrap(); + crate::open_or_initialize_project(root.path()).unwrap(); + let operation = Uuid::new_v4(); + let budgets = GraphConstructionBudgets::default(); + let mut session = create(&root, operation, version, budgets, None); + session + .append(ConstructionChunkKind::Node, "nodes", &node_batch(1, 2)) + .unwrap(); + let private = session.root.path().to_path_buf(); + let partial = private.join(control_temp(CHECKPOINT)); + std::fs::write(&partial, b"{\"format_version\":").unwrap(); + drop(session); + let session = + GraphConstructionSession::open(root.path(), operation, 0, budgets).unwrap(); + assert_eq!(session.checkpoint.format_version, version); + assert_eq!(std::fs::read(&partial).unwrap(), b"{\"format_version\":"); + let mut wrong = session.checkpoint.clone(); + wrong.format_version = if version == 6 { 7 } else { 6 }; + let mixed = private.join(control_temp(CHECKPOINT)); + let bytes = serde_json::to_vec(&wrong).unwrap(); + std::fs::write(&mixed, &bytes).unwrap(); + let current = std::fs::read(root.path().join("CURRENT")).unwrap(); + let checkpoint = std::fs::read(private.join(CHECKPOINT)).unwrap(); + drop(session); + let error = GraphConstructionSession::open(root.path(), operation, 0, budgets) + .err() + .expect("mixed control rejected"); + assert!( + error + .to_string() + .contains("temporary control version differs"), + "{error}" + ); + assert_eq!(std::fs::read(root.path().join("CURRENT")).unwrap(), current); + assert_eq!(std::fs::read(private.join(CHECKPOINT)).unwrap(), checkpoint); + assert_eq!(std::fs::read(&mixed).unwrap(), bytes); + std::fs::remove_file(mixed).unwrap(); + let mut invalid: serde_json::Value = serde_json::from_slice(&checkpoint).unwrap(); + invalid["format_version"] = 99.into(); + let invalid = serde_json::to_vec(&invalid).unwrap(); + std::fs::write(private.join(CHECKPOINT), &invalid).unwrap(); + assert!(GraphConstructionSession::open(root.path(), operation, 0, budgets).is_err()); + assert_eq!(std::fs::read(root.path().join("CURRENT")).unwrap(), current); + assert_eq!(std::fs::read(private.join(CHECKPOINT)).unwrap(), invalid); + } + } + #[test] + fn detail_codec_crash_child() { + let Ok(path) = std::env::var("GF_DETAIL_CODEC_CRASH_ROOT") else { + return; + }; + let version: u32 = std::env::var("GF_DETAIL_CODEC_VERSION") + .unwrap() + .parse() + .unwrap(); + let root = Path::new(&path); + let mut session = GraphConstructionSession::open_internal_with_format( + root, + root, + Uuid::from_u128(117_200), + 0, + graphforge_core::OntologyMode::Exploratory, + None, + GraphConstructionBudgets::default(), + crate::filesystem_admission::ProjectLifecycleMode::Durable, + None, + version, + ) + .unwrap(); + session + .append(ConstructionChunkKind::Node, "nodes", &node_batch(1, 2)) + .unwrap(); + session + .append(ConstructionChunkKind::Edge, "edges", &edge_batch(100, 1)) + .unwrap(); + session.seal().unwrap(); + let encoded = session.prepare_canonical_encoding(1).unwrap(); + session + .publish_canonical(&encoded, Uuid::from_u128(117_201), Uuid::from_u128(117_202)) + .unwrap(); + } + + #[test] + fn detail_codec_both_versions_crash_replay_preserves_authority() { + let boundaries = [ + "control.install.after_temp_fsync.checkpoint.json", + "control.install.after_install.intent.json", + "artifact.after_install.chunk-00000000000000000000-node.node-details.run", + "shape.after_complete_inventory", + "publication.after_current_before_receipt", + ]; + let mut failed = Vec::new(); + for version in [6, 7] { + for boundary in boundaries { + let root = TempDir::new().unwrap(); + crate::open_or_initialize_project(root.path()).unwrap(); + let status = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "graph_construction::tests::compact_details::detail_codec_crash_child", + "--nocapture", + ]) + .env("GF_DETAIL_CODEC_CRASH_ROOT", root.path()) + .env("GF_DETAIL_CODEC_VERSION", version.to_string()) + .env( + "GF_CONSTRUCTION_FAILPOINT_COOKIE", + "graphforge-construction-test-v1", + ) + .env("GF_CONSTRUCTION_FAILPOINT", boundary) + .status() + .unwrap(); + if status.code() != Some(86) { + failed.push(format!("v{version} {boundary}: {status}")); + continue; + } + let mut resumed = if boundary == "control.install.after_temp_fsync.checkpoint.json" + { + // No installed checkpoint exists yet: use the ordinary create-or-reopen entry. + GraphConstructionSession::open( + root.path(), + Uuid::from_u128(117_200), + 0, + GraphConstructionBudgets::default(), + ) + .unwrap() + } else { + GraphConstructionSession::resume_with_mode_and_lifecycle( + root.path(), + Uuid::from_u128(117_200), + graphforge_core::OntologyMode::Exploratory, + GraphConstructionBudgets::default(), + crate::filesystem_admission::ProjectLifecycleMode::Durable, + ) + .unwrap() + }; + assert_eq!(resumed.checkpoint.format_version, version); + if resumed.state() == GraphConstructionState::Staging { + if resumed.accepted_chunks() == 0 { + resumed + .append(ConstructionChunkKind::Node, "nodes", &node_batch(1, 2)) + .unwrap(); + } + if resumed.accepted_chunks() == 1 { + resumed + .append(ConstructionChunkKind::Edge, "edges", &edge_batch(100, 1)) + .unwrap(); + } + resumed.seal().unwrap(); + } + let encoded = resumed.prepare_canonical_encoding(1).unwrap(); + let receipt = resumed + .publish_canonical(&encoded, Uuid::from_u128(117_201), Uuid::from_u128(117_202)) + .unwrap(); + assert_eq!(receipt.generation_uuid, Uuid::from_u128(117_201)); + let repeated = resumed + .publish_canonical(&encoded, Uuid::from_u128(117_201), Uuid::from_u128(117_202)) + .unwrap(); + assert_eq!(repeated.generation_uuid, receipt.generation_uuid); + assert_eq!( + crate::resolve_project_generation(root.path()) + .unwrap() + .generation_uuid(), + receipt.generation_uuid + ); + } + } + assert!(failed.is_empty(), "{failed:?}"); + } + #[test] + fn detail_codec_both_versions_cancel_corrupt_copy_and_retry() { + for version in [6, 7] { + let root = TempDir::new().unwrap(); + crate::open_or_initialize_project(root.path()).unwrap(); + let mut session = create( + &root, + Uuid::new_v4(), + version, + GraphConstructionBudgets::default(), + None, + ); + session + .append(ConstructionChunkKind::Node, "nodes", &node_batch(1, 2)) + .unwrap(); + let receipt = session.read_receipt(0).unwrap().details; + let path = session.root.path().join(&receipt.name); + let original = std::fs::read(&path).unwrap(); + let current = std::fs::read(root.path().join("CURRENT")).unwrap(); + let codec = DetailCodec::from_version(version).unwrap(); + let output = "merge-node-codec-retry.run"; + let cancelled = copy_authenticated_run_with_codec::( + &session.root, + &receipt, + output, + &mut || true, + &mut session.checkpoint.evidence, + Some(codec), + ) + .unwrap_err(); + assert!(cancelled.to_string().contains("construction cancelled")); + assert!(shape_temporary_names(&session.root).is_empty()); + assert!(!session.root.path().join(output).exists()); + assert_eq!(std::fs::read(&path).unwrap(), original); + let mut corrupt = original.clone(); + corrupt[17] = b'X'; // Still valid UTF-8 and ordering: digest must detect it. + std::fs::write(&path, &corrupt).unwrap(); + let corrupted = copy_authenticated_run_with_codec::( + &session.root, + &receipt, + output, + &mut || false, + &mut session.checkpoint.evidence, + Some(codec), + ) + .unwrap_err(); + assert!( + corrupted.to_string().contains("source content changed"), + "{corrupted}" + ); + assert!(shape_temporary_names(&session.root).is_empty()); + assert!(!session.root.path().join(output).exists()); + assert_eq!(std::fs::read(root.path().join("CURRENT")).unwrap(), current); + std::fs::write(&path, &original).unwrap(); + copy_authenticated_run_with_codec::( + &session.root, + &receipt, + output, + &mut || false, + &mut session.checkpoint.evidence, + Some(codec), + ) + .unwrap(); + assert_eq!( + std::fs::read(session.root.path().join(output)).unwrap(), + original + ); + assert!(shape_temporary_names(&session.root).is_empty()); + } + } + #[test] + fn detail_codec_initial_temporary_rejects_unknown_conflicting_and_changed_binding() { + for case in ["unknown", "conflict", "budgets", "noninitial"] { + let root = TempDir::new().unwrap(); + crate::open_or_initialize_project(root.path()).unwrap(); + let status = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "graph_construction::tests::compact_details::detail_codec_crash_child", + "--nocapture", + ]) + .env("GF_DETAIL_CODEC_CRASH_ROOT", root.path()) + .env("GF_DETAIL_CODEC_VERSION", "6") + .env( + "GF_CONSTRUCTION_FAILPOINT_COOKIE", + "graphforge-construction-test-v1", + ) + .env( + "GF_CONSTRUCTION_FAILPOINT", + "control.install.after_temp_fsync.checkpoint.json", + ) + .status() + .unwrap(); + assert_eq!(status.code(), Some(86)); + let private = root + .path() + .join(PRIVATE_ROOT) + .join(Uuid::from_u128(117_200).simple().to_string()); + let temporary = std::fs::read_dir(&private) + .unwrap() + .map(|entry| entry.unwrap().path()) + .find(|path| { + path.file_name() + .unwrap() + .to_str() + .unwrap() + .starts_with(".checkpoint.json-") + }) + .unwrap(); + let mut candidate: Checkpoint = + serde_json::from_slice(&std::fs::read(&temporary).unwrap()).unwrap(); + match case { + "unknown" => candidate.format_version = 99, + "conflict" => candidate.format_version = 7, + "budgets" => candidate.budgets.max_batch_rows /= 2, + "noninitial" => candidate.saw_edge = true, + _ => unreachable!(), + } + let destination = if case == "conflict" { + private.join(control_temp(CHECKPOINT)) + } else { + temporary + }; + let bytes = serde_json::to_vec(&candidate).unwrap(); + std::fs::write(&destination, &bytes).unwrap(); + let current = std::fs::read(root.path().join("CURRENT")).unwrap(); + let error = GraphConstructionSession::open( + root.path(), + Uuid::from_u128(117_200), + 0, + GraphConstructionBudgets::default(), + ) + .err() + .expect("invalid initial authority rejected"); + assert!( + error.to_string().contains("checkpoint"), + "case={case}: {error}" + ); + assert!(!private.join(CHECKPOINT).exists()); + assert_eq!(std::fs::read(destination).unwrap(), bytes); + assert_eq!(std::fs::read(root.path().join("CURRENT")).unwrap(), current); + } + } +} diff --git a/crates/graphforge-storage/src/graph_construction.rs b/crates/graphforge-storage/src/graph_construction.rs index af105743..2922c5b6 100644 --- a/crates/graphforge-storage/src/graph_construction.rs +++ b/crates/graphforge-storage/src/graph_construction.rs @@ -36,9 +36,10 @@ use sha2::{Digest, Sha256}; use uuid::Uuid; use crate::UuidIndexKind; +use crate::construction_detail_codec::{DetailCodec, DetailValidator}; use crate::uuid_membership::{AuthenticatedUuidIndexSnapshot, UuidConstructionSnapshotWork}; -const FORMAT_VERSION: u32 = 6; +const FORMAT_VERSION: u32 = 7; const PRIVATE_ROOT: &str = ".graphforge-construction"; const SESSION_LOCK: &str = "session.lock"; const CHECKPOINT: &str = "checkpoint.json"; @@ -1288,7 +1289,7 @@ impl GraphConstructionSession { .clone() .ok_or_else(|| storage("publication requires encoded inventory authority"))?; let intent = ConstructionPublicationIntent { - format_version: FORMAT_VERSION, + format_version: self.checkpoint.format_version, operation_uuid: self.checkpoint.operation_uuid, project_identity: self.checkpoint.project_identity.clone(), session_identity: self.checkpoint.session_identity.clone(), @@ -1404,6 +1405,7 @@ impl GraphConstructionSession { } let encoded = crate::graph_construction_encoding::encode( &self.root, + DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?, shape, generation, self.checkpoint.ontology_mode, @@ -2076,6 +2078,34 @@ impl GraphConstructionSession { lifecycle_mode: crate::filesystem_admission::ProjectLifecycleMode, allocation: Option<&crate::StorageAllocationOperation>, ) -> Result { + Self::open_internal_with_format( + project_dir, + graph_source_dir, + operation_uuid, + parent_topology_generation, + ontology_mode, + semantic_authority, + budgets, + lifecycle_mode, + allocation, + FORMAT_VERSION, + ) + } + + #[allow(clippy::too_many_arguments, clippy::too_many_lines)] + fn open_internal_with_format( + project_dir: &Path, + graph_source_dir: &Path, + operation_uuid: Uuid, + parent_topology_generation: u64, + ontology_mode: graphforge_core::OntologyMode, + semantic_authority: Option, + budgets: GraphConstructionBudgets, + lifecycle_mode: crate::filesystem_admission::ProjectLifecycleMode, + allocation: Option<&crate::StorageAllocationOperation>, + initial_format_version: u32, + ) -> Result { + DetailCodec::from_version(initial_format_version).map_err(storage)?; let budgets = budgets.validate()?; let semantic_authority_sha256 = semantic_authority .as_ref() @@ -2114,13 +2144,6 @@ impl GraphConstructionSession { )); } let session_identity = root.identity(); - cleanup_authenticated_control_temps( - &root, - operation_uuid, - project_identity, - session_identity, - )?; - cleanup_owned_artifact_temps(&root)?; // Authenticate private recovery authority before consulting the // mutable public pointer. A publishing/published replay resolves its // exact immutable parent and therefore remains recoverable after @@ -2130,14 +2153,23 @@ impl GraphConstructionSession { Err(error) if error.kind() == std::io::ErrorKind::NotFound => None, Err(error) => return Err(storage(error)), }; - if let Some(checkpoint) = recovered_checkpoint.as_mut() { - if checkpoint.format_version != FORMAT_VERSION + if let Some(checkpoint) = recovered_checkpoint.as_mut() + && (DetailCodec::from_version(checkpoint.format_version).is_err() || checkpoint.operation_uuid != operation_uuid || !checkpoint.project_identity.matches(project_identity) - || !checkpoint.session_identity.matches(session_identity) - { - return Err(storage("checkpoint private authority changed")); - } + || !checkpoint.session_identity.matches(session_identity)) + { + return Err(storage("checkpoint private authority changed")); + } + if let Some(checkpoint) = recovered_checkpoint.as_mut() { + cleanup_authenticated_control_temps( + &root, + operation_uuid, + project_identity, + session_identity, + checkpoint.format_version, + )?; + cleanup_owned_artifact_temps(&root)?; recover_publication(project_dir, &root, checkpoint)?; } let (parent_generation_uuid, parent_generation_manifest_sha256) = @@ -2292,8 +2324,8 @@ impl GraphConstructionSession { peak_catalog_identifier_bytes: parent_catalog.retained_identifier_bytes() as u64, ..GraphConstructionEvidence::default() }; - let initial = Checkpoint { - format_version: FORMAT_VERSION, + let mut initial = Checkpoint { + format_version: initial_format_version, operation_uuid, project_identity: project_identity.into(), session_identity: session_identity.into(), @@ -2324,6 +2356,15 @@ impl GraphConstructionSession { base_work, evidence, }; + initial.format_version = initial_checkpoint_format(&root, project_identity, &initial)?; + cleanup_authenticated_control_temps( + &root, + operation_uuid, + project_identity, + session_identity, + initial.format_version, + )?; + cleanup_owned_artifact_temps(&root)?; install_control(&root, CHECKPOINT, &initial)?; initial }; @@ -2683,7 +2724,11 @@ impl GraphConstructionSession { && receipt.input_sha256 == input_sha256 && receipt.schema_sha256 == schema_sha256 { - let work = validate_receipt_artifacts(&self.root, &receipt)?; + let work = validate_receipt_artifacts( + &self.root, + &receipt, + DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?, + )?; self.checkpoint.evidence.replay_validation_read_bytes = self .checkpoint .evidence @@ -2739,7 +2784,7 @@ impl GraphConstructionSession { } let sequence = self.checkpoint.next_sequence; let mut intent = ChunkIntent { - format_version: FORMAT_VERSION, + format_version: self.checkpoint.format_version, operation_uuid: self.checkpoint.operation_uuid, project_identity: self.checkpoint.project_identity.clone(), session_identity: self.checkpoint.session_identity.clone(), @@ -2830,17 +2875,19 @@ impl GraphConstructionSession { reject_cancelled(&mut cancelled)?; } intent.details = Some(match &arrays.details { - DetailRuns::Node(records) => write_fixed_run( + DetailRuns::Node(records) => write_run( &self.root, &format!("{stem}.node-details.run"), records, &mut self.checkpoint.evidence, + Some(DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?), )?, - DetailRuns::Edge(records) => write_fixed_run( + DetailRuns::Edge(records) => write_run( &self.root, &format!("{stem}.edge-details.run"), records, &mut self.checkpoint.evidence, + Some(DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?), )?, }); replace_control(&self.root, INTENT, &intent)?; @@ -2880,7 +2927,12 @@ impl GraphConstructionSession { let mut read_operations = 0_u64; for sequence in 0..self.checkpoint.next_sequence { let receipt = self.read_receipt(sequence)?; - validate_receipt_semantics(&receipt, sequence, self.checkpoint.budgets)?; + validate_receipt_semantics( + &receipt, + sequence, + self.checkpoint.budgets, + DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?, + )?; if receipt.kind == ConstructionChunkKind::Node && saw_edge { return Err(storage("node receipt follows edge receipt")); } @@ -2889,7 +2941,11 @@ impl GraphConstructionSession { return Err(storage("receipt journal chain is discontinuous")); } if authenticate_artifacts { - let work = validate_receipt_artifacts(&self.root, &receipt)?; + let work = validate_receipt_artifacts( + &self.root, + &receipt, + DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?, + )?; account_cache_release(work.cache_release, &mut self.checkpoint.evidence)?; read_bytes = read_bytes .checked_add(work.bytes) @@ -2963,13 +3019,17 @@ impl GraphConstructionSession { let fan_in = self.checkpoint.budgets.merge_fan_in; let mut unified = FixedMergeAccumulator::new("merge-identities", fan_in, true); - let mut node_details = FixedMergeAccumulator::new("merge-node-details", fan_in, true); - let mut edge_details = FixedMergeAccumulator::new("merge-edge-details", fan_in, true); + let detail_codec = + DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?; + let mut node_details = FixedMergeAccumulator::new("merge-node-details", fan_in, true) + .with_detail_codec(detail_codec); + let mut edge_details = FixedMergeAccumulator::new("merge-edge-details", fan_in, true) + .with_detail_codec(detail_codec); let mut endpoints = FixedMergeAccumulator::new("merge-endpoints", fan_in, false); let mut row_groups: BTreeMap<(u8, String), RowMergeAccumulator> = BTreeMap::new(); let mut catalog_authority = Sha256::new(); let shape_intent = ShapeIntent { - format_version: FORMAT_VERSION, + format_version: self.checkpoint.format_version, operation_uuid: self.checkpoint.operation_uuid, project_identity: self.checkpoint.project_identity.clone(), session_identity: self.checkpoint.session_identity.clone(), @@ -2993,7 +3053,11 @@ impl GraphConstructionSession { // digest in the merge consumers below. Parquet's range-oriented // decoder cannot establish a whole-file digest, so retain exactly // one explicit whole-file authentication pass for that artifact. - let mut work = authenticate_artifact(&self.root, &receipt.parquet)?; + let mut work = authenticate_artifact( + &self.root, + &receipt.parquet, + DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?, + )?; account_cache_release(work.cache_release, &mut self.checkpoint.evidence)?; let metadata_work = validate_parquet_metadata(&self.root, &receipt)?; account_cache_release(metadata_work.cache_release, &mut self.checkpoint.evidence)?; @@ -3053,12 +3117,13 @@ impl GraphConstructionSession { match receipt.kind { ConstructionChunkKind::Node => { let name = format!("merge-node-source-{sequence:020}.run"); - copy_authenticated_run::( + copy_authenticated_run_with_codec::( &self.root, &receipt.details, &name, &mut cancelled, &mut self.checkpoint.evidence, + Some(detail_codec), )?; node_details.push::( &self.root, @@ -3069,12 +3134,13 @@ impl GraphConstructionSession { } ConstructionChunkKind::Edge => { let detail = format!("merge-edge-source-{sequence:020}.run"); - copy_authenticated_run::( + copy_authenticated_run_with_codec::( &self.root, &receipt.details, &detail, &mut cancelled, &mut self.checkpoint.evidence, + Some(detail_codec), )?; edge_details.push::( &self.root, @@ -3163,6 +3229,7 @@ impl GraphConstructionSession { &staged_identities, node_details.as_deref(), edge_details.as_deref(), + detail_codec, &mut cancelled, &mut self.checkpoint.evidence, )?; @@ -3319,7 +3386,7 @@ impl GraphConstructionSession { &self.root, SHAPE_INTENT, &ShapeIntent { - format_version: FORMAT_VERSION, + format_version: self.checkpoint.format_version, operation_uuid: self.checkpoint.operation_uuid, project_identity: self.checkpoint.project_identity.clone(), session_identity: self.checkpoint.session_identity.clone(), @@ -3414,11 +3481,20 @@ impl GraphConstructionSession { match self.root.open_child_file(OsStr::new(&receipt_name)) { Ok(mut receipt_file) => { let receipt: ConstructionChunkReceipt = decode_bounded(&mut receipt_file)?; - validate_receipt_semantics(&receipt, intent.sequence, self.checkpoint.budgets)?; + validate_receipt_semantics( + &receipt, + intent.sequence, + self.checkpoint.budgets, + DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?, + )?; if receipt != receipt_from_intent(&intent)? { return Err(storage("recovered receipt differs from durable intent")); } - let recovery_work = validate_receipt_artifacts(&self.root, &receipt)?; + let recovery_work = validate_receipt_artifacts( + &self.root, + &receipt, + DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?, + )?; self.checkpoint.evidence.recovery_application_read_bytes = self .checkpoint .evidence @@ -3478,7 +3554,12 @@ impl GraphConstructionSession { .into_iter() .flatten() { - let recovery_work = authenticate_artifact(&self.root, &artifact)?; + let recovery_work = authenticate_artifact( + &self.root, + &artifact, + DetailCodec::from_version(self.checkpoint.format_version) + .map_err(storage)?, + )?; account_cache_release( recovery_work.cache_release, &mut self.checkpoint.evidence, @@ -3506,6 +3587,8 @@ impl GraphConstructionSession { &format!("{stem}.parquet"), intent.kind, intent.rows, + DetailCodec::from_version(self.checkpoint.format_version) + .map_err(storage)?, )?; } if intent.identities.is_none() { @@ -3514,6 +3597,8 @@ impl GraphConstructionSession { &format!("{stem}.identities.run"), intent.kind, intent.rows, + DetailCodec::from_version(self.checkpoint.format_version) + .map_err(storage)?, )?; } if intent.kind == ConstructionChunkKind::Edge && intent.endpoints.is_none() { @@ -3522,6 +3607,8 @@ impl GraphConstructionSession { &format!("{stem}.endpoints.run"), intent.kind, intent.rows, + DetailCodec::from_version(self.checkpoint.format_version) + .map_err(storage)?, )?; } if intent.details.is_none() { @@ -3537,6 +3624,8 @@ impl GraphConstructionSession { ), intent.kind, intent.rows, + DetailCodec::from_version(self.checkpoint.format_version) + .map_err(storage)?, )?; } } @@ -3551,7 +3640,12 @@ impl GraphConstructionSession { .open_child_file(OsStr::new(&receipt_name(sequence))) .map_err(storage)?; let receipt = decode_bounded(&mut file)?; - validate_receipt_semantics(&receipt, sequence, self.checkpoint.budgets)?; + validate_receipt_semantics( + &receipt, + sequence, + self.checkpoint.budgets, + DetailCodec::from_version(self.checkpoint.format_version).map_err(storage)?, + )?; if receipt.operation_uuid != self.checkpoint.operation_uuid || receipt.project_identity != self.checkpoint.project_identity || receipt.session_identity != self.checkpoint.session_identity @@ -4361,6 +4455,16 @@ fn write_fixed_run( name: &str, records: &[[u8; N]], evidence: &mut GraphConstructionEvidence, +) -> Result { + write_run(root, name, records, evidence, None) +} + +fn write_run( + root: &StableDirectory, + name: &str, + records: &[[u8; N]], + evidence: &mut GraphConstructionEvidence, + codec: Option, ) -> Result { let temporary = artifact_temp(name); let file = root @@ -4373,7 +4477,7 @@ fn write_fixed_run( for group in records.chunks(records_per_block) { block.clear(); for record in group { - block.extend_from_slice(record); + block.extend_from_slice(run_record_bytes(record, codec)?); } writer.write_all(&block).map_err(storage)?; } @@ -4683,7 +4787,7 @@ fn cleanup_incomplete_shape_capabilities(root: &StableDirectory) -> Result<(), G } fn validate_shape_binding(intent: &ShapeIntent, checkpoint: &Checkpoint) -> Result<(), GfError> { - if intent.format_version != FORMAT_VERSION + if intent.format_version != checkpoint.format_version || intent.operation_uuid != checkpoint.operation_uuid || intent.project_identity != checkpoint.project_identity || intent.session_identity != checkpoint.session_identity @@ -4730,7 +4834,7 @@ fn validate_publication_intent( )?; validate_sha256(&intent.shape_authority_sha256, "shape authority")?; validate_sha256(&intent.encoding_inventory_sha256, "encoding inventory")?; - if intent.format_version != FORMAT_VERSION + if intent.format_version != checkpoint.format_version || intent.operation_uuid != checkpoint.operation_uuid || intent.project_identity != checkpoint.project_identity || intent.session_identity != checkpoint.session_identity @@ -5600,7 +5704,13 @@ fn unlink_shape_artifact( fn account_merge_read( evidence: &mut GraphConstructionEvidence, ) -> Result<(), GfError> { - let _ = N; + account_merge_read_bytes(evidence, N as u64) +} + +fn account_merge_read_bytes( + evidence: &mut GraphConstructionEvidence, + _bytes: u64, +) -> Result<(), GfError> { evidence.merge_read_records = evidence .merge_read_records .checked_add(1) @@ -5610,6 +5720,13 @@ fn account_merge_read( fn account_merge_write( evidence: &mut GraphConstructionEvidence, +) -> Result<(), GfError> { + account_merge_write_bytes(evidence, N as u64) +} + +fn account_merge_write_bytes( + evidence: &mut GraphConstructionEvidence, + bytes: u64, ) -> Result<(), GfError> { evidence.merge_written_records = evidence .merge_written_records @@ -5617,7 +5734,7 @@ fn account_merge_write( .ok_or_else(|| storage("merge written record count overflows"))?; evidence.merge_written_bytes = evidence .merge_written_bytes - .checked_add(N as u64) + .checked_add(bytes) .ok_or_else(|| storage("merge written byte count overflows"))?; Ok(()) } @@ -6076,13 +6193,24 @@ fn convert_identity_run( Ok(()) } -#[allow(clippy::too_many_lines)] fn copy_authenticated_run( root: &StableDirectory, receipt: &ArtifactReceipt, output: &str, cancelled: &mut impl FnMut() -> bool, evidence: &mut GraphConstructionEvidence, +) -> Result<(), GfError> { + copy_authenticated_run_with_codec::(root, receipt, output, cancelled, evidence, None) +} + +#[allow(clippy::too_many_lines)] // Retain the existing coupled authentication/publication cleanup scope. +fn copy_authenticated_run_with_codec( + root: &StableDirectory, + receipt: &ArtifactReceipt, + output: &str, + cancelled: &mut impl FnMut() -> bool, + evidence: &mut GraphConstructionEvidence, + codec: Option, ) -> Result<(), GfError> { let input = root .open_child_file(OsStr::new(&receipt.name)) @@ -6150,14 +6278,15 @@ fn copy_authenticated_run( let mut digest = Sha256::new(); let mut bytes = 0_u64; let copied = (|| -> Result<(), GfError> { - while let Some(record) = read_fixed::(&mut reader)? { - digest.update(record); + while let Some(record) = read_run_record::(&mut reader, codec)? { + let wire = run_record_bytes(&record, codec)?; + digest.update(wire); bytes = bytes - .checked_add(N as u64) + .checked_add(wire.len() as u64) .ok_or_else(|| storage("bytes overflows"))?; - writer.write_all(&record).map_err(storage)?; - account_merge_read::(evidence)?; - account_merge_write::(evidence)?; + writer.write_all(wire).map_err(storage)?; + account_merge_read_bytes(evidence, wire.len() as u64)?; + account_merge_write_bytes(evidence, wire.len() as u64)?; reject_cancelled(cancelled)?; } if bytes != receipt.bytes || hex(&digest.finalize()) != receipt.sha256 { @@ -6295,6 +6424,7 @@ fn copy_authenticated_run( /// logarithmic level rather than one name per accepted chunk. struct FixedMergeAccumulator { prefix: &'static str, + detail_codec: Option, fan_in: usize, reject_duplicates: bool, levels: Vec>, @@ -6319,6 +6449,7 @@ impl FixedMergeAccumulator { fn new(prefix: &'static str, fan_in: usize, reject_duplicates: bool) -> Self { Self { prefix, + detail_codec: None, fan_in, reject_duplicates, levels: Vec::new(), @@ -6327,6 +6458,11 @@ impl FixedMergeAccumulator { } } + fn with_detail_codec(mut self, codec: DetailCodec) -> Self { + self.detail_codec = Some(codec); + self + } + fn slot_count(&self) -> usize { self.levels.iter().map(Vec::len).sum() } @@ -6369,6 +6505,7 @@ impl FixedMergeAccumulator { &inputs, &name, self.reject_duplicates, + self.detail_codec, cancelled, evidence, )?; @@ -6435,6 +6572,7 @@ impl FixedMergeAccumulator { &inputs, &output, self.reject_duplicates, + self.detail_codec, cancelled, evidence, )?; @@ -6461,6 +6599,7 @@ fn merge_fixed_group( inputs: &[String], output: &str, reject_duplicates: bool, + codec: Option, cancelled: &mut impl FnMut() -> bool, evidence: &mut GraphConstructionEvidence, ) -> Result { @@ -6500,9 +6639,9 @@ fn merge_fixed_group( let mut writer = BufWriter::with_capacity(BLOCK_BYTES, hashing); let mut heap = BinaryHeap::new(); for (index, reader) in readers.iter_mut().enumerate() { - if let Some(record) = read_fixed::(reader)? { + if let Some(record) = read_run_record::(reader, codec)? { heap.push(Reverse((record, index))); - account_merge_read::(evidence)?; + account_merge_read_bytes(evidence, run_record_bytes(&record, codec)?.len() as u64)?; } } let mut previous = None; @@ -6514,15 +6653,16 @@ fn merge_fixed_group( { return Err(storage("duplicate identity across construction runs")); } - writer.write_all(&record).map_err(storage)?; - account_merge_write::(evidence)?; + let wire = run_record_bytes(&record, codec)?; + writer.write_all(wire).map_err(storage)?; + account_merge_write_bytes(evidence, wire.len() as u64)?; previous = Some(record); if evidence.merge_written_records.is_multiple_of(4096) { reject_cancelled(cancelled)?; } - if let Some(next) = read_fixed::(&mut readers[index])? { + if let Some(next) = read_run_record::(&mut readers[index], codec)? { heap.push(Reverse((next, index))); - account_merge_read::(evidence)?; + account_merge_read_bytes(evidence, run_record_bytes(&next, codec)?.len() as u64)?; } } writer.flush().map_err(storage)?; @@ -7431,6 +7571,7 @@ where catalog, Some(digest.to_owned()), ReadWork { + detail_records: 0, bytes, operations, cache_release: cache_release.evidence(), @@ -7438,6 +7579,26 @@ where )) } +fn read_run_record( + reader: &mut impl Read, + codec: Option, +) -> Result, GfError> { + match codec { + Some(codec) => codec.read(reader).map_err(storage), + None => read_fixed(reader), + } +} + +fn run_record_bytes( + record: &[u8; N], + codec: Option, +) -> Result<&[u8], GfError> { + match codec { + Some(codec) => codec.bytes(record).map_err(storage), + None => Ok(record), + } +} + fn read_fixed(reader: &mut impl Read) -> Result, GfError> { let mut record = [0_u8; N]; let mut filled = 0; @@ -7456,6 +7617,7 @@ fn validate_staged_details( identities_name: &str, node_details_name: Option<&str>, edge_details_name: Option<&str>, + codec: DetailCodec, cancelled: &mut impl FnMut() -> bool, evidence: &mut GraphConstructionEvidence, ) -> Result<(u64, u64), GfError> { @@ -7490,40 +7652,27 @@ fn validate_staged_details( } } account_fixed_read_operations(&identities_counter, evidence)?; - let count = |name: Option<&str>, width: u64| -> Result { - let Some(name) = name else { return Ok(0) }; - let bytes = root - .open_child_file(OsStr::new(name)) - .map_err(storage)? - .metadata() - .map_err(storage)? - .len(); - if bytes % width != 0 { - return Err(storage("truncated canonical detail run")); - } - Ok(bytes / width) - }; - if count(node_details_name, NODE_DETAIL_WIDTH as u64)? != nodes - || count(edge_details_name, EDGE_DETAIL_WIDTH as u64)? != edges - { - return Err(storage("staged identity and detail domains disagree")); - } - validate_detail_domain::( + let actual_nodes = validate_detail_domain::( root, identities_name, node_details_name, 0, + codec, cancelled, evidence, )?; - validate_detail_domain::( + let actual_edges = validate_detail_domain::( root, identities_name, edge_details_name, 1, + codec, cancelled, evidence, )?; + if actual_nodes != nodes || actual_edges != edges { + return Err(storage("staged identity and detail domains disagree")); + } Ok((nodes, edges)) } @@ -7633,6 +7782,7 @@ fn validate_unified_and_details( identities_name, node_details_name, 0, + DetailCodec::Legacy, cancelled, evidence, )?; @@ -7641,6 +7791,7 @@ fn validate_unified_and_details( identities_name, edge_details_name, 1, + DetailCodec::Legacy, cancelled, evidence, )?; @@ -7668,11 +7819,12 @@ fn validate_detail_domain( identities_name: &str, details_name: Option<&str>, kind: u8, + codec: DetailCodec, cancelled: &mut impl FnMut() -> bool, evidence: &mut GraphConstructionEvidence, -) -> Result<(), GfError> { +) -> Result { let Some(details_name) = details_name else { - return Ok(()); + return Ok(0); }; let (mut identities, identities_counter) = open_counted_fixed_reader(root, identities_name, evidence)?; @@ -7682,7 +7834,7 @@ fn validate_detail_domain( account_merge_read::(evidence)?; } let mut count = 0_u64; - while let Some(detail) = read_fixed::(&mut details)? { + while let Some(detail) = codec.read::(&mut details).map_err(storage)? { while identity .as_ref() .is_some_and(|item| item[17] == 1 || item[16] != kind || item[..16] < detail[..16]) @@ -7721,7 +7873,7 @@ fn validate_detail_domain( account_fixed_read_operations(&details_counter, evidence)?; release_counted_reader_cache(&mut identities, evidence)?; release_counted_reader_cache(&mut details, evidence)?; - Ok(()) + Ok(count) } #[cfg(test)] @@ -8030,6 +8182,7 @@ fn account_probe_work( #[derive(Clone, Copy, Debug, Default)] struct ReadWork { + detail_records: u64, bytes: u64, operations: u64, cache_release: graphforge_filesystem::FileCacheReleaseEvidence, @@ -8039,8 +8192,9 @@ struct ReadWork { fn authenticate_artifact( root: &StableDirectory, receipt: &ArtifactReceipt, + codec: DetailCodec, ) -> Result { - validate_artifact_name(receipt)?; + validate_artifact_name(receipt, codec)?; let file = root .open_child_file(OsStr::new(&receipt.name)) .map_err(storage)?; @@ -8053,7 +8207,7 @@ fn authenticate_artifact( } let releasing = graphforge_filesystem::FileCacheReleasingReader::new(file).map_err(storage)?; let mut reader = BufReader::with_capacity(BLOCK_BYTES, releasing); - let result = (|| -> Result<(u64, u64), GfError> { + let result = (|| -> Result<(u64, u64, u64), GfError> { let mut block = vec![0_u8; BLOCK_BYTES]; let mut digest = Sha256::new(); let mut bytes = 0_u64; @@ -8069,6 +8223,11 @@ fn authenticate_artifact( } else { None }; + let mut detail = width + .filter(|width| matches!(*width, NODE_DETAIL_WIDTH | EDGE_DETAIL_WIDTH)) + .map(|width| DetailValidator::new(codec, width)) + .transpose() + .map_err(storage)?; let mut pending = Vec::new(); let mut previous: Option> = None; loop { @@ -8083,7 +8242,9 @@ fn authenticate_artifact( operations = operations .checked_add(1) .ok_or_else(|| storage("operations overflows"))?; - if let Some(width) = width { + if let Some(detail) = detail.as_mut() { + detail.consume(&block[..count]).map_err(storage)?; + } else if let Some(width) = width { pending.extend_from_slice(&block[..count]); let complete = pending.len() / width * width; for record in pending[..complete].chunks_exact(width) { @@ -8133,10 +8294,16 @@ fn authenticate_artifact( if bytes != receipt.bytes || hex(&digest.finalize()) != receipt.sha256 { return Err(storage("artifact digest or size changed")); } - Ok((bytes, operations)) + let records = detail + .as_ref() + .map(DetailValidator::finish) + .transpose() + .map_err(storage)? + .unwrap_or(0); + Ok((bytes, operations, records)) })(); let release = reader.get_mut().finish().map_err(storage); - let (bytes, operations) = match (result, release) { + let (bytes, operations, detail_records) = match (result, release) { (Ok(value), Ok(_)) => value, (Ok(_), Err(error)) => return Err(error), (Err(primary), Ok(_)) => return Err(primary), @@ -8148,13 +8315,14 @@ fn authenticate_artifact( }; let cache_release = reader.get_ref().tracker().evidence(); Ok(ReadWork { + detail_records, bytes, operations, cache_release, }) } -fn validate_artifact_name(receipt: &ArtifactReceipt) -> Result<(), GfError> { +fn validate_artifact_name(receipt: &ArtifactReceipt, codec: DetailCodec) -> Result<(), GfError> { let valid_suffix = receipt.name.ends_with(".parquet") || receipt.name.ends_with(".identities.run") || receipt.name.ends_with(".endpoints.run") @@ -8182,12 +8350,14 @@ fn validate_artifact_name(receipt: &ArtifactReceipt) -> Result<(), GfError> { { return Err(storage("truncated endpoint run")); } - if receipt.name.ends_with(".node-details.run") + if codec == DetailCodec::Legacy + && receipt.name.ends_with(".node-details.run") && !receipt.bytes.is_multiple_of(NODE_DETAIL_WIDTH as u64) { return Err(storage("truncated node detail run")); } - if receipt.name.ends_with(".edge-details.run") + if codec == DetailCodec::Legacy + && receipt.name.ends_with(".edge-details.run") && !receipt.bytes.is_multiple_of(EDGE_DETAIL_WIDTH as u64) { return Err(storage("truncated edge detail run")); @@ -8233,6 +8403,7 @@ fn validate_receipt_semantics( receipt: &ConstructionChunkReceipt, sequence: u64, budgets: GraphConstructionBudgets, + codec: DetailCodec, ) -> Result<(), GfError> { validate_chunk_id(&receipt.chunk_id)?; if receipt.sequence != sequence @@ -8281,14 +8452,14 @@ fn validate_receipt_semantics( } else { EDGE_DETAIL_WIDTH }; - if receipt.details.bytes / detail_width as u64 != receipt.rows { - return Err(storage("detail receipt semantics are inconsistent")); - } - validate_artifact_name(&receipt.parquet)?; - validate_artifact_name(&receipt.identities)?; - validate_artifact_name(&receipt.details)?; + codec + .validate_size(detail_width, receipt.rows, receipt.details.bytes) + .map_err(storage)?; + validate_artifact_name(&receipt.parquet, codec)?; + validate_artifact_name(&receipt.identities, codec)?; + validate_artifact_name(&receipt.details, codec)?; if let Some(endpoints) = &receipt.endpoints { - validate_artifact_name(endpoints)?; + validate_artifact_name(endpoints, codec)?; } Ok(()) } @@ -8296,13 +8467,19 @@ fn validate_receipt_semantics( fn validate_receipt_artifacts( root: &StableDirectory, receipt: &ConstructionChunkReceipt, + codec: DetailCodec, ) -> Result { let mut work = ReadWork::default(); for artifact in [&receipt.parquet, &receipt.identities, &receipt.details] .into_iter() .chain(receipt.endpoints.iter()) { - let artifact_work = authenticate_artifact(root, artifact)?; + let artifact_work = authenticate_artifact(root, artifact, codec)?; + if artifact.name == receipt.details.name && artifact_work.detail_records != receipt.rows { + return Err(storage( + "detail receipt row count differs from authenticated records", + )); + } work.bytes = work .bytes .checked_add(artifact_work.bytes) @@ -8405,6 +8582,7 @@ fn validate_parquet_shape( "Parquet shape", )?; Ok(ReadWork { + detail_records: 0, bytes: counter.bytes.load(Ordering::Relaxed), operations: counter.operations.load(Ordering::Relaxed), cache_release: cache_release.evidence(), @@ -8452,6 +8630,7 @@ fn validate_parquet_metadata( "Parquet metadata", )?; Ok(ReadWork { + detail_records: 0, bytes: counter.bytes.load(Ordering::Relaxed), operations: counter.operations.load(Ordering::Relaxed), cache_release: cache_release.evidence(), @@ -8483,11 +8662,8 @@ fn validate_intent(intent: &ChunkIntent, checkpoint: &Checkpoint) -> Result<(), } else { EDGE_DETAIL_WIDTH }; - let expected_detail_bytes = intent - .rows - .checked_mul(detail_width as u64) - .ok_or_else(|| storage("intent detail byte count overflow"))?; - if intent.format_version != FORMAT_VERSION + let codec = DetailCodec::from_version(checkpoint.format_version).map_err(storage)?; + if intent.format_version != checkpoint.format_version || intent.operation_uuid != checkpoint.operation_uuid || intent.project_identity != checkpoint.project_identity || intent.session_identity != checkpoint.session_identity @@ -8530,7 +8706,9 @@ fn validate_intent(intent: &ChunkIntent, checkpoint: &Checkpoint) -> Result<(), "edge" } ) - || artifact.bytes != expected_detail_bytes + || codec + .validate_size(detail_width, intent.rows, artifact.bytes) + .is_err() }) { return Err(storage("durable intent is inconsistent with checkpoint")); @@ -8566,7 +8744,7 @@ fn validate_checkpoint( .len() .checked_add(checkpoint.edge_schema_sha256.len()) .ok_or_else(|| storage("checkpoint schema-group count overflow"))?; - if checkpoint.format_version != FORMAT_VERSION + if DetailCodec::from_version(checkpoint.format_version).is_err() || checkpoint.operation_uuid != operation || !checkpoint.project_identity.matches(project) || !checkpoint.session_identity.matches(session) @@ -8647,11 +8825,74 @@ fn validate_checkpoint( Ok(()) } +/// A completed initial checkpoint temporary pins only the codec choice. It is +/// never promoted: the ordinary writer creates the initial checkpoint after +/// verifying the candidate against the current admitted parent and parameters. +fn initial_checkpoint_format( + root: &StableDirectory, + project_identity: FileIdentity, + initial: &Checkpoint, +) -> Result { + let mut selected = None; + for name in root.child_names().map_err(storage)? { + let Some(text) = name.to_str() else { continue }; + if !text.starts_with(".checkpoint.json-") || !is_control_temp(text) { + continue; + } + let mut file = root.open_child_file(&name).map_err(storage)?; + let body = read_bounded_limit(&mut file, MAX_CONTROL_BYTES)?; + let Ok(candidate) = serde_json::from_slice::(&body) else { + continue; + }; + if candidate.operation_uuid != initial.operation_uuid + || candidate.project_identity != initial.project_identity + || candidate.session_identity != initial.session_identity + { + continue; + } + if file_link_count(&file).map_err(storage)? != 1 { + return Err(storage("initial checkpoint temporary has unexpected links")); + } + validate_checkpoint( + &candidate, + initial.operation_uuid, + project_identity, + root.identity(), + initial.parent_topology_generation, + initial.ontology_mode, + initial.lifecycle_mode, + initial.semantic_authority_sha256.as_deref(), + initial.budgets, + initial.parent_catalog_sha256.as_deref(), + initial.parent_generation_uuid, + &initial.parent_generation_manifest_sha256, + )?; + if candidate.state != GraphConstructionState::Staging + || candidate.next_sequence != 0 + || candidate.saw_edge + || candidate.last_receipt_sha256.is_some() + || candidate.publication_state.is_some() + || candidate.shape_authority_sha256.is_some() + || candidate.encoding_inventory_sha256.is_some() + || !candidate.node_schema_sha256.is_empty() + || !candidate.edge_schema_sha256.is_empty() + { + return Err(storage("checkpoint temporary is not initial authority")); + } + if selected.is_some_and(|version| version != candidate.format_version) { + return Err(storage("initial checkpoint temporary versions disagree")); + } + selected = Some(candidate.format_version); + } + Ok(selected.unwrap_or(initial.format_version)) +} + fn cleanup_authenticated_control_temps( root: &StableDirectory, operation: Uuid, project: FileIdentity, session: FileIdentity, + format_version: u32, ) -> Result<(), GfError> { for name in root.child_names().map_err(storage)? { let Some(text) = name.to_str() else { continue }; @@ -8661,13 +8902,11 @@ fn cleanup_authenticated_control_temps( let mut file = root.open_child_file(&name).map_err(storage)?; let body = read_bounded_limit(&mut file, MAX_SHAPE_CONTROL_BYTES)?; let authenticated = serde_json::from_slice::(&body).is_ok_and(|value| { - value.format_version == FORMAT_VERSION - && value.operation_uuid == operation + value.operation_uuid == operation && value.project_identity.matches(project) && value.session_identity.matches(session) }) || serde_json::from_slice::(&body).is_ok_and(|value| { - value.format_version == FORMAT_VERSION - && value.operation_uuid == operation + value.operation_uuid == operation && value.project_identity.matches(project) && value.session_identity.matches(session) }) || serde_json::from_slice::(&body) @@ -8682,14 +8921,12 @@ fn cleanup_authenticated_control_temps( && value.session_identity.matches(session) }) || serde_json::from_slice::(&body).is_ok_and(|value| { - value.format_version == FORMAT_VERSION - && value.operation_uuid == operation + value.operation_uuid == operation && value.project_identity.matches(project) && value.session_identity.matches(session) }) || serde_json::from_slice::(&body).is_ok_and(|value| { - value.format_version == FORMAT_VERSION - && value.operation_uuid == operation + value.operation_uuid == operation && value.project_identity.matches(project) && value.session_identity.matches(session) }) @@ -8698,6 +8935,16 @@ fn cleanup_authenticated_control_temps( && value.project_identity.matches(project) && value.session_identity.matches(session) }); + if authenticated { + // A complete intent can also deserialize as a versionless receipt. + // Pin its version before that structural overlap can authorize cleanup. + let control: serde_json::Value = serde_json::from_slice(&body).map_err(storage)?; + if let Some(version) = control.get("format_version") + && version.as_u64() != Some(u64::from(format_version)) + { + return Err(storage("temporary control version differs from session")); + } + } if authenticated && file_link_count(&file).map_err(storage)? == 1 { let identity = file_identity(&file).map_err(storage)?; drop(file); @@ -8935,6 +9182,7 @@ fn remove_unrecorded_artifact( name: &str, kind: ConstructionChunkKind, rows: u64, + codec: DetailCodec, ) -> Result<(), GfError> { let file = match root.open_child_file(OsStr::new(name)) { Ok(file) => file, @@ -8990,10 +9238,13 @@ fn remove_unrecorded_artifact( let expected_bytes = expected_records .checked_mul(u64::try_from(width).map_err(storage)?) .ok_or_else(|| storage("unrecorded fixed run byte count overflow"))?; - if file.metadata().map_err(storage)?.len() != expected_bytes { + if !(codec == DetailCodec::Compact + && matches!(width, NODE_DETAIL_WIDTH | EDGE_DETAIL_WIDTH)) + && file.metadata().map_err(storage)?.len() != expected_bytes + { return Err(storage("unrecorded fixed run row count changed")); } - validate_sorted_run(file, width)?; + validate_sorted_run(file, width, codec, expected_records)?; } unlink_writer_capability(root, name, None)?; root.unlink_child_if_identity(OsStr::new(name), identity) @@ -9001,18 +9252,32 @@ fn remove_unrecorded_artifact( root.sync().map_err(storage) } -fn validate_sorted_run(file: File, width: usize) -> Result<(), GfError> { +fn validate_sorted_run( + file: File, + width: usize, + codec: DetailCodec, + expected_records: u64, +) -> Result<(), GfError> { let releasing = graphforge_filesystem::FileCacheReleasingReader::new(file).map_err(storage)?; let mut reader = BufReader::with_capacity(BLOCK_BYTES, releasing); let mut block = vec![0_u8; BLOCK_BYTES]; let mut pending = Vec::new(); let mut previous: Option> = None; + let mut detail = if matches!(width, NODE_DETAIL_WIDTH | EDGE_DETAIL_WIDTH) { + Some(DetailValidator::new(codec, width).map_err(storage)?) + } else { + None + }; let validated = (|| -> Result<(), GfError> { loop { let count = reader.read(&mut block).map_err(storage)?; if count == 0 { break; } + if let Some(detail) = detail.as_mut() { + detail.consume(&block[..count]).map_err(storage)?; + continue; + } pending.extend_from_slice(&block[..count]); let complete = pending.len() / width * width; for record in pending[..complete].chunks_exact(width) { @@ -9045,6 +9310,11 @@ fn validate_sorted_run(file: File, width: usize) -> Result<(), GfError> { } pending.drain(..complete); } + if let Some(detail) = detail.as_ref() + && detail.finish().map_err(storage)? != expected_records + { + return Err(storage("unrecorded detail row count changed")); + } if !pending.is_empty() { return Err(storage("unrecorded fixed run has truncated tail")); } @@ -9074,7 +9344,12 @@ fn authenticated_receipt_artifact_count( .open_child_file(OsStr::new(&receipt_name(sequence))) .map_err(storage)?; let receipt: ConstructionChunkReceipt = decode_bounded(&mut file)?; - validate_receipt_semantics(&receipt, sequence, checkpoint.budgets)?; + validate_receipt_semantics( + &receipt, + sequence, + checkpoint.budgets, + DetailCodec::from_version(checkpoint.format_version).map_err(storage)?, + )?; if receipt.kind == ConstructionChunkKind::Node && saw_edge { return Err(storage("node receipt follows edge receipt")); } @@ -9195,6 +9470,7 @@ fn hex(bytes: &[u8]) -> String { #[cfg(test)] mod tests { + include!("construction_detail_tests.rs"); use std::sync::Arc; use arrow::array::{BinaryArray, FixedSizeBinaryArray, Int64Array, StringArray}; @@ -10937,12 +11213,13 @@ mod tests { inject_shape_cleanup_failures(true, true); let mut never_cancelled = || false; - let error = copy_authenticated_run::( + let error = copy_authenticated_run_with_codec::( &session.root, &receipt.details, "failed-details.run", &mut never_cancelled, &mut session.checkpoint.evidence, + Some(DetailCodec::from_version(session.checkpoint.format_version).unwrap()), ) .unwrap_err() .to_string(); @@ -11027,12 +11304,13 @@ mod tests { inject_shape_cleanup_failures(false, true); } inject_shape_publication_failure(point); - let error = copy_authenticated_run::( + let error = copy_authenticated_run_with_codec::( &session.root, &receipt.details, &output, &mut || false, &mut session.checkpoint.evidence, + Some(DetailCodec::from_version(session.checkpoint.format_version).unwrap()), ) .unwrap_err() .to_string(); diff --git a/crates/graphforge-storage/src/graph_construction_encoding.rs b/crates/graphforge-storage/src/graph_construction_encoding.rs index 7484ebfc..23f7936e 100644 --- a/crates/graphforge-storage/src/graph_construction_encoding.rs +++ b/crates/graphforge-storage/src/graph_construction_encoding.rs @@ -14,6 +14,7 @@ use std::sync::Arc; #[cfg(test)] use std::cell::RefCell; +use crate::construction_detail_codec::DetailCodec; use crate::construction_directory::ConstructionDirectory as StableDirectory; use arrow::array::{ Array, ArrayRef, BooleanArray, FixedSizeBinaryArray, ListArray, StringArray, @@ -541,6 +542,7 @@ pub(crate) fn inventory_authority_sha256( #[allow(clippy::too_many_arguments, clippy::too_many_lines)] pub(crate) fn encode( source: &StableDirectory, + detail_codec: DetailCodec, shape: &ConstructionShape, generation: u64, ontology_mode: OntologyMode, @@ -721,6 +723,7 @@ pub(crate) fn encode( let v4 = encode_nodes( source, + detail_codec, &output, shape, shape_outputs, @@ -791,6 +794,7 @@ pub(crate) fn encode( } encode_edges( source, + detail_codec, &output, shape, shape_outputs, @@ -1005,6 +1009,7 @@ fn property_projections( #[allow(clippy::too_many_arguments, clippy::too_many_lines)] fn encode_nodes( source: &StableDirectory, + detail_codec: DetailCodec, output: &StableDirectory, shape: &ConstructionShape, shape_outputs: &[ArtifactReceipt], @@ -1078,8 +1083,13 @@ fn encode_nodes( ) }) .transpose()?; - let mut details = - FixedReader::::open(source, shape_outputs, details_name, cache_window)?; + let mut details = FixedReader::::open_with_codec( + source, + shape_outputs, + details_name, + cache_window, + Some(detail_codec), + )?; let rows_per_window = budgets .max_batch_rows .min((budgets.max_batch_bytes / 128).max(1)); @@ -1361,6 +1371,7 @@ fn encode_node_properties( #[allow(clippy::too_many_arguments, clippy::too_many_lines)] fn encode_edges( source: &StableDirectory, + detail_codec: DetailCodec, output: &StableDirectory, shape: &ConstructionShape, shape_outputs: &[ArtifactReceipt], @@ -1391,8 +1402,13 @@ fn encode_edges( &shape.identities, cache_window, )?; - let mut details = - FixedReader::::open(source, shape_outputs, details_name, cache_window)?; + let mut details = FixedReader::::open_with_codec( + source, + shape_outputs, + details_name, + cache_window, + Some(detail_codec), + )?; let mut endpoints = FixedReader::::open( source, shape_outputs, @@ -2606,6 +2622,7 @@ fn install_json( } struct FixedReader { + detail_codec: Option, reader: BufReader>, counter: IoCounter, identity: graphforge_filesystem::FileIdentity, @@ -2621,14 +2638,27 @@ impl FixedReader { outputs: &[ArtifactReceipt], name: &str, cache_window: std::num::NonZeroU64, + ) -> Result { + Self::open_with_codec(root, outputs, name, cache_window, None) + } + + fn open_with_codec( + root: &StableDirectory, + outputs: &[ArtifactReceipt], + name: &str, + cache_window: std::num::NonZeroU64, + detail_codec: Option, ) -> Result { let authenticated = open_authenticated_shape_source(root, outputs, name)?; let file = authenticated.file; - if file.metadata().map_err(storage)?.len() % N as u64 != 0 { + if detail_codec != Some(DetailCodec::Compact) + && file.metadata().map_err(storage)?.len() % N as u64 != 0 + { return Err(storage("fixed-width construction stream is truncated")); } let counter = IoCounter::default(); Ok(Self { + detail_codec, reader: BufReader::with_capacity( COPY_BUFFER_BYTES, CountingInput { @@ -2651,22 +2681,34 @@ impl FixedReader { } fn next(&mut self) -> Result, GfError> { - let mut record = [0_u8; N]; - let mut read = 0; - while read < N { - let amount = self.reader.read(&mut record[read..]).map_err(storage)?; - if amount == 0 { - if read == 0 { - return Ok(None); + let record = if let Some(codec) = self.detail_codec { + let Some(record) = codec.read::(&mut self.reader).map_err(storage)? else { + return Ok(None); + }; + record + } else { + let mut record = [0_u8; N]; + let mut read = 0; + while read < N { + let amount = self.reader.read(&mut record[read..]).map_err(storage)?; + if amount == 0 { + if read == 0 { + return Ok(None); + } + return Err(storage("fixed-width construction stream is truncated")); } - return Err(storage("fixed-width construction stream is truncated")); + read += amount; } - read += amount; - } - self.digest.update(record); + record + }; + let wire = match self.detail_codec { + Some(codec) => codec.bytes(&record).map_err(storage)?, + None => &record, + }; + self.digest.update(wire); self.consumed_bytes = self .consumed_bytes - .checked_add(u64::try_from(N).map_err(storage)?) + .checked_add(u64::try_from(wire.len()).map_err(storage)?) .ok_or_else(|| storage("fixed-width source byte count overflow"))?; Ok(Some(record)) } diff --git a/crates/graphforge-storage/src/lib.rs b/crates/graphforge-storage/src/lib.rs index 296c58d7..39e6c2b9 100644 --- a/crates/graphforge-storage/src/lib.rs +++ b/crates/graphforge-storage/src/lib.rs @@ -55,6 +55,7 @@ pub use graph_projection::{ materialize_graph_projection, materialize_portable_graph_tree_projection, }; +mod construction_detail_codec; pub mod graph_construction; mod graph_construction_encoding; pub use graph_construction::{ diff --git a/docs/adr/0024-storage-format-exceptions.md b/docs/adr/0024-storage-format-exceptions.md index 047615fc..49a6eb99 100644 --- a/docs/adr/0024-storage-format-exceptions.md +++ b/docs/adr/0024-storage-format-exceptions.md @@ -107,3 +107,37 @@ of unknown Parquet schemas is not. | Encode each small mutation as Parquet | Reintroduces the rewrite amplification ADR 0019 was designed to avoid. | | Treat compiled ontology Parquet as source authority | Creates two competing ontology definitions and makes checksum mismatch ambiguous. | | Promise transparent pre-v1 migration | The old GFDR payload is not losslessly recoverable, and the compiled snapshot lacks an enforced format-version gate. | + +### Private construction detail runs retain session-bound codecs + +Construction checkpoints and intents bind the private detail codec. New sessions +use version 7: a node record contains its 16-byte UUID, a one-byte UTF-8 byte +length, and exactly that many label bytes; an edge record contains its edge, +source and target UUIDs (48 bytes), the length byte and exact route bytes. Names +remain nonempty and at most 255 bytes. Records are strictly ordered by UUID. +There is no dictionary or substitution of runtime or ontology IDs for names. + +Version 6 sessions continue their original 272-byte node and 304-byte edge +records, including zero-filled name padding, throughout append, resume, shaping +and publication replay. The initial version is chosen before writing a new +checkpoint; resume does not upgrade or relabel authenticated artifacts. Unknown +versions and disagreement between a session and its controls fail closed. +Incomplete, unauthenticated control temporaries retain the existing recovery +policy and do not authorize deleting another writer's file. +If no checkpoint was installed before a crash, a complete initial checkpoint +temporary may select its codec only after matching the admitted parent, catalog, +budgets and session identity. Its contents are not promoted. Conflicting or +noninitial candidates are rejected. + +Both codecs use the same bounded merge fan-in and maximum in-memory record +size. Compact byte counts come from actual encoded names; authenticated row +counts come from decoding records, not dividing file size by a fixed width. +Identity, endpoint, canonical Parquet, portable and published ordinal formats +are unchanged. + +The equal-input regression uses 1,024 nodes and 1,024 edges in eight chunks per +kind, crossing three merge levels with fan-in two. Detail EOF falls from 589,824 +to 74,752 bytes. Filesystem allocation and complete construction peak are +measured separately because block rounding and concurrent representations also +matter. This establishes a bounded representation improvement; it does not +establish S26 capacity admission, which requires the shared host projection.