From 328f9eadc59492b3c033e67cbdb990d284169722 Mon Sep 17 00:00:00 2001 From: forhappy Date: Wed, 14 Oct 2026 19:26:26 -0700 Subject: [PATCH 1/3] fix(push): retain and retire capsule staging Pin exact staging owners through remote publication. Preserve committed recipes before rolling back interrupted native index intents, and reject uncommitted protected finalize states before releasing local payloads. --- .../docs/architecture/indexed-remote-reads.md | 55 +++++ .../e2e/run_add_commit_push_rustfs_smoke.py | 14 ++ .../e2e/run_large_file_edits_rustfs.py | 16 +- crab/src/cmd/add.rs | 36 ++- crab/src/git/capsule_push.rs | 207 ++++++++++++++++++ crab/src/git/index.rs | 173 ++++++++++++++- crab/src/git/push.rs | 1 + crates/crab-auth/README.md | 5 + crates/crab-auth/src/protected_push.rs | 44 ++++ crates/crab-staging/AGENTS.md | 5 + crates/crab-staging/README.md | 13 ++ crates/crab-staging/src/lib.rs | 151 ++++++++++--- crates/crab-staging/src/push_snapshot.rs | 70 ++++++ .../tests/publication_history_recovery.rs | 128 +++++++++++ .../crab-staging/tests/push_snapshot_guard.rs | 174 +++++++++++++++ 15 files changed, 1045 insertions(+), 47 deletions(-) create mode 100644 crates/crab-staging/src/push_snapshot.rs create mode 100644 crates/crab-staging/tests/publication_history_recovery.rs create mode 100644 crates/crab-staging/tests/push_snapshot_guard.rs diff --git a/crab/docs/architecture/indexed-remote-reads.md b/crab/docs/architecture/indexed-remote-reads.md index a10a26ba6..b99f91da3 100644 --- a/crab/docs/architecture/indexed-remote-reads.md +++ b/crab/docs/architecture/indexed-remote-reads.md @@ -2016,3 +2016,58 @@ therefore do not qualify capsule lifecycle. A separate follow-up must bind exact selected owners before publication and release only proven committed owners, including cancellation, partial/ref-CAS failure and concurrent readers. Until that is implemented and tested, local post-push reclamation is not qualified. + +### Capsule snapshot lifecycle follow-up + +Capsule push fences retirement before selecting local recipes, pins selected +published/delegated owners, and retains that snapshot through upload and ref +publication. Confirmed remote commit precedes the local committed journal and +exact retirement. Local cleanup errors leave maintenance work and preserve the +successful remote result. Ref rejection or failure retains unpublished leases. + +`StagingPushSnapshot` owns marker/snapshot cleanup on drop, including task abort. +Sibling readers defer retirement to exclusive maintenance. It shares the existing +index and physical-payload implementation with the released asynchronous API; +there is no storage-format or dependency change. Guard tests cover drop, task +abort, shared native ownership, sibling deferral and local journal failure. + +Native publication-intent recovery now runs in capsule push before snapshot +admission. Recovery uses Git's immediate index lock. If the complete prior index +state is restored, a bounded-memory local Git object walk checks refs and reflogs +before native rollback. Recipes still referenced there acquire canonical history +owners before native leases are released. Cancellation or an incomplete history +walk fails without authorizing rollback. Ordinary pushes without unresolved +intents do not perform this history scan; this does not replace the historical +remote-catalog indexing workstream. + +Both exclusive native-add recovery and shared push recovery use this ownership +rule. The shared staging helper validates all requested recipes against the +intent before adding history owners, preserves the original intent across an +interrupted prefix, and permits an idempotent retry. Prepared-only recovery tests +cover subset retention, full reconstruction and physical xorb deletion only after +confirmed snapshot commit. + +Both protected finalize clients use the shared auth response validator. A response +with coordinator metadata must attest `Committed` or `Materialized` before it can +permit local retirement. `Pending`, `ObjectsUploaded` and `Aborted` are rejected; +the existing successful response without coordinator metadata remains supported. +The minimal protected-push suite passes 30 tests. + +The existing managed-file smoke checks post-push owners, snapshots, raw payloads +and prepared ownership bytes for native and Git helper push. The frozen #566 CLI +baseline completes native add/commit/push and verifies the remote tip, then fails +the retirement assertion with retained owners/prepared payload. Corrected CLI +proof and final-source 10-GiB qualification remain required. + +The large-file harness adds `--pattern entropy|repeated|zero`; entropy preserves +its original seed/default. Repeated input uses a deterministic one-MiB block; +zero input writes real blocks. All patterns retain four alternating native/Git +versions, beginning/middle/end edits, two-file identity checks, source removal and +cold historical reconstruction. Syntax checks pass; the new patterns are unrun. + +The frozen #570 release build succeeds with `simd-accel,gix-pathmatch` and no +default features. Its binary hash is recorded externally. Resumed workload +`unified-ingestion-10g-r3` passes the formerly failing v1 Git-clean push and the +subsequent v2 native push, with 2,216,109 and 2,133,870 incremental xorb bytes. +The full workload is still running; cold historical reconstruction is unproven. +This artifact precedes the lifecycle and finalize-contract changes above. diff --git a/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py b/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py index 82c378e42..50a70f816 100755 --- a/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py +++ b/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py @@ -1607,6 +1607,20 @@ def run_case(self, case_name: str, use_crab_add: bool) -> None: ) self.head_repository_state(repo_prefix) + ownership = self.run_crab( + repo, ["staging", "stats", "--json"], + name=f"{case_name} staging ownership after push", + ) + lifecycle = json.loads(self.read_stdout(ownership))["data"]["lifecycle"] + self.check( + f"{case_name}-retires-published-staging-owners", + all(lifecycle[field] == 0 for field in ( + "path_heads", "path_leases", "open_push_snapshots", + "committed_push_snapshots", "payloads", + "current_head_prepared_bytes", "snapshot_pinned_prepared_bytes", + )), + lifecycle, + ) after_xorbs = self.list_keys(".crab/xorbs/") after_shards = self.list_keys(".crab/shards/") new_xorbs = len(after_xorbs - before_xorbs) diff --git a/crab/scripts/e2e/run_large_file_edits_rustfs.py b/crab/scripts/e2e/run_large_file_edits_rustfs.py index 71323b5bd..c81cf662c 100644 --- a/crab/scripts/e2e/run_large_file_edits_rustfs.py +++ b/crab/scripts/e2e/run_large_file_edits_rustfs.py @@ -1,7 +1,8 @@ #!/usr/bin/env python3 """Qualify repeated 10-GiB edits through both Git clean and native Crab push. -Two copy-on-write files start with identical, non-sparse deterministic entropy. +Two copy-on-write files start with identical, non-sparse deterministic content. +Choose entropy, a repeated one-MiB block, or zero-filled bytes. Edits diverge at the beginning, middle and end. Each committed version must hydrate byte-identically from a fresh cache after the source repository is removed. Only this invocation's disposable data is released; logs and receipts remain. @@ -77,9 +78,15 @@ def qualify(runner: AddCommitPushSmoke, proxy: RequestCountingProxy) -> None: repo, remote, prefix = runner.prepare_repo("source") cap_cache(repo) paths = [repo / "model-a.bin", repo / "model-b.bin"] + repeated_block = None + if args.pattern == "zero": + repeated_block = bytes(MIB) + elif args.pattern == "repeated": + repeated_block = hashlib.shake_256(b"large-edits:repeated").digest(MIB) with paths[0].open("wb") as output: for block in range(args.file_mib): - output.write(hashlib.shake_256(f"large-edits:{block}".encode()).digest(MIB)) + output.write(repeated_block if repeated_block is not None else + hashlib.shake_256(f"large-edits:{block}".encode()).digest(MIB)) clone_flag = "-c" if sys.platform == "darwin" else "--reflink=always" runner.run_cmd("copy-on-write second large file", ["cp", clone_flag, str(paths[0]), str(paths[1])], repo) @@ -89,13 +96,13 @@ def qualify(runner: AddCommitPushSmoke, proxy: RequestCountingProxy) -> None: (code / f"module-{index:05}.rs").write_text(f"pub const VALUE: u64 = {index};\n") runner.check("workload", all(path.stat().st_size == size for path in paths), {"file_bytes": size, "files": 2, "initial_unique_bytes": size, - "versions": 4, "code_files": args.code_files}) + "versions": 4, "code_files": args.code_files, "pattern": args.pattern}) history = [] phases = [] def record() -> None: receipt = runner.artifacts / "large-file-edits.json" - receipt.write_text(json.dumps({"history": history, "phases": phases, + receipt.write_text(json.dumps({"history": history, "phases": phases, "pattern": args.pattern, "transport": proxy.snapshot()}, indent=2) + "\n") runner.report.artifacts["large_file_edits"] = str(receipt) runner.write_report() @@ -194,6 +201,7 @@ def main() -> None: parser.add_argument("--endpoint-url", required=True) parser.add_argument("--file-mib", type=int, default=10240) parser.add_argument("--code-files", type=int, default=5000) + parser.add_argument("--pattern", choices=("entropy", "repeated", "zero"), default="entropy") args = parser.parse_args() if args.file_mib < 16 or args.code_files < 1 or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", args.run_id): parser.error("require >=16 MiB/file, positive code files and a safe run id") diff --git a/crab/src/cmd/add.rs b/crab/src/cmd/add.rs index 006dc45b8..5dc8759fc 100644 --- a/crab/src/cmd/add.rs +++ b/crab/src/cmd/add.rs @@ -1050,7 +1050,7 @@ async fn execute_add( // pointers published by the process that just released the staging lock. let staging_root = worktree_ctx.shared_staging_dir(); let lock_wait_start = Instant::now(); - let staging = Arc::new(open_add_staging(&staging_root, &repo_root).await?); + let staging = Arc::new(open_add_staging(&staging_root, &repo_root, cancel).await?); let lock_wait_duration_ms = lock_wait_start.elapsed().as_millis() as u64; let (candidates, clean_skipped) = filter_clean_indexed_candidates(&repo_root, candidates, jobs, cancel).await?; @@ -1656,12 +1656,16 @@ async fn execute_add( Ok(summary) } -async fn open_add_staging(staging_root: &Path, repo_root: &Path) -> Result { +async fn open_add_staging( + staging_root: &Path, + repo_root: &Path, + cancel: &CancellationToken, +) -> Result { let staging = StagingArea::open_blocking(staging_root.to_path_buf(), ADD_STAGING_LOCK_WAIT_BUDGET) .await .map_err(CrabError::from)?; - reconcile_publication_intents_writer(&staging, repo_root)?; + reconcile_publication_intents_writer(&staging, repo_root, cancel)?; Ok(staging) } @@ -5891,7 +5895,8 @@ mod tests { let repo_root = repo_root.clone(); let path = path.clone(); async move { - let staging = open_add_staging(&root, &repo_root).await?; + let staging = + open_add_staging(&root, &repo_root, &CancellationToken::new()).await?; let filtered = filter_clean_indexed_candidates( &repo_root, vec![(path, data.len() as u64)], @@ -6225,7 +6230,9 @@ mod tests { assert!(intent.is_some()); staging.close().await.unwrap(); - let recovered = open_add_staging(&staging_root, &repo).await.unwrap(); + let recovered = open_add_staging(&staging_root, &repo, &CancellationToken::new()) + .await + .unwrap(); assert!( recovered @@ -6271,7 +6278,7 @@ mod tests { staging.close().await.unwrap(); let recovered = StagingAreaReadOnly::open(staging_root).await.unwrap(); - reconcile_publication_intents(&recovered, &repo) + reconcile_publication_intents(&recovered, &repo, &CancellationToken::new()) .await .unwrap(); @@ -6323,14 +6330,19 @@ mod tests { let (error, intents, bytes) = if read_only { staging.close().await.unwrap(); let staging = StagingAreaReadOnly::open(root).await.unwrap(); - let result = reconcile_publication_intents(&staging, &repo).await; + let result = + reconcile_publication_intents(&staging, &repo, &CancellationToken::new()).await; ( result.unwrap_err(), staging.unresolved_publication_intents().unwrap().len(), staging.get_chunk(&compute_data_hash(data)).await.unwrap(), ) } else { - let result = reconcile_publication_intents_writer(&staging, &repo); + let result = reconcile_publication_intents_writer( + &staging, + &repo, + &CancellationToken::new(), + ); let state = ( result.unwrap_err(), staging.unresolved_publication_intents().unwrap().len(), @@ -6428,7 +6440,7 @@ mod tests { }) .await .unwrap(); - let blocked = reconcile_publication_intents(&staging, repo) + let blocked = reconcile_publication_intents(&staging, repo, &CancellationToken::new()) .await .unwrap_err(); let retained = staging.unresolved_publication_intents().unwrap().len() == 1 @@ -6450,7 +6462,9 @@ mod tests { .await .unwrap(); assert!(status.success(), "Git index installation failed"); - reconcile_publication_intents(&staging, repo).await.unwrap(); + reconcile_publication_intents(&staging, repo, &CancellationToken::new()) + .await + .unwrap(); assert!( matches!(blocked, CrabError::Io(_)) && retained @@ -6510,7 +6524,7 @@ mod tests { staging.close().await.unwrap(); let recovered = StagingAreaReadOnly::open(staging_root).await.unwrap(); - let error = reconcile_publication_intents(&recovered, &repo) + let error = reconcile_publication_intents(&recovered, &repo, &CancellationToken::new()) .await .expect_err("mixed publication state must fail closed"); diff --git a/crab/src/git/capsule_push.rs b/crab/src/git/capsule_push.rs index 23ae886f9..ae657e236 100644 --- a/crab/src/git/capsule_push.rs +++ b/crab/src/git/capsule_push.rs @@ -6,6 +6,7 @@ use std::sync::Arc; use bytes::Bytes; use crab_staging::StagingAreaReadOnly; +use crab_xet::hash::MerkleHash; use gix_object::{Exists, Find, FindHeader}; use tokio_util::sync::CancellationToken; @@ -322,6 +323,25 @@ async fn run_inner( config.force_full_graph, ) .await?; + if !prepared.pointers.is_empty() + && let Some(staging) = staging + && !staging.unresolved_publication_intents()?.is_empty() + { + // A native add can install Git's pointer before its local publication + // acknowledgement. Recover under the canonical index lock before admission. + let worktree = gix::open(&git_dir) + .map_err(|error| CrabError::Io(std::io::Error::other(error)))? + .workdir() + .map(Path::to_path_buf) + .ok_or_else(|| { + CrabError::StagingCorrupt( + "native publication recovery requires a Git working tree".to_owned(), + ) + })?; + super::index::reconcile_publication_intents(staging, &worktree, cancel).await?; + } + + let staging_snapshot = capture_staging_snapshot(staging, &prepared.pointers)?; let visibility_delta = prepare_visibility_delta( &common_git_dir, &edits, @@ -519,6 +539,11 @@ async fn run_inner( } }; if let Some(active_active_commit) = committed { + if let Some(snapshot) = staging_snapshot + && let Err(error) = snapshot.commit() + { + tracing::warn!(%error, "capsule committed; local staging cleanup needs maintenance"); + } for ref_name in ref_names { outcomes.insert(ref_name, RefPushOutcome::Ok); } @@ -544,6 +569,41 @@ async fn run_inner( Ok((PushResult::new(outcomes), None)) } +fn capture_staging_snapshot( + staging: Option<&Arc>, + pointers: &[crab_types::pointer::Pointer], +) -> Result> { + let Some(staging) = staging.filter(|_| !pointers.is_empty()) else { + return Ok(None); + }; + // Fence sibling retirement before selecting local roots; snapshot pins + // then protect exactly those leases through upload and the ref transition. + let snapshot = crab_staging::StagingPushSnapshot::begin( + Arc::clone(staging), + uuid::Uuid::now_v7().to_string(), + )?; + let hashes = pointers + .iter() + .map(|pointer| MerkleHash::from(pointer.file_hash)) + .collect::>(); + let selected = super::push_staging::recipes_for_selected_pointers(staging, &hashes)?; + for pointer in pointers { + if let Some(Some(recipe)) = selected.get(&MerkleHash::from(pointer.file_hash)) + && recipe.file_size() != pointer.size + { + return Err(CrabError::StagingCorrupt( + "selected push recipe disagrees with pointer size".to_owned(), + )); + } + } + let recipes = selected.into_values().flatten().collect::>(); + if recipes.is_empty() { + return Ok(None); + } + snapshot.capture(&recipes)?; + Ok(Some(snapshot)) +} + enum CapsulePublishAttempt { Committed(Option), RefChanged, @@ -2006,6 +2066,153 @@ mod tests { ); } + #[tokio::test(flavor = "multi_thread")] + async fn capsule_push_retires_staging_only_after_confirmed_ref_publication() { + use bstr::ByteSlice; + use crab_staging::stream::{StreamStageProgress, stage_reader_streaming}; + let _git_env = crate::test::git_repo::CleanGitEnvGuard::new(); + for (stale, native_intent, restored_index) in [ + (false, false, false), + (true, false, false), + (false, true, false), + (true, true, false), + (false, true, true), + ] { + let source = tempfile::tempdir().unwrap(); + git(source.path(), &["init", "-b", "main"]); + git(source.path(), &["config", "user.name", "Crab Test"]); + git( + source.path(), + &["config", "user.email", "crab@example.invalid"], + ); + let first = commit(source.path(), "initial"); + let staging_root = source.path().join(".crab/staging"); + let staging = crab_staging::StagingArea::open(staging_root.clone()) + .await + .unwrap(); + let staged = stage_reader_streaming( + &mut std::io::Cursor::new(b"capsule payload"), + Path::new("model.bin"), + &staging, + StreamStageProgress::default(), + &CancellationToken::new(), + ) + .await + .unwrap(); + if !native_intent { + staging + .delegate_publication(&[(&staged.batch_id, Path::new("model.bin"))]) + .unwrap(); + } + let pointer = crab_types::pointer::Pointer { + file_hash: staged.file_hash, + size: staged.size, + shard_hint: None, + }; + std::fs::write(source.path().join("model.bin"), pointer.serialize()).unwrap(); + if native_intent { + let index = gix_index::File::at_or_default( + source.path().join(".git/index"), + gix_hash::Kind::Sha1, + false, + gix_index::decode::Options::default(), + ) + .unwrap(); + let previous_index_state = + super::super::index::git_index_path_state(&index, b"model.bin".as_bstr()); + staging + .create_publication_intent(&[crab_staging::PublicationIntentEntry { + batch_id: staged.batch_id.clone(), + path: "model.bin".into(), + expected_pointer_oid: git( + source.path(), + &["hash-object", "-w", "model.bin"], + ), + previous_index_state, + }]) + .unwrap(); + } + git(source.path(), &["add", "model.bin"]); + git(source.path(), &["commit", "-m", "managed payload"]); + let tip = git(source.path(), &["rev-parse", "HEAD"]); + if restored_index { + git(source.path(), &["read-tree", "HEAD^"]); + } + staging.close().await.unwrap(); + let staging = Arc::new(StagingAreaReadOnly::open(staging_root).await.unwrap()); + let storage = crab_storage::Store::new(Arc::new(InMemory::new())); + let store = crate::storage::store::Store::from_storage(storage); + let router = crate::storage::StoreLayout::new(store.clone(), "repos/test".to_owned()); + let layout = + crab_storage::StoreLayout::new(store.as_storage().clone(), "repos/test".to_owned()); + let root = crab_write::capsule_protocol::initialize( + &layout, + &"1".repeat(64), + "refs/heads/main", + ) + .await + .unwrap(); + let limits = crab_read::capsule_protocol::CapsuleReadLimits { + max_capsule_bytes: 16 * 1024 * 1024, + max_frontier_bytes: 16 * 1024 * 1024, + }; + let initial = crab_read::capsule_protocol::open_view_from_root(&layout, root, limits) + .await + .unwrap(); + if stale { + publish_ref_edits_with_visibility( + &layout, + &source.path().join(".git"), + vec![crab_metadata::capsule_protocol::CapsuleRefEdit::new( + "refs/heads/main", + None, + Some(first.clone()), + None, + )], + ) + .await; + } + let config = PushConfig { + git_dir: Some(source.path().join(".git")), + ..Default::default() + }; + let (result, _) = run( + &config, + &[PushSpec { + force: false, + src: "refs/heads/main".to_owned(), + dst: "refs/heads/main".to_owned(), + }], + &store, + &router, + Some(initial.into()), + &[], + Some(&staging), + None, + None, + &CancellationToken::new(), + ) + .await + .unwrap(); + let health = staging.lifecycle_health().unwrap(); + let view = crab_read::capsule_protocol::open_view(&layout, limits) + .await + .unwrap(); + assert!( + result.all_ok() != stale + && view.refs().get("refs/heads/main") + == Some(if stale { &first } else { &tip }) + && health.path_leases == u64::from(stale) + && health.payloads == u64::from(stale) + && health.open_push_snapshots == 0 + && health.committed_push_snapshots == 0 + && health.unresolved_publications == 0 + && staging.list_inflight().unwrap().is_empty(), + "stale={stale}, native_intent={native_intent}, restored_index={restored_index}, health={health:?}" + ); + } + } + #[tokio::test] async fn single_ref_successor_extends_committed_multi_ref_history() { let source = tempfile::tempdir().expect("source repository"); diff --git a/crab/src/git/index.rs b/crab/src/git/index.rs index bba69e0c3..d3c80ac24 100644 --- a/crab/src/git/index.rs +++ b/crab/src/git/index.rs @@ -1,5 +1,8 @@ +use crab_xet::hash::MerkleHash; +use std::collections::HashSet; use std::path::Path; use std::process::Command; +use tokio_util::sync::CancellationToken; use crate::core::error::{CrabError, Result}; use crab_staging::{PublicationIntentId, StagingArea, StagingAreaReadOnly}; @@ -52,20 +55,31 @@ pub(crate) fn git_index_path_state(index: &gix_index::File, path: &bstr::BStr) - pub(crate) async fn reconcile_publication_intents( staging: &StagingAreaReadOnly, repo_root: &Path, + cancel: &CancellationToken, ) -> Result<()> { + crate::core::error::check_cancelled(cancel)?; if staging.unresolved_publication_intents()?.is_empty() { return Ok(()); } let (_index_lock, index) = lock_publication_index(repo_root)?; // Another shared recovery may have completed before lock acquisition. // Read intents again under the lock rather than acting on a stale list. - let actions = publication_recovery_actions(staging.unresolved_publication_intents()?, &index)?; + let actions = publication_recovery_actions( + staging.unresolved_publication_intents()?, + &index, + repo_root, + cancel, + )?; for action in actions { + crate::core::error::check_cancelled(cancel)?; match action { PublicationRecoveryAction::Publish(intent_id) => { staging.publish_publication_intent(&intent_id)?; } - PublicationRecoveryAction::Rollback(intent_id) => { + PublicationRecoveryAction::Rollback(intent_id, history) => { + if !history.is_empty() { + staging.preserve_publication_history(&intent_id, &history, cancel)?; + } staging.rollback_publication_intent(&intent_id).await?; } } @@ -77,20 +91,31 @@ pub(crate) async fn reconcile_publication_intents( pub(crate) fn reconcile_publication_intents_writer( staging: &StagingArea, repo_root: &Path, + cancel: &CancellationToken, ) -> Result<()> { + crate::core::error::check_cancelled(cancel)?; if staging.unresolved_publication_intents()?.is_empty() { return Ok(()); } let (_index_lock, index) = lock_publication_index(repo_root)?; // Another shared recovery may have completed before lock acquisition. // Read intents again under the lock rather than acting on a stale list. - let actions = publication_recovery_actions(staging.unresolved_publication_intents()?, &index)?; + let actions = publication_recovery_actions( + staging.unresolved_publication_intents()?, + &index, + repo_root, + cancel, + )?; for action in actions { + crate::core::error::check_cancelled(cancel)?; match action { PublicationRecoveryAction::Publish(intent_id) => { staging.publish_publication_intent(&intent_id)?; } - PublicationRecoveryAction::Rollback(intent_id) => { + PublicationRecoveryAction::Rollback(intent_id, history) => { + if !history.is_empty() { + staging.preserve_publication_history(&intent_id, &history, cancel)?; + } staging.rollback_publication_intent(&intent_id)?; } } @@ -191,7 +216,7 @@ fn delegated_pointer_matches( enum PublicationRecoveryAction { Publish(PublicationIntentId), - Rollback(PublicationIntentId), + Rollback(PublicationIntentId, Vec), } fn lock_publication_index(repo_root: &Path) -> Result<(gix_lock::Marker, gix_index::File)> { @@ -226,11 +251,14 @@ fn lock_publication_index(repo_root: &Path) -> Result<(gix_lock::Marker, gix_ind fn publication_recovery_actions( intents: Vec, index: &gix_index::File, + repo_root: &Path, + cancel: &CancellationToken, ) -> Result> { use bstr::ByteSlice; use gix_index::entry; let mut actions = Vec::with_capacity(intents.len()); + let mut rollbacks = Vec::new(); for intent in intents { let new_matches = intent .entries @@ -255,7 +283,7 @@ fn publication_recovery_actions( if new_matches == intent.entries.len() { actions.push(PublicationRecoveryAction::Publish(intent.intent_id)); } else if old_matches == intent.entries.len() { - actions.push(PublicationRecoveryAction::Rollback(intent.intent_id)); + rollbacks.push(intent); } else { return Err(CrabError::StagingCorrupt(format!( "add publication {} has an ambiguous Git index state ({new_matches}/{} new paths, {old_matches}/{} prior paths); restore one complete index state, then retry", @@ -265,14 +293,147 @@ fn publication_recovery_actions( ))); } } + let wanted = rollbacks + .iter() + .flat_map(|intent| intent.entries.iter()) + .map(|entry| entry.expected_pointer_oid.clone()) + .collect::>(); + let committed = committed_pointer_oids(repo_root, &wanted, cancel)?; + for intent in rollbacks { + let history = intent + .entries + .iter() + .filter(|entry| committed.contains(&entry.expected_pointer_oid)) + .map(|entry| entry.recipe_hash) + .collect(); + actions.push(PublicationRecoveryAction::Rollback( + intent.intent_id, + history, + )); + } Ok(actions) } +fn committed_pointer_oids( + repo_root: &Path, + wanted: &HashSet, + cancel: &CancellationToken, +) -> Result> { + use std::io::{BufRead, Read}; + if wanted.is_empty() { + return Ok(HashSet::new()); + } + // Only crash recovery needs this local history walk. Stream OIDs from refs + // and reflogs: an old index alone cannot authorize deleting committed bytes. + let mut command = Command::new("git"); + command + .current_dir(repo_root) + .env("GIT_NO_LAZY_FETCH", "1") + .env("GIT_ALLOW_PROTOCOL", "") + .args([ + "--no-replace-objects", + "rev-list", + "--all", + "--reflog", + "--objects", + "--no-object-names", + "--use-bitmap-index", + ]); + for name in super::process::GIT_ENV_REMOVALS { + command.env_remove(name); + } + let output = super::process::run( + command, + cancel, + None:: Result<()>>, + |stdout| { + let mut found = HashSet::new(); + let mut reader = std::io::BufReader::new(stdout); + let mut line = String::with_capacity(41); + loop { + crate::core::error::check_cancelled(cancel)?; + line.clear(); + let count = (&mut reader).take(42).read_line(&mut line)?; + if count == 0 { + break; + } + let oid = line + .strip_suffix('\n') + .filter(|oid| oid.len() == 40 && oid.bytes().all(|b| b.is_ascii_hexdigit())) + .ok_or_else(|| { + CrabError::Internal( + "Git recovery history scan returned an invalid object ID".to_owned(), + ) + })?; + if wanted.contains(oid) { + found.insert(oid.to_owned()); + } + } + Ok(found) + }, + )?; + if !output.status.success() { + return Err(CrabError::Internal(format!( + "Git recovery history scan failed ({}): {}", + output.status, + String::from_utf8_lossy(&output.stderr).trim() + ))); + } + Ok(output.stdout) +} + #[cfg(test)] #[allow(clippy::unwrap_used, reason = "test assertions")] mod tests { use super::*; + #[test] + fn recovery_history_includes_reflogs_and_rejects_incomplete_walks() { + let _env = crate::test::git_repo::CleanGitEnvGuard::new(); + let temp = tempfile::tempdir().unwrap(); + let git = |args: &[&str]| { + let output = Command::new("git") + .args(args) + .current_dir(temp.path()) + .output() + .unwrap(); + assert!(output.status.success(), "{:?}", output.stderr); + String::from_utf8(output.stdout).unwrap().trim().to_owned() + }; + git(&["init", "-q", "-b", "main"]); + git(&["config", "user.name", "Crab Test"]); + git(&["config", "user.email", "crab@example.invalid"]); + std::fs::write(temp.path().join("file"), b"initial").unwrap(); + git(&["add", "file"]); + git(&["commit", "-qm", "initial"]); + let initial = git(&["rev-parse", "HEAD"]); + std::fs::write(temp.path().join("file"), b"committed pointer").unwrap(); + git(&["add", "file"]); + git(&["commit", "-qm", "pointer"]); + let pointer = git(&["rev-parse", "HEAD:file"]); + let tree = git(&["rev-parse", "HEAD^{tree}"]); + git(&["reset", "--hard", &initial]); + let wanted = HashSet::from([pointer]); + let cancel = CancellationToken::new(); + assert_eq!( + committed_pointer_oids(temp.path(), &wanted, &cancel).unwrap(), + wanted + ); + cancel.cancel(); + assert!(matches!( + committed_pointer_oids(temp.path(), &wanted, &cancel), + Err(CrabError::Cancelled) + )); + std::fs::remove_file( + temp.path() + .join(".git/objects") + .join(&tree[..2]) + .join(&tree[2..]), + ) + .unwrap(); + assert!(committed_pointer_oids(temp.path(), &wanted, &CancellationToken::new()).is_err()); + } + #[test] fn metadata_staging_propagates_git_index_failure() { let dir = tempfile::tempdir().unwrap(); diff --git a/crab/src/git/push.rs b/crab/src/git/push.rs index f5bb8df89..d9e7b97fe 100644 --- a/crab/src/git/push.rs +++ b/crab/src/git/push.rs @@ -9716,6 +9716,7 @@ impl PushPipeline { crate::git::index::reconcile_publication_intents( staging, &worktree.current_worktree_root, + &self.cancel, ) .await?; } diff --git a/crates/crab-auth/README.md b/crates/crab-auth/README.md index 56681e7e0..034f1643d 100644 --- a/crates/crab-auth/README.md +++ b/crates/crab-auth/README.md @@ -18,6 +18,11 @@ The crate also owns the wire contracts used by protected pushes and managed transfer grants. A caller can therefore validate an auth response before any credential is handed to an object-store builder. +A successful protected finalize response requires `status: updated`. If it +includes coordinator metadata, the complete metadata must attest `Committed` +or `Materialized`; pending, uploaded-only and aborted states are rejected before +a caller can retire its local staging payloads. + ## Architecture The main flow is: diff --git a/crates/crab-auth/src/protected_push.rs b/crates/crab-auth/src/protected_push.rs index b796f6419..96e855346 100644 --- a/crates/crab-auth/src/protected_push.rs +++ b/crates/crab-auth/src/protected_push.rs @@ -162,6 +162,17 @@ pub fn validate_push_finalize_response(response: &PushFinalizeResponse) -> Resul return invalid_finalize_response("partial active-active commit metadata"); } + // A successful finalize authorizes releasing the client's local payloads. + // Pending or aborted coordinator state cannot attest remote publication. + if let Some(state) = response.commit_state + && !matches!( + state, + PushTransactionState::Committed | PushTransactionState::Materialized + ) + { + return invalid_finalize_response("active-active push has not committed"); + } + if let Some(operation_id) = response.operation_id.as_deref() && operation_id.trim().is_empty() { @@ -560,6 +571,39 @@ mod tests { assert!(err.to_string().contains("partial active-active")); } + #[test] + fn finalize_success_requires_committed_coordinator_state() { + for state in [ + PushTransactionState::Pending, + PushTransactionState::ObjectsUploaded, + PushTransactionState::Committed, + PushTransactionState::Materialized, + PushTransactionState::Aborted, + ] { + let outcome = CommitOutcome { + operation_id: "operation".to_owned(), + coordinator_epoch: 1, + writer: "writer".to_owned(), + region: "region".to_owned(), + manifest_generation: 1, + commit_sequence: 1, + state, + }; + let response = PushFinalizeResponse::updated_with_commit_outcome( + vec![valid_ref_update()], + Some(&outcome), + ); + assert_eq!( + validate_push_finalize_response(&response).is_ok(), + matches!( + state, + PushTransactionState::Committed | PushTransactionState::Materialized + ), + "state={state:?}", + ); + } + } + #[test] fn validate_push_finalize_response_rejects_unexpected_status() { let response = PushFinalizeResponse { diff --git a/crates/crab-staging/AGENTS.md b/crates/crab-staging/AGENTS.md index b083d1aa6..3818b2299 100644 --- a/crates/crab-staging/AGENTS.md +++ b/crates/crab-staging/AGENTS.md @@ -62,6 +62,11 @@ Trace one path: `crab/src/cmd/add.rs` (`close_staging_before_indexing`) → active native owners. Layout v3 upgrades released v1 atomically and rejects the unreleased v2 prototype. Sources: `index.rs`, `lib.rs`, and crate README. +- `StagingPushSnapshot` owns reader fencing and exact snapshot cleanup. Product + callers establish remote commit; the guard retains deferred committed journals + and preserves unpublished leases on cancellation. Read `src/push_snapshot.rs` + and the capsule consumer before changing this boundary. + ## Features and platform No declared Cargo features. Requires writable local filesystem/SQLite and platform lock semantics. Scale tests under `crates/crab-staging/tests` need dedicated disk/time budgets; do not run them as an incidental unit check. diff --git a/crates/crab-staging/README.md b/crates/crab-staging/README.md index be12d13db..7b954a0dc 100644 --- a/crates/crab-staging/README.md +++ b/crates/crab-staging/README.md @@ -94,6 +94,19 @@ and active native transactions survive. Publishing a newer head never cancels another caller's open lease. Failed Git conversion can leave a retained handoff; reconciliation and abandoned-handoff reclamation still need qualification. +`StagingPushSnapshot::begin` fences retirement before recipe selection; +`capture` pins exact selected leases. Call `commit` only after confirmed remote +publication. It records a committed journal before retiring captured owners. +Drop clears the reader marker and discards an uncommitted snapshot when no +sibling reader prevents cleanup. Deferred journals remain for exclusive +maintenance; native and uncaptured leases remain owned throughout. + +Native intent recovery may preserve caller-attested committed recipes under +canonical history owners before rolling back an old Git index state. +`preserve_publication_history` accepts only verified recipe roots belonging to +that unresolved intent. It leaves the original intent intact on cancellation or +error, so interrupted preservation can be retried before rollback. + ## Recovery contract Opening writable staging runs recovery before creating or reusing its writer. diff --git a/crates/crab-staging/src/lib.rs b/crates/crab-staging/src/lib.rs index 4ffd03668..025bbbc30 100644 --- a/crates/crab-staging/src/lib.rs +++ b/crates/crab-staging/src/lib.rs @@ -44,6 +44,7 @@ mod index; pub mod metrics; pub mod multipart_resume; pub mod push_plan; +mod push_snapshot; pub mod recipe; mod recovery; mod segment; @@ -69,6 +70,7 @@ mod prop_torn_tail_recovery; pub use config::StagingConfig; pub use error::{Result, StagingError}; pub use metrics::StagingMetrics; +pub use push_snapshot::StagingPushSnapshot; pub use stats::{ CompactionStats, RetireStats, StagingCleanStats, StagingLifecycleHealth, StagingStats, StagingVerifyStats, @@ -256,6 +258,97 @@ fn unresolved_publication_intents( .collect() } +fn publish_verified_recipe_owner( + root: &Path, + index: &Mutex, + owner: &Path, + recipe: &crate::recipe::FileRecipe, + display_path: Option<&str>, +) -> Result { + let batch_id = new_staging_batch_id(); + let file_hash: [u8; 32] = recipe.file_hash().into(); + let result = (|| { + let guard = lock_index(index)?; + guard.insert_batch(batch_id.as_str())?; + guard.insert_file(&file_hash, recipe.file_size())?; + guard.insert_recipe_lease( + batch_id.as_str(), + &staging_path_bytes(owner), + recipe, + RecipeVerification::CallerVerified, + )?; + if let Some(display_path) = display_path { + guard.insert_file_path(&file_hash, display_path)?; + } + guard.mark_batch_published(batch_id.as_str()) + })(); + if let Err(error) = result.and_then(|unleased| remove_indexed_files(root, index, &unleased)) { + let unowned = lock_index(index).and_then(|guard| { + guard.rollback_batch(batch_id.as_str())?; + guard.has_file_owner(&file_hash).map(|owned| !owned) + }); + if matches!(unowned, Ok(true)) { + let _ = remove_indexed_files(root, index, &[file_hash]); + } + return Err(error); + } + Ok(batch_id) +} + +fn preserve_publication_history( + root: &Path, + index: &Mutex, + intent_id: &PublicationIntentId, + recipe_hashes: &[MerkleHash], + cancel: &tokio_util::sync::CancellationToken, +) -> Result<()> { + if cancel.is_cancelled() { + return Err(StagingError::Cancelled); + } + let intent = unresolved_publication_intents(index)? + .into_iter() + .find(|intent| &intent.intent_id == intent_id) + .ok_or_else(|| StagingError::NotFound { + path: format!("publication intent {}", intent_id.as_str()), + })?; + let members = intent + .entries + .iter() + .map(|entry| entry.recipe_hash) + .collect::>(); + let recipes = { + let guard = lock_index(index)?; + recipe_hashes + .iter() + .copied() + .collect::>() + .into_iter() + .map(|hash| { + if !members.contains(&hash) { + return Err(StagingError::StagingCorrupt( + "history recipe is not owned by publication intent".to_owned(), + )); + } + guard.verified_local_recipe(&hash.into())?.ok_or_else(|| { + StagingError::StagingCorrupt( + "publication history recipe is not verified".to_owned(), + ) + }) + }) + .collect::>>()? + }; + // Commit every history owner before rollback can remove native leases. + // An interruption leaves the original intent and its remaining payloads owned. + for recipe in recipes { + if cancel.is_cancelled() { + return Err(StagingError::Cancelled); + } + let owner = PathBuf::from(".crab/history").join(recipe.file_hash().hex()); + publish_verified_recipe_owner(root, index, &owner, &recipe, None)?; + } + Ok(()) +} + #[cfg(unix)] fn staging_path_bytes(path: &Path) -> Vec { use std::os::unix::ffi::OsStrExt; @@ -2707,6 +2800,16 @@ impl StagingArea { unresolved_publication_intents(&self.index) } + /// Preserve caller-attested committed recipe roots before rolling back a native intent. + pub fn preserve_publication_history( + &self, + intent_id: &PublicationIntentId, + recipe_hashes: &[MerkleHash], + cancel: &tokio_util::sync::CancellationToken, + ) -> Result<()> { + preserve_publication_history(&self.root, &self.index, intent_id, recipe_hashes, cancel) + } + /// Atomically remove an unresolved intent and every open batch it owns. pub fn rollback_publication_intent( &self, @@ -2726,31 +2829,13 @@ impl StagingArea { path: &Path, recipe: &crate::recipe::FileRecipe, ) -> Result { - self.publish_verified_recipe_owner(path, recipe, Some(&path.to_string_lossy())) - } - - fn publish_verified_recipe_owner( - &self, - owner: &Path, - recipe: &crate::recipe::FileRecipe, - display_path: Option<&str>, - ) -> Result { - let batch_id = self.create_batch()?; - let file_hash = recipe.file_hash(); - let publish = (|| -> Result<()> { - self.pre_register_file(&file_hash, recipe.file_size())?; - self.record_verified_recipe_lease(&batch_id, owner, recipe)?; - if let Some(display_path) = display_path { - self.record_file_path(&file_hash, display_path)?; - } - self.mark_batch_published(&batch_id) - })(); - if let Err(error) = publish { - let _ = self.rollback_batch(&batch_id); - let _ = self.retire_file_if_unleased(&file_hash); - return Err(error); - } - Ok(batch_id) + publish_verified_recipe_owner( + &self.root, + &self.index, + path, + recipe, + Some(&path.to_string_lossy()), + ) } /// Publish one immutable recipe needed by not-yet-pushed Git history. @@ -2763,7 +2848,7 @@ impl StagingArea { recipe: &crate::recipe::FileRecipe, ) -> Result { let owner = PathBuf::from(".crab/history").join(recipe.file_hash().hex()); - self.publish_verified_recipe_owner(&owner, recipe, None) + publish_verified_recipe_owner(&self.root, &self.index, &owner, recipe, None) } /// Preserve a locally published recipe referenced by not-yet-pushed Git history. @@ -4134,6 +4219,16 @@ impl StagingAreaReadOnly { Ok(()) } + /// Preserve caller-attested committed recipe roots before rolling back a native intent. + pub fn preserve_publication_history( + &self, + intent_id: &PublicationIntentId, + recipe_hashes: &[MerkleHash], + cancel: &tokio_util::sync::CancellationToken, + ) -> Result<()> { + preserve_publication_history(&self.root, &self.index, intent_id, recipe_hashes, cancel) + } + /// Atomically remove an unresolved intent and every open batch it owns. pub async fn rollback_publication_intent( &self, @@ -4165,6 +4260,10 @@ impl StagingAreaReadOnly { /// Retire the exact recipe ownership pinned by one committed push snapshot. pub async fn retire_push_snapshot(&self, push_id: &str) -> Result> { + self.retire_push_snapshot_now(push_id) + } + + fn retire_push_snapshot_now(&self, push_id: &str) -> Result> { let unleased = lock_index(&self.index)?.retire_push_snapshot(push_id)?; remove_indexed_files(&self.root, &self.index, &unleased) } diff --git a/crates/crab-staging/src/push_snapshot.rs b/crates/crab-staging/src/push_snapshot.rs new file mode 100644 index 000000000..2437a4634 --- /dev/null +++ b/crates/crab-staging/src/push_snapshot.rs @@ -0,0 +1,70 @@ +use std::sync::Arc; + +use crate::{Result, StagingAreaReadOnly, recipe::FileRecipe}; + +/// Own one push reader's marker and exact immutable recipe snapshot. +#[must_use = "retain the snapshot through remote publication"] +pub struct StagingPushSnapshot { + staging: Arc, + push_id: String, + committed: bool, +} + +impl StagingPushSnapshot { + /// Fence retirement before selecting recipes for a uniquely identified push. + /// + /// The caller must supply a unique push ID. + /// Failure or drop preserves their original leases; another reader can defer cleanup. + pub fn begin(staging: Arc, push_id: String) -> Result { + let snapshot = Self { + staging, + push_id, + committed: false, + }; + snapshot.staging.mark_push_inflight(&snapshot.push_id)?; + Ok(snapshot) + } + + /// Pin published/delegated recipes selected from verified Git pointers. + pub fn capture(&self, recipes: &[FileRecipe]) -> Result<()> { + self.staging.create_push_snapshot(&self.push_id, recipes) + } + + /// Record confirmed remote publication and retire only captured local owners. + /// + /// The caller must establish remote commit before calling this method. Contention + /// leaves a committed journal for exclusive maintenance; cleanup errors retain it. + pub fn commit(mut self) -> Result<()> { + // Remote commit is already authoritative. Even if the local transition + // fails, Drop must retain its journal rather than treat it as a failed push. + self.committed = true; + self.staging.commit_push_snapshot(&self.push_id)?; + let Some(_retirement) = self.staging.begin_retirement(&self.push_id)? else { + return Ok(()); + }; + self.staging.retire_push_snapshot_now(&self.push_id)?; + self.staging.sweep_orphans()?; + self.staging.remove_push_snapshot(&self.push_id) + } +} + +impl Drop for StagingPushSnapshot { + fn drop(&mut self) { + if !self.committed { + let discard = (|| -> Result<()> { + // A sibling may still read a version this snapshot alone pins. + // Exclusive maintenance consumes the journal if cleanup is deferred. + if let Some(_retirement) = self.staging.begin_retirement(&self.push_id)? { + self.staging.discard_open_push_snapshot_now(&self.push_id)?; + } + Ok(()) + })(); + if let Err(error) = discard { + tracing::warn!(push_id = %self.push_id, %error, "staging snapshot cleanup needs maintenance"); + } + } + if let Err(error) = self.staging.clear_push_inflight(&self.push_id) { + tracing::warn!(push_id = %self.push_id, %error, "staging push marker cleanup failed"); + } + } +} diff --git a/crates/crab-staging/tests/publication_history_recovery.rs b/crates/crab-staging/tests/publication_history_recovery.rs new file mode 100644 index 000000000..7afe7de79 --- /dev/null +++ b/crates/crab-staging/tests/publication_history_recovery.rs @@ -0,0 +1,128 @@ +//! Committed history ownership must outlive rollback of its native index intent. + +#![expect(clippy::unwrap_used, reason = "test assertions")] + +use std::path::Path; +use std::sync::Arc; + +use crab_staging::stream::{StreamStageProgress, StreamStageXorbBuilder, stage_reader_streaming}; +use crab_staging::{ + PublicationIntentEntry, StagingArea, StagingAreaReadOnly, StagingError, StagingPushSnapshot, +}; +use crab_xet::hash::MerkleHash; +use tokio_util::sync::CancellationToken; + +#[tokio::test(flavor = "multi_thread")] +async fn committed_history_preserves_only_selected_intent_recipes_through_rollback() { + let temp = tempfile::tempdir().unwrap(); + let staging = StagingArea::open(temp.path().to_owned()).await.unwrap(); + let mut staged = Vec::new(); + for (path, byte) in [("committed", 0x71), ("uncommitted", 0x72)] { + staged.push( + stage_reader_streaming( + &mut std::io::Cursor::new(vec![byte; 2 * 1024 * 1024]), + Path::new(path), + &staging, + StreamStageProgress { + xorb_builder: Some(StreamStageXorbBuilder::new( + 1, + crab_xet::xorb::builder::XorbBuilder::new, + )), + ..Default::default() + }, + &CancellationToken::new(), + ) + .await + .unwrap(), + ); + } + let entries = staged + .iter() + .enumerate() + .map(|(index, staged)| PublicationIntentEntry { + batch_id: staged.batch_id.clone(), + path: if index == 0 { + "committed" + } else { + "uncommitted" + } + .into(), + expected_pointer_oid: format!("{}", index + 1).repeat(40), + previous_index_state: "0".repeat(64), + }) + .collect::>(); + let intent = staging.create_publication_intent(&entries).unwrap(); + let recipe_hash = MerkleHash::from(staged[0].recipe.hash()); + let invalid = staging.preserve_publication_history( + &intent, + &[recipe_hash, MerkleHash::from([0xff; 32])], + &CancellationToken::new(), + ); + assert!(invalid.is_err() && staging.lifecycle_health().unwrap().path_heads == 0); + let cancelled = CancellationToken::new(); + cancelled.cancel(); + assert!(matches!( + staging.preserve_publication_history(&intent, &[recipe_hash], &cancelled), + Err(StagingError::Cancelled) + )); + for _ in 0..2 { + staging + .preserve_publication_history(&intent, &[recipe_hash], &CancellationToken::new()) + .unwrap(); + } + let health = staging.lifecycle_health().unwrap(); + assert!( + health.path_heads == 1 + && health.path_leases == 3 + && health.payloads == 0 + && health.current_head_prepared_bytes > 0, + "{health:?}" + ); + staging.rollback_publication_intent(&intent).unwrap(); + assert!( + staging + .published_recipe_for_file(&staged[0].recipe.file_hash()) + .unwrap() + == Some(staged[0].recipe.clone()) + && staging + .chunks_for_file(&staged[0].recipe.file_hash()) + .unwrap() + .len() + == 16 + && staging + .chunks_for_file(&staged[1].recipe.file_hash()) + .unwrap() + .is_empty() + ); + staging.close().await.unwrap(); + let reader = Arc::new( + StagingAreaReadOnly::open(temp.path().to_owned()) + .await + .unwrap(), + ); + reader + .verify_file_reconstruction(&staged[0].recipe.file_hash(), staged[0].size) + .await + .unwrap(); + let prepared = reader.prepared_xorbs_for_recipes(&[recipe_hash]).unwrap(); + let bodies = prepared + .values() + .flatten() + .map(|xorb| { + crab_staging::push_plan::prepared_xorb_path( + temp.path(), + &MerkleHash::from_hex(&xorb.hash).unwrap(), + ) + }) + .collect::>(); + assert!(!bodies.is_empty() && bodies.iter().all(|path| path.is_file())); + let snapshot = + StagingPushSnapshot::begin(Arc::clone(&reader), "history-push".to_owned()).unwrap(); + snapshot + .capture(std::slice::from_ref(&staged[0].recipe)) + .unwrap(); + snapshot.commit().unwrap(); + let health = reader.lifecycle_health().unwrap(); + assert!(health.path_heads == 0 && health.path_leases == 0 && health.payloads == 0); + assert!(bodies.iter().all(|path| !path.try_exists().unwrap())); +} diff --git a/crates/crab-staging/tests/push_snapshot_guard.rs b/crates/crab-staging/tests/push_snapshot_guard.rs new file mode 100644 index 000000000..909622b41 --- /dev/null +++ b/crates/crab-staging/tests/push_snapshot_guard.rs @@ -0,0 +1,174 @@ +//! Reader fences and exact lease cleanup across push completion and cancellation. + +#![expect(clippy::unwrap_used, reason = "test assertions")] + +use std::path::Path; +use std::sync::Arc; + +use crab_staging::stream::{StreamStageProgress, stage_reader_streaming}; +use crab_staging::{StagingArea, StagingAreaReadOnly, StagingPushSnapshot}; +use tokio_util::sync::CancellationToken; + +async fn fixture( + root: &Path, + native_owner: bool, +) -> ( + Arc, + crab_staging::stream::StreamStageResult, + Option, +) { + let staging = StagingArea::open(root.to_owned()).await.unwrap(); + let staged = stage_reader_streaming( + &mut std::io::Cursor::new(b"shared fixture payload"), + Path::new("model.bin"), + &staging, + StreamStageProgress::default(), + &CancellationToken::new(), + ) + .await + .unwrap(); + staging + .delegate_publication(&[(&staged.batch_id, Path::new("model.bin"))]) + .unwrap(); + let native = if native_owner { + let batch = staging.create_batch().unwrap(); + staging + .record_verified_recipe_lease(&batch, Path::new("native.bin"), &staged.recipe) + .unwrap(); + Some(batch) + } else { + None + }; + staging.close().await.unwrap(); + ( + Arc::new(StagingAreaReadOnly::open(root.to_owned()).await.unwrap()), + staged, + native, + ) +} + +#[tokio::test(flavor = "multi_thread")] +async fn dropped_snapshot_keeps_delegated_payload_and_clears_its_reader() { + let temp = tempfile::tempdir().unwrap(); + let (staging, staged, _) = fixture(temp.path(), false).await; + let snapshot = StagingPushSnapshot::begin(Arc::clone(&staging), "failed".to_owned()).unwrap(); + snapshot + .capture(std::slice::from_ref(&staged.recipe)) + .unwrap(); + drop(snapshot); + let health = staging.lifecycle_health().unwrap(); + assert!( + health.open_push_snapshots == 0 + && health.path_leases == 1 + && health.payloads == 1 + && staging.list_inflight().unwrap().is_empty() + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn committed_snapshot_preserves_active_native_owner_of_shared_payload() { + let temp = tempfile::tempdir().unwrap(); + let (staging, staged, native) = fixture(temp.path(), true).await; + let snapshot = + StagingPushSnapshot::begin(Arc::clone(&staging), "committed".to_owned()).unwrap(); + snapshot + .capture(std::slice::from_ref(&staged.recipe)) + .unwrap(); + snapshot.commit().unwrap(); + let health = staging.lifecycle_health().unwrap(); + assert!( + health.path_leases == 1 + && health.payloads == 1 + && health.open_push_snapshots == 0 + && health.committed_push_snapshots == 0 + && staging.list_inflight().unwrap().is_empty() + ); + drop(staging); + let writer = StagingArea::open(temp.path().to_owned()).await.unwrap(); + writer.rollback_batch(&native.unwrap()).unwrap(); + assert!( + writer + .chunks_for_file(&staged.recipe.file_hash()) + .unwrap() + .is_empty() + ); + writer.close().await.unwrap(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn sibling_reader_defers_retirement_until_exclusive_maintenance() { + let temp = tempfile::tempdir().unwrap(); + let (staging, staged, _) = fixture(temp.path(), false).await; + let first = StagingPushSnapshot::begin(Arc::clone(&staging), "first".to_owned()).unwrap(); + first.capture(std::slice::from_ref(&staged.recipe)).unwrap(); + let second = StagingPushSnapshot::begin(Arc::clone(&staging), "second".to_owned()).unwrap(); + second + .capture(std::slice::from_ref(&staged.recipe)) + .unwrap(); + first.commit().unwrap(); + let health = staging.lifecycle_health().unwrap(); + assert!( + health.committed_push_snapshots == 1 + && health.open_push_snapshots == 1 + && health.payloads == 1 + ); + drop(second); + drop(staging); + let writer = StagingArea::open(temp.path().to_owned()).await.unwrap(); + writer.clean().unwrap(); + let health = writer.lifecycle_health().unwrap(); + assert!( + health.path_leases == 0 + && health.payloads == 0 + && health.open_push_snapshots == 0 + && health.committed_push_snapshots == 0 + && writer.list_inflight().unwrap().is_empty() + ); + writer.close().await.unwrap(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn task_abort_drops_snapshot_without_retiring_unpublished_lease() { + let temp = tempfile::tempdir().unwrap(); + let (staging, staged, _) = fixture(temp.path(), false).await; + let reader = Arc::clone(&staging); + let (ready, received) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(async move { + let snapshot = StagingPushSnapshot::begin(reader, "aborted".to_owned()).unwrap(); + snapshot + .capture(std::slice::from_ref(&staged.recipe)) + .unwrap(); + ready.send(()).unwrap(); + std::future::pending::<()>().await; + drop(snapshot); + }); + received.await.unwrap(); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + let health = staging.lifecycle_health().unwrap(); + assert!( + health.path_leases == 1 + && health.open_push_snapshots == 0 + && health.payloads == 1 + && staging.list_inflight().unwrap().is_empty() + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn local_commit_failure_releases_marker_and_preserves_original_lease() { + let temp = tempfile::tempdir().unwrap(); + let (staging, staged, _) = fixture(temp.path(), false).await; + let snapshot = + StagingPushSnapshot::begin(Arc::clone(&staging), "missing-journal".to_owned()).unwrap(); + snapshot + .capture(std::slice::from_ref(&staged.recipe)) + .unwrap(); + staging.remove_push_snapshot("missing-journal").unwrap(); + assert!(snapshot.commit().is_err()); + let health = staging.lifecycle_health().unwrap(); + assert!( + health.path_leases == 1 + && health.payloads == 1 + && staging.list_inflight().unwrap().is_empty() + ); +} From 19c88b54e99df8806ff6b22633514e34f30219c1 Mon Sep 17 00:00:00 2001 From: forhappy Date: Wed, 14 Oct 2026 19:32:57 -0700 Subject: [PATCH 2/3] ci: qualify capsule staging lifecycle --- .../workflows/git-protocol-v2-partial-clone.yml | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/.github/workflows/git-protocol-v2-partial-clone.yml b/.github/workflows/git-protocol-v2-partial-clone.yml index 827223172..554748afd 100644 --- a/.github/workflows/git-protocol-v2-partial-clone.yml +++ b/.github/workflows/git-protocol-v2-partial-clone.yml @@ -45,6 +45,8 @@ on: - "crab/tests/**" - "crates/crab-git/**" - "crates/crab-auth-server/**" + - "crates/crab-auth/**" + - "crates/crab-staging/**" - "crates/crab-metadata/**" - "crates/crab-read/**" - "crates/crab-remote-git/**" @@ -99,6 +101,8 @@ on: - "crab/tests/**" - "crates/crab-git/**" - "crates/crab-auth-server/**" + - "crates/crab-auth/**" + - "crates/crab-staging/**" - "crates/crab-metadata/**" - "crates/crab-read/**" - "crates/crab-remote-git/**" @@ -176,6 +180,11 @@ jobs: cargo test -p crab --bin crab pull_machine_errors_keep_the_selected_mode_and_schema --locked --features gix-transport cargo test -p crab --lib git::push_native::tests --locked --features gix-transport cargo test -p crab --lib git::push_staging::tests --locked --features gix-transport + cargo test -p crab-staging --locked --test push_snapshot_guard --test publication_history_recovery + cargo test -p crab-auth --lib protected_push --locked --features crab-auth-client + cargo test -p crab --lib git::index::tests --locked --features gix-transport + cargo test -p crab --lib publication_recovery --locked --features gix-transport + cargo test -p crab --lib git::capsule_push::tests --locked --features gix-transport cargo test -p crab --lib --locked --features gix-transport -- \ git::push::tests::lookup_staging \ git::push::tests::under_lock_refresh \ @@ -269,6 +278,11 @@ jobs: cargo test -p crab --lib cmd::lfs::install::tests --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport cargo test -p crab --lib git::push_native::tests --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport cargo test -p crab --lib git::push_staging::tests --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport + cargo test -p crab-staging --locked --test push_snapshot_guard --test publication_history_recovery + cargo test -p crab-auth --lib protected_push --locked --features crab-auth-client + cargo test -p crab --lib git::index::tests --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport + cargo test -p crab --lib publication_recovery --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport + cargo test -p crab --lib capsule_push_retires_staging_only_after_confirmed_ref_publication --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport cargo test -p crab --lib git::remote_helper::tests::list_session_does_not_open_a_damaged_staging_index --locked --no-default-features --features simd-accel,tier,watch,nfs,gix-pathmatch,gix-transport - name: Run upload-pack session contracts From 7abb53e0b85b6d86a3f5708e877fc1b586b29c57 Mon Sep 17 00:00:00 2001 From: forhappy Date: Wed, 14 Oct 2026 20:16:00 -0700 Subject: [PATCH 3/3] test: verify root faults and both add orders --- .../docs/architecture/indexed-remote-reads.md | 59 ++++++++++++ .../e2e/run_add_commit_push_rustfs_smoke.py | 37 +++++--- .../e2e/run_large_file_edits_rustfs.py | 19 +++- crab/src/git/index.rs | 94 ++++++++++--------- 4 files changed, 150 insertions(+), 59 deletions(-) diff --git a/crab/docs/architecture/indexed-remote-reads.md b/crab/docs/architecture/indexed-remote-reads.md index b99f91da3..1968be230 100644 --- a/crab/docs/architecture/indexed-remote-reads.md +++ b/crab/docs/architecture/indexed-remote-reads.md @@ -2040,6 +2040,15 @@ walk fails without authorizing rollback. Ordinary pushes without unresolved intents do not perform this history scan; this does not replace the historical remote-catalog indexing workstream. +The minimum Git 2.30 dependency contract supports the selected walk flags and +includes other worktrees with `--all --reflog` by default +([revision options](https://github.com/git/git/blob/v2.30.0/Documentation/rev-list-options.txt)). +Its transport allowlist takes precedence over configured transport permissions +([transport implementation](https://github.com/git/git/blob/v2.30.0/transport.c)); +the empty allowlist prevents an implicit fetch during local recovery. Windows +startup sets stdout to binary mode, matching the OID parser's LF delimiter +([MinGW startup](https://github.com/git/git/blob/v2.30.0/compat/mingw.c)). + Both exclusive native-add recovery and shared push recovery use this ownership rule. The shared staging helper validates all requested recipes against the intent before adding history owners, preserves the original intent across an @@ -2071,3 +2080,53 @@ default features. Its binary hash is recorded externally. Resumed workload subsequent v2 native push, with 2,216,109 and 2,133,870 incremental xorb bytes. The full workload is still running; cold historical reconstruction is unproven. This artifact precedes the lifecycle and finalize-contract changes above. + + +The 10-GiB harness also selects `--first-ingestion native|git`, covering both +native→clean→native and clean→native→clean. Receipts include the selected order +and each add's ingestion path. After each add, the harness rejects local +segment/prepared storage (including temporary stream bodies) beyond one initial +representation, cumulative edited +bytes and metadata allowance (10% of source size, at least 32 MiB). This detects +the observed extra full 10-GiB representation; it is not a latency target or a +claim that all allowed overhead is necessary. Existing runs retain their +original harness identity and cannot prove these newly added checks. + +The local history dependency probe additionally reproduces exit zero with a +missing reflog commit and an omitted expected pointer. Git reports that partial +walk only through diagnostics; its +[reflog traversal implementation](https://github.com/git/git/blob/v2.30.0/revision.c) +confirms that contract. Recovery therefore requires a diagnostic-free successful +walk before accepting absence. A warning retains the unresolved intent and its +payloads; there is no log-message matching or alternate deletion path. Both a +missing reflog tip and a missing referenced tree are regression cases. Earlier +consumer builds do not qualify this tightened scanner until rerun. + +An extracted-source probe compiles the exact helper and canonical subprocess +owner against the frozen release's actual error/dependency types. The old helper +accepts the incomplete walk (negative runtime exits 1); the tightened helper +preserves the healthy reflog result and rejects the missing tip (positive runtime +exits 0). Source hashes are recorded with both probes. This is function-level +runtime proof; the full updated consumer still needs its own Cargo/CLI run. + + +Reduced crossover qualification passes with the frozen #570 release: Git-first +repeated data and native-first zero data, each with two 128-MiB files, four +versions, source removal, eight cold file digests and remote fsck. Each run records +71 commands and 53 passing checks. An older debug binary fails the new storage +assertion after its first native-to-clean switch: 67,156,162 prepared bytes plus +68,224,532 segment bytes exceed the 102,760,448-byte bound for its 64-MiB input. +These runs qualify the order/storage oracle; they do not qualify the final +lifecycle source or replace the full 10-GiB matrix. The later inclusion of +stream-temporary bytes in the accounting remains part of that final matrix. + +The dedicated #572 provider run passes its provider contract and the initial +native cutover push's retirement check, then stops at the old missing-manifest +fixture. It deleted a retired key while `v2/root` remained authoritative. Missing +and corrupt fixtures now initialize a healthy repository, attest the live root, +and verify their deletion/overwrite took effect before requiring list/clone to +fail. Corruption still requires the integrity error; expectations were not +relaxed. Both updated fixtures pass against the unchanged #570 root consumer +(23 commands, 23 checks), and the 11 source-isolation tests pass. The explicit +v1 upgrade/reset fixtures remain separate. Full final-source canary proof is +still pending. diff --git a/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py b/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py index 50a70f816..17f2f63d9 100755 --- a/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py +++ b/crab/scripts/e2e/run_add_commit_push_rustfs_smoke.py @@ -8,7 +8,7 @@ The large-file cases prove pointer staging, object-store publication, fresh clone, hydration, and byte identity. A separate ordinary-Git matrix proves -missing/corrupt manifest handling, update/fetch, deletion, force and +missing/corrupt metadata-root handling, update/fetch, deletion, force and non-fast-forward behavior, atomic rejection, tag following, shallow history, immutable-pack failures, and concurrent push serialization. @@ -850,19 +850,25 @@ def clone_git( check=check, ) - def run_missing_manifest_case(self) -> None: - case_name = "missing-manifest" + def run_missing_root_case(self) -> None: + case_name = "missing-root" _repo, remote_url, repo_prefix = self.prepare_git_repo(case_name) + root_key = f"{repo_prefix}/v2/root" + self.ls_remote(remote_url, name=f"{case_name} healthy list") + self.head_key(root_key) self.run_aws( - "delete canonical manifest fixture", + "delete canonical root fixture", [ "delete-object", "--bucket", self.args.bucket, "--key", - f"{repo_prefix}/manifest", + root_key, ], ) + status, _, _ = self.signed_s3_request("HEAD", root_key) + self.check(f"{case_name}-fixture-removed", status == 404, + {"key": root_key, "status": status}) listing = self.run_git( self.run_root, ["ls-remote", remote_url], @@ -1174,13 +1180,15 @@ def run_shallow_case(self) -> None: {"commit_count": full_depth}, ) - def run_corrupt_manifest_case(self) -> None: - case_name = "corrupt-manifest" - remote_url, repo_prefix = self.remote_for_case(case_name) - manifest_key = f"{repo_prefix}/manifest" + def run_corrupt_root_case(self) -> None: + case_name = "corrupt-root" + _repo, remote_url, repo_prefix = self.prepare_git_repo(case_name) + root_key = f"{repo_prefix}/v2/root" + self.ls_remote(remote_url, name=f"{case_name} healthy list") + self.head_key(root_key) status, _, response_body = self.signed_s3_request( "PUT", - manifest_key, + root_key, body=b'{"version":', ) self.check( @@ -1193,6 +1201,11 @@ def run_corrupt_manifest_case(self) -> None: }, ) + status, _, body = self.signed_s3_request("GET", root_key) + self.check(f"{case_name}-fixture-overwritten", + status == 200 and body == b'{"version":', + {"key": root_key, "status": status}) + ls_remote = self.run_git( self.run_root, ["ls-remote", remote_url], @@ -2207,8 +2220,8 @@ def run(self) -> None: self.write_report() return self.run_v1_hard_cutover_reset_case() - self.run_missing_manifest_case() - self.run_corrupt_manifest_case() + self.run_missing_root_case() + self.run_corrupt_root_case() self.run_ref_update_delete_force_and_tag_case() self.run_atomic_rejection_case() self.run_shallow_case() diff --git a/crab/scripts/e2e/run_large_file_edits_rustfs.py b/crab/scripts/e2e/run_large_file_edits_rustfs.py index c81cf662c..4b24ea36d 100644 --- a/crab/scripts/e2e/run_large_file_edits_rustfs.py +++ b/crab/scripts/e2e/run_large_file_edits_rustfs.py @@ -3,6 +3,7 @@ Two copy-on-write files start with identical, non-sparse deterministic content. Choose entropy, a repeated one-MiB block, or zero-filled bytes. +Choose native-first or Git-first alternating ingestion. Edits diverge at the beginning, middle and end. Each committed version must hydrate byte-identically from a fresh cache after the source repository is removed. Only this invocation's disposable data is released; logs and receipts remain. @@ -96,13 +97,15 @@ def qualify(runner: AddCommitPushSmoke, proxy: RequestCountingProxy) -> None: (code / f"module-{index:05}.rs").write_text(f"pub const VALUE: u64 = {index};\n") runner.check("workload", all(path.stat().st_size == size for path in paths), {"file_bytes": size, "files": 2, "initial_unique_bytes": size, - "versions": 4, "code_files": args.code_files, "pattern": args.pattern}) + "versions": 4, "code_files": args.code_files, "pattern": args.pattern, + "first_ingestion": args.first_ingestion}) history = [] phases = [] def record() -> None: receipt = runner.artifacts / "large-file-edits.json" receipt.write_text(json.dumps({"history": history, "phases": phases, "pattern": args.pattern, + "first_ingestion": args.first_ingestion, "transport": proxy.snapshot()}, indent=2) + "\n") runner.report.artifacts["large_file_edits"] = str(receipt) runner.write_report() @@ -116,7 +119,8 @@ def record() -> None: stream.write(hashlib.shake_256(f"edit:{version}:{index}".encode()).digest(MIB)) (code / "module-00000.rs").write_text(f"pub const VERSION: u64 = {version};\n") before = proxy.snapshot() - if version % 2: + native_add = (version % 2 == 0) == (args.first_ingestion == "native") + if not native_add: add = runner.run_git(repo, ["add", "--", *[p.name for p in paths]], name=f"v{version} Git clean-filter add") else: @@ -124,9 +128,17 @@ def record() -> None: staging = repo / ".crab" / "staging" local_bytes = { name: sum(path.stat().st_size for path in (staging / name).rglob("*") if path.is_file()) - for name in ("segments", "push-plans") + for name in ("segments", "push-plans", "stream-prepared-xorbs") } + # One initial representation plus all edited bytes is sufficient. Allow + # encoded metadata overhead, while rejecting the observed second full copy. + max_local_bytes = size + 2 * version * MIB + max(size // 10, 32 * MIB) + runner.check(f"v{version}-no-second-full-local-payload", + sum(local_bytes.values()) <= max_local_bytes, + {"staging_directory_bytes": local_bytes, + "max_bytes": max_local_bytes}) phases.append({"version": version, "phase": "add", "duration_ms": add.duration_ms, + "ingestion": "native" if native_add else "git", "staging_directory_bytes": local_bytes, "transport": RequestCountingProxy.delta(before, proxy.snapshot())}) for path in paths: @@ -202,6 +214,7 @@ def main() -> None: parser.add_argument("--file-mib", type=int, default=10240) parser.add_argument("--code-files", type=int, default=5000) parser.add_argument("--pattern", choices=("entropy", "repeated", "zero"), default="entropy") + parser.add_argument("--first-ingestion", choices=("native", "git"), default="native") args = parser.parse_args() if args.file_mib < 16 or args.code_files < 1 or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", args.run_id): parser.error("require >=16 MiB/file, positive code files and a safe run id") diff --git a/crab/src/git/index.rs b/crab/src/git/index.rs index d3c80ac24..4af4fed25 100644 --- a/crab/src/git/index.rs +++ b/crab/src/git/index.rs @@ -372,9 +372,11 @@ fn committed_pointer_oids( Ok(found) }, )?; - if !output.status.success() { + // Git can exit zero while skipping unavailable reflog tips. A diagnostic + // means this walk cannot attest absence and authorize payload deletion. + if !output.status.success() || !output.stderr.is_empty() { return Err(CrabError::Internal(format!( - "Git recovery history scan failed ({}): {}", + "Git recovery history scan did not complete cleanly ({}): {}", output.status, String::from_utf8_lossy(&output.stderr).trim() ))); @@ -390,48 +392,52 @@ mod tests { #[test] fn recovery_history_includes_reflogs_and_rejects_incomplete_walks() { let _env = crate::test::git_repo::CleanGitEnvGuard::new(); - let temp = tempfile::tempdir().unwrap(); - let git = |args: &[&str]| { - let output = Command::new("git") - .args(args) - .current_dir(temp.path()) - .output() - .unwrap(); - assert!(output.status.success(), "{:?}", output.stderr); - String::from_utf8(output.stdout).unwrap().trim().to_owned() - }; - git(&["init", "-q", "-b", "main"]); - git(&["config", "user.name", "Crab Test"]); - git(&["config", "user.email", "crab@example.invalid"]); - std::fs::write(temp.path().join("file"), b"initial").unwrap(); - git(&["add", "file"]); - git(&["commit", "-qm", "initial"]); - let initial = git(&["rev-parse", "HEAD"]); - std::fs::write(temp.path().join("file"), b"committed pointer").unwrap(); - git(&["add", "file"]); - git(&["commit", "-qm", "pointer"]); - let pointer = git(&["rev-parse", "HEAD:file"]); - let tree = git(&["rev-parse", "HEAD^{tree}"]); - git(&["reset", "--hard", &initial]); - let wanted = HashSet::from([pointer]); - let cancel = CancellationToken::new(); - assert_eq!( - committed_pointer_oids(temp.path(), &wanted, &cancel).unwrap(), - wanted - ); - cancel.cancel(); - assert!(matches!( - committed_pointer_oids(temp.path(), &wanted, &cancel), - Err(CrabError::Cancelled) - )); - std::fs::remove_file( - temp.path() - .join(".git/objects") - .join(&tree[..2]) - .join(&tree[2..]), - ) - .unwrap(); - assert!(committed_pointer_oids(temp.path(), &wanted, &CancellationToken::new()).is_err()); + for revision in ["HEAD^{tree}", "HEAD"] { + let temp = tempfile::tempdir().unwrap(); + let git = |args: &[&str]| { + let output = Command::new("git") + .args(args) + .current_dir(temp.path()) + .output() + .unwrap(); + assert!(output.status.success(), "{:?}", output.stderr); + String::from_utf8(output.stdout).unwrap().trim().to_owned() + }; + git(&["init", "-q", "-b", "main"]); + git(&["config", "user.name", "Crab Test"]); + git(&["config", "user.email", "crab@example.invalid"]); + std::fs::write(temp.path().join("file"), b"initial").unwrap(); + git(&["add", "file"]); + git(&["commit", "-qm", "initial"]); + let initial = git(&["rev-parse", "HEAD"]); + std::fs::write(temp.path().join("file"), b"committed pointer").unwrap(); + git(&["add", "file"]); + git(&["commit", "-qm", "pointer"]); + let pointer = git(&["rev-parse", "HEAD:file"]); + let missing = git(&["rev-parse", revision]); + git(&["reset", "--hard", &initial]); + let wanted = HashSet::from([pointer]); + let cancel = CancellationToken::new(); + assert_eq!( + committed_pointer_oids(temp.path(), &wanted, &cancel).unwrap(), + wanted + ); + cancel.cancel(); + assert!(matches!( + committed_pointer_oids(temp.path(), &wanted, &cancel), + Err(CrabError::Cancelled) + )); + std::fs::remove_file( + temp.path() + .join(".git/objects") + .join(&missing[..2]) + .join(&missing[2..]), + ) + .unwrap(); + assert!( + committed_pointer_oids(temp.path(), &wanted, &CancellationToken::new()).is_err() + ); + } } #[test]