From 9fb02239613fd490114cb180c444947f8cfb5558 Mon Sep 17 00:00:00 2001 From: Yad Konrad Date: Thu, 1 Oct 2026 13:45:23 -0400 Subject: [PATCH] Don't delete a committed log segment when the manifest CAS answers 412 The S3 SDK retries failed calls three times by default and the GCS client retries uploads that carry a precondition, the manifest CAS included. When a CAS lands and its reply is lost, the retry gets a 412 from our own write. process_batch, publish_compact_classified and publish_settings_impl all read any 412 as a lost race and delete their log segment, which the committed manifest now lists: the writer reports a failure for a durable change and every reader then fails on the missing segment. Run the existing cas_landed check on a 412 too before dropping the segment, and keep an unknown outcome unknown when that check fails. The fault store gains the missing fault kind (applied, then 412), sharing the existing CAS roll so seeded simulations keep their sequences, and a test drives a push, a compaction and a settings change through it. One extra manifest GET on the 412 path only. Co-Authored-By: Claude Opus 5.5 --- crates/walgit-store/src/fault.rs | 39 ++++++++++- crates/walgit-wal/src/publish.rs | 41 ++++++++++- crates/walgit-wal/tests/wal.rs | 116 +++++++++++++++++++++++++++++++ docs/ROUNDTRIPS.md | 1 + docs/spec/README.md | 2 +- 5 files changed, 194 insertions(+), 5 deletions(-) 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 |