Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions crab/src/cmd/mount.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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 {
Expand Down
6 changes: 3 additions & 3 deletions crab/src/read/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)))
}

Expand Down
8 changes: 7 additions & 1 deletion crates/crab-metadata/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
114 changes: 55 additions & 59 deletions crates/crab-metadata/src/file_index_lookup.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Arc<crate::capsule_protocol::PointerCatalogLookup>>,
static_entries: Option<Arc<HashMap<MerkleHash, MerkleHash>>>,
reader: Option<Arc<slatedb::DbReader>>,
anchor: Option<CommittedShardAnchor>,
Expand All @@ -244,6 +245,7 @@ impl FileIndexLookupSession {
snapshot: &RepositorySnapshot,
) -> Result<Self> {
Ok(Self {
catalog: None,
static_entries: None,
reader: None,
anchor: CommittedShardAnchor::from_snapshot(snapshot)?,
Expand Down Expand Up @@ -307,6 +309,7 @@ impl FileIndexLookupSession {
)?;
let anchor = CommittedShardAnchor::from_snapshot(snapshot)?;
Ok(Self {
catalog: None,
static_entries: None,
reader: None,
anchor,
Expand Down Expand Up @@ -367,6 +370,7 @@ impl FileIndexLookupSession {
})
};
Ok(Self {
catalog: None,
static_entries: None,
reader: None,
anchor,
Expand Down Expand Up @@ -414,6 +418,7 @@ impl FileIndexLookupSession {
})
.collect::<Result<HashMap<_, _>>>()?;
return Ok(Self {
catalog: None,
static_entries: Some(Arc::new(entries)),
reader: None,
anchor: None,
Expand All @@ -438,6 +443,7 @@ impl FileIndexLookupSession {
Err(error) => return Err(error),
};
let mut session = Self {
catalog: None,
static_entries: None,
reader: None,
anchor,
Expand Down Expand Up @@ -477,6 +483,14 @@ impl FileIndexLookupSession {

/// Look up one file hash in the open session.
pub async fn lookup(&self, file_hash: &MerkleHash) -> Result<Option<MerkleHash>> {
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());
}
Expand Down Expand Up @@ -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::<Vec<_>>();
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()
Expand Down Expand Up @@ -739,9 +777,9 @@ fn spawn_shard_parse(
}

enum LookupSource {
Static {
Catalog {
router: crab_storage::StoreLayout<crab_storage::Store>,
entries: Arc<HashMap<MerkleHash, MerkleHash>>,
lookup: Arc<crate::capsule_protocol::PointerCatalogLookup>,
},
Current {
store: crab_storage::Store,
Expand All @@ -759,8 +797,9 @@ enum LookupSource {
impl LookupSource {
async fn open(&self) -> Result<FileIndexLookupSession> {
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(),
Expand Down Expand Up @@ -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<crab_storage::Store>,
catalog: &crate::capsule_protocol::PointerCatalog,
) -> Result<Self> {
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::<Result<HashMap<_, _>>>()?;
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.
Expand Down Expand Up @@ -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)]
Expand Down Expand Up @@ -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<dyn ObjectStore> = 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();
}
}
Loading
Loading