diff --git a/crates/tracedecay-graph-db/src/generation.rs b/crates/tracedecay-graph-db/src/generation.rs index 39e63cc578..01a16c482b 100644 --- a/crates/tracedecay-graph-db/src/generation.rs +++ b/crates/tracedecay-graph-db/src/generation.rs @@ -49,6 +49,7 @@ pub use identity::{ #[cfg(test)] pub(crate) use recovered::recovered_generation_digest_chunked; pub(crate) use recovered::recovered_generation_digest_from_database; +pub(crate) use recovered::recovered_relation_lanes; pub(crate) use replay::InlineOnlyGraphGenerationManifestProvider; use replay::validate_sealed_replay; pub use replay::{ @@ -157,6 +158,10 @@ pub struct GraphGenerationManifestIdentity { /// it. Re-validated against `dependencies` on every read and invisible to /// equality and clones. digest_memo: DependencyClosureDigestMemo, + /// The namespace this identity's rows are stored under when another + /// projection sealed them: a sibling scope's base, read and proven + /// under this projection. `None` for rows stored under their own. + stored_namespace: Option, } impl GraphGenerationManifestIdentity { @@ -177,9 +182,20 @@ impl GraphGenerationManifestIdentity { watermark, dependencies, digest_memo: DependencyClosureDigestMemo::default(), + stored_namespace: None, } } + /// This identity over rows stored under `physical`, the namespace its + /// generation was sealed under, possibly by another projection. + pub(crate) fn stored_under(mut self, physical: GraphNamespace) -> Result { + self.stored_namespace = None; + if physical != self.physical_namespace()? { + self.stored_namespace = Some(physical); + } + Ok(self) + } + pub fn dependency_closure_digest( &self, check: &dyn Fn() -> Result<(), GraphDbError>, @@ -188,6 +204,9 @@ impl GraphGenerationManifestIdentity { } pub(crate) fn physical_namespace(&self) -> Result { + if let Some(stored) = &self.stored_namespace { + return Ok(stored.clone()); + } physical_namespace( &self.projection.namespace, &self.projection.projection, @@ -592,6 +611,7 @@ impl GraphGenerationManifest { .digest_memo .dependency_closure .propagated(&self.dependencies), + stored_namespace: None, } } diff --git a/crates/tracedecay-graph-db/src/generation/recovered.rs b/crates/tracedecay-graph-db/src/generation/recovered.rs index d2fefc7eb7..bb7e2ae2a1 100644 --- a/crates/tracedecay-graph-db/src/generation/recovered.rs +++ b/crates/tracedecay-graph-db/src/generation/recovered.rs @@ -20,9 +20,9 @@ use crate::{GraphDbError, GraphNamespace}; use super::{ CheckedDigestWriter, CheckedVecWriter, GraphEntityRef, GraphGenerationManifestIdentity, - GraphGenerationRelation, GraphProjectionIdentity, frame_length_headers, - physical_namespace_projection_map, recovered_entity_ref, write_canonical_row_frame, - write_generation_identity_frames, + GraphGenerationRelation, GraphProjectionIdentity, RowLanes, frame_length_headers, + physical_namespace_projection_map, recovered_entity_ref, row_frame_lanes, + write_canonical_row_frame, write_generation_identity_frames, }; /// Rows per encode chunk. Sized so one chunk is a few milliseconds of decode @@ -144,6 +144,46 @@ pub(crate) fn recovered_generation_digest_chunked( Ok((encode_lowercase_hex(&digest.finalize()), canonical_bytes)) } +/// Streams each stored relation of `identity`'s generation to `emit` with +/// the row-sum lanes of its frame as `identity` recovers it, in identity +/// order. A relation's frame names its endpoints' projection, so the same +/// stored rows hash differently under each projection that reads them. +pub(crate) fn recovered_relation_lanes( + database: &GrafeoDB, + identity: &GraphGenerationManifestIdentity, + check: &dyn Fn() -> Result<(), GraphDbError>, + emit: &mut dyn FnMut(&str, RowLanes) -> Result<(), GraphDbError>, +) -> Result<(), GraphDbError> { + let store = database.graph_store(); + let relations = projection_relation_nodes_sorted_checked( + database, + &identity.physical_namespace()?, + &identity.projection.projection, + check, + )?; + let namespace_projection = physical_namespace_projection_map(identity)?; + let mut canonical = CheckedVecWriter::new(check, MAX_GRAPH_REPLAY_SOURCE_BYTES_V1)?; + let mut endpoints = EndpointIdentityCache::default(); + let mut endpoint_refs = HashMap::new(); + for (sorted_identity, locator) in &relations { + check()?; + let relation = decode_sorted_relation( + store.as_ref(), + sorted_identity, + *locator, + &namespace_projection, + &mut endpoints, + &mut endpoint_refs, + )?; + let bytes = canonical.encode(&relation, "recovered generation relation")?; + emit( + sorted_identity.as_str(), + row_frame_lanes("relation", bytes)?, + )?; + } + Ok(()) +} + /// The single-pass stream for generations at or below one chunk: one decoded /// row resident at a time, every frame hashed as it is encoded. #[tracing::instrument( diff --git a/crates/tracedecay-graph-db/src/lib.rs b/crates/tracedecay-graph-db/src/lib.rs index c207ab9095..65d6df6070 100644 --- a/crates/tracedecay-graph-db/src/lib.rs +++ b/crates/tracedecay-graph-db/src/lib.rs @@ -99,7 +99,7 @@ pub use runtime::{GraphDb, GraphDbRuntimeState, GraphServingEnginePin, GraphSnap pub use schema::graph_stable_identity; pub use sealed_layer::{ GraphLayeredRowSpill, GraphLayeredRowsV1, GraphSealedBaseAbsenceV1, GraphSealedBaseV1, - LayeredGraphGeneration, + GraphSiblingSealedBaseV1, LayeredGraphGeneration, }; pub use sealed_store::{SealedStoreCensusV1, census_sealed_store}; diff --git a/crates/tracedecay-graph-db/src/registry/publication.rs b/crates/tracedecay-graph-db/src/registry/publication.rs index 5d7d114080..27e2284aaa 100644 --- a/crates/tracedecay-graph-db/src/registry/publication.rs +++ b/crates/tracedecay-graph-db/src/registry/publication.rs @@ -1306,6 +1306,32 @@ impl GraphDbRegistry { operation.database().layered_row_spill(projection, base) } + /// The sealed cold bases other scopes of this store serve for the same + /// projector, which a scope with no parent graph may layer over. + pub fn sibling_sealed_bases( + &self, + registration: GraphDbRegistration, + projection: &GraphProjectionIdentity, + check: &dyn Fn() -> Result<(), GraphDbError>, + ) -> Result, GraphDbError> { + let operation = self.registered_operation(registration)?; + operation.database().sibling_sealed_bases(projection, check) + } + + /// A row spill for a delta over a sibling scope's base, which it pins + /// until it seals. + pub fn sibling_layered_row_spill( + &self, + registration: GraphDbRegistration, + projection: GraphProjectionIdentity, + sibling: crate::GraphSiblingSealedBaseV1, + ) -> Result { + let operation = self.registered_operation(registration)?; + operation + .database() + .sibling_layered_row_spill(projection, sibling) + } + /// Publishes through an already-issued, registry-validated graph lease. /// /// The caller retains the exact operation lease through the publication; diff --git a/crates/tracedecay-graph-db/src/row_index.rs b/crates/tracedecay-graph-db/src/row_index.rs index a32006b13f..a13d6026d4 100644 --- a/crates/tracedecay-graph-db/src/row_index.rs +++ b/crates/tracedecay-graph-db/src/row_index.rs @@ -292,13 +292,26 @@ impl RowIndex { count: u64, key: RowKey, ) -> Result, GraphDbError> { + Ok(self + .position(first, count, key)? + .map(|(_, lanes, first_word, second_word)| (lanes, first_word, second_word))) + } + + fn position( + &self, + first: u64, + count: u64, + key: RowKey, + ) -> Result, GraphDbError> { let (mut low, mut high) = (0_u64, count); while low < high { let middle = low + (high - low) / 2; let (found, lanes, first_word, second_word) = record_parts(&self.record(first + middle)?); match found.cmp(&key) { - std::cmp::Ordering::Equal => return Ok(Some((lanes, first_word, second_word))), + std::cmp::Ordering::Equal => { + return Ok(Some((first + middle, lanes, first_word, second_word))); + } std::cmp::Ordering::Less => low = middle + 1, std::cmp::Ordering::Greater => high = middle, } @@ -322,6 +335,50 @@ impl RowIndex { .map(|(lanes, from, to)| IndexedRelation { lanes, from, to })) } + /// Writes this index to `path` with every relation's lanes replaced by + /// the lanes `relane` emits for it, entity records and endpoint ordinals + /// unchanged: the index of the same rows read under another projection. + /// `relane` streams each recorded relation exactly once, in strictly + /// increasing identity order, so only one record is resident at a time. + pub(crate) fn write_relaned( + &self, + path: &Path, + relane: impl FnOnce( + &mut dyn FnMut(&str, RowLanes) -> Result<(), GraphDbError>, + ) -> Result<(), GraphDbError>, + ) -> Result<(), GraphDbError> { + std::fs::copy(&self.path, path).map_err(|error| index_io("copy", error))?; + let mut target = std::fs::OpenOptions::new() + .write(true) + .open(path) + .map_err(|error| index_io("open", error))?; + let mut relaned = 0_u64; + let mut previous: Option = None; + relane(&mut |identity, lanes| { + if previous.as_deref().is_some_and(|last| last >= identity) { + return Err(corrupt("relanes relations out of identity order")); + } + previous = Some(identity.to_owned()); + let Some((position, ..)) = + self.position(self.entities, self.relations, row_key("relation", identity))? + else { + return Err(corrupt("relanes a relation it does not record")); + }; + target + .seek(SeekFrom::Start(HEADER_BYTES + position * RECORD_BYTES + 16)) + .map_err(|error| index_io("seek", error))?; + target + .write_all(&lanes_bytes(lanes)) + .map_err(|error| index_io("write", error))?; + relaned += 1; + Ok(()) + })?; + if relaned != self.relations { + return Err(corrupt("relanes a different relation set than it records")); + } + target.sync_all().map_err(|error| index_io("sync", error)) + } + /// The sum of every recorded row: equal to the generation's row sum /// exactly when the index records the generation's rows. pub(crate) fn row_sum( diff --git a/crates/tracedecay-graph-db/src/sealed_layer.rs b/crates/tracedecay-graph-db/src/sealed_layer.rs index 0c0927f138..dc6621edb3 100644 --- a/crates/tracedecay-graph-db/src/sealed_layer.rs +++ b/crates/tracedecay-graph-db/src/sealed_layer.rs @@ -92,13 +92,14 @@ impl SealedBaseReceiptV1 { &self, projection: &GraphProjectionIdentity, ) -> Result { - Ok(GraphGenerationManifestIdentity::new( + GraphGenerationManifestIdentity::new( projection.clone(), GraphGenerationId::new(self.generation.clone())?, SourceGeneration::new(self.source_generation.clone())?, GraphWatermark::new(self.watermark.clone())?, Vec::new(), - )) + ) + .stored_under(GraphNamespace::new(self.physical_namespace.clone())?) } } @@ -286,6 +287,64 @@ impl GraphSealedBaseV1 { })) } + /// This base as `projection` reads it: the same stored rows, with each + /// relation's endpoints under `projection`, which changes its frame. + /// `engine` serves the base's proven rows; their relation lanes under + /// `projection` replace those of the index pinned in `directory`, and + /// the row sum and digest follow from it, exactly what a cold build of + /// these rows under `projection` records. Only the pinned files are + /// read, so the sibling may retire its own meanwhile. + fn reattested( + &self, + projection: &GraphProjectionIdentity, + engine: &GraphDb, + directory: &Path, + check: &dyn Fn() -> Result<(), GraphDbError>, + ) -> Result { + let base = &self.inner; + let index_path = directory.join(LAYERED_BASE_ROW_INDEX_FILE); + let entity_rows = directory.join(LAYERED_BASE_ENTITY_ROWS_FILE); + let identity = GraphGenerationManifestIdentity::new( + projection.clone(), + base.identity.generation.clone(), + base.identity.source_generation.clone(), + base.identity.watermark.clone(), + Vec::new(), + ) + .stored_under(base.physical_namespace.clone())?; + let staged = index_path.with_extension("reattested"); + let relaned = RowIndex::open(&index_path)?.write_relaned(&staged, |emit| { + engine.read_intact(&NeverCancelled, |native| { + crate::generation::recovered_relation_lanes(native, &identity, check, emit) + }) + }); + let _ = engine.hibernate_if_lazy_when_idle(); + relaned?; + std::fs::rename(&staged, &index_path) + .map_err(|error| layered_io("reattested row index install", error))?; + let index = RowIndex::open(&index_path)?; + let row_sum = index.row_sum(check)?; + let recovered_digest = recovered_digest_from_row_sum(&identity, row_sum, check)?; + Ok(Self { + inner: Arc::new(SealedBaseInner { + physical_namespace: base.physical_namespace.clone(), + recovered_digest: recovered_digest.as_str().to_owned(), + entities: base.entities, + relations: base.relations, + row_sum, + container: base.container.clone(), + attachment: base.attachment.clone(), + index, + entity_row_offsets: EntityRowOffsets::open( + &directory.join(LAYERED_BASE_ENTITY_ROW_OFFSETS_FILE), + &entity_rows, + )?, + entity_rows, + identity, + }), + }) + } + /// The base's graph generation. #[must_use] pub fn generation(&self) -> &GraphGenerationId { @@ -312,6 +371,31 @@ fn generation_relation( ) } +/// A sealed cold generation of the same projector that another scope of +/// this graph store serves, with the engine serving it: a linked worktree's +/// candidate base. +pub struct GraphSiblingSealedBaseV1 { + pub(crate) base: GraphSealedBaseV1, + pub(crate) engine: Arc, +} + +impl std::fmt::Debug for GraphSiblingSealedBaseV1 { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("GraphSiblingSealedBaseV1") + .field("base", &self.base) + .field("namespace", &self.base.inner.identity.projection.namespace) + .finish_non_exhaustive() + } +} + +impl GraphSiblingSealedBaseV1 { + #[must_use] + pub fn base(&self) -> &GraphSealedBaseV1 { + &self.base + } +} + /// Rows a refresh changes relative to a sealed base, spilled like a cold /// generation's rows, plus the base identities it removes. /// @@ -321,6 +405,9 @@ fn generation_relation( pub struct GraphLayeredRowSpill { spill: GraphGenerationRowSpill, base: GraphSealedBaseV1, + /// The engine serving a sibling scope's base, whose relations the + /// finished layer re-attests under its own projection. + sibling: Option>, hidden_entities: BTreeSet, hidden_relations: BTreeSet, } @@ -347,6 +434,35 @@ impl GraphLayeredRowSpill { "a layered generation names a projection its base does not serve", )); } + Self::pin(directory, projection, base, None) + } + + /// A delta of `projection` over `base`, a generation another scope of + /// the same projector sealed and `engine` serves: a linked worktree at + /// the same tree layers over its sibling's graph instead of sealing the + /// same rows again. + pub(crate) fn create_over_sibling( + directory: PathBuf, + projection: GraphProjectionIdentity, + base: GraphSealedBaseV1, + engine: Arc, + ) -> Result { + if projection.projection != base.inner.identity.projection.projection + || projection.namespace == base.inner.identity.projection.namespace + { + return Err(GraphDbError::invalid( + "a sibling base must serve the same projector under another namespace", + )); + } + Self::pin(directory, projection, base, Some(engine)) + } + + fn pin( + directory: PathBuf, + projection: GraphProjectionIdentity, + base: GraphSealedBaseV1, + sibling: Option>, + ) -> Result { let spill = GraphGenerationRowSpill::create_layered(directory, projection)?; let pinned_container = spill.directory().join(LAYERED_BASE_CONTAINER_FILE); std::fs::hard_link(&base.inner.container, &pinned_container) @@ -380,6 +496,7 @@ impl GraphLayeredRowSpill { Ok(Self { spill, base, + sibling, hidden_entities: BTreeSet::new(), hidden_relations: BTreeSet::new(), }) @@ -492,6 +609,14 @@ impl GraphLayeredRowSpill { check: &dyn Fn() -> Result<(), GraphDbError>, ) -> Result { check()?; + if let Some(engine) = self.sibling.take() { + self.base = self.base.reattested( + &identity.projection, + &engine, + self.spill.directory(), + check, + )?; + } if identity.projection != self.base.inner.identity.projection { return Err(GraphDbError::invalid( "a layered generation names a projection its base does not serve", @@ -505,6 +630,7 @@ impl GraphLayeredRowSpill { let Self { spill, base, + sibling: _, hidden_entities, hidden_relations, } = self; diff --git a/crates/tracedecay-graph-db/src/sealed_store.rs b/crates/tracedecay-graph-db/src/sealed_store.rs index 64b62f875e..3a35f1158c 100644 --- a/crates/tracedecay-graph-db/src/sealed_store.rs +++ b/crates/tracedecay-graph-db/src/sealed_store.rs @@ -69,8 +69,8 @@ use crate::schema::{ relation_type_for_kind, }; use crate::sealed_layer::{ - GraphLayeredRowSpill, GraphSealedBaseAbsenceV1, GraphSealedBaseV1, LayeredGraphGeneration, - LayeredReads, SealedBaseFilesV1, SealedBaseReceiptV1, SealedLayer, + GraphLayeredRowSpill, GraphSealedBaseAbsenceV1, GraphSealedBaseV1, GraphSiblingSealedBaseV1, + LayeredGraphGeneration, LayeredReads, SealedBaseFilesV1, SealedBaseReceiptV1, SealedLayer, }; use crate::state::{ EndpointIdentityCache, latest_projection, load_relation_by_locator_cached, @@ -1451,6 +1451,54 @@ impl GraphDb { GraphLayeredRowSpill::create(self.row_spill_directory()?, projection, base) } + /// The cold bases a first generation of `projection` may layer over: + /// every flat sealed generation of the same projector this store has + /// installed for another scope. Linked worktrees share one store, so a + /// worktree at its sibling's tree finds that sibling's graph here. + pub(crate) fn sibling_sealed_bases( + &self, + projection: &GraphProjectionIdentity, + check: &dyn Fn() -> Result<(), GraphDbError>, + ) -> Result, GraphDbError> { + let siblings = self + .inner + .sealed_generations + .read() + .map_err(|_| GraphDbError::unavailable("sealed generation map lock is poisoned"))? + .iter() + .filter(|(locator, store)| { + locator.projection.projection == projection.projection + && locator.projection.namespace != projection.namespace + && store.layer.is_none() + }) + .map(|(_, store)| Arc::clone(store)) + .collect::>(); + let mut bases = Vec::new(); + for store in siblings { + if let Ok(base) = store.sealed_base(check)? { + bases.push(GraphSiblingSealedBaseV1 { + base, + engine: Arc::clone(&store.database), + }); + } + } + Ok(bases) + } + + /// A row spill for a delta of `projection` over a sibling scope's base. + pub(crate) fn sibling_layered_row_spill( + &self, + projection: GraphProjectionIdentity, + sibling: GraphSiblingSealedBaseV1, + ) -> Result { + GraphLayeredRowSpill::create_over_sibling( + self.row_spill_directory()?, + projection, + sibling.base, + sibling.engine, + ) + } + /// Opens an existing sealed store for `identity` without building one. /// /// Used on the recovery path: a matching artifact on disk is installed diff --git a/crates/tracedecay-store-runtime/src/session_registry/code_graph.rs b/crates/tracedecay-store-runtime/src/session_registry/code_graph.rs index 3ea0885e23..2883e930f0 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/code_graph.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/code_graph.rs @@ -1906,24 +1906,34 @@ impl RetainedCodeGraphRuntimeV1 { check: &dyn Fn() -> std::result::Result<(), GraphDbError>, ) -> std::result::Result<(GraphGenerationRows, Option), GraphDbError> { - let layered_spill = |parent: &CodeGenerationId| { - let parent = tracedecay_code_index::graph_projection::code_graph_generation_id( - parent, - projector_revision, - ) - .map_err(|error| GraphDbError::invalid(error.to_string()))?; - match self.graph_registry.sealed_generation_base( - registration(), - projection.clone(), - parent, - check, - )? { - Ok(base) => self - .graph_registry - .layered_row_spill(registration(), projection.clone(), base) - .map(Ok), - Err(absence) => Ok(Err(absence)), + let layered_spill = |base: super::code_graph_manifest::LayeredBaseV1<'_>| match base { + super::code_graph_manifest::LayeredBaseV1::Parent(parent) => { + let parent = tracedecay_code_index::graph_projection::code_graph_generation_id( + parent, + projector_revision, + ) + .map_err(|error| GraphDbError::invalid(error.to_string()))?; + match self.graph_registry.sealed_generation_base( + registration(), + projection.clone(), + parent, + check, + )? { + Ok(base) => self + .graph_registry + .layered_row_spill(registration(), projection.clone(), base) + .map(Ok), + Err(absence) => Ok(Err(absence)), + } } + super::code_graph_manifest::LayeredBaseV1::Sibling(sibling) => self + .graph_registry + .sibling_layered_row_spill(registration(), projection.clone(), sibling) + .map(Ok), + }; + let sibling_bases = || { + self.graph_registry + .sibling_sealed_bases(registration(), projection, check) }; let cold_spill = || { self.graph_registry @@ -1937,6 +1947,7 @@ impl RetainedCodeGraphRuntimeV1 { projection.clone(), projector_revision, &layered_spill, + &sibling_bases, &cold_spill, admit, check, diff --git a/crates/tracedecay-store-runtime/src/session_registry/code_graph/layered_refresh_tests.rs b/crates/tracedecay-store-runtime/src/session_registry/code_graph/layered_refresh_tests.rs index 710c8c5a00..2ff64920ff 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/code_graph/layered_refresh_tests.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/code_graph/layered_refresh_tests.rs @@ -323,6 +323,35 @@ impl RefreshFixture { fixture } + /// A linked worktree of this fixture's repository at its `main` tree, + /// indexed into its own scope of the same code-index store. + fn linked(&self, name: &str) -> Self { + let project_root = self.root.join(name); + git( + &self.project_root, + &[ + "worktree", + "add", + "-q", + "-b", + name, + project_root.to_str().expect("linked worktree path"), + "main", + ], + ); + let canonical_project = project_root.canonicalize().expect("canonical linked root"); + Self { + scoped_store: scoped_code_index_store_root( + &self.root.join("code-index-store"), + &canonical_project, + ), + root: self.root.clone(), + project_root, + canonical_project, + project_id: self.project_id.clone(), + } + } + fn commit(&self, message: &str) { git(&self.project_root, &["add", "-A"]); git(&self.project_root, &["commit", "-qm", message]); @@ -695,6 +724,141 @@ async fn small_refreshes_seal_deltas_that_serve_like_their_cold_builds() { drop((base_runtime, child_runtime, grandchild_runtime)); } +/// The sealed receipt of `generation` in `profile` sealed as `form`. +fn receipt_in_form(profile: &Path, generation: &str, form: &str) -> (PathBuf, serde_json::Value) { + sealed_receipts(profile) + .into_iter() + .find(|(_, receipt)| receipt["generation"] == generation && receipt["form"] == form) + .unwrap_or_else(|| panic!("no {form} sealed receipt for {generation}")) +} + +/// Fails when a linked worktree at its sibling's tree seals the whole graph +/// again (#2402): its first generation must layer over the sibling's sealed +/// container by hard link with a delta of just its generation marker, and +/// still record the digest, serve the rows, and answer every read exactly as +/// a cold build in its own namespace. Its later edits stay deltas over that +/// base and leave the sibling's graph untouched. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_linked_worktree_at_the_same_tree_layers_over_its_siblings_sealed_graph() { + let temporary = tempfile::tempdir().expect("temporary fixture parent"); + let root = temporary + .path() + .canonicalize() + .expect("canonical fixture root"); + let primary_fixture = RefreshFixture::create(&root, MODULES); + let (primary_source, primary_generation, _, primary_binding) = primary_fixture.seal(); + let (_scope, registry, database) = primary_fixture.open_profile("profile", 45).await; + let primary_runtime = primary_source + .retain(®istry, &database, &primary_generation, primary_binding) + .await + .expect("retain the primary graph runtime"); + let primary = primary_runtime + .publish_verified_snapshot(Arc::new(AtomicBool::new(false))) + .expect("publish the primary graph cold"); + let profile = root.join("profile"); + let (primary_directory, _) = + receipt_in_form(&profile, primary.generation().as_str(), "compact"); + #[cfg(unix)] + let primary_container = file_identity(&primary_directory.join("generation.grafeo")); + #[cfg(not(unix))] + let _ = primary_directory; + let primary_rows = scan_rows(&primary); + + let linked_fixture = primary_fixture.linked("linked"); + let (linked_source, linked_generation, linked_parent, linked_binding) = linked_fixture.seal(); + assert_eq!(linked_parent, None, "a new worktree's first generation"); + assert_ne!(linked_source.worktree, primary_source.worktree); + let linked_runtime = linked_source + .retain( + ®istry, + &database, + &linked_generation, + linked_binding.clone(), + ) + .await + .expect("retain the linked graph runtime"); + let report = + layered_report(&linked_runtime).expect("the linked worktree layers over its sibling"); + assert_eq!(report.reextracted_files, 0, "{report:?}"); + assert_eq!(report.removed_files, 0, "{report:?}"); + let linked = linked_runtime + .publish_verified_snapshot(Arc::new(AtomicBool::new(false))) + .expect("publish the linked worktree's graph"); + assert_ne!(linked.projection(), primary.projection()); + let (linked_directory, receipt) = + receipt_in_form(&profile, linked.generation().as_str(), "layered"); + #[cfg(not(unix))] + let _ = linked_directory; + assert_eq!(receipt["base"]["generation"], primary.generation().as_str()); + #[cfg(unix)] + assert_eq!( + file_identity(&linked_directory.join("base.grafeo")), + primary_container, + "the linked worktree references its sibling's container instead of re-encoding it" + ); + let delta = rows_in(&receipt["row_sum"]) + rows_in(&receipt["base"]["hidden_row_sum"]) + - rows_in(&receipt["base"]["row_sum"]); + assert_eq!( + delta, 1, + "the delta is the linked generation's marker alone" + ); + assert_matches_cold_build( + &linked_fixture, + &linked_source, + &linked_runtime, + linked, + linked_binding, + "profile-linked-cold", + 46, + ) + .await; + + // The linked worktree diverges: still a delta over the shared base, and + // the primary serves exactly what it served. + edit_three_files(&linked_fixture.project_root, MODULES); + linked_fixture.commit("edit three files in the linked worktree"); + let (linked_source, edited_generation, edited_parent, edited_binding) = linked_fixture.seal(); + assert_eq!(edited_parent.as_ref(), Some(&linked_generation)); + let edited_runtime = linked_source + .retain( + ®istry, + &database, + &edited_generation, + edited_binding.clone(), + ) + .await + .expect("retain the edited linked graph runtime"); + let edited = edited_runtime + .publish_verified_snapshot(Arc::new(AtomicBool::new(false))) + .expect("publish the linked worktree's edit"); + let edited_rows = scan_rows(&edited); + assert_layered_over( + &profile, + edited.generation().as_str(), + primary.generation().as_str(), + #[cfg(unix)] + primary_container, + ); + assert_ne!(edited_rows, primary_rows, "the linked edit diverges"); + assert_eq!( + scan_rows(&primary), + primary_rows, + "the primary is untouched" + ); + assert_matches_cold_build( + &linked_fixture, + &linked_source, + &edited_runtime, + edited, + edited_binding, + "profile-linked-edit-cold", + 47, + ) + .await; + + drop((primary, primary_runtime, linked_runtime, edited_runtime)); +} + /// A serving store over `snapshot` whose interactive catalog is warmed from /// `predecessor`, as graph activation warms it. fn warmed_store( diff --git a/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs b/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs index 625437767f..5c224bd7eb 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs @@ -19,7 +19,8 @@ use tracedecay_graph_db::{ GraphBudgetKind, GraphDbError, GraphGenerationManifestProvider, GraphGenerationRowSpill, GraphGenerationRows, GraphLayeredRowSpill, GraphNamespace, GraphProjectionId, GraphProjectionIdentity, GraphProjectorRevision, GraphSealedBaseAbsenceV1, - SealedCodeGenerationReplay, SealedGraphStateDigest, SpilledGraphGeneration, + GraphSiblingSealedBaseV1, SealedCodeGenerationReplay, SealedGraphStateDigest, + SpilledGraphGeneration, }; use tracedecay_runtime_core::resident_memory::ResidentMemoryPressureV1; use tracedecay_store::{GraphProjectionIdentityV1, StoreShardIdV1}; @@ -606,9 +607,19 @@ fn open_seal_from_roots( pub(super) type LayeredRowSpillV1 = Result, GraphDbError>; +/// The sealed graph a layered refresh builds its delta over. +pub(super) enum LayeredBaseV1<'a> { + /// The graph the scope sealed for the seal's parent code generation. + Parent(&'a tracedecay_domain::CodeGenerationId), + /// A cold graph another scope of the same store sealed, the base of a + /// scope's first generation: a linked worktree at its sibling's tree. + Sibling(GraphSiblingSealedBaseV1), +} + /// A seal's graph rows as its publication seals them: a delta over the -/// sealed graph of the seal's parent code generation when `layered_spill` -/// resolves one and the base's inputs admit it, the cold rows otherwise. +/// sealed graph of the seal's parent code generation, or for a seal with no +/// parent over a sibling scope's sealed graph, when `layered_spill` resolves +/// one and the base's inputs admit it, the cold rows otherwise. /// /// A layered attempt that fails for any reason but an interruption is /// reported and replaced by the cold build, which is the authority a @@ -626,7 +637,8 @@ pub(super) fn graph_rows_from_roots( generation: &tracedecay_domain::CodeGenerationId, projection: GraphProjectionIdentity, projector_revision: &GraphProjectorRevision, - layered_spill: &dyn Fn(&tracedecay_domain::CodeGenerationId) -> LayeredRowSpillV1, + layered_spill: &dyn Fn(LayeredBaseV1<'_>) -> LayeredRowSpillV1, + sibling_bases: &dyn Fn() -> Result, GraphDbError>, cold_spill: &dyn Fn() -> Result, admit: &mut dyn FnMut( tracedecay_code_index::production::CodeGraphBuildBoundV1, @@ -641,50 +653,44 @@ pub(super) fn graph_rows_from_roots( check, "code_graph_manifest.graph_rows", )?; + let mut layered = |base: &str, spill: LayeredRowSpillV1| { + layered_graph_rows( + &seal, + generation, + &projection, + projector_revision, + base, + spill, + admit, + check, + ) + }; if let Some(parent) = seal.source.manifest().parent_generation.clone() { - let layered = layered_spill(&parent).and_then(|spill| match spill { - Ok(spill) => with_verified_segments( - &seal.sealed_state_digest, - &seal.routes, - check, - |read_segment| { - build_layered_code_graph_rows( - projection.clone(), - &seal.source, - read_segment, - projector_revision, - spill, - admit, - check, - ) - }, - ) - .map(|built| built.map_err(|decline| format!("{decline:?}"))), - Err(absence) => Ok(Err(format!("{absence:?}"))), - }); - match layered { - Ok(Ok(built)) => return Ok((built.generation.into(), Some(built.report))), - Ok(Err(reason)) => tracing::info!( - event = "code_graph_layered_refresh_declined", - generation = %generation, - parent = %parent, - reason = %reason, - "the refresh seals cold rather than as a delta over its parent's graph" - ), - Err( - error @ (GraphDbError::Cancelled - | GraphDbError::DeadlineExceeded - | GraphDbError::SealSuperseded), - ) => { - return Err(error); + if let Some(built) = layered( + parent.as_str(), + layered_spill(LayeredBaseV1::Parent(&parent)), + )? { + return Ok(built); + } + } else { + let siblings = match sibling_bases() { + Ok(siblings) => siblings, + Err(error) if is_interruption(&error) => return Err(error), + Err(error) => { + tracing::warn!( + event = "code_graph_layered_refresh_unavailable", + generation = %generation, + error = %error, + "sibling sealed graphs could not be listed; sealing the generation cold" + ); + Vec::new() + } + }; + for sibling in siblings { + let base = sibling.base().generation().as_str().to_owned(); + if let Some(built) = layered(&base, layered_spill(LayeredBaseV1::Sibling(sibling)))? { + return Ok(built); } - Err(error) => tracing::warn!( - event = "code_graph_layered_refresh_unavailable", - generation = %generation, - parent = %parent, - error = %error, - "a layered graph refresh could not be built; sealing the generation cold" - ), } } let spilled = spill_verified_seal_graph( @@ -698,6 +704,69 @@ pub(super) fn graph_rows_from_roots( Ok((spilled.into(), None)) } +fn is_interruption(error: &GraphDbError) -> bool { + matches!( + error, + GraphDbError::Cancelled | GraphDbError::DeadlineExceeded | GraphDbError::SealSuperseded + ) +} + +/// The seal's rows as a delta over `base`, or `None` when the base declines +/// or the attempt fails for any reason but an interruption. +#[allow(clippy::too_many_arguments)] +fn layered_graph_rows( + seal: &VerifiedSealGraphSourceV1, + generation: &tracedecay_domain::CodeGenerationId, + projection: &GraphProjectionIdentity, + projector_revision: &GraphProjectorRevision, + base: &str, + spill: LayeredRowSpillV1, + admit: &mut dyn FnMut( + tracedecay_code_index::production::CodeGraphBuildBoundV1, + ) -> Result<(), GraphDbError>, + check: &dyn Fn() -> Result<(), GraphDbError>, +) -> Result)>, GraphDbError> { + let layered = spill.and_then(|spill| match spill { + Ok(spill) => with_verified_segments( + &seal.sealed_state_digest, + &seal.routes, + check, + |read_segment| { + build_layered_code_graph_rows( + projection.clone(), + &seal.source, + read_segment, + projector_revision, + spill, + admit, + check, + ) + }, + ) + .map(|built| built.map_err(|decline| format!("{decline:?}"))), + Err(absence) => Ok(Err(format!("{absence:?}"))), + }); + match layered { + Ok(Ok(built)) => return Ok(Some((built.generation.into(), Some(built.report)))), + Ok(Err(reason)) => tracing::info!( + event = "code_graph_layered_refresh_declined", + generation = %generation, + base = %base, + reason = %reason, + "the refresh seals cold rather than as a delta over this base's graph" + ), + Err(error) if is_interruption(&error) => return Err(error), + Err(error) => tracing::warn!( + event = "code_graph_layered_refresh_unavailable", + generation = %generation, + base = %base, + error = %error, + "a layered graph refresh could not be built; sealing the generation cold" + ), + } + Ok(None) +} + struct PinnedPartitionedSegmentV1 { digest: String, size_bytes: u64,