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..d5e48f4d8 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(&crab_xet::hash::MerkleHash::from(shard_hash))) + .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!(