From 7944e379726662feb999f6c1121ce8598baaf3a6 Mon Sep 17 00:00:00 2001 From: forhappy Date: Wed, 14 Oct 2026 22:09:29 -0700 Subject: [PATCH 1/2] perf(read): page pinned hydration lookups Keep captured catalog roots in the shared file lookup instead of copying all file bindings. Mount capture uses the paged Git view so checkpoint bodies stay cold. Preserve close ownership and authenticate the selected dependency closure. --- crab/src/cmd/mount.rs | 11 +- crab/src/read/mod.rs | 6 +- crates/crab-metadata/README.md | 8 +- crates/crab-metadata/src/file_index_lookup.rs | 114 +++++++------- .../src/file_index_lookup/catalog_tests.rs | 140 ++++++++++++++++++ crates/crab-read/README.md | 5 + crates/crab-read/src/hydrator.rs | 88 ++++++++++- 7 files changed, 302 insertions(+), 70 deletions(-) create mode 100644 crates/crab-metadata/src/file_index_lookup/catalog_tests.rs diff --git a/crab/src/cmd/mount.rs b/crab/src/cmd/mount.rs index 0ed8c170a..6f414255d 100644 --- a/crab/src/cmd/mount.rs +++ b/crab/src/cmd/mount.rs @@ -2077,7 +2077,9 @@ async fn resolve_mount_read_context_from_remote_url( } else { config.uploadpack_max_egress_bytes }; - let view = crab_read::capsule_protocol::open_view_from_root_with_control( + // The mount needs captured catalog roots, not complete checkpoint bodies. + // Keep pages cold until hydration requests their dependency closure. + let view = crab_read::capsule_protocol::open_git_view_from_root( &read_layout, root, crab_read::capsule_protocol::CapsuleReadLimits { @@ -2087,13 +2089,12 @@ async fn resolve_mount_read_context_from_remote_url( ) .await .ok()?; - let catalog = view.pointer_catalog().await.ok()?; + let catalog = view.pointer_catalog_lookup(&read_layout, maximum).ok()?; Some( crab_metadata::file_index_lookup::SharedFileIndexLookup::for_pointer_catalog( read_layout.clone(), - &catalog, - ) - .ok()?, + catalog, + ), ) } Err(crab_metadata::error::MetadataError::Storage { diff --git a/crab/src/read/mod.rs b/crab/src/read/mod.rs index ee5a83395..0a5c729d9 100644 --- a/crab/src/read/mod.rs +++ b/crab/src/read/mod.rs @@ -531,12 +531,12 @@ impl Inner { &self.cancel, ) .await?; - let catalog = view.pointer_catalog().await?; + let catalog = view.pointer_catalog_lookup(&read_layout, maximum)?; let file_index_lookup = crab_metadata::file_index_lookup::SharedFileIndexLookup::for_pointer_catalog( read_layout, - &catalog, - )?; + catalog, + ); Ok((resolved, git_dir, Some(file_index_lookup))) } diff --git a/crates/crab-metadata/README.md b/crates/crab-metadata/README.md index c7752c5a2..84a0a6de5 100644 --- a/crates/crab-metadata/README.md +++ b/crates/crab-metadata/README.md @@ -59,7 +59,13 @@ its footer. `LayeredCheckpoint::pointer_catalog_lookup` captures these roots; `PointerCatalogLookup` batches file/chunk/xorb keys through one bounded page cache and byte budget, validates selected dependency closure and keeps immutable descriptor conflicts distinct from absent keys. Full inspection remains an -explicit maintenance operation. All 19 audited local release tags omit these +explicit maintenance operation. `SharedFileIndexLookup::for_pointer_catalog` +accepts captured lookup roots and resolves requested files with their authenticated +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 capsule checkpoint formats; unshipped format 6 is rejected without a fallback. Publication deltas are still materialized by the caller's captured view; their run-level aggregation and lazy capture remain required for the complete gate. diff --git a/crates/crab-metadata/src/file_index_lookup.rs b/crates/crab-metadata/src/file_index_lookup.rs index 86fef2ef5..872418e4b 100644 --- a/crates/crab-metadata/src/file_index_lookup.rs +++ b/crates/crab-metadata/src/file_index_lookup.rs @@ -224,6 +224,7 @@ async fn lookup_committed_record( /// Opens the repo's `file_index_db` once, serves one or many point lookups, and /// closes the underlying SlateDB reader when consumed. pub struct FileIndexLookupSession { + catalog: Option>, static_entries: Option>>, reader: Option>, anchor: Option, @@ -244,6 +245,7 @@ impl FileIndexLookupSession { snapshot: &RepositorySnapshot, ) -> Result { Ok(Self { + catalog: None, static_entries: None, reader: None, anchor: CommittedShardAnchor::from_snapshot(snapshot)?, @@ -307,6 +309,7 @@ impl FileIndexLookupSession { )?; let anchor = CommittedShardAnchor::from_snapshot(snapshot)?; Ok(Self { + catalog: None, static_entries: None, reader: None, anchor, @@ -367,6 +370,7 @@ impl FileIndexLookupSession { }) }; Ok(Self { + catalog: None, static_entries: None, reader: None, anchor, @@ -414,6 +418,7 @@ impl FileIndexLookupSession { }) .collect::>>()?; return Ok(Self { + catalog: None, static_entries: Some(Arc::new(entries)), reader: None, anchor: None, @@ -438,6 +443,7 @@ impl FileIndexLookupSession { Err(error) => return Err(error), }; let mut session = Self { + catalog: None, static_entries: None, reader: None, anchor, @@ -477,6 +483,14 @@ impl FileIndexLookupSession { /// Look up one file hash in the open session. pub async fn lookup(&self, file_hash: &MerkleHash) -> Result> { + if self.catalog.is_some() { + return Ok(self + .lookup_batch(&[*file_hash]) + .await? + .into_iter() + .next() + .flatten()); + } if let Some(entries) = &self.static_entries { return Ok(entries.get(file_hash).copied()); } @@ -505,6 +519,30 @@ impl FileIndexLookupSession { if file_hashes.is_empty() { 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(); + } if let Some(entries) = &self.static_entries { return Ok(file_hashes .iter() @@ -739,9 +777,9 @@ fn spawn_shard_parse( } enum LookupSource { - Static { + Catalog { router: crab_storage::StoreLayout, - entries: Arc>, + lookup: Arc, }, Current { store: crab_storage::Store, @@ -759,8 +797,9 @@ enum LookupSource { impl LookupSource { async fn open(&self) -> Result { match self { - Self::Static { router, entries } => Ok(FileIndexLookupSession { - static_entries: Some(Arc::clone(entries)), + Self::Catalog { router, lookup } => Ok(FileIndexLookupSession { + catalog: Some(Arc::clone(lookup)), + static_entries: None, reader: None, anchor: None, storage: router.store().clone(), @@ -807,45 +846,25 @@ pub struct SharedFileIndexLookup { } impl SharedFileIndexLookup { - /// Lazily resolve files from an already authenticated pointer catalog. + /// Resolve files lazily from the caller's captured authenticated catalog roots. /// - /// This binds the lookup to the catalog captured by a caller's immutable - /// v2 view. Lookups never open the mutable file-index database or widen to - /// a later repository state. + /// The caller retains these roots against GC. Lookups never open mutable + /// metadata or widen to a later repository state; dependency errors stay errors. + #[must_use] pub fn for_pointer_catalog( router: crab_storage::StoreLayout, - catalog: &crate::capsule_protocol::PointerCatalog, - ) -> Result { - let entries = catalog - .files() - .iter() - .map(|(file_hash, entry)| { - Ok(( - MerkleHash::from_hex(file_hash).map_err(|source| { - MetadataError::CorruptObject { - path: "capsule-protocol pointer catalog".to_owned(), - reason: format!("invalid file hash {file_hash}: {source}"), - } - })?, - 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()), - } - })?, - )) - }) - .collect::>>()?; - Ok(Self { + lookup: crate::capsule_protocol::PointerCatalogLookup, + ) -> Self { + Self { inner: Arc::new(SharedFileIndexLookupInner { - source: LookupSource::Static { + source: LookupSource::Catalog { router, - entries: Arc::new(entries), + lookup: Arc::new(lookup), }, session: tokio::sync::RwLock::new(tokio::sync::OnceCell::new()), closed: AtomicBool::new(false), }), - }) + } } /// Lazily resolve files from one captured immutable shard-index root without writes. @@ -976,6 +995,8 @@ pub async fn resolve_file_hash_to_shard( result } +#[cfg(test)] +mod catalog_tests; #[cfg(test)] mod shared_tests; #[cfg(test)] @@ -1832,29 +1853,4 @@ mod tests { .expect_err("closed clone must reject future lookups"); assert!(matches!(err, MetadataError::Internal(_))); } - - #[tokio::test] - async fn shared_lookup_from_pointer_catalog_never_reads_latest_index() { - let store: Arc = Arc::new(InMemory::new()); - let storage = crab_storage::Store::new(Arc::clone(&store)); - let router = crab_storage::StoreLayout::new(storage, "org/pinned".to_owned()); - let file_hash = hash_from_seed(91); - let shard_hash = hash_from_seed(92); - let mut catalog = crate::capsule_protocol::PointerCatalog::new(); - catalog - .insert_file( - file_hash.hex(), - crate::capsule_protocol::FileCatalogEntry::new(16, shard_hash.hex()), - ) - .unwrap(); - - let lookup = SharedFileIndexLookup::for_pointer_catalog(router, &catalog).unwrap(); - assert_eq!(lookup.lookup(&file_hash).await.unwrap(), Some(shard_hash)); - assert_eq!( - lookup.lookup(&hash_from_seed(93)).await.unwrap(), - None, - "the captured catalog is authoritative for this mount" - ); - lookup.close().await.unwrap(); - } } diff --git a/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs b/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs new file mode 100644 index 000000000..a035cf9cb --- /dev/null +++ b/crates/crab-metadata/src/file_index_lookup/catalog_tests.rs @@ -0,0 +1,140 @@ +use super::*; +use crate::capsule_protocol::{ + CapsuleGitPack, FileCatalogEntry, LayeredCheckpoint, PackLayer, PointerCatalog, + PointerCatalogLookup, ShardCatalogEntry, XorbCatalogEntry, XorbChunkEntry, +}; +use bytes::Bytes; +use object_store::memory::InMemory; +use std::sync::atomic::AtomicUsize; + +async fn capture( + layout: &crab_storage::StoreLayout, + catalog: PointerCatalog, +) -> (PointerCatalogLookup, object_store::path::Path) { + let pack = CapsuleGitPack::new( + Bytes::from_static(b"pack"), + Bytes::from_static(b"index"), + Bytes::from_static(b"reverse"), + Bytes::from_static(b"locator"), + "1".repeat(40), + 1, + ) + .unwrap(); + let source = PackLayer::build(&pack) + .unwrap() + .source_descriptor() + .unwrap(); + let checkpoint = + LayeredCheckpoint::build(1, &"3".repeat(64), vec![source], catalog, None).unwrap(); + let path = layout.capsule_checkpoint_path(checkpoint.hash()); + layout + .store() + .put_overwrite(&path, checkpoint.bytes().clone()) + .await + .unwrap(); + let control = LayeredCheckpoint::decode_control( + checkpoint + .bytes() + .slice(checkpoint.control_offset() as usize..), + checkpoint.bytes().len() as u64, + checkpoint.control_offset(), + checkpoint.hash(), + checkpoint.footer_hash(), + ) + .unwrap(); + ( + control.pointer_catalog_lookup(layout, 512 * 1024).unwrap(), + path, + ) +} + +fn catalog(count: usize) -> PointerCatalog { + let mut result = PointerCatalog::new(); + for index in 0..count { + let file = super::tests::hash_from_seed(index as u64); + let shard = super::tests::hash_from_seed(1_000_000 + index as u64); + let xorb = super::tests::hash_from_seed(2_000_000 + index as u64); + result + .insert_file(file.hex(), FileCatalogEntry::new(16, shard.hex())) + .unwrap(); + result + .insert_shard(shard.hex(), ShardCatalogEntry::new(80, vec![xorb.hex()])) + .unwrap(); + result + .insert_xorb( + xorb.hex(), + XorbCatalogEntry::new( + 99, + "e".repeat(64), + vec![XorbChunkEntry::new( + super::tests::hash_from_seed(3_000_000 + index as u64).hex(), + 16, + )], + ), + ) + .unwrap(); + } + result +} + +#[tokio::test] +async fn pinned_batches_preserve_order_misses_and_page_bounds_as_history_grows() { + for count in [100, 20_000] { + 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, "pinned-lookup".to_owned()); + let (captured, _) = capture(&layout, catalog(count)).await; + let lookup = SharedFileIndexLookup::for_pointer_catalog(layout.clone(), captured); + assert_eq!( + reads.load(Ordering::SeqCst), + 0, + "construction must leave pages cold" + ); + // A later authenticated catalog must not change an existing capture. + capture(&layout, catalog(count + 1)).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); + assert_eq!( + lookup.lookup_batch(&[file, later, file]).await.unwrap(), + vec![Some(shard), None, Some(shard)] + ); + let cold = reads.load(Ordering::SeqCst); + assert!( + cold <= 24, + "fixed file lookup exceeded its page bound: {cold}" + ); + assert_eq!(lookup.lookup(&file).await.unwrap(), Some(shard)); + assert_eq!( + reads.load(Ordering::SeqCst), + cold, + "warm lookup reread pages" + ); + let clone = lookup.clone(); + lookup.close().await.unwrap(); + assert!(clone.lookup(&file).await.is_err()); + } +} + +#[tokio::test] +async fn missing_captured_pages_are_errors_even_when_a_new_catalog_exists() { + let layout = crab_storage::StoreLayout::new( + 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; + layout.store().delete(&path).await.unwrap(); + let lookup = SharedFileIndexLookup::for_pointer_catalog(layout, captured); + assert!( + lookup + .lookup(&super::tests::hash_from_seed(42)) + .await + .is_err() + ); + lookup.close().await.unwrap(); +} diff --git a/crates/crab-read/README.md b/crates/crab-read/README.md index 6fe09c3ad..22f08e250 100644 --- a/crates/crab-read/README.md +++ b/crates/crab-read/README.md @@ -61,6 +61,11 @@ 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. +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 +checks. It never substitutes a later catalog or mutable file-index session. + `ShardHydrator` provides memory, file, and half-open byte-range reconstruction. Before full-file reconstruction it checks that the pointer's shard terms are covered; after reconstruction it verifies the requested file hash. A shared diff --git a/crates/crab-read/src/hydrator.rs b/crates/crab-read/src/hydrator.rs index 8c1adcd21..c2a1cf139 100644 --- a/crates/crab-read/src/hydrator.rs +++ b/crates/crab-read/src/hydrator.rs @@ -1209,10 +1209,94 @@ mod tests { hydrator.router.repo_prefix().to_owned(), hydrator.router.global_prefix().to_owned(), ); - let lookup = crab_metadata::file_index_lookup::SharedFileIndexLookup::for_pointer_catalog( - router, &catalog, + let (body, _) = router + .store() + .get_with_etag(&router.shard_path(&shard_hash.into())) + .await + .unwrap(); + let mut chunks = std::collections::BTreeMap::<_, Vec<_>>::new(); + crab_xet::shard_parse::visit_xorb_chunks_from_reader( + &mut std::io::Cursor::new(&body), + |xorb, _, chunk, size| { + chunks.entry(xorb).or_default().push( + crab_metadata::capsule_protocol::XorbChunkEntry::new(chunk.hex(), size), + ); + Ok(()) + }, ) .unwrap(); + let xorb_hashes = chunks.keys().map(|hash| hash.hex()).collect(); + catalog + .insert_shard( + crab_xet::hash::MerkleHash::from(shard_hash).hex(), + crab_metadata::capsule_protocol::ShardCatalogEntry::new( + body.len() as u64, + xorb_hashes, + ), + ) + .unwrap(); + for (hash, entries) in chunks { + let (body, _) = router + .store() + .get_with_etag(&router.xorb_path(&hash)) + .await + .unwrap(); + catalog + .insert_xorb( + hash.hex(), + crab_metadata::capsule_protocol::XorbCatalogEntry::new( + body.len() as u64, + blake3::hash(&body).to_hex().to_string(), + entries, + ), + ) + .unwrap(); + } + let pack = crab_metadata::capsule_protocol::CapsuleGitPack::new( + bytes::Bytes::from_static(b"pack"), + bytes::Bytes::from_static(b"index"), + bytes::Bytes::from_static(b"reverse"), + bytes::Bytes::from_static(b"locator"), + "1".repeat(40), + 1, + ) + .unwrap(); + let source = crab_metadata::capsule_protocol::PackLayer::build(&pack) + .unwrap() + .source_descriptor() + .unwrap(); + let checkpoint = crab_metadata::capsule_protocol::LayeredCheckpoint::build( + 1, + &"3".repeat(64), + vec![source], + catalog, + None, + ) + .unwrap(); + router + .store() + .put_overwrite( + &router.capsule_checkpoint_path(checkpoint.hash()), + checkpoint.bytes().clone(), + ) + .await + .unwrap(); + let control = crab_metadata::capsule_protocol::LayeredCheckpoint::decode_control( + checkpoint + .bytes() + .slice(checkpoint.control_offset() as usize..), + checkpoint.bytes().len() as u64, + checkpoint.control_offset(), + checkpoint.hash(), + checkpoint.footer_hash(), + ) + .unwrap(); + let captured = control + .pointer_catalog_lookup(&router, 1024 * 1024) + .unwrap(); + let lookup = crab_metadata::file_index_lookup::SharedFileIndexLookup::for_pointer_catalog( + router, captured, + ); let hydrator = hydrator.with_file_index_lookup(lookup); assert_eq!( From 43203ff48f6aa7e9b7baf4391892d3535d43f42e Mon Sep 17 00:00:00 2001 From: forhappy Date: Wed, 14 Oct 2026 22:21:23 -0700 Subject: [PATCH 2/2] test(read): name the fixture shard hash type --- crates/crab-read/src/hydrator.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/crab-read/src/hydrator.rs b/crates/crab-read/src/hydrator.rs index c2a1cf139..d5e48f4d8 100644 --- a/crates/crab-read/src/hydrator.rs +++ b/crates/crab-read/src/hydrator.rs @@ -1211,7 +1211,7 @@ mod tests { ); let (body, _) = router .store() - .get_with_etag(&router.shard_path(&shard_hash.into())) + .get_with_etag(&router.shard_path(&crab_xet::hash::MerkleHash::from(shard_hash))) .await .unwrap(); let mut chunks = std::collections::BTreeMap::<_, Vec<_>>::new();