diff --git a/crates/crab-metadata/README.md b/crates/crab-metadata/README.md index ecbe60207..7632f6091 100644 --- a/crates/crab-metadata/README.md +++ b/crates/crab-metadata/README.md @@ -92,6 +92,10 @@ 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. +Ref deletion removes its live visibility closure even when the capsule has no +visibility payload. Retained dictionary objects cannot supply a new ref's base; +complete and paged reconciliation enforce the same deletion boundary. + External pointer catalogs use `CRBPCP01` authenticated pages. A bounded descriptor commits file, shard, xorb and reverse-chunk directories; nested shard closures and ordered xorb chunk lists are paged too. Keys preserve canonical lowercase Xet hex diff --git a/crates/crab-metadata/src/capsule_protocol/captured_visibility.rs b/crates/crab-metadata/src/capsule_protocol/captured_visibility.rs index 60dd7d5cb..f4b2ed08e 100644 --- a/crates/crab-metadata/src/capsule_protocol/captured_visibility.rs +++ b/crates/crab-metadata/src/capsule_protocol/captured_visibility.rs @@ -42,7 +42,16 @@ impl VisibilityReconciliation { index, ) } - Self::Complete(_) => Ok(()), + Self::Complete(index) => { + // A deletion-only capsule has no visibility payload. Its ref + // closure must still disappear before another control borrows it. + for edit in control.transaction().edits() { + if edit.new_oid().is_none() { + index.remove_ref(edit.ref_name()); + } + } + Ok(()) + } Self::Paged(index) => index.apply(control).await, } } @@ -335,6 +344,13 @@ impl CapturedVisibility { /// 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 { + // No payload skips OID changes, not deletion of an entire ref. + // Keep historical dictionary entries separate from live closures. + for edit in control.transaction().edits() { + if edit.new_oid().is_none() { + self.refs.remove(edit.ref_name()); + } + } return Ok(()); }; for (edit, header) in control.transaction().edits().iter().zip(headers) { diff --git a/crates/crab-metadata/src/capsule_protocol/captured_visibility/tests.rs b/crates/crab-metadata/src/capsule_protocol/captured_visibility/tests.rs index e7cd4593b..164fa3215 100644 --- a/crates/crab-metadata/src/capsule_protocol/captured_visibility/tests.rs +++ b/crates/crab-metadata/src/capsule_protocol/captured_visibility/tests.rs @@ -169,6 +169,51 @@ async fn removed_invisible_objects_do_not_supply_new_ref_bases() { ); } +#[tokio::test] +async fn deleted_ref_objects_remain_in_dictionary_but_cannot_supply_new_bases() { + let layout = StoreLayout::new( + Store::new(Arc::new(InMemory::new())), + "repositories/deleted".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 transaction = CapsuleTransaction::new( + &"1".repeat(64), + vec![CapsuleRefEdit::new( + "refs/heads/main", + Some(hex(1)), + None, + None, + )], + ) + .unwrap(); + let run = + CapsuleRun::leaf(Capsule::build(&transaction, Vec::new(), Vec::new()).unwrap()).unwrap(); + let (_, mut controls) = run.verified_controls().unwrap(); + lazy.apply(&controls.remove(0)).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::new()).await.unwrap() + ), + (vec![true], false) + ); +} + #[tokio::test] async fn removal_outside_captured_prior_closure_is_rejected() { let layout = StoreLayout::new( diff --git a/crates/crab-metadata/src/capsule_protocol/store.rs b/crates/crab-metadata/src/capsule_protocol/store.rs index 8cc7fe82f..d3a6ad266 100644 --- a/crates/crab-metadata/src/capsule_protocol/store.rs +++ b/crates/crab-metadata/src/capsule_protocol/store.rs @@ -15,6 +15,9 @@ use crate::error::Result; #[cfg(test)] mod tests; +#[cfg(test)] +mod catalog_order_tests; + /// A verified stored root and the provider token protecting its next update. #[derive(Debug, Clone)] pub struct RootSnapshot { diff --git a/crates/crab-metadata/src/capsule_protocol/store/catalog_order_tests.rs b/crates/crab-metadata/src/capsule_protocol/store/catalog_order_tests.rs new file mode 100644 index 000000000..384127111 --- /dev/null +++ b/crates/crab-metadata/src/capsule_protocol/store/catalog_order_tests.rs @@ -0,0 +1,583 @@ +#![expect(clippy::unwrap_used, reason = "test assertions")] + +use super::*; +use crate::capsule_protocol::{ + Capsule, CapsuleGitPack, CapsulePointer, CapsuleRefEdit, CapsuleSection, CapsuleTransaction, + CapsuleVisibilityDelta, CheckpointPointer, FileCatalogEntry, HistorySegmentState, + LayeredObjectMember, LayeredVisibilitySnapshot, PackSourceDescriptor, RepositoryRoot, + ShardCatalogEntry, VisibilityReconciliation, XorbCatalogEntry, XorbChunkEntry, + apply_control_refs, load_pointer_catalog_lookup_from_root, order_ref_controls, + source_catalog_digest, +}; +use crate::git_visibility::{GitVisibilityEdit, GitVisibilityIndex}; +use bytes::Bytes; +use std::sync::Arc; + +async fn expose( + layout: &StoreLayout, + root: &RootSnapshot, + transaction: &CapsuleTransaction, + pointer: CapsulePointer, +) { + 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(); +} + +struct BorrowingFixture { + layout: StoreLayout, + root: RootSnapshot, + original: CapsuleTransaction, + borrowed: CapsuleTransaction, + earlier: CapsuleTransaction, + controls: BTreeMap, + pointers: BTreeMap, + original_source: PackSourceDescriptor, + file: String, + original_shard: String, + borrowed_shard: String, +} + +// Fixed publication identities keep ordering coverage independent of nonce sampling. +fn ordering_transaction(root: &RootSnapshot, edit: CapsuleRefEdit) -> CapsuleTransaction { + let edits = vec![edit]; + let identity = blake3::hash(&serde_json::to_vec(&edits).unwrap()); + CapsuleTransaction::for_protected_source(root.record().digest(), &identity.to_hex(), edits) + .unwrap() +} + +async fn borrowing_fixture() -> BorrowingFixture { + let layout = StoreLayout::new( + Store::new(Arc::new(object_store::memory::InMemory::new())), + "checkpoint-precedence".to_owned(), + ); + let root = create_root( + &layout, + RootRecord::encode(RepositoryRoot::initial(&"1".repeat(64), "refs/heads/main").unwrap()) + .unwrap(), + ) + .await + .unwrap(); + let tip = "1".repeat(40); + let transaction = |name: String| { + ordering_transaction( + &root, + CapsuleRefEdit::new(name, None, Some(tip.clone()), None), + ) + }; + let borrowed = (0..64) + .map(|i| transaction(format!("refs/heads/borrow-{i}"))) + .find(|t| (b'4'..b'8').contains(&t.id().unwrap().as_bytes()[0])) + .unwrap(); + let borrowed_id = borrowed.id().unwrap(); + let mut producers = (0..256) + .map(|i| transaction(format!("refs/heads/source-{i}"))) + .filter(|t| t.id().unwrap() > borrowed_id) + .take(2) + .collect::>(); + producers.sort_by_key(|t| t.id().unwrap()); + let earlier = producers.remove(0); + let original = producers.remove(0); + let original_id = original.id().unwrap(); + let earlier_id = earlier.id().unwrap(); + let file = "3".repeat(64); + let original_shard = "4".repeat(64); + let borrowed_shard = "5".repeat(64); + let xorb = "6".repeat(64); + let mut descriptors = PointerCatalog::new(); + descriptors + .insert_xorb( + xorb.clone(), + XorbCatalogEntry::new( + 99, + "8".repeat(64), + vec![XorbChunkEntry::new("7".repeat(64), 16)], + ), + ) + .unwrap(); + for shard in [&original_shard, &borrowed_shard] { + descriptors + .insert_shard( + shard.clone(), + ShardCatalogEntry::new(80, vec![xorb.clone()]), + ) + .unwrap(); + } + let mut controls = BTreeMap::new(); + let mut pointers = BTreeMap::new(); + let mut original_source = None; + for t in [&original, &borrowed, &earlier] { + let id = t.id().unwrap(); + let edit = if id == borrowed_id { + GitVisibilityEdit::from_delta_objects(Some(tip.clone()), tip.clone(), vec![], vec![]) + } else { + GitVisibilityEdit::from_replacement_objects(None, tip.clone(), vec![tip.clone()]) + }; + let visibility = CapsuleVisibilityDelta::new(BTreeMap::from([( + t.edits()[0].ref_name().to_owned(), + edit, + )])) + .unwrap(); + let mut catalog = if id == original_id { + descriptors.clone() + } else { + PointerCatalog::new() + }; + if id != earlier_id { + catalog + .insert_file( + file.clone(), + FileCatalogEntry::new( + 16, + if id == borrowed_id { + borrowed_shard.clone() + } else { + original_shard.clone() + }, + ), + ) + .unwrap(); + } + // These bodies exercise metadata codecs; this is not native Git or + // reconstruction qualification. The single object admits the shared tip. + let packs = if id == original_id { + vec![ + CapsuleGitPack::new( + Bytes::from_static(b"pack"), + Bytes::from_static(b"index"), + Bytes::from_static(b"reverse"), + Bytes::from_static(b"locator"), + tip.clone(), + 1, + ) + .unwrap(), + ] + } else { + vec![] + }; + let run = CapsuleRun::leaf( + Capsule::build( + t, + packs, + vec![ + CapsuleSection::new( + CapsuleSectionKind::VisibilityDelta, + visibility.encode().unwrap(), + ), + CapsuleSection::new( + CapsuleSectionKind::CatalogDelta, + catalog.encode_delta().unwrap(), + ), + ], + ) + .unwrap(), + ) + .unwrap(); + layout + .store() + .create_strict(&layout.capsule_path(run.hash()), run.bytes().clone()) + .await + .unwrap(); + let (control, mut capsules) = run.verified_controls().unwrap(); + if id == original_id { + original_source = Some(control.source_descriptor().unwrap()); + } + controls.insert(id.clone(), capsules.remove(0)); + pointers.insert( + id, + 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(), + ); + } + BorrowingFixture { + layout, + root, + original, + borrowed, + earlier, + controls, + pointers, + original_source: original_source.unwrap(), + file, + original_shard, + borrowed_shard, + } +} + +#[tokio::test] +async fn checkpoint_replay_base_binds_catalog_precedence() { + let BorrowingFixture { + layout, + root, + original, + borrowed, + earlier, + controls, + pointers, + original_source, + file, + original_shard, + borrowed_shard, + } = borrowing_fixture().await; + let original_id = original.id().unwrap(); + let borrowed_id = borrowed.id().unwrap(); + let earlier_id = earlier.id().unwrap(); + for t in [&original, &borrowed] { + expose(&layout, &root, t, pointers[&t.id().unwrap()].clone()).await; + } + let before = load_pointer_catalog_from_root(&layout, &root) + .await + .unwrap(); + let empty = + GitVisibilityIndex::new(0, "0".repeat(64), "0".repeat(64), BTreeMap::new()).unwrap(); + let ordered = order_ref_controls( + &BTreeMap::new(), + &BTreeMap::new(), + VisibilityReconciliation::Complete(empty.clone()), + &BTreeMap::new(), + controls + .iter() + .filter(|(id, _)| **id != earlier_id) + .map(|(id, c)| (id.clone(), c.clone())) + .collect(), + ) + .await + .unwrap(); + let mut visibility = VisibilityReconciliation::Complete(empty); + let mut refs = BTreeMap::new(); + let mut peeled = BTreeMap::new(); + for control in &ordered { + apply_control_refs(control, &mut refs, &mut peeled).unwrap(); + visibility.apply(control).await.unwrap(); + } + let VisibilityReconciliation::Complete(index) = visibility else { + unreachable!() + }; + let sources = vec![original_source]; + let snapshot = LayeredVisibilitySnapshot::from_index_with_member_admission( + &index, + &source_catalog_digest(&sources).unwrap(), + vec![LayeredObjectMember::new(0, 0)], + ) + .unwrap(); + let checkpoint = LayeredCheckpoint::build( + root.record().root().generation(), + root.record().digest(), + sources, + before.clone(), + Some(snapshot), + ) + .unwrap(); + layout + .store() + .create_strict( + &layout.capsule_checkpoint_path(checkpoint.hash()), + checkpoint.bytes().clone(), + ) + .await + .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 positions = [&original, &borrowed] + .into_iter() + .map(|t| (t.edits()[0].ref_name().to_owned(), t.id().unwrap())) + .collect::>(); + let history = HistorySegment::build( + pointer.clone(), + None, + HistorySegmentState::new( + refs.clone(), + peeled.clone(), + root.record().root().head().to_owned(), + positions.clone(), + vec![ + pointers[&original_id].clone(), + pointers[&borrowed_id].clone(), + ], + ), + ) + .unwrap(); + layout + .store() + .create_strict( + &layout.capsule_history_segment_path(history.hash()), + history.bytes().clone(), + ) + .await + .unwrap(); + let next = RootRecord::encode( + root.record() + .root() + .install_ref_checkpoint( + root.record().digest(), + pointer, + history.pointer().unwrap(), + refs, + peeled, + positions, + ) + .unwrap(), + ) + .unwrap(); + let etag = layout + .store() + .update( + &layout.capsule_root_path(), + next.bytes().clone(), + root.etag().clone(), + ) + .await + .unwrap(); + let compacted = root.committed_ref_checkpoint(next, etag).unwrap(); + // Keep the original base to characterize explicitly pinned snapshot inputs; + // concurrent writer admission is a separate contract from this metadata API. + expose(&layout, &root, &earlier, pointers[&earlier_id].clone()).await; + let mut winners = Vec::new(); + for captured in [&root, &compacted] { + let full = load_pointer_catalog_from_root(&layout, captured) + .await + .unwrap(); + let selected = load_pointer_catalog_lookup_from_root(&layout, captured) + .await + .unwrap() + .files(layout.store(), std::slice::from_ref(&file)) + .await + .unwrap(); + assert_eq!(selected.files().get(&file), full.files().get(&file)); + winners.push(selected.files()[&file].shard_hash().to_owned()); + } + // A checkpoint is an authority boundary, not a transparent reader shortcut. + // It freezes the old replay base even when later visible refs supply an old tip. + assert_eq!( + (before.files()[&file].shard_hash().to_owned(), winners), + (borrowed_shard.clone(), vec![original_shard, borrowed_shard]) + ); +} + +#[tokio::test] +async fn independent_visibility_can_reorder_prior_borrowing_controls() { + let fixture = borrowing_fixture().await; + let original_id = fixture.original.id().unwrap(); + let borrowed_id = fixture.borrowed.id().unwrap(); + let earlier_id = fixture.earlier.id().unwrap(); + let mut paged = BTreeMap::new(); + for pointer in fixture.pointers.values() { + let (_, mut controls) = load_capsule_run_git_control(&fixture.layout, pointer) + .await + .unwrap(); + let control = controls.remove(0); + paged.insert(control.transaction_id().to_owned(), control); + } + for (is_paged, controls) in [(false, fixture.controls), (true, paged)] { + for with_earlier in [false, true] { + let index = GitVisibilityIndex::new(0, "0".repeat(64), "0".repeat(64), BTreeMap::new()) + .unwrap(); + let visibility = if is_paged { + VisibilityReconciliation::Paged(Box::new( + crate::capsule_protocol::CapturedVisibility::new( + fixture.layout.store().clone(), + index, + std::iter::empty(), + ), + )) + } else { + VisibilityReconciliation::Complete(index) + }; + let pending = controls + .iter() + .filter(|(id, _)| with_earlier || **id != earlier_id) + .map(|(id, c)| (id.clone(), c.clone())) + .collect(); + let ordered = order_ref_controls( + &BTreeMap::new(), + &BTreeMap::new(), + visibility, + &BTreeMap::new(), + pending, + ) + .await + .unwrap(); + let expected = if with_earlier { + vec![ + earlier_id.as_str(), + borrowed_id.as_str(), + original_id.as_str(), + ] + } else { + vec![original_id.as_str(), borrowed_id.as_str()] + }; + // Another visible producer can release an old borrowing control sooner; + // physical run presence alone must never change the captured selection. + assert_eq!( + ordered + .iter() + .map(CapsuleControl::transaction_id) + .collect::>(), + expected, + "paged={is_paged}, with_earlier={with_earlier}" + ); + } + } +} + +#[tokio::test] +async fn deleting_a_visible_producer_can_delay_borrowing_controls() { + let mut fixture = borrowing_fixture().await; + let borrowed_id = fixture.borrowed.id().unwrap(); + let original_id = fixture.original.id().unwrap(); + let tip = fixture.original.edits()[0].new_oid().unwrap().to_owned(); + let (producer, deletion) = (0..16_384) + .find_map(|i| { + let name = format!("refs/heads/transient-{i}"); + let producer = ordering_transaction( + &fixture.root, + CapsuleRefEdit::new(name.clone(), None, Some(tip.clone()), None), + ); + let deletion = ordering_transaction( + &fixture.root, + CapsuleRefEdit::new(name, Some(tip.clone()), None, None), + ); + let producer_id = producer.id().unwrap(); + let deletion_id = deletion.id().unwrap(); + (producer_id < deletion_id && deletion_id < borrowed_id).then_some((producer, deletion)) + }) + .unwrap(); + let producer_id = producer.id().unwrap(); + let deletion_id = deletion.id().unwrap(); + for (transaction, edits) in [ + ( + &producer, + BTreeMap::from([( + producer.edits()[0].ref_name().to_owned(), + GitVisibilityEdit::from_replacement_objects(None, tip.clone(), vec![tip.clone()]), + )]), + ), + (&deletion, BTreeMap::new()), + ] { + let sections = if edits.is_empty() { + Vec::new() + } else { + vec![CapsuleSection::new( + CapsuleSectionKind::VisibilityDelta, + CapsuleVisibilityDelta::new(edits) + .unwrap() + .encode() + .unwrap(), + )] + }; + let run = + CapsuleRun::leaf(Capsule::build(transaction, Vec::new(), sections).unwrap()).unwrap(); + fixture + .layout + .store() + .create_strict( + &fixture.layout.capsule_path(run.hash()), + run.bytes().clone(), + ) + .await + .unwrap(); + let (_, mut controls) = run.verified_controls().unwrap(); + let id = transaction.id().unwrap(); + fixture.controls.insert(id.clone(), controls.remove(0)); + fixture.pointers.insert( + id, + 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(), + ); + } + fixture.controls.remove(&fixture.earlier.id().unwrap()); + let mut paged = BTreeMap::new(); + for id in fixture.controls.keys() { + let (_, mut controls) = + load_capsule_run_git_control(&fixture.layout, &fixture.pointers[id]) + .await + .unwrap(); + paged.insert(id.clone(), controls.remove(0)); + } + let mut observed = Vec::new(); + for (is_paged, controls) in [(false, fixture.controls), (true, paged)] { + let index = + GitVisibilityIndex::new(0, "0".repeat(64), "0".repeat(64), BTreeMap::new()).unwrap(); + let visibility = if is_paged { + VisibilityReconciliation::Paged(Box::new( + crate::capsule_protocol::CapturedVisibility::new( + fixture.layout.store().clone(), + index, + std::iter::empty(), + ), + )) + } else { + VisibilityReconciliation::Complete(index) + }; + let ordered = order_ref_controls( + &BTreeMap::new(), + &BTreeMap::new(), + visibility, + &BTreeMap::new(), + controls, + ) + .await + .unwrap(); + // The transient source disappears before borrowing wins admission. + // Treating visibility as monotone would incorrectly put borrowing + // before the surviving producer and reverse their file-binding winner. + observed.push( + ordered + .iter() + .map(|control| control.transaction_id().to_owned()) + .collect::>(), + ); + } + let expected = vec![producer_id, deletion_id, original_id, borrowed_id]; + assert_eq!(observed, vec![expected.clone(), expected]); +} diff --git a/crates/crab-metadata/src/capsule_protocol/store/tests.rs b/crates/crab-metadata/src/capsule_protocol/store/tests.rs index 45a3729dd..7443a559b 100644 --- a/crates/crab-metadata/src/capsule_protocol/store/tests.rs +++ b/crates/crab-metadata/src/capsule_protocol/store/tests.rs @@ -589,35 +589,23 @@ async fn full_catalog_matches_selected_closure_across_independent_ref_dependenci ) .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( + let transaction = |name, tip| { + CapsuleTransaction::new( root.record().digest(), - vec![CapsuleRefEdit::new( - format!("refs/tags/consumer-{index}"), - None, - Some("2".repeat(40)), - None, - )], + vec![CapsuleRefEdit::new(name, None, Some(tip), 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"); + .unwrap() + }; + let left = transaction("refs/tags/left", "1".repeat(40)); + let right = transaction("refs/tags/right", "2".repeat(40)); + // Assign roles after nonce generation; a bounded search against one random + // descriptor could fail before testing dependency assembly at all. + let (binding, descriptors) = if left.id().unwrap() < right.id().unwrap() { + (left, right) + } else { + (right, left) + }; + assert!(binding.id().unwrap() < descriptors.id().unwrap()); let file = "3".repeat(64); let shard = "4".repeat(64); let xorb = "5".repeat(64); diff --git a/crates/crab-read/README.md b/crates/crab-read/README.md index eb53e0b23..b7e996abd 100644 --- a/crates/crab-read/README.md +++ b/crates/crab-read/README.md @@ -55,7 +55,9 @@ capsule-relative, while transport ranges stay on the captured run. Shared transactions deduplicate without discarding conflicting content evidence. Paged closure layers preserve the inherited base when its source ref later moves, and dictionary retention -never supplies visibility for a new ref. Initial rooted additions use their +never supplies visibility for a new ref. Deletion-only capsules remove their +live ref closure even without an OID payload, in both full-body and control reads. +Initial rooted additions use their authenticated count; differing actual delta additions/removals remain admitted work. `git_visibility_index`, `candidate_git_visibility` and `tip_bound_transitions` asynchronously materialize captured controls only when diff --git a/crates/crab-read/src/capsule_protocol.rs b/crates/crab-read/src/capsule_protocol.rs index 1272dd352..d2dba2a94 100644 --- a/crates/crab-read/src/capsule_protocol.rs +++ b/crates/crab-read/src/capsule_protocol.rs @@ -4649,9 +4649,7 @@ async fn assemble_view( } let mut visibility_index = visibility_index_for_checkpoint(checkpoint.as_ref())?; for capsule in &root_capsules { - if capsule.visibility_delta()?.is_some() { - apply_capsule_visibility_index(capsule, &mut visibility_index)?; - } + reconcile_capsule_visibility(capsule, &mut visibility_index)?; } let ordered = order_ref_capsules( snapshot.record().root().refs(), @@ -5160,9 +5158,7 @@ fn order_ref_capsules( .remove(&ready) .ok_or_else(|| ReadError::internal("ready capsule disappeared"))?; apply_capsule_refs(&capsule, &mut refs, &mut peeled)?; - if capsule.visibility_delta()?.is_some() { - apply_capsule_visibility_index(&capsule, &mut visibility)?; - } + reconcile_capsule_visibility(&capsule, &mut visibility)?; applied.insert(capsule.transaction_id().to_owned()); ordered.push(capsule); } @@ -5198,6 +5194,27 @@ fn apply_capsule_visibility_index( .map_err(metadata_control_error) } +fn reconcile_capsule_visibility( + capsule: &Capsule, + index: &mut crab_metadata::git_visibility::GitVisibilityIndex, +) -> Result<()> { + let transaction = capsule.transaction()?; + let delta = capsule.visibility_delta()?; + if delta.is_some() { + return apply_transaction_visibility_index(&transaction, delta.as_ref(), index) + .map_err(metadata_control_error); + } + // Reconciliation permits metadata-only live edits; deep inspection still + // requires their evidence. Deletions revoke closures without an OID payload, + // or a later borrower could inherit a deleted ref. + for edit in transaction.edits() { + if edit.new_oid().is_none() { + index.remove_ref(edit.ref_name()); + } + } + Ok(()) +} + fn apply_capsule_refs( capsule: &Capsule, refs: &mut BTreeMap, @@ -7376,6 +7393,86 @@ mod tests { ); } + #[test] + fn full_ref_ordering_cannot_borrow_from_a_deleted_ref() { + let tip = "1".repeat(40); + let mut pending = BTreeMap::new(); + for (key, name, old, new, evidence) in [ + ( + "a-delete", + "refs/heads/transient", + Some(tip.clone()), + None, + None, + ), + ( + "b-borrow", + "refs/heads/borrow", + None, + Some(tip.clone()), + Some(GitVisibilityEdit::from_delta_objects( + Some(tip.clone()), + tip.clone(), + Vec::new(), + Vec::new(), + )), + ), + ( + "z-source", + "refs/heads/source", + None, + Some(tip.clone()), + Some(GitVisibilityEdit::from_replacement_objects( + None, + tip.clone(), + vec![tip.clone()], + )), + ), + ] { + let transaction = CapsuleTransaction::new( + &"a".repeat(64), + vec![CapsuleRefEdit::new(name, old, new, None)], + ) + .unwrap(); + let sections = evidence + .map(|edit| { + CapsuleSection::new( + CapsuleSectionKind::VisibilityDelta, + CapsuleVisibilityDelta::new(BTreeMap::from([(name.to_owned(), edit)])) + .unwrap() + .encode() + .unwrap(), + ) + }) + .into_iter() + .collect(); + pending.insert( + key.to_owned(), + Capsule::build(&transaction, Vec::new(), sections).unwrap(), + ); + } + let expected = ["a-delete", "z-source", "b-borrow"] + .map(|key| pending[key].transaction_id().to_owned()); + let ordered = order_ref_capsules( + &BTreeMap::from([("refs/heads/transient".to_owned(), tip.clone())]), + &BTreeMap::new(), + visibility_index(BTreeMap::from([( + "refs/heads/transient".to_owned(), + vec![tip], + )])), + &BTreeMap::new(), + pending, + ) + .unwrap(); + assert_eq!( + ordered + .iter() + .map(|capsule| capsule.transaction_id().to_owned()) + .collect::>(), + expected, + ); + } + #[test] fn ref_capsules_follow_authenticated_predecessors_when_oid_repeats() { let oid_a = "1".repeat(40); diff --git a/crates/crab-remote/README.md b/crates/crab-remote/README.md index 5c391f6d2..0364edf60 100644 --- a/crates/crab-remote/README.md +++ b/crates/crab-remote/README.md @@ -70,6 +70,9 @@ verified create/read identities with their source catalog, publishes rebuildable chunk hints only after proven ref visibility, and preserves a committed outcome if hint-index maintenance fails. Uncertain publication does not publish hints. Caller-owned ref leases and GC fences cover this complete operation. +New refs and explicit cross-ref visibility bases publish complete replacement +closures from the captured proof. Only same-ref updates retain additive evidence; +a different ref's later deletion must not invalidate an acknowledged branch. `objects` owns bounded tree reads, deterministic tree edits and encoding, commit encoding, and Git object identities. `prepare` owns normalized packs and diff --git a/crates/crab-remote/src/prepare.rs b/crates/crab-remote/src/prepare.rs index 7210a1e20..b96de1ab0 100644 --- a/crates/crab-remote/src/prepare.rs +++ b/crates/crab-remote/src/prepare.rs @@ -910,6 +910,47 @@ impl Source<'_> { } } + fn prior_objects(&self, maximum: usize) -> Result> { + let name = self.prior.as_ref().map(|(name, _)| name.as_str()); + let Some(name) = name else { + return Err(Error::Request("Additive visibility has no captured base")); + }; + match &self.proof { + Some(VisibilityProof::Materialized(proof)) => { + if proof.object_count_for_refs([name]) > maximum { + return Err(Error::Request( + "Visibility base exceeds the graph object limit", + )); + } + proof.objects_for_ref(name).ok_or(Error::Request( + "Visibility base is absent from its captured proof", + )) + } + Some(VisibilityProof::Catalog(proof)) => { + if proof.object_count_for_refs([name]) > maximum { + return Err(Error::Request( + "Visibility base exceeds the graph object limit", + )); + } + let ordinals = proof.ordinals_for_refs([name]); + let objects = self + .handle + .block_on(self.operation.catalog_object_ids_by_ordinal(&ordinals))?; + if objects.len() != ordinals.len() || objects.iter().any(Option::is_none) { + return Err(Error::Request( + "Visibility base has missing catalog objects", + )); + } + Ok(objects + .into_iter() + .flatten() + .map(|oid| oid.to_string()) + .collect()) + } + None => Err(Error::Request("Visibility base has no captured proof")), + } + } + fn contains_ref(&self, name: &str) -> bool { match &self.proof { Some(VisibilityProof::Catalog(proof)) => proof.contains_ref(name), @@ -1173,6 +1214,24 @@ where || cancel.is_cancelled(), )?; let evidence = match proof { + RefVisibility::Additive { added, .. } + if update.old.is_none() + || source + .prior + .as_ref() + .is_some_and(|(name, _)| name != &update.name) => + { + // A different ref can disappear before canonical replay. + // Copy its pinned closure into replacement evidence so an + // acknowledged ref never borrows that mutable lifetime. + let mut objects = source.prior_objects(options.graph.max_graph_steps)?; + objects.extend(added.into_iter().map(|oid| oid.to_string())); + GitVisibilityEdit::from_replacement_objects( + update.old.map(|oid| oid.to_string()), + new.to_string(), + objects, + ) + } RefVisibility::Additive { base, added } => { GitVisibilityEdit::from_delta_objects( Some(base.to_string()), diff --git a/crates/crab-remote/tests/capsule_preparation.rs b/crates/crab-remote/tests/capsule_preparation.rs index 52d9db6fe..76539482e 100644 --- a/crates/crab-remote/tests/capsule_preparation.rs +++ b/crates/crab-remote/tests/capsule_preparation.rs @@ -195,12 +195,34 @@ async fn prepare( crab_read::capsule_protocol::CapsuleRepositoryView, Arc, crab_remote::publication::PublicationLeases, +) { + let oid = gix_hash::ObjectId::from_hex( + String::from_utf8(git(directory, &["rev-parse", "HEAD"])) + .unwrap() + .trim() + .as_bytes(), + ) + .unwrap(); + prepare_update(layout, directory, content, "refs/heads/main", Some(oid)).await +} + +async fn prepare_update( + layout: &crab_storage::StoreLayout, + directory: &std::path::Path, + content: Option, + ref_name: &str, + new: Option, +) -> ( + crab_remote::prepare::Prepared, + crab_read::capsule_protocol::CapsuleRepositoryView, + Arc, + crab_remote::publication::PublicationLeases, ) { let cancel = CancellationToken::new(); let leases = crab_remote::publication::acquire_leases( layout.store(), layout, - ["refs/heads/main".to_owned()], + [ref_name.to_owned()], crab_remote::publication::LeaseOptions::renewing(std::time::Duration::from_secs(60)), &cancel, ) @@ -231,24 +253,24 @@ async fn prepare( ) .await .unwrap(); - let oid = gix_hash::ObjectId::from_hex( - String::from_utf8(git(directory, &["rev-parse", "HEAD"])) - .unwrap() - .trim() - .as_bytes(), - ) - .unwrap(); - let mut revisions = String::from("HEAD\n"); - if let Some(old) = view.refs().get("refs/heads/main") { - revisions.push_str(&format!("^{old}\n")); - } - let pack = git_input( - directory, - &["pack-objects", "--stdout", "--revs"], - revisions.as_bytes(), - ); - let incoming = directory.join("incoming.pack"); - std::fs::write(&incoming, pack).unwrap(); + let incoming = if let Some(oid) = new { + let mut revisions = format!("{oid}\n"); + for old in view.refs().values() { + revisions.push_str(&format!("^{old}\n")); + } + let pack = git_input( + directory, + &["pack-objects", "--stdout", "--revs"], + revisions.as_bytes(), + ); + let incoming = directory.join("incoming.pack"); + std::fs::write(&incoming, pack).unwrap(); + Some(std::io::BufReader::new( + std::fs::File::open(incoming).unwrap(), + )) + } else { + None + }; let limits = crab_git::incoming_pack::ReceiveLimits { max_pack_bytes: LIMIT, max_objects: 1024, @@ -260,16 +282,14 @@ async fn prepare( repository, view.git_visibility_index().await.unwrap(), crab_git::incoming_pack::ScratchSpace::new(directory, limits), - Some(std::io::BufReader::new( - std::fs::File::open(incoming).unwrap(), - )), + incoming, vec![crab_git::receive_plan::RefUpdate { - name: "refs/heads/main".to_owned(), + name: ref_name.to_owned(), old: view .refs() - .get("refs/heads/main") + .get(ref_name) .map(|oid| gix_hash::ObjectId::from_hex(oid.as_bytes()).unwrap()), - new: Some(oid), + new, }], std::collections::BTreeMap::new(), &cancel, @@ -619,3 +639,6 @@ async fn historical_pointer_lookup_preserves_declared_size_checks() { ); } } + +#[path = "capsule_preparation/ref_visibility.rs"] +mod ref_visibility; diff --git a/crates/crab-remote/tests/capsule_preparation/ref_visibility.rs b/crates/crab-remote/tests/capsule_preparation/ref_visibility.rs new file mode 100644 index 000000000..841fcbe63 --- /dev/null +++ b/crates/crab-remote/tests/capsule_preparation/ref_visibility.rs @@ -0,0 +1,294 @@ +use super::*; + +#[tokio::test(flavor = "multi_thread")] +async fn acknowledged_branch_survives_source_deletion() { + branch_survives_source_deletion(layout().await.0, false).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn prepared_branch_survives_source_deletion_before_commit() { + branch_survives_source_deletion(layout().await.0, true).await; +} + +#[tokio::test(flavor = "multi_thread")] +#[ignore = "requires CRAB_E2E_FLAT_BUCKET and explicit scratch S3 credentials"] +async fn live_branch_survives_source_deletion_and_source_removal() { + let bucket = std::env::var("CRAB_E2E_FLAT_BUCKET").expect("explicit scratch bucket required"); + let store = + crab_storage::build_static_env_store(&bucket, crab_storage::StorageProviderKind::S3) + .unwrap(); + let nonce = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + for before_commit in [false, true] { + let prefix = format!("ref-lifetime/{nonce}/{before_commit}"); + let layout = crab_storage::StoreLayout::with_global_prefix( + store.clone(), + format!("{prefix}/repo"), + format!("{prefix}/global"), + ); + crab_write::capsule_protocol::initialize(&layout, &"1".repeat(64), "refs/heads/main") + .await + .unwrap(); + branch_survives_source_deletion(layout, before_commit).await; + } +} + +async fn branch_survives_source_deletion( + layout: crab_storage::StoreLayout, + delete_before_commit: bool, +) { + use crab_metadata::capsule_protocol::{CapsuleRefEdit, CapsuleTransaction}; + use crab_remote::prepare::CapsuleCommitOutcome; + let cancel = CancellationToken::new(); + let (directory, content) = content_fixture().await; + source_commit(directory.path(), &content); + let (prepared, view, runtime, leases) = prepare(&layout, directory.path(), Some(content)).await; + let artifacts = prepared + .upload_capsule(&view, proof_limits(), &cancel) + .await + .unwrap(); + let committed = artifacts + .commit(None, std::time::Duration::from_secs(60), &cancel) + .await; + leases.release_after_commit().await; + runtime.shutdown().await; + assert!(matches!( + committed.unwrap(), + CapsuleCommitOutcome::Committed { .. } + )); + let tip = gix_hash::ObjectId::from_hex( + String::from_utf8(git(directory.path(), &["rev-parse", "HEAD"])) + .unwrap() + .trim() + .as_bytes(), + ) + .unwrap(); + let mut expected_objects = String::from_utf8(git( + directory.path(), + &["rev-list", "--objects", "--no-object-names", "HEAD"], + )) + .unwrap() + .lines() + .map(str::to_owned) + .collect::>(); + expected_objects.sort_unstable(); + let expected_bodies = expected_objects + .iter() + .map(|oid| { + let kind = String::from_utf8(git(directory.path(), &["cat-file", "-t", oid])).unwrap(); + ( + gix_hash::ObjectId::from_hex(oid.as_bytes()).unwrap(), + git(directory.path(), &["cat-file", kind.trim(), oid]), + ) + }) + .collect::>(); + let crab_git::pointer_detect::PointerKind::Crab(pointer) = crab_git::pointer_detect::classify( + &std::fs::read(directory.path().join("model.bin")).unwrap(), + ) else { + panic!("fixture pointer"); + }; + let borrowed = "refs/heads/borrowed"; + let (prepared, view, runtime, leases) = + prepare_update(&layout, directory.path(), None, borrowed, Some(tip)).await; + let edit = CapsuleRefEdit::new( + borrowed, + None, + Some(tip.to_string()), + prepared + .plan() + .peeled() + .get(borrowed) + .map(ToString::to_string), + ); + let (plan, expected_borrow) = (0..256) + .find_map(|i| { + let plan = blake3::hash(format!("borrow-lifetime:{i}").as_bytes()) + .to_hex() + .to_string(); + let transaction = CapsuleTransaction::for_plan( + view.root_snapshot().record().digest(), + &plan, + vec![edit.clone()], + ) + .unwrap(); + let id = transaction.id().unwrap(); + (id.as_bytes()[0] >= b'8').then_some((plan, id)) + }) + .unwrap(); + let artifacts = prepared + .upload_capsule(&view, proof_limits(), &cancel) + .await + .unwrap(); + if delete_before_commit { + delete_source(&layout, directory.path(), tip, &expected_borrow).await; + } + let committed = artifacts + .commit(Some(&plan), std::time::Duration::from_secs(60), &cancel) + .await; + leases.release_after_commit().await; + runtime.shutdown().await; + let CapsuleCommitOutcome::Committed { transaction_id } = committed.unwrap() else { + panic!("borrowed ref publication must have a proven terminal outcome"); + }; + assert_eq!(transaction_id, expected_borrow); + let root = crab_write::capsule_protocol::open_root(&layout) + .await + .unwrap(); + let (heads, active) = + crab_metadata::capsule_protocol::capture_ref_heads(&layout, root.record().root()) + .await + .unwrap(); + let head = heads + .iter() + .find(|head| head.ref_name() == borrowed) + .unwrap(); + let capsule_pointer = &head.visible(&active).frontier()[0]; + let (_, controls) = + crab_metadata::capsule_protocol::load_capsule_run_git_control(&layout, capsule_pointer) + .await + .unwrap(); + let control = controls[0] + .materialize_visibility(layout.store()) + .await + .unwrap(); + let evidence = &control.visibility_delta().unwrap().unwrap().edits()[borrowed]; + assert_eq!( + (evidence.replaces, &evidence.old_oid, &evidence.added), + (true, &None, &expected_objects), + ); + if !delete_before_commit { + delete_source(&layout, directory.path(), tip, &expected_borrow).await; + } + let root = crab_write::capsule_protocol::open_root(&layout) + .await + .unwrap(); + std::fs::remove_dir_all(directory.path()).unwrap(); + let limits = crab_read::capsule_protocol::CapsuleReadLimits { + max_capsule_bytes: LIMIT, + max_frontier_bytes: LIMIT, + }; + for mode in ["full", "control", "paged"] { + let view = match mode { + "full" => { + crab_read::capsule_protocol::open_view_from_root(&layout, root.clone(), limits) + .await + } + "control" => { + crab_read::capsule_protocol::open_view_from_root_with_metadata( + &layout, + root.clone(), + limits, + ) + .await + } + _ => { + crab_read::capsule_protocol::open_git_view_from_root(&layout, root.clone(), limits) + .await + } + } + .unwrap_or_else(|error| panic!("acknowledged state must reopen through {mode}: {error}")); + let index = view.git_visibility_index().await.unwrap(); + assert_eq!( + ( + view.refs().get(borrowed).map(String::as_str), + view.refs().contains_key("refs/heads/main"), + index.objects_for_ref(borrowed) + ), + ( + Some(tip.to_string().as_str()), + false, + Some(expected_objects.clone()) + ), + ); + let runtime = Arc::new(crab_remote_git::RemoteGitRuntime::default()); + let repository = view + .git_repository_from_store( + layout.clone(), + crab_remote_git::RepositoryIdentity::new("qualification", "branch", 1).unwrap(), + runtime.clone(), + crab_remote_git::RepositoryOptions::default(), + LIMIT, + &cancel, + ) + .await + .unwrap(); + let operation = repository + .operation(crab_remote_git::OperationKind::Repository, &cancel) + .await + .unwrap(); + let oids = expected_bodies + .iter() + .map(|(oid, _)| *oid) + .collect::>(); + let result = operation.read_objects(&oids).await; + let result = operation.finish(result).await; + runtime.shutdown().await; + assert_eq!( + result + .unwrap() + .into_iter() + .map(|object| (object.oid, object.data.to_vec())) + .collect::>(), + expected_bodies, + "cold {mode} object bytes after source removal", + ); + let catalog = view.pointer_catalog().await.unwrap(); + let file_hash = crab_xet::hash::merkle_hex_from_bytes(&pointer.file_hash); + let shard = MerkleHash::from_hex(catalog.files()[&file_hash].shard_hash()).unwrap(); + crab_read::pointer_proof::verify_crab_pointer( + &layout, + &pointer, + shard, + proof_limits().content, + &cancel, + ) + .await + .unwrap(); + } +} + +async fn delete_source( + layout: &crab_storage::StoreLayout, + directory: &std::path::Path, + tip: gix_hash::ObjectId, + expected_borrow: &str, +) { + use crab_metadata::capsule_protocol::{CapsuleRefEdit, CapsuleTransaction}; + use crab_remote::prepare::CapsuleCommitOutcome; + let cancel = CancellationToken::new(); + let (prepared, view, runtime, leases) = + prepare_update(layout, directory, None, "refs/heads/main", None).await; + let edit = CapsuleRefEdit::new("refs/heads/main", Some(tip.to_string()), None, None); + // Reviewed plans provide stable transaction identities without a runtime + // nonce hook. Force deletion ahead of borrowing in canonical ID order. + let (plan, expected_delete) = (0..256) + .find_map(|i| { + let plan = blake3::hash(format!("delete-lifetime:{i}").as_bytes()) + .to_hex() + .to_string(); + let transaction = CapsuleTransaction::for_plan( + view.root_snapshot().record().digest(), + &plan, + vec![edit.clone()], + ) + .unwrap(); + let id = transaction.id().unwrap(); + (id.as_str() < expected_borrow).then_some((plan, id)) + }) + .unwrap(); + let artifacts = prepared + .upload_capsule(&view, proof_limits(), &cancel) + .await + .unwrap(); + let committed = artifacts + .commit(Some(&plan), std::time::Duration::from_secs(60), &cancel) + .await; + leases.release_after_commit().await; + runtime.shutdown().await; + let CapsuleCommitOutcome::Committed { transaction_id } = committed.unwrap() else { + panic!("source deletion must have a proven terminal outcome"); + }; + assert_eq!(transaction_id, expected_delete); +}