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.
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<Store>, 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<Store>, 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<Store>, st: &Arc<StreamState>, lsn: u64) {
 /// awaited in `wait_durable_lsn`.
 fn write_wire(st: &StreamState, ap: &mut Appender, wire: &Bytes) -> std::io::Result<u64> {
     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<Store>, 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<Store>, 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<Self> {
         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<String, (Meta, PathBuf)> = HashMap::new();
         let mut data_files: Vec<PathBuf> = Vec::new();
+        let mut quarantined: Vec<PathBuf> = Vec::new();
         let mut entries: Vec<PathBuf> = 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::<Meta>(&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::<Meta>(&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<MetaSegment>, 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<ResolvedSlice>) -> Result<Vec<Segment>, 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<StreamState>, lo: u64, hi: u64, out: &mut Vec<Resolve
                 // lock), so we always open the chunk file *before* it is unlinked
                 // (and the fd stays valid after unlink). Do not move the open
                 // outside the lock: that reintroduces an open-vs-unlink race.
-                if let Ok(f) = OpenOptions::new().read(true).open(path) {
-                    out.push(ResolvedSlice::Local(Segment {
+                match OpenOptions::new().read(true).open(path) {
+                    Ok(f) => 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<u64> {
+        // 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<u64> {
+            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<Self>,
+        drained: &[Arc<StreamState>],
+        checkpoint_lsn: u64,
+    ) -> io::Result<u64> {
+        {
             // 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
             }
         };