From 94758f19374bb11e21eb35bfaaea62dd74f8e9ad Mon Sep 17 00:00:00 2001 From: forhappy Date: Wed, 14 Oct 2026 22:23:22 -0700 Subject: [PATCH 1/6] refactor(metadata): share ref capture authority --- crates/crab-metadata/README.md | 5 + .../crab-metadata/src/capsule_protocol/mod.rs | 4 +- .../src/capsule_protocol/store.rs | 22 ++- .../src/capsule_protocol/store/tests.rs | 37 +++++ crates/crab-read/src/capsule_protocol.rs | 145 +----------------- 5 files changed, 66 insertions(+), 147 deletions(-) diff --git a/crates/crab-metadata/README.md b/crates/crab-metadata/README.md index 7678cf595..522ee677c 100644 --- a/crates/crab-metadata/README.md +++ b/crates/crab-metadata/README.md @@ -46,6 +46,11 @@ Catalog readers resolve only ref heads in that root's authority epoch, before loading multi-ref activation records. Restore can leave retired heads in storage; their checkpoint positions and dependencies must not enter the restored catalog. Current-epoch activation and checkpoint-chain errors still fail closed. +`capture_ref_heads` and `resolve_referenced_activations` own that stable capture +for metadata and read orchestration. Transactionless heads must match the +compacted root; selected-ref capture reuses the same activation validator while +keeping its targeted ETag probes. This extraction does not yet remove the eager +current-state file-lookup catalog loader. Run controls retain each capsule's own pack directory in capsule-relative coordinates. The run directory keeps physical transport coordinates separately. Logical capsule equality compares authenticated visibility section identities diff --git a/crates/crab-metadata/src/capsule_protocol/mod.rs b/crates/crab-metadata/src/capsule_protocol/mod.rs index a93cdb337..c435b426a 100644 --- a/crates/crab-metadata/src/capsule_protocol/mod.rs +++ b/crates/crab-metadata/src/capsule_protocol/mod.rs @@ -64,10 +64,10 @@ pub use run::{ pub use run::{CapsuleRunAdmissionReader, RunCatalogSource}; #[cfg(feature = "storage")] pub use store::{ - RootSnapshot, create_root, load_capsule_run, load_capsule_run_control, + RootSnapshot, capture_ref_heads, create_root, load_capsule_run, load_capsule_run_control, load_capsule_run_git_control, load_history_chain, load_history_segment, load_layered_checkpoint, load_layered_checkpoint_control, load_pointer_catalog, - load_pointer_catalog_from_root, load_root, + load_pointer_catalog_from_root, load_root, resolve_referenced_activations, }; pub use transaction::{CapsuleRefEdit, CapsuleTransaction}; pub use transaction_record::{ diff --git a/crates/crab-metadata/src/capsule_protocol/store.rs b/crates/crab-metadata/src/capsule_protocol/store.rs index c5c35d069..d3ed7d433 100644 --- a/crates/crab-metadata/src/capsule_protocol/store.rs +++ b/crates/crab-metadata/src/capsule_protocol/store.rs @@ -674,7 +674,11 @@ async fn load_visible_ref_capsules( Ok(ordered) } -async fn capture_ref_heads( +/// Capture one stable, epoch-filtered ref-head view and its committed activations. +/// +/// Rejects mismatched object identities, activation records and uncommitted base +/// heads. Concurrent changes retry within the existing bounded capture policy. +pub async fn capture_ref_heads( router: &StoreLayout, root: &super::RepositoryRoot, ) -> Result<(Vec, BTreeSet)> { @@ -717,6 +721,16 @@ async fn capture_ref_heads( if before != after { continue; } + for head in &heads { + let visible = head.visible(&active); + let base_oid = root.refs().get(head.ref_name()).map(String::as_str); + if visible.transaction_id().is_none() && visible.oid() != base_oid { + return Err(corrupt( + &router.capsule_ref_head_path(&capsule_ref_name_key(head.ref_name())), + "capsule ref head without a transaction differs from the compacted root", + )); + } + } return Ok((heads, active)); } Err(contract_error( @@ -745,7 +759,11 @@ fn listed_version_matches(object: &ObjectMeta, etag: &ETag) -> bool { .is_none_or(|listed| etag.version.as_ref() == Some(listed)) } -async fn resolve_referenced_activations( +/// Validate referenced activation records and select those committed for these heads. +/// +/// Callers filter retired epochs before resolving activations so stale heads +/// cannot introduce authority or demand retired dependencies. +pub async fn resolve_referenced_activations( router: &StoreLayout, heads: &[CapsuleRefHead], ) -> Result> { diff --git a/crates/crab-metadata/src/capsule_protocol/store/tests.rs b/crates/crab-metadata/src/capsule_protocol/store/tests.rs index 70560bb41..e407e0061 100644 --- a/crates/crab-metadata/src/capsule_protocol/store/tests.rs +++ b/crates/crab-metadata/src/capsule_protocol/store/tests.rs @@ -534,3 +534,40 @@ async fn visibility_lookup_keeps_oid_pages_cold_until_requested() { .is_err() ); } + +#[tokio::test] +async fn ref_capture_rejects_uncommitted_base_changes_only_in_the_active_epoch() { + for retired in [false, true] { + let layout = StoreLayout::new( + Store::new(Arc::new(object_store::memory::InMemory::new())), + "repositories/ref-capture".to_owned(), + ); + let root = RepositoryRoot::initial(&"1".repeat(64), "refs/heads/main").unwrap(); + let name = "refs/tags/uncommitted"; + let head = CapsuleRefHead::from_root( + name, + if retired { + "2".repeat(64) + } else { + root.ref_epoch().to_owned() + }, + Some("3".repeat(40)), + None, + ) + .unwrap(); + layout + .store() + .create_strict( + &layout.capsule_ref_head_path(&capsule_ref_name_key(name)), + head.encode().unwrap(), + ) + .await + .unwrap(); + let result = capture_ref_heads(&layout, &root).await; + if retired { + assert!(result.unwrap().0.is_empty()); + } else { + assert!(matches!(result, Err(MetadataError::CorruptObject { .. }))); + } + } +} diff --git a/crates/crab-read/src/capsule_protocol.rs b/crates/crab-read/src/capsule_protocol.rs index f52bb40d4..8fb1df7cb 100644 --- a/crates/crab-read/src/capsule_protocol.rs +++ b/crates/crab-read/src/capsule_protocol.rs @@ -4,13 +4,13 @@ use bytes::Bytes; use crab_metadata::capsule_protocol::{ Capsule, CapsuleControl, CapsulePointer, CapsuleRun, CapsuleRunControl, CheckpointPointer, LayeredCheckpoint, PackMemberDescriptor, PackRange, PackSourceDescriptor, PackSourceKind, - PointerCatalog, RootRecord, load_root, visibility_object_set_digest, + PointerCatalog, RootRecord, capture_ref_heads, load_root, resolve_referenced_activations, + visibility_object_set_digest, }; use crab_storage::{Store, StoreLayout}; use futures_util::future::try_join_all; use futures_util::{StreamExt, TryStreamExt}; use gix_hash::ObjectId; -use object_store::ObjectMeta; use std::collections::{BTreeMap, BTreeSet}; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -5231,73 +5231,6 @@ fn materialize_visible_ref_heads( }) } -async fn load_ref_heads( - router: &StoreLayout, - objects: &[ObjectMeta], -) -> Result>> { - let loaded = futures_util::stream::iter(objects.iter().cloned().map(|object| async move { - let (body, etag) = router.store().get_with_etag(&object.location).await?; - let head = crab_metadata::capsule_protocol::CapsuleRefHead::decode(&body)?; - let expected = router.capsule_ref_head_path( - &crab_metadata::capsule_protocol::capsule_ref_name_key(head.ref_name()), - ); - if expected != object.location { - return Err(corrupt( - &object.location, - "capsule ref-head key does not match its ref name", - )); - } - Ok::<_, ReadError>((head, listed_version_matches(&object, &etag))) - })) - .buffer_unordered(32) - .try_collect::>() - .await?; - if loaded.iter().any(|(_, matched)| !matched) { - return Ok(None); - } - let mut heads = loaded.into_iter().map(|(head, _)| head).collect::>(); - heads.sort_unstable_by(|left, right| left.ref_name().cmp(right.ref_name())); - Ok(Some(heads)) -} - -async fn capture_ref_heads( - router: &StoreLayout, - root: &crab_metadata::capsule_protocol::RepositoryRoot, -) -> Result<( - Vec, - BTreeSet, -)> { - for _ in 0..8 { - let before = list_ref_head_objects(router).await?; - let Some(heads) = load_ref_heads(router, &before).await? else { - continue; - }; - let heads = heads - .into_iter() - .filter(|head| head.ref_epoch() == root.ref_epoch()) - .collect::>(); - let active = resolve_referenced_activations(router, &heads).await?; - let after = list_ref_head_objects(router).await?; - if before != after { - continue; - } - for head in &heads { - let visible = head.visible(&active); - let base_oid = root.refs().get(head.ref_name()).map(String::as_str); - if visible.transaction_id().is_none() && visible.oid() != base_oid { - return Err(corrupt_path( - "capsule-protocol ref heads", - "capsule ref head without a transaction differs from the compacted root", - )); - } - } - return Ok((heads, active)); - } - Err(ReadError::internal( - "capsule ref snapshot changed during every bounded capture attempt", - )) -} - async fn capture_selected_ref_heads( router: &StoreLayout, root: &crab_metadata::capsule_protocol::RepositoryRoot, @@ -5370,80 +5303,6 @@ async fn load_selected_ref_heads( .await } -async fn list_ref_head_objects(router: &StoreLayout) -> Result> { - let mut objects = router - .store() - .list_prefix_bounded( - &router.capsule_ref_heads_prefix(), - crab_metadata::capsule_protocol::MAX_CAPSULE_REF_HEADS, - ) - .await? - .ok_or_else(|| ReadError::internal("capsule ref-head limit exceeded"))?; - objects.sort_unstable_by(|left, right| left.location.cmp(&right.location)); - Ok(objects) -} - -fn listed_version_matches(object: &ObjectMeta, etag: &crab_storage::ETag) -> bool { - object - .e_tag - .as_ref() - .is_none_or(|listed| etag.e_tag.as_ref() == Some(listed)) - && object - .version - .as_ref() - .is_none_or(|listed| etag.version.as_ref() == Some(listed)) -} - -async fn resolve_referenced_activations( - router: &StoreLayout, - heads: &[crab_metadata::capsule_protocol::CapsuleRefHead], -) -> Result> { - let referenced = heads - .iter() - .filter_map(|head| head.prepared_activation_id()) - .map(str::to_owned) - .collect::>(); - futures_util::stream::iter(referenced.into_iter().map(|activation_id| async move { - let path = router.capsule_transaction_path(&activation_id); - let (body, _) = router - .store() - .get_with_etag_bounded( - &path, - crab_metadata::capsule_protocol::MAX_CAPSULE_TRANSACTION_RECORD_BYTES, - ) - .await?; - let record = crab_metadata::capsule_protocol::CapsuleTransactionRecord::decode(&body)?; - if record.activation_id() != activation_id { - return Err(corrupt( - &path, - "capsule transaction record does not match its activation key", - )); - } - if heads.iter().any(|head| { - head.prepared_activation_id() == Some(activation_id.as_str()) - && head - .visible(&BTreeSet::from([activation_id.clone()])) - .transaction_id() - != Some(record.transaction_id()) - }) { - return Err(corrupt( - &path, - "capsule transaction record does not match its prepared ref heads", - )); - } - Ok::<_, ReadError>(( - activation_id, - record.status() == crab_metadata::capsule_protocol::CapsuleTransactionStatus::Committed, - )) - })) - .buffer_unordered(32) - .try_filter_map( - |(activation_id, committed)| async move { Ok(committed.then_some(activation_id)) }, - ) - .try_collect() - .await -} - fn order_ref_capsules( base_refs: &BTreeMap, base_peeled: &BTreeMap, From dc7f5dd5de0996234bca23b5b4ed53a7abde9ec5 Mon Sep 17 00:00:00 2001 From: forhappy Date: Wed, 14 Oct 2026 23:15:23 -0700 Subject: [PATCH 2/6] perf(metadata): page current catalog lookups Share captured ref ordering and visibility with Git readers. Validate full catalog dependencies after assembling the visible cross-ref set. --- crates/crab-metadata/README.md | 34 +- .../capsule_protocol/captured_visibility.rs} | 502 ++++++----------- .../captured_visibility/tests.rs | 195 +++++++ .../src/capsule_protocol/catalog_capture.rs | 175 ++++++ .../crab-metadata/src/capsule_protocol/mod.rs | 21 +- .../src/capsule_protocol/pointer.rs | 44 +- .../src/capsule_protocol/pointer/paged.rs | 2 +- .../src/capsule_protocol/ref_view.rs | 330 +++++++++++ .../src/capsule_protocol/store.rs | 195 +------ .../src/capsule_protocol/store/tests.rs | 138 ++++- crates/crab-metadata/src/file_index_lookup.rs | 47 +- .../src/file_index_lookup/catalog_tests.rs | 162 +++++- crates/crab-read/README.md | 6 + crates/crab-read/src/capsule_protocol.rs | 511 ++---------------- .../src/capsule_protocol/visibility_tests.rs | 106 ++++ 15 files changed, 1416 insertions(+), 1052 deletions(-) rename crates/{crab-read/src/capsule_protocol/visibility.rs => crab-metadata/src/capsule_protocol/captured_visibility.rs} (55%) create mode 100644 crates/crab-metadata/src/capsule_protocol/captured_visibility/tests.rs create mode 100644 crates/crab-metadata/src/capsule_protocol/catalog_capture.rs create mode 100644 crates/crab-metadata/src/capsule_protocol/ref_view.rs create mode 100644 crates/crab-read/src/capsule_protocol/visibility_tests.rs diff --git a/crates/crab-metadata/README.md b/crates/crab-metadata/README.md index 522ee677c..79984ed78 100644 --- a/crates/crab-metadata/README.md +++ b/crates/crab-metadata/README.md @@ -49,8 +49,9 @@ Current-epoch activation and checkpoint-chain errors still fail closed. `capture_ref_heads` and `resolve_referenced_activations` own that stable capture for metadata and read orchestration. Transactionless heads must match the compacted root; selected-ref capture reuses the same activation validator while -keeping its targeted ETag probes. This extraction does not yet remove the eager -current-state file-lookup catalog loader. +keeping its targeted ETag probes. Current-state catalog lookups reuse that owner together with the shared +control-selection and visibility machinery; full maintenance materialization +remains explicit. Run controls retain each capsule's own pack directory in capsule-relative coordinates. The run directory keeps physical transport coordinates separately. Logical capsule equality compares authenticated visibility section identities @@ -71,15 +72,18 @@ accepts captured lookup roots and resolves requested files with their authentica shard/xorb closure. Point and batch calls share the page cache and retain the session read guard through I/O, so close waits for active reads. Missing catalog pages remain errors rather than reopening the mutable file-index database. -The current-state opener still materializes a complete v2 catalog; callers with -an authenticated view should pass its captured roots. All 19 audited local release tags omit these +The current-state opener also captures authenticated catalog roots and leaves +catalog bodies cold. Explicitly pinned callers pass their existing capture, while +current-state sessions bind once to the first root they open. Catalog and Git +views share stable ref capture, ref-frontier selection, visibility reconciliation +and transaction ordering; neither path invents a separate precedence policy. All 19 audited local release tags omit these capsule checkpoint formats; unshipped format 6 is rejected without a fallback. -Captured run-v4 roots provide the same selected lookup over exact logical -capsule occurrences; catalog bodies remain cold until their keys are selected. +Checkpoint and selected run catalog roots share one lookup budget and page cache. CLI and shared capsule-publication preparation use these captured queries. Protected receive selects from separately authorized source/filtered captures. -Captured revision and remote-mount hydration pass the same paged lookup to the -shared hydrator; current-state hydration still requires capture integration. +Captured revision, remote-mount and current-state hydration use the shared paged +lookup. Control capture and selected closure queries are measured separately; +complete opening and history-amplification qualification remain required. External pointer catalogs use `CRBPCP01` authenticated pages. A bounded descriptor commits file, shard, xorb and reverse-chunk directories; nested shard closures and @@ -88,12 +92,16 @@ ordering, including the previous lowest-xorb/first-occurrence placement rule. Complete inspection verifies the body hash, complete dependency closure and canonical page layout. Delta inspection defers base-owned closure to `apply`, which retains later file bindings at equal size and rejects immutable shard/xorb -conflicts. Encoded catalogs are limited to 2 GiB and descriptors to 4 KiB. +conflicts. +Full inspection of a captured visible set uses `PointerCatalogAssembly`: binding +order need not be dependency order across independent refs. Rows and immutable +conflicts are checked while collecting deltas, then the complete closure is +checked before returning the catalog. Publication-time `PointerCatalog::apply` +still requires each delta's dependencies to exist in its base. Encoded catalogs are limited to 2 GiB and descriptors to 4 KiB. The retired monolithic JSON encoding was absent from all 19 audited release tags -and is rejected. Foreground consumers must use captured selected lookups rather -than infer bounded work from the page encoding alone. Maintenance retains explicit full -inspection; shared/protected hydration and lifecycle qualification remain -separate gates. +and is rejected. Shared file lookup uses paged capture for pinned and current-state sessions. +Complete catalog inspection remains explicit; consumer, historical amplification +and lifecycle qualification must not be inferred from the page representation. `v2/browse-indexes` is a bounded, mutable derived record, not ref authority. Its version 3 record binds a durable immutable commit graph, optional complete diff --git a/crates/crab-read/src/capsule_protocol/visibility.rs b/crates/crab-metadata/src/capsule_protocol/captured_visibility.rs similarity index 55% rename from crates/crab-read/src/capsule_protocol/visibility.rs rename to crates/crab-metadata/src/capsule_protocol/captured_visibility.rs index 4682e2fa2..60dd7d5cb 100644 --- a/crates/crab-read/src/capsule_protocol/visibility.rs +++ b/crates/crab-metadata/src/capsule_protocol/captured_visibility.rs @@ -1,30 +1,46 @@ //! Captured visibility closure layers; physical presence never grants membership. -use super::*; -use crab_metadata::capsule_protocol::{ - CapsuleVisibilityReader, LayeredVisibilityReader, VisibilityChange, VisibilityEditHeader, +use crate::capsule_protocol::{ + CapsuleControl, CapsuleTransaction, CapsuleVisibilityDelta, CapsuleVisibilityReader, + LayeredVisibilityReader, VisibilityChange, VisibilityEditHeader, }; -use crab_metadata::git_visibility::{GitVisibilityIndex, MAX_GIT_VISIBILITY_OBJECTS}; +use crate::error::{MetadataError, Result}; +use crate::git_visibility::{GitVisibilityIndex, MAX_GIT_VISIBILITY_OBJECTS, decode_oid}; +use crab_storage::Store; +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Arc; -pub(super) enum Reconciliation { +/// Reconcile captured controls against complete or paged visibility evidence. +pub enum VisibilityReconciliation { Complete(GitVisibilityIndex), - Paged(Box), + Paged(Box), } -impl Reconciliation { - pub(super) async fn ready( +impl VisibilityReconciliation { + /// Check expected ref values and captured prior-closure availability. + pub async fn ready( &self, control: &CapsuleControl, refs: &BTreeMap, ) -> Result { match self { - Self::Complete(index) => control_is_ready(control, refs, index), + Self::Complete(index) => visibility_is_ready( + control.transaction(), + control.visibility_delta()?, + refs, + index, + ), Self::Paged(index) => index.ready(control, refs).await, } } - pub(super) async fn apply(&mut self, control: &CapsuleControl) -> Result<()> { + /// Apply one authenticated control without widening to mutable ref state. + pub async fn apply(&mut self, control: &CapsuleControl) -> Result<()> { match self { Self::Complete(index) if control.has_visibility() => { - apply_capsule_control_visibility_index(control, index) + apply_transaction_visibility_index( + control.transaction(), + control.visibility_delta()?, + index, + ) } Self::Complete(_) => Ok(()), Self::Paged(index) => index.apply(control).await, @@ -77,7 +93,8 @@ impl Initial { } } -pub(super) struct Visibility { +/// Retain immutable prior ref closures while applying authenticated visibility edits. +pub struct CapturedVisibility { store: Store, initial: Initial, layers: Vec, @@ -87,16 +104,10 @@ pub(super) struct Visibility { dictionary_count: u64, } -fn oid(value: &str) -> Result<[u8; 20]> { - ObjectId::from_hex(value.as_bytes()) - .map_err(|_| corrupt_path("capsule Git visibility", "invalid object identity"))? - .as_bytes() - .try_into() - .map_err(|_| corrupt_path("capsule Git visibility", "object format is unsupported")) -} - -impl Visibility { - pub(super) fn new( +impl CapturedVisibility { + /// Seed captured closures from complete authenticated visibility evidence. + #[must_use] + pub fn new( store: Store, initial: GitVisibilityIndex, names: impl Iterator, @@ -107,7 +118,8 @@ impl Visibility { Self::seed(store, Initial::Complete(initial), names) } - pub(super) async fn from_checkpoint( + /// Seed captured closures from the checkpoint's authenticated visibility pages. + pub async fn from_checkpoint( store: Store, reader: Arc, names: impl Iterator, @@ -285,7 +297,8 @@ impl Visibility { Ok(Some((reader.clone(), headers))) } - pub(super) async fn ready( + /// Check expected ref values and captured prior-closure availability. + pub async fn ready( &self, control: &CapsuleControl, refs: &BTreeMap, @@ -311,7 +324,7 @@ impl Visibility { if edit.expected_old().is_none() && !header.replaces() && let Some(old) = header.old_oid() - && self.first_ref(oid(old)?).await?.is_none() + && self.first_ref(decode_oid(old)?).await?.is_none() { return Ok(false); } @@ -319,7 +332,8 @@ impl Visibility { Ok(true) } - pub(super) async fn apply(&mut self, control: &CapsuleControl) -> Result<()> { + /// Apply one authenticated control without widening to mutable ref state. + pub async fn apply(&mut self, control: &CapsuleControl) -> Result<()> { let Some((reader, headers)) = self.headers(control).await? else { return Ok(()); }; @@ -333,17 +347,17 @@ impl Visibility { && !header.replaces() && let Some(old) = header.old_oid() { - parent = self.first_ref(oid(old)?).await?; + parent = self.first_ref(decode_oid(old)?).await?; } if let Some(old) = header.old_oid() - && !self.contains(parent, &[oid(old)?]).await?[0] + && !self.contains(parent, &[decode_oid(old)?]).await?[0] { return Err(corrupt_path( "capsule Git visibility", "visibility delta prior closure does not contain its old ref tip", )); } - let new = oid(header.new_oid())?; + let new = decode_oid(header.new_oid())?; let change = reader .lookup(&self.store, edit.ref_name(), &[new]) .await? @@ -420,315 +434,145 @@ impl Visibility { } } -#[cfg(test)] -mod tests { - use super::*; - use crab_metadata::capsule_protocol::{ - CapsuleRefEdit, CapsuleSection, CapsuleSectionKind, CapsuleTransaction, - CapsuleVisibilityDelta, - }; - use crab_metadata::git_visibility::GitVisibilityEdit; - use object_store::memory::InMemory; - - fn hex(value: u8) -> String { - format!("{value:040x}") - } - fn object(value: u8) -> [u8; 20] { - let mut oid = [0; 20]; - oid[19] = value; - oid +/// Validate publication evidence and check its prior ref dependencies. +pub fn visibility_is_ready( + transaction: &CapsuleTransaction, + delta: Option<&CapsuleVisibilityDelta>, + refs: &BTreeMap, + visibility_index: &crate::git_visibility::GitVisibilityIndex, +) -> Result { + if transaction + .edits() + .iter() + .any(|edit| refs.get(edit.ref_name()).map(String::as_str) != edit.expected_old()) + { + return Ok(false); } - async fn control( - layout: &StoreLayout, - name: &str, - expected: Option, - edit: GitVisibilityEdit, - ) -> CapsuleControl { - let transaction = CapsuleTransaction::new( - &"1".repeat(64), - vec![CapsuleRefEdit::new( - name, - expected.map(hex), - Some(edit.new_oid.clone()), - None, - )], - ) - .unwrap(); - let delta = CapsuleVisibilityDelta::new(BTreeMap::from([(name.to_owned(), edit)])).unwrap(); - let capsule = Capsule::build( - &transaction, - Vec::new(), - vec![CapsuleSection::new( - CapsuleSectionKind::VisibilityDelta, - delta.encode().unwrap(), - )], - ) - .unwrap(); - let run = CapsuleRun::leaf(capsule).unwrap(); - layout - .store() - .create_strict(&layout.capsule_path(run.hash()), run.bytes().clone()) - .await - .unwrap(); - let pointer = CapsulePointer::new( - run.hash(), - run.bytes().len() as u64, - run.control_offset(), - run.control_size(), - run.footer_hash(), - run.level(), - run.transaction_ids(), - run.newest_base_root_digest(), - ) - .unwrap(); - let (_, mut controls) = - crab_metadata::capsule_protocol::load_capsule_run_git_control(layout, &pointer) - .await - .unwrap(); - controls.remove(0) + let Some(delta) = delta else { + return Ok(true); + }; + // Transactions have unique refs. Count matched rows without copying OID lists; + // unmatched evidence still fails before it can authorize a publication. + let evidence = delta.edits(); + let mut matched = 0; + for edit in transaction.edits() { + let Some(new_oid) = edit.new_oid() else { + if evidence.get(edit.ref_name()).is_some() { + return Err(corrupt_path( + "capsule Git visibility", + "deleted ref has visibility evidence", + )); + } + continue; + }; + let Some(visibility) = evidence.get(edit.ref_name()) else { + return Err(corrupt_path( + "capsule Git visibility", + "live ref edit has no visibility evidence", + )); + }; + matched += 1; + visibility.validate()?; + if visibility.new_oid != new_oid { + return Err(corrupt_path( + "capsule Git visibility", + "visibility evidence does not match its ref edit", + )); + } + match edit.expected_old() { + Some(expected_old) => { + if visibility.old_oid.as_deref() != Some(expected_old) { + return Err(corrupt_path( + "capsule Git visibility", + "visibility evidence does not match the expected old ref", + )); + } + if !visibility_index.contains_ref(edit.ref_name()) { + return Ok(false); + } + } + None if visibility.replaces => {} + None => { + let Some(old_oid) = visibility.old_oid.as_deref() else { + continue; + }; + if !visibility_index.contains_hex_in_any_ref(old_oid) { + return Ok(false); + } + } + } } - fn initial() -> GitVisibilityIndex { - GitVisibilityIndex::new(0, "0".repeat(64), "0".repeat(64), BTreeMap::new()).unwrap() + if evidence.len() != matched { + return Err(corrupt_path( + "capsule Git visibility", + "visibility evidence contains an uncommitted ref", + )); } + Ok(true) +} - #[tokio::test] - async fn initial_replacement_opens_without_materializing_its_oid_set() { - use crab_metadata::capsule_protocol::{RepositoryRoot, RootRecord}; - use std::sync::atomic::{AtomicU64, Ordering}; - let bytes = Arc::new(AtomicU64::new(0)); - let observed = bytes.clone(); - let store = - Store::new(Arc::new(InMemory::new())).with_read_byte_observer(Arc::new(move |size| { - observed.fetch_add(size, Ordering::Relaxed); - })); - let layout = StoreLayout::new(store, "repositories/large-root".into()); - let initial = RootRecord::encode( - RepositoryRoot::initial(&"1".repeat(64), "refs/heads/main").unwrap(), - ) - .unwrap(); - crab_metadata::capsule_protocol::create_root(&layout, initial.clone()) - .await - .unwrap(); - let objects = (0u32..10000) - .map(|index| format!("{index:040x}")) - .collect::>(); - let tip = objects[9000].clone(); - let transaction = CapsuleTransaction::new( - initial.digest(), - vec![CapsuleRefEdit::new( - "refs/heads/main", - None, - Some(tip.clone()), - None, - )], - ) - .unwrap(); - let delta = CapsuleVisibilityDelta::new(BTreeMap::from([( - "refs/heads/main".into(), - GitVisibilityEdit::from_replacement_objects(None, tip.clone(), objects), - )])) - .unwrap(); - let capsule = Capsule::build( - &transaction, - Vec::new(), - vec![CapsuleSection::new( - CapsuleSectionKind::VisibilityDelta, - delta.encode().unwrap(), - )], - ) - .unwrap(); - let run = CapsuleRun::leaf(capsule).unwrap(); - layout - .store() - .create_strict(&layout.capsule_path(run.hash()), run.bytes().clone()) - .await - .unwrap(); - let pointer = CapsulePointer::new( - run.hash(), - run.bytes().len() as u64, - run.control_offset(), - run.control_size(), - run.footer_hash(), - run.level(), - run.transaction_ids(), - run.newest_base_root_digest(), - ) - .unwrap(); - let next = RootRecord::encode( - initial - .root() - .advance( - initial.digest(), - BTreeMap::from([("refs/heads/main".into(), tip)]), - BTreeMap::new(), - vec![pointer], - transaction.id().unwrap().as_str(), +/// Apply transaction-bound visibility evidence to a complete captured index. +pub fn apply_transaction_visibility_index( + transaction: &CapsuleTransaction, + delta: Option<&CapsuleVisibilityDelta>, + index: &mut crate::git_visibility::GitVisibilityIndex, +) -> Result<()> { + let evidence = delta.map(|delta| delta.edits()); + let mut matched = 0; + for edit in transaction.edits() { + let Some(new_oid) = edit.new_oid() else { + if evidence + .and_then(|rows| rows.get(edit.ref_name())) + .is_some() + { + return Err(corrupt_path( + "capsule Git visibility", + "deleted ref has visibility evidence", + )); + } + index.remove_ref(edit.ref_name()); + continue; + }; + let visibility = evidence + .and_then(|rows| rows.get(edit.ref_name())) + .ok_or_else(|| { + corrupt_path( + "capsule Git visibility", + "live ref edit has no visibility evidence", ) - .unwrap(), - ) - .unwrap(); - layout - .store() - .put_overwrite(&layout.capsule_root_path(), next.bytes().clone()) - .await - .unwrap(); - let root = crab_metadata::capsule_protocol::load_root(&layout) - .await - .unwrap(); - bytes.store(0, Ordering::Relaxed); - let view = open_git_view_from_root( - &layout, - root, - CapsuleReadLimits { - max_capsule_bytes: 1024 * 1024, - max_frontier_bytes: 1024 * 1024, - }, - ) - .await - .unwrap(); - assert!( - bytes.load(Ordering::Relaxed) < 64 * 1024 - && view.capsule_controls[0].visibility_delta().is_err() - ); - } - - #[tokio::test] - async fn layered_closures_match_complete_edits_and_preserve_captured_bases() { - let layout = StoreLayout::new( - Store::new(Arc::new(InMemory::new())), - "repositories/layers".into(), - ); - let mut full = initial(); - let mut lazy = Visibility::new(layout.store().clone(), full.clone(), std::iter::empty()); - let edits = [ - ( - "refs/heads/main", - None, - GitVisibilityEdit::from_replacement_objects( - None, - hex(1), - vec![hex(1), hex(2), hex(3)], - ), - ), - ( - "refs/heads/feature", - None, - GitVisibilityEdit::from_delta_objects( - Some(hex(1)), - hex(2), - vec![hex(4)], - vec![hex(3)], - ), - ), - ( - "refs/heads/main", - Some(1), - GitVisibilityEdit::from_delta_objects( - Some(hex(1)), - hex(5), - vec![hex(5)], - vec![hex(3)], - ), - ), - ]; - for (name, expected, edit) in edits { - let control = control(&layout, name, expected, edit.clone()).await; - full.apply_ref_edit(name.to_owned(), &edit).unwrap(); - lazy.apply(&control).await.unwrap(); + })?; + matched += 1; + if visibility.new_oid != new_oid { + return Err(corrupt_path( + "capsule Git visibility", + "visibility evidence does not match its ref edit", + )); } - let requested = (1..7).map(object).collect::>(); - let mut actual = BTreeMap::new(); - for (name, index) in &lazy.refs { - actual.insert( - name.clone(), - lazy.contains(Some(*index), &requested).await.unwrap(), - ); + if let Some(expected_old) = edit.expected_old() + && visibility.old_oid.as_deref() != Some(expected_old) + { + return Err(corrupt_path( + "capsule Git visibility", + "visibility evidence does not match the expected old ref", + )); } - let expected = lazy - .refs - .keys() - .map(|name| { - ( - name.clone(), - requested - .iter() - .map(|oid| full.contains_in_ref(name, oid)) - .collect::>(), - ) - }) - .collect::>(); - assert_eq!( - (actual, lazy.dictionary_count), - (expected, full.objects().len() as u64) - ); + index.apply_ref_edit(edit.ref_name().to_owned(), visibility)?; } - - #[tokio::test] - async fn removed_invisible_objects_do_not_supply_new_ref_bases() { - let layout = StoreLayout::new( - Store::new(Arc::new(InMemory::new())), - "repositories/invisible".into(), - ); - let mut lazy = Visibility::new(layout.store().clone(), initial(), std::iter::empty()); - let create = control( - &layout, - "refs/heads/main", - None, - GitVisibilityEdit::from_replacement_objects(None, hex(1), vec![hex(1), hex(3)]), - ) - .await; - lazy.apply(&create).await.unwrap(); - let remove = control( - &layout, - "refs/heads/main", - Some(1), - GitVisibilityEdit::from_delta_objects(Some(hex(1)), hex(2), vec![hex(2)], vec![hex(3)]), - ) - .await; - lazy.apply(&remove).await.unwrap(); - let branch = control( - &layout, - "refs/heads/feature", - None, - GitVisibilityEdit::from_delta_objects(Some(hex(3)), hex(4), vec![hex(4)], vec![]), - ) - .await; - assert_eq!( - ( - lazy.dictionary_contains(&[object(3)]).await.unwrap(), - lazy.ready( - &branch, - &BTreeMap::from([("refs/heads/main".into(), hex(2))]) - ) - .await - .unwrap() - ), - (vec![true], false) - ); + if evidence.is_some_and(|rows| rows.len() != matched) { + return Err(corrupt_path( + "capsule Git visibility", + "visibility evidence contains an uncommitted ref", + )); } + Ok(()) +} - #[tokio::test] - async fn removal_outside_captured_prior_closure_is_rejected() { - let layout = StoreLayout::new( - Store::new(Arc::new(InMemory::new())), - "repositories/bad-removal".into(), - ); - let mut lazy = Visibility::new(layout.store().clone(), initial(), std::iter::empty()); - let create = control( - &layout, - "refs/heads/main", - None, - GitVisibilityEdit::from_replacement_objects(None, hex(1), vec![hex(1)]), - ) - .await; - lazy.apply(&create).await.unwrap(); - let bad = control( - &layout, - "refs/heads/main", - Some(1), - GitVisibilityEdit::from_delta_objects(Some(hex(1)), hex(2), vec![hex(2)], vec![hex(9)]), - ) - .await; - assert!(lazy.apply(&bad).await.is_err()); +fn corrupt_path(path: impl Into, reason: impl Into) -> MetadataError { + MetadataError::CorruptObject { + path: path.into(), + reason: reason.into(), } } + +#[cfg(test)] +mod tests; diff --git a/crates/crab-metadata/src/capsule_protocol/captured_visibility/tests.rs b/crates/crab-metadata/src/capsule_protocol/captured_visibility/tests.rs new file mode 100644 index 000000000..e7cd4593b --- /dev/null +++ b/crates/crab-metadata/src/capsule_protocol/captured_visibility/tests.rs @@ -0,0 +1,195 @@ +use super::*; +use crate::capsule_protocol::{ + Capsule, CapsulePointer, CapsuleRefEdit, CapsuleRun, CapsuleSection, CapsuleSectionKind, + CapsuleTransaction, CapsuleVisibilityDelta, +}; +use crate::git_visibility::GitVisibilityEdit; +use crab_storage::StoreLayout; +use object_store::memory::InMemory; + +fn hex(value: u8) -> String { + format!("{value:040x}") +} +fn object(value: u8) -> [u8; 20] { + let mut oid = [0; 20]; + oid[19] = value; + oid +} +async fn control( + layout: &StoreLayout, + name: &str, + expected: Option, + edit: GitVisibilityEdit, +) -> CapsuleControl { + let transaction = CapsuleTransaction::new( + &"1".repeat(64), + vec![CapsuleRefEdit::new( + name, + expected.map(hex), + Some(edit.new_oid.clone()), + None, + )], + ) + .unwrap(); + let delta = CapsuleVisibilityDelta::new(BTreeMap::from([(name.to_owned(), edit)])).unwrap(); + let capsule = Capsule::build( + &transaction, + Vec::new(), + vec![CapsuleSection::new( + CapsuleSectionKind::VisibilityDelta, + delta.encode().unwrap(), + )], + ) + .unwrap(); + let run = CapsuleRun::leaf(capsule).unwrap(); + layout + .store() + .create_strict(&layout.capsule_path(run.hash()), run.bytes().clone()) + .await + .unwrap(); + let pointer = CapsulePointer::new( + run.hash(), + run.bytes().len() as u64, + run.control_offset(), + run.control_size(), + run.footer_hash(), + run.level(), + run.transaction_ids(), + run.newest_base_root_digest(), + ) + .unwrap(); + let (_, mut controls) = crate::capsule_protocol::load_capsule_run_git_control(layout, &pointer) + .await + .unwrap(); + controls.remove(0) +} +fn initial() -> GitVisibilityIndex { + GitVisibilityIndex::new(0, "0".repeat(64), "0".repeat(64), BTreeMap::new()).unwrap() +} + +#[tokio::test] +async fn layered_closures_match_complete_edits_and_preserve_captured_bases() { + let layout = StoreLayout::new( + Store::new(Arc::new(InMemory::new())), + "repositories/layers".into(), + ); + let mut full = initial(); + let mut lazy = + CapturedVisibility::new(layout.store().clone(), full.clone(), std::iter::empty()); + let edits = [ + ( + "refs/heads/main", + None, + GitVisibilityEdit::from_replacement_objects(None, hex(1), vec![hex(1), hex(2), hex(3)]), + ), + ( + "refs/heads/feature", + None, + GitVisibilityEdit::from_delta_objects(Some(hex(1)), hex(2), vec![hex(4)], vec![hex(3)]), + ), + ( + "refs/heads/main", + Some(1), + GitVisibilityEdit::from_delta_objects(Some(hex(1)), hex(5), vec![hex(5)], vec![hex(3)]), + ), + ]; + for (name, expected, edit) in edits { + let control = control(&layout, name, expected, edit.clone()).await; + full.apply_ref_edit(name.to_owned(), &edit).unwrap(); + lazy.apply(&control).await.unwrap(); + } + let requested = (1..7).map(object).collect::>(); + let mut actual = BTreeMap::new(); + for (name, index) in &lazy.refs { + actual.insert( + name.clone(), + lazy.contains(Some(*index), &requested).await.unwrap(), + ); + } + let expected = lazy + .refs + .keys() + .map(|name| { + ( + name.clone(), + requested + .iter() + .map(|oid| full.contains_in_ref(name, oid)) + .collect::>(), + ) + }) + .collect::>(); + assert_eq!( + (actual, lazy.dictionary_count), + (expected, full.objects().len() as u64) + ); +} + +#[tokio::test] +async fn removed_invisible_objects_do_not_supply_new_ref_bases() { + let layout = StoreLayout::new( + Store::new(Arc::new(InMemory::new())), + "repositories/invisible".into(), + ); + let mut lazy = CapturedVisibility::new(layout.store().clone(), initial(), std::iter::empty()); + let create = control( + &layout, + "refs/heads/main", + None, + GitVisibilityEdit::from_replacement_objects(None, hex(1), vec![hex(1), hex(3)]), + ) + .await; + lazy.apply(&create).await.unwrap(); + let remove = control( + &layout, + "refs/heads/main", + Some(1), + GitVisibilityEdit::from_delta_objects(Some(hex(1)), hex(2), vec![hex(2)], vec![hex(3)]), + ) + .await; + lazy.apply(&remove).await.unwrap(); + let branch = control( + &layout, + "refs/heads/feature", + None, + GitVisibilityEdit::from_delta_objects(Some(hex(3)), hex(4), vec![hex(4)], vec![]), + ) + .await; + assert_eq!( + ( + lazy.dictionary_contains(&[object(3)]).await.unwrap(), + lazy.ready( + &branch, + &BTreeMap::from([("refs/heads/main".into(), hex(2))]) + ) + .await + .unwrap() + ), + (vec![true], false) + ); +} + +#[tokio::test] +async fn removal_outside_captured_prior_closure_is_rejected() { + let layout = StoreLayout::new( + Store::new(Arc::new(InMemory::new())), + "repositories/bad-removal".into(), + ); + let mut lazy = CapturedVisibility::new(layout.store().clone(), initial(), std::iter::empty()); + let create = control( + &layout, + "refs/heads/main", + None, + GitVisibilityEdit::from_replacement_objects(None, hex(1), vec![hex(1)]), + ) + .await; + lazy.apply(&create).await.unwrap(); + let bad = control( + &layout, + "refs/heads/main", + Some(1), + GitVisibilityEdit::from_delta_objects(Some(hex(1)), hex(2), vec![hex(2)], vec![hex(9)]), + ) + .await; + assert!(lazy.apply(&bad).await.is_err()); +} diff --git a/crates/crab-metadata/src/capsule_protocol/catalog_capture.rs b/crates/crab-metadata/src/capsule_protocol/catalog_capture.rs new file mode 100644 index 000000000..78c162da2 --- /dev/null +++ b/crates/crab-metadata/src/capsule_protocol/catalog_capture.rs @@ -0,0 +1,175 @@ +//! One shared authenticated control capture for current-state catalog queries. +use super::pointer::paged::MAX_CATALOG_BYTES; +use super::{ + CapsuleControl, CapturedVisibility, LayeredCheckpoint, PointerCatalog, PointerCatalogAssembly, + PointerCatalogLookup, RootSnapshot, RunCatalogSource, VisibilityReconciliation, + apply_control_refs, capture_ref_heads, load_capsule_run_git_control, load_layered_checkpoint, + load_layered_checkpoint_control, materialize_visible_ref_heads, order_ref_controls, + select_ref_control_frontiers, select_ref_journal_controls, +}; +use crate::error::{MetadataError, Result}; +use crate::git_visibility::GitVisibilityIndex; +use crab_storage::{Store, StoreLayout}; +use futures_util::{StreamExt, TryStreamExt}; +use std::collections::BTreeMap; +use std::sync::Arc; + +/// Capture current-state catalog roots without decoding catalog or Git pack bodies. +/// +/// The caller retains the captured roots against GC. Missing dependencies remain +/// errors. Control intake and each selected catalog query use the format byte bound. +pub async fn load_pointer_catalog_lookup_from_root( + layout: &StoreLayout, + root: &RootSnapshot, +) -> Result { + capture(layout, root).await?.lookup(layout) +} + +pub(super) async fn materialize_from_root( + layout: &StoreLayout, + root: &RootSnapshot, +) -> Result { + let captured = capture(layout, root).await?; + let base_catalog = match root.record().root().checkpoint() { + Some(pointer) => load_layered_checkpoint(layout, pointer) + .await? + .pointer_catalog()?, + None => PointerCatalog::new(), + }; + // Concurrent ref histories have a canonical binding order, not a global + // payload dependency order. Validate completeness after collecting the set. + let mut catalog = PointerCatalogAssembly::new(base_catalog); + for control in captured.controls { + if let Some(delta) = control.materialize_catalog(layout.store()).await? { + catalog.apply(&delta)?; + } + } + catalog.finish() +} + +struct Capture { + checkpoint: Option, + controls: Vec, + sources: Vec, +} + +impl Capture { + fn lookup(self, layout: &StoreLayout) -> Result { + let lookup = self + .checkpoint + .as_ref() + .map(|checkpoint| checkpoint.pointer_catalog_lookup(layout, MAX_CATALOG_BYTES)) + .transpose()? + .unwrap_or_else(|| PointerCatalogLookup::empty(MAX_CATALOG_BYTES)); + let order = self + .controls + .iter() + .map(|control| control.hash().to_owned()) + .collect::>(); + lookup.with_run_sources(self.sources, &order) + } +} + +async fn capture(layout: &StoreLayout, snapshot: &RootSnapshot) -> Result { + let root = snapshot.record().root(); + let (heads, active) = capture_ref_heads(layout, root).await?; + let visible = materialize_visible_ref_heads(root, &heads, &active)?; + let mut intake = root + .checkpoint() + .map_or(0, |pointer| pointer.control_size()); + for pointer in &visible.pointers { + intake = intake + .checked_add(pointer.control_size()) + .ok_or_else(|| corrupt("catalog control intake overflows"))?; + } + if intake > MAX_CATALOG_BYTES { + return Err(corrupt( + "catalog control intake exceeds the format byte bound", + )); + } + let checkpoint = match root.checkpoint() { + Some(pointer) => Some(load_layered_checkpoint_control(layout, pointer).await?), + None => None, + }; + let run_layout = layout.clone(); + let runs = futures_util::stream::iter(visible.pointers.into_iter().map(move |pointer| { + let layout = run_layout.clone(); + async move { + let (control, capsules) = load_capsule_run_git_control(&layout, &pointer).await?; + if !control.git_packs().is_empty() { + control.source_descriptor()?; + } + Ok::<_, MetadataError>((control.hash().to_owned(), (control, capsules))) + } + })) + .buffer_unordered(32) + .try_collect::>() + .await?; + let selected = select_ref_control_frontiers(root, &visible.frontiers, &runs)?; + let pending = select_ref_journal_controls(root, &selected.required, &runs)?; + let initial = match checkpoint + .as_ref() + .map(|checkpoint| checkpoint.visibility_reader(layout, MAX_CATALOG_BYTES)) + .transpose()? + .flatten() + { + Some(reader) => { + CapturedVisibility::from_checkpoint( + layout.store().clone(), + Arc::new(reader), + root.refs().keys().cloned(), + ) + .await? + } + None => CapturedVisibility::new( + layout.store().clone(), + GitVisibilityIndex::new(0, "0".repeat(64), "0".repeat(64), BTreeMap::new())?, + root.refs().keys().cloned(), + ), + }; + let mut visibility = VisibilityReconciliation::Paged(Box::new(initial)); + let mut controls = selected.root_controls; + for control in &controls { + visibility.apply(control).await?; + } + let ordered = order_ref_controls( + root.refs(), + root.peeled_refs(), + visibility, + &selected.predecessors, + pending, + ) + .await?; + let mut refs = root.refs().clone(); + let mut peeled = root.peeled_refs().clone(); + // Root-owned controls are already reflected in the root's ref values. + // Only journal controls advance them during this captured reconciliation. + for control in &ordered { + apply_control_refs(control, &mut refs, &mut peeled)?; + } + controls.extend(ordered); + if refs != visible.refs || peeled != visible.peeled_refs { + return Err(corrupt( + "materialized controls differ from the captured ref-head state", + )); + } + let sources = runs + .values() + .map(|(run, _)| run.catalog_source(layout)) + .collect::>>()? + .into_iter() + .flatten() + .collect(); + Ok(Capture { + checkpoint, + controls, + sources, + }) +} + +fn corrupt(reason: impl Into) -> MetadataError { + MetadataError::CorruptObject { + path: "capsule-protocol catalog capture".to_owned(), + reason: reason.into(), + } +} diff --git a/crates/crab-metadata/src/capsule_protocol/mod.rs b/crates/crab-metadata/src/capsule_protocol/mod.rs index c435b426a..0da15ce80 100644 --- a/crates/crab-metadata/src/capsule_protocol/mod.rs +++ b/crates/crab-metadata/src/capsule_protocol/mod.rs @@ -4,12 +4,18 @@ use bstr::ByteSlice; mod browse; mod capsule; +#[cfg(feature = "storage")] +mod captured_visibility; +#[cfg(feature = "storage")] +mod catalog_capture; mod history; mod layered; #[cfg(feature = "storage")] mod plan; mod pointer; mod ref_head; +#[cfg(feature = "storage")] +mod ref_view; mod root; mod run; #[cfg(feature = "storage")] @@ -25,6 +31,13 @@ pub use capsule::{ Capsule, CapsuleGitPack, CapsuleGitPackDescriptor, CapsuleSection, CapsuleSectionKind, CapsuleSectionLocation, }; +#[cfg(feature = "storage")] +pub use captured_visibility::{ + CapturedVisibility, VisibilityReconciliation, apply_transaction_visibility_index, + visibility_is_ready, +}; +#[cfg(feature = "storage")] +pub use catalog_capture::load_pointer_catalog_lookup_from_root; pub use history::{ HistorySegment, HistorySegmentPointer, HistorySegmentState, MAX_HISTORY_CHAIN_BYTES, MAX_HISTORY_CHAIN_SEGMENTS, MAX_HISTORY_SEGMENT_BYTES, @@ -45,13 +58,19 @@ pub use plan::{ #[cfg(feature = "storage")] pub use pointer::PointerCatalogLookup; pub use pointer::{ - FileCatalogEntry, PointerCatalog, ShardCatalogEntry, XorbCatalogEntry, XorbChunkEntry, + FileCatalogEntry, PointerCatalog, PointerCatalogAssembly, ShardCatalogEntry, XorbCatalogEntry, + XorbChunkEntry, }; pub use ref_head::{ CAPSULE_REF_COMPACTION_FAN_IN, CAPSULE_REF_ROLLUP_WINDOW, CapsuleRefHead, CapsuleRefState, MAX_CAPSULE_REF_FRONTIER, MAX_CAPSULE_REF_HEADS, capsule_ref_name_from_key, capsule_ref_name_key, }; +#[cfg(feature = "storage")] +pub use ref_view::{ + RefControlFrontiers, VisibleRefHeads, apply_control_refs, materialize_visible_ref_heads, + order_ref_controls, select_ref_control_frontiers, select_ref_journal_controls, +}; pub use root::{ CapsulePointer, CheckpointPointer, GcFence, MAX_CAPSULE_FRONTIER, MAX_ROOT_BYTES, RepositoryRoot, RootRecord, diff --git a/crates/crab-metadata/src/capsule_protocol/pointer.rs b/crates/crab-metadata/src/capsule_protocol/pointer.rs index 2e9363aed..2b359154a 100644 --- a/crates/crab-metadata/src/capsule_protocol/pointer.rs +++ b/crates/crab-metadata/src/capsule_protocol/pointer.rs @@ -158,6 +158,36 @@ impl Default for PointerCatalog { } } +/// Assemble a captured visible catalog without assuming dependency order across refs. +/// +/// File bindings retain capture order; immutable conflicts fail immediately. +/// Dependency completeness is required before the catalog can leave the assembly. +#[must_use] +pub struct PointerCatalogAssembly { + catalog: PointerCatalog, +} + +impl PointerCatalogAssembly { + /// Start with the captured checkpoint catalog, or an empty catalog before checkpointing. + pub fn new(catalog: PointerCatalog) -> Self { + Self { catalog } + } + + /// Add one ordered visible delta, validating its rows and immutable conflicts. + /// + /// Discard the assembly if a delta fails. Dependencies may come from another + /// visible ref; callers must finish before using the resulting catalog. + pub fn apply(&mut self, delta: &PointerCatalog) -> Result<()> { + self.catalog.merge_delta(delta) + } + + /// Return the captured catalog only if its complete dependency closure is valid. + pub fn finish(self) -> Result { + self.catalog.validate(true, false)?; + Ok(self.catalog) + } +} + impl PointerCatalog { #[must_use] pub fn new() -> Self { @@ -190,6 +220,11 @@ impl PointerCatalog { /// Apply one publication delta, rejecting immutable identity conflicts. pub fn apply(&mut self, delta: &Self) -> Result<()> { + self.merge_delta(delta)?; + self.validate(true, false) + } + + fn merge_delta(&mut self, delta: &Self) -> Result<()> { delta.validate(false, false)?; for (hash, entry) in &delta.xorbs { self.insert_xorb(hash.clone(), entry.clone())?; @@ -209,7 +244,7 @@ impl PointerCatalog { } } } - self.validate(true, false) + Ok(()) } /// Encode this complete catalog as bounded authenticated page trees. @@ -446,6 +481,13 @@ mod tests { assert!(current.apply(&delta_with_base_xorb()).is_err()); } + #[test] + fn visible_assembly_rejects_unresolved_dependencies_at_finish() { + let mut assembly = PointerCatalogAssembly::new(PointerCatalog::new()); + assembly.apply(&delta_with_base_xorb()).unwrap(); + assert!(assembly.finish().is_err()); + } + #[test] fn apply_allows_new_file_mapping_but_not_new_size() { let mut current = catalog(); diff --git a/crates/crab-metadata/src/capsule_protocol/pointer/paged.rs b/crates/crab-metadata/src/capsule_protocol/pointer/paged.rs index 61346bb59..b7e3e9518 100644 --- a/crates/crab-metadata/src/capsule_protocol/pointer/paged.rs +++ b/crates/crab-metadata/src/capsule_protocol/pointer/paged.rs @@ -10,7 +10,7 @@ pub use reader::PointerCatalogLookup; const MAGIC: &[u8; 8] = b"CRBPCP01"; const TRAILER_BYTES: usize = 12; const MAX_DESCRIPTOR_BYTES: usize = 4096; -const MAX_CATALOG_BYTES: u64 = 2 * 1024 * 1024 * 1024; +pub(in crate::capsule_protocol) const MAX_CATALOG_BYTES: u64 = 2 * 1024 * 1024 * 1024; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] diff --git a/crates/crab-metadata/src/capsule_protocol/ref_view.rs b/crates/crab-metadata/src/capsule_protocol/ref_view.rs new file mode 100644 index 000000000..4230bc951 --- /dev/null +++ b/crates/crab-metadata/src/capsule_protocol/ref_view.rs @@ -0,0 +1,330 @@ +//! Shared captured ref values and transaction ordering for Git and catalog readers. +use crate::capsule_protocol::{ + CapsuleControl, CapsulePointer, CapsuleRunControl, RepositoryRoot, VisibilityReconciliation, +}; +use crate::error::{MetadataError, Result}; +use std::collections::{BTreeMap, BTreeSet}; + +/// Visible ref values, publication positions and immutable frontiers for one capture. +pub struct VisibleRefHeads { + pub refs: BTreeMap, + pub peeled_refs: BTreeMap, + pub transactions: BTreeMap, + /// First uncompacted transaction offset and its ordered physical runs per ref. + pub frontiers: BTreeMap)>, + pub pointers: Vec, +} + +/// Materialize only ref-head metadata from one epoch-filtered stable capture. +pub fn materialize_visible_ref_heads( + root: &crate::capsule_protocol::RepositoryRoot, + heads: &[crate::capsule_protocol::CapsuleRefHead], + active: &BTreeSet, +) -> Result { + let mut refs = root.refs().clone(); + let mut peeled_refs = root.peeled_refs().clone(); + let mut transactions = BTreeMap::new(); + let mut frontiers = BTreeMap::new(); + let mut pointers = root.capsule_frontier().to_vec(); + for head in heads { + let state = head.visible(active); + if let Some(transaction_id) = state.transaction_id() { + transactions.insert(head.ref_name().to_owned(), transaction_id.to_owned()); + } + if state.transaction_id() + == root + .compacted_ref_transactions() + .get(head.ref_name()) + .map(String::as_str) + { + continue; + } + match state.oid() { + Some(oid) => { + refs.insert(head.ref_name().to_owned(), oid.to_owned()); + match state.peeled_oid() { + Some(peeled) => { + peeled_refs.insert(head.ref_name().to_owned(), peeled.to_owned()); + } + None => { + peeled_refs.remove(head.ref_name()); + } + } + } + None => { + refs.remove(head.ref_name()); + peeled_refs.remove(head.ref_name()); + } + } + for pointer in state.frontier() { + match pointers + .iter() + .find(|candidate| candidate.hash() == pointer.hash()) + { + Some(candidate) if candidate != pointer => { + return Err(corrupt_path( + "capsule-protocol ref heads", + "capsule run identity has conflicting authenticated metadata", + )); + } + Some(_) => {} + None => pointers.push(pointer.clone()), + } + } + // Pointers commit to exact ordered transaction IDs; both run loaders + // verify that binding. Resolve the checkpoint offset once for all readers. + let ids = state + .frontier() + .iter() + .flat_map(|pointer| pointer.transaction_ids()) + .collect::>(); + let start = match root.compacted_ref_transactions().get(head.ref_name()) { + Some(compacted) => match ids.iter().position(|id| *id == compacted) { + Some(index) => index + 1, + None if state.checkpoint_transaction_id() == Some(compacted.as_str()) => 0, + None => { + return Err(frontier_contract(format!( + "ref {} does not extend its compacted transaction", + head.ref_name() + ))); + } + }, + None if state.checkpoint_transaction_id().is_none() => 0, + None => { + return Err(frontier_contract(format!( + "ref {} names a checkpoint absent from the repository root", + head.ref_name() + ))); + } + }; + frontiers.insert( + head.ref_name().to_owned(), + (start, state.frontier().to_vec()), + ); + } + Ok(VisibleRefHeads { + refs, + peeled_refs, + transactions, + frontiers, + pointers, + }) +} + +/// Order selected controls by captured ref predecessors and visibility readiness. +pub async fn order_ref_controls( + base_refs: &BTreeMap, + base_peeled: &BTreeMap, + mut visibility: VisibilityReconciliation, + predecessors: &BTreeMap>, + mut pending: BTreeMap, +) -> Result> { + let mut refs = base_refs.clone(); + let mut peeled = base_peeled.clone(); + let tracked = pending.keys().cloned().collect::>(); + let mut applied = BTreeSet::new(); + let mut ordered = Vec::with_capacity(pending.len()); + while !pending.is_empty() { + let mut ready = None; + for (id, capsule) in &pending { + if predecessors + .get(capsule.transaction_id()) + .is_some_and(|required| { + required + .iter() + .any(|id| tracked.contains(id) && !applied.contains(id)) + }) + { + continue; + } + if visibility.ready(capsule, &refs).await? { + ready = Some(id.clone()); + break; + } + } + let id = ready.ok_or_else(|| { + corrupt_path( + "capsule-protocol ref heads", + "capsule ref history is cyclic or does not extend the compacted root", + ) + })?; + let capsule = pending + .remove(&id) + .ok_or_else(|| internal("ready capsule control disappeared"))?; + apply_control_refs(&capsule, &mut refs, &mut peeled)?; + visibility.apply(&capsule).await?; + applied.insert(capsule.transaction_id().to_owned()); + ordered.push(capsule); + } + Ok(ordered) +} + +/// Apply expected-old-checked ref edits to captured ref and peeled-ref maps. +pub fn apply_control_refs( + capsule: &CapsuleControl, + refs: &mut BTreeMap, + peeled_refs: &mut BTreeMap, +) -> Result<()> { + for edit in capsule.transaction().edits() { + if refs.get(edit.ref_name()).map(String::as_str) != edit.expected_old() { + return Err(corrupt_path( + "capsule-protocol ref heads", + "capsule expected-old ref does not match its parent state", + )); + } + match edit.new_oid() { + Some(oid) => { + refs.insert(edit.ref_name().to_owned(), oid.to_owned()); + match edit.peeled_oid() { + Some(peeled) => { + peeled_refs.insert(edit.ref_name().to_owned(), peeled.to_owned()); + } + None => { + peeled_refs.remove(edit.ref_name()); + } + } + } + None => { + refs.remove(edit.ref_name()); + peeled_refs.remove(edit.ref_name()); + } + } + } + Ok(()) +} + +fn internal(reason: impl Into) -> MetadataError { + MetadataError::Internal(reason.into()) +} +fn corrupt_path(path: impl Into, reason: impl Into) -> MetadataError { + MetadataError::CorruptObject { + path: path.into(), + reason: reason.into(), + } +} + +/// Selected root controls and ref-frontier dependencies before visibility reconciliation. +pub struct RefControlFrontiers { + pub root_controls: Vec, + pub required: BTreeSet, + pub predecessors: BTreeMap>, + pub counts: BTreeMap, +} + +/// Resolve captured frontier offsets into verified physical run occurrences. +pub fn select_ref_control_frontiers( + root: &RepositoryRoot, + ref_frontiers: &BTreeMap)>, + runs: &BTreeMap)>, +) -> Result { + let mut required_transactions = BTreeSet::new(); + let mut transaction_predecessors = BTreeMap::>::new(); + let mut ref_capsule_counts = BTreeMap::new(); + for (ref_name, (start, frontier)) in ref_frontiers { + let transaction_ids = frontier + .iter() + .map(|pointer| { + runs.get(pointer.hash()) + .ok_or_else(|| internal("loaded capsule run control disappeared")) + }) + .collect::>>()? + .into_iter() + .flat_map(|(_, capsules)| capsules.iter().map(|capsule| capsule.transaction_id())) + .collect::>(); + let required = transaction_ids.get(*start..).ok_or_else(|| { + corrupt_path( + "capsule-protocol ref heads", + "captured frontier offset exceeds its validated transaction count", + ) + })?; + ref_capsule_counts.insert( + ref_name.clone(), + u32::try_from(required.len()).map_err(|_| internal("ref capsule count overflowed"))?, + ); + required_transactions.extend( + required + .iter() + .map(|transaction_id| (*transaction_id).to_owned()), + ); + for pair in required.windows(2) { + transaction_predecessors + .entry(pair[1].to_owned()) + .or_default() + .insert(pair[0].to_owned()); + } + } + let mut root_capsules = Vec::new(); + let mut root_capsule_identities = BTreeMap::new(); + for pointer in root.capsule_frontier() { + let (_, capsules) = runs + .get(pointer.hash()) + .ok_or_else(|| internal("loaded root capsule run control disappeared"))?; + for capsule in capsules { + match root_capsule_identities.get(capsule.transaction_id()) { + Some(existing) if existing != capsule => { + return Err(corrupt_path( + "capsule-protocol root frontier", + "transaction identity names conflicting capsules", + )); + } + Some(_) => {} + None => { + root_capsule_identities + .insert(capsule.transaction_id().to_owned(), capsule.clone()); + root_capsules.push(capsule.clone()); + } + } + } + } + + Ok(RefControlFrontiers { + root_controls: root_capsules, + required: required_transactions, + predecessors: transaction_predecessors, + counts: ref_capsule_counts, + }) +} + +/// Deduplicate only the journal transactions required by the captured ref frontiers. +pub fn select_ref_journal_controls( + root: &RepositoryRoot, + required: &BTreeSet, + runs: &BTreeMap)>, +) -> Result> { + let root_transactions = root + .capsule_frontier() + .iter() + .flat_map(|pointer| pointer.transaction_ids().iter().cloned()) + .collect::>(); + let mut journal_capsules = BTreeMap::new(); + for (_, capsules) in runs.values() { + for capsule in capsules { + if root_transactions.contains(capsule.transaction_id()) + || !required.contains(capsule.transaction_id()) + { + continue; + } + match journal_capsules.get(capsule.transaction_id()) { + Some(existing) if existing != capsule => { + return Err(corrupt_path( + "capsule-protocol ref heads", + "transaction identity names conflicting capsules", + )); + } + Some(_) => {} + None => { + journal_capsules.insert(capsule.transaction_id().to_owned(), capsule.clone()); + } + } + } + } + + Ok(journal_capsules) +} + +fn frontier_contract(reason: impl Into) -> MetadataError { + MetadataError::CapsuleContract { + record: "stored root", + reason: reason.into(), + } +} diff --git a/crates/crab-metadata/src/capsule_protocol/store.rs b/crates/crab-metadata/src/capsule_protocol/store.rs index d3ed7d433..8cc7fe82f 100644 --- a/crates/crab-metadata/src/capsule_protocol/store.rs +++ b/crates/crab-metadata/src/capsule_protocol/store.rs @@ -5,7 +5,7 @@ use futures_util::{StreamExt, TryStreamExt}; use object_store::ObjectMeta; use crate::capsule_protocol::{ - Capsule, CapsuleControl, CapsuleRefHead, CapsuleRun, CapsuleRunControl, CapsuleSectionKind, + CapsuleControl, CapsuleRefHead, CapsuleRun, CapsuleRunControl, CapsuleSectionKind, HistorySegment, HistorySegmentPointer, LayeredCheckpoint, MAX_CAPSULE_REF_HEADS, MAX_HISTORY_SEGMENT_BYTES, MAX_ROOT_BYTES, PointerCatalog, RootRecord, capsule_ref_name_key, }; @@ -480,198 +480,7 @@ pub async fn load_pointer_catalog_from_root( router: &StoreLayout, snapshot: &RootSnapshot, ) -> Result { - let root = snapshot.record().root().clone(); - let mut catalog = if let Some(pointer) = root.checkpoint() { - load_layered_checkpoint(router, pointer) - .await? - .pointer_catalog()? - } else { - PointerCatalog::new() - }; - for pointer in root.capsule_frontier() { - let run = load_run(router, pointer).await?; - for capsule in run.capsules() { - if let Some(delta) = capsule.pointer_catalog_delta()? { - catalog.apply(&delta)?; - } - } - } - for capsule in load_visible_ref_capsules(router, &root, &catalog).await? { - if let Some(delta) = capsule.pointer_catalog_delta()? { - catalog.apply(&delta)?; - } - } - Ok(catalog) -} - -async fn load_visible_ref_capsules( - router: &StoreLayout, - root: &super::RepositoryRoot, - base_catalog: &PointerCatalog, -) -> Result> { - let (heads, active) = capture_ref_heads(router, root).await?; - - let mut pointers = BTreeMap::new(); - let mut sequences = BTreeMap::new(); - let mut expected_refs = root.refs().clone(); - for head in heads { - let state = head.visible(&active); - if state.transaction_id() - == root - .compacted_ref_transactions() - .get(head.ref_name()) - .map(String::as_str) - { - continue; - } - match state.oid() { - Some(oid) => { - expected_refs.insert(head.ref_name().to_owned(), oid.to_owned()); - } - None => { - expected_refs.remove(head.ref_name()); - } - } - for pointer in state.frontier() { - match pointers.get(pointer.hash()) { - Some(existing) if existing != pointer => { - return Err(contract_error( - "capsule run identity has conflicting authenticated metadata", - )); - } - Some(_) => {} - None => { - pointers.insert(pointer.hash().to_owned(), pointer.clone()); - } - } - } - sequences.insert( - head.ref_name().to_owned(), - ( - state.checkpoint_transaction_id().map(str::to_owned), - state.frontier().to_vec(), - ), - ); - } - - let run_router = router.clone(); - let runs = futures_util::stream::iter(pointers.into_values().map(move |pointer| { - let router = run_router.clone(); - async move { - load_run(&router, &pointer) - .await - .map(|run| (run.hash().to_owned(), run)) - } - })) - .buffer_unordered(32) - .try_collect::>() - .await?; - let mut required = BTreeSet::new(); - for (ref_name, (checkpoint_transaction_id, frontier)) in sequences { - let transaction_ids = frontier - .iter() - .flat_map(|pointer| { - runs.get(pointer.hash()) - .into_iter() - .flat_map(|run| run.capsules().iter().map(Capsule::transaction_id)) - }) - .collect::>(); - let start = match root.compacted_ref_transactions().get(&ref_name) { - Some(compacted) => match transaction_ids - .iter() - .position(|transaction_id| *transaction_id == compacted) - { - Some(index) => index + 1, - None if checkpoint_transaction_id.as_deref() == Some(compacted.as_str()) => 0, - None => { - return Err(contract_error(format!( - "ref {ref_name} does not extend its compacted transaction" - ))); - } - }, - None if checkpoint_transaction_id.is_none() => 0, - None => { - return Err(contract_error(format!( - "ref {ref_name} names a checkpoint absent from the repository root" - ))); - } - }; - required.extend( - transaction_ids[start..] - .iter() - .map(|transaction_id| (*transaction_id).to_owned()), - ); - } - let mut pending = BTreeMap::new(); - for capsule in runs - .values() - .flat_map(|run| run.capsules()) - .filter(|capsule| required.contains(capsule.transaction_id())) - { - match pending.get(capsule.transaction_id()) { - Some(existing) if existing != capsule => { - return Err(contract_error( - "transaction identity names conflicting capsules", - )); - } - Some(_) => {} - None => { - pending.insert(capsule.transaction_id().to_owned(), capsule.clone()); - } - } - } - let mut refs = root.refs().clone(); - let mut catalog = base_catalog.clone(); - let mut ordered = Vec::with_capacity(pending.len()); - while !pending.is_empty() { - let mut ready = None; - for (id, capsule) in &pending { - let transaction = capsule.transaction()?; - if transaction - .edits() - .iter() - .any(|edit| refs.get(edit.ref_name()).map(String::as_str) != edit.expected_old()) - { - continue; - } - let mut candidate = catalog.clone(); - if let Some(delta) = capsule.pointer_catalog_delta()? - && candidate.apply(&delta).is_err() - { - continue; - } - ready = Some(id.clone()); - break; - } - let Some(ready) = ready else { - return Err(contract_error( - "capsule ref history is cyclic or has an unsatisfied catalog dependency", - )); - }; - let capsule = pending - .remove(&ready) - .ok_or_else(|| contract_error("ready capsule disappeared"))?; - for edit in capsule.transaction()?.edits() { - match edit.new_oid() { - Some(oid) => { - refs.insert(edit.ref_name().to_owned(), oid.to_owned()); - } - None => { - refs.remove(edit.ref_name()); - } - } - } - if let Some(delta) = capsule.pointer_catalog_delta()? { - catalog.apply(&delta)?; - } - ordered.push(capsule); - } - if refs != expected_refs { - return Err(contract_error( - "materialized capsules do not match visible ref-head state", - )); - } - Ok(ordered) + super::catalog_capture::materialize_from_root(router, snapshot).await } /// Capture one stable, epoch-filtered ref-head view and its committed activations. diff --git a/crates/crab-metadata/src/capsule_protocol/store/tests.rs b/crates/crab-metadata/src/capsule_protocol/store/tests.rs index e407e0061..45a3729dd 100644 --- a/crates/crab-metadata/src/capsule_protocol/store/tests.rs +++ b/crates/crab-metadata/src/capsule_protocol/store/tests.rs @@ -2,7 +2,7 @@ use super::*; use crate::capsule_protocol::{ - CapsuleRefEdit, CapsuleRefState, CapsuleTransaction, RepositoryRoot, + Capsule, CapsuleRefEdit, CapsuleRefState, CapsuleTransaction, RepositoryRoot, }; use std::sync::Arc; @@ -571,3 +571,139 @@ async fn ref_capture_rejects_uncommitted_base_changes_only_in_the_active_epoch() } } } + +#[tokio::test] +async fn full_catalog_matches_selected_closure_across_independent_ref_dependencies() { + use crate::capsule_protocol::{ + CapsulePointer, CapsuleSection, FileCatalogEntry, ShardCatalogEntry, XorbCatalogEntry, + XorbChunkEntry, load_pointer_catalog_lookup_from_root, + }; + let layout = StoreLayout::new( + Store::new(Arc::new(object_store::memory::InMemory::new())), + "repositories/catalog-order".to_owned(), + ); + let root = create_root( + &layout, + RootRecord::encode(RepositoryRoot::initial(&"1".repeat(64), "refs/heads/main").unwrap()) + .unwrap(), + ) + .await + .unwrap(); + let descriptors = CapsuleTransaction::new( + root.record().digest(), + vec![CapsuleRefEdit::new( + "refs/tags/source", + None, + Some("1".repeat(40)), + None, + )], + ) + .unwrap(); + let descriptor_id = descriptors.id().unwrap(); + let mut binding = None; + for index in 0..256 { + let candidate = CapsuleTransaction::new( + root.record().digest(), + vec![CapsuleRefEdit::new( + format!("refs/tags/consumer-{index}"), + None, + Some("2".repeat(40)), + None, + )], + ) + .unwrap(); + if candidate.id().unwrap() < descriptor_id { + binding = Some(candidate); + break; + } + } + let binding = binding.expect("fixture must put its dependent binding first in hash order"); + let file = "3".repeat(64); + let shard = "4".repeat(64); + let xorb = "5".repeat(64); + let mut complete = PointerCatalog::new(); + complete + .insert_xorb( + xorb.clone(), + XorbCatalogEntry::new( + 99, + "6".repeat(64), + vec![XorbChunkEntry::new("7".repeat(64), 16)], + ), + ) + .unwrap(); + complete + .insert_shard(shard.clone(), ShardCatalogEntry::new(80, vec![xorb])) + .unwrap(); + let mut file_delta = PointerCatalog::new(); + file_delta + .insert_file(file.clone(), FileCatalogEntry::new(16, shard)) + .unwrap(); + for (transaction, delta) in [(&descriptors, &complete), (&binding, &file_delta)] { + let capsule = Capsule::build( + transaction, + vec![], + vec![CapsuleSection::new( + CapsuleSectionKind::CatalogDelta, + delta.encode_delta().unwrap(), + )], + ) + .unwrap(); + let run = CapsuleRun::leaf(capsule).unwrap(); + layout + .store() + .create_strict(&layout.capsule_path(run.hash()), run.bytes().clone()) + .await + .unwrap(); + let pointer = CapsulePointer::new( + run.hash(), + run.bytes().len() as u64, + run.control_offset(), + run.control_size(), + run.footer_hash(), + run.level(), + run.transaction_ids(), + run.newest_base_root_digest(), + ) + .unwrap(); + let edit = &transaction.edits()[0]; + let head = CapsuleRefHead::from_root( + edit.ref_name(), + root.record().root().ref_epoch().to_owned(), + None, + None, + ) + .unwrap(); + let state = head + .successor_state( + &BTreeSet::new(), + edit.new_oid().map(str::to_owned), + None, + transaction.id().unwrap(), + vec![pointer], + ) + .unwrap(); + layout + .store() + .create_strict( + &layout.capsule_ref_head_path(&capsule_ref_name_key(edit.ref_name())), + head.commit(state).unwrap().encode().unwrap(), + ) + .await + .unwrap(); + } + complete.apply(&file_delta).unwrap(); + let selected = load_pointer_catalog_lookup_from_root(&layout, &root) + .await + .unwrap() + .files(layout.store(), &[file]) + .await + .unwrap(); + assert_eq!(selected, complete); + assert_eq!( + load_pointer_catalog_from_root(&layout, &root) + .await + .unwrap(), + selected + ); +} diff --git a/crates/crab-metadata/src/file_index_lookup.rs b/crates/crab-metadata/src/file_index_lookup.rs index 872418e4b..c8f3f4c9b 100644 --- a/crates/crab-metadata/src/file_index_lookup.rs +++ b/crates/crab-metadata/src/file_index_lookup.rs @@ -1,4 +1,4 @@ -//! Read-only file-index lookup over a repository's SlateDB metadata. +//! Snapshot-bound file/shard lookup over authenticated repository metadata. //! //! This Module owns the read Interface for `file_hash -> shard_hash` point //! lookups. It does not own metadata writes, CLI error presentation, metrics, @@ -225,7 +225,6 @@ async fn lookup_committed_record( /// closes the underlying SlateDB reader when consumed. pub struct FileIndexLookupSession { catalog: Option>, - static_entries: Option>>, reader: Option>, anchor: Option, storage: crab_storage::Store, @@ -246,7 +245,6 @@ impl FileIndexLookupSession { ) -> Result { Ok(Self { catalog: None, - static_entries: None, reader: None, anchor: CommittedShardAnchor::from_snapshot(snapshot)?, storage: router.store().clone(), @@ -310,7 +308,6 @@ impl FileIndexLookupSession { let anchor = CommittedShardAnchor::from_snapshot(snapshot)?; Ok(Self { catalog: None, - static_entries: None, reader: None, anchor, storage: router.store().clone(), @@ -371,7 +368,6 @@ impl FileIndexLookupSession { }; Ok(Self { catalog: None, - static_entries: None, reader: None, anchor, storage: router.store().clone(), @@ -393,33 +389,10 @@ impl FileIndexLookupSession { // Only an absent root permits v1 lookup. Missing dependencies // under a verified v2 root must not revive older v1 authority. let catalog = - crate::capsule_protocol::load_pointer_catalog_from_root(&router, &root).await?; - let entries = catalog - .files() - .iter() - .map(|(file_hash, entry)| { - Ok(( - MerkleHash::from_hex(file_hash).map_err(|error| { - MetadataError::CorruptObject { - path: "capsule-protocol pointer catalog".to_owned(), - reason: format!("invalid file hash {file_hash}: {error}"), - } - })?, - MerkleHash::from_hex(entry.shard_hash()).map_err(|error| { - MetadataError::CorruptObject { - path: "capsule-protocol pointer catalog".to_owned(), - reason: format!( - "invalid shard hash {}: {error}", - entry.shard_hash() - ), - } - })?, - )) - }) - .collect::>>()?; + crate::capsule_protocol::load_pointer_catalog_lookup_from_root(&router, &root) + .await?; return Ok(Self { - catalog: None, - static_entries: Some(Arc::new(entries)), + catalog: Some(Arc::new(catalog)), reader: None, anchor: None, storage, @@ -444,7 +417,6 @@ impl FileIndexLookupSession { }; let mut session = Self { catalog: None, - static_entries: None, reader: None, anchor, storage, @@ -491,9 +463,6 @@ impl FileIndexLookupSession { .next() .flatten()); } - if let Some(entries) = &self.static_entries { - return Ok(entries.get(file_hash).copied()); - } if let Some(reader) = self.reader.as_ref() && let Some(record) = lookup_committed_record(reader, *file_hash, self.anchor.as_ref()).await? @@ -543,13 +512,6 @@ impl FileIndexLookupSession { }) .collect(); } - if let Some(entries) = &self.static_entries { - return Ok(file_hashes - .iter() - .map(|file_hash| entries.get(file_hash).copied()) - .collect()); - } - let records = self.lookup_committed_records_batch(file_hashes).await?; let mut out = records .into_iter() @@ -799,7 +761,6 @@ impl LookupSource { match self { Self::Catalog { router, lookup } => Ok(FileIndexLookupSession { catalog: Some(Arc::clone(lookup)), - static_entries: None, reader: None, anchor: None, storage: router.store().clone(), diff --git a/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs b/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs index 458aacdb7..f656df3f9 100644 --- a/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs +++ b/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs @@ -1,16 +1,18 @@ use super::*; use crate::capsule_protocol::{ - CapsuleGitPack, FileCatalogEntry, LayeredCheckpoint, PackLayer, PointerCatalog, - PointerCatalogLookup, ShardCatalogEntry, XorbCatalogEntry, XorbChunkEntry, + CapsuleGitPack, CheckpointPointer, FileCatalogEntry, LayeredCheckpoint, PackLayer, + PointerCatalog, PointerCatalogLookup, RepositoryRoot, RootRecord, ShardCatalogEntry, + XorbCatalogEntry, XorbChunkEntry, }; use bytes::Bytes; use object_store::memory::InMemory; +use std::collections::BTreeMap; use std::sync::atomic::AtomicUsize; pub(super) async fn capture( layout: &crab_storage::StoreLayout, catalog: PointerCatalog, -) -> (PointerCatalogLookup, object_store::path::Path) { +) -> (PointerCatalogLookup, object_store::path::Path, RootRecord) { let pack = CapsuleGitPack::new( Bytes::from_static(b"pack"), Bytes::from_static(b"index"), @@ -24,8 +26,103 @@ pub(super) async fn capture( .unwrap() .source_descriptor() .unwrap(); - let checkpoint = - LayeredCheckpoint::build(1, &"3".repeat(64), vec![source], catalog, None).unwrap(); + let initial = + RootRecord::encode(RepositoryRoot::initial(&"1".repeat(64), "refs/heads/main").unwrap()) + .unwrap(); + let transaction = crate::capsule_protocol::CapsuleTransaction::new( + initial.digest(), + vec![crate::capsule_protocol::CapsuleRefEdit::new( + "refs/heads/main", + None, + Some("1".repeat(40)), + None, + )], + ) + .unwrap(); + let run = crate::capsule_protocol::CapsuleRun::leaf( + crate::capsule_protocol::Capsule::build(&transaction, vec![], vec![]).unwrap(), + ) + .unwrap(); + let run_pointer = crate::capsule_protocol::CapsulePointer::new( + run.hash(), + run.bytes().len() as u64, + run.control_offset(), + run.control_size(), + run.footer_hash(), + run.level(), + run.transaction_ids(), + run.newest_base_root_digest(), + ) + .unwrap(); + layout + .store() + .put_overwrite(&layout.capsule_path(run.hash()), run.bytes().clone()) + .await + .unwrap(); + let base = RootRecord::encode( + initial + .root() + .advance( + initial.digest(), + BTreeMap::from([("refs/heads/main".to_owned(), "1".repeat(40))]), + BTreeMap::new(), + vec![run_pointer], + transaction.id().unwrap().as_str(), + ) + .unwrap(), + ) + .unwrap(); + let checkpoint = LayeredCheckpoint::build( + base.root().generation(), + base.digest(), + vec![source], + catalog, + None, + ) + .unwrap(); + let pointer = CheckpointPointer::new_layered( + checkpoint.hash(), + checkpoint.bytes().len() as u64, + checkpoint.control_offset(), + checkpoint.control_size(), + checkpoint.footer_hash(), + checkpoint.covered_generation(), + checkpoint.covered_root_digest(), + checkpoint.pack_count().unwrap(), + checkpoint.object_count().unwrap(), + ) + .unwrap(); + let history = crate::capsule_protocol::HistorySegment::build( + pointer.clone(), + None, + crate::capsule_protocol::HistorySegmentState::new( + base.root().refs().clone(), + base.root().peeled_refs().clone(), + base.root().head().to_owned(), + base.root().compacted_ref_transactions().clone(), + base.root().capsule_frontier().to_vec(), + ), + ) + .unwrap(); + layout + .store() + .put_overwrite( + &layout.capsule_history_segment_path(history.hash()), + history.bytes().clone(), + ) + .await + .unwrap(); + let root = RootRecord::encode( + base.root() + .install_checkpoint(base.digest(), pointer, Some(history.pointer().unwrap())) + .unwrap(), + ) + .unwrap(); + layout + .store() + .put_overwrite(&layout.capsule_root_path(), root.bytes().clone()) + .await + .unwrap(); let path = layout.capsule_checkpoint_path(checkpoint.hash()); layout .store() @@ -45,6 +142,7 @@ pub(super) async fn capture( ( control.pointer_catalog_lookup(layout, 512 * 1024).unwrap(), path, + root, ) } @@ -87,7 +185,7 @@ async fn pinned_batches_preserve_order_misses_and_page_bounds_as_history_grows() observed.fetch_add(1, Ordering::SeqCst); })); let layout = crab_storage::StoreLayout::new(storage, "pinned-lookup".to_owned()); - let (captured, _) = capture(&layout, catalog(count)).await; + let (captured, _, _) = capture(&layout, catalog(count)).await; let lookup = SharedFileIndexLookup::for_pointer_catalog(layout.clone(), captured); assert_eq!( reads.load(Ordering::SeqCst), @@ -126,7 +224,7 @@ async fn missing_captured_pages_are_errors_even_when_a_new_catalog_exists() { crab_storage::Store::new(Arc::new(InMemory::new())), "missing-page".to_owned(), ); - let (captured, path) = capture(&layout, catalog(100)).await; + let (captured, path, _) = capture(&layout, catalog(100)).await; capture(&layout, catalog(101)).await; layout.store().delete(&path).await.unwrap(); let lookup = SharedFileIndexLookup::for_pointer_catalog(layout, captured); @@ -138,3 +236,53 @@ async fn missing_captured_pages_are_errors_even_when_a_new_catalog_exists() { ); lookup.close().await.unwrap(); } + +#[tokio::test] +async fn current_state_opener_keeps_catalog_pages_selected_and_pins_its_first_root() { + use std::sync::atomic::AtomicU64; + for count in [100, 20_000] { + let reads = Arc::new(AtomicUsize::new(0)); + let bytes = Arc::new(AtomicU64::new(0)); + let observed_reads = reads.clone(); + let observed_bytes = bytes.clone(); + let storage = crab_storage::Store::new(Arc::new(InMemory::new())) + .with_read_request_observer(Arc::new(move |_| { + observed_reads.fetch_add(1, Ordering::SeqCst); + })) + .with_read_byte_observer(Arc::new(move |size| { + observed_bytes.fetch_add(size, Ordering::SeqCst); + })); + let layout = crab_storage::StoreLayout::new(storage.clone(), "current-catalog".to_owned()); + let (_, _, root) = capture(&layout, catalog(count)).await; + assert_eq!(root.root().generation(), 1); + let lookup = SharedFileIndexLookup::new_for_storage(&storage, layout.repo_prefix()); + assert_eq!(reads.load(Ordering::SeqCst), 0); + let file = super::tests::hash_from_seed(42); + let shard = super::tests::hash_from_seed(1_000_042); + assert_eq!( + lookup.lookup_batch(&[file, file]).await.unwrap(), + vec![Some(shard), Some(shard)] + ); + assert!( + reads.load(Ordering::SeqCst) <= 32 && bytes.load(Ordering::SeqCst) <= 512 * 1024, + "current-state fixed query materialized unrelated catalog history: count={count}, reads={}, bytes={}", + reads.load(Ordering::SeqCst), + bytes.load(Ordering::SeqCst) + ); + let (_, _, later_root) = capture(&layout, catalog(count + 1)).await; + layout + .store() + .put_overwrite(&layout.capsule_root_path(), later_root.bytes().clone()) + .await + .unwrap(); + let later_file = super::tests::hash_from_seed(count as u64); + assert_eq!(lookup.lookup(&later_file).await.unwrap(), None); + let latest = SharedFileIndexLookup::new_for_storage(&storage, layout.repo_prefix()); + assert_eq!( + latest.lookup(&later_file).await.unwrap(), + Some(super::tests::hash_from_seed(1_000_000 + count as u64)) + ); + lookup.close().await.unwrap(); + latest.close().await.unwrap(); + } +} diff --git a/crates/crab-read/README.md b/crates/crab-read/README.md index 8a6ea11d4..eb53e0b23 100644 --- a/crates/crab-read/README.md +++ b/crates/crab-read/README.md @@ -61,6 +61,12 @@ work. `git_visibility_index`, `candidate_git_visibility` and `tip_bound_transitions` asynchronously materialize captured controls only when bulk inspection needs them. They do not recapture refs or switch origins. +Metadata owns the stable ref capture, control selection, visibility reconciliation +and transaction ordering used by Git views and current-state catalog lookup. +Current-state file sessions capture their first root once and query selected +file/shard/xorb pages; they do not materialize the complete catalog or open +SlateDB under a v2 root. Full maintenance inspection remains explicit. + Pinned hydration and remote-mount setup pass captured paged catalog roots into metadata's shared file lookup. A lookup loads only its selected file dependency closure, preserving the captured publication order and immutable descriptor diff --git a/crates/crab-read/src/capsule_protocol.rs b/crates/crab-read/src/capsule_protocol.rs index 8fb1df7cb..17941f462 100644 --- a/crates/crab-read/src/capsule_protocol.rs +++ b/crates/crab-read/src/capsule_protocol.rs @@ -2,10 +2,15 @@ use bytes::Bytes; use crab_metadata::capsule_protocol::{ - Capsule, CapsuleControl, CapsulePointer, CapsuleRun, CapsuleRunControl, CheckpointPointer, - LayeredCheckpoint, PackMemberDescriptor, PackRange, PackSourceDescriptor, PackSourceKind, - PointerCatalog, RootRecord, capture_ref_heads, load_root, resolve_referenced_activations, - visibility_object_set_digest, + Capsule, CapsuleControl, CapsulePointer, CapsuleRun, CapsuleRunControl, CapturedVisibility, + CheckpointPointer, LayeredCheckpoint, PackMemberDescriptor, PackRange, PackSourceDescriptor, + PackSourceKind, PointerCatalog, PointerCatalogAssembly, RootRecord, VisibilityReconciliation, + apply_transaction_visibility_index, capture_ref_heads, load_root, + resolve_referenced_activations, visibility_is_ready, visibility_object_set_digest, +}; +use crab_metadata::capsule_protocol::{ + apply_control_refs, materialize_visible_ref_heads, order_ref_controls, + select_ref_control_frontiers, select_ref_journal_controls, }; use crab_storage::{Store, StoreLayout}; use futures_util::future::try_join_all; @@ -18,7 +23,8 @@ use tokio_util::sync::CancellationToken; use crate::{ReadError, Result}; -mod visibility; +#[cfg(test)] +mod visibility_tests; const LAYERED_SIDECAR_COALESCE_GAP_BYTES: u64 = 64 * 1024; const LAYERED_SIDECAR_MAX_EXTRA_BYTES: u64 = 4 * 1024 * 1024; @@ -783,7 +789,7 @@ impl CapsuleRepositoryView { self.catalog_sources .as_ref() .ok_or(ReadError::IncompleteCatalogView)?; - let mut catalog = + let base_catalog = match self.checkpoint.as_ref() { Some(checkpoint) if checkpoint.is_control_only() => { let layout = self.metadata_layout.as_ref().ok_or_else(|| { @@ -800,6 +806,7 @@ impl CapsuleRepositoryView { Some(checkpoint) => checkpoint.pointer_catalog()?, None => PointerCatalog::new(), }; + let mut catalog = PointerCatalogAssembly::new(base_catalog); for capsule in &self.capsules { if let Some(delta) = capsule.pointer_catalog_delta()? { catalog.apply(&delta)?; @@ -813,7 +820,7 @@ impl CapsuleRepositoryView { catalog.apply(&delta)?; } } - Ok(catalog) + Ok(catalog.finish()?) } /// Open the authenticated embedded Git packs as a filesystem-free repository. @@ -4553,7 +4560,7 @@ async fn assemble_view( let mut required_transactions = BTreeSet::new(); let mut transaction_predecessors = BTreeMap::>::new(); let mut ref_capsule_counts = BTreeMap::new(); - for (ref_name, (checkpoint_transaction_id, frontier)) in &ref_frontiers { + for (ref_name, (start, frontier)) in &ref_frontiers { let transaction_ids = frontier .iter() .map(|pointer| { @@ -4564,34 +4571,12 @@ async fn assemble_view( .into_iter() .flat_map(|run| run.capsules().iter().map(Capsule::transaction_id)) .collect::>(); - let start = match snapshot - .record() - .root() - .compacted_ref_transactions() - .get(ref_name) - { - Some(compacted) => match transaction_ids - .iter() - .position(|transaction_id| *transaction_id == compacted) - { - Some(index) => index + 1, - None if checkpoint_transaction_id.as_deref() == Some(compacted.as_str()) => 0, - None => { - return Err(corrupt_path( - "capsule-protocol ref heads", - format!("ref {ref_name} does not extend its compacted transaction"), - )); - } - }, - None if checkpoint_transaction_id.is_none() => 0, - None => { - return Err(corrupt_path( - "capsule-protocol ref heads", - format!("ref {ref_name} names a checkpoint absent from the repository root"), - )); - } - }; - let required = &transaction_ids[start..]; + let required = transaction_ids.get(*start..).ok_or_else(|| { + corrupt_path( + "capsule-protocol ref heads", + "captured frontier offset exceeds its validated transaction count", + ) + })?; ref_capsule_counts.insert( ref_name.clone(), u32::try_from(required.len()) @@ -4851,95 +4836,9 @@ async fn assemble_layered_control_view( capsule_run_sources.push(source); } } - let root_transactions = snapshot - .record() - .root() - .capsule_frontier() - .iter() - .flat_map(|pointer| pointer.transaction_ids().iter().cloned()) - .collect::>(); - let mut required_transactions = BTreeSet::new(); - let mut transaction_predecessors = BTreeMap::>::new(); - let mut ref_capsule_counts = BTreeMap::new(); - for (ref_name, (checkpoint_transaction_id, frontier)) in &ref_frontiers { - let transaction_ids = frontier - .iter() - .map(|pointer| { - runs.get(pointer.hash()) - .ok_or_else(|| ReadError::internal("loaded capsule run control disappeared")) - }) - .collect::>>()? - .into_iter() - .flat_map(|(_, capsules)| capsules.iter().map(|capsule| capsule.transaction_id())) - .collect::>(); - let start = match snapshot - .record() - .root() - .compacted_ref_transactions() - .get(ref_name) - { - Some(compacted) => match transaction_ids - .iter() - .position(|transaction_id| *transaction_id == compacted) - { - Some(index) => index + 1, - None if checkpoint_transaction_id.as_deref() == Some(compacted.as_str()) => 0, - None => { - return Err(corrupt_path( - "capsule-protocol ref heads", - format!("ref {ref_name} does not extend its compacted transaction"), - )); - } - }, - None if checkpoint_transaction_id.is_none() => 0, - None => { - return Err(corrupt_path( - "capsule-protocol ref heads", - format!("ref {ref_name} names a checkpoint absent from the repository root"), - )); - } - }; - let required = &transaction_ids[start..]; - ref_capsule_counts.insert( - ref_name.clone(), - u32::try_from(required.len()) - .map_err(|_| ReadError::internal("ref capsule count overflowed"))?, - ); - required_transactions.extend( - required - .iter() - .map(|transaction_id| (*transaction_id).to_owned()), - ); - for pair in required.windows(2) { - transaction_predecessors - .entry(pair[1].to_owned()) - .or_default() - .insert(pair[0].to_owned()); - } - } - let mut root_capsules = Vec::new(); - let mut root_capsule_identities = BTreeMap::new(); - for pointer in snapshot.record().root().capsule_frontier() { - let (_, capsules) = runs - .get(pointer.hash()) - .ok_or_else(|| ReadError::internal("loaded root capsule run control disappeared"))?; - for capsule in capsules { - match root_capsule_identities.get(capsule.transaction_id()) { - Some(existing) if existing != capsule => { - return Err(corrupt_path( - "capsule-protocol root frontier", - "transaction identity names conflicting capsules", - )); - } - Some(_) => {} - None => { - root_capsule_identities - .insert(capsule.transaction_id().to_owned(), capsule.clone()); - root_capsules.push(capsule.clone()); - } - } - } - } + let selected = select_ref_control_frontiers(snapshot.record().root(), &ref_frontiers, &runs)?; + let mut root_capsules = selected.root_controls; + let ref_capsule_counts = selected.counts; if footer_only { return Ok(CapsuleRepositoryView { root: snapshot, @@ -4964,28 +4863,9 @@ async fn assemble_layered_control_view( frontier_object_admission, }); } - let mut journal_capsules = BTreeMap::new(); - for (_, capsules) in runs.values() { - for capsule in capsules { - if root_transactions.contains(capsule.transaction_id()) - || !required_transactions.contains(capsule.transaction_id()) - { - continue; - } - match journal_capsules.get(capsule.transaction_id()) { - Some(existing) if existing != capsule => { - return Err(corrupt_path( - "capsule-protocol ref heads", - "transaction identity names conflicting capsules", - )); - } - Some(_) => {} - None => { - journal_capsules.insert(capsule.transaction_id().to_owned(), capsule.clone()); - } - } - } - } + let journal_capsules = + select_ref_journal_controls(snapshot.record().root(), &selected.required, &runs)?; + let transaction_predecessors = selected.predecessors; let checkpoint_visibility = if matches!(read, ControlRead::Git) { checkpoint .as_ref() @@ -5000,22 +4880,18 @@ async fn assemble_layered_control_view( let names = snapshot.record().root().refs().keys().cloned(); let initial = match &checkpoint_visibility { Some(reader) => { - visibility::Visibility::from_checkpoint( - router.store().clone(), - reader.clone(), - names, - ) - .await? + CapturedVisibility::from_checkpoint(router.store().clone(), reader.clone(), names) + .await? } - None => visibility::Visibility::new( + None => CapturedVisibility::new( router.store().clone(), visibility_index_for_checkpoint(None)?, names, ), }; - visibility::Reconciliation::Paged(Box::new(initial)) + VisibilityReconciliation::Paged(Box::new(initial)) } else { - visibility::Reconciliation::Complete(visibility_index_for_checkpoint(checkpoint.as_ref())?) + VisibilityReconciliation::Complete(visibility_index_for_checkpoint(checkpoint.as_ref())?) }; for capsule in &root_capsules { visibility.apply(capsule).await?; @@ -5139,14 +5015,6 @@ fn extend_frontier_object_admission( } } -struct VisibleRefHeads { - refs: BTreeMap, - peeled_refs: BTreeMap, - transactions: BTreeMap, - frontiers: BTreeMap, Vec)>, - pointers: Vec, -} - fn state_digest(root: &RootRecord, transactions: &BTreeMap) -> String { let mut hasher = blake3::Hasher::new(); hasher.update(b"crab capsule repository view v2\0"); @@ -5159,78 +5027,6 @@ fn state_digest(root: &RootRecord, transactions: &BTreeMap) -> S hasher.finalize().to_hex().to_string() } -fn materialize_visible_ref_heads( - root: &crab_metadata::capsule_protocol::RepositoryRoot, - heads: &[crab_metadata::capsule_protocol::CapsuleRefHead], - active: &BTreeSet, -) -> Result { - let mut refs = root.refs().clone(); - let mut peeled_refs = root.peeled_refs().clone(); - let mut transactions = BTreeMap::new(); - let mut frontiers = BTreeMap::new(); - let mut pointers = root.capsule_frontier().to_vec(); - for head in heads { - let state = head.visible(active); - if let Some(transaction_id) = state.transaction_id() { - transactions.insert(head.ref_name().to_owned(), transaction_id.to_owned()); - } - if state.transaction_id() - == root - .compacted_ref_transactions() - .get(head.ref_name()) - .map(String::as_str) - { - continue; - } - match state.oid() { - Some(oid) => { - refs.insert(head.ref_name().to_owned(), oid.to_owned()); - match state.peeled_oid() { - Some(peeled) => { - peeled_refs.insert(head.ref_name().to_owned(), peeled.to_owned()); - } - None => { - peeled_refs.remove(head.ref_name()); - } - } - } - None => { - refs.remove(head.ref_name()); - peeled_refs.remove(head.ref_name()); - } - } - for pointer in state.frontier() { - match pointers - .iter() - .find(|candidate| candidate.hash() == pointer.hash()) - { - Some(candidate) if candidate != pointer => { - return Err(corrupt_path( - "capsule-protocol ref heads", - "capsule run identity has conflicting authenticated metadata", - )); - } - Some(_) => {} - None => pointers.push(pointer.clone()), - } - } - frontiers.insert( - head.ref_name().to_owned(), - ( - state.checkpoint_transaction_id().map(str::to_owned), - state.frontier().to_vec(), - ), - ); - } - Ok(VisibleRefHeads { - refs, - peeled_refs, - transactions, - frontiers, - pointers, - }) -} - async fn capture_selected_ref_heads( router: &StoreLayout, root: &crab_metadata::capsule_protocol::RepositoryRoot, @@ -5358,175 +5154,6 @@ fn order_ref_capsules( Ok(ordered) } -async fn order_ref_controls( - base_refs: &BTreeMap, - base_peeled: &BTreeMap, - mut visibility: visibility::Reconciliation, - predecessors: &BTreeMap>, - mut pending: BTreeMap, -) -> Result> { - let mut refs = base_refs.clone(); - let mut peeled = base_peeled.clone(); - let tracked = pending.keys().cloned().collect::>(); - let mut applied = BTreeSet::new(); - let mut ordered = Vec::with_capacity(pending.len()); - while !pending.is_empty() { - let mut ready = None; - for (id, capsule) in &pending { - if predecessors - .get(capsule.transaction_id()) - .is_some_and(|required| { - required - .iter() - .any(|id| tracked.contains(id) && !applied.contains(id)) - }) - { - continue; - } - if visibility.ready(capsule, &refs).await? { - ready = Some(id.clone()); - break; - } - } - let id = ready.ok_or_else(|| { - corrupt_path( - "capsule-protocol ref heads", - "capsule ref history is cyclic or does not extend the compacted root", - ) - })?; - let capsule = pending - .remove(&id) - .ok_or_else(|| ReadError::internal("ready capsule control disappeared"))?; - apply_control_refs(&capsule, &mut refs, &mut peeled)?; - visibility.apply(&capsule).await?; - applied.insert(capsule.transaction_id().to_owned()); - ordered.push(capsule); - } - Ok(ordered) -} - -fn control_is_ready( - capsule: &CapsuleControl, - refs: &BTreeMap, - visibility_index: &crab_metadata::git_visibility::GitVisibilityIndex, -) -> Result { - visibility_is_ready( - capsule.transaction(), - capsule.visibility_delta()?, - refs, - visibility_index, - ) -} - -fn visibility_is_ready( - transaction: &crab_metadata::capsule_protocol::CapsuleTransaction, - delta: Option<&crab_metadata::capsule_protocol::CapsuleVisibilityDelta>, - refs: &BTreeMap, - visibility_index: &crab_metadata::git_visibility::GitVisibilityIndex, -) -> Result { - if transaction - .edits() - .iter() - .any(|edit| refs.get(edit.ref_name()).map(String::as_str) != edit.expected_old()) - { - return Ok(false); - } - let Some(delta) = delta else { - return Ok(true); - }; - // Transactions have unique refs. Count matched rows without copying OID lists; - // unmatched evidence still fails before it can authorize a publication. - let evidence = delta.edits(); - let mut matched = 0; - for edit in transaction.edits() { - let Some(new_oid) = edit.new_oid() else { - if evidence.get(edit.ref_name()).is_some() { - return Err(corrupt_path( - "capsule Git visibility", - "deleted ref has visibility evidence", - )); - } - continue; - }; - let Some(visibility) = evidence.get(edit.ref_name()) else { - return Err(corrupt_path( - "capsule Git visibility", - "live ref edit has no visibility evidence", - )); - }; - matched += 1; - visibility.validate()?; - if visibility.new_oid != new_oid { - return Err(corrupt_path( - "capsule Git visibility", - "visibility evidence does not match its ref edit", - )); - } - match edit.expected_old() { - Some(expected_old) => { - if visibility.old_oid.as_deref() != Some(expected_old) { - return Err(corrupt_path( - "capsule Git visibility", - "visibility evidence does not match the expected old ref", - )); - } - if !visibility_index.contains_ref(edit.ref_name()) { - return Ok(false); - } - } - None if visibility.replaces => {} - None => { - let Some(old_oid) = visibility.old_oid.as_deref() else { - continue; - }; - if !visibility_index.contains_hex_in_any_ref(old_oid) { - return Ok(false); - } - } - } - } - if evidence.len() != matched { - return Err(corrupt_path( - "capsule Git visibility", - "visibility evidence contains an uncommitted ref", - )); - } - Ok(true) -} - -fn apply_control_refs( - capsule: &CapsuleControl, - refs: &mut BTreeMap, - peeled_refs: &mut BTreeMap, -) -> Result<()> { - for edit in capsule.transaction().edits() { - if refs.get(edit.ref_name()).map(String::as_str) != edit.expected_old() { - return Err(corrupt_path( - "capsule-protocol ref heads", - "capsule expected-old ref does not match its parent state", - )); - } - match edit.new_oid() { - Some(oid) => { - refs.insert(edit.ref_name().to_owned(), oid.to_owned()); - match edit.peeled_oid() { - Some(peeled) => { - peeled_refs.insert(edit.ref_name().to_owned(), peeled.to_owned()); - } - None => { - peeled_refs.remove(edit.ref_name()); - } - } - } - None => { - refs.remove(edit.ref_name()); - peeled_refs.remove(edit.ref_name()); - } - } - } - Ok(()) -} - fn capsule_is_ready( capsule: &Capsule, refs: &BTreeMap, @@ -5534,68 +5161,19 @@ fn capsule_is_ready( ) -> Result { let transaction = capsule.transaction()?; let delta = capsule.visibility_delta()?; - visibility_is_ready(&transaction, delta.as_ref(), refs, visibility_index) + Ok(visibility_is_ready( + &transaction, + delta.as_ref(), + refs, + visibility_index, + )?) } fn apply_capsule_control_visibility_index( capsule: &CapsuleControl, index: &mut crab_metadata::git_visibility::GitVisibilityIndex, ) -> Result<()> { - apply_transaction_visibility_index(capsule.transaction(), capsule.visibility_delta()?, index) -} - -fn apply_transaction_visibility_index( - transaction: &crab_metadata::capsule_protocol::CapsuleTransaction, - delta: Option<&crab_metadata::capsule_protocol::CapsuleVisibilityDelta>, - index: &mut crab_metadata::git_visibility::GitVisibilityIndex, -) -> Result<()> { - let evidence = delta.map(|delta| delta.edits()); - let mut matched = 0; - for edit in transaction.edits() { - let Some(new_oid) = edit.new_oid() else { - if evidence - .and_then(|rows| rows.get(edit.ref_name())) - .is_some() - { - return Err(corrupt_path( - "capsule Git visibility", - "deleted ref has visibility evidence", - )); - } - index.remove_ref(edit.ref_name()); - continue; - }; - let visibility = evidence - .and_then(|rows| rows.get(edit.ref_name())) - .ok_or_else(|| { - corrupt_path( - "capsule Git visibility", - "live ref edit has no visibility evidence", - ) - })?; - matched += 1; - if visibility.new_oid != new_oid { - return Err(corrupt_path( - "capsule Git visibility", - "visibility evidence does not match its ref edit", - )); - } - if let Some(expected_old) = edit.expected_old() - && visibility.old_oid.as_deref() != Some(expected_old) - { - return Err(corrupt_path( - "capsule Git visibility", - "visibility evidence does not match the expected old ref", - )); - } - index.apply_ref_edit(edit.ref_name().to_owned(), visibility)?; - } - if evidence.is_some_and(|rows| rows.len() != matched) { - return Err(corrupt_path( - "capsule Git visibility", - "visibility evidence contains an uncommitted ref", - )); - } + apply_transaction_visibility_index(capsule.transaction(), capsule.visibility_delta()?, index)?; Ok(()) } @@ -5605,7 +5183,8 @@ fn apply_capsule_visibility_index( ) -> Result<()> { let transaction = capsule.transaction()?; let delta = capsule.visibility_delta()?; - apply_transaction_visibility_index(&transaction, delta.as_ref(), index) + apply_transaction_visibility_index(&transaction, delta.as_ref(), index)?; + Ok(()) } fn apply_capsule_refs( @@ -7670,7 +7249,13 @@ mod tests { assert_eq!( ( capsule_is_ready(&capsule, &refs, &index).is_err(), - control_is_ready(&controls[0], &refs, &index).is_err(), + visibility_is_ready( + controls[0].transaction(), + controls[0].visibility_delta().unwrap(), + &refs, + &index + ) + .is_err(), apply_capsule_visibility_index(&capsule, &mut index.clone()).is_err(), apply_capsule_control_visibility_index(&controls[0], &mut index.clone()) .is_err(), diff --git a/crates/crab-read/src/capsule_protocol/visibility_tests.rs b/crates/crab-read/src/capsule_protocol/visibility_tests.rs new file mode 100644 index 000000000..9b11ca598 --- /dev/null +++ b/crates/crab-read/src/capsule_protocol/visibility_tests.rs @@ -0,0 +1,106 @@ +use super::*; +use crab_metadata::capsule_protocol::{ + CapsuleRefEdit, CapsuleSection, CapsuleSectionKind, CapsuleTransaction, CapsuleVisibilityDelta, +}; +use crab_metadata::git_visibility::GitVisibilityEdit; +use object_store::memory::InMemory; + +#[tokio::test] +async fn initial_replacement_opens_without_materializing_its_oid_set() { + use crab_metadata::capsule_protocol::{RepositoryRoot, RootRecord}; + use std::sync::atomic::{AtomicU64, Ordering}; + let bytes = Arc::new(AtomicU64::new(0)); + let observed = bytes.clone(); + let store = + Store::new(Arc::new(InMemory::new())).with_read_byte_observer(Arc::new(move |size| { + observed.fetch_add(size, Ordering::Relaxed); + })); + let layout = StoreLayout::new(store, "repositories/large-root".into()); + let initial = + RootRecord::encode(RepositoryRoot::initial(&"1".repeat(64), "refs/heads/main").unwrap()) + .unwrap(); + crab_metadata::capsule_protocol::create_root(&layout, initial.clone()) + .await + .unwrap(); + let objects = (0u32..10000) + .map(|index| format!("{index:040x}")) + .collect::>(); + let tip = objects[9000].clone(); + let transaction = CapsuleTransaction::new( + initial.digest(), + vec![CapsuleRefEdit::new( + "refs/heads/main", + None, + Some(tip.clone()), + None, + )], + ) + .unwrap(); + let delta = CapsuleVisibilityDelta::new(BTreeMap::from([( + "refs/heads/main".into(), + GitVisibilityEdit::from_replacement_objects(None, tip.clone(), objects), + )])) + .unwrap(); + let capsule = Capsule::build( + &transaction, + Vec::new(), + vec![CapsuleSection::new( + CapsuleSectionKind::VisibilityDelta, + delta.encode().unwrap(), + )], + ) + .unwrap(); + let run = CapsuleRun::leaf(capsule).unwrap(); + layout + .store() + .create_strict(&layout.capsule_path(run.hash()), run.bytes().clone()) + .await + .unwrap(); + let pointer = CapsulePointer::new( + run.hash(), + run.bytes().len() as u64, + run.control_offset(), + run.control_size(), + run.footer_hash(), + run.level(), + run.transaction_ids(), + run.newest_base_root_digest(), + ) + .unwrap(); + let next = RootRecord::encode( + initial + .root() + .advance( + initial.digest(), + BTreeMap::from([("refs/heads/main".into(), tip)]), + BTreeMap::new(), + vec![pointer], + transaction.id().unwrap().as_str(), + ) + .unwrap(), + ) + .unwrap(); + layout + .store() + .put_overwrite(&layout.capsule_root_path(), next.bytes().clone()) + .await + .unwrap(); + let root = crab_metadata::capsule_protocol::load_root(&layout) + .await + .unwrap(); + bytes.store(0, Ordering::Relaxed); + let view = open_git_view_from_root( + &layout, + root, + CapsuleReadLimits { + max_capsule_bytes: 1024 * 1024, + max_frontier_bytes: 1024 * 1024, + }, + ) + .await + .unwrap(); + assert!( + bytes.load(Ordering::Relaxed) < 64 * 1024 + && view.capsule_controls[0].visibility_delta().is_err() + ); +} From 78dac584d449b93e206b54228b17e4cb034927c5 Mon Sep 17 00:00:00 2001 From: forhappy Date: Wed, 14 Oct 2026 23:28:50 -0700 Subject: [PATCH 3/6] fix(read): retain control corruption errors Preserve read-side corruption classification when shared metadata code detects conflicting captured controls. Keep transport errors as metadata sources. --- crates/crab-read/src/capsule_protocol.rs | 88 ++++++++++++++---------- 1 file changed, 51 insertions(+), 37 deletions(-) diff --git a/crates/crab-read/src/capsule_protocol.rs b/crates/crab-read/src/capsule_protocol.rs index 17941f462..1272dd352 100644 --- a/crates/crab-read/src/capsule_protocol.rs +++ b/crates/crab-read/src/capsule_protocol.rs @@ -4077,7 +4077,8 @@ pub async fn open_ref_view_from_root( snapshot: crab_metadata::capsule_protocol::RootSnapshot, ) -> Result { let (heads, active) = capture_ref_heads(router, snapshot.record().root()).await?; - let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active)?; + let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active) + .map_err(metadata_control_error)?; Ok(CapsuleRefView { root: snapshot, refs: visible.refs, @@ -4096,7 +4097,8 @@ pub async fn read_activity_from_root( snapshot: &crab_metadata::capsule_protocol::RootSnapshot, ) -> Result { let (heads, active) = capture_ref_heads(router, snapshot.record().root()).await?; - let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active)?; + let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active) + .map_err(metadata_control_error)?; let capsule_count = visible.pointers.iter().try_fold(0_u64, |total, pointer| { total .checked_add(u64::from(pointer.capsule_count())) @@ -4137,7 +4139,8 @@ pub async fn open_ref_view_from_root_for_refs( ) -> Result { let (heads, active) = capture_selected_ref_heads(router, snapshot.record().root(), ref_names).await?; - let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active)?; + let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active) + .map_err(metadata_control_error)?; Ok(CapsuleRefView { root: snapshot, refs: visible @@ -4191,7 +4194,8 @@ pub async fn open_ref_view_from_root_for_push( )); } } - let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active)?; + let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active) + .map_err(metadata_control_error)?; let push_ref_head_bases = ref_names .iter() .cloned() @@ -4251,7 +4255,9 @@ pub async fn open_view_from_root_for_repack( ) -> Result { let root = snapshot.record().root(); let (heads, active) = capture_ref_heads(router, root).await?; - let pointers = materialize_visible_ref_heads(root, &heads, &active)?.pointers; + let pointers = materialize_visible_ref_heads(root, &heads, &active) + .map_err(metadata_control_error)? + .pointers; let frontier_bytes = pointers.iter().try_fold(0_u64, |total, pointer| { total .checked_add(pointer.size()) @@ -4486,7 +4492,8 @@ async fn assemble_view( } let mut refs = snapshot.record().root().refs().clone(); let mut peeled_refs = snapshot.record().root().peeled_refs().clone(); - let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active)?; + let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active) + .map_err(metadata_control_error)?; let expected_refs = visible.refs; let expected_peeled = visible.peeled_refs; let visible_ref_transactions = visible.transactions; @@ -4699,7 +4706,8 @@ async fn assemble_layered_control_view( let footer_only = matches!(read, ControlRead::Fetch); let mut refs = snapshot.record().root().refs().clone(); let mut peeled_refs = snapshot.record().root().peeled_refs().clone(); - let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active)?; + let visible = materialize_visible_ref_heads(snapshot.record().root(), &heads, &active) + .map_err(metadata_control_error)?; let expected_refs = visible.refs; let expected_peeled = visible.peeled_refs; let visible_ref_transactions = visible.transactions; @@ -4836,7 +4844,8 @@ async fn assemble_layered_control_view( capsule_run_sources.push(source); } } - let selected = select_ref_control_frontiers(snapshot.record().root(), &ref_frontiers, &runs)?; + let selected = select_ref_control_frontiers(snapshot.record().root(), &ref_frontiers, &runs) + .map_err(metadata_control_error)?; let mut root_capsules = selected.root_controls; let ref_capsule_counts = selected.counts; if footer_only { @@ -4864,7 +4873,8 @@ async fn assemble_layered_control_view( }); } let journal_capsules = - select_ref_journal_controls(snapshot.record().root(), &selected.required, &runs)?; + select_ref_journal_controls(snapshot.record().root(), &selected.required, &runs) + .map_err(metadata_control_error)?; let transaction_predecessors = selected.predecessors; let checkpoint_visibility = if matches!(read, ControlRead::Git) { checkpoint @@ -4881,7 +4891,8 @@ async fn assemble_layered_control_view( let initial = match &checkpoint_visibility { Some(reader) => { CapturedVisibility::from_checkpoint(router.store().clone(), reader.clone(), names) - .await? + .await + .map_err(metadata_control_error)? } None => CapturedVisibility::new( router.store().clone(), @@ -4894,7 +4905,10 @@ async fn assemble_layered_control_view( VisibilityReconciliation::Complete(visibility_index_for_checkpoint(checkpoint.as_ref())?) }; for capsule in &root_capsules { - visibility.apply(capsule).await?; + visibility + .apply(capsule) + .await + .map_err(metadata_control_error)?; } let ordered = order_ref_controls( snapshot.record().root().refs(), @@ -4903,9 +4917,10 @@ async fn assemble_layered_control_view( &transaction_predecessors, journal_capsules, ) - .await?; + .await + .map_err(metadata_control_error)?; for capsule in &ordered { - apply_control_refs(capsule, &mut refs, &mut peeled_refs)?; + apply_control_refs(capsule, &mut refs, &mut peeled_refs).map_err(metadata_control_error)?; } if refs != expected_refs || peeled_refs != expected_peeled { return Err(corrupt_path( @@ -5161,20 +5176,16 @@ fn capsule_is_ready( ) -> Result { let transaction = capsule.transaction()?; let delta = capsule.visibility_delta()?; - Ok(visibility_is_ready( - &transaction, - delta.as_ref(), - refs, - visibility_index, - )?) + visibility_is_ready(&transaction, delta.as_ref(), refs, visibility_index) + .map_err(metadata_control_error) } fn apply_capsule_control_visibility_index( capsule: &CapsuleControl, index: &mut crab_metadata::git_visibility::GitVisibilityIndex, ) -> Result<()> { - apply_transaction_visibility_index(capsule.transaction(), capsule.visibility_delta()?, index)?; - Ok(()) + apply_transaction_visibility_index(capsule.transaction(), capsule.visibility_delta()?, index) + .map_err(metadata_control_error) } fn apply_capsule_visibility_index( @@ -5183,8 +5194,8 @@ fn apply_capsule_visibility_index( ) -> Result<()> { let transaction = capsule.transaction()?; let delta = capsule.visibility_delta()?; - apply_transaction_visibility_index(&transaction, delta.as_ref(), index)?; - Ok(()) + apply_transaction_visibility_index(&transaction, delta.as_ref(), index) + .map_err(metadata_control_error) } fn apply_capsule_refs( @@ -5282,28 +5293,31 @@ fn admit_frontier(pointers: &[CapsulePointer], limits: CapsuleReadLimits) -> Res Ok(()) } -async fn load_run(router: &StoreLayout, pointer: &CapsulePointer) -> Result { - match crab_metadata::capsule_protocol::load_capsule_run(router, pointer).await { - Ok(run) => Ok(run), - // Read callers distinguish malformed committed data from transport errors. - Err(crab_metadata::error::MetadataError::CorruptObject { path, reason }) => { - Err(ReadError::CorruptObject { path, reason }) +// Read consumers classify corrupt captured state separately from metadata I/O. +// Keep that boundary when shared metadata code detects the same invalid control. +fn metadata_control_error(error: crab_metadata::error::MetadataError) -> ReadError { + match error { + crab_metadata::error::MetadataError::CorruptObject { path, reason } => { + ReadError::CorruptObject { path, reason } } - Err(error) => Err(error.into()), + crab_metadata::error::MetadataError::Internal(reason) => ReadError::internal(reason), + error => error.into(), } } +async fn load_run(router: &StoreLayout, pointer: &CapsulePointer) -> Result { + crab_metadata::capsule_protocol::load_capsule_run(router, pointer) + .await + .map_err(metadata_control_error) +} + async fn load_run_control( router: &StoreLayout, pointer: &CapsulePointer, ) -> Result<(CapsuleRunControl, Vec)> { - match crab_metadata::capsule_protocol::load_capsule_run_control(router, pointer).await { - Ok(value) => Ok(value), - Err(crab_metadata::error::MetadataError::CorruptObject { path, reason }) => { - Err(ReadError::CorruptObject { path, reason }) - } - Err(error) => Err(error.into()), - } + crab_metadata::capsule_protocol::load_capsule_run_control(router, pointer) + .await + .map_err(metadata_control_error) } fn corrupt(path: &object_store::path::Path, reason: impl Into) -> ReadError { From f74f279fc92260c3dbb67563b77da996458bdf36 Mon Sep 17 00:00:00 2001 From: forhappy Date: Wed, 14 Oct 2026 23:56:21 -0700 Subject: [PATCH 4/6] test(remote): measure current catalog sessions Measure opening before cold and warm selected queries. Compare duplicate results with full inspection only after closing the session. --- crates/crab-remote/examples/qualify_remote.rs | 50 ++++++++++++++++++- 1 file changed, 49 insertions(+), 1 deletion(-) diff --git a/crates/crab-remote/examples/qualify_remote.rs b/crates/crab-remote/examples/qualify_remote.rs index 07142244b..56661eeb9 100644 --- a/crates/crab-remote/examples/qualify_remote.rs +++ b/crates/crab-remote/examples/qualify_remote.rs @@ -36,13 +36,14 @@ async fn main() -> Result<()> { [bucket, prefix, path] => (bucket, prefix, path, "all"), [bucket, prefix, path, selection] => (bucket, prefix, path, selection.as_str()), _ => return Err(io::Error::other( - "usage: qualify_remote [operation|all|build-indexes|checkpoint|catalog|ref-point|resolve-oid|blame-history|blame-history-uncached]", + "usage: qualify_remote [operation|all|build-indexes|checkpoint|catalog|catalog-current|ref-point|resolve-oid|blame-history|blame-history-uncached]", ) .into()), }; if selection != "all" && selection != "build-indexes" && selection != "catalog" + && selection != "catalog-current" && selection != "checkpoint" && selection != "graph-pages" && selection != "graph-ancestry" @@ -101,6 +102,53 @@ async fn main() -> Result<()> { "{}", json!({"kind":"result", "operation":"ref-point", "value":previous}) ); + } else if selection == "catalog-current" { + let hash = crab_xet::hash::MerkleHash::from_hex(std::str::from_utf8(path.as_bytes())?)?; + let session = counters + .measure("open", "cold", async { + Ok( + crab_metadata::file_index_lookup::FileIndexLookupSession::open_for_storage( + layout.store(), + layout.repo_prefix(), + ) + .await?, + ) + }) + .await?; + let result: Result<_> = async { + let mut previous = None; + for cache in ["cold", "warm"] { + let selected = counters + .measure("catalog-current", cache, async { + Ok(session.lookup_batch(&[hash, hash]).await?) + }) + .await?; + if previous.as_ref().is_some_and(|value| value != &selected) { + return Err(io::Error::other("cold/warm current catalog results differ").into()); + } + previous = Some(selected); + } + Ok(previous) + } + .await; + let closed = session.close().await; + let selected = result?; + closed?; + // Run the bulk maintenance oracle only after measured reads and close: + // it must not populate page caches before the cold selected query. + let full = crab_metadata::capsule_protocol::load_pointer_catalog(&layout).await?; + let expected = full + .files() + .get(&hash.hex()) + .ok_or_else(|| io::Error::other("requested catalog file is absent"))?; + let expected = crab_xet::hash::MerkleHash::from_hex(expected.shard_hash())?; + if selected.as_ref() != Some(&vec![Some(expected), Some(expected)]) { + return Err(io::Error::other("current catalog batch differs from full catalog").into()); + } + println!( + "{}", + json!({"kind":"result", "operation":"catalog-current", "files":2, "shard":expected.hex()}) + ); } else if selection == "catalog" { let view = counters .measure("open", "cold", async { From 6a8a8e75b8a0fba391fb6b2f8aaab7b9626610dc Mon Sep 17 00:00:00 2001 From: forhappy Date: Thu, 15 Oct 2026 00:52:57 -0700 Subject: [PATCH 5/6] perf(metadata): cache verified catalog bindings --- crates/crab-metadata/README.md | 6 + .../capsule_protocol/pointer/paged/reader.rs | 35 ++++- crates/crab-metadata/src/file_index_lookup.rs | 107 +++++++++++----- .../src/file_index_lookup/catalog_tests.rs | 120 ++++++++++++++++-- .../src/file_index_lookup/shared_tests.rs | 8 +- 5 files changed, 234 insertions(+), 42 deletions(-) diff --git a/crates/crab-metadata/README.md b/crates/crab-metadata/README.md index 79984ed78..395091efe 100644 --- a/crates/crab-metadata/README.md +++ b/crates/crab-metadata/README.md @@ -72,6 +72,12 @@ accepts captured lookup roots and resolves requested files with their authentica shard/xorb closure. Point and batch calls share the page cache and retain the session read guard through I/O, so close waits for active reads. Missing catalog pages remain errors rather than reopening the mutable file-index database. +A session caches at most 4,096 verified catalog bindings or authenticated misses, +using the same result-cache owner as canonical shard lookup. Cold queries still +validate the full selected dependency closure before publishing cache entries; +errors are never cached as absence. Cached inputs retain the original admission +charge. Catalog cache locks cover memory work only, so unrelated warm lookups can finish +while a cold origin read is paused. Large cold batches do not retain their results. The current-state opener also captures authenticated catalog roots and leaves catalog bodies cold. Explicitly pinned callers pass their existing capture, while current-state sessions bind once to the first root they open. Catalog and Git diff --git a/crates/crab-metadata/src/capsule_protocol/pointer/paged/reader.rs b/crates/crab-metadata/src/capsule_protocol/pointer/paged/reader.rs index 14449fba8..f9e3bf451 100644 --- a/crates/crab-metadata/src/capsule_protocol/pointer/paged/reader.rs +++ b/crates/crab-metadata/src/capsule_protocol/pointer/paged/reader.rs @@ -10,6 +10,8 @@ use std::sync::Arc; #[cfg(test)] mod tests; +const QUERY_KEY_BYTES: u64 = 96; + /// Snapshot-bound catalog queries with one cache and aggregate byte budget. /// /// The owner retains the captured objects against GC. Missing keys return an @@ -236,6 +238,17 @@ impl PointerCatalogLookup { self.dependencies(store, hashes, &[], &[]).await } + #[cfg(feature = "file-index-reader")] + pub(crate) async fn files_for_lookup( + &self, + store: &Store, + hashes: &[String], + cached_files: usize, + ) -> Result { + self.dependencies_with_admission(store, hashes, &[], &[], cached_files) + .await + } + /// Select files and explicit shard/xorb identities with their complete closure. /// /// All keys and dependency pages share one query budget. Absent requested @@ -246,11 +259,31 @@ impl PointerCatalogLookup { files: &[String], shards: &[String], xorbs: &[String], + ) -> Result { + self.dependencies_with_admission(store, files, shards, xorbs, 0) + .await + } + + async fn dependencies_with_admission( + &self, + store: &Store, + files: &[String], + shards: &[String], + xorbs: &[String], + cached_files: usize, ) -> Result { let mut work = self.work(); + // Cached bindings skip page decoding, not the aggregate input budget. + admit( + &mut work, + (cached_files as u64).saturating_mul(QUERY_KEY_BYTES), + )?; let hashes = requested(files, &mut work)?; let mut shards = requested(shards, &mut work)?; let mut xorbs = requested(xorbs, &mut work)?; + if hashes.is_empty() && shards.is_empty() && xorbs.is_empty() { + return Ok(PointerCatalog::new()); + } let root = self .checkpoint .as_ref() @@ -425,7 +458,7 @@ fn admit(work: &mut PageWork, bytes: u64) -> Result<()> { } fn requested(hashes: &[String], work: &mut PageWork) -> Result> { - admit(work, (hashes.len() as u64).saturating_mul(96))?; + admit(work, (hashes.len() as u64).saturating_mul(QUERY_KEY_BYTES))?; for value in hashes { hash(value)?; } diff --git a/crates/crab-metadata/src/file_index_lookup.rs b/crates/crab-metadata/src/file_index_lookup.rs index c8f3f4c9b..a9a433b24 100644 --- a/crates/crab-metadata/src/file_index_lookup.rs +++ b/crates/crab-metadata/src/file_index_lookup.rs @@ -22,6 +22,7 @@ use crab_xet::xorb::format::MerkleHash; const DB_LABEL: &str = "file_index_db"; const GET_BATCH_CONCURRENCY: usize = 256; const SHARD_SEARCH_CONCURRENCY: usize = 4; +const MAX_VERIFIED_CATALOG_FILES: usize = 4096; static SHARD_SCANS: LazyLock> = LazyLock::new(|| Arc::new(tokio::sync::Semaphore::new(SHARD_SEARCH_CONCURRENCY))); @@ -145,7 +146,7 @@ impl CommittedShardAnchor { } #[derive(Debug, Default)] -struct ManifestFallbackCache { +struct FileLookupCache { found: HashMap, missing: HashSet, shard_visits: usize, @@ -230,7 +231,7 @@ pub struct FileIndexLookupSession { storage: crab_storage::Store, router: crab_storage::StoreLayout, parsers: tokio_util::task::TaskTracker, - manifest_fallback: tokio::sync::Mutex, + resolved_files: tokio::sync::Mutex, limits: FileIndexLookupLimits, } @@ -250,7 +251,7 @@ impl FileIndexLookupSession { storage: router.store().clone(), router, parsers: tokio_util::task::TaskTracker::new(), - manifest_fallback: tokio::sync::Mutex::new(ManifestFallbackCache::default()), + resolved_files: tokio::sync::Mutex::new(FileLookupCache::default()), limits: FileIndexLookupLimits::CURRENT_STATE, }) } @@ -313,7 +314,7 @@ impl FileIndexLookupSession { storage: router.store().clone(), router: router.clone(), parsers: tokio_util::task::TaskTracker::new(), - manifest_fallback: tokio::sync::Mutex::new(ManifestFallbackCache::default()), + resolved_files: tokio::sync::Mutex::new(FileLookupCache::default()), limits, }) } @@ -373,7 +374,7 @@ impl FileIndexLookupSession { storage: router.store().clone(), router: router.clone(), parsers: tokio_util::task::TaskTracker::new(), - manifest_fallback: tokio::sync::Mutex::new(ManifestFallbackCache::default()), + resolved_files: tokio::sync::Mutex::new(FileLookupCache::default()), limits, }) } @@ -398,7 +399,7 @@ impl FileIndexLookupSession { storage, router, parsers: tokio_util::task::TaskTracker::new(), - manifest_fallback: tokio::sync::Mutex::new(ManifestFallbackCache::default()), + resolved_files: tokio::sync::Mutex::new(FileLookupCache::default()), limits: FileIndexLookupLimits::CURRENT_STATE, }); } @@ -422,7 +423,7 @@ impl FileIndexLookupSession { storage, router, parsers: tokio_util::task::TaskTracker::new(), - manifest_fallback: tokio::sync::Mutex::new(ManifestFallbackCache::default()), + resolved_files: tokio::sync::Mutex::new(FileLookupCache::default()), limits: FileIndexLookupLimits::CURRENT_STATE, }; if use_acceleration { @@ -489,28 +490,7 @@ impl FileIndexLookupSession { return Ok(Vec::new()); } if let Some(lookup) = &self.catalog { - let hashes = file_hashes.iter().map(MerkleHash::hex).collect::>(); - let selected = lookup.files(&self.storage, &hashes).await?; - return hashes - .iter() - .map(|hash| { - selected - .files() - .get(hash) - .map(|entry| { - MerkleHash::from_hex(entry.shard_hash()).map_err(|source| { - MetadataError::CorruptObject { - path: "capsule-protocol pointer catalog".to_owned(), - reason: format!( - "invalid shard hash {}: {source}", - entry.shard_hash() - ), - } - }) - }) - .transpose() - }) - .collect(); + return self.catalog_lookup_batch(lookup, file_hashes).await; } let records = self.lookup_committed_records_batch(file_hashes).await?; let mut out = records @@ -533,6 +513,71 @@ impl FileIndexLookupSession { Ok(out) } + async fn catalog_lookup_batch( + &self, + lookup: &crate::capsule_protocol::PointerCatalogLookup, + hashes: &[MerkleHash], + ) -> Result>> { + let mut output = vec![None; hashes.len()]; + let missing = { + let cache = self.resolved_files.lock().await; + hashes + .iter() + .enumerate() + .filter_map(|(index, hash)| { + if let Some(shard) = cache.found.get(hash) { + output[index] = Some(*shard); + return None; + } + (!cache.missing.contains(hash)).then_some((index, *hash)) + }) + .collect::>() + }; + let keys = missing + .iter() + .map(|(_, hash)| hash.hex()) + .collect::>(); + let cached_files = hashes.len() - missing.len(); + // Verified bindings belong to this immutable capture. Keep only the + // small result, but admit every input and validate cold dependency closure. + let selected = lookup + .files_for_lookup(&self.storage, &keys, cached_files) + .await?; + for (index, hash) in &missing { + output[*index] = selected + .files() + .get(&hash.hex()) + .map(|entry| { + MerkleHash::from_hex(entry.shard_hash()).map_err(|source| { + MetadataError::CorruptObject { + path: "capsule-protocol pointer catalog".to_owned(), + reason: format!("invalid shard hash {}: {source}", entry.shard_hash()), + } + }) + }) + .transpose()?; + } + if !missing.is_empty() && missing.len() <= MAX_VERIFIED_CATALOG_FILES { + let mut cache = self.resolved_files.lock().await; + let cached = cache.found.len().saturating_add(cache.missing.len()); + if cached.saturating_add(missing.len()) > MAX_VERIFIED_CATALOG_FILES { + cache.found.clear(); + cache.missing.clear(); + } + for (index, hash) in missing { + match output[index] { + Some(shard) => { + cache.found.insert(hash, shard); + } + None => { + cache.missing.insert(hash); + } + } + } + } + Ok(output) + } + /// Return only generation-pinned acceleration hits, without invoking the /// manifest-scoped correctness fallback. pub async fn lookup_committed_records_batch( @@ -581,7 +626,7 @@ impl FileIndexLookupSession { let Some(anchor) = self.anchor.as_ref() else { return Ok(vec![None; file_hashes.len()]); }; - let mut cache = self.manifest_fallback.lock().await; + let mut cache = self.resolved_files.lock().await; let unresolved = file_hashes .iter() .filter(|file_hash| { @@ -766,7 +811,7 @@ impl LookupSource { storage: router.store().clone(), router: router.clone(), parsers: tokio_util::task::TaskTracker::new(), - manifest_fallback: tokio::sync::Mutex::new(ManifestFallbackCache::default()), + resolved_files: tokio::sync::Mutex::new(FileLookupCache::default()), limits: FileIndexLookupLimits::CURRENT_STATE, }), Self::Current { diff --git a/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs b/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs index f656df3f9..95b847f4f 100644 --- a/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs +++ b/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs @@ -12,6 +12,7 @@ use std::sync::atomic::AtomicUsize; pub(super) async fn capture( layout: &crab_storage::StoreLayout, catalog: PointerCatalog, + maximum: u64, ) -> (PointerCatalogLookup, object_store::path::Path, RootRecord) { let pack = CapsuleGitPack::new( Bytes::from_static(b"pack"), @@ -140,7 +141,7 @@ pub(super) async fn capture( ) .unwrap(); ( - control.pointer_catalog_lookup(layout, 512 * 1024).unwrap(), + control.pointer_catalog_lookup(layout, maximum).unwrap(), path, root, ) @@ -185,7 +186,7 @@ async fn pinned_batches_preserve_order_misses_and_page_bounds_as_history_grows() observed.fetch_add(1, Ordering::SeqCst); })); let layout = crab_storage::StoreLayout::new(storage, "pinned-lookup".to_owned()); - let (captured, _, _) = capture(&layout, catalog(count)).await; + let (captured, _, _) = capture(&layout, catalog(count), 512 * 1024).await; let lookup = SharedFileIndexLookup::for_pointer_catalog(layout.clone(), captured); assert_eq!( reads.load(Ordering::SeqCst), @@ -193,7 +194,7 @@ async fn pinned_batches_preserve_order_misses_and_page_bounds_as_history_grows() "construction must leave pages cold" ); // A later authenticated catalog must not change an existing capture. - capture(&layout, catalog(count + 1)).await; + capture(&layout, catalog(count + 1), 512 * 1024).await; let file = super::tests::hash_from_seed(42); let later = super::tests::hash_from_seed(count as u64); let shard = super::tests::hash_from_seed(1_000_042); @@ -224,16 +225,30 @@ async fn missing_captured_pages_are_errors_even_when_a_new_catalog_exists() { crab_storage::Store::new(Arc::new(InMemory::new())), "missing-page".to_owned(), ); - let (captured, path, _) = capture(&layout, catalog(100)).await; - capture(&layout, catalog(101)).await; + let (captured, path, _) = capture(&layout, catalog(100), 512 * 1024).await; + let (original, _) = layout + .store() + .get_with_etag_bounded(&path, 512 * 1024) + .await + .unwrap(); + capture(&layout, catalog(101), 512 * 1024).await; layout.store().delete(&path).await.unwrap(); - let lookup = SharedFileIndexLookup::for_pointer_catalog(layout, captured); + let lookup = SharedFileIndexLookup::for_pointer_catalog(layout.clone(), captured); assert!( lookup .lookup(&super::tests::hash_from_seed(42)) .await .is_err() ); + layout.store().put_overwrite(&path, original).await.unwrap(); + assert_eq!( + lookup + .lookup(&super::tests::hash_from_seed(42)) + .await + .unwrap(), + Some(super::tests::hash_from_seed(1_000_042)), + "failed closure reads must not cache absence" + ); lookup.close().await.unwrap(); } @@ -253,7 +268,7 @@ async fn current_state_opener_keeps_catalog_pages_selected_and_pins_its_first_ro observed_bytes.fetch_add(size, Ordering::SeqCst); })); let layout = crab_storage::StoreLayout::new(storage.clone(), "current-catalog".to_owned()); - let (_, _, root) = capture(&layout, catalog(count)).await; + let (_, _, root) = capture(&layout, catalog(count), 512 * 1024).await; assert_eq!(root.root().generation(), 1); let lookup = SharedFileIndexLookup::new_for_storage(&storage, layout.repo_prefix()); assert_eq!(reads.load(Ordering::SeqCst), 0); @@ -269,7 +284,7 @@ async fn current_state_opener_keeps_catalog_pages_selected_and_pins_its_first_ro reads.load(Ordering::SeqCst), bytes.load(Ordering::SeqCst) ); - let (_, _, later_root) = capture(&layout, catalog(count + 1)).await; + let (_, _, later_root) = capture(&layout, catalog(count + 1), 512 * 1024).await; layout .store() .put_overwrite(&layout.capsule_root_path(), later_root.bytes().clone()) @@ -286,3 +301,92 @@ async fn current_state_opener_keeps_catalog_pages_selected_and_pins_its_first_ro latest.close().await.unwrap(); } } + +#[tokio::test] +async fn repeated_large_recipe_binding_queries_reuse_verified_results() { + let reads = Arc::new(AtomicUsize::new(0)); + let observed = reads.clone(); + let storage = crab_storage::Store::new(Arc::new(InMemory::new())).with_read_request_observer( + Arc::new(move |_| { + observed.fetch_add(1, Ordering::SeqCst); + }), + ); + let layout = crab_storage::StoreLayout::new(storage, "large-recipe-binding".to_owned()); + let file = super::tests::hash_from_seed(42); + let shard = super::tests::hash_from_seed(1_000_042); + let mut catalog = PointerCatalog::new(); + let mut xorbs = Vec::new(); + for index in 0..160_u64 { + let xorb = super::tests::hash_from_seed(2_000_000 + index).hex(); + let chunks = (0..1024) + .map(|offset| { + XorbChunkEntry::new( + super::tests::hash_from_seed(3_000_000 + index * 1024 + offset).hex(), + 64 * 1024, + ) + }) + .collect(); + catalog + .insert_xorb( + xorb.clone(), + XorbCatalogEntry::new(64 * 1024 * 1024, "e".repeat(64), chunks), + ) + .unwrap(); + xorbs.push(xorb); + } + catalog + .insert_shard(shard.hex(), ShardCatalogEntry::new(80, xorbs)) + .unwrap(); + // This is metadata for a 10 GiB recipe, not a payload reconstruction fixture. + // Its verified closure outgrows the ordinary decoded-page cache. + catalog + .insert_file( + file.hex(), + FileCatalogEntry::new(10 * 1024 * 1024 * 1024, shard.hex()), + ) + .unwrap(); + let (captured, _, _) = capture(&layout, catalog, 16 * 1024 * 1024).await; + let lookup = SharedFileIndexLookup::for_pointer_catalog(layout, captured); + let cold = lookup.lookup(&file).await; + let cold_reads = reads.load(Ordering::SeqCst); + let warm = lookup.lookup_batch(&[file, file]).await; + let warm_reads = reads.load(Ordering::SeqCst); + lookup.close().await.unwrap(); + assert_eq!( + (cold.unwrap(), warm.unwrap()), + (Some(shard), vec![Some(shard), Some(shard)]) + ); + assert!(cold_reads > 0); + assert_eq!( + warm_reads, cold_reads, + "verified immutable bindings must not re-read the large recipe" + ); +} + +#[tokio::test] +async fn cached_binding_batches_keep_the_original_input_admission() { + let reads = Arc::new(AtomicUsize::new(0)); + let observed = reads.clone(); + let storage = crab_storage::Store::new(Arc::new(InMemory::new())).with_read_request_observer( + Arc::new(move |_| { + observed.fetch_add(1, Ordering::SeqCst); + }), + ); + let layout = crab_storage::StoreLayout::new(storage, "cached-admission".to_owned()); + let file = super::tests::hash_from_seed(42); + let (captured, _, _) = capture(&layout, catalog(100), 512 * 1024).await; + let queries = vec![file; 10_000]; + let keys = queries.iter().map(MerkleHash::hex).collect::>(); + assert!(captured.files(layout.store(), &keys).await.is_err()); + let lookup = SharedFileIndexLookup::for_pointer_catalog(layout, captured); + lookup.lookup(&file).await.unwrap().unwrap(); + let cold_reads = reads.load(Ordering::SeqCst); + let result = lookup.lookup_batch(&queries).await; + lookup.close().await.unwrap(); + assert!(matches!(result, Err(MetadataError::CorruptObject { .. }))); + assert_eq!( + reads.load(Ordering::SeqCst), + cold_reads, + "cached input rejection must precede I/O" + ); +} diff --git a/crates/crab-metadata/src/file_index_lookup/shared_tests.rs b/crates/crab-metadata/src/file_index_lookup/shared_tests.rs index 2a52edc02..2f7c5a1b2 100644 --- a/crates/crab-metadata/src/file_index_lookup/shared_tests.rs +++ b/crates/crab-metadata/src/file_index_lookup/shared_tests.rs @@ -352,8 +352,12 @@ async fn captured_catalog_close_waits_for_active_reads_and_abandonment_releases_ let inner: Arc = Arc::new(InMemory::new()); let storage = crab_storage::Store::new(inner.clone()); let router = crab_storage::StoreLayout::new(storage, "pinned/close".to_owned()); - let (catalog, path) = - super::catalog_tests::capture(&router, super::catalog_tests::catalog(10_000)).await; + let (catalog, path, _) = super::catalog_tests::capture( + &router, + super::catalog_tests::catalog(10_000), + 512 * 1024, + ) + .await; let store = Arc::new(PausedLookupStore { inner, shard: path, From 6fab757a39751a0b0f52b334f08eca949207a373 Mon Sep 17 00:00:00 2001 From: forhappy Date: Thu, 15 Oct 2026 01:20:28 -0700 Subject: [PATCH 6/6] fix(metadata): retain duplicate batch bindings --- crates/crab-metadata/README.md | 3 +- crates/crab-metadata/src/file_index_lookup.rs | 15 ++++++---- .../src/file_index_lookup/catalog_tests.rs | 29 +++++++++++++++++-- 3 files changed, 38 insertions(+), 9 deletions(-) diff --git a/crates/crab-metadata/README.md b/crates/crab-metadata/README.md index 395091efe..676c023e5 100644 --- a/crates/crab-metadata/README.md +++ b/crates/crab-metadata/README.md @@ -77,7 +77,8 @@ using the same result-cache owner as canonical shard lookup. Cold queries still validate the full selected dependency closure before publishing cache entries; errors are never cached as absence. Cached inputs retain the original admission charge. Catalog cache locks cover memory work only, so unrelated warm lookups can finish -while a cold origin read is paused. Large cold batches do not retain their results. +while a cold origin read is paused. Retention counts distinct file identities; +duplicate occurrences do not consume additional entries or prevent caching. The current-state opener also captures authenticated catalog roots and leaves catalog bodies cold. Explicitly pinned callers pass their existing capture, while current-state sessions bind once to the first root they open. Catalog and Git diff --git a/crates/crab-metadata/src/file_index_lookup.rs b/crates/crab-metadata/src/file_index_lookup.rs index a9a433b24..dbf710cc9 100644 --- a/crates/crab-metadata/src/file_index_lookup.rs +++ b/crates/crab-metadata/src/file_index_lookup.rs @@ -557,14 +557,17 @@ impl FileIndexLookupSession { }) .transpose()?; } - if !missing.is_empty() && missing.len() <= MAX_VERIFIED_CATALOG_FILES { + if !missing.is_empty() { let mut cache = self.resolved_files.lock().await; - let cached = cache.found.len().saturating_add(cache.missing.len()); - if cached.saturating_add(missing.len()) > MAX_VERIFIED_CATALOG_FILES { - cache.found.clear(); - cache.missing.clear(); - } for (index, hash) in missing { + if cache.found.contains_key(&hash) || cache.missing.contains(&hash) { + continue; + } + // Duplicate occurrences consume input budget, but retain one fact. + if cache.found.len() + cache.missing.len() == MAX_VERIFIED_CATALOG_FILES { + cache.found.clear(); + cache.missing.clear(); + } match output[index] { Some(shard) => { cache.found.insert(hash, shard); diff --git a/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs b/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs index 95b847f4f..3c4efd60e 100644 --- a/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs +++ b/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs @@ -347,14 +347,18 @@ async fn repeated_large_recipe_binding_queries_reuse_verified_results() { .unwrap(); let (captured, _, _) = capture(&layout, catalog, 16 * 1024 * 1024).await; let lookup = SharedFileIndexLookup::for_pointer_catalog(layout, captured); - let cold = lookup.lookup(&file).await; + let duplicates = vec![file; super::MAX_VERIFIED_CATALOG_FILES + 1]; + let cold = lookup.lookup_batch(&duplicates).await; let cold_reads = reads.load(Ordering::SeqCst); let warm = lookup.lookup_batch(&[file, file]).await; let warm_reads = reads.load(Ordering::SeqCst); lookup.close().await.unwrap(); assert_eq!( (cold.unwrap(), warm.unwrap()), - (Some(shard), vec![Some(shard), Some(shard)]) + ( + vec![Some(shard); duplicates.len()], + vec![Some(shard), Some(shard)] + ) ); assert!(cold_reads > 0); assert_eq!( @@ -363,6 +367,27 @@ async fn repeated_large_recipe_binding_queries_reuse_verified_results() { ); } +#[tokio::test] +async fn binding_cache_stays_bounded_for_distinct_misses() { + let layout = crab_storage::StoreLayout::new( + crab_storage::Store::new(Arc::new(InMemory::new())), + "bounded-binding-cache".to_owned(), + ); + let (captured, _, _) = capture(&layout, PointerCatalog::new(), 16 * 1024 * 1024).await; + let lookup = SharedFileIndexLookup::for_pointer_catalog(layout, captured); + let files = (0..super::MAX_VERIFIED_CATALOG_FILES * 2) + .map(|index| super::tests::hash_from_seed(index as u64)) + .collect::>(); + lookup.lookup_batch(&files).await.unwrap(); + let retained = { + let session = lookup.session().await.unwrap(); + let cache = session.resolved_files.lock().await; + cache.found.len() + cache.missing.len() + }; + lookup.close().await.unwrap(); + assert!(retained <= super::MAX_VERIFIED_CATALOG_FILES); +} + #[tokio::test] async fn cached_binding_batches_keep_the_original_input_admission() { let reads = Arc::new(AtomicUsize::new(0));