diff --git a/crates/walgit-server/tests/sim.rs b/crates/walgit-server/tests/sim.rs index 7998424c..d96f438c 100644 --- a/crates/walgit-server/tests/sim.rs +++ b/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/.pb` orphaned. Every /// later writer used to 412 on that key forever ("retry exhausted"): one crash 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/classification.rs b/crates/walgit-wal/src/classification.rs index 637b77bc..057e3142 100644 --- a/crates/walgit-wal/src/classification.rs +++ b/crates/walgit-wal/src/classification.rs @@ -376,7 +376,9 @@ impl RepoHandle { snapshots: &[CoverageSnapshot], lease: &tokio::sync::Mutex, ) -> Result { - 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()), diff --git a/crates/walgit-wal/src/publish.rs b/crates/walgit-wal/src/publish.rs index 07a23bda..d6d7db4e 100644 --- a/crates/walgit-wal/src/publish.rs +++ b/crates/walgit-wal/src/publish.rs @@ -23,12 +23,17 @@ //! log is strictly increasing, not dense — readers follow //! `manifest.log_segments`) and the orphan is CAS-deleted after commit. //! 5. CAS the manifest: `head_seq = last_seq`, `packs += new PackRefs`, -//! `log_segments += new segment`. On 412: re-sync (applies others' -//! entries), re-verify, CAS-delete *our* segment, retry with new seq. On -//! any other error the outcome is ambiguous (the bucket may have applied -//! the write and lost the response): re-read the manifest; if it lists -//! our segment the publish succeeded, else leave the segment in place -//! (an orphan is harmless, a deleted committed segment is not). +//! `log_segments += new segment`. On 412 the CAS lost a race, or a +//! transport retry of our own landed CAS answered it: re-read the +//! manifest first ([`slot_evidence`]). If it lists our segment the publish +//! succeeded; if it proves the segment uncommitted (key unlisted, nothing +//! folded at our seq), CAS-delete *our* segment, re-sync (applies +//! others' entries), re-verify and retry with a new seq; otherwise the +//! outcome is unknown and the segment stays. On any other error the +//! outcome is ambiguous (the bucket may have applied the write and lost +//! the response, or may still): re-read the manifest; if it lists our +//! segment the publish succeeded, else leave the segment in place (an +//! orphan is harmless, a deleted committed segment is not). //! 6. On success: apply all valid txns locally, update handle state, send //! success responses. //! @@ -324,37 +329,81 @@ pub(crate) async fn sweep_burned(store: &Prefixed, slot: &LogSlot) { } } -/// After a manifest CAS `PreconditionFailed`: CAS-delete *our* segment (only -/// the version we wrote; never anything a later writer put there). +/// CAS-delete *our* segment (only the version we wrote; never anything a later +/// writer put there). Only while no manifest can list it: before our manifest +/// CAS was sent, or after a 412 that [`SlotEvidence::Unlisted`] resolved. pub(crate) async fn drop_own_slot(store: &Prefixed, slot: &LogSlot) { let _ = store.delete(&slot.key, Some(slot.version.clone())).await; } -/// After a manifest CAS failed with a non-412 error: did the write land? -/// `Ok(Some(manifest))` when the fresh manifest lists our segment (committed), -/// `Ok(None)` when evidence is absent (unknown, not rejection; leave it alone). -pub(crate) async fn cas_landed( +/// What a fresh manifest proves about a slot whose manifest CAS did not report +/// success. +pub(crate) enum SlotEvidence { + /// The manifest lists this exact segment (key, first seq, size, bytes): the + /// CAS landed. + Committed(Box, walgit_store::Version), + /// The manifest does not list the segment's key and no checkpoint folded + /// the log at or past its first seq (folding drops segment descriptors), so + /// no committed manifest ever listed it. Together with a 412 this proves the + /// CAS lost: the request was conditioned on a manifest version that is gone, + /// so no copy of it can still land. After any other error it proves nothing + /// (the request may still be in flight). + Unlisted, + /// No proof either way: the log was folded at or past the segment, the + /// manifest lists the key with other contents, or there is no manifest. The + /// segment may be committed; it must stay and the outcome is unknown. + Unknown, +} + +/// Read the manifest fresh and classify `slot` (see [`SlotEvidence`]). +pub(crate) async fn slot_evidence( store: &Prefixed, slot: &LogSlot, -) -> Result, WalError> { +) -> Result { use walgit_store::ObjectStoreExt; let Some((meta, manifest)) = crate::store_proto::get_message::(store, keys::MANIFEST).await? else { - return Ok(None); + return Ok(SlotEvidence::Unknown); }; crate::validate_manifest(&manifest)?; - let listed = manifest.log_segments.iter().any(|s| { - s.key == slot.key && s.first_seq == slot.first_seq && s.size == slot.bytes.len() as u64 - }); - if !listed { - return Ok(None); + if let Some(listed) = manifest.log_segments.iter().find(|s| s.key == slot.key) { + // The key is referenced: whatever it holds, it is never ours to delete. + let descriptor = + listed.first_seq == slot.first_seq && listed.size == slot.bytes.len() as u64; + let exact = descriptor + && store + .get_bytes(&slot.key) + .await? + .is_some_and(|(_, bytes)| bytes == slot.bytes); + return Ok(if exact { + SlotEvidence::Committed(Box::new(manifest), meta.version) + } else { + SlotEvidence::Unknown + }); } - let exact = store - .get_bytes(&slot.key) - .await? - .is_some_and(|(_, bytes)| bytes == slot.bytes); - Ok(exact.then_some((manifest, meta.version))) + let folded = manifest + .checkpoint + .as_ref() + .is_some_and(|cp| cp.seq >= slot.first_seq); + Ok(if folded { + SlotEvidence::Unknown + } else { + SlotEvidence::Unlisted + }) +} + +/// After a manifest CAS failed with a non-412 error: did the write land? +/// `Ok(Some(manifest))` when the fresh manifest lists our segment (committed), +/// `Ok(None)` when evidence is absent (unknown, not rejection; leave it alone). +pub(crate) async fn cas_landed( + store: &Prefixed, + slot: &LogSlot, +) -> Result, WalError> { + Ok(match slot_evidence(store, slot).await? { + SlotEvidence::Committed(manifest, version) => Some((*manifest, version)), + SlotEvidence::Unlisted | SlotEvidence::Unknown => None, + }) } /// Verify a ref transaction against a working ref map. Returns per-ref results. @@ -748,7 +797,39 @@ 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 slot_evidence(&handle.store, &slot) + .instrument(span.clone()) + .await + { + Ok(SlotEvidence::Committed(fresh, version)) => { + tracing::warn!(repo = %handle.id, seq = last_seq, "manifest CAS answered 412 but landed"); + Some((*fresh, version)) + } + Ok(SlotEvidence::Unlisted) => None, + Ok(SlotEvidence::Unknown) => { + // A checkpoint may have folded our landed entry away before we looked: the + // segment stays, and re-verifying would report our own push as a conflict. + let msg = format!( + "manifest CAS answered 412 and the manifest no longer shows whether seq {first_seq} landed" + ); + return finish_with_error_msg( + batch, + &valid_indices, + &msg, + WalError::CommitUnknown(msg.clone()), + ); + } + 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 +1261,22 @@ 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 slot_evidence(&handle.store, &slot).await? { + SlotEvidence::Committed(fresh, version) => { + tracing::warn!(repo = %handle.id, seq, "compact manifest CAS answered 412 but landed"); + Some((*fresh, version)) + } + SlotEvidence::Unlisted => None, + SlotEvidence::Unknown => { + return Err(WalError::CommitUnknown(format!( + "compact manifest CAS answered 412 and the manifest no longer shows whether seq {seq} landed" + ))); + } + } + } 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 +1597,22 @@ 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. + match slot_evidence(&handle.store, &slot).await? { + SlotEvidence::Committed(..) => { + 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); + } + SlotEvidence::Unlisted => {} + SlotEvidence::Unknown => { + return Err(WalError::CommitUnknown(format!( + "settings manifest CAS answered 412 and the manifest no longer shows whether seq {seq} landed" + ))); + } + } drop_own_slot(&handle.store, &slot).await; attempts += 1; if attempts >= max_retries { @@ -1565,4 +1677,113 @@ mod claim_tests { assert!(cas_landed(&store, &second).await?.is_some()); Ok(()) } + + /// After a 412 the segment may be dropped only when the fresh manifest proves no committed + /// manifest ever listed it. A checkpoint folding the log at or past its seq drops segment + /// descriptors, so an unlisted segment there may still be committed. + #[tokio::test] + async fn only_an_unlisted_unfolded_segment_is_proven_uncommitted() -> anyhow::Result<()> { + let truth = walgit_store::memory::MemoryStore::shared(); + let store = Prefixed::new(truth, "repos/o/evidence/"); + let encode = |seq| { + frame::encode_entries([&LogEntry { + seq, + kind: EntryKind::Push as i32, + ..Default::default() + }]) + }; + // Our segment at seq 3, written on a basis of head 2. + let ClaimOutcome::Claimed(slot) = claim_log_slot(&store, 2, encode).await? else { + anyhow::bail!("uncontended claim expected"); + }; + let ours = LogSegmentRef { + key: slot.key.clone(), + first_seq: 3, + last_seq: 3, + size: slot.bytes.len() as u64, + sealed: true, + }; + let checkpoint = |seq: u64| walgit_proto::v1::CheckpointRef { + seq, + key: format!("checkpoints/{seq}/checkpoint.pb"), + refs_key: format!("checkpoints/{seq}/refs.pb"), + ..Default::default() + }; + let manifest = |head_seq: u64, segments: Vec, cp: Option| Manifest { + format_version: walgit_proto::WAL_FORMAT_VERSION, + object_format: "sha1".into(), + head_seq, + min_seq: cp.map_or(1, |seq| seq + 1), + log_segments: segments, + checkpoint: cp.map(checkpoint), + ..Default::default() + }; + let cases = [ + ( + "lists our exact segment", + manifest(3, vec![ours.clone()], None), + "committed", + ), + ( + "lists our segment after an older fold", + manifest(4, vec![ours.clone()], Some(2)), + "committed", + ), + ( + "lists our key with another size", + manifest( + 3, + vec![LogSegmentRef { + size: ours.size + 1, + ..ours.clone() + }], + None, + ), + "unknown", + ), + ( + "moved without reaching our seq", + manifest(2, vec![], None), + "unlisted", + ), + ( + "burned our seq and committed past it", + manifest( + 4, + vec![LogSegmentRef { + key: keys::log_segment_key(4), + first_seq: 4, + last_seq: 4, + size: 1, + sealed: true, + }], + Some(2), + ), + "unlisted", + ), + ("folded at our seq", manifest(3, vec![], Some(3)), "unknown"), + ( + "folded past our seq", + manifest(5, vec![], Some(5)), + "unknown", + ), + ]; + for (what, m, want) in cases { + store + .put_bytes(keys::MANIFEST, m.encode_to_vec(), PutMode::Overwrite) + .await?; + let got = match slot_evidence(&store, &slot).await? { + SlotEvidence::Committed(..) => "committed", + SlotEvidence::Unlisted => "unlisted", + SlotEvidence::Unknown => "unknown", + }; + assert_eq!(got, want, "{what}"); + } + store.delete(keys::MANIFEST, None).await?; + assert!( + matches!(slot_evidence(&store, &slot).await?, SlotEvidence::Unknown), + "no manifest proves nothing" + ); + Ok(()) + } } diff --git a/crates/walgit-wal/tests/coverage.rs b/crates/walgit-wal/tests/coverage.rs index 99b6f25a..39e5a9cf 100644 --- a/crates/walgit-wal/tests/coverage.rs +++ b/crates/walgit-wal/tests/coverage.rs @@ -280,6 +280,90 @@ async fn replacement_seal_requires_exact_outputs_and_keeps_inputs_until_success( ); } +/// A client library retrying the seal's manifest CAS after its first reply was lost gets 412 from +/// the seal's own landed write. Like every other publisher, the seal must find its segment listed +/// and keep it: deleting it leaves the committed manifest pointing at a missing log segment. +#[tokio::test] +async fn a_landed_seal_answered_412_keeps_the_commit() { + use walgit_store::fault::{FaultPlan, FaultStore}; + let store = MemoryStore::shared(); + let dir = tempfile::tempdir().unwrap(); + let link = FaultStore::new(store.clone(), "seal-landed-412", 1); + let registry = Registry::new(link.clone(), Arc::new(config(dir.path()))); + let h = registry.create(&id(), ObjectFormat::Sha1).await.unwrap(); + h.publish_settings("", "test", "generation").await.unwrap(); + let mut manifest = (*h.manifest()).clone(); + let input = real_pack(&store, &["conserved"]).await; + let mut output = real_pack(&store, &["conserved", "extra"]).await; + output.published_at = Some(walgit_proto::time::now()); + output.pack_groups = vec!["_retained".into()]; + output.audience = PackAudience::Retained as i32; + output.ref_policy = h.effective_config().refs.policy_identity(); + manifest.packs = vec![input.clone()]; + manifest.revision += 1; + seed(&store, &manifest).await; + let captured = h.publication_view().await.unwrap(); + manifest.packs.push(output.clone()); + manifest.revision += 1; + seed(&store, &manifest).await; + let lease = tokio::sync::Mutex::new( + walgit_store::coord::try_acquire( + store.clone(), + "test-seal-lease", + "test", + "seal", + std::time::Duration::from_mins(1), + ) + .await + .unwrap() + .unwrap(), + ); + link.set(FaultPlan { + p_cas_fail_after: 1.0, + only_keys: Some(vec!["manifest.pb".into()]), + ..Default::default() + }); + let sealed = h + .seal_pack_replacement( + &captured, + std::slice::from_ref(&input.checksum), + std::slice::from_ref(&output), + &[], + &lease, + ) + .await; + link.heal(); + let (_, bytes) = store + .get_bytes(&format!("{}manifest.pb", id().store_prefix())) + .await + .unwrap() + .unwrap(); + let committed = Manifest::decode(bytes.as_ref()).unwrap(); + for segment in &committed.log_segments { + assert!( + store + .get_bytes(&format!("{}{}", id().store_prefix(), segment.key)) + .await + .unwrap() + .is_some(), + "manifest lists {} but it is gone", + segment.key + ); + } + let seq = sealed.expect("a durable seal must be acknowledged"); + assert_eq!(committed.head_seq, seq); + assert!( + committed + .retired_packs + .iter() + .any(|p| p.checksum == input.checksum) + ); + let reader_dir = tempfile::tempdir().unwrap(); + let reader = fresh(&store, config(reader_dir.path())).await; + drop(reader.sync_refs().await.unwrap()); + assert_eq!(reader.manifest().head_seq, seq); +} + #[tokio::test] async fn exact_snapshot_is_not_authority_until_cas_and_corruption_is_rejected() { let store = MemoryStore::shared(); diff --git a/crates/walgit-wal/tests/wal.rs b/crates/walgit-wal/tests/wal.rs index ce85399e..a544e1f7 100644 --- a/crates/walgit-wal/tests/wal.rs +++ b/crates/walgit-wal/tests/wal.rs @@ -2910,6 +2910,213 @@ 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" + ); +} + +/// The manifest as the bucket holds it, read past every link. +async fn truth_manifest(truth: &MemoryStore, id: &RepoId) -> walgit_proto::v1::Manifest { + use walgit_store::ObjectStoreExt; + let key = format!("{}{}", id.store_prefix(), walgit_proto::keys::MANIFEST); + let (_, bytes) = truth.get_bytes(&key).await.unwrap().unwrap(); + ::decode(bytes.as_ref()).unwrap() +} + +/// A landed CAS answered 412, and before the writer reads the manifest back another instance +/// commits on top and a checkpoint folds the log past the writer's entry, dropping its segment +/// descriptor. The manifest then cannot show whether the push landed: the writer must keep the +/// segment and answer "commit unknown", not delete it and report its own durable push as a +/// conflict. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_landed_cas_folded_before_the_412_is_resolved_stays_unknown() { + use walgit_store::ObjectStoreExt; + use walgit_store::fault::{FaultPlan, FaultStore}; + let truth = MemoryStore::shared(); + let id = repo_id("o", "folded"); + let writer_cache = tempfile::tempdir().unwrap(); + let link = FaultStore::new(truth.clone(), "landed-412-folded", 1); + let writer = Registry::new(link.clone(), Arc::new(make_config(writer_cache.path(), 0))) + .create(&id, ObjectFormat::Sha1) + .await + .unwrap(); + + let work = WorkRepo::new(); + let first = work.commit("first", "content"); + let pack = ingest_pack_data(&writer, work.create_pack()).await.unwrap(); + link.set(FaultPlan { + p_cas_fail_after: 1.0, + only_keys: Some(vec!["manifest.pb".into()]), + ..Default::default() + }); + // The first manifest read after the CAS is the writer's evidence check: hold it. + let gate = link.gate_next("get", "manifest.pb"); + let push = { + let writer = writer.clone(); + let txn = make_txn(vec![("refs/heads/main", "", &first)]); + tokio::spawn(async move { + writer + .publish_push_synced(Some(pack), txn, HashMap::new()) + .await + }) + }; + gate.entered().await; + link.heal(); + let landed = truth_manifest(&truth, &id).await; + assert_eq!(landed.head_seq, 1, "the writer's CAS landed"); + let segment = format!("{}{}", id.store_prefix(), landed.log_segments[0].key); + + // Meanwhile another instance commits on top and folds the log past seq 1. + let other_cache = tempfile::tempdir().unwrap(); + let other = Registry::new(truth.clone(), Arc::new(make_config(other_cache.path(), 0))) + .open(&id) + .await + .unwrap(); + other + .publish_push( + None, + make_txn(vec![("refs/heads/other", "", &first)]), + HashMap::new(), + ) + .await + .unwrap(); + let cp = other.write_checkpoint().await.unwrap(); + assert_eq!(cp.seq, 2); + assert!( + truth_manifest(&truth, &id).await.log_segments.is_empty(), + "the fold dropped every segment descriptor" + ); + gate.release(); + + let outcome = push.await.unwrap(); + assert!( + matches!(outcome, Err(walgit_wal::WalError::CommitUnknown(_))), + "an unprovable outcome is unknown, never a conflict: {:?}", + outcome.map(|r| r.per_ref) + ); + assert!( + truth.get_bytes(&segment).await.unwrap().is_some(), + "the writer deleted its committed (folded) segment" + ); + // The push is durable, and the writer itself converges on it. + drop(writer.sync_refs().await.unwrap()); + assert_eq!( + writer.local().ref_view().unwrap().get("refs/heads/main"), + Some(first) + ); +} + #[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..bed87112 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). The segment is dropped only when the fresh manifest lists neither it nor a checkpoint at or past its seq (a fold drops descriptors); folded or otherwise unproven evidence stays unknown and keeps the segment | +1 manifest GET on the 412 path only (+1 log GET when our descriptor is listed); successful publishes unchanged | `publish.rs::slot_evidence`, `process_batch`, `publish_compact_classified`, `publish_settings_impl`, `classification.rs::seal_pack_replacement` | | 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..4b643b9f 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`; `a_landed_cas_folded_before_the_412_is_resolved_stays_unknown`; `a_landed_seal_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 |