Don't delete a committed log segment when the manifest CAS answers 412

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
             }
         }

crates/walgit-wal/src/publish.rs

@@ -748,7 +748,26 @@ async fn process_batch(handle: &RepoHandle, batch: Vec<PublishRequest>) -> 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 {

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();

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` |

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 |