diff --git a/crab/src/cmd/gc/capsule_cleanup_tests.rs b/crab/src/cmd/gc/capsule_cleanup_tests.rs index 7d49556c4..a501550aa 100644 --- a/crab/src/cmd/gc/capsule_cleanup_tests.rs +++ b/crab/src/cmd/gc/capsule_cleanup_tests.rs @@ -226,7 +226,6 @@ async fn capsule_gc_does_not_report_revalidated_retained_objects_as_deleted() { .unwrap(); let mut manifest = crab_metadata::manifests::Manifest::default_for_repo("refs/heads/main"); manifest.seal_git_validation(); - let state_digest = "a".repeat(64); let outcome = sweep_capsule_objects( &GcArgs { force, @@ -236,7 +235,6 @@ async fn capsule_gc_does_not_report_revalidated_retained_objects_as_deleted() { &layout, &root, &[], - &state_digest, &manifest, &HashSet::new(), &CancellationToken::new(), diff --git a/crab/src/cmd/gc/captured_index_tests.rs b/crab/src/cmd/gc/captured_index_tests.rs new file mode 100644 index 000000000..f135038bc --- /dev/null +++ b/crab/src/cmd/gc/captured_index_tests.rs @@ -0,0 +1,417 @@ +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime}; + +use bytes::Bytes; +use crab_metadata::capsule_protocol::{ + Capsule, CapsuleGitPack, CapsuleRefEdit, CapsuleSection, CapsuleSectionKind, + CapsuleTransaction, CapsuleVisibilityDelta, +}; +use crab_metadata::git_object_locator::{GitPackInventoryEntry, PagedGitObjectIndex}; +use crab_metadata::git_visibility::GitVisibilityEdit; +use crab_read::capsule_protocol::{CapsuleReadLimits, CapsuleRepositoryView}; +use crab_remote_git::{RemoteGitRuntime, RepositoryIdentity, RepositoryOptions}; +use crab_storage::StoreLayout; +use crab_xet::hash::MerkleHash; +use futures_util::{TryStreamExt, stream::BoxStream}; +use object_store::{ + CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, + PutMultipartOptions, PutOptions, PutPayload, PutResult, memory::InMemory, path::Path, +}; +use tokio_util::sync::CancellationToken; + +use super::{GcArgs, sweep_capsule_objects}; +use crate::storage::store::Store; + +const LIMIT: u64 = 8 * 1024 * 1024; + +// Immutable bodies can predate their first reader by years. Alter only their +// provider timestamps; native pack validation and conditional publication stay real. +#[derive(Debug)] +struct AgedObjects { + inner: InMemory, + paths: Mutex>, + modified: SystemTime, + list_calls: Mutex>, +} + +impl std::fmt::Display for AgedObjects { + fn fmt(&self, output: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + output.write_str("aged index fixture") + } +} + +#[async_trait::async_trait] +impl ObjectStore for AgedObjects { + async fn put_opts( + &self, + path: &Path, + payload: PutPayload, + options: PutOptions, + ) -> object_store::Result { + self.inner.put_opts(path, payload, options).await + } + + async fn put_multipart_opts( + &self, + path: &Path, + options: PutMultipartOptions, + ) -> object_store::Result> { + self.inner.put_multipart_opts(path, options).await + } + + async fn get_opts(&self, path: &Path, options: GetOptions) -> object_store::Result { + let mut result = self.inner.get_opts(path, options).await?; + if self.paths.lock().unwrap().contains(path) { + result.meta.last_modified = self.modified.into(); + } + Ok(result) + } + + fn delete_stream( + &self, + paths: BoxStream<'static, object_store::Result>, + ) -> BoxStream<'static, object_store::Result> { + self.inner.delete_stream(paths) + } + + fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result> { + if let Some(prefix) = prefix { + self.list_calls.lock().unwrap().push(prefix.clone()); + } + let paths = self.paths.lock().unwrap().clone(); + let modified = self.modified; + Box::pin(self.inner.list(prefix).map_ok(move |mut meta| { + if paths.contains(&meta.location) { + meta.last_modified = modified.into(); + } + meta + })) + } + + async fn list_with_delimiter(&self, prefix: Option<&Path>) -> object_store::Result { + let mut result = self.inner.list_with_delimiter(prefix).await?; + let paths = self.paths.lock().unwrap(); + for meta in &mut result.objects { + if paths.contains(&meta.location) { + meta.last_modified = self.modified.into(); + } + } + Ok(result) + } + + async fn copy_opts( + &self, + from: &Path, + to: &Path, + options: CopyOptions, + ) -> object_store::Result<()> { + self.inner.copy_opts(from, to, options).await + } +} + +async fn publish_blob(layout: &StoreLayout, name: &str) -> String { + let payload = name.as_bytes(); + let oid = crab_remote::objects::object_id(gix_object::Kind::Blob, payload) + .unwrap() + .to_string(); + let mut bytes = Vec::new(); + crab_git::pack_writer::write_pack( + &mut bytes, + std::iter::once(Ok((gix_object::Kind::Blob, payload.len() as u64, payload))), + LIMIT, + || false, + ) + .unwrap(); + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("incoming.pack"); + std::fs::write(&path, &bytes).unwrap(); + let installed = crab_git::pack::install_pack_file_from_path( + &directory.path().join("indexed"), + &path, + blake3::hash(&bytes).to_hex().as_ref(), + LIMIT, + true, + ) + .unwrap(); + let checksum = gix_hash::ObjectId::from_hex(installed.git_sha1.as_bytes()).unwrap(); + let kinds = + crab_git::pack_locator::encode_pack_kind_metadata(checksum, &[gix_object::Kind::Blob]) + .unwrap(); + let pack = CapsuleGitPack::new( + bytes.into(), + std::fs::read(installed.idx_path).unwrap().into(), + std::fs::read(installed.rev_path).unwrap().into(), + Bytes::from(kinds), + installed.git_sha1, + 1, + ) + .unwrap(); + let root = crab_write::capsule_protocol::open_root(layout) + .await + .unwrap(); + let name = format!("refs/tags/{name}"); + let transaction = CapsuleTransaction::new( + root.record().digest(), + vec![CapsuleRefEdit::new(&name, None, Some(oid.clone()), None)], + ) + .unwrap(); + let visibility = CapsuleVisibilityDelta::new(BTreeMap::from([( + name, + GitVisibilityEdit::from_replacement_objects(None, oid.clone(), vec![oid.clone()]), + )])) + .unwrap(); + let capsule = Capsule::build( + &transaction, + vec![pack], + vec![CapsuleSection::new( + CapsuleSectionKind::VisibilityDelta, + visibility.encode().unwrap(), + )], + ) + .unwrap(); + crab_write::capsule_protocol::publish(layout, root, &transaction, &capsule) + .await + .unwrap(); + oid +} + +async fn view(layout: &StoreLayout) -> CapsuleRepositoryView { + let root = crab_write::capsule_protocol::open_root(layout) + .await + .unwrap(); + crab_read::capsule_protocol::open_git_view_from_root( + layout, + root, + CapsuleReadLimits { + max_capsule_bytes: LIMIT, + max_frontier_bytes: LIMIT, + }, + ) + .await + .unwrap() +} + +async fn cold_capture_after_gc(replace_readiness: bool) -> crab_metadata::error::Result { + let backend = Arc::new(AgedObjects { + inner: InMemory::new(), + paths: Mutex::new(HashSet::new()), + modified: SystemTime::now() - Duration::from_secs(7200), + list_calls: Mutex::new(Vec::new()), + }); + let store = Store::new(backend.clone()); + let layout = StoreLayout::new(store.as_storage().clone(), "org/captured-gc".to_owned()); + crab_write::capsule_protocol::initialize(&layout, &"1".repeat(64), "refs/heads/main") + .await + .unwrap(); + let oid = publish_blob(&layout, "original").await; + let runtime = Arc::new(RemoteGitRuntime::default()); + let identity = RepositoryIdentity::new("local", "captured-gc", 1).unwrap(); + let cancel = CancellationToken::new(); + let indexes = crab_remote::browse_indexes::ensure( + &layout, + &identity, + runtime.clone(), + RepositoryOptions::default(), + LIMIT, + &cancel, + ) + .await + .unwrap() + .unwrap(); + let captured = view(&layout).await; + let snapshot = captured.git_snapshot().unwrap(); + let reader = PagedGitObjectIndex::load( + layout.store(), + &layout, + indexes.object_index_hash().unwrap(), + &captured.state_digest(), + &snapshot.manifest, + LIMIT, + ) + .await + .unwrap(); + // Age only this immutable body. The capture, ref change, readiness + // replacement and GC all happen now, within a fresh reader's lifetime. + backend + .paths + .lock() + .unwrap() + .insert(reader.body_path().clone()); + let inventory = snapshot + .journal + .packs + .iter() + .map(|pack| { + let id = MerkleHash::from_hex(&pack.pack_id).unwrap(); + ( + id, + GitPackInventoryEntry { + pack_id: id, + object_count: pack.object_count, + pack_size: pack.size, + }, + ) + }) + .collect::>(); + publish_blob(&layout, "later").await; + if replace_readiness { + let replacement = crab_remote::browse_indexes::ensure( + &layout, + &identity, + runtime.clone(), + RepositoryOptions::default(), + LIMIT, + &cancel, + ) + .await + .unwrap() + .unwrap(); + assert_ne!(replacement.object_index_hash(), indexes.object_index_hash()); + } + let current = view(&layout).await; + let current_snapshot = current.git_snapshot().unwrap(); + sweep_capsule_objects( + &GcArgs::default(), + &store, + &layout, + current.root().root(), + current.capsule_run_pointers(), + ¤t_snapshot.manifest, + &HashSet::new(), + &cancel, + SystemTime::now(), + Duration::from_secs(3600), + Instant::now(), + ) + .await + .unwrap(); + let oid = gix_hash::ObjectId::from_hex(oid.as_bytes()).unwrap(); + let oid: [u8; 20] = oid.as_bytes().try_into().unwrap(); + let result = reader.lookup(layout.store(), &[oid], &inventory).await; + runtime.shutdown().await; + result.map(|entries| entries[0].is_some()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn gc_preserves_cold_captures_after_ref_movement() { + let result = cold_capture_after_gc(false).await; + assert!( + matches!(result, Ok(true)), + "cold capture lost after ref movement: {result:?}" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn gc_preserves_cold_captures_after_readiness_replacement() { + let result = cold_capture_after_gc(true).await; + assert!( + matches!(result, Ok(true)), + "cold capture lost after readiness replacement: {result:?}" + ); +} + +#[tokio::test] +async fn capsule_gc_lists_each_candidate_namespace_once() { + let backend = Arc::new(AgedObjects { + inner: InMemory::new(), + paths: Mutex::new(HashSet::new()), + modified: SystemTime::now(), + list_calls: Mutex::new(Vec::new()), + }); + let store = Store::new(backend.clone()); + let layout = StoreLayout::new(store.as_storage().clone(), "org/gc-listing".to_owned()); + let root = + crab_write::capsule_protocol::initialize(&layout, &"1".repeat(64), "refs/heads/main") + .await + .unwrap(); + let mut manifest = crab_metadata::manifests::Manifest::default_for_repo("refs/heads/main"); + manifest.seal_git_validation(); + backend.list_calls.lock().unwrap().clear(); + let outcome = sweep_capsule_objects( + &GcArgs::default(), + &store, + &layout, + root.record().root(), + &[], + &manifest, + &HashSet::new(), + &CancellationToken::new(), + SystemTime::now(), + Duration::from_secs(3600), + Instant::now(), + ) + .await + .unwrap(); + let mut actual = backend.list_calls.lock().unwrap().clone(); + actual.sort(); + let mut expected = [ + "v2/capsules/", + "v2/checkpoints/", + "v2/pack-layers/", + "v2/history/", + "metadata/", + "v2/browse-roots/", + "manifests/", + ] + .map(|suffix| layout.repo_path(suffix)) + .to_vec(); + expected.sort(); + assert_eq!( + (actual, outcome.list_requests, outcome.list_parallelism), + (expected, 7, 7) + ); +} + +#[tokio::test] +async fn capsule_gc_preserves_metadata_outside_owned_index_directories() { + let backend = Arc::new(AgedObjects { + inner: InMemory::new(), + paths: Mutex::new(HashSet::new()), + modified: SystemTime::now() - Duration::from_secs(7200), + list_calls: Mutex::new(Vec::new()), + }); + let store = Store::new(backend.clone()); + let layout = StoreLayout::new(store.as_storage().clone(), "org/gc-metadata".to_owned()); + let root = + crab_write::capsule_protocol::initialize(&layout, &"1".repeat(64), "refs/heads/main") + .await + .unwrap(); + let unrelated = [ + "unrelated/keep", + "path-state-backups/keep", + "commit-graph-backups/keep", + "git-object-index-backups/keep", + ] + .map(|suffix| layout.repo_path(&format!("metadata/{suffix}"))); + for path in &unrelated { + store + .put(path, Bytes::from_static(b"unowned metadata")) + .await + .unwrap(); + backend.paths.lock().unwrap().insert(path.clone()); + } + let mut manifest = crab_metadata::manifests::Manifest::default_for_repo("refs/heads/main"); + manifest.seal_git_validation(); + sweep_capsule_objects( + &GcArgs::default(), + &store, + &layout, + root.record().root(), + &[], + &manifest, + &HashSet::new(), + &CancellationToken::new(), + SystemTime::now(), + Duration::from_secs(3600), + Instant::now(), + ) + .await + .unwrap(); + let retained = + futures_util::future::try_join_all(unrelated.iter().map(|path| store.head(path))).await; + assert!( + retained.is_ok(), + "GC removed metadata outside its owned directories: {retained:?}" + ); +} diff --git a/crab/src/cmd/gc/mod.rs b/crab/src/cmd/gc/mod.rs index 00304d8d4..1f4a279f8 100644 --- a/crab/src/cmd/gc/mod.rs +++ b/crab/src/cmd/gc/mod.rs @@ -22,6 +22,8 @@ pub mod parallel_enum; #[cfg(test)] mod capsule_cleanup_tests; +#[cfg(test)] +mod captured_index_tests; use std::collections::HashSet; #[cfg(test)] @@ -3134,14 +3136,12 @@ async fn run_capsule_gc( ) .await?; let snapshot = view.git_snapshot()?; - let state_digest = view.state_digest().to_owned(); sweep_capsule_objects( args, store, &layout, fenced.record().root(), view.capsule_run_pointers(), - &state_digest, &snapshot.manifest, coordinator_protected_keys, cancel, @@ -3183,7 +3183,6 @@ async fn sweep_capsule_objects( layout: &crab_storage::StoreLayout, root: &crab_metadata::capsule_protocol::RepositoryRoot, capsule_runs: &[crab_metadata::capsule_protocol::CapsulePointer], - state_digest: &str, manifest: &crab_metadata::manifests::Manifest, coordinator_protected_keys: &HashSet, cancel: &CancellationToken, @@ -3235,22 +3234,16 @@ async fn sweep_capsule_objects( } } reachable.extend(coordinator_protected_keys.iter().cloned()); - mark_derived_index_objects( - store.as_storage(), - layout, - state_digest, - manifest, - &mut reachable, - ) - .await?; + mark_derived_index_objects(store.as_storage(), layout, manifest, &mut reachable).await?; let capsule_prefix = layout.repo_path("v2/capsules/"); let checkpoint_prefix = layout.repo_path("v2/checkpoints/"); let pack_layer_prefix = layout.repo_path("v2/pack-layers/"); let history_prefix = layout.repo_path("v2/history/"); - let path_state_prefix = layout.repo_path("metadata/path-state/"); - let commit_graph_prefix = layout.repo_path("metadata/commit-graph/"); - let object_index_prefix = layout.repo_path("metadata/git-object-index/"); + let metadata_prefix = layout.repo_path("metadata/"); + let index_prefixes = ["path-state", "commit-graph", "git-object-index"] + .map(|kind| format!("{metadata_prefix}/{kind}/")); + let browse_root_prefix = layout.repo_path("v2/browse-roots/"); let manifests_prefix = layout.repo_path("manifests/"); let path_state_descriptor_prefix = layout.repo_path("manifests/path-state-"); let commit_graph_descriptor_prefix = layout.repo_path("manifests/commit-graph-"); @@ -3260,24 +3253,28 @@ async fn sweep_capsule_objects( checkpoints, pack_layers, history, - path_state_objects, - commit_graph_objects, - object_index_objects, + metadata_objects, + browse_root_objects, manifests, ) = tokio::try_join!( store.list_prefix(&capsule_prefix), store.list_prefix(&checkpoint_prefix), store.list_prefix(&pack_layer_prefix), store.list_prefix(&history_prefix), - store.list_prefix(&path_state_prefix), - store.list_prefix(&commit_graph_prefix), - store.list_prefix(&object_index_prefix), + store.list_prefix(&metadata_prefix), + store.list_prefix(&browse_root_prefix), store.list_prefix(&manifests_prefix), )?; - let derived_index_objects = path_state_objects + // One metadata scan covers derived index bodies. Lifetime roots belong to + // protocol control; unrelated metadata never becomes a deletion candidate. + let derived_index_objects = metadata_objects .into_iter() - .chain(commit_graph_objects) - .chain(object_index_objects) + .filter(|object| { + index_prefixes + .iter() + .any(|prefix| object.location.as_ref().starts_with(prefix)) + }) + .chain(browse_root_objects) .chain(manifests.into_iter().filter(|object| { let key = object.location.as_ref(); key.starts_with(path_state_descriptor_prefix.as_ref()) @@ -3384,13 +3381,30 @@ async fn sweep_capsule_objects( async fn mark_derived_index_objects( store: &crab_storage::Store, layout: &crab_storage::StoreLayout, - state_digest: &str, manifest: &crab_metadata::manifests::Manifest, reachable: &mut HashSet, ) -> Result<()> { - if let Some(indexes) = crab_metadata::capsule_protocol::load_browse_indexes(layout).await? - && indexes.state_digest() == state_digest + let current = crab_metadata::capsule_protocol::load_browse_indexes(layout).await?; + let roots = crab_metadata::capsule_protocol::load_retained_browse_indexes(layout).await?; + if current + .as_ref() + .is_some_and(|current| !roots.iter().any(|root| root.indexes() == current)) { + return Err(CrabError::CorruptObject { + path: layout.capsule_browse_indexes_path().to_string(), + reason: "published browse indexes have no retained capture; rebuild browse indexes before GC".to_owned(), + }); + } + // Captured readers own original source ranges as well as cold index + // pages. Repacking cannot substitute a new source behind those ranges. + for root in roots { + reachable.insert(layout.browse_retained_root_path(root.hash()).to_string()); + reachable.extend( + root.source_paths(layout) + .into_iter() + .map(|path| path.to_string()), + ); + let indexes = root.indexes(); let graph_hash = indexes.commit_graph_hash(); let graph_path = layout.bulk_manifest_path("commit-graph", graph_hash); let graph = crab_metadata::split_commit_graph::load_split_commit_graph_descriptor( @@ -3409,23 +3423,15 @@ async fn mark_derived_index_objects( ); if let Some(hash) = indexes.object_index_hash() { - let index = crab_metadata::git_object_locator::PagedGitObjectIndex::load( + let paths = crab_metadata::git_object_locator::PagedGitObjectIndex::retained_paths( store, layout, hash, - state_digest, - manifest, + indexes.state_digest(), crab_metadata::split_commit_graph::DEFAULT_MAX_SPLIT_COMMIT_GRAPH_BYTES, ) .await?; - // Every authenticated physical page lives in this immutable body. - // Retain it with readiness so pruning cannot strand future delta reads. - reachable.insert( - layout - .bulk_manifest_path("git-object-index", hash) - .to_string(), - ); - reachable.insert(index.body_path().to_string()); + reachable.extend(paths.into_iter().map(|path| path.to_string())); } if let Some(path_state_hash) = indexes.path_state_hash() { @@ -5517,7 +5523,6 @@ mod tests { .await .unwrap(); let snapshot = current_view.git_snapshot().unwrap(); - let state_digest = current_view.state_digest().to_owned(); assert_eq!( root.record().root().compacted_ref_transactions(), captured.visible_ref_transactions() @@ -5558,7 +5563,6 @@ mod tests { &layout, root.record().root(), &[], - &state_digest, &snapshot.manifest, &protected, &CancellationToken::new(), @@ -5586,7 +5590,6 @@ mod tests { &layout, root.record().root(), &[], - &state_digest, &snapshot.manifest, &protected, &CancellationToken::new(), @@ -5811,6 +5814,15 @@ mod tests { .unwrap() .with_object_index(physical_hash.clone()) .unwrap(); + crab_metadata::capsule_protocol::retain_browse_indexes( + &layout, + &indexes, + manifest.generation, + None, + &[], + ) + .await + .unwrap(); let index_path = layout.capsule_browse_indexes_path(); let bytes = indexes.encode().unwrap(); index_etag = Some(match index_etag { @@ -5821,15 +5833,9 @@ mod tests { .unwrap(), }); let mut retained = HashSet::new(); - mark_derived_index_objects( - store.as_storage(), - &layout, - &state_digest, - &manifest, - &mut retained, - ) - .await - .unwrap(); + mark_derived_index_objects(store.as_storage(), &layout, &manifest, &mut retained) + .await + .unwrap(); assert!( retained.contains( &layout @@ -5957,7 +5963,6 @@ mod tests { &layout, root.record().root(), &[], - &state_digest, &manifest, &HashSet::new(), &CancellationToken::new(), @@ -5975,7 +5980,6 @@ mod tests { &layout, root.record().root(), &[], - &state_digest, &manifest, &HashSet::new(), &CancellationToken::new(), diff --git a/crab/src/cmd/history_recovery_v2.rs b/crab/src/cmd/history_recovery_v2.rs index f46a75818..ac82e116e 100644 --- a/crab/src/cmd/history_recovery_v2.rs +++ b/crab/src/cmd/history_recovery_v2.rs @@ -459,7 +459,18 @@ async fn prune_history( let base = crab_write::capsule_protocol::open_root(layout).await?; let chain = history_chain(layout, &base).await?; let mut payload = prune_payload(&chain, args.keep_last, true); + let minimum = chain + .iter() + .take(args.keep_last) + .next_back() + .map(|segment| segment.checkpoint().covered_generation()); if payload.roots_pruned == 0 { + // Retry capture cleanup after an earlier history CAS committed but + // its derived-retention update failed. Empty history has no cutoff. + if let Some(minimum) = minimum { + crab_metadata::capsule_protocol::prune_retained_browse_indexes(layout, minimum) + .await?; + } return Ok(payload); } // Finish fallible local preparation before acquiring the root fence; @@ -484,7 +495,18 @@ async fn prune_history( let replacement = crab_write::capsule_protocol::replace_history(layout, fenced.clone(), &rebuilt).await; let (primary, release_base) = match replacement { - Ok(root) => (None, root), + Ok(root) => { + let cleanup = match minimum { + Some(minimum) => { + crab_metadata::capsule_protocol::prune_retained_browse_indexes( + layout, minimum, + ) + .await + } + None => Ok(()), + }; + (cleanup.err().map(CrabError::from), root) + } Err(error) => (Some(CrabError::from(error)), fenced), }; let release = crab_write::capsule_protocol::end_gc(layout, release_base, &fence_id).await; diff --git a/crates/crab-metadata/README.md b/crates/crab-metadata/README.md index 676c023e5..961517e1c 100644 --- a/crates/crab-metadata/README.md +++ b/crates/crab-metadata/README.md @@ -118,7 +118,28 @@ requires complete verified pack coverage before publication; path completion retains that binding. Graph publication does not wait for path construction. The prior versions were never reachable from a release tag; maintenance rebuilds this derived record in place, without a legacy reader. Readers ignore stale -records; absent indexes remain a readiness state. Rebuilding an immutable index +records; absent indexes remain a readiness state. GC retains dependencies of the currently +published readiness record even when newer refs make it unusable for new reads. +Storage retention does not grant inventory membership or ref visibility. +The separate `v2/browse-retention` CAS head owns a bounded chain of authenticated +`v2/browse-roots/` captures. Maintenance records original checkpoint, Git source and all pending capsule-run +identities before advertising replacement readiness. Ref-only visibility/catalog +runs remain dependencies even when they contain no Git pack members. GC follows every +retained capture's graph, path-state, physical pages and original source bodies; +foreground reads still load only the current bounded readiness record. An +unchanged capture does not append a node. Missing/corrupt lifetime dependencies +stop GC before deletion. A valid readiness record without lifetime ownership +requires background browse maintenance before GC; there is no scan fallback. + +Explicit history pruning trims superseded captures below the oldest retained +checkpoint generation only after the authoritative history CAS succeeds. Current +readiness and the newest lifetime capture remain rooted. A failed cleanup leaves +extra retention and is retried even if the history chain already has the requested +length. Pruning is the boundary that can invalidate captures from discarded +generations; ordinary rebuilding, checkpointing and GC do not expire readers. +The new lifetime format is version 1; no release tag contains capsule browse +readiness, and its existing version 3 encoding remains unchanged. +Rebuilding an immutable index verifies the generated content identity and repairs corrupt stored bytes only with a conditional update against the observed version, followed by readback. This applies to graph/path-state/physical-index descriptors and bodies, not Git/Xet data. diff --git a/crates/crab-metadata/src/capsule_protocol/browse_retention.rs b/crates/crab-metadata/src/capsule_protocol/browse_retention.rs new file mode 100644 index 000000000..89a06c4b9 --- /dev/null +++ b/crates/crab-metadata/src/capsule_protocol/browse_retention.rs @@ -0,0 +1,339 @@ +use std::collections::{BTreeMap, BTreeSet}; + +use bytes::Bytes; +use crab_storage::{ETag, Store, StoreLayout}; +use serde::{Deserialize, Serialize}; + +use super::{ + BrowseIndexes, CapsulePointer, LayeredCheckpoint, MAX_BROWSE_INDEXES_BYTES, + MAX_HISTORY_CHAIN_BYTES, MAX_HISTORY_CHAIN_SEGMENTS, MAX_HISTORY_SEGMENT_BYTES, PackSourceKind, +}; +use crate::error::{MetadataError, Result}; +use crate::validation::{corrupt_object, validate_content_hash}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Head { + version: u32, + hash: String, + roots: usize, + bytes: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Payload { + version: u32, + generation: u64, + indexes: BrowseIndexes, + previous: Option, + checkpoint: Option, + sources: BTreeMap, +} + +/// Immutable capture of browse pages and the original Git sources used by their readers. +#[derive(Debug)] +pub struct RetainedBrowseIndexes { + hash: String, + payload: Payload, +} + +impl RetainedBrowseIndexes { + /// Return this immutable retention record's identity. + #[must_use] + pub fn hash(&self) -> &str { + &self.hash + } + + /// Return the source generation used by explicit history pruning. + #[must_use] + pub fn generation(&self) -> u64 { + self.payload.generation + } + + /// Return captured index dependencies without granting current visibility. + #[must_use] + pub fn indexes(&self) -> &BrowseIndexes { + &self.payload.indexes + } + + /// Return original source bodies, including a captured checkpoint when present. + #[must_use] + pub fn source_paths(&self, layout: &StoreLayout) -> Vec { + self.payload + .checkpoint + .iter() + .map(|hash| layout.capsule_checkpoint_path(hash)) + .chain(self.payload.sources.iter().map(|(hash, kind)| match kind { + PackSourceKind::CapsuleRun => layout.capsule_path(hash), + PackSourceKind::PackLayer => layout.capsule_pack_layer_path(hash), + })) + .collect() + } +} + +impl Payload { + fn encode(&self) -> Result { + self.validate()?; + let bytes = serde_json::to_vec(self) + .map_err(|source| MetadataError::BrowseIndexRecord { source })?; + if bytes.len() as u64 > MAX_HISTORY_SEGMENT_BYTES { + return Err(corrupt("retained browse record exceeds byte limit")); + } + Ok(bytes.into()) + } + + fn validate(&self) -> Result<()> { + if self.version != 1 { + return Err(corrupt("unsupported retained browse version")); + } + self.indexes.encode()?; + for hash in self + .previous + .iter() + .chain(self.checkpoint.iter()) + .chain(self.sources.keys()) + { + validate_content_hash(hash, "retention dependency", "browse retention")?; + } + Ok(()) + } +} + +impl Head { + fn append(previous: Option<&Self>, bytes: &Bytes) -> Result { + let value = Self { + version: 1, + hash: blake3::hash(bytes).to_hex().to_string(), + roots: previous.map_or(0, |head| head.roots).saturating_add(1), + bytes: previous + .map_or(0, |head| head.bytes) + .saturating_add(bytes.len() as u64), + }; + value.validate()?; + Ok(value) + } + + fn validate(&self) -> Result<()> { + validate_content_hash(&self.hash, "retained root", "browse retention")?; + if self.version != 1 + || self.roots == 0 + || self.roots > MAX_HISTORY_CHAIN_SEGMENTS + || self.bytes == 0 + || self.bytes > MAX_HISTORY_CHAIN_BYTES + { + return Err(corrupt( + "retained browse head exceeds version or resource bounds", + )); + } + Ok(()) + } +} + +async fn load_head(layout: &StoreLayout) -> Result> { + let (bytes, etag) = match layout + .store() + .get_with_etag_bounded( + &layout.browse_retained_head_path(), + MAX_BROWSE_INDEXES_BYTES, + ) + .await + { + Ok(value) => value, + Err(crab_storage::StorageError::NotFound { .. }) => return Ok(None), + Err(error) => return Err(error.into()), + }; + let head: Head = serde_json::from_slice(&bytes) + .map_err(|source| MetadataError::BrowseIndexRecord { source })?; + head.validate()?; + Ok(Some((head, etag))) +} + +async fn load_payload( + layout: &StoreLayout, + hash: &str, + maximum: u64, +) -> Result<(Payload, u64)> { + validate_content_hash(hash, "retained root", "browse retention")?; + let expected = *blake3::Hash::from_hex(hash) + .map_err(|_| corrupt("invalid retained browse hash"))? + .as_bytes(); + let bytes = layout + .store() + .verify_bounded( + &layout.browse_retained_root_path(hash), + &expected, + maximum.min(MAX_HISTORY_SEGMENT_BYTES), + ) + .await?; + let payload: Payload = serde_json::from_slice(&bytes) + .map_err(|source| MetadataError::BrowseIndexRecord { source })?; + payload.validate()?; + Ok((payload, bytes.len() as u64)) +} + +async fn publish_head(layout: &StoreLayout, head: &Head, etag: Option) -> Result<()> { + let bytes = + serde_json::to_vec(head).map_err(|source| MetadataError::BrowseIndexRecord { source })?; + let path = layout.browse_retained_head_path(); + match etag { + Some(etag) => layout.store().update(&path, bytes.into(), etag).await?, + None => { + layout + .store() + .create_strict_with_etag(&path, bytes.into()) + .await? + } + }; + Ok(()) +} + +/// Retain original dependencies before publishing replacement browse readiness. +/// +/// The caller holds generation ownership and both GC writer fences. Lifetime +/// CAS precedes readiness CAS: a failed readiness publication may retain extra +/// data, but a crash can never advertise unretained pages or source ranges. +pub async fn retain_browse_indexes( + layout: &StoreLayout, + indexes: &BrowseIndexes, + generation: u64, + checkpoint: Option<&LayeredCheckpoint>, + run_pointers: &[CapsulePointer], +) -> Result<()> { + let previous = load_head(layout).await?; + let mut sources = BTreeMap::new(); + let checkpoint_sources = checkpoint + .into_iter() + .flat_map(LayeredCheckpoint::sources) + .map(|source| (source.object_hash().to_owned(), source.kind())); + let capsule_sources = run_pointers + .iter() + .map(|pointer| (pointer.hash().to_owned(), PackSourceKind::CapsuleRun)); + // Ref-only runs carry visibility/catalog pages despite having no Git + // members. Captures own those lazy reads as well as physical pack ranges. + for (hash, kind) in checkpoint_sources.chain(capsule_sources) { + if sources.insert(hash, kind).is_some_and(|old| old != kind) { + return Err(corrupt("retained source has conflicting kinds")); + } + } + let mut payload = Payload { + version: 1, + generation, + indexes: indexes.clone(), + previous: previous.as_ref().map(|(head, _)| head.hash.clone()), + checkpoint: checkpoint.map(|value| value.hash().to_owned()), + sources, + }; + if let Some((head, _)) = &previous { + let (old, _) = load_payload(layout, &head.hash, head.bytes).await?; + if old.generation > generation { + return Err(corrupt("retained browse generation moved backwards")); + } + payload.previous = old.previous.clone(); + if payload == old { + return Ok(()); + } + payload.previous = Some(head.hash.clone()); + } + let bytes = payload.encode()?; + let head = Head::append(previous.as_ref().map(|(head, _)| head), &bytes)?; + layout + .store() + .put_if_absent_verified(&layout.browse_retained_root_path(&head.hash), bytes) + .await?; + publish_head(layout, &head, previous.map(|(_, etag)| etag)).await +} + +async fn load_chain( + layout: &StoreLayout, + head: &Head, +) -> Result> { + let mut next = Some(head.hash.clone()); + let mut seen = BTreeSet::new(); + let mut records: Vec = Vec::new(); + let mut remaining = head.bytes; + while let Some(hash) = next { + if records.len() == head.roots || !seen.insert(hash.clone()) { + return Err(corrupt("retained browse count or cycle mismatch")); + } + let (payload, bytes) = load_payload(layout, &hash, remaining).await?; + remaining = remaining + .checked_sub(bytes) + .ok_or_else(|| corrupt("retained browse chain exceeds byte limit"))?; + if records + .last() + .is_some_and(|newer| newer.generation() < payload.generation) + { + return Err(corrupt("retained browse generations are not ordered")); + } + next = payload.previous.clone(); + records.push(RetainedBrowseIndexes { hash, payload }); + } + if remaining != 0 || records.len() != head.roots { + return Err(corrupt("retained browse accounting mismatch")); + } + Ok(records) +} + +/// Authenticate retained captures for maintenance; foreground lookup never walks this chain. +pub async fn load_retained_browse_indexes( + layout: &StoreLayout, +) -> Result> { + match load_head(layout).await? { + Some((head, _)) => load_chain(layout, &head).await, + None => Ok(Vec::new()), + } +} + +/// Trim captured generations after authoritative history pruning commits. +/// +/// The caller holds the exclusive repository GC lease. Failure leaves extra +/// retention and is retryable; trimming before history CAS could strand readers. +pub async fn prune_retained_browse_indexes( + layout: &StoreLayout, + minimum_generation: u64, +) -> Result<()> { + let Some((head, etag)) = load_head(layout).await? else { + return Ok(()); + }; + let current = super::load_browse_indexes(layout).await?; + let records = load_chain(layout, &head).await?; + let keep = |position: usize, record: &RetainedBrowseIndexes| { + position == 0 + || record.generation() >= minimum_generation + || current.as_ref() == Some(record.indexes()) + }; + if records + .iter() + .enumerate() + .all(|(position, record)| keep(position, record)) + { + return Ok(()); + } + let mut previous: Option = None; + for (position, record) in records.into_iter().enumerate().rev() { + if !keep(position, &record) { + continue; + } + let payload = Payload { + previous: previous.as_ref().map(|head| head.hash.clone()), + ..record.payload + }; + let bytes = payload.encode()?; + let head = Head::append(previous.as_ref(), &bytes)?; + layout + .store() + .put_if_absent_verified(&layout.browse_retained_root_path(&head.hash), bytes) + .await?; + previous = Some(head); + } + let head = previous.ok_or_else(|| corrupt("pruning removed current browse retention"))?; + publish_head(layout, &head, Some(etag)).await +} + +fn corrupt(reason: &str) -> MetadataError { + corrupt_object("browse retention", reason) +} + +#[cfg(test)] +mod tests; diff --git a/crates/crab-metadata/src/capsule_protocol/browse_retention/tests.rs b/crates/crab-metadata/src/capsule_protocol/browse_retention/tests.rs new file mode 100644 index 000000000..61412516f --- /dev/null +++ b/crates/crab-metadata/src/capsule_protocol/browse_retention/tests.rs @@ -0,0 +1,234 @@ +use std::sync::Arc; + +use object_store::memory::InMemory; + +use super::*; + +fn fixture() -> StoreLayout { + StoreLayout::new( + Store::new(Arc::new(InMemory::new())), + "browse-retention".into(), + ) +} + +fn indexes(byte: char) -> BrowseIndexes { + BrowseIndexes::new(byte.to_string().repeat(64), "b".repeat(64), None).unwrap() +} + +#[tokio::test] +async fn unchanged_capture_does_not_grow_retention_or_rewrite_its_head() { + let layout = fixture(); + retain_browse_indexes(&layout, &indexes('a'), 1, None, &[]) + .await + .unwrap(); + let before = layout + .store() + .get_with_etag_bounded(&layout.browse_retained_head_path(), 4096) + .await + .unwrap(); + retain_browse_indexes(&layout, &indexes('a'), 1, None, &[]) + .await + .unwrap(); + let after = layout + .store() + .get_with_etag_bounded(&layout.browse_retained_head_path(), 4096) + .await + .unwrap(); + assert_eq!(before, after); +} + +#[tokio::test] +async fn replacement_retains_prior_coverage_even_without_query_readiness() { + let layout = fixture(); + retain_browse_indexes(&layout, &indexes('a'), 1, None, &[]) + .await + .unwrap(); + retain_browse_indexes(&layout, &indexes('c'), 2, None, &[]) + .await + .unwrap(); + let roots = load_retained_browse_indexes(&layout).await.unwrap(); + assert_eq!( + roots + .iter() + .map(|root| root.indexes().state_digest()) + .collect::>(), + vec!["c".repeat(64), "a".repeat(64)] + ); +} + +#[tokio::test] +async fn explicit_pruning_keeps_current_readiness_and_relinks_retained_generations() { + let layout = fixture(); + for (generation, byte) in [(1, 'a'), (2, 'c'), (3, 'd'), (4, 'e')] { + retain_browse_indexes(&layout, &indexes(byte), generation, None, &[]) + .await + .unwrap(); + } + // A lifetime publication may precede a failed readiness CAS. Preserve the + // older advertised root even when it is below the explicit pruning floor. + layout + .store() + .create_strict_with_etag( + &layout.capsule_browse_indexes_path(), + indexes('a').encode().unwrap(), + ) + .await + .unwrap(); + prune_retained_browse_indexes(&layout, 3).await.unwrap(); + let roots = load_retained_browse_indexes(&layout).await.unwrap(); + assert_eq!( + roots + .iter() + .map(RetainedBrowseIndexes::generation) + .collect::>(), + vec![4, 3, 1] + ); +} + +#[tokio::test] +async fn corrupt_or_missing_retained_roots_fail_closed() { + for missing in [false, true] { + let layout = fixture(); + retain_browse_indexes(&layout, &indexes('a'), 1, None, &[]) + .await + .unwrap(); + let roots = load_retained_browse_indexes(&layout).await.unwrap(); + let path = layout.browse_retained_root_path(roots[0].hash()); + if missing { + layout.store().delete(&path).await.unwrap(); + } else { + layout + .store() + .put_overwrite(&path, Bytes::from_static(b"corrupt")) + .await + .unwrap(); + } + assert!(load_retained_browse_indexes(&layout).await.is_err()); + } +} + +#[tokio::test] +async fn backwards_publication_preserves_the_lifetime_head() { + let layout = fixture(); + retain_browse_indexes(&layout, &indexes('a'), 2, None, &[]) + .await + .unwrap(); + let before = layout + .store() + .get_with_etag_bounded(&layout.browse_retained_head_path(), 4096) + .await + .unwrap(); + assert!( + retain_browse_indexes(&layout, &indexes('c'), 1, None, &[]) + .await + .is_err() + ); + assert_eq!( + layout + .store() + .get_with_etag_bounded(&layout.browse_retained_head_path(), 4096) + .await + .unwrap(), + before + ); +} + +#[test] +fn retention_head_bounds_are_admitted_before_publishing() { + for (roots, bytes) in [ + (MAX_HISTORY_CHAIN_SEGMENTS, 1), + (1, MAX_HISTORY_CHAIN_BYTES), + ] { + let old = Head { + version: 1, + hash: "a".repeat(64), + roots, + bytes, + }; + assert!(Head::append(Some(&old), &Bytes::from_static(b"new root")).is_err()); + } +} + +#[tokio::test] +async fn an_expired_publisher_cannot_replace_a_successor_lifetime_head() { + let layout = fixture(); + retain_browse_indexes(&layout, &indexes('a'), 1, None, &[]) + .await + .unwrap(); + let (old, etag) = load_head(&layout).await.unwrap().unwrap(); + retain_browse_indexes(&layout, &indexes('c'), 2, None, &[]) + .await + .unwrap(); + let result = publish_head(&layout, &old, Some(etag)).await; + assert!(matches!( + result, + Err(MetadataError::Storage { + source: crab_storage::StorageError::StateConflict { .. } + }) + )); +} + +#[tokio::test] +async fn captures_keep_ref_only_visibility_sources_independently_of_current_packaging() { + use crate::capsule_protocol::{Capsule, CapsuleRefEdit, CapsuleRun, CapsuleTransaction}; + let layout = fixture(); + let transaction = CapsuleTransaction::new( + &"a".repeat(64), + vec![CapsuleRefEdit::new( + "refs/heads/main", + None, + Some("1".repeat(40)), + None, + )], + ) + .unwrap(); + let capsule = Capsule::build(&transaction, Vec::new(), Vec::new()).unwrap(); + let run = CapsuleRun::leaf(capsule).unwrap(); + let source = 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(); + retain_browse_indexes(&layout, &indexes('a'), 1, None, &[source]) + .await + .unwrap(); + retain_browse_indexes(&layout, &indexes('c'), 2, None, &[]) + .await + .unwrap(); + let roots = load_retained_browse_indexes(&layout).await.unwrap(); + assert_eq!( + roots[1].source_paths(&layout), + vec![layout.capsule_path(run.hash())] + ); +} + +#[tokio::test] +async fn retrying_completed_pruning_does_not_rewrite_the_lifetime_head() { + let layout = fixture(); + for (generation, byte) in [(1, 'a'), (2, 'c'), (3, 'd')] { + retain_browse_indexes(&layout, &indexes(byte), generation, None, &[]) + .await + .unwrap(); + } + prune_retained_browse_indexes(&layout, 2).await.unwrap(); + let before = layout + .store() + .get_with_etag_bounded(&layout.browse_retained_head_path(), 4096) + .await + .unwrap(); + prune_retained_browse_indexes(&layout, 2).await.unwrap(); + assert_eq!( + layout + .store() + .get_with_etag_bounded(&layout.browse_retained_head_path(), 4096) + .await + .unwrap(), + before + ); +} diff --git a/crates/crab-metadata/src/capsule_protocol/mod.rs b/crates/crab-metadata/src/capsule_protocol/mod.rs index 0da15ce80..0f9bda89d 100644 --- a/crates/crab-metadata/src/capsule_protocol/mod.rs +++ b/crates/crab-metadata/src/capsule_protocol/mod.rs @@ -3,6 +3,8 @@ use bstr::ByteSlice; mod browse; +#[cfg(feature = "storage")] +mod browse_retention; mod capsule; #[cfg(feature = "storage")] mod captured_visibility; @@ -27,6 +29,11 @@ mod visibility; #[cfg(feature = "storage")] pub use browse::load_browse_indexes; pub use browse::{BrowseIndexes, MAX_BROWSE_INDEXES_BYTES}; +#[cfg(feature = "storage")] +pub use browse_retention::{ + RetainedBrowseIndexes, load_retained_browse_indexes, prune_retained_browse_indexes, + retain_browse_indexes, +}; pub use capsule::{ Capsule, CapsuleGitPack, CapsuleGitPackDescriptor, CapsuleSection, CapsuleSectionKind, CapsuleSectionLocation, diff --git a/crates/crab-metadata/src/git_object_locator/paged/reader.rs b/crates/crab-metadata/src/git_object_locator/paged/reader.rs index c1f39af8d..f26164933 100644 --- a/crates/crab-metadata/src/git_object_locator/paged/reader.rs +++ b/crates/crab-metadata/src/git_object_locator/paged/reader.rs @@ -12,18 +12,7 @@ impl PagedGitObjectIndex { manifest: &Manifest, maximum: u64, ) -> Result { - validate_content_hash(hash, "object index descriptor", "Git object index")?; - let expected = hash_bytes(hash)?; - let bytes = store - .verify_bounded( - &layout.bulk_manifest_path("git-object-index", hash), - &expected, - maximum.min(MAX_DESCRIPTOR_BYTES), - ) - .await?; - let descriptor: Descriptor = serde_json::from_slice(&bytes) - .map_err(|source| crate::error::MetadataError::GitObjectIndexRecord { source })?; - descriptor.validate()?; + let (descriptor, descriptor_bytes) = Self::descriptor(store, layout, hash, maximum).await?; if descriptor.state_digest != state_digest || descriptor.generation != manifest.generation || descriptor.pack_index_hash != manifest.pack_index_hash @@ -34,7 +23,7 @@ impl PagedGitObjectIndex { let path = body_path(layout, &descriptor.body_hash); Ok(Self { descriptor, - descriptor_bytes: bytes.len() as u64, + descriptor_bytes, path, origin: Arc::clone(layout.store().inner()), global_prefix: layout.global_prefix().to_owned(), @@ -43,6 +32,48 @@ impl PagedGitObjectIndex { }) } + /// Resolve retained readiness to its authenticated descriptor and complete page body. + /// + /// The state digest must come from the retained root. This proves storage + /// dependencies for GC, not current inventory membership or ref visibility. + pub async fn retained_paths( + store: &Store, + layout: &StoreLayout, + hash: &str, + state_digest: &str, + maximum: u64, + ) -> Result<[object_store::path::Path; 2]> { + let (descriptor, _) = Self::descriptor(store, layout, hash, maximum).await?; + if descriptor.state_digest != state_digest { + return bad("retained object index does not match captured state"); + } + Ok([ + layout.bulk_manifest_path("git-object-index", hash), + body_path(layout, &descriptor.body_hash), + ]) + } + + async fn descriptor( + store: &Store, + layout: &StoreLayout, + hash: &str, + maximum: u64, + ) -> Result<(Descriptor, u64)> { + validate_content_hash(hash, "object index descriptor", "Git object index")?; + let expected = hash_bytes(hash)?; + let bytes = store + .verify_bounded( + &layout.bulk_manifest_path("git-object-index", hash), + &expected, + maximum.min(MAX_DESCRIPTOR_BYTES), + ) + .await?; + let descriptor: Descriptor = serde_json::from_slice(&bytes) + .map_err(|source| crate::error::MetadataError::GitObjectIndexRecord { source })?; + descriptor.validate()?; + Ok((descriptor, bytes.len() as u64)) + } + /// Check captured state and in-process origin before attaching lookup to a repository. #[must_use] pub fn matches_binding( diff --git a/crates/crab-metadata/src/git_object_locator/paged/tests.rs b/crates/crab-metadata/src/git_object_locator/paged/tests.rs index bef5084ef..a7bd74db0 100644 --- a/crates/crab-metadata/src/git_object_locator/paged/tests.rs +++ b/crates/crab-metadata/src/git_object_locator/paged/tests.rs @@ -246,6 +246,52 @@ async fn descriptors_cannot_cross_captured_git_states() { ); } +#[tokio::test] +async fn retained_paths_cover_the_complete_body_without_loading_pages() { + let (manifest, inventory, entries) = inputs(10_000); + let state = "c".repeat(64); + let write = PagedGitObjectIndexWrite::build(&state, &manifest, &inventory, &entries).unwrap(); + let reads = Arc::new(AtomicU64::new(0)); + let observed = reads.clone(); + let store = Store::new(Arc::new(InMemory::new())).with_read_byte_observer(Arc::new(move |n| { + observed.fetch_add(n, Ordering::Relaxed); + })); + let layout = StoreLayout::new(store.clone(), "retained-object-index".into()); + let hash = write.upload(&store, &layout).await.unwrap(); + reads.store(0, Ordering::Relaxed); + let paths = PagedGitObjectIndex::retained_paths(&store, &layout, &hash, &state, 256 * 1024) + .await + .unwrap(); + assert_eq!( + (paths, reads.load(Ordering::Relaxed) <= MAX_DESCRIPTOR_BYTES), + ( + [ + layout.bulk_manifest_path("git-object-index", &hash), + body_path(&layout, &write.descriptor.body_hash) + ], + true + ) + ); +} + +#[tokio::test] +async fn retained_paths_require_the_captured_state_and_descriptor_budget() { + let (manifest, inventory, entries) = inputs(10); + let state = "c".repeat(64); + let write = PagedGitObjectIndexWrite::build(&state, &manifest, &inventory, &entries).unwrap(); + let store = Store::new(Arc::new(InMemory::new())); + let layout = StoreLayout::new(store.clone(), "retained-object-binding".into()); + let hash = write.upload(&store, &layout).await.unwrap(); + let foreign = "d".repeat(64); + for (state, maximum) in [(&foreign, 256 * 1024), (&state, 1)] { + assert!( + PagedGitObjectIndex::retained_paths(&store, &layout, &hash, state, maximum) + .await + .is_err() + ); + } +} + #[tokio::test] async fn object_aliases_prefer_full_encodings_and_keep_every_physical_offset() { let (mut manifest, mut inventory, mut entries) = inputs(1); diff --git a/crates/crab-remote/README.md b/crates/crab-remote/README.md index 20c9bf085..43b41b9a4 100644 --- a/crates/crab-remote/README.md +++ b/crates/crab-remote/README.md @@ -95,6 +95,10 @@ Foreground qualification uses `open_git_view_from_root`; browse-index builders use the complete metadata entry point. Pack payloads stay in authenticated origin ranges before and after checkpoints. Checkpoint catalog/visibility decoding and ref capture remain measured opening work, not a completed bounded-opening gate. +`capture-browse` captures the immutable view and readiness before its stdin +handshake, then constructs the repository after external GC. Its separate +`reader-open` measurement includes normal tip validation, keeping graph, physical +and path pages cold until the external lifecycle transition completes. `qualify_remote graph-pages` measures sampled ordinal/parent records and duplicate OID batches through `CommitGraphReader`. It verifies the derived graph's captured state and Git identity before reads. diff --git a/crates/crab-remote/examples/qualify_remote.rs b/crates/crab-remote/examples/qualify_remote.rs index 56661eeb9..cd5eff2b2 100644 --- a/crates/crab-remote/examples/qualify_remote.rs +++ b/crates/crab-remote/examples/qualify_remote.rs @@ -1,4 +1,6 @@ //! Measure isolated cold/warm capsule reads and emit independent-oracle evidence. +#[path = "support/captured_browse.rs"] +mod captured_browse; mod support; use std::io; @@ -36,7 +38,7 @@ async fn main() -> Result<()> { [bucket, prefix, path] => (bucket, prefix, path, "all"), [bucket, prefix, path, selection] => (bucket, prefix, path, selection.as_str()), _ => return Err(io::Error::other( - "usage: qualify_remote [operation|all|build-indexes|checkpoint|catalog|catalog-current|ref-point|resolve-oid|blame-history|blame-history-uncached]", + "usage: qualify_remote [operation|all|build-indexes|checkpoint|catalog|catalog-current|capture-browse|ref-point|resolve-oid|blame-history|blame-history-uncached]", ) .into()), }; @@ -44,6 +46,7 @@ async fn main() -> Result<()> { && selection != "build-indexes" && selection != "catalog" && selection != "catalog-current" + && selection != "capture-browse" && selection != "checkpoint" && selection != "graph-pages" && selection != "graph-ancestry" @@ -74,7 +77,20 @@ async fn main() -> Result<()> { .with_storage_observer(counters.clone()); let layout = StoreLayout::new(store, prefix.to_owned()); let identity = RepositoryIdentity::new(format!("s3:{bucket}"), prefix.as_str(), 1)?; - if selection == "ref-point" { + if selection == "capture-browse" { + let runtime = Arc::new(RemoteGitRuntime::default()); + let result = captured_browse::qualify( + &layout, + &identity, + runtime.clone(), + options, + &path, + &counters, + ) + .await; + runtime.shutdown().await; + result?; + } else if selection == "ref-point" { let root = counters .measure("open", "cold", async { Ok(crab_metadata::capsule_protocol::load_root(&layout).await?) diff --git a/crates/crab-remote/examples/support/captured_browse.rs b/crates/crab-remote/examples/support/captured_browse.rs new file mode 100644 index 000000000..2f6ba4c9d --- /dev/null +++ b/crates/crab-remote/examples/support/captured_browse.rs @@ -0,0 +1,76 @@ +//! Hold a cold reader while an external driver moves refs, rebuilds, and runs repository GC. +use std::io::{self, Write}; +use std::sync::Arc; + +use crab_remote_git::{GitPath, RemoteGitRuntime, RepositoryIdentity, RepositoryOptions, Revision}; +use crab_storage::{Store, StoreLayout}; +use serde_json::json; +use tokio_util::sync::CancellationToken; + +use super::support::{Measurements, Result, capture_repository_view}; + +pub async fn qualify( + layout: &StoreLayout, + identity: &RepositoryIdentity, + runtime: Arc, + options: RepositoryOptions, + path: &GitPath, + counters: &Measurements, +) -> Result<()> { + let cancel = CancellationToken::new(); + let view = counters + .measure("open", "cold", capture_repository_view(layout)) + .await?; + let head = view + .refs() + .get(view.head()) + .ok_or_else(|| io::Error::other("repository has no HEAD"))?; + let revision = Revision::Reference(view.head().to_owned()); + // Repository construction proves visible tips through graph pages. Delay + // it until after GC so even those pages stay cold in the captured view. + println!( + "{}", + json!({"kind":"captured", "schema":1, "commit":head, "generation":view.root().root().generation()}) + ); + io::stdout().flush()?; + let command = tokio::task::spawn_blocking(|| -> io::Result { + let mut command = String::new(); + io::stdin().read_line(&mut command)?; + Ok(command) + }) + .await??; + if command.trim() != "continue" { + return Err(io::Error::other( + "capture probe requires a continue command after external GC", + ) + .into()); + } + let repository = counters + .measure("reader-open", "after-external-gc", async { + Ok(view + .git_repository_from_store( + layout.clone(), + identity.clone(), + runtime, + options, + 2 * 1024 * 1024 * 1024, + &cancel, + ) + .await?) + }) + .await?; + for name in ["blob", "attribution", "path-history"] { + let value = counters + .measure( + name, + "after-external-gc", + super::query(name, &repository, &revision, path, &cancel), + ) + .await?; + println!( + "{}", + json!({"kind":"captured-result", "operation":name, "value":value}) + ); + } + Ok(()) +} diff --git a/crates/crab-remote/examples/support/mod.rs b/crates/crab-remote/examples/support/mod.rs index 221add4c4..d8c490bfb 100644 --- a/crates/crab-remote/examples/support/mod.rs +++ b/crates/crab-remote/examples/support/mod.rs @@ -233,18 +233,7 @@ pub async fn open_repository( options: RepositoryOptions, cancel: &CancellationToken, ) -> Result { - let root = crab_metadata::capsule_protocol::load_root(&layout).await?; - let indexes = crab_metadata::capsule_protocol::load_browse_indexes(&layout).await?; - let view = crab_read::capsule_protocol::open_git_view_from_root( - &layout, - root, - crab_read::capsule_protocol::CapsuleReadLimits { - max_capsule_bytes: 2 * 1024 * 1024 * 1024, - max_frontier_bytes: 2 * 1024 * 1024 * 1024, - }, - ) - .await? - .with_browse_indexes(indexes); + let view = capture_repository_view(&layout).await?; Ok(view .git_repository_from_store( layout, @@ -257,6 +246,24 @@ pub async fn open_repository( .await?) } +pub async fn capture_repository_view( + layout: &StoreLayout, +) -> Result { + let root = crab_metadata::capsule_protocol::load_root(layout).await?; + let indexes = crab_metadata::capsule_protocol::load_browse_indexes(layout).await?; + let view = crab_read::capsule_protocol::open_git_view_from_root( + layout, + root, + crab_read::capsule_protocol::CapsuleReadLimits { + max_capsule_bytes: 2 * 1024 * 1024 * 1024, + max_frontier_bytes: 2 * 1024 * 1024 * 1024, + }, + ) + .await? + .with_browse_indexes(indexes); + Ok(view) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/crab-remote/src/browse_indexes.rs b/crates/crab-remote/src/browse_indexes.rs index ebf1fa4f2..14b7b07e5 100644 --- a/crates/crab-remote/src/browse_indexes.rs +++ b/crates/crab-remote/src/browse_indexes.rs @@ -2,7 +2,9 @@ use std::{sync::Arc, time::Duration}; -use crab_metadata::capsule_protocol::{BrowseIndexes, MAX_BROWSE_INDEXES_BYTES, load_root}; +use crab_metadata::capsule_protocol::{ + BrowseIndexes, MAX_BROWSE_INDEXES_BYTES, load_root, retain_browse_indexes, +}; use crab_read::capsule_protocol::{ CapsuleReadLimits, open_view_from_root_with_metadata, read_activity_from_root, }; @@ -51,7 +53,7 @@ pub async fn ensure( max_capsule_bytes: max_bytes, max_frontier_bytes: max_bytes, }; - let view = open_view_from_root_with_metadata(layout, root, limits).await?; + let mut view = open_view_from_root_with_metadata(layout, root, limits).await?; let snapshot = view.git_snapshot()?; if snapshot.manifest.refs.is_empty() { return Ok(()); @@ -97,6 +99,12 @@ pub async fn ensure( } indexes = indexes.with_object_index(hash)?; } + // Lifetime publication precedes query readiness. If the later CAS + // loses, extra retained data is safe; the opposite order can strand readers. + retain_browse_indexes( + layout, &indexes, snapshot.manifest.generation, view.layered_checkpoint(), + view.capsule_run_pointers(), + ).await?; if previous.as_ref() == Some(&indexes) && indexes.path_state_hash().is_some() { ready = Some(indexes); return Ok(()); @@ -111,8 +119,8 @@ pub async fn ensure( }; // Bind the just-verified physical pages to this same view before path // construction. Reopening a fresh root could mix index and ref identities. - let repository = view.with_browse_indexes(Some(indexes.clone())) - .git_repository_from_store(layout.clone(), identity.clone(), runtime, options, max_bytes, cancel) + view = view.with_browse_indexes(Some(indexes.clone())); + let repository = view.git_repository_from_store(layout.clone(), identity.clone(), runtime, options, max_bytes, cancel) .await?; let complete = crab_write::generation::browse::build_path_state( layout, @@ -123,6 +131,10 @@ pub async fn ensure( cancel, ) .await?; + retain_browse_indexes( + layout, &complete, snapshot.manifest.generation, view.layered_checkpoint(), + view.capsule_run_pointers(), + ).await?; if publish_current(layout, &complete, etag, cancel) .await? .is_some() diff --git a/crates/crab-storage/src/layout.rs b/crates/crab-storage/src/layout.rs index 35c9a5aee..2e4641f8d 100644 --- a/crates/crab-storage/src/layout.rs +++ b/crates/crab-storage/src/layout.rs @@ -188,6 +188,18 @@ impl StoreLayout { self.repo_path("v2/browse-indexes") } + /// Path to the mutable lifetime root for published browse captures. + #[must_use] + pub fn browse_retained_head_path(&self) -> ObjectPath { + self.repo_path("v2/browse-retention") + } + + /// Path to one authenticated browse capture retained until explicit history pruning. + #[must_use] + pub fn browse_retained_root_path(&self, hash: &str) -> ObjectPath { + self.repo_path(&format!("v2/browse-roots/{hash}")) + } + /// Path to one immutable capsule-protocol capsule. #[must_use] pub fn capsule_path(&self, hash: &str) -> ObjectPath {