diff --git a/crates/crab-metadata/README.md b/crates/crab-metadata/README.md index 84a0a6de5..197d37621 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 edad84b9d..83c5a9b95 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; @@ -5196,73 +5196,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, @@ -5335,80 +5268,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,