From bd472831ebf9eab706c4d255a2ae05b518828e31 Mon Sep 17 00:00:00 2001 From: Valter Balegas Date: Tue, 14 Jul 2026 12:43:59 +0100 Subject: [PATCH 1/2] =?UTF-8?q?fix(durable-streams-rust):=20recovery=20har?= =?UTF-8?q?dening=20=E2=80=94=20fail-stop=20barriers,=20crash-safe=20metad?= =?UTF-8?q?ata,=20rollback-on-error?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deep failure-mode review (4 focused audits) findings, all data-loss or availability class: FAIL-STOP durability barriers (fsyncgate family): a failed fdatasync/syncfs/ seal must never be retried in place — Linux marks the failed pages clean and clears the fd error state, so the retry falsely succeeds and (a) the committer would ack lost bytes, (b) the next checkpoint would recycle the WAL records that were the only durable copy, (c) a retried seal could graft a zero tail that silently ends replay. All three now abort the process (nothing acked is at risk at first-error time; restart replays the durable log). Committer thread is supervised: a panic aborts instead of freezing durable_lsn forever (silent whole-server DoS). WAL stage write failures retry bounded (data still in hand) then abort — no more permanent watermark gap that wedged the shard. Checkpoint error handling: a failed checkpoint re-registers the drained dirty set — previously the streams' tails proofs were dropped, the NEXT checkpoint recycled their WAL records anyway, and any later restart truncated acked, durable bytes back to a stale frontier. Crash-safe metadata: sidecar tmp is always synced before rename (a torn rename target used to get the sidecar AND the data file deleted at boot); unparsable sidecars now QUARANTINE (.meta.corrupt, stream skipped, data kept) instead of deleting the stream; sidecar read errors fail the boot; recovery data-file open errors fail the boot loudly (EMFILE used to silently drop streams whose WAL records the reset then destroyed); wal/shards and streams/.lanes written tmp+sync+rename+dir-fsync; dir fsyncs added for segment create/roll/reset (same-name inode clobber) and the tails rename before recycle; per-lane .lane mount markers refuse to boot when a lane's device mount is missing (previously the empty mountpoint booted 'fine' and the WAL reset destroyed the lane's acked records); create rolls back the map entry + file + parent refcount when the durable meta write fails (phantom streams whose acked appends the next boot deleted); recovered compaction intents are durably cleared before appends (double-crash file_base mis-derivation shifted all reads). Append rollback: a stage failure now rolls back the data-file write, tail, and producer/seq/close state — 500'd bytes used to be durably resurrected by the next append, and the client's retry was swallowed as a producer duplicate. write_wire truncates partial writes (ENOSPC desynced logical/physical offsets for all later data). Tier: seal cuts at the DURABLE tail, not the writer tail (sealing un-acked bytes made them served-as-durable after a crash and permanently shadowed the retried acked bytes); boot re-offload unlinks the local chunk only after the manifest flip is durable; an unreadable sealed chunk now poisons the read (was: silently omitted interior bytes in a well-formed 200); Meta::capture snapshots segments+sealed_offset under one manifest lock (torn captures persisted manifest holes). 3 new e2e regression tests (stage rollback, sidecar quarantine, lane-mount guard); the stage-failure injection hook moved pre-reservation to model the now-reachable semantics. 109/109 tests. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y3x7bcT9vLGiT4tXZeQpnk --- packages/durable-streams-rust/src/handlers.rs | 87 +++++- packages/durable-streams-rust/src/store.rs | 246 +++++++++++++--- packages/durable-streams-rust/src/tier.rs | 61 +++- .../durable-streams-rust/src/wal/e2e_tests.rs | 99 +++++++ .../durable-streams-rust/src/wal/shard.rs | 273 ++++++++++++------ .../durable-streams-rust/src/wal/walset.rs | 15 +- 6 files changed, 637 insertions(+), 144 deletions(-) diff --git a/packages/durable-streams-rust/src/handlers.rs b/packages/durable-streams-rust/src/handlers.rs index 29d1dc974e..30866e0c98 100644 --- a/packages/durable-streams-rust/src/handlers.rs +++ b/packages/durable-streams-rust/src/handlers.rs @@ -585,6 +585,7 @@ async fn handle_create(store: Arc, req: Req, path: String) -> Resp { let lock_t0 = crate::telemetry::Timer::start(); let mut ap = st.appender.lock().await; crate::telemetry::record_append_lock_wait(lock_t0.elapsed_secs()); + let pre_written = ap.written; let new_tail = match write_wire(&st, &mut ap, &wire) { Ok(t) => t, Err(_) => return text_response(500, "write failed"), @@ -599,7 +600,19 @@ async fn handle_create(store: Arc, req: Req, path: String) -> Resp { // durability wait runs after the lock is dropped. let staged_lsn = match stage_for_durability(&store, &st, &wire, stream_offset) { Ok(lsn) => lsn, - Err(_) => return text_response(500, "wal stage failed"), + Err(_) => { + // ROLL BACK the data-file write: the bytes were 500'd but + // already sit in the file — left in place they would be + // durably resurrected by the next successful append / + // checkpoint (client told "failed", bytes served anyway). + let _ = ap.file.set_len(pre_written); + ap.written = pre_written; + { + let mut sh = st.shared.write().unwrap(); + sh.tail = sh.file_base + pre_written; + } + return text_response(500, "wal stage failed"); + } }; drop(ap); if let Some(lsn) = staged_lsn { @@ -735,7 +748,16 @@ async fn wait_durable_lsn(store: &Arc, st: &Arc, lsn: u64) { /// awaited in `wait_durable_lsn`. fn write_wire(st: &StreamState, ap: &mut Appender, wire: &Bytes) -> std::io::Result { use std::io::Write; - (&*ap.file).write_all(wire)?; + if let Err(e) = (&*ap.file).write_all(wire) { + // A partial write (ENOSPC mid-slice) leaves garbage bytes in the file + // PAST `ap.written` while the logical offsets don't advance — every + // later append would land after the garbage (O_APPEND) with a logical + // offset that assumes it landed at `ap.written`: silent, permanent + // offset desync for all subsequent data. Truncate back to the exact + // pre-write length so physical == logical again. + let _ = ap.file.set_len(ap.written); + return Err(e); + } ap.written += wire.len() as u64; let tail = { let mut s = st.shared.write().unwrap(); @@ -1082,6 +1104,20 @@ async fn handle_append_inner(store: Arc, req: Req, path: String) -> (Resp // readers only AFTER durability (below), so a live reader never observes // bytes a crash could roll back (PROTOCOL.md §4.1). let mut new_tail = None; + let pre_written = ap.written; + // Pre-mutation snapshots for the stage-failure rollback below: a 500'd + // append must leave NO trace — neither bytes (resurrected by the next + // append/checkpoint) nor producer/seq dedup state (which would swallow the + // client's retry as a duplicate: silent loss from the client's view). + let (prev_producer, prev_seq_header) = { + let sh = st.shared.read().unwrap(); + ( + producer + .as_ref() + .map(|p| (p.id.clone(), sh.producers.get(&p.id).cloned())), + sh.last_seq_header.clone(), + ) + }; if !wire.is_empty() { match write_wire(&st, &mut ap, &wire) { Ok(t) => new_tail = Some(t), @@ -1136,7 +1172,39 @@ async fn handle_append_inner(store: Arc, req: Req, path: String) -> (Resp let staged_lsn = if !wire.is_empty() { match stage_for_durability(&store, &st, &wire, stream_offset) { Ok(lsn) => lsn, - Err(_) => ret!(text_response(500, "wal stage failed"), Conflict), + Err(_) => { + // ROLL BACK everything this append changed (still under the + // appender lock, so no concurrent appender observed it): + // 1) the data-file bytes — otherwise the next successful append + // advances the durable frontier over them and they are served + // (and checkpoint-persisted) despite the 500; + // 2) the in-memory tail; + // 3) producer/seq/closed state — otherwise the client's RETRY of + // this failed append is deduplicated as "already seen" and + // silently dropped. + let _ = ap.file.set_len(pre_written); + ap.written = pre_written; + { + let mut sh = st.shared.write().unwrap(); + sh.tail = sh.file_base + pre_written; + if let Some((id, prev)) = &prev_producer { + match prev { + Some(ps) => { + sh.producers.insert(id.clone(), ps.clone()); + } + None => { + sh.producers.remove(id); + } + } + } + sh.last_seq_header = prev_seq_header.clone(); + if close_req { + sh.closed = false; + sh.closed_by = None; + } + } + ret!(text_response(500, "wal stage failed"), Conflict) + } } } else { None @@ -1338,6 +1406,13 @@ fn stream_resolved_body( } for sl in slices { match sl { + ResolvedSlice::Missing => { + // Poison slice (unreadable sealed chunk): abort the + // connection — a response missing interior bytes must never + // terminate cleanly. + fail(); + return; + } ResolvedSlice::Local(seg) => { // Window the (possibly large) local slice so we never hold // more than COLD_LOCAL_WINDOW of it in memory at once. @@ -1410,6 +1485,12 @@ async fn materialize_resolved( out.put_slice(prefix); for sl in slices { match sl { + ResolvedSlice::Missing => { + return Err(Error::new( + ErrorKind::NotFound, + "sealed chunk unreadable (poison slice)", + )); + } ResolvedSlice::Local(seg) => { let want = seg.len; let bytes = tokio::task::spawn_blocking(move || { diff --git a/packages/durable-streams-rust/src/store.rs b/packages/durable-streams-rust/src/store.rs index 70cabde507..4ec43781ba 100644 --- a/packages/durable-streams-rust/src/store.rs +++ b/packages/durable-streams-rust/src/store.rs @@ -513,6 +513,10 @@ impl Store { ) -> std::io::Result { let streams_dir = data_dir.join("streams"); std::fs::create_dir_all(&streams_dir)?; + // Whether this store existed before THIS boot — captured before the + // `.lanes` block below writes the marker on first initialization (the + // lane-mount guard must not fire on a genuinely fresh store). + let store_initialized = streams_dir.join(".lanes").exists(); // Persist + validate the stream-lane count (mirrors the WAL shard count's // persisted-N contract): opening a laned layout with a different // `--stream-lanes` would make every existing stream silently invisible @@ -531,7 +535,7 @@ impl Store { return Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, format!( - "--stream-lanes {} does not match this data dir's on-disk layout ({} lanes, recorded in {}). The lane count is a layout choice and must match across restarts.", + "--stream-lanes {} does not match this data dir's on-disk layout ({} lanes, recorded in {}). The lane count is a layout choice and must match across restarts.", stream_lanes(), on_disk, marker.display() @@ -549,10 +553,21 @@ impl Store { { return Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, - "--stream-lanes > 1 over an existing flat streams/ layout; this data dir was created with 1 lane", + "--stream-lanes > 1 over an existing flat streams/ layout; this data dir was created with 1 lane", )); } - std::fs::write(&marker, format!("{}\n", stream_lanes()))?; + // Durable write (tmp + sync + rename + dir fsync): losing + // this marker while laned dirs full of data survive would + // let a later boot mis-read the layout. + let tmp = streams_dir.join(".lanes.tmp"); + { + use std::io::Write; + let mut f = File::create(&tmp)?; + f.write_all(format!("{}\n", stream_lanes()).as_bytes())?; + f.sync_all()?; + } + std::fs::rename(&tmp, &marker)?; + fsync_parent_dir(&marker)?; } Err(e) => return Err(e), } @@ -563,7 +578,44 @@ impl Store { let mut lane_fds = Vec::with_capacity(stream_lanes()); for lane in 0..stream_lanes() { let d = lane_dir(&data_dir, lane); + // MOUNT GUARD: each lane dir carries a `.lane` marker written at + // first initialization. Lane dirs are mountpoints for independent + // devices in the intended layout — if a lane's mount is absent at + // boot, `create_dir_all` silently recreates an EMPTY dir on the + // parent fs, every stream on that lane vanishes from recovery, + // and the WAL reset then destroys their acked records. So: once + // the store is initialized (the `.lanes` count marker exists), a + // lane whose `.lane` marker is missing AND whose dir is empty is + // treated as a missing mount and boot is refused. (A missing + // marker with contents present = pre-marker layout: adopt it.) + let marker = d.join(".lane"); + if store_initialized && !marker.exists() { + let has_contents = std::fs::read_dir(&d) + .map(|mut it| it.next().is_some()) + .unwrap_or(false); + if !has_contents { + return Err(std::io::Error::new( + std::io::ErrorKind::NotFound, + format!( + "stream lane {lane} at {} is empty and unmarked on an \ + initialized store — its device mount is likely missing. \ + Refusing to boot: continuing would drop every stream on \ + this lane and the WAL reset would destroy their acked \ + records. Mount the lane device (or restore its data) and \ + restart.", + d.display() + ), + )); + } + } std::fs::create_dir_all(&d)?; + if !marker.exists() { + let mut f = File::create(&marker)?; + use std::io::Write; + f.write_all(lane.to_string().as_bytes())?; + f.sync_all()?; + fsync_parent_dir(&marker)?; + } lane_fds.push(File::open(&d)?); } *LANE_SYNC_FDS.lock().unwrap() = Some(Arc::new(lane_fds)); @@ -610,6 +662,7 @@ impl Store { let _ = streams_dir; // root; per-lane dirs derived below (lane 0 == root when lanes == 1) let mut metas: HashMap = HashMap::new(); let mut data_files: Vec = Vec::new(); + let mut quarantined: Vec = Vec::new(); let mut entries: Vec = Vec::new(); for lane in 0..stream_lanes() { for entry in std::fs::read_dir(lane_dir(&self.data_dir, lane))? { @@ -618,8 +671,8 @@ impl Store { } for p in entries { let name = p.file_name().and_then(|n| n.to_str()).unwrap_or(""); - if name == ".lanes" { - // stream-lane layout marker, not stream data + if name == ".lanes" || name == ".lane" { + // stream-lane layout / lane-mount markers, not stream data continue; } if name.ends_with(".meta.tmp") { @@ -633,21 +686,49 @@ impl Store { } else if name.ends_with(".meta") { let data_path = PathBuf::from(p.as_os_str().to_str().unwrap().trim_end_matches(".meta")); if data_path.exists() { - if let Ok(bytes) = std::fs::read(&p) { - if let Ok(meta) = serde_json::from_slice::(&bytes) { - metas.insert(meta.path.clone(), (meta, data_path)); - continue; + match std::fs::read(&p) { + Ok(bytes) => { + if let Ok(meta) = serde_json::from_slice::(&bytes) { + metas.insert(meta.path.clone(), (meta, data_path)); + } else { + // QUARANTINE, never delete: an unparsable sidecar + // next to a data file is far more likely a torn + // write than garbage worth destroying — deleting + // it (and then the "orphaned" data file) would + // silently erase a fully-acked stream. Park the + // sidecar, keep the data file untouched, and skip + // the stream loudly so an operator can repair. + eprintln!( + "WARN: quarantining unparsable stream sidecar {} \ + (stream skipped this boot; data file kept)", + p.display() + ); + let _ = std::fs::rename(&p, p.with_extension("meta.corrupt")); + quarantined.push(data_path); + } + } + Err(e) => { + // A transient READ error (EIO/EACCES) is not + // corruption: fail the boot rather than misclassify + // and destroy data. + return Err(std::io::Error::new( + e.kind(), + format!("failed to read stream sidecar {}: {e}", p.display()), + )); } } + continue; } + // Sidecar with NO data file: stale leftover, safe to remove. let _ = std::fs::remove_file(&p); } else { data_files.push(p); } } - // Drop orphan data files (no usable sidecar). + // Drop orphan data files (no usable sidecar) — but NEVER a data file + // whose sidecar was quarantined above. for p in data_files { - if !metas.values().any(|(_, dp)| *dp == p) { + if !metas.values().any(|(_, dp)| *dp == p) && !quarantined.contains(&p) { let _ = std::fs::remove_file(&p); } } @@ -733,14 +814,35 @@ impl Store { } // Remove any temp not promoted above (post-rename leftover, or a partial). let _ = std::fs::remove_file(&tmp_path); + // A failed open/stat here is a RESOURCE error (EMFILE, EIO, perms) on a + // data file whose sidecar just parsed — silently skipping the stream + // (the old `.ok()?`) meant its WAL records were skipped at replay and + // then destroyed by reset_after_recovery: acked-data loss with no log + // line. Boot must fail loudly instead; the operator fixes the resource + // limit and the data is still intact. let file = Arc::new( OpenOptions::new() .read(true) .append(true) .open(data_path) - .ok()?, + .unwrap_or_else(|e| { + panic!( + "recovery: cannot open stream data file {} ({e}); refusing \ + to boot without it — skipping would let WAL reset destroy \ + its acked records", + data_path.display() + ) + }), ); - let written = file.metadata().ok()?.len(); + let written = file + .metadata() + .unwrap_or_else(|e| { + panic!( + "recovery: cannot stat stream data file {} ({e})", + data_path.display() + ) + }) + .len(); // `file_base` is the live file's logical start. With a `pending_compaction` // intent and the durable temp promoted above, the live file IS the full // residual `[new_file_base, tail)` — so `file_base = new_file_base`, derived @@ -830,6 +932,23 @@ impl Store { // Re-enqueue any sealed-but-not-yet-offloaded segments left by a crash // mid-offload (placement still Local while a remote tier is configured). self.reconcile_manifest_on_boot(&state); + // A recovered `pending_compaction` intent must be durably CLEARED (with + // the derived `file_base`) before this stream can accept appends: the + // derivation branches assume the file length still matches the crash + // moment, so "appends after boot + a second crash before any sidecar + // write" would re-enter them with a grown file and mis-derive + // `file_base` — shifting every subsequent live read AND replay write by + // the appended delta (silent corruption). Persisting now closes that + // double-crash window; a persist failure fails the boot loudly. + if meta.pending_compaction.is_some() { + write_meta_sync(&state, true).unwrap_or_else(|e| { + panic!( + "recovery: cannot durably clear the compaction intent for {} ({e}); \ + booting without it risks a mis-derived file_base after another crash", + state.file_path.display() + ) + }); + } self.streams.insert(path.to_string(), state.clone()); Some(state) } @@ -1037,11 +1156,34 @@ impl Store { v.insert(state.clone()); // Take the fork reference only once insertion has succeeded, so // rejected/raced creates never leak a refcount on the source. - if let Some(p) = &parent { - p.shared.write().unwrap().ref_count += 1; - write_meta_sync(p, true)?; + let created = (|| -> std::io::Result<()> { + if let Some(p) = &parent { + p.shared.write().unwrap().ref_count += 1; + if let Err(e) = write_meta_sync(p, true) { + p.shared.write().unwrap().ref_count -= 1; + return Err(e); + } + } + if let Err(e) = write_meta_sync(&state, true) { + if let Some(p) = &parent { + p.shared.write().unwrap().ref_count -= 1; + let _ = write_meta_sync(p, true); + } + return Err(e); + } + Ok(()) + })(); + if let Err(e) = created { + // UNDO the create: without a durable sidecar the stream must + // not stay live — WAL mode would happily ack appends to it, + // and the next boot would treat the sidecar-less data file as + // an orphan and delete it (acked appends destroyed after a + // create the client saw fail). + self.streams + .remove_if(&state.path, |_, cur| Arc::ptr_eq(cur, &state)); + let _ = std::fs::remove_file(&state.file_path); + return Err(e); } - write_meta_sync(&state, true)?; Ok(CreateResult::Created(state)) } } @@ -1234,6 +1376,32 @@ fn unix_secs(t: SystemTime) -> u64 { impl Meta { fn capture(st: &StreamState) -> Meta { + let seg_snapshot: (Vec, u64) = { + let m = st.tier.manifest.lock().unwrap(); + ( + m.segments + .iter() + .map(|seg| match &seg.placement { + crate::tier::Placement::Local(p) => MetaSegment { + logical_start: seg.logical_start, + len: seg.len, + remote_key: None, + local_file: p + .file_name() + .and_then(|n| n.to_str()) + .map(|s| s.to_string()), + }, + crate::tier::Placement::Remote(key) => MetaSegment { + logical_start: seg.logical_start, + len: seg.len, + remote_key: Some(key.clone()), + local_file: None, + }, + }) + .collect(), + m.sealed_offset, + ) + }; let s = st.shared.read().unwrap(); Meta { id: st.id, @@ -1254,30 +1422,14 @@ impl Meta { last_access_unix: unix_secs(s.last_access), ref_count: s.ref_count, soft_deleted: s.soft_deleted, - segments: { - let m = st.tier.manifest.lock().unwrap(); - m.segments - .iter() - .map(|seg| match &seg.placement { - crate::tier::Placement::Local(p) => MetaSegment { - logical_start: seg.logical_start, - len: seg.len, - remote_key: None, - local_file: p - .file_name() - .and_then(|n| n.to_str()) - .map(|s| s.to_string()), - }, - crate::tier::Placement::Remote(key) => MetaSegment { - logical_start: seg.logical_start, - len: seg.len, - remote_key: Some(key.clone()), - local_file: None, - }, - }) - .collect() - }, - sealed_offset: st.tier.manifest.lock().unwrap().sealed_offset, + // segments + sealed_offset MUST come from ONE lock acquisition: a + // seal pass interleaving between two separate acquisitions would + // yield a capture whose sealed_offset covers a region absent from + // `segments` — persisted, that is a permanent manifest hole below + // the watermark (reads resolve nothing there; the sealer never + // re-seals it). + segments: seg_snapshot.0, + sealed_offset: seg_snapshot.1, file_base: Some(s.file_base), pending_compaction: *st.compaction.lock().unwrap(), durable_tail: Some(s.durable_tail), @@ -1304,9 +1456,14 @@ pub fn write_meta_sync(st: &StreamState, durable: bool) -> std::io::Result<()> { use std::io::Write; let mut f = File::create(&tmp)?; f.write_all(&bytes)?; - if durable { - f.sync_all()?; - } + // ALWAYS sync the tmp's data before the rename — even for the + // "non-durable" lagging flushes. Renaming an unsynced tmp over the + // previously-durable sidecar lets a power crash land the rename with + // zero-length/garbage content (no ext4-style rename heuristic on all + // filesystems), and boot treats an unparsable sidecar as corruption. + // The `durable` flag now only gates the parent-dir fsync (rename + // persistence), preserving the lagging-flush contract's cheapness. + f.sync_all()?; } std::fs::rename(&tmp, &final_path)?; // A rename is crash-durable only once the parent dir entry is fsynced. @@ -1499,6 +1656,7 @@ mod tier_tests { let b = bs.get_range(&key, offset, len).await.unwrap(); out.extend_from_slice(&b); } + ResolvedSlice::Missing => panic!("test read hit a poison slice"), } } out diff --git a/packages/durable-streams-rust/src/tier.rs b/packages/durable-streams-rust/src/tier.rs index b7a08f3fd1..82a5e85a6a 100644 --- a/packages/durable-streams-rust/src/tier.rs +++ b/packages/durable-streams-rust/src/tier.rs @@ -521,7 +521,15 @@ impl Store { }; let (tail, file_base, file) = { let s = st.shared.read().unwrap(); - (s.tail, s.file_base, s.file.clone()) + // Seal at most up to the DURABLE frontier, never the writer + // tail: `s.tail` includes bytes whose WAL fdatasync hasn't + // completed (un-acked). Sealing them makes never-acked bytes + // durably part of the sealed record — after a crash, recovery + // truncates the live file to the durable frontier but nothing + // trims sealed segments, so the torn bytes would be SERVED as + // durable and would permanently shadow the client's retried + // (acked) bytes at the same offsets. + (s.durable_tail, s.file_base, s.file.clone()) }; let unsealed = tail.saturating_sub(sealed_offset); if unsealed < seg_bytes { @@ -827,8 +835,29 @@ impl TierTask { } } let stc = st.clone(); - let _ = tokio::task::spawn_blocking(move || write_meta_sync(&stc, true)).await; - let _ = tokio::fs::remove_file(&path).await; + // The unlink is safe ONLY after the Local->Remote flip is durable: + // unlinking first and crashing leaves a durable sidecar that says + // Local pointing at a file that no longer exists — the segment + // becomes permanently unreadable even though the object is in S3. + match tokio::task::spawn_blocking(move || write_meta_sync(&stc, true)).await { + Ok(Ok(())) => { + let _ = tokio::fs::remove_file(&path).await; + } + Ok(Err(e)) => { + eprintln!( + "WARN: boot offload of {} uploaded but could not durably flip \ + the manifest ({e}); keeping the local chunk for retry", + path.display() + ); + } + Err(e) => { + eprintln!( + "WARN: boot offload manifest flip task failed for {} ({e}); \ + keeping the local chunk", + path.display() + ); + } + } } st.tier.manifest.lock().unwrap().offloading = false; } @@ -840,6 +869,10 @@ impl TierTask { pub enum ResolvedSlice { Local(Segment), Remote { key: String, offset: u64, len: u64 }, + /// A slice whose backing local chunk could not be opened (missing/EIO). + /// Poison: any read resolving this must ERROR, never serve around it — + /// omitting it would produce a well-formed 200 missing interior bytes. + Missing, } /// If every resolved slice is `Local` (the live data file and/or sealed chunk @@ -854,7 +887,9 @@ pub fn into_local_segments(slices: Vec) -> Result, V .into_iter() .map(|s| match s { ResolvedSlice::Local(seg) => seg, - ResolvedSlice::Remote { .. } => unreachable!("checked all-Local above"), + ResolvedSlice::Remote { .. } | ResolvedSlice::Missing => { + unreachable!("checked all-Local above") + } }) .collect()) } else { @@ -933,12 +968,24 @@ fn resolve_sealed(st: &Arc, lo: u64, hi: u64, out: &mut Vec out.push(ResolvedSlice::Local(Segment { file: Arc::new(f), file_start: off_in_seg, len, - })); + })), + Err(e) => { + // NEVER silently omit a slice: Content-Length is computed + // from the resolved slices, so a dropped slice produces a + // well-formed 200 that is simply MISSING interior bytes — + // the client's next offset then skips them forever. + // Surface a poison slice so the read errors instead. + eprintln!( + "ERROR: sealed chunk {} unreadable ({e}) — failing the read", + path.display() + ); + out.push(ResolvedSlice::Missing); + } } } Placement::Remote(key) => { diff --git a/packages/durable-streams-rust/src/wal/e2e_tests.rs b/packages/durable-streams-rust/src/wal/e2e_tests.rs index 8b0c1511aa..b138ef3d8a 100644 --- a/packages/durable-streams-rust/src/wal/e2e_tests.rs +++ b/packages/durable-streams-rust/src/wal/e2e_tests.rs @@ -952,6 +952,105 @@ async fn e2e_recycled_first_segment_acked_records_survive_crash() { let _ = std::fs::remove_dir_all(&dir); } +/// Recovery-hardening: a stage failure must leave NO trace — the 500'd bytes +/// must not be resurrected by later successful appends, neither live nor +/// across a crash/recovery. (Uses the shard's test-only write-failure +/// injection, which surfaces exactly like a production stage error.) +#[tokio::test] +async fn e2e_stage_failure_rolls_back_data_write() { + let _guard = DurabilityGuard::wal(); + let dir = tmp("stage-rollback"); + let h = Harness::boot(&dir, Some(1), 1).unwrap(); + create_stream(&h.store, "rb", OCTET).await; + + let mut expected = Vec::new(); + append_acked(&h.store, "rb", OCTET, b"first|").await; + expected.extend_from_slice(b"first|"); + + // Inject: the next WAL stage write fails -> the append must 500 and the + // data-file write must be rolled back. + h.walset.shards()[0].fail_next_write(); + let resp = handlers::handle( + Arc::clone(&h.store), + post_req("rb", OCTET, b"LOST-must-not-resurrect|"), + ) + .await; + assert_eq!(resp.status, 500, "injected stage failure must 500"); + + // A later append succeeds; the failed bytes must NOT appear before it. + append_acked(&h.store, "rb", OCTET, b"second|").await; + expected.extend_from_slice(b"second|"); + + let live = stream_file_bytes(&h.store, "rb"); + assert_eq!(live, expected, "500'd bytes must not persist in the live file"); + + h.crash(); + let h2 = Harness::boot(&dir, None, 1).unwrap(); + let got = stream_file_bytes(&h2.store, "rb"); + assert_eq!(got, expected, "500'd bytes must not resurrect across recovery"); + h2.crash(); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Recovery-hardening: an unparsable sidecar must QUARANTINE the stream (skip +/// + keep the data file + park the sidecar as .meta.corrupt), never delete the +/// data file — a torn sidecar next to real data is a torn write, not garbage. +#[tokio::test] +async fn e2e_corrupt_sidecar_quarantines_instead_of_deleting() { + let _guard = DurabilityGuard::wal(); + let dir = tmp("sidecar-quarantine"); + let h = Harness::boot(&dir, Some(1), 1).unwrap(); + create_stream(&h.store, "q", OCTET).await; + append_acked(&h.store, "q", OCTET, b"precious|").await; + let data_path = h.store.get("q").unwrap().file_path.clone(); + let meta_path = std::path::PathBuf::from(format!("{}.meta", data_path.display())); + h.crash(); + + // Tear the sidecar (simulates a crash-torn rename target). + std::fs::write(&meta_path, b"{ this is not json").unwrap(); + + let h2 = Harness::boot(&dir, None, 1).unwrap(); + assert!(h2.store.get("q").is_none(), "stream is skipped this boot"); + assert!(data_path.exists(), "data file must NOT be deleted"); + assert!( + meta_path.with_extension("meta.corrupt").exists(), + "sidecar parked as .meta.corrupt for repair" + ); + h2.crash(); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Recovery-hardening: on an initialized store, a stream lane whose dir is +/// empty and unmarked (= its device mount is missing) must REFUSE to boot — +/// continuing would drop the lane's streams and let the WAL reset destroy +/// their acked records. +#[tokio::test] +async fn e2e_missing_lane_mount_refuses_boot() { + let _guard = DurabilityGuard::wal(); + crate::store::set_stream_lanes(3); + let dir = tmp("lane-mount-guard"); + { + let h = Harness::boot(&dir, Some(1), 1).unwrap(); + create_stream(&h.store, "lm", OCTET).await; + append_acked(&h.store, "lm", OCTET, b"x|").await; + h.crash(); + } + // Simulate a missing mount: replace lane 1's dir with a fresh empty dir. + let lane1 = dir.join("streams").join("1"); + std::fs::remove_dir_all(&lane1).unwrap(); + std::fs::create_dir_all(&lane1).unwrap(); + + let err = Store::new_with_tier(dir.clone(), TierConfig::default()) + .err() + .expect("boot must refuse when a lane mount is missing"); + assert!( + err.to_string().contains("mount"), + "error should name the missing mount: {err}" + ); + crate::store::set_stream_lanes(1); + let _ = std::fs::remove_dir_all(&dir); +} + /// `--stream-lanes N`: stream data files hash across `streams/<0..N>/` subdirs /// (one per device in the intended deployment — the ~1M-stream writeback-wall /// fix). Crash recovery must find every file in its lane dir, and the diff --git a/packages/durable-streams-rust/src/wal/shard.rs b/packages/durable-streams-rust/src/wal/shard.rs index eefbf3208d..4f123a597a 100644 --- a/packages/durable-streams-rust/src/wal/shard.rs +++ b/packages/durable-streams-rust/src/wal/shard.rs @@ -228,20 +228,6 @@ impl CommitSignal { g.work_pending = false; g.stop } - - /// Error-backoff wait: park up to `timeout`, returning early if stop is - /// requested. Returns whether `stop` was requested. Does **not** clear - /// `work_pending` — we want to retry the same un-acked watermark, and a - /// pending stage should still drive the next wake. Using the condvar (rather - /// than `thread::sleep`) lets shutdown interrupt a long backoff. - fn backoff_wait(&self, timeout: std::time::Duration) -> bool { - let g = self.state.lock().unwrap(); - if g.stop { - return true; - } - let (g, _timed_out) = self.cv.wait_timeout(g, timeout).unwrap(); - g.stop - } } /// Handle to a shard's dedicated committer OS thread. Signalling stop + joining @@ -492,7 +478,12 @@ impl Shard { Some((start, path)) => (start, Arc::new(FileSegment::open_existing(path)?)), None => ( 1, - Arc::new(FileSegment::create(seg_path(&dir, 1), segment_size)?), + { + let seg = Arc::new(FileSegment::create(seg_path(&dir, 1), segment_size)?); + // Durable dirent for the fresh segment (see roll-site note). + crate::store::fsync_parent_dir(&seg_path(&dir, 1))?; + seg + }, ), }; Ok(std::sync::Arc::new(Shard { @@ -583,6 +574,11 @@ impl Shard { seg_path(&self.dir, seg_start_lsn), self.segment_size, )?); + // Barrier the shard DIR: reset unlinked the old segments and created a + // fresh one under the SAME name — without a dir fsync a crash can leave + // the dirent pointing at the OLD inode, and the next recovery would + // replay stale pre-reset records as if they were the fresh log. + crate::store::fsync_parent_dir(&seg_path(&self.dir, 1))?; let mut g = self.inner.lock().unwrap(); g.active = active; g.seg_start_lsn = seg_start_lsn; @@ -616,6 +612,15 @@ impl Shard { stream_offset: u64, payload: &[u8], ) -> io::Result { + // Test-only fault injection: simulate a stage failure. Fires BEFORE the + // lsn reservation: production write failures retry-then-abort (see + // below), so a reserved-but-unwritten gap is unreachable in production — + // the hook models the reachable outcome (a clean Err, no lsn consumed). + #[cfg(test)] + if self.fail_next_write.swap(false, std::sync::atomic::Ordering::SeqCst) { + return Err(io::Error::other("injected WAL stage failure")); + } + // Test-only ordering seam (CQ-1): fires before any lsn is reserved, i.e. // before this record can ever become durable. A test uses it to assert the // stream was already registered dirty by the caller (register-before-stage). @@ -668,7 +673,22 @@ impl Shard { // packed seam). The committer additionally fsyncs this sealed // segment via `sealed_pending` to cover any record still being // written into it off-lock (see `sealed_pending` docs). - g.active.seal_to(g.write_pos)?; + if let Err(e) = g.active.seal_to(g.write_pos) { + // FAIL-STOP: seal_to is a durability barrier (set_len + + // fsync). A failed one retried on the next overflowing + // append can falsely succeed (kernel consumed the error); + // after a crash the un-truncated segment reads with a + // grafted zero tail that silently ENDS replay — discarding + // every acked record in every later segment. Nothing acked + // is at risk at abort time (no lsn reserved yet). + eprintln!( + "FATAL: WAL segment seal (truncate+fsync) failed for shard {:?}: {e}. \ + Aborting: a retried seal can falsely succeed and a later \ + crash would silently truncate the replayed log.", + self.dir + ); + std::process::abort(); + } let next_lsn = g.next_lsn; // The highest lsn already reserved in the old segment is // `next_lsn - 1` (the rolling record itself, `next_lsn`, lands in @@ -681,6 +701,11 @@ impl Shard { seg_path(&self.dir, next_lsn), self.segment_size, )?); + // Make the new segment's DIR ENTRY durable: fdatasync of record + // bytes persists file data, not the dirent — without this, a + // crash can orphan the inode and replay never sees the segment + // (acked records in it silently vanish). + crate::store::fsync_parent_dir(&seg_path(&self.dir, next_lsn))?; g.active = new_seg; g.seg_start_lsn = next_lsn; g.write_pos = 0; @@ -703,13 +728,32 @@ impl Shard { &Record { lsn, kind, stream_id, stream_offset, payload }, ); - // Test-only fault injection: simulate a write_at failure. - #[cfg(test)] - if self.fail_next_write.swap(false, std::sync::atomic::Ordering::SeqCst) { - return Err(io::Error::other("injected WAL segment write_at failure")); - } - seg.write_at(off, &buf)?; + // A failed positioned WRITE is soundly retryable (unlike fsync: the + // encoded bytes are still in hand) — but an UNRECOVERED failure leaves a + // permanent gap lsn that freezes the shard's contiguous watermark + // forever: every later append would stage fine and then hang in + // wait_durable with no ack and no error (silent whole-shard wedge). + // Bounded retries close the transient case; a persistent write error is + // fail-stop, mirroring the committer's barrier policy. + let mut write_result = seg.write_at(off, &buf); + for _ in 0..2 { + if write_result.is_ok() { + break; + } + std::thread::sleep(std::time::Duration::from_millis(5)); + write_result = seg.write_at(off, &buf); + } + if let Err(e) = write_result { + eprintln!( + "FATAL: WAL segment write failed for shard {:?} lsn {lsn} after retries: {e}. \ + Aborting: the reserved lsn range would otherwise become a permanent \ + gap that silently wedges the shard's durability watermark (all later \ + appends would hang un-acked forever).", + self.dir + ); + std::process::abort(); + } { let mut g = super::telemetry::timed_lock( |ns| self.stats.record_inner_lock_wait(ns), @@ -845,6 +889,35 @@ impl Shard { // checkpoint_lsn → recycle. Acks never gate on any of this. let this = Arc::clone(self); tokio::task::spawn_blocking(move || -> io::Result { + let result = Self::checkpoint_blocking(&this, &drained, checkpoint_lsn); + if result.is_err() { + // A failed checkpoint must NOT drop the drained dirty set: these + // streams' tails proofs were never persisted, and nothing + // re-registers them until their NEXT append — a later successful + // checkpoint would recycle their WAL records anyway and a + // subsequent restart would truncate acked bytes back to a stale + // frontier. Re-register so the next checkpoint re-barriers and + // re-records their tails. (register_dirty is epoch-CAS'd, so a + // stream that re-appended meanwhile is not double-added.) + for st in drained.iter() { + this.register_dirty(st.id, Arc::clone(st)); + } + } + result + }) + .await + .expect("checkpoint task panicked") + } + + /// The blocking body of [`Self::checkpoint`]: capture → barrier → tails → + /// checkpoint_lsn → recycle → sidecars. Split out so the caller can + /// re-register `drained` on error (see checkpoint()). + fn checkpoint_blocking( + this: &Arc, + drained: &[Arc], + checkpoint_lsn: u64, + ) -> io::Result { + { // Phase timing for the `WAL_CKPT` line (`--wal-stats`). One clock // read per phase, once per ~3 s per shard — nowhere near the hot path. let t_start = std::time::Instant::now(); @@ -874,12 +947,30 @@ impl Shard { // syncfs off) fallback. Serial on purpose: parallel fan-out // regressed in every controlled test (it steals device budget // from the commit path). - if cfg!(target_os = "linux") && n_touched > 0 { - crate::store::syncfs_stream_lanes(&touched[0].2)?; + // FAIL-STOP on barrier error: like the committer's fdatasync, a + // failed syncfs/fsync must not be retried in place — the kernel + // consumes the error and drops the dirty pages, so the NEXT + // checkpoint's barrier can falsely succeed and recycle the WAL + // records that were the only durable copy (acked-data loss). + // Nothing acked is at risk at abort time: acks never gate on the + // checkpoint, and the WAL still holds every record. + let barrier = if cfg!(target_os = "linux") && n_touched > 0 { + crate::store::syncfs_stream_lanes(&touched[0].2) } else { - for (_, _, f) in &touched { - crate::store::barrier_fsync(f)?; - } + touched + .iter() + .try_for_each(|(_, _, f)| crate::store::barrier_fsync(f)) + }; + if let Err(e) = barrier { + eprintln!( + "FATAL: checkpoint durability barrier failed for shard {:?}: {e}. \ + Aborting: a retried barrier can falsely succeed (the kernel \ + drops dirty pages and clears the error) and the next \ + checkpoint would recycle the WAL over lost bytes. Restart \ + recovery replays the retained WAL.", + this.dir + ); + std::process::abort(); } let t_fsync = t_start.elapsed(); @@ -919,7 +1010,7 @@ impl Shard { // must never delay the recycle floor. Errors are ignored exactly // like the debounced flush ignored them. let mut n_meta = 0u64; - for st in &drained { + for st in drained { if st.meta_dirty.swap(false, Ordering::AcqRel) { let _ = crate::store::write_meta_sync(st, false); n_meta += 1; @@ -944,9 +1035,7 @@ impl Shard { } Ok(checkpoint_lsn) - }) - .await - .expect("checkpoint task panicked") + } } /// Merge `touched` `(stream_id, durable_tail)` pairs into the persisted @@ -997,6 +1086,11 @@ impl Shard { // crash-durable BEFORE recycle (the whole point of 11b). std::fs::File::open(&tmp)?.sync_all()?; std::fs::rename(&tmp, &path)?; + // The rename is crash-durable only once the shard DIR entry is fsynced — + // and recycle's unlinks in the same tick hit this same directory. Without + // this barrier a crash could persist the unlinks but not the rename: + // stale tails proof + WAL gone = recovery truncates acked bytes. + crate::store::fsync_parent_dir(&path)?; Ok(n) } @@ -1393,44 +1487,33 @@ impl Shard { /// we always re-snapshot the watermark off-lock after waking. A stage racing /// the park is therefore never lost (see [`CommitSignal`]). /// - /// **fsync-error path:** on `fdatasync` failure we do **not** advance - /// `durable_lsn` (no ack) and back off with bounded exponential delay - /// (interruptible by shutdown), exactly as the no-loss invariant requires. + /// **fsync-error path: FAIL-STOP.** A failed WAL `fdatasync` must never be + /// retried in place: on Linux (>=4.13) a failed fsync marks the dirty pages + /// clean and consumes the fd's error state, so a retry on the same fd can + /// return success without the bytes ever reaching stable storage — the + /// retry would then publish `durable_lsn` over lost bytes and ack them + /// (the PostgreSQL "fsyncgate" failure mode). Aborting here is provably + /// safe: `durable_lsn` was not advanced, so everything at risk is un-acked, + /// and restart replay ends at the torn point exactly as a crash would. /// /// **Shutdown:** on a stop signal the loop performs a **final drain** /// (commits everything already contiguous-written, so in-flight commits are - /// not dropped) and then returns so the thread can be joined. + /// not dropped) and then returns so the thread can be joined. A drain-time + /// fsync error gives up WITHOUT acking (no-loss holds; the process is + /// exiting anyway, so no abort is needed to prevent a later false ack). pub fn run_committer(&self) { - // Backoff state for the fsync-error path: a persistently failing disk - // (ENOSPC/EIO/read-only volume) must not busy-spin a core, hammer the disk, - // and flood stderr. The no-loss invariant holds throughout — `durable_lsn` - // is never advanced on failure, so nothing is ever acked. - const RETRY_BACKOFF_MIN: std::time::Duration = std::time::Duration::from_millis(5); - const RETRY_BACKOFF_MAX: std::time::Duration = std::time::Duration::from_secs(1); - const LOG_EVERY: u64 = 100; - let mut backoff = RETRY_BACKOFF_MIN; - let mut consecutive_errors: u64 = 0; loop { match self.commit_once() { Ok(Some(_)) => { // Advanced — re-snapshot immediately; more may have arrived // while we were fsyncing (group commit naturally batches them). - consecutive_errors = 0; - backoff = RETRY_BACKOFF_MIN; continue; } Ok(None) => { - // Caught up — reset the error backoff and park for the next - // stage (or a stop signal). - consecutive_errors = 0; - backoff = RETRY_BACKOFF_MIN; + // Caught up — park for the next stage (or a stop signal). if self.commit_signal.wait_for_work() { // Stop requested: drain any records that became - // contiguous-written before/at the stop, then exit. We do - // NOT retry fsync errors forever here — on error we log and - // give up (no-loss holds: durable_lsn is not advanced, so - // the un-drained tail simply stays un-acked, exactly as a - // crash would leave it). + // contiguous-written before/at the stop, then exit. loop { match self.commit_once() { Ok(Some(_)) => continue, @@ -1447,23 +1530,17 @@ impl Shard { } } Err(e) => { - // Rate-limited log + bounded exponential backoff. The backoff - // parks on the condvar (not a raw sleep) so shutdown can - // interrupt it; a stream of concurrent stages cannot turn the - // retry into a hot loop (we do not clear `work_pending` here, - // so the same un-acked watermark is retried). - consecutive_errors += 1; - if consecutive_errors == 1 || consecutive_errors % LOG_EVERY == 0 { - eprintln!( - "WAL committer fdatasync failed (attempt {consecutive_errors}): {e}" - ); - } - if self.commit_signal.backoff_wait(backoff) { - // Stop requested during backoff: exit without acking the - // failing watermark (no-loss preserved). - return; - } - backoff = (backoff * 2).min(RETRY_BACKOFF_MAX); + // FAIL-STOP (see doc above): no in-place fsync retry is + // sound on Linux, and the design keeps no in-memory copy to + // re-write the range. Nothing at risk was acked. + eprintln!( + "FATAL: WAL committer fdatasync failed for shard {:?}: {e}. \ + Aborting: a retried fsync can falsely succeed (the kernel \ + drops dirty pages and clears the error) and would ack lost \ + bytes. Restart recovery replays the durable log.", + self.dir + ); + std::process::abort(); } } } @@ -1476,7 +1553,29 @@ impl Shard { let me = Arc::clone(self); let join = std::thread::Builder::new() .name("wal-committer".to_string()) - .spawn(move || me.run_committer()) + .spawn(move || { + // SUPERVISION: a committer that dies of a PANIC (poisoned lock, + // bug) must not leave a half-alive server — durable_lsn freezes, + // every in-flight wait_durable parks forever, appends keep being + // staged and their connections never release their permits, and + // the server degrades into a silent total denial of service. + // Fail-stop instead: nothing at risk was acked (durable_lsn only + // advances on a successful barrier), so restart recovery is + // strictly better than the frozen half-life. A normal return + // (stop-signal shutdown drain) is NOT a failure. + let dir = me.dir.clone(); + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + me.run_committer() + })); + if r.is_err() { + eprintln!( + "FATAL: WAL committer thread for shard {dir:?} panicked. \ + Aborting: a dead committer freezes durable_lsn and turns \ + every append into a silent, permanent hang." + ); + std::process::abort(); + } + }) .expect("spawn WAL committer thread"); CommitterHandle { shard: Arc::clone(self), @@ -1814,28 +1913,24 @@ mod tests { #[tokio::test] async fn write_error_fails_the_stage_without_panicking() { - // A transient WAL `write_at` failure must surface as a `Result::Err` - // (so the caller fails the ack) — NOT a process panic. The reserved lsn - // stays a permanent gap, so a LATER good stage can never become durable - // past it (the committer's contiguous watermark is blocked). + // An injected stage failure must surface as a `Result::Err` (so the + // caller fails the ack) — NOT a process panic — and must consume no + // lsn: the next good stage proceeds and becomes durable normally. + // (Production `write_at` failures after reservation retry then + // fail-stop, so a reserved-but-unwritten gap is unreachable; the pure + // gap-blocks-watermark invariant is covered by the reserve_only test + // above.) let sh = Shard::open(tmp("write-err")).unwrap(); sh.fail_next_write(); let err = sh.reserve_and_stage(RecordKind::Append, 1, 0, b"boom"); - assert!(err.is_err(), "an injected write_at failure must return Err, not panic"); + assert!(err.is_err(), "an injected stage failure must return Err, not panic"); - // The failed lsn (1) was reserved but never written → permanent gap. A - // subsequent good stage (lsn 2) is on disk, but the committer must NOT - // advance durable_lsn past the gap at lsn 1. - let l2 = sh.reserve_and_stage(RecordKind::Append, 1, 4, b"ok").unwrap(); - assert_eq!(l2, 2, "the failed stage still consumed lsn 1 (it stays a gap)"); + let l1 = sh.reserve_and_stage(RecordKind::Append, 1, 0, b"ok").unwrap(); + assert_eq!(l1, 1, "the failed stage consumed no lsn"); let h = sh.spawn_committer(); - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - assert_eq!( - sh.durable_lsn(), - 0, - "durable_lsn cannot advance past the unwritten (failed) lsn-1 gap" - ); + sh.wait_durable(l1).await; + assert!(sh.durable_lsn() >= l1, "the shard is fully functional after the error"); h.stop(); } diff --git a/packages/durable-streams-rust/src/wal/walset.rs b/packages/durable-streams-rust/src/wal/walset.rs index b8fe5609ea..e40a97904e 100644 --- a/packages/durable-streams-rust/src/wal/walset.rs +++ b/packages/durable-streams-rust/src/wal/walset.rs @@ -107,7 +107,20 @@ impl WalSet { None => { // Fresh data dir: persist the requested N, or the caller's default. let n = requested_n.unwrap_or(default_n).max(1); - std::fs::write(&shards_path, n.to_string())?; + // Durable write (tmp + sync + rename + dir fsync): a lost + // `shards` file next to surviving shard dirs would re-derive a + // DIFFERENT N on the next boot — shard dirs >= N would silently + // never be opened or replayed (acked-data loss) and stream->shard + // routing would shift. + let tmp = wal_dir.join("shards.tmp"); + { + use std::io::Write; + let mut f = std::fs::File::create(&tmp)?; + f.write_all(n.to_string().as_bytes())?; + f.sync_all()?; + } + std::fs::rename(&tmp, &shards_path)?; + crate::store::fsync_parent_dir(&shards_path)?; n } }; From abca3a5f7327dd60a6ffd615d17321d034eba868 Mon Sep 17 00:00:00 2001 From: Valter Balegas Date: Tue, 14 Jul 2026 12:44:52 +0100 Subject: [PATCH 2/2] chore: changeset for recovery hardening Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y3x7bcT9vLGiT4tXZeQpnk --- .changeset/recovery-hardening.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/recovery-hardening.md diff --git a/.changeset/recovery-hardening.md b/.changeset/recovery-hardening.md new file mode 100644 index 0000000000..c2379a0baa --- /dev/null +++ b/.changeset/recovery-hardening.md @@ -0,0 +1,5 @@ +--- +"@electric-ax/durable-streams-server-rust": patch +--- + +Recovery hardening: durability barriers (committer fdatasync, checkpoint syncfs, segment seal) are fail-stop instead of retryable-in-place (a retried fsync can falsely succeed on Linux and ack/recycle lost bytes); failed checkpoints re-register their dirty set (previously a transient error + restart truncated acked bytes); torn sidecars are quarantined instead of deleting the stream's data file; missing stream-lane mounts refuse to boot instead of letting the WAL reset destroy the lane's records; append stage failures roll back the data write and producer state (500'd bytes no longer resurrect; retries no longer swallowed as duplicates); sealing cuts at the durable frontier; unreadable sealed chunks fail the read instead of serving a response with missing interior bytes; dir fsyncs added across the WAL metadata lifecycle.