diff --git a/crates/walgit-store/src/fault.rs b/crates/walgit-store/src/fault.rs index a3e95742..68469ca1 100644 --- a/crates/walgit-store/src/fault.rs +++ b/crates/walgit-store/src/fault.rs @@ -19,6 +19,7 @@ //! Fault kinds (see [`FaultPlan`]): latency, transient error *before* the op //! (nothing applied), transient error *after* a mutation (applied, response //! lost — the one that breaks "PUT then CAS" protocols), spurious CAS failure, +//! a CAS answered 412 after it applied (a client library retrying a write that landed), //! stale 304, truncated bodies, hanging futures, black hole, missing objects, //! and panics (a process crash in the middle of a protocol step). //! @@ -58,6 +59,10 @@ pub struct FaultPlan { pub p_err_after: f64, /// Conditional PUT/DELETE answers `PreconditionFailed` without applying. pub p_cas_fail: f64, + /// Conditional mutation applied, then `PreconditionFailed` returned: what a client library's + /// own retry of a write that landed (first reply lost) answers. S3's SDK retries in standard + /// mode; GCS retries uploads that carry a precondition as idempotent. + pub p_cas_fail_after: f64, /// `get` with `if_none_match` answers `NotModified` regardless of the /// real version (a replica that never sees anyone else's writes). pub p_stale_304: f64, @@ -121,6 +126,7 @@ pub struct Stats { pub err_before: AtomicU64, pub err_after: AtomicU64, pub cas_fail: AtomicU64, + pub cas_fail_after: AtomicU64, pub stale_304: AtomicU64, pub truncate: AtomicU64, pub hang: AtomicU64, @@ -133,6 +139,7 @@ impl Stats { self.err_before.load(Ordering::Relaxed) + self.err_after.load(Ordering::Relaxed) + self.cas_fail.load(Ordering::Relaxed) + + self.cas_fail_after.load(Ordering::Relaxed) + self.stale_304.load(Ordering::Relaxed) + self.truncate.load(Ordering::Relaxed) + self.hang.load(Ordering::Relaxed) @@ -141,11 +148,12 @@ impl Stats { } pub fn summary(&self) -> String { format!( - "ops={} err_before={} err_after={} cas_fail={} stale_304={} truncate={} hang={} denied={} panics={}", + "ops={} err_before={} err_after={} cas_fail={} cas_fail_after={} stale_304={} truncate={} hang={} denied={} panics={}", self.ops.load(Ordering::Relaxed), self.err_before.load(Ordering::Relaxed), self.err_after.load(Ordering::Relaxed), self.cas_fail.load(Ordering::Relaxed), + self.cas_fail_after.load(Ordering::Relaxed), self.stale_304.load(Ordering::Relaxed), self.truncate.load(Ordering::Relaxed), self.hang.load(Ordering::Relaxed), @@ -214,6 +222,7 @@ enum Decision { ErrBefore, ErrAfter, CasFail, + CasFailAfter, Stale, Truncate(usize), Hang, @@ -375,6 +384,10 @@ impl FaultStore { } else if mutation && conditional && roll[2] < plan.p_cas_fail { self.stats.cas_fail.fetch_add(1, Ordering::Relaxed); Decision::CasFail + } else if mutation && conditional && roll[2] < plan.p_cas_fail + plan.p_cas_fail_after { + // Shares the CAS roll: a plan without this fault keeps every seeded sequence. + self.stats.cas_fail_after.fetch_add(1, Ordering::Relaxed); + Decision::CasFailAfter } else if mutation && roll[3] < plan.p_err_after { self.stats.err_after.fetch_add(1, Ordering::Relaxed); Decision::ErrAfter @@ -393,6 +406,7 @@ impl FaultStore { Decision::ErrBefore => "err-before", Decision::CasFail => "cas-fail", Decision::ErrAfter => "err-after", + Decision::CasFailAfter => "cas-fail-after (applied)", Decision::Stale => "stale-304", Decision::Truncate(_) => "truncate", _ => "?", @@ -493,6 +507,13 @@ impl ObjectStore for FaultStore { let _ = self.inner.put(key, body, opts).await?; Err(self.retryable("put", key, "after (applied)")) } + Decision::CasFailAfter => { + let _ = self.inner.put(key, body, opts).await?; + Err(StoreError::PreconditionFailed { + key: key.into(), + current: None, + }) + } _ => self.inner.put(key, body, opts).await, } } @@ -510,6 +531,13 @@ impl ObjectStore for FaultStore { self.inner.delete(key, if_version).await?; Err(self.retryable("delete", key, "after (applied)")) } + Decision::CasFailAfter => { + self.inner.delete(key, if_version).await?; + Err(StoreError::PreconditionFailed { + key: key.into(), + current: None, + }) + } _ => self.inner.delete(key, if_version).await, } } @@ -572,6 +600,13 @@ impl ObjectStore for FaultStore { let _ = self.inner.compose(dest, sources, opts).await?; Err(self.retryable("compose", dest, "after (applied)")) } + Decision::CasFailAfter => { + let _ = self.inner.compose(dest, sources, opts).await?; + Err(StoreError::PreconditionFailed { + key: dest.into(), + current: None, + }) + } _ => self.inner.compose(dest, sources, opts).await, } } @@ -623,7 +658,7 @@ impl FaultStore { } r @ GetResult::NotModified { .. } => Ok(r), }, - Decision::Proceed | Decision::ErrAfter | Decision::CasFail => { + Decision::Proceed | Decision::ErrAfter | Decision::CasFail | Decision::CasFailAfter => { self.inner.get(key, opts).await } } diff --git a/crates/walgit-wal/src/publish.rs b/crates/walgit-wal/src/publish.rs index 07a23bda..764d141d 100644 --- a/crates/walgit-wal/src/publish.rs +++ b/crates/walgit-wal/src/publish.rs @@ -748,7 +748,26 @@ async fn process_batch(handle: &RepoHandle, batch: Vec) -> Resul // lost the response. Look before deciding. let committed: Option<(Manifest, walgit_store::Version)> = match cas { Ok(meta) => Some((updated, meta.version)), - Err(StoreError::PreconditionFailed { .. }) => None, + // A client library that retries a CAS whose first reply was lost gets 412 from our own + // landed write (S3's SDK in standard mode; GCS for uploads with a precondition). Look + // before dropping the segment the committed manifest may already list. + Err(StoreError::PreconditionFailed { .. }) => { + match cas_landed(&handle.store, &slot) + .instrument(span.clone()) + .await + { + Ok(Some((fresh, version))) => { + tracing::warn!(repo = %handle.id, seq = last_seq, "manifest CAS answered 412 but landed"); + Some((fresh, version)) + } + Ok(None) => None, + Err(e) => { + let msg = + format!("manifest CAS answered 412 and re-reading it failed: {e}"); + return finish_with_error_msg(batch, &valid_indices, &msg, e); + } + } + } Err(e) => match cas_landed(&handle.store, &slot) .instrument(span.clone()) .await @@ -1180,7 +1199,17 @@ pub(crate) async fn publish_compact_classified( .await; let committed = match cas { Ok(meta) => Some((updated, meta.version)), - Err(StoreError::PreconditionFailed { .. }) => None, + // A 412 can be our own landed write answering a client library's retry (see + // `process_batch`): look before dropping the segment. + Err(StoreError::PreconditionFailed { .. }) => { + match cas_landed(&handle.store, &slot).await? { + Some((fresh, version)) => { + tracing::warn!(repo = %handle.id, seq, "compact manifest CAS answered 412 but landed"); + Some((fresh, version)) + } + None => None, + } + } Err(e) => match cas_landed(&handle.store, &slot).await { Ok(Some((fresh, version))) => { tracing::warn!(repo = %handle.id, seq, "compact manifest CAS errored but landed: {e}"); @@ -1501,6 +1530,14 @@ pub(crate) async fn publish_settings_impl( return Ok(revision); } Err(StoreError::PreconditionFailed { .. }) => { + // A 412 can be our own landed write answering a client library's retry (see + // `process_batch`). Committed: leave the segment and let the next sync adopt it. + if cas_landed(&handle.store, &slot).await?.is_some() { + handle.manifest_version.lock().take(); + sweep_burned(&handle.store, &slot).await; + tracing::warn!(repo = %handle.id, seq, revision, "settings CAS answered 412 but landed"); + return Ok(revision); + } drop_own_slot(&handle.store, &slot).await; attempts += 1; if attempts >= max_retries { diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index ce85399e..f4a77cd6 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -2910,6 +2910,122 @@ async fn lost_cas_reply_is_resolved_or_unknown_without_losing_the_commit() { assert_eq!(handle.read_log(1, None).await.unwrap().len(), 3); } +/// A client library that retries a conditional write whose first reply was lost gets 412 from its +/// own landed write. No publisher may read that as a lost race and delete the segment the committed +/// manifest now lists: not a push, a compaction or a settings change. +#[tokio::test] +async fn a_landed_cas_answered_412_keeps_the_commit() { + use prost::Message; + use walgit_store::ObjectStoreExt; + use walgit_store::fault::{FaultPlan, FaultStore}; + let landed_412 = || FaultPlan { + p_cas_fail_after: 1.0, + only_keys: Some(vec!["manifest.pb".into()]), + ..Default::default() + }; + let cache = tempfile::tempdir().unwrap(); + let truth = MemoryStore::shared(); + let link = FaultStore::new(truth.clone(), "landed-412", 1); + let registry = Registry::new(link.clone(), Arc::new(make_config(cache.path(), 0))); + let id = repo_id("o", "landed"); + let handle = registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + // Every segment the manifest lists exists, and a fresh instance syncs to `tip`. + let intact = |what: &'static str, tip: String| { + let truth = truth.clone(); + let id = id.clone(); + async move { + let mkey = format!("{}{}", id.store_prefix(), walgit_proto::keys::MANIFEST); + let (_, bytes) = truth.get_bytes(&mkey).await.unwrap().unwrap(); + let manifest = walgit_proto::v1::Manifest::decode(bytes.as_ref()).unwrap(); + for segment in &manifest.log_segments { + let key = format!("{}{}", id.store_prefix(), segment.key); + assert!( + truth.get_bytes(&key).await.unwrap().is_some(), + "{what}: manifest lists {} but it is gone", + segment.key + ); + } + let reader_cache = tempfile::tempdir().unwrap(); + let reader = Registry::new(truth, Arc::new(make_config(reader_cache.path(), 0))) + .open(&id) + .await + .unwrap(); + drop(reader.sync_refs().await.unwrap()); + assert_eq!( + reader.local().ref_view().unwrap().get("refs/heads/main"), + Some(tip), + "{what}" + ); + } + }; + + let work = WorkRepo::new(); + let first = work.commit("first", "content"); + let pack = ingest_pack_data(&handle, work.create_pack()).await.unwrap(); + link.set(landed_412()); + let pushed = handle + .publish_push( + Some(pack), + make_txn(vec![("refs/heads/main", "", &first)]), + HashMap::new(), + ) + .await; + link.heal(); + intact("push", first.clone()).await; + assert!( + pushed + .unwrap() + .per_ref + .iter() + .all(|(_, status)| status.is_ok()), + "a durable push must be acknowledged" + ); + + let second = work.commit("second", "more"); + let pack = ingest_pack_data(&handle, work.create_incremental_pack(&second, &first)) + .await + .unwrap(); + handle + .publish_push( + Some(pack), + make_txn(vec![("refs/heads/main", &first, &second)]), + HashMap::new(), + ) + .await + .unwrap(); + let repack = handle + .local() + .repack(walgit_git::RepackOptions { + mode: walgit_git::RepackMode::Full, + write_bitmap: false, + write_midx: false, + keep: vec![], + }) + .await + .unwrap(); + link.set(landed_412()); + let compacted = handle + .publish_compact(repack.new_packs[0].clone(), repack.removed.clone(), 2) + .await; + link.heal(); + intact("compaction", second.clone()).await; + assert!( + compacted.is_ok(), + "a durable compaction must be acknowledged" + ); + + link.set(landed_412()); + let settings = handle + .publish_settings("[packs]\nenabled = false\n", "test", "landed 412") + .await; + link.heal(); + intact("settings", second).await; + assert!( + settings.is_ok(), + "a durable settings change must be acknowledged" + ); +} + #[tokio::test] async fn readiness_carries_only_proven_inventory_and_rechecks_revision_only_restart() { let writer_cache = tempfile::tempdir().unwrap(); diff --git a/docs/ROUNDTRIPS.md b/docs/ROUNDTRIPS.md index 447af901..7b1255b9 100644 --- a/docs/ROUNDTRIPS.md +++ b/docs/ROUNDTRIPS.md @@ -52,6 +52,7 @@ right shape. This document is the thinking tool; apply it to every protocol chan | Validated remote index admission | warm valid index: 0 store requests; absent/corrupt index: index GET (HEAD first only for old descriptors with unknown size), at most four indexes downloading concurrently | unchanged healthy store depth; corruption now triggers repair instead of admitting bad evidence. Checksums, lock, mmap and cleanup are local work; no LIST. Per-index checksum CPU cost occurs when constructing a new remote inventory, not per object lookup. | `index_cache.rs`, `remote.rs` | | Final replacement evidence | coherent refs-level manifest capture → verified current index mappings → conservation/current-tip scan → existing log claim/CAS | warm indexes add 0 store requests; cold/bad evidence adds index GETs (HEAD only for unknown sizes), four concurrent downloads. O(refs) local capture and streaming index-union scan per CAS attempt; no pack download or LIST. This cost is not zero-read or constant-time seal certification. | `classification.rs`, `closure.rs`, `index_cache.rs` | | Lost CAS response resolution | fresh manifest GET; only if the exact segment descriptor is listed, GET and compare the claimed log bytes | normally +1 GET on this failure path compared with key/sequence-only resolution; no added successful-push requests. Missing/folded evidence remains unknown. Per-attempt nonce and actual frame sizes are computed locally. | `publish.rs::cas_landed` | +| Manifest CAS answered 412 | the same evidence check before dropping the own segment: a 412 is contention, or a client library retrying a write that already landed (S3 SDK standard mode, GCS uploads with a precondition) | +1 manifest GET on the 412 path only (+1 log GET when our descriptor is listed); successful publishes unchanged | `publish.rs::process_batch`, `publish_compact_classified`, `publish_settings_impl` | | Settings publish (D24) | refs sync (conditional GET) → log slot PUT → manifest CAS; readers pay nothing extra (settings ride inline on the manifest) | 3 rounds; read: 0 | `publish.rs::publish_settings_impl` | | Lease acquire | 1 GET → 1 CAS put (or 1 Create when absent) | 2 | `coord.rs::try_acquire` | | Publish, local commit (2026-08-23) | unchanged in round trips: after the manifest CAS the ref txns are applied to the local copy **before** the new manifest version is advertised, both under `sync_mutex` (the refs phase of every sync); the reverse order let a reader cache the old refs under the new version, and without the lock a concurrent sync replayed the same entry (two `update-ref`, a lock collision). A landed CAS is answered `ok` whatever the local apply does — the next sync replays (one conditional GET that then returns 200, no extra write). | 0 extra | `publish.rs::process_batch` | diff --git a/docs/spec/README.md b/docs/spec/README.md index c3adef73..dae3a664 100644 --- a/docs/spec/README.md +++ b/docs/spec/README.md @@ -65,7 +65,7 @@ required implementation obligation even while the corresponding Rust audit remai | C6b/C6c, B2 | Exact policy/member/dependency selection, shared snapshot budget, native splice and engine gates | PackCoverage: retire/scope/policy/dependency/generation/companion; retry/old/new/concurrent witnesses | `packfile_uri` selector tests; stock-Git SHA-1/SHA-256 filtered clone and checkout; native residual, metadata isolation and gix/protected fallback | | C6d/C10a | Exact captured inputs, raw indexed-link validation in isolated inputs, conserving disk-spooled families, additive output commits and per-attempt indexed conservation/current-tip checks at final seal | PackReplacement: raw/conserve/replace/supersedes; seal/refusal/race/retry witnesses. ClosureProvenance: boundary/tip/seal/retry/rival controls | `pack_segments` conservation/resume fixtures and WAL exact-seal tests; SHA-1/SHA-256 unreachable broken-link rejection and bounded metadata validation; real-index seal regression rejects lost unreachable objects and post-plan uncovered tips; candidate publication/raw-boundary gates and adversarial CAS-race twins remain open | | C2/C3 | Checkpoint anti-regression and composition against the current CAS basis | WALCheckpoint fixed/bug; ManifestComposition stale-base/no-checkpoint/no-tips and composite-story witnesses | Delayed-checkpoint test and simulation replay oracle; combined publisher/checkpointer/sealer twin remains required | -| C5/C7/C8 | No-op receipts use seq 0 and publish no entry; empty-ref pack-only requests reject; unknown CAS remains typed unknown | BatchPublication: partial/isolation/retry/phantom/absence/noopphantom; atomic, late, folded and no-op witnesses | `noop_receipts_never_claim_a_siblings_log_entry`; `lost_cas_reply_is_resolved_or_unknown_without_losing_the_commit`; further late/overlaid failure twins remain required | +| C5/C7/C8 | No-op receipts use seq 0 and publish no entry; empty-ref pack-only requests reject; unknown CAS remains typed unknown | BatchPublication: partial/isolation/retry/phantom/absence/noopphantom; atomic, late, folded and no-op witnesses | `noop_receipts_never_claim_a_siblings_log_entry`; `lost_cas_reply_is_resolved_or_unknown_without_losing_the_commit`; `a_landed_cas_answered_412_keeps_the_commit`; further late/overlaid failure twins remain required | | C3/C7, B5 | Per-claim nonce prevents content-token aliasing; lost-response resolution checks exact bytes | LogSlotClaim: burned retry/early sweep/ambiguous delete/deterministic bytes; retry/sweep/late witnesses | `recreated_claims_have_distinct_bytes_and_resolution_checks_identity`; orphan WAL and fault simulation tests; delayed-delete ABA twin remains required | | C9 | Disable redirects and transport retries; only pre-delivery failures permit local fallback | FrontReplay: ambiguous replay/connect taxonomy/no precondition/no-op publishing; fallback/recovery/quiet no-op witnesses | `ambiguous_delivery_and_gateway_responses_never_publish_locally`; no-op WAL receipt test | | C10b | Exact retired membership and retained bucket bytes | MCTrim: reader mutation and historical-trim witness | Protected retired pack/index download test; explicit pinned historical local-reader twin remains required |