kernel-wal: fsyncs fail loudly; poison survives restart (round 2)

crates/busbar-kernel-wal/Cargo.toml

@@ -6,9 +6,11 @@
 # to end. Folding it into the ledger would mix "what the money is" with "how the bytes landed".
 #
 # DEPENDENCIES. `busbar-contract`, because the one thing the log hands back on a failure is a
-# `DurabilityLost`, and only a holder of the durability token may mint one. `sha2`, because a torn
-# tail is truncated at recovery to the last record whose digest verifies, and a digest is not a
-# thing to hand-roll. Nothing else: no async runtime, no serializer, no allocator opinion.
+# `DurabilityLost`, and only a holder of the durability token may mint one. `ring`, for SHA-256:
+# a torn tail is truncated at recovery to the last record whose digest verifies, the journal chains
+# on the same digest, and a digest is not a thing to hand-roll; ring is the workspace's one crypto
+# backend. Nothing else: no async runtime, no serializer, no allocator opinion. `sha2` is a
+# dev-dependency only, the independent witness the tests check ring's digests against.
 [package]
 name = "busbar-kernel-wal"
 version = "1.6.0"

crates/busbar-kernel-wal/src/backend.rs

@@ -436,36 +436,6 @@ impl SegmentBackend for FileSegment {
     }
 }
 
-/// Fsync the directory HOLDING `path`, which is what makes `path`'s own directory entry durable.
-///
-/// Best effort by construction: a filesystem that refuses to open a directory as a file (or refuses
-/// to fsync one) cannot be made to promise more than it does, and failing construction of the log
-/// over it would trade a weaker durability story for no log at all. The promise this crate makes —
-/// a sync that RETURNS SUCCESS was really durable — is carried by [`SegmentBackend::sync`], which
-/// does report its failure.
-fn sync_holding_dir(path: &Path) {
-    if let Some(parent) = path.parent() {
-        let parent = if parent.as_os_str().is_empty() {
-            Path::new(".")
-        } else {
-            parent
-        };
-        #[cfg(test)]
-        DIR_SYNCS.with(|c| c.borrow_mut().push(parent.to_path_buf()));
-        if let Ok(dir) = std::fs::File::open(parent) {
-            let _ = dir.sync_all();
-        }
-    }
-}
-
-#[cfg(test)]
-thread_local! {
-    /// Every directory this thread fsynced to make an entry in it durable, in order — the
-    /// observation the segment-creation battery reads. Test builds only.
-    pub(crate) static DIR_SYNCS: std::cell::RefCell<Vec<PathBuf>> =
-        const { std::cell::RefCell::new(Vec::new()) };
-}
-
 /// Hands out file segments named `<index>.wal` inside one directory.
 #[derive(Debug)]
 pub struct DirectoryFactory {
@@ -477,19 +447,15 @@ impl DirectoryFactory {
     /// this type is already the decision to write to a disk. A node with no data directory never
     /// builds one, which is why it leaves no files behind.
     ///
-    /// The creation is made DURABLE here rather than left to `create_dir_all`'s default, which
-    /// leaves the new directory ENTRY unfsynced: a power loss right after the first group commit
-    /// could then take the whole directory with it, segments included, even though every segment's
-    /// own `sync` returned success. A log that reports a durable write and then loses it to its own
-    /// holding directory has broken the one promise this crate exists to keep, so the parent gets
-    /// the fsync that makes the entry survive.
+    /// The creation is made DURABLE here, through [`create_dir_durably`], rather than left to
+    /// `std::fs::create_dir_all`, which leaves every new directory ENTRY unfsynced: a power
+    /// loss right after the first group commit could then take the directory with it — or an
+    /// ancestor this call created, and the directory under it — segments included, even though
+    /// every segment's own `sync` returned success. Every directory created gets the fsync of the
+    /// directory holding it, and a failed fsync is this call's error.
     pub fn new(dir: impl Into<PathBuf>) -> io::Result<Self> {
         let dir = dir.into();
-        let fresh = !dir.exists();
-        std::fs::create_dir_all(&dir)?;
-        if fresh {
-            sync_holding_dir(&dir);
-        }
+        create_dir_durably(&dir)?;
         Ok(DirectoryFactory { dir })
     }
 
@@ -520,11 +486,21 @@ impl SegmentFactory for DirectoryFactory {
     /// missing from the directory, taking acknowledged records with it. So a segment this call
     /// created has its holding directory fsynced here, once, before anything can be written to it;
     /// one that already existed was made durable by whoever created it.
+    ///
+    /// A directory fsync that fails is this call's error, and the file it created is removed
+    /// again: left in place, the next open would find it existing, skip the fsync, and hand out a
+    /// segment whose entry nothing ever made durable.
     fn open(&mut self, index: u64) -> io::Result<Box<dyn SegmentBackend>> {
         let path = self.segment_path(index);
         let (segment, created) = FileSegment::open_reporting(&path)?;
         if created {
-            sync_holding_dir(&path);
+            if let Err(e) = sync_holding_dir(&path) {
+                drop(segment);
+                // The fsync's error is the one returned; a removal that fails too leaves an empty
+                // file the next open reads as a segment with nothing in it.
+                let _removed = std::fs::remove_file(&path);
+                return Err(e);
+            }
         }
         Ok(Box::new(segment))
     }
@@ -560,7 +536,7 @@ impl SegmentFactory for DirectoryFactory {
     /// `<segment>.quarantine-<unix_ms>` beside the segment, created fresh — never over an existing
     /// file, so a second recovery of the same damage writes a second copy rather than replacing the
     /// first. The file's data is synced, then the directory holding it, and only then does this
-    /// return. The name does not end in `.wal`, so [`DirectoryFactory::highest_index`] never reads
+    /// return; a failure of either is this call's error, so recovery does not cut. The name does not end in `.wal`, so [`DirectoryFactory::highest_index`] never reads
     /// a quarantine as a segment.
     fn quarantine(
         &mut self,
@@ -601,7 +577,83 @@ impl SegmentFactory for DirectoryFactory {
         file.write_all(bytes)?;
         // The full sync, not the data sync: the file is new, and its length is metadata.
         file.sync_all()?;
-        sync_holding_dir(&path);
+        sync_holding_dir(&path)?;
         Ok(Some(path))
     }
 }
+
+// ── Directory durability ─────────────────────────────────────────────────────────────────────────
+// The rule the loader's `durable` module states for every other file busbar publishes, kept here
+// for the log's own segment files and quarantine copies: this crate takes no edge onto the loader
+// (lane-gb-loader-wal-edge), so the log keeps its own copy of the one rule rather than a dependency.
+
+/// The directory whose ENTRY creating or removing `path` changes. A relative `path` with no parent
+/// resolves to ".", where it actually lives.
+fn holding_dir(path: &Path) -> &Path {
+    path.parent()
+        .filter(|p| !p.as_os_str().is_empty())
+        .unwrap_or_else(|| Path::new("."))
+}
+
+/// fsync the directory that holds `path`, so creating it is itself durable.
+///
+/// A failure is the caller's error: opening the directory, or fsyncing it, failing means the entry
+/// is NOT durable, and a segment or a quarantine copy reported made would be a durability the
+/// medium did not give. The one exception is a filesystem that says the operation is not supported
+/// on a directory (`EINVAL`, or an error the platform reports as unsupported): it cannot promise
+/// more than it does, and refusing every segment on it would leave no log at all.
+fn sync_holding_dir(path: &Path) -> io::Result<()> {
+    match sync_dir(holding_dir(path)) {
+        Err(e) if fsync_unsupported(&e) => Ok(()),
+        other => other,
+    }
+}
+
+/// fsync `dir` itself, every failure reported. [`sync_holding_dir`] decides which are the caller's.
+fn sync_dir(dir: &Path) -> io::Result<()> {
+    #[cfg(test)]
+    crate::tests::hooks::parent_fsync(dir)?;
+    #[cfg(unix)]
+    {
+        std::fs::File::open(dir)?.sync_all()
+    }
+    // There is no directory-entry barrier to reach for on this platform.
+    #[cfg(not(unix))]
+    {
+        let _no_barrier = dir;
+        Ok(())
+    }
+}
+
+/// Whether `e` is a filesystem saying a directory cannot be fsynced at all, as opposed to failing to
+/// do it.
+fn fsync_unsupported(e: &io::Error) -> bool {
+    matches!(
+        e.kind(),
+        io::ErrorKind::Unsupported | io::ErrorKind::InvalidInput
+    )
+}
+
+/// DURABLY create `path` and any missing ancestors: each directory actually created has its PARENT
+/// fsynced, so the new entry itself survives a power loss, shallowest first. Directories that
+/// already exist are left alone. A directory fsync that fails is this call's error.
+fn create_dir_durably(path: &Path) -> io::Result<()> {
+    let mut missing: Vec<&Path> = Vec::new();
+    let mut cur = Some(path);
+    while let Some(p) = cur {
+        if p.as_os_str().is_empty() || p.exists() {
+            break;
+        }
+        missing.push(p);
+        cur = p.parent();
+    }
+    for dir in missing.iter().rev() {
+        match std::fs::create_dir(dir) {
+            Ok(()) => sync_holding_dir(dir)?,
+            // A concurrent creator won the race; the entry is theirs to make durable.
+            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {}
+            Err(e) => return Err(e),
+        }
+    }
+    Ok(())
+}

crates/busbar-kernel-wal/src/journal.rs

@@ -730,14 +730,9 @@ impl std::fmt::Debug for Journal {
 }
 
 impl Journal {
-    /// A journal over a memory-buffered log that ships nowhere and touches no disk.
-    #[must_use]
-    pub fn memory_buffered(node: u64, clock: Clock) -> Self {
-        Journal::over(Wal::memory_buffered(clock), node)
-    }
-
-    /// A journal over a memory-buffered log shipping through `shipper` — the shape a deployment that
-    /// names a store but no data directory runs.
+    /// A journal over a memory-buffered log shipping through `shipper` — the default, and the shape a
+    /// deployment that names no data directory runs. It starts a fresh chain each process: what such
+    /// a node keeps across a restart is what the previous release kept, not the journal.
     #[must_use]
     pub fn memory_buffered_to(node: u64, shipper: Box<dyn Shipper<Record>>, clock: Clock) -> Self {
         Journal::over(Wal::memory_buffered_to(shipper, clock), node)
@@ -823,25 +818,6 @@ impl Journal {
         journal
     }
 
-    /// A journal continuing a chain somebody else is holding — the store, on a node that keeps no
-    /// data directory and whose own buffer therefore did not survive the restart.
-    ///
-    /// The head and the sequence number come from the records that were SHIPPED, which is the only
-    /// honest place for them on such a node: what the store acknowledged is what exists.
-    #[must_use]
-    pub fn resuming(log: Wal, node: u64, head: [u8; 32], next_seq: u64) -> Self {
-        let mut journal = Journal::over(log, node);
-        journal.head = head;
-        journal.next_seq = next_seq.max(1);
-        // The caller's number is where the STORE got to, and on the node this constructor is for the
-        // log holds nothing, so the two agree. Where they do not — a store that acknowledged less
-        // than the local log took — the caller's number would re-use identities the log already
-        // carries, and the log answers a duplicate identity by passing over the record and reporting
-        // success. The floor is what keeps that from being a settlement lost behind an `Ok`.
-        journal.floor_seq_to_the_log();
-        journal
-    }
-
     /// A memory-buffered journal RESUMED from the chain the configured store kept: `records` are
     /// that chain's records as the store returned them, oldest first. They go into the log's buffer
     /// without being shipped again ([`Wal::memory_seeded`]), so [`Journal::replay`] answers them, and
@@ -866,7 +842,15 @@ impl Journal {
             .ok()
             .and_then(|decoded| tail_of(&decoded, node))
             .unwrap_or(([0u8; 32], 1));
-        Ok(Journal::resuming(log, node, head, next_seq))
+        let mut journal = Journal::over(log, node);
+        journal.head = head;
+        journal.next_seq = next_seq.max(1);
+        // The number is where the store's chain got to, and the seeded log holds that chain, so the
+        // two agree. Where they do not, the store's number would re-use identities the log already
+        // carries, and the log answers a duplicate identity by passing over the record and reporting
+        // success. The floor is what keeps that from being a settlement lost behind an `Ok`.
+        journal.floor_seq_to_the_log();
+        Ok(journal)
     }
 
     /// The same journal KEEPING every record at its bound instead of dropping the oldest.
@@ -1006,8 +990,8 @@ impl Journal {
     /// # Errors
     ///
     /// The log could not make the batch durable. Without a data directory that means the store
-    /// refused it; with one it means a write or a sync failed. Either way the records are retained
-    /// and re-offered on the next append, up to the bound.
+    /// refused it; with one it means a write, a sync or the roll to a fresh segment failed. Either
+    /// way the records are retained and re-offered on the next append, up to the bound.
     pub fn append(
         &mut self,
         token: &Grant<DurableWrite>,

crates/busbar-kernel-wal/src/lib.rs

@@ -18,16 +18,18 @@
 //!
 //! ## The four rules
 //!
-//! **Records are fixed-size frames.** Every frame is 512 bytes on the medium, header and payload
-//! together. A body longer than one frame's payload area continues into further frames carrying the
+//! **Records are fixed-size frames.** Every frame is 512 bytes on the medium, header, payload and
+//! end mark together. A body longer than one frame's payload area continues into further frames carrying the
 //! same writer identity and an increasing part index. Nothing about finding the next frame depends
 //! on a length field a reader might have read out of a half-written region.
 //!
 //! **A torn tail is cut at recovery to the last record that verifies.** Whole records, not whole
-//! frames: a record whose last part was lost is absent, not short. The battery for this truncates a
-//! written segment at every single byte offset, recovers, and asserts the result is the longest
-//! complete prefix — there is no interesting subset of the offsets a crash can stop at, so none is
-//! chosen.
+//! frames: a record whose last part was lost is absent, not short. Torn means what a crash leaves —
+//! the last group commit, no subsequent commit vouching for it, damaged only by writes that never landed
+//! or stopped part way — and nothing else: an acknowledged record that fails a check is quarantined,
+//! never cut. The battery for this truncates a written segment at every single byte offset, and
+//! zeroes it from every byte offset on, recovers, and asserts the result is the longest complete
+//! prefix — there is no interesting subset of the offsets a crash can stop at, so none is chosen.
 //!
 //! **A group commit is one write and one sync.** Records are framed into one buffer and put down
 //! with a single positional write, then made durable with a single data sync. Whatever the batch
@@ -36,14 +38,17 @@
 //! **A failed sync poisons the segment.** Permanently. Every byte after the last good commit is of
 //! unknown state, and the only safe reading of unknown is absent. The caller is handed a
 //! [`busbar_contract::caps::DurabilityLost`] — which only a holder of the durability token can mint, so the
-//! claim "a durable write was lost" is not something any other part of the system can invent — and
-//! the batch that failed is written again, along with the one after it, on a fresh segment.
+//! claim "a durable write was lost" is not something any other part of the system can invent — the
+//! failed batch's bytes are cut off the segment, and the batch is written again, along with the one
+//! after it, on a fresh segment. That fresh segment is opened the moment the sync fails, and its
+//! directory entry is the durable record that the poisoned one is finished: a restart appends to
+//! the newest segment, so a segment that lost a sync is never written to again.
 //!
 //! ## The default is no disk at all
 //!
-//! [`Wal::memory_buffered`] is the default, and it is what a deployment that names no data directory
-//! gets. There is no directory probe, no file, no preallocation and no boot warning. It is not a
-//! degraded mode with the file writing switched off; a memory-buffered log holds no directory
+//! [`Wal::memory_buffered_to`] is the default, and it is what a deployment that names no data
+//! directory gets. There is no directory probe, no file, no preallocation and no boot warning. It is
+//! not a degraded mode with the file writing switched off; a memory-buffered log holds no directory
 //! factory, so there is no code path from it to a file system. The battery for this runs a whole
 //! append-and-recover cycle with a temporary directory in view and asserts the directory is still
 //! empty afterwards.
@@ -91,15 +96,15 @@ pub use journal::{
     JOURNAL_HEADER_BYTES, JOURNAL_MAGIC, JOURNAL_VERSION, MEMORY_BUFFER_RECORDS,
 };
 pub use record::{
-    decode_frame, FrameError, FrameHeader, Record, FRAME_BYTES, FRAME_HEADER_BYTES, FRAME_MAGIC,
-    FRAME_PAYLOAD_BYTES, FRAME_VERSION,
+    decode_frame, FrameError, FrameHeader, Record, FRAME_BYTES, FRAME_END_MARK, FRAME_HEADER_BYTES,
+    FRAME_MAGIC, FRAME_PAYLOAD_BYTES, FRAME_TRAILER_BYTES, FRAME_VERSION,
 };
 pub use recover::{
     recover_and_truncate, scan, Quarantine, QuarantineKept, Recovered, TailVerdict,
     QUARANTINE_BODY_TAG,
 };
 pub use segment::{Segment, SegmentError, GROWTH_STEP_BYTES, SEGMENT_BYTES};
-pub use ship::{BufferShipper, NullShipper, ShipError, Shipper};
+pub use ship::{NullShipper, ShipError, Shipper};
 pub use wal::{BatchAck, Clock, Mode, OpenError, Wal};
 
 /// The largest record the log will carry, header and payload together — the contract's own cap on a