Keep a 412'd log segment unless the manifest proves it uncommitted

crates/walgit-server/tests/sim.rs

@@ -1213,6 +1213,60 @@ async fn liveness_after_a_lost_cas_response() -> Result<()> {
     Ok(())
 }
 
+/// Liveness 4b: the bucket applies the manifest CAS, the reply is lost, and the client library's
+/// automatic retry of the same conditional PUT answers 412 (our own write moved the `ETag`). The
+/// writer must recognise its own commit: the push is acknowledged, the segment the manifest lists
+/// stays, and everybody (a cold instance too) syncs to it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn liveness_after_a_landed_cas_answered_412() -> Result<()> {
+    let mut c = Cluster::new(15, 2).await?;
+    let mut p = Pusher::new(0);
+    ensure!(
+        p.push_once(&c.instances[0], &c.id, Duration::from_secs(10))
+            .await?
+    );
+    c.instances[0].link.set(
+        FaultPlan {
+            p_cas_fail_after: 1.0,
+            ..Default::default()
+        }
+        .with_only(&["manifest.pb"]),
+    );
+    let acked = p
+        .push_once(&c.instances[0], &c.id, Duration::from_secs(10))
+        .await?;
+    c.instances[0].link.heal();
+    let r = check_truth(&c, std::slice::from_ref(&p)).await;
+    if let Err(e) = &r {
+        eprintln!("{}", c.dump_traces());
+        bail!("truth inconsistent after a landed CAS answered 412: {e:#}");
+    }
+    ensure!(
+        acked,
+        "a durable push was not acknowledged: {:?}",
+        p.errors.last()
+    );
+    let head = c.truth_manifest().await?.head_seq;
+    let cold = c.add_instance("cold", &|_| {});
+    let h = c.instances[cold].open(&c.id).await?;
+    tokio::time::timeout(Duration::from_secs(10), h.sync_full())
+        .await
+        .map_err(|_| anyhow!("cold sync hung"))??;
+    ensure!(
+        h.applied_seq() == head,
+        "cold at {} != {head}",
+        h.applied_seq()
+    );
+    ensure!(
+        p.push_once(&c.instances[0], &c.id, Duration::from_secs(10))
+            .await?,
+        "writer wedged: {:?}",
+        p.errors.last()
+    );
+    check_truth(&c, std::slice::from_ref(&p)).await?;
+    Ok(())
+}
+
 /// Liveness 7 (found by `sim_safety_then_liveness`): a writer crashes between
 /// its log PUT and its manifest CAS, leaving `log/<head+1>.pb` orphaned. Every
 /// later writer used to 412 on that key forever ("retry exhausted"): one crash

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/classification.rs

@@ -376,7 +376,9 @@ impl RepoHandle {
         snapshots: &[CoverageSnapshot],
         lease: &tokio::sync::Mutex<walgit_store::coord::LeaseGuard>,
     ) -> Result<u64, WalError> {
-        use crate::publish::{ClaimOutcome, claim_log_slot, drop_own_slot, sweep_burned};
+        use crate::publish::{
+            ClaimOutcome, SlotEvidence, claim_log_slot, drop_own_slot, slot_evidence, sweep_burned,
+        };
         use walgit_proto::v1::{EntryKind, LogEntry, LogSegmentRef};
         use walgit_proto::{frame, keys, time};
         if inputs.is_empty() || outputs.is_empty() || inputs.len() > 4096 || outputs.len() > 4096 {
@@ -528,7 +530,22 @@ impl RepoHandle {
                 .await;
             let committed = match result {
                 Ok(meta) => Some((updated, meta.version)),
-                Err(StoreError::PreconditionFailed { .. }) => None,
+                // A 412 may answer a transport retry of our own landed CAS: look before
+                // dropping the segment (see `publish::process_batch`).
+                Err(StoreError::PreconditionFailed { .. }) => {
+                    match slot_evidence(&self.store, &slot).await? {
+                        SlotEvidence::Committed(fresh, version) => {
+                            tracing::warn!(repo = %self.id, seq, "seal manifest CAS answered 412 but landed");
+                            Some((*fresh, version))
+                        }
+                        SlotEvidence::Unlisted => None,
+                        SlotEvidence::Unknown => {
+                            return Err(WalError::CommitUnknown(format!(
+                                "seal manifest CAS answered 412 and the manifest no longer shows whether seq {seq} landed"
+                            )));
+                        }
+                    }
+                }
                 Err(error) => match crate::publish::cas_landed(&self.store, &slot).await? {
                     Some(pair) => Some(pair),
                     None => return Err(error.into()),