diff --git a/crates/busbar-kernel-wal/Cargo.toml b/crates/busbar-kernel-wal/Cargo.toml index d0a5891100..8d64850e99 100644 --- a/crates/busbar-kernel-wal/Cargo.toml +++ b/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" diff --git a/crates/busbar-kernel-wal/src/backend.rs b/crates/busbar-kernel-wal/src/backend.rs index d315aec0bf..ef19938b80 100644 --- a/crates/busbar-kernel-wal/src/backend.rs +++ b/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> = - const { std::cell::RefCell::new(Vec::new()) }; -} - /// Hands out file segments named `.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) -> io::Result { 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> { 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 { /// `.quarantine-` 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(()) +} diff --git a/crates/busbar-kernel-wal/src/journal.rs b/crates/busbar-kernel-wal/src/journal.rs index ae77b55a93..2b155f3ba0 100644 --- a/crates/busbar-kernel-wal/src/journal.rs +++ b/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>, 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, diff --git a/crates/busbar-kernel-wal/src/lib.rs b/crates/busbar-kernel-wal/src/lib.rs index 9273ead5a1..5c2debf51f 100644 --- a/crates/busbar-kernel-wal/src/lib.rs +++ b/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 diff --git a/crates/busbar-kernel-wal/src/record.rs b/crates/busbar-kernel-wal/src/record.rs index f254720e57..c7d4c62fd6 100644 --- a/crates/busbar-kernel-wal/src/record.rs +++ b/crates/busbar-kernel-wal/src/record.rs @@ -6,8 +6,18 @@ //! Every frame on disk is exactly [`FRAME_BYTES`] long. That is the whole reason a torn tail is //! recoverable: a reader never has to trust a length field it just read from a half-written region //! to find where the next frame starts, because the next frame starts at a fixed stride. The length -//! field only says how much of the frame's payload area is real, and the digest covers both the -//! header and that payload, so a frame that was partly written fails its digest and the scan stops. +//! field only says how much of the frame's payload area is real, and the digest covers the header, +//! that payload and the end mark, so a frame that was partly written fails and the scan stops. +//! +//! The frame's LAST byte is its end mark, [`FRAME_END_MARK`], which is never zero. A write runs +//! front to back through a frame, so a write that stopped anywhere inside one leaves that byte as +//! the zero the segment claimed ahead of it: a frame whose end mark is there was written to its end, +//! and one whose end mark is zero was not. That is what tells a torn write from a whole frame that +//! changed afterwards, wherever inside the frame the write stopped (see `crate::recover`). +//! +//! The first frame of every group commit carries [`FrameHeader::opens_commit`]. A verifying frame +//! that opens a commit was only ever written after every commit before it had synced, so it is the +//! on-medium proof that those earlier commits were acknowledged. //! //! A body longer than one frame's payload area is split across frames that carry the same //! `(node, node_seq)` and an increasing part index. A record is only considered present once every @@ -22,42 +32,61 @@ pub const FRAME_BYTES: usize = 512; /// How much of a frame the header takes. The last 32 bytes of it are the digest. pub const FRAME_HEADER_BYTES: usize = 96; +/// How much of a frame follows the payload: the one end-mark byte. +pub const FRAME_TRAILER_BYTES: usize = 1; + /// How many payload bytes one frame carries. -pub const FRAME_PAYLOAD_BYTES: usize = FRAME_BYTES - FRAME_HEADER_BYTES; +pub const FRAME_PAYLOAD_BYTES: usize = FRAME_BYTES - FRAME_HEADER_BYTES - FRAME_TRAILER_BYTES; + +/// The byte every frame ends with. Never zero, so a frame whose last byte is zero is one whose write +/// did not reach its end. +pub const FRAME_END_MARK: u8 = 0xE5; + +/// Where the end mark sits: the frame's last byte. +const END_MARK_OFFSET: usize = FRAME_BYTES - FRAME_TRAILER_BYTES; /// The four bytes every frame opens with. A frame that does not start with these is not a frame — /// which is exactly what the zero-filled preallocated tail of a segment looks like. pub const FRAME_MAGIC: [u8; 4] = *b"BWAL"; -/// The layout version this build WRITES. A reader that meets a version it does not know stops there -/// rather than guessing at the field offsets. +/// The layout version this build WRITES, and the only one it reads. A reader that meets a version it +/// does not know stops there rather than guessing at the field offsets. +/// +/// Version 3 is version 2 with two additions, and the layout changed, so the number did: +/// +/// - the END MARK: the frame's last byte is [`FRAME_END_MARK`], taken from the payload area (which +/// is one byte shorter for it). A torn write and a whole frame altered afterwards are told apart +/// by it wherever inside the frame the write stopped, not only inside the header; +/// - the COMMIT FLAG: the first frame of each group commit says so, which is how recovery knows a +/// damaged commit was acknowledged (a subsequent commit verifies past it) or may have been torn (no +/// subsequent commit does). /// -/// Version 2 adds the HEADER CHECK (bytes `[HEADER_CHECK_OFFSET, +4)`): the first four bytes of a +/// Version 2 added the HEADER CHECK (bytes `[HEADER_CHECK_OFFSET, +4)`): the first four bytes of a /// SHA-256 over the frame's fixed header fields `[0, HEADER_CHECKED_BYTES)` — magic, version, -/// flags, identity, part numbers and the payload length. It is what tells a TORN write from a -/// COMPLETE frame whose bytes changed afterwards: a write that stopped inside the header fails the -/// header check, while a frame whose header checks and whose digest does not was written whole and -/// then altered — which is never cut silently (see `crate::recover`). +/// flags, identity, part numbers and the payload length. Version 3 keeps it. /// -/// It is the ONLY version this build reads. Version 1 (the layout before the header check) was -/// never released — 1.5.5 had no log at all — so no segment in the field holds it, and reading it -/// would only give an attacker a downgrade: a version-2 frame relabelled 1 skips its header check -/// and has its digest failure cut as a torn tail. A version-1 frame is an unknown layout, which -/// recovery quarantines rather than cuts. -pub const FRAME_VERSION: u16 = 2; +/// Neither version 1 (before the header check) nor version 2 (before the end mark) was released — +/// 1.5.5 had no log at all — so no segment in the field holds either, and this build reads neither: +/// a frame claiming one is an unknown layout, which recovery quarantines rather than cuts. Reading +/// an older layout would only hand an attacker a downgrade (a frame relabelled to it skips a check +/// the current layout makes). +pub const FRAME_VERSION: u16 = 3; /// How many leading header bytes the header check covers: magic, version, flags, node, node_seq, /// part index, part count and payload length. const HEADER_CHECKED_BYTES: usize = 34; /// Where the header check sits. -const HEADER_CHECK_OFFSET: usize = 34; +pub(crate) const HEADER_CHECK_OFFSET: usize = 34; /// The flag bit that says another part of this record follows. const FLAG_MORE_PARTS: u8 = 1 << 0; +/// The flag bit that says this frame is the first one of a group commit. +const FLAG_OPENS_COMMIT: u8 = 1 << 1; + /// Where the digest sits inside the header, and therefore how much of the header it covers. -const DIGEST_OFFSET: usize = 64; +pub(crate) const DIGEST_OFFSET: usize = 64; /// One record as a caller hands it in: who wrote it, that writer's own sequence number, and the /// bytes of the entry. @@ -101,8 +130,15 @@ impl Record { } /// Frame this record, in order. Every frame is fully written, including the zero padding after - /// a short payload, so the bytes on disk are a function of the record alone. + /// a short payload and the end mark, so the bytes on disk are a function of the record alone. + /// No frame of it opens a commit; [`Record::encode_in_commit`] is the framing a segment writes. pub fn encode(&self) -> Vec<[u8; FRAME_BYTES]> { + self.encode_in_commit(false) + } + + /// Frame this record as a group commit puts it down: when `opens_commit`, its first frame is + /// flagged as the first frame of the commit. + pub fn encode_in_commit(&self, opens_commit: bool) -> Vec<[u8; FRAME_BYTES]> { let parts = self.frame_count(); let mut frames = Vec::with_capacity(parts); for index in 0..parts { @@ -118,6 +154,7 @@ impl Record { self.node_seq, index as u32, parts as u32, + opens_commit && index == 0, payload, )); } @@ -147,6 +184,8 @@ pub struct FrameHeader { pub payload_len: u16, /// Whether another part follows. pub more_parts: bool, + /// Whether this frame is the first one of a group commit. + pub opens_commit: bool, } /// Why a frame could not be read back. @@ -175,16 +214,19 @@ pub enum FrameError { /// The digest over the frame's own bytes is not the digest stored in it. Reported only once the /// frame's header check has passed, so the frame was written whole and then altered. DigestMismatch, - /// A version-2 frame's header check does not match its header. When nothing was written past - /// the check, the write stopped inside the header, which is what a torn write looks like; a - /// frame written past it was written whole and then altered (see [`written_whole`]). + /// The frame's header check does not match its header. A frame whose end mark is there was + /// written whole and then altered; one whose end mark is zero is a write that stopped (see + /// [`written_whole`]). HeaderMismatch, + /// The frame does not end with [`FRAME_END_MARK`]. Zero there is a write that did not reach the + /// frame's end; anything else is a last byte that changed after it was written. + EndMarkMissing, } impl FrameError { - /// Whether this is a WHOLE frame whose bytes changed after they were written: a version-2 - /// frame whose header check passed and whose digest did not. Such a frame is never a torn - /// write, wherever it sits. + /// Whether this is a WHOLE frame whose bytes changed after they were written: a frame of this + /// layout whose header check and end mark passed and whose digest did not. Such a frame is never + /// a torn write, wherever it sits. #[must_use] pub fn is_altered_whole_frame(self, version: u16) -> bool { version == FRAME_VERSION && self == FrameError::DigestMismatch @@ -214,8 +256,9 @@ impl std::fmt::Display for FrameError { f.write_str("the frame's own bytes do not hash to the digest stored in it") } FrameError::HeaderMismatch => { - f.write_str("the frame's header does not match its header check (a torn write)") + f.write_str("the frame's header does not match its header check") } + FrameError::EndMarkMissing => f.write_str("the frame does not end with its end mark"), } } } @@ -229,17 +272,21 @@ fn encode_frame( node_seq: u64, part_index: u32, part_count: u32, + opens_commit: bool, payload: &[u8], ) -> [u8; FRAME_BYTES] { debug_assert!(payload.len() <= FRAME_PAYLOAD_BYTES); let mut frame = [0u8; FRAME_BYTES]; frame[0..4].copy_from_slice(&FRAME_MAGIC); frame[4..6].copy_from_slice(&FRAME_VERSION.to_le_bytes()); - frame[6] = if part_index + 1 < part_count { - FLAG_MORE_PARTS - } else { - 0 - }; + let mut flags = 0; + if part_index + 1 < part_count { + flags |= FLAG_MORE_PARTS; + } + if opens_commit { + flags |= FLAG_OPENS_COMMIT; + } + frame[6] = flags; frame[8..16].copy_from_slice(&node.to_le_bytes()); frame[16..24].copy_from_slice(&node_seq.to_le_bytes()); frame[24..28].copy_from_slice(&part_index.to_le_bytes()); @@ -248,23 +295,14 @@ fn encode_frame( let check = header_check(&frame); frame[HEADER_CHECK_OFFSET..HEADER_CHECK_OFFSET + 4].copy_from_slice(&check); frame[FRAME_HEADER_BYTES..FRAME_HEADER_BYTES + payload.len()].copy_from_slice(payload); + frame[END_MARK_OFFSET] = FRAME_END_MARK; let digest = frame_digest(&frame); frame[DIGEST_OFFSET..DIGEST_OFFSET + 32].copy_from_slice(&digest); frame } -/// TEST ONLY: re-seal a frame whose header fields a test edited — its header check and its digest -/// recomputed over the bytes as they now stand — so the field bounds behind both checks are reached. -#[cfg(test)] -pub(crate) fn reseal(frame: &mut [u8; FRAME_BYTES]) { - let check = header_check(frame); - frame[HEADER_CHECK_OFFSET..HEADER_CHECK_OFFSET + 4].copy_from_slice(&check); - let digest = frame_digest(frame); - frame[DIGEST_OFFSET..DIGEST_OFFSET + 32].copy_from_slice(&digest); -} - -/// The header check a version-2 frame carries over its fixed header fields. -fn header_check(frame: &[u8; FRAME_BYTES]) -> [u8; 4] { +/// The header check a frame carries over its fixed header fields. +pub(crate) fn header_check(frame: &[u8; FRAME_BYTES]) -> [u8; 4] { let mut hasher = Context::new(&SHA256); hasher.update(&frame[0..HEADER_CHECKED_BYTES]); let digest = digest32(hasher); @@ -293,29 +331,57 @@ pub fn checked_header(frame: &[u8; FRAME_BYTES]) -> Option<(FrameHeader, &[u8])> Some((header, payload)) } -/// Whether `frame` was WRITTEN WHOLE: it holds a byte past its header check. +/// Whether `frame` was WRITTEN WHOLE: the write that made it reached its last byte. /// -/// A write that stops inside a frame's header leaves the rest of the frame as the zeros the segment -/// claimed ahead of it, so it has nothing past the check; a frame that does was written in full -/// (its digest alone is 32 bytes that are never all zero). So a frame whose header does not check -/// — its magic gone, a header field changed — but which answers yes here was acknowledged and then -/// altered, and is never a torn tail. +/// The last byte of every frame is the end mark, which is never zero, and the space a write lands +/// in is the zeros the segment claimed ahead of it. A write runs front to back, so one that stopped +/// anywhere inside the frame — in the header, in the digest, in the payload — leaves that byte zero. +/// A frame whose last byte is not zero was written in full, so if it fails any check it was +/// acknowledged and then altered, and is never a torn tail. #[must_use] pub fn written_whole(frame: &[u8; FRAME_BYTES]) -> bool { - frame[HEADER_CHECK_OFFSET + 4..].iter().any(|&b| b != 0) + frame[END_MARK_OFFSET] != 0 +} + +/// Whether `frame` is what a write of a frame of THIS layout leaves when it stops before the frame's +/// end: all zeros (nothing landed), or the frame's leading bytes followed by zeros to its end. +/// +/// The leading bytes it can check are checked: the magic and the version, which every frame of this +/// layout opens with, and the header check once the write got past it. Anything else is not a torn +/// write of this layout — a whole frame that changed, or a frame of a layout this build does not +/// read, whose bytes another build can still read and which is therefore never cut. +#[must_use] +pub fn unfinished_write(frame: &[u8; FRAME_BYTES]) -> bool { + if written_whole(frame) { + return false; + } + let Some(last) = frame.iter().rposition(|&b| b != 0) else { + return true; + }; + let reached = last + 1; + let mut opening = [0u8; 6]; + opening[0..4].copy_from_slice(&FRAME_MAGIC); + opening[4..6].copy_from_slice(&FRAME_VERSION.to_le_bytes()); + let n = usize::min(reached, opening.len()); + if frame[..n] != opening[..n] { + return false; + } + reached < HEADER_CHECK_OFFSET + 4 + || header_check(frame)[..] == frame[HEADER_CHECK_OFFSET..HEADER_CHECK_OFFSET + 4] } -/// The header of a frame that was [`written_whole`] but whose header does not check, read as the -/// bytes now stand; `None` for anything else. Whatever layout version it now claims: the damaged -/// byte may be the version itself (a frame relabelled to a layout this build does not read is still -/// an acknowledged record, and its identity is still taken). +/// The header of a frame that is not an [`unfinished_write`] but whose header does not check, read +/// as the bytes now stand; `None` for anything else. Whatever layout version it now claims: the +/// damaged byte may be the version itself (a frame relabelled to a layout this build does not read +/// is still an acknowledged record, and its identity is still taken), and a frame of a layout this +/// build does not read keeps its identity at the same offsets every layout has put it. /// /// The identity and payload are UNVERIFIED — the damaged byte may sit in them — so they are good /// for one thing only: naming, conservatively, what a quarantine set aside. The payload length is /// clamped to the payload area, because the damaged byte may be the length itself. #[must_use] pub fn unchecked_whole(frame: &[u8; FRAME_BYTES]) -> Option<(FrameHeader, &[u8])> { - if !written_whole(frame) || checked_header(frame).is_some() { + if unfinished_write(frame) || checked_header(frame).is_some() { return None; } let payload_len = u16::from_le_bytes([frame[32], frame[33]]); @@ -326,6 +392,7 @@ pub fn unchecked_whole(frame: &[u8; FRAME_BYTES]) -> Option<(FrameHeader, &[u8]) part_count: u32::from_le_bytes([frame[28], frame[29], frame[30], frame[31]]), payload_len, more_parts: frame[6] & FLAG_MORE_PARTS != 0, + opens_commit: frame[6] & FLAG_OPENS_COMMIT != 0, }; let end = FRAME_HEADER_BYTES + usize::min(usize::from(payload_len), FRAME_PAYLOAD_BYTES); Some((header, &frame[FRAME_HEADER_BYTES..end])) @@ -353,14 +420,16 @@ fn read_header(frame: &[u8; FRAME_BYTES]) -> Result<(FrameHeader, u16), FrameErr part_count, payload_len, more_parts: frame[6] & FLAG_MORE_PARTS != 0, + opens_commit: frame[6] & FLAG_OPENS_COMMIT != 0, }, payload_len, )) } -/// The digest a frame carries: over the header up to the digest field, then over the payload area. -/// The digest field itself is skipped, which is what lets it be filled in afterwards. -fn frame_digest(frame: &[u8; FRAME_BYTES]) -> [u8; 32] { +/// The digest a frame carries: over the header up to the digest field, then over the payload area +/// and the end mark. The digest field itself is skipped, which is what lets it be filled in +/// afterwards. +pub(crate) fn frame_digest(frame: &[u8; FRAME_BYTES]) -> [u8; 32] { let mut hasher = Context::new(&SHA256); hasher.update(&frame[0..DIGEST_OFFSET]); hasher.update(&frame[FRAME_HEADER_BYTES..]); @@ -386,12 +455,16 @@ pub fn decode_frame(frame: &[u8; FRAME_BYTES]) -> Result<(FrameHeader, &[u8]), F if version != FRAME_VERSION { return Err(FrameError::UnknownVersion { found: version }); } - // THE HEADER CHECK FIRST: a write that stopped inside the header is a torn write, and is told - // apart from a whole frame that was altered (which fails only its digest). + // THE HEADER CHECK FIRST: a header whose fields do not agree with their check is not read. if header_check(frame)[..] != frame[HEADER_CHECK_OFFSET..HEADER_CHECK_OFFSET + 4] { return Err(FrameError::HeaderMismatch); } let (header, payload_len) = read_header(frame)?; + // THE END MARK BEFORE THE DIGEST: a write that stopped inside the frame is told apart from a + // whole frame whose bytes changed (which keeps its end mark and fails only its digest). + if frame[END_MARK_OFFSET] != FRAME_END_MARK { + return Err(FrameError::EndMarkMissing); + } // The digest field is not itself hashed (see `frame_digest`), so the frame can be digested as // it stands rather than having to be copied with the field blanked first. if frame_digest(frame)[..] != frame[DIGEST_OFFSET..DIGEST_OFFSET + 32] { diff --git a/crates/busbar-kernel-wal/src/recover.rs b/crates/busbar-kernel-wal/src/recover.rs index dad0406d9c..436be5ef61 100644 --- a/crates/busbar-kernel-wal/src/recover.rs +++ b/crates/busbar-kernel-wal/src/recover.rs @@ -25,32 +25,43 @@ //! //! ## Two ways a scan stops on damage, and only one of them may be cut //! -//! A crash in the middle of a group commit leaves a TORN tail: the last batch's bytes are partly on -//! the medium and nothing after them was ever written. Cutting that tail loses nothing that was -//! acknowledged — the sync never returned — so it is cut silently, exactly as it always was. +//! A crash in the middle of a group commit leaves a TORN tail: the commit's bytes are partly on the +//! medium and nothing after them was ever written. Cutting that tail loses nothing that was +//! acknowledged — the sync never returned — so it is cut silently, exactly as it always was. A +//! record that was acknowledged and then damaged is something else, and is never cut silently +//! wherever it sits, the tail included: a silent cut would drop a settlement the node had already +//! made, and let recovery settle the unit again at its checkpoint. So it is CORRUPT. Telling the two +//! apart is a question about what was ACKNOWLEDGED, and the medium answers it two ways. //! -//! A WHOLE frame that was altered after it was written is something else again, and is never cut -//! silently wherever it sits, the tail included. A version-2 frame carries a header check over its -//! fixed header fields: a write that stopped inside the header fails that check and leaves nothing -//! past it but zeros (torn), while a frame whose header checks and whose digest does not — or whose -//! header does not check but which holds bytes past the check, its digest at least — was written in -//! full and changed afterwards. "Torn" is a statement about the WRITE: a frame that is short, or -//! zeros past the point the write stopped. A whole frame is never torn, whichever check it fails. -//! Its record was acknowledged; a silent cut would drop a settlement the node had already made, -//! and let recovery settle the unit again at its checkpoint. So it is CORRUPT, and the record's -//! bytes as they now read are handed to the caller (`Quarantine::damaged`), which is the one place -//! that knows what they named. +//! **A subsequent commit verifying past the damage.** The first frame of every group commit says so, and +//! a commit is only ever written after the one before it synced. So a verifying frame that opens a +//! commit, past the damage, proves the damaged commit's sync returned: it was acknowledged, and the +//! damage is the medium, or something outside this process, changing bytes that were already +//! durable. Corrupt, whatever the damage looks like. A verifying frame of the SAME commit proves +//! nothing — writeback puts a commit's pages down in any order, so a crash mid-commit leaves good +//! frames past missing ones as a matter of course — and is cut with the torn commit it belongs to. //! -//! A frame that fails its digest with WHOLE, VERIFYING frames after it is something else. Nothing -//! about a crash writes a good frame past a bad one: the writes are one contiguous run from the -//! durable end, and the space past them is zeros. So damage with a verifying frame beyond it is the -//! medium, or something outside this process, changing bytes that were already durable, and the -//! records behind it were acknowledged. That verdict is CORRUPT, and a corrupt segment is never cut -//! on its own say-so: the damaged remainder is first copied, byte for byte, into a quarantine the -//! factory makes durable, and only then is the segment cut back to the verified prefix. A crash -//! between the two steps leaves the copy AND the uncut segment, so it loses nothing. A frame written -//! under a layout version this build does not read is treated the same way — its bytes are whole -//! and another build can read them, so they are never simply cut. +//! **The shape of the damage, in the commit nothing acknowledges.** The last commit has no subsequent one +//! to vouch for it; its sync may or may not have returned. A crash leaves only one shape behind: +//! frames that never landed (the zeros the segment claimed ahead of the writes) and frames whose +//! write stopped part way (their leading bytes, then zeros). Every frame ends with an end mark that +//! is never zero, so a frame whose write stopped anywhere inside it — header, digest or payload — +//! has a zero last byte, and one whose last byte is there was written to its end. If every frame of +//! the damaged tail is one a write could have left unfinished, the tail is TORN. If any is a WHOLE +//! frame that fails a check — its end mark is there, so the write that made it completed — it was +//! written in full and changed afterwards: CORRUPT, and the record's bytes as they now read are +//! handed to the caller (`Quarantine::damaged`), which is the one place that knows what they named. +//! +//! What this cannot see, nothing can: bytes changed in the last commit into exactly the shape of an +//! unfinished write — zeros to the end of a frame — read as a torn tail. That is the power to cut +//! the file short, which a torn tail has always been indistinguishable from. +//! +//! A corrupt segment is never cut on its own say-so: the damaged remainder is first copied, byte for +//! byte, into a quarantine the factory makes durable, and only then is the segment cut back to the +//! verified prefix. A crash between the two steps leaves the copy AND the uncut segment, so it loses +//! nothing. A frame written under a layout version this build does not read is never one a write of +//! this layout left unfinished, so it is treated the same way — its bytes are whole and another +//! build can read them, so they are never simply cut. //! //! The verdict is returned rather than acted on further here. Raising the alarm — the log line, the //! counter, the durable record naming what was set aside — is the caller's, which is the one place @@ -61,8 +72,8 @@ use std::path::PathBuf; use crate::backend::SegmentFactory; use crate::record::{ - checked_header, decode_frame, frame_version, unchecked_whole, written_whole, FrameError, - Record, FRAME_BYTES, + checked_header, decode_frame, unchecked_whole, unfinished_write, FrameError, Record, + FRAME_BYTES, }; use crate::segment::Segment; @@ -72,12 +83,13 @@ pub enum TailVerdict { /// The writes end where the verified records end. Anything past them is space claimed ahead of /// the writes, which is zeros. Clean, - /// The scan stopped on an incomplete final record — a crash mid-append. Nothing past the damage - /// verifies, so nothing acknowledged is behind it: cut silently. + /// The scan stopped in the last group commit, on damage a crash mid-commit leaves: frames that + /// never landed and frames whose write stopped part way. Nothing acknowledged is in it: cut + /// silently, with any frame of the same commit that did land. Torn, - /// The scan stopped on damage with whole, verifying frames beyond it (or on a frame written - /// under a layout this build does not read). Records that were acknowledged are behind it. Never - /// cut without a quarantine copy first. + /// The scan stopped on damage a crash does not leave: a commit a subsequent one acknowledges, a whole + /// frame that fails a check, or a frame written under a layout this build does not read. + /// Acknowledged records are in it. Never cut without a quarantine copy first. Corrupt { /// The byte offset, inside the segment, of the frame the scan stopped on. at: u64, @@ -134,10 +146,10 @@ pub struct Quarantine { /// Every record the set-aside bytes hold whose frames' HEADERS check, assembled as the bytes now /// read: a whole altered record (header checks, digest does not) and every acknowledged record /// behind the damage. They are gone from the log, and the caller reads what they named — a - /// settlement the node made must not be settled again as if it never happened. A whole - /// version-2 frame whose header does not check contributes its record as its bytes now read - /// (unverified, so a caller that cannot decode it holds back what it cannot rule out); a torn - /// or unreadable frame contributes nothing. + /// settlement the node made must not be settled again as if it never happened. A whole frame + /// whose header does not check — of this layout or another — contributes its record as its + /// bytes now read (unverified, so a caller that cannot decode it holds back what it cannot rule + /// out); a frame a write left unfinished contributes nothing. pub damaged: Vec, /// When recovery set them aside, in milliseconds since the Unix epoch. The quarantine file's /// name carries the same number. @@ -329,18 +341,10 @@ pub fn scan(segment: &Segment) -> io::Result { let len = segment.len()?; let mut beyond = look_past(segment, durable_end, stop_at, len)?; - let unreadable_layout = matches!(stopped_because, Some(FrameError::UnknownVersion { .. })); - // A WHOLE frame altered after it was written: its header checks and its digest does not, or its - // header does not check and it holds bytes past the check. The write that made it completed, so - // this is never a torn tail. - let altered = stopped_because.is_some_and(|why| { - let mut stop_frame = [0u8; FRAME_BYTES]; - read_full(segment, stop_at, &mut stop_frame).unwrap_or(false) - && (why.is_altered_whole_frame(frame_version(&stop_frame)) - || (matches!(why, FrameError::HeaderMismatch | FrameError::NotAFrame) - && written_whole(&stop_frame))) - }); - let corrupt = beyond.verifies || unreadable_layout || altered; + // Corrupt when anything is past the verified prefix and either a subsequent commit acknowledges the + // damaged one or the damage is not a shape a crash leaves. See the module preamble. + let corrupt = + beyond.written_end > durable_end && (beyond.acknowledged || !beyond.unfinished_only); let damaged = if corrupt { let records = read_set_aside(segment, durable_end, len)?; // Every set-aside identity is taken, the altered record's first: it was acknowledged, and @@ -426,23 +430,30 @@ fn read_set_aside(segment: &Segment, from: u64, len: u64) -> io::Result, } -/// Walk `[from, len)` a frame at a time: note where the written bytes end, and whether any frame at -/// or after `stop_at` verifies on its own. Read in blocks, because on an ordinary boot this walks -/// the zeros claimed ahead of the writes and a read per frame would be a read per 512 bytes. +/// Walk `[from, len)` a frame at a time: note where the written bytes end, whether a subsequent commit +/// verifies past `stop_at`, and whether every frame is one a crash can leave. Read in blocks, +/// because on an ordinary boot this walks the zeros claimed ahead of the writes and a read per +/// frame would be a read per 512 bytes. fn look_past(segment: &Segment, from: u64, stop_at: u64, len: u64) -> io::Result { const BLOCK_FRAMES: usize = 128; let mut beyond = Beyond { - verifies: false, written_end: from, + acknowledged: false, + unfinished_only: true, identities: Vec::new(), }; let mut block = vec![0u8; BLOCK_FRAMES * FRAME_BYTES]; @@ -468,17 +479,26 @@ fn look_past(segment: &Segment, from: u64, stop_at: u64, len: u64) -> io::Result let Ok(whole) = <&[u8; FRAME_BYTES]>::try_from(chunk) else { continue; }; - if let Ok((header, _)) = decode_frame(whole) { - // Frames before the stop are the head of a record that never completed: they - // verify on their own, say nothing about what lies past the damage, and still - // carry an identity the set-aside bytes hold. - if chunk_at >= stop_at { - beyond.verifies = true; - } - let id = (header.node, header.node_seq); - if !beyond.identities.contains(&id) { - beyond.identities.push(id); + match decode_frame(whole) { + Ok((header, _)) => { + // A verifying frame AT the stop is one the scan refused for where it sits (a + // part out of order), which no crash writes. + if chunk_at == stop_at { + beyond.unfinished_only = false; + } + // Frames before the stop are the head of a record that never completed: they + // verify on their own, belong to the damaged commit, and still carry an + // identity the set-aside bytes hold. + if chunk_at > stop_at && header.opens_commit { + beyond.acknowledged = true; + } + let id = (header.node, header.node_seq); + if !beyond.identities.contains(&id) { + beyond.identities.push(id); + } } + Err(_) if unfinished_write(whole) => {} + Err(_) => beyond.unfinished_only = false, } } at += filled as u64; @@ -489,7 +509,8 @@ fn look_past(segment: &Segment, from: u64, stop_at: u64, len: u64) -> io::Result /// Scan, then cut the segment back to the last complete record — branching on WHY the scan /// stopped. /// -/// - A torn tail, or a clean end with space claimed ahead of it, is cut silently. +/// - A torn tail (the last commit, torn by a crash), or a clean end with space claimed ahead of +/// it, is cut silently. /// - A corrupt one first has its damaged remainder copied, byte for byte, into a quarantine /// `factory` makes durable, and only then is the segment cut. If the copy cannot be made, the /// segment is NOT cut: it is closed to writes instead, so the next append rolls past it and the @@ -507,7 +528,7 @@ pub fn recover_and_truncate( ) -> io::Result { let mut recovered = scan(segment)?; if recovered.was_torn() { - // A crash mid-append. Nothing past the damage was acknowledged. + // A crash mid-commit. Nothing past the verified prefix was acknowledged. segment.truncate_to(recovered.durable_end)?; return Ok(recovered); } diff --git a/crates/busbar-kernel-wal/src/segment.rs b/crates/busbar-kernel-wal/src/segment.rs index dc72dc28e6..b7abb9021b 100644 --- a/crates/busbar-kernel-wal/src/segment.rs +++ b/crates/busbar-kernel-wal/src/segment.rs @@ -14,9 +14,17 @@ //! **A failed sync poisons the segment, permanently.** Not "retries", not "degrades" — poisons. The //! moment a sync reports an error, every byte in this segment after the last good commit is of //! unknown state, and the one safe reading of unknown is that it is not there. So the segment is -//! closed to further writes, the caller is handed a durability loss, and the batch that failed goes -//! to a fresh segment along with the one after it. A segment that has been poisoned never un-poisons -//! itself, because nothing that happens later can tell you what landed. +//! closed to further writes, the failed batch's bytes are cut off it (the segment is shortened back +//! to the last good commit and that is synced), the caller is handed a durability loss, and the +//! batch that failed goes to a fresh segment along with the one after it. A segment that has been +//! poisoned never un-poisons itself, because nothing that happens afterwards can tell you what landed. +//! +//! The cut is what keeps a batch the caller was told was lost from coming back: after a write error +//! its pages can still sit in the page cache, readable and verifying, without ever reaching the +//! medium, and a restart that read them would put on the book records that are on no disk and chain +//! the next ones onto them. That the segment is finished is recorded by the log opening the next +//! segment at once (see `crate::wal`): a restart appends to the newest segment, never to one below +//! it, so a poisoned segment is never written to again by this process or the next one. use std::io; @@ -167,8 +175,10 @@ impl Segment { return Err(SegmentError::Poisoned); } let mut buf: Vec = Vec::new(); - for record in records { - for frame in record.encode() { + for (i, record) in records.iter().enumerate() { + // The commit's first frame says so: a subsequent commit verifying past a damaged one is the + // proof the damaged one was acknowledged. + for frame in record.encode_in_commit(i == 0) { buf.extend_from_slice(&frame); } } @@ -181,21 +191,36 @@ impl Segment { } if let Err(e) = self.claim_through(end) { // A claim that fails is a write that may have half happened. Treated as a loss. - self.poisoned = true; + self.poison(); return Err(SegmentError::Write(e)); } if let Err(e) = self.backend.write_all_at(self.write_offset, &buf) { - self.poisoned = true; + self.poison(); return Err(SegmentError::Write(e)); } if let Err(e) = self.backend.sync() { - self.poisoned = true; + self.poison(); return Err(SegmentError::Sync(e)); } self.write_offset = end; Ok(self.write_offset) } + /// Close the segment for good after a failed write or sync, and cut what the failed batch may + /// have left on the backing: shorten it to the last good commit and sync the new length. + /// + /// The write's own error is what the caller reports as the durability loss. A cut that fails + /// too has nothing further to try on a medium that refuses both: the failed batch's bytes may + /// then still read back until the retried batch is durable in the next segment, which is the + /// one case `crate::wal` names as beyond what any record of the loss could cover. + fn poison(&mut self) { + self.poisoned = true; + if self.backend.set_len(self.write_offset).is_ok() { + self.claimed = self.write_offset; + let _synced = self.backend.sync().is_ok(); + } + } + /// Claim space forward so that `end` is inside the backing, a step at a time. fn claim_through(&mut self, end: u64) -> io::Result<()> { if end <= self.claimed { diff --git a/crates/busbar-kernel-wal/src/ship.rs b/crates/busbar-kernel-wal/src/ship.rs index 30b4913806..ffbcdae6fd 100644 --- a/crates/busbar-kernel-wal/src/ship.rs +++ b/crates/busbar-kernel-wal/src/ship.rs @@ -35,41 +35,3 @@ impl Shipper for NullShipper { Ok(()) } } - -/// A shipper that keeps every record it was handed, in order. -/// -/// This is what a memory-buffered node's "store" looks like when the store is itself in memory: the -/// records are the system of record for exactly as long as the process lives, and the crate says so -/// rather than implying more. -/// -/// The records are held behind a shared handle so that whoever configured the shipper can still see -/// what reached it after handing ownership to the log. -#[derive(Debug, Default, Clone)] -pub struct BufferShipper { - records: std::sync::Arc>>, -} - -impl BufferShipper { - /// A fresh one. - pub fn new() -> Self { - BufferShipper::default() - } - - /// A snapshot of everything shipped so far, in order. - pub fn records(&self) -> Vec { - self.records - .lock() - .unwrap_or_else(|e| e.into_inner()) - .clone() - } -} - -impl Shipper for BufferShipper { - fn ship(&mut self, records: &[Record]) -> Result<(), ShipError> { - self.records - .lock() - .unwrap_or_else(|e| e.into_inner()) - .extend_from_slice(records); - Ok(()) - } -} diff --git a/crates/busbar-kernel-wal/src/tests/dir_fsync.rs b/crates/busbar-kernel-wal/src/tests/dir_fsync.rs new file mode 100644 index 0000000000..cda853f87d --- /dev/null +++ b/crates/busbar-kernel-wal/src/tests/dir_fsync.rs @@ -0,0 +1,151 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! A directory fsync that fails is an error, wherever the log makes directory durability. +//! +//! The entry a creation, a rename or an unlink changes is only durable once the directory holding +//! it is fsynced. Swallowing that fsync's failure hands the caller a success the medium did not +//! give: a publish reported durable whose rename a power loss can undo, a segment handed out whose +//! name a power loss can take with every acknowledged record in it, a quarantine copy reported kept +//! while recovery cuts the only other copy of the bytes. The one failure that is not the caller's +//! is a filesystem saying it does not support the operation at all. (The same rule for every other +//! file busbar publishes is the loader's `durable` module's, and its cases are there.) + +use std::path::{Path, PathBuf}; + +use crate::backend::{DirectoryFactory, SegmentFactory}; +use crate::record::{Record, FRAME_BYTES}; +use crate::recover::QuarantineKept; +use crate::segment::SEGMENT_BYTES; +use crate::ship::NullShipper; +use crate::tests::hooks::{fault_arm, fault_reset, parents_fsynced, FaultStep}; +use crate::wal::{Mode, Wal}; + +/// Linux and macOS share this value. Injected via `from_raw_os_error`. +const EIO: i32 = 5; + +/// A scratch path under the system temp directory, removed on drop. Not created: a test that wants +/// it to exist creates it. +struct Scratch(PathBuf); + +impl Scratch { + fn new(tag: &str) -> Self { + let path = std::env::temp_dir().join(format!( + "busbar-wal-dirfsync-{tag}-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + let _ = std::fs::remove_dir_all(&path); + Scratch(path) + } + + fn made(tag: &str) -> Self { + let scratch = Scratch::new(tag); + std::fs::create_dir_all(&scratch.0).expect("the scratch directory"); + scratch + } + + fn path(&self) -> &Path { + &self.0 + } +} + +impl Drop for Scratch { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +/// **A SEGMENT WHOSE DIRECTORY FSYNC FAILS IS NOT HANDED OUT, AND NOT LEFT BEHIND.** RED before the +/// fix: the factory swallowed the failure and handed out a segment whose name a power loss could +/// take with every record acknowledged in it. Left behind, the file would be found existing by the +/// next open, which would skip the fsync for good. +#[test] +fn a_segment_whose_directory_fsync_fails_is_not_handed_out() { + let scratch = Scratch::new("segment"); + let mut factory = DirectoryFactory::new(scratch.path()).expect("the data directory opens"); + let dir = scratch.path().to_path_buf(); + + fault_reset(); + fault_arm(FaultStep::DirSync, EIO); + let err = factory + .open(0) + .err() + .expect("a segment whose entry is not durable is not handed out"); + assert_eq!(err.raw_os_error(), Some(EIO)); + assert!( + !factory.segment_path(0).exists(), + "the file is not left for the next open to find already existing" + ); + + fault_reset(); + factory.open(0).expect("the next open creates it again"); + assert_eq!( + parents_fsynced(), + vec![dir], + "and makes its entry durable this time" + ); +} + +/// **A QUARANTINE COPY WHOSE DIRECTORY FSYNC FAILS IS NOT KEPT, SO THE SEGMENT IS NOT CUT.** RED +/// before the fix: the quarantine answered `Ok` over a copy whose name a power loss could take, and +/// recovery then cut the only other copy of the damaged bytes out of the segment. +#[test] +fn a_quarantine_whose_directory_fsync_fails_leaves_the_segment_uncut() { + let scratch = Scratch::made("quarantine"); + let factory = DirectoryFactory::new(scratch.path()).expect("the data directory opens"); + let segment = factory.segment_path(0); + let mut bytes = Vec::new(); + for (i, record) in (1..=4u64) + .map(|seq| Record::new(3, seq, vec![seq as u8; 200])) + .enumerate() + { + for frame in record.encode_in_commit(i == 0) { + bytes.extend_from_slice(&frame); + } + } + // The second record's payload is damaged; the two behind it verify. + bytes[FRAME_BYTES + 200] ^= 0xFF; + std::fs::write(&segment, &bytes).unwrap(); + + fault_reset(); + fault_arm(FaultStep::DirSync, EIO); + let wal = Wal::with_parts( + Box::new(factory), + Box::new(NullShipper::new()), + Mode::OnDisk, + SEGMENT_BYTES, + crate::tests::fixtures::wall_ms, + ) + .expect("a failed copy does not stop the log opening"); + let q = wal.quarantined(); + assert_eq!(q.len(), 1, "the damage is reported"); + assert!( + matches!(q[0].kept, QuarantineKept::InPlace(_)), + "a copy that was not made durable is not a copy kept: {:?}", + q[0].kept + ); + assert_eq!( + std::fs::read(&segment).unwrap(), + bytes, + "the segment still holds every damaged byte" + ); + assert!(wal.is_poisoned(), "and is closed to writes"); +} + +/// **A DATA DIRECTORY CREATED SEVERAL LEVELS DEEP HAS EVERY NEW DIRECTORY'S ENTRY FSYNCED.** RED +/// before the fix: the factory created the tree with `std::fs::create_dir_all` and fsynced only the +/// leaf's parent, so an intermediate entry — and the whole log under it — could be lost to the first +/// power loss. It now goes through the one durable `create_dir_all`. +#[test] +fn a_deep_data_directory_has_every_directory_it_creates_fsynced() { + let scratch = Scratch::made("deep"); + let root = scratch.path().to_path_buf(); + fault_reset(); + DirectoryFactory::new(root.join("a").join("b").join("journal")).expect("the tree is created"); + assert_eq!( + parents_fsynced(), + vec![root.clone(), root.join("a"), root.join("a").join("b")], + "one parent fsync per created directory, shallowest first" + ); +} diff --git a/crates/busbar-kernel-wal/src/tests/fixtures.rs b/crates/busbar-kernel-wal/src/tests/fixtures.rs index f70d4a539a..32b57ce594 100644 --- a/crates/busbar-kernel-wal/src/tests/fixtures.rs +++ b/crates/busbar-kernel-wal/src/tests/fixtures.rs @@ -1,7 +1,7 @@ // SPDX-License-Identifier: Apache-2.0 // Copyright (C) 2026 Busbar Inc and contributors -//! The shared props: a wall clock. +//! The shared props: a wall clock, and a re-seal for a frame a test edited. /// The wall clock a test hands a log, in unix milliseconds — the reading the composition root /// hands a production one. A test may read a clock; the crate under test may not. @@ -10,3 +10,13 @@ pub fn wall_ms() -> u64 { .duration_since(std::time::UNIX_EPOCH) .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX)) } + +/// Re-seal a frame whose header fields a test edited — its header check and its digest recomputed +/// over the bytes as they now stand — so the field bounds behind both checks are reached. +pub fn reseal(frame: &mut [u8; crate::record::FRAME_BYTES]) { + use crate::record::{frame_digest, header_check, DIGEST_OFFSET, HEADER_CHECK_OFFSET}; + let check = header_check(frame); + frame[HEADER_CHECK_OFFSET..HEADER_CHECK_OFFSET + 4].copy_from_slice(&check); + let digest = frame_digest(frame); + frame[DIGEST_OFFSET..DIGEST_OFFSET + 32].copy_from_slice(&digest); +} diff --git a/crates/busbar-kernel-wal/src/tests/hooks.rs b/crates/busbar-kernel-wal/src/tests/hooks.rs new file mode 100644 index 0000000000..503ac153e8 --- /dev/null +++ b/crates/busbar-kernel-wal/src/tests/hooks.rs @@ -0,0 +1,58 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! The test side of `backend.rs`'s `#[cfg(test)]` hook point: the one directory fsync the log makes +//! for its own segment files and quarantine copies. +//! +//! `backend.rs` keeps only the call. What it consults — the armed fault and the recorded fsyncs — +//! lives here, per thread, so a test arms exactly the case it means and no other test sees it. + +use std::cell::RefCell; +use std::io; +use std::path::{Path, PathBuf}; + +/// The step a fault can be armed at: the fsync of the directory holding an entry the log created. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub(crate) enum FaultStep { + /// The fsync of the directory holding a created segment, quarantine copy or data directory. + DirSync, +} + +thread_local! { + /// `(step, raw_os_errno)` — the next matching step on this thread returns that errno once. + static FAULT_INJECT: RefCell> = const { RefCell::new(None) }; + /// Every directory the log fsynced on this thread, in order. + static PARENT_FSYNCS: RefCell> = const { RefCell::new(Vec::new()) }; +} + +/// Arm a one-shot fault: the next matching step returns `io::Error::from_raw_os_error(errno)`. +pub(crate) fn fault_arm(step: FaultStep, errno: i32) { + FAULT_INJECT.with(|c| *c.borrow_mut() = Some((step, errno))); +} + +/// Clear any armed fault and every recorded directory fsync (call at the start of each case). +pub(crate) fn fault_reset() { + FAULT_INJECT.with(|c| *c.borrow_mut() = None); + PARENT_FSYNCS.with(|c| c.borrow_mut().clear()); +} + +/// The directory-fsync hook: record `parent` as fsynced on this thread, then fail if a +/// [`FaultStep::DirSync`] fault is armed. +pub(crate) fn parent_fsync(parent: &Path) -> io::Result<()> { + PARENT_FSYNCS.with(|c| c.borrow_mut().push(parent.to_path_buf())); + FAULT_INJECT.with(|c| { + let mut slot = c.borrow_mut(); + match *slot { + Some((FaultStep::DirSync, errno)) => { + *slot = None; + Err(io::Error::from_raw_os_error(errno)) + } + None => Ok(()), + } + }) +} + +/// Every directory fsynced, in order, for the ancestor-walk and segment-creation assertions. +pub(crate) fn parents_fsynced() -> Vec { + PARENT_FSYNCS.with(|c| c.borrow().clone()) +} diff --git a/crates/busbar-kernel-wal/src/tests/journal_chain.rs b/crates/busbar-kernel-wal/src/tests/journal_chain.rs index e51709e5c2..8a935fd11f 100644 --- a/crates/busbar-kernel-wal/src/tests/journal_chain.rs +++ b/crates/busbar-kernel-wal/src/tests/journal_chain.rs @@ -34,7 +34,12 @@ fn every_class_is_pinned_to_its_byte() { fn the_bound_is_pinned() { assert_eq!(MEMORY_BUFFER_RECORDS, 8192); assert_eq!( - Journal::memory_buffered(1, crate::tests::fixtures::wall_ms).capacity(), + Journal::memory_buffered_to( + 1, + Box::new(crate::ship::NullShipper::new()), + crate::tests::fixtures::wall_ms + ) + .capacity(), MEMORY_BUFFER_RECORDS ); } @@ -75,6 +80,25 @@ fn a_body_reads_back_what_was_written() { assert_eq!(lying.text(), None); } +/// A shipper that acknowledges everything and counts what it was offered, readable after the log +/// has taken ownership of it. +#[derive(Clone, Default)] +struct CountingShipper(std::sync::Arc); + +impl CountingShipper { + fn shipped(&self) -> u64 { + self.0.load(std::sync::atomic::Ordering::SeqCst) + } +} + +impl crate::ship::Shipper for CountingShipper { + fn ship(&mut self, records: &[crate::Record]) -> Result<(), crate::ship::ShipError> { + self.0 + .fetch_add(records.len() as u64, std::sync::atomic::Ordering::SeqCst); + Ok(()) + } +} + /// A run of `n` chained journal records for `node`, numbered from one, as a store hands them back. fn chained_run(node: u64, n: u64) -> Vec { let mut head = [0u8; 32]; @@ -107,7 +131,7 @@ fn chained_run(node: u64, n: u64) -> Vec { #[test] fn a_resumed_memory_journal_reads_back_and_continues_the_stored_chain() { let run = chained_run(5, 40); - let shipper = crate::BufferShipper::new(); + let shipper = CountingShipper::default(); let journal = Journal::memory_resumed( 5, &run, @@ -127,7 +151,7 @@ fn a_resumed_memory_journal_reads_back_and_continues_the_stored_chain() { "the chain continues from the newest stored record" ); assert!( - shipper.records().is_empty(), + shipper.shipped() == 0, "a seeded record is never shipped back to the store it came from" ); assert_eq!(journal.mode(), crate::Mode::MemoryBuffered); @@ -137,9 +161,13 @@ fn a_resumed_memory_journal_reads_back_and_continues_the_stored_chain() { /// that asked for it refuses new work instead. #[test] fn a_retaining_journal_reports_its_bound_and_drops_nothing() { - let journal = Journal::memory_buffered(1, crate::tests::fixtures::wall_ms) - .with_capacity(1) - .retaining_at_bound(); + let journal = Journal::memory_buffered_to( + 1, + Box::new(crate::ship::NullShipper::new()), + crate::tests::fixtures::wall_ms, + ) + .with_capacity(1) + .retaining_at_bound(); assert!(!journal.at_bound(), "an empty buffer is not at its bound"); assert_eq!(journal.dropped_total(), 0); assert!(journal.overflows().is_empty()); diff --git a/crates/busbar-kernel-wal/src/tests/mod.rs b/crates/busbar-kernel-wal/src/tests/mod.rs index b2145df601..acc1ad460e 100644 --- a/crates/busbar-kernel-wal/src/tests/mod.rs +++ b/crates/busbar-kernel-wal/src/tests/mod.rs @@ -3,7 +3,9 @@ //! The batteries. +mod dir_fsync; mod fixtures; +pub(crate) mod hooks; mod journal_chain; mod no_disk; mod record_layout; diff --git a/crates/busbar-kernel-wal/src/tests/no_disk.rs b/crates/busbar-kernel-wal/src/tests/no_disk.rs index 6929b244c8..121b84f42f 100644 --- a/crates/busbar-kernel-wal/src/tests/no_disk.rs +++ b/crates/busbar-kernel-wal/src/tests/no_disk.rs @@ -18,9 +18,15 @@ use crate::wal::{Mode, Wal}; #[test] fn the_default_log_is_the_memory_buffered_one() { // Stated as a test because "the default is no disk" is a product claim, and a default that - // quietly changed would otherwise be found by an operator rather than by the suite. + // quietly changed would otherwise be found by an operator rather than by the suite. The + // default is the log a node with no data directory is built with: memory-buffered, shipping to + // whatever store the composition root binds. assert_eq!( - Wal::memory_buffered(crate::tests::fixtures::wall_ms).mode(), + Wal::memory_buffered_to( + Box::new(crate::ship::NullShipper::new()), + crate::tests::fixtures::wall_ms + ) + .mode(), Mode::MemoryBuffered ); } diff --git a/crates/busbar-kernel-wal/src/tests/record_layout.rs b/crates/busbar-kernel-wal/src/tests/record_layout.rs index 5b4e79ab8d..a7cf1eb13b 100644 --- a/crates/busbar-kernel-wal/src/tests/record_layout.rs +++ b/crates/busbar-kernel-wal/src/tests/record_layout.rs @@ -4,8 +4,8 @@ //! The frame itself: fixed stride, continuation, and a digest that catches an edit. use crate::record::{ - decode_frame, FrameError, Record, FRAME_BYTES, FRAME_HEADER_BYTES, FRAME_MAGIC, - FRAME_PAYLOAD_BYTES, + decode_frame, unfinished_write, FrameError, Record, FRAME_BYTES, FRAME_END_MARK, + FRAME_HEADER_BYTES, FRAME_MAGIC, FRAME_PAYLOAD_BYTES, FRAME_TRAILER_BYTES, FRAME_VERSION, }; #[test] @@ -84,11 +84,11 @@ fn zeros_are_read_as_the_end_of_the_writes_and_not_as_damage() { #[test] fn a_frame_whose_payload_length_is_impossible_is_refused_before_it_is_used() { // Re-sealed after the edit, so the header check passes and the bound behind it is what refuses - // (an unsealed edit fails its header check first: `a_version_2_header_edit_fails_its_header_check`). + // (an unsealed edit fails its header check first: `a_header_edit_fails_its_header_check`). let record = Record::new(1, 1, vec![1u8; 10]); let mut frame = record.encode().remove(0); frame[32..34].copy_from_slice(&((FRAME_PAYLOAD_BYTES + 1) as u16).to_le_bytes()); - crate::record::reseal(&mut frame); + crate::tests::fixtures::reseal(&mut frame); assert!(matches!( decode_frame(&frame), Err(FrameError::PayloadTooLong { .. }) @@ -107,15 +107,15 @@ fn a_frame_from_a_layout_this_build_does_not_know_stops_the_scan() { } /// Version 1 — the layout before the header check — was never released, so this build does not read -/// it: a frame claiming it is an unknown layout, sealed digest or not. Reading it would let a -/// version-2 frame relabelled 1 skip its header check. +/// it: a frame claiming it is an unknown layout, sealed digest or not. Reading it would let a frame +/// relabelled 1 skip its header check. #[test] fn a_version_1_frame_is_a_layout_this_build_does_not_read() { let record = Record::new(1, 1, vec![1u8; 10]); let mut frame = record.encode().remove(0); frame[4..6].copy_from_slice(&1u16.to_le_bytes()); // Re-sealed under the edit, so nothing but the version can refuse it. - crate::record::reseal(&mut frame); + crate::tests::fixtures::reseal(&mut frame); assert_eq!( decode_frame(&frame), Err(FrameError::UnknownVersion { found: 1 }) @@ -129,7 +129,7 @@ fn a_frame_claiming_a_part_outside_its_own_count_is_refused() { let record = Record::new(1, 1, vec![1u8; 10]); let mut frame = record.encode().remove(0); frame[24..28].copy_from_slice(&5u32.to_le_bytes()); - crate::record::reseal(&mut frame); + crate::tests::fixtures::reseal(&mut frame); assert!(matches!( decode_frame(&frame), Err(FrameError::BadParts { .. }) @@ -138,26 +138,117 @@ fn a_frame_claiming_a_part_outside_its_own_count_is_refused() { #[test] fn the_header_leaves_the_documented_amount_of_room_for_a_payload() { - // The two constants are load-bearing for the fixed stride, so they are pinned rather than + // The constants are load-bearing for the fixed stride, so they are pinned rather than // recomputed from each other at every call site. - assert_eq!(FRAME_HEADER_BYTES + FRAME_PAYLOAD_BYTES, FRAME_BYTES); + assert_eq!( + FRAME_HEADER_BYTES + FRAME_PAYLOAD_BYTES + FRAME_TRAILER_BYTES, + FRAME_BYTES + ); + assert_eq!(FRAME_HEADER_BYTES, 96); + assert_eq!(FRAME_PAYLOAD_BYTES, 415); + assert_eq!(FRAME_TRAILER_BYTES, 1); + assert_eq!(FRAME_VERSION, 3); assert_eq!(FRAME_BYTES, crate::MAX_RECORD_BYTES); } -/// A version-2 frame whose header field is edited fails its HEADER CHECK before anything else is -/// read: an edit to the header reads as a torn write, never as a different record. +/// Every frame ENDS with its end mark, and the end mark is never zero: it is the byte that says the +/// write reached the frame's end. +#[test] +fn every_frame_ends_with_its_end_mark() { + assert_ne!(FRAME_END_MARK, 0); + for body_len in [ + 0usize, + 1, + FRAME_PAYLOAD_BYTES, + FRAME_PAYLOAD_BYTES + 1, + 1000, + ] { + let body: Vec = vec![0u8; body_len]; + for frame in Record::new(1, 1, body).encode() { + assert_eq!(frame[FRAME_BYTES - 1], FRAME_END_MARK, "body of {body_len}"); + } + } +} + +/// **A WRITE THAT STOPS ANYWHERE INSIDE A FRAME LEAVES AN UNFINISHED WRITE; A WHOLE FRAME THAT +/// CHANGED NEVER IS ONE.** The two are what recovery tells a torn tail from an altered record by, so +/// both directions are walked at every byte: a frame cut at each offset (the rest the zeros the +/// segment claimed ahead) is unfinished and does not decode, and the same frame with any one byte +/// changed is not unfinished, wherever that byte is. +#[test] +fn a_write_stopped_at_any_byte_is_unfinished_and_a_changed_whole_frame_is_not() { + // A payload ending in zeros: the case where the bytes alone could not have told a tear in the + // payload from a whole frame. + let mut body = vec![7u8; 300]; + body.extend_from_slice(&[0u8; 50]); + let frame = Record::new(9, 4, body).encode_in_commit(true).remove(0); + for stop in 0..FRAME_BYTES { + let mut torn = [0u8; FRAME_BYTES]; + torn[..stop].copy_from_slice(&frame[..stop]); + assert!( + unfinished_write(&torn), + "a write that stopped at byte {stop} is an unfinished write" + ); + assert!( + decode_frame(&torn).is_err(), + "and is not read, at byte {stop}" + ); + } + assert!(decode_frame(&frame).is_ok()); + assert!(!unfinished_write(&frame), "a whole frame is not unfinished"); + for at in 0..FRAME_BYTES { + let mut changed = frame; + changed[at] ^= 0xFF; + assert!( + !unfinished_write(&changed), + "a whole frame with byte {at} changed reads as an unfinished write" + ); + } +} + +/// A frame of a layout this build does not read is never an unfinished write of this one, even +/// with a zero last byte: its bytes are another build's to read, and recovery never cuts them. +#[test] +fn a_frame_of_another_layout_is_never_an_unfinished_write() { + let mut frame = Record::new(1, 1, vec![1u8; 10]).encode().remove(0); + frame[FRAME_BYTES - 1] = 0; + for version in [1u16, 2] { + frame[4..6].copy_from_slice(&version.to_le_bytes()); + assert!(!unfinished_write(&frame), "version {version}"); + } +} + +/// The first frame of a group commit says so, and no other frame does: it is the frame whose +/// verifying presence past a damaged commit proves that commit was acknowledged. +#[test] +fn only_the_first_frame_of_a_commit_opens_it() { + let record = Record::new(2, 5, vec![3u8; 1000]); + let opening: Vec = record + .encode_in_commit(true) + .iter() + .map(|f| decode_frame(f).unwrap().0.opens_commit) + .collect(); + assert_eq!(opening, vec![true, false, false]); + assert!(record + .encode() + .iter() + .all(|f| !decode_frame(f).unwrap().0.opens_commit)); +} + +/// A frame whose header field is edited fails its HEADER CHECK before anything else is read: an edit +/// to the header never reads as a different record. #[test] -fn a_version_2_header_edit_fails_its_header_check() { +fn a_header_edit_fails_its_header_check() { let record = Record::new(1, 1, vec![1u8; 10]); let mut frame = record.encode().remove(0); frame[24..28].copy_from_slice(&5u32.to_le_bytes()); assert_eq!(decode_frame(&frame), Err(FrameError::HeaderMismatch)); } -/// A version-2 frame whose PAYLOAD is edited keeps its header check and fails its digest: a whole -/// frame altered after it was written. +/// A frame whose PAYLOAD is edited keeps its header check and its end mark and fails its digest: a +/// whole frame altered after it was written. #[test] -fn a_version_2_payload_edit_is_an_altered_whole_frame() { +fn a_payload_edit_is_an_altered_whole_frame() { let record = Record::new(1, 1, vec![1u8; 10]); let mut frame = record.encode().remove(0); frame[crate::record::FRAME_HEADER_BYTES] ^= 0xFF; @@ -167,10 +258,10 @@ fn a_version_2_payload_edit_is_an_altered_whole_frame() { } /// THE ON-DISK DIGESTS ARE SHA-256, BYTE FOR BYTE. The frame digest, the header check and the -/// journal's body digest moved from RustCrypto `sha2` to ring (ONE crypto backend = ring). Every -/// segment already on disk was written with the old one, so a frame the new code writes must carry -/// exactly the bytes RustCrypto computes over the same regions: the digest (header bytes 64..96) -/// over the header up to it plus the payload area, and the 4-byte check (34..38) over bytes 0..34. +/// journal's body digest are computed with ring (ONE crypto backend = ring), and a frame must carry +/// exactly the bytes an independent SHA-256 (RustCrypto `sha2`, a dev-dependency only) computes over +/// the same regions: the digest (header bytes 64..96) over the header up to it plus the payload area +/// and the end mark, and the 4-byte check (34..38) over bytes 0..34. #[test] fn the_frame_digest_and_header_check_are_sha256_byte_for_byte() { use sha2::Digest as _; diff --git a/crates/busbar-kernel-wal/src/tests/segment_dir_sync.rs b/crates/busbar-kernel-wal/src/tests/segment_dir_sync.rs index 4c38f2deb1..586211bd55 100644 --- a/crates/busbar-kernel-wal/src/tests/segment_dir_sync.rs +++ b/crates/busbar-kernel-wal/src/tests/segment_dir_sync.rs @@ -5,14 +5,15 @@ use std::path::PathBuf; -use crate::backend::{DirectoryFactory, SegmentFactory, DIR_SYNCS}; +use crate::backend::{DirectoryFactory, SegmentFactory}; +use crate::tests::hooks::{fault_reset, parents_fsynced}; fn syncs() -> Vec { - DIR_SYNCS.with(|c| c.borrow().clone()) + parents_fsynced() } fn reset() { - DIR_SYNCS.with(|c| c.borrow_mut().clear()); + fault_reset(); } /// A scratch data directory, removed on drop. diff --git a/crates/busbar-kernel-wal/src/wal.rs b/crates/busbar-kernel-wal/src/wal.rs index 64173b585d..c41370ba5d 100644 --- a/crates/busbar-kernel-wal/src/wal.rs +++ b/crates/busbar-kernel-wal/src/wal.rs @@ -41,11 +41,29 @@ //! //! ## What happens when a sync fails //! -//! The segment is poisoned and the caller is handed a `DurabilityLost`. The failed batch is retained -//! whole. The next append rolls to a fresh segment and writes the retained batch first, then the new -//! one — batches *n* and *n+1*, in order, on a segment that has not lost anything. If the fresh -//! segment fails too, the node has a disk that cannot be written to, and the log says so on every -//! subsequent call rather than accumulating silently. +//! The segment is poisoned, the failed batch's bytes are cut off it, the next segment is opened, and +//! the caller is handed a `DurabilityLost`. The failed batch is retained whole. The next append moves +//! to the fresh segment and writes the retained batch first, then the new one — batches *n* and +//! *n+1*, in order, on a segment that has not lost anything. If the fresh segment cannot be opened, or fails too, the node has a +//! disk that cannot be written to: the batch is retained with the one that could not follow it, and +//! the log says so on every subsequent call rather than accumulating silently or dropping either. +//! +//! ## A segment that lost a sync is never written to again, across a restart too +//! +//! Poison is a flag in the process that saw the sync fail, and the flag dies with it. So the log +//! does not leave the poisoned segment on the next append, it opens the next segment AT ONCE, before +//! the loss is reported — and the next segment's directory entry, made durable when it is created, +//! is the durable record that the poisoned one is finished. A restart appends to the newest segment +//! and never to one below it, even when the newest holds nothing yet and the tail of the log is read +//! from the one before; so the segment that lost a sync is not resumed into. The failed batch's +//! bytes were cut off it before the loss was reported, so the restart does not read them back +//! either. +//! +//! What that cannot cover is a medium that refuses everything at once: if the cut fails as well as +//! the sync, the failed batch's bytes may still be readable at a restart until the retried batch is +//! durable in the next segment; and if the next segment cannot be created either, a restart resumes +//! in the segment that lost the sync. A disk that will make nothing durable cannot be made to +//! record that it lost something; every call on it reports the loss. use std::collections::{HashMap, HashSet, VecDeque}; use std::io; @@ -159,6 +177,10 @@ pub struct Wal { segments_used: u64, /// Every corrupt remainder recovery set aside, at open and at any roll since, oldest first. quarantined: Vec, + /// The segment after a poisoned one, opened the moment the poison happened so that its + /// directory entry records, durably, that the poisoned one is finished. The next append moves + /// to it. + next_segment: Option, } impl std::fmt::Debug for Wal { @@ -174,24 +196,12 @@ impl std::fmt::Debug for Wal { } impl Wal { - /// The default log: memory-buffered, shipping nowhere, touching no disk. + /// A memory-buffered log shipping to `shipper`: the default, and the shape a deployment that + /// names no data directory runs. The buffer stages, and the shipper decides what, if anything, + /// is kept (the shipped store adapter keeps a count and the last identity only). /// /// A node built this way cannot create a file even by mistake, because the only thing that /// knows how to open one is the directory factory and this log does not hold one. - pub fn memory_buffered(clock: Clock) -> Self { - Wal::with_parts( - Box::new(MemoryFactory::new()), - Box::new(NullShipper::new()), - Mode::MemoryBuffered, - SEGMENT_BYTES, - clock, - ) - .expect("a memory segment cannot fail to open") - } - - /// A memory-buffered log shipping to `shipper`. This is the shape a deployment that names a - /// store but no data directory runs: the buffer stages, and the shipper decides what, if - /// anything, is kept (the shipped store adapter keeps a count and the last identity only). pub fn memory_buffered_to(shipper: Box>, clock: Clock) -> Self { Wal::with_parts( Box::new(MemoryFactory::new()), @@ -258,7 +268,7 @@ impl Wal { /// /// Constructing this IS the decision to write to a disk. Nothing here probes for a directory or /// guesses at one: a caller that has no data directory configured calls - /// [`Wal::memory_buffered`] and never reaches this function. + /// [`Wal::memory_buffered_to`] and never reaches this function. pub fn in_directory( dir: impl AsRef, shipper: Box>, @@ -305,6 +315,7 @@ impl Wal { recovered: Vec::new(), segments_used, quarantined: Vec::new(), + next_segment: None, }; for record in &recovered.records { wal.mark_written(record.node, record.node_seq); @@ -482,9 +493,14 @@ impl Wal { ) -> Result { let replaying = !self.lost_batch.is_empty(); // A poisoned segment is left behind before anything else happens. If a fresh one cannot be - // opened either, the node has a disk it cannot write to and the caller is told so. - if self.segment.is_poisoned() && self.roll().is_err() { - return Err(DurabilityLost::observed(token, at)); + // opened, the node has a disk it cannot write to and the caller is told so; the records are + // retained exactly as on every other arm, because the journal has already chained them and + // a record that is on no medium and in no retry queue is a hole in the chain nothing names. + if self.segment.is_poisoned() { + if let Err(_e) = self.roll() { + self.lost_batch.extend(records.iter().cloned()); + return Err(DurabilityLost::observed(token, at)); + } } // Batch n first, then batch n+1, so the order records went in is the order they come back. @@ -575,8 +591,15 @@ impl Wal { } Err(_poisoned_or_io) => { // The write or the sync failed. Everything after the last good commit in this - // segment is of unknown state, so the whole batch is owed again, on a fresh segment. + // segment is of unknown state — the segment has already cut it off — so the whole + // batch is owed again, on a fresh segment. self.lost_batch = owed.into_iter().chain(records.iter().cloned()).collect(); + // The fresh segment is opened NOW, not on the next append: its directory entry is + // what tells a restart, durably, not to resume in this one. If it cannot be opened, + // the next append tries again and reports the loss again. + if self.next_segment.is_none() { + self.next_segment = self.open_next().ok(); + } Err(DurabilityLost::observed(token, at)) } } @@ -643,6 +666,17 @@ impl Wal { /// Move to the next segment. Called when the current one is poisoned or full. fn roll(&mut self) -> io::Result<()> { + let segment = match self.next_segment.take() { + Some(opened) => opened, + None => self.open_next()?, + }; + self.segment = segment; + self.segments_used += 1; + Ok(()) + } + + /// Open the segment after the current one. + fn open_next(&mut self) -> io::Result { let next = self.segment.index() + 1; let backend = self.factory.open(next)?; let mut segment = Segment::open_at(backend, next, 0, self.ceiling)?; @@ -653,15 +687,13 @@ impl Wal { if let Some(q) = recovered.quarantined { self.take_quarantine(q); } - self.segment = segment; - self.segments_used += 1; - Ok(()) + Ok(segment) } } -/// Open the segment a restart has to resume in, and hand back the records on its tail. +/// Open the segment a restart appends to, and hand back the records on the log's tail. /// -/// ## Which segment that is +/// ## Which segment the tail is in /// /// The highest one the factory has a backing for — a directory listing on disk, the resident slots /// in memory — and NOT index zero. A log that has rolled has its newest records in its newest @@ -676,6 +708,13 @@ impl Wal { /// actually end in, or reaches the first. In practice that is one step at most; it is a loop because /// nothing forbids a run of them. /// +/// ## Which segment the appends go to +/// +/// The highest one, ALWAYS — even when it holds nothing and the tail was read from below it. An empty +/// highest segment is one a roll opened: either the segment below it was full, or it lost a sync and +/// the log opened this one the moment it did. Either way the segment below takes no more writes, +/// and resuming in it would hand a restart the one segment a previous process knew was finished. +/// /// ## Why the seeding does not scan the earlier segments /// /// Recovery is O(the last segment), and it is meant to stay that way: a node that has been up for a @@ -709,6 +748,9 @@ fn open_tail( ) -> Result<(Segment, Recovered, Vec), OpenError> { let mut index = factory.highest_index()?.unwrap_or(0); let mut quarantined = Vec::new(); + // The highest segment, held while the walk steps back below it for the tail: the appends go to + // it, never to the segment the tail is found in. + let mut newest: Option = None; loop { let backend = factory.open(index)?; let mut segment = Segment::open_at(backend, index, 0, ceiling)?; @@ -722,7 +764,10 @@ fn open_tail( quarantined.push(q); } if !recovered.records.is_empty() || index == 0 { - return Ok((segment, recovered, quarantined)); + return Ok((newest.unwrap_or(segment), recovered, quarantined)); + } + if newest.is_none() { + newest = Some(segment); } index -= 1; } diff --git a/crates/busbar-kernel/src/tests/members/wal/corruption_verdict_tests.rs b/crates/busbar-kernel/src/tests/members/wal/corruption_verdict_tests.rs index beb9f32ea5..c69e256f21 100644 --- a/crates/busbar-kernel/src/tests/members/wal/corruption_verdict_tests.rs +++ b/crates/busbar-kernel/src/tests/members/wal/corruption_verdict_tests.rs @@ -216,6 +216,8 @@ const VERSION_1_SEGMENT: &[u8] = include_bytes!("vectors/legacy_segment.bin"); /// A VERSION-1 SEGMENT IS A LAYOUT THIS BUILD DOES NOT READ: quarantined whole, never read and never /// cut. Version 1 was never released, so no segment in the field holds it; reading it under its old /// rule (no header check, a digest failure cut as a torn tail) was a downgrade path and nothing else. +/// Its frames end in zeros, as every layout before the end mark did; they are still not unfinished +/// writes of this layout, because their version is not this layout's. /// Every record it holds is set aside with its identity taken, so no acknowledged number is handed /// out again. #[test] @@ -234,7 +236,9 @@ fn a_version_1_segment_is_quarantined_never_read() { assert_eq!(frame[0..4], current[0..4]); assert_eq!(frame[6..34], current[6..34]); assert_eq!(frame[34..64], [0u8; 30]); - assert_eq!(frame[96..], current[96..]); + // Every payload byte; the current layout's last byte is its end mark, which version 1 did + // not have. + assert_eq!(frame[96..FRAME_BYTES - 1], current[96..FRAME_BYTES - 1]); assert_eq!( decode_frame(frame), Err(FrameError::UnknownVersion { found: 1 }) @@ -334,14 +338,17 @@ fn a_whole_final_frame_with_an_altered_header_is_quarantined_not_cut() { } } -/// A tear INSIDE a version-2 frame's payload (the header was written whole) cannot be told from a -/// whole frame altered afterwards, so it is quarantined rather than cut: the loud side of the -/// ambiguity. Nothing is lost either way; the quarantine holds the bytes and says so. +/// **A TEAR INSIDE A FRAME'S PAYLOAD IS A TORN TAIL, CUT SILENTLY** (Q128 kernel-wal 5). The write +/// got past the header and the digest and stopped in the payload, leaving zeros to the frame's end. +/// The frame's end mark is one of those zeros, so the write that made it never completed: nothing +/// in it was acknowledged, and an ordinary crash is not reported as corruption. RED before the fix: +/// the header checked and the digest did not, which read as a whole frame altered afterwards, and +/// the crash raised a quarantine. #[test] -fn a_tear_inside_a_whole_header_frame_is_quarantined_not_cut() { +fn a_tear_inside_a_frames_payload_is_a_torn_tail_cut_silently() { let dir = TempDir::new("tear-payload"); let written = four(); - lay_down(dir.path(), &written); + let original = lay_down(dir.path(), &written); let path = segment_path(dir.path()); let mut bytes = std::fs::read(&path).unwrap(); for b in &mut bytes[3 * FRAME_BYTES + 100..] { @@ -355,7 +362,118 @@ fn a_tear_inside_a_whole_header_frame_is_quarantined_not_cut() { ) .unwrap(); assert_eq!(wal.recovered(), &written[..3]); - assert_eq!(wal.quarantined().len(), 1); + assert!(wal.quarantined().is_empty(), "a crash is not corruption"); + assert!(quarantine_files(dir.path()).is_empty()); + assert_eq!( + std::fs::read(&path).unwrap(), + original[..3 * FRAME_BYTES].to_vec(), + "the torn frame is cut" + ); + assert_eq!( + wal.next_free_seq(3), + 4, + "an identity nothing acknowledged is not taken" + ); +} + +/// Write each of `commits` as its own group commit to a fresh log in `dir`, close it, and return +/// the committed bytes of segment 0. +fn lay_down_commits(dir: &Path, commits: &[Vec]) -> Vec { + let mut wal = + Wal::in_directory(dir, Box::new(NullShipper::new()), super::fixtures::wall_ms).unwrap(); + let mut end = 0; + for commit in commits { + end = wal + .append_batch(&durability_token(), METER, commit) + .unwrap() + .durable_end; + } + drop(wal); + let bytes = std::fs::read(segment_path(dir)).unwrap(); + bytes[..usize::try_from(end).unwrap()].to_vec() +} + +/// **A COMMIT WHOSE PAGES LANDED OUT OF ORDER IS A TORN TAIL, NOT CORRUPTION** (Q128 kernel-wal 5). +/// +/// A group commit spanning more than one page is written back in whatever order the kernel picks. +/// A crash during its sync can land one page and not the page before it: frames that verify sit +/// past frames that never landed. They are frames of the SAME commit, which never synced, so +/// nothing in it was acknowledged — the commit is cut silently and its identities stay free. RED +/// before the fix: any verifying frame past the stop meant "corrupt", whichever commit it was in. +#[test] +fn a_commit_whose_pages_landed_out_of_order_is_a_torn_tail() { + let dir = TempDir::new("pages-out-of-order"); + // Two acknowledged records, then a commit of twelve single-frame records: frames 2..14, across + // the 4 KiB page boundary at frame 8. + let acknowledged = records(3, 1, 2, 200); + let torn_commit = records(3, 3, 12, 200); + let original = lay_down_commits(dir.path(), &[acknowledged.clone(), torn_commit.clone()]); + assert_eq!(original.len(), 14 * FRAME_BYTES); + + // The second page landed; the first page's share of the torn commit did not. + let path = segment_path(dir.path()); + let mut bytes = std::fs::read(&path).unwrap(); + for b in &mut bytes[2 * FRAME_BYTES..8 * FRAME_BYTES] { + *b = 0; + } + std::fs::write(&path, bytes).unwrap(); + + let wal = Wal::in_directory( + dir.path(), + Box::new(NullShipper::new()), + super::fixtures::wall_ms, + ) + .unwrap(); + assert_eq!(wal.recovered(), acknowledged.as_slice()); + assert!( + wal.quarantined().is_empty(), + "a crash mid-commit is not corruption" + ); + assert!(quarantine_files(dir.path()).is_empty()); + assert_eq!( + std::fs::read(&path).unwrap(), + original[..2 * FRAME_BYTES].to_vec(), + "the whole torn commit is cut, the frames of it that landed included" + ); + assert_eq!( + wal.next_free_seq(3), + 3, + "no identity of the commit nothing acknowledged is taken" + ); +} + +/// The other side of the same rule: damage in a commit that a SUBSEQUENT commit acknowledges is +/// corruption whatever it looks like — even a frame of zeros, the shape a crash leaves — because a +/// commit is only written after the one before it synced. +#[test] +fn damage_in_a_commit_a_subsequent_commit_acknowledges_is_corrupt_whatever_its_shape() { + let dir = TempDir::new("acknowledged-zeros"); + let first = records(3, 1, 4, 200); + let next = records(3, 5, 2, 200); + lay_down_commits(dir.path(), &[first.clone(), next.clone()]); + + let path = segment_path(dir.path()); + let mut bytes = std::fs::read(&path).unwrap(); + for b in &mut bytes[FRAME_BYTES..2 * FRAME_BYTES] { + *b = 0; + } + std::fs::write(&path, bytes).unwrap(); + + let wal = Wal::in_directory( + dir.path(), + Box::new(NullShipper::new()), + super::fixtures::wall_ms, + ) + .unwrap(); + assert_eq!(wal.recovered(), &first[..1]); + let q = wal.quarantined(); + assert_eq!(q.len(), 1, "acknowledged records lost are reported"); + assert_eq!(q[0].damage_at, FRAME_BYTES as u64); + let mut ids = q[0].identities.clone(); + ids.sort_unstable(); + assert_eq!(ids, vec![(3, 3), (3, 4), (3, 5), (3, 6)]); + assert_eq!(quarantine_files(dir.path()).len(), 1); + assert_eq!(wal.next_free_seq(3), 7); } #[test] diff --git a/crates/busbar-kernel/src/tests/members/wal/fixtures.rs b/crates/busbar-kernel/src/tests/members/wal/fixtures.rs index ad0817c5f6..ffdd4a484f 100644 --- a/crates/busbar-kernel/src/tests/members/wal/fixtures.rs +++ b/crates/busbar-kernel/src/tests/members/wal/fixtures.rs @@ -34,11 +34,13 @@ pub enum Fault { WriteEio, } -/// A disk that can be told, from outside, to fail its next sync or write. +/// A disk that can be told, from outside, to fail its next sync or write — or to refuse to create a +/// new segment at all. #[derive(Debug, Clone, Default)] pub struct FaultSwitch { fault: Arc>, syncs: Arc>, + refuse_opens: Arc>, } impl FaultSwitch { @@ -47,6 +49,7 @@ impl FaultSwitch { FaultSwitch { fault: Arc::new(Mutex::new(Fault::None)), syncs: Arc::new(Mutex::new(0)), + refuse_opens: Arc::new(Mutex::new(false)), } } @@ -60,6 +63,12 @@ impl FaultSwitch { self.arm(Fault::None); } + /// While set, the factory refuses to open a segment it does not already hold — what a full + /// volume or an exhausted descriptor table does to a roll. + pub fn refuse_new_segments(&self, refuse: bool) { + *self.refuse_opens.lock().unwrap() = refuse; + } + /// How many syncs the disk has been asked for. This is what makes "one sync per group commit" /// a measurement rather than a claim. pub fn syncs(&self) -> u64 { @@ -128,7 +137,16 @@ pub struct FaultyFactory { impl FaultyFactory { /// A factory and the switch that drives it. pub fn new() -> (Self, FaultSwitch, MemoryFactory) { - let inner = MemoryFactory::new(); + FaultyFactory::over(MemoryFactory::new()) + } + + /// The same over memory that keeps every segment it ever handed out, the way a data directory + /// keeps its files across a restart. + pub fn retaining() -> (Self, FaultSwitch, MemoryFactory) { + FaultyFactory::over(MemoryFactory::retaining()) + } + + fn over(inner: MemoryFactory) -> (Self, FaultSwitch, MemoryFactory) { let switch = FaultSwitch::new(); ( FaultyFactory { @@ -143,6 +161,12 @@ impl FaultyFactory { impl SegmentFactory for FaultyFactory { fn open(&mut self, index: u64) -> io::Result> { + if *self.switch.refuse_opens.lock().unwrap() && self.inner.existing(index)?.is_none() { + return Err(io::Error::new( + io::ErrorKind::StorageFull, + "simulated ENOSPC creating the next segment", + )); + } Ok(Box::new(FaultyBackend { inner: busbar_kernel_wal::backend::MemorySegment::over(self.inner.segment_bytes(index)), switch: self.switch.clone(), @@ -176,6 +200,32 @@ impl SegmentFactory for FaultyFactory { } } +/// A store that keeps every record it was handed, in order, behind a shared handle so whoever +/// configured it can see what reached it after handing ownership to the log. +#[derive(Debug, Default, Clone)] +pub struct KeepingShipper { + records: Arc>>, +} + +impl KeepingShipper { + /// A fresh one. + pub fn new() -> Self { + KeepingShipper::default() + } + + /// A snapshot of everything shipped so far, in order. + pub fn records(&self) -> Vec { + self.records.lock().unwrap().clone() + } +} + +impl busbar_kernel_wal::ship::Shipper for KeepingShipper { + fn ship(&mut self, records: &[Record]) -> Result<(), busbar_kernel_wal::ship::ShipError> { + self.records.lock().unwrap().extend_from_slice(records); + Ok(()) + } +} + /// `n` records with bodies of the given length, numbered from `first_seq`. pub fn records(node: u64, first_seq: u64, n: u64, body_len: usize) -> Vec { (0..n) diff --git a/crates/busbar-kernel/src/tests/members/wal/journal_chain_tests.rs b/crates/busbar-kernel/src/tests/members/wal/journal_chain_tests.rs index cb2a101592..028b6c3d7b 100644 --- a/crates/busbar-kernel/src/tests/members/wal/journal_chain_tests.rs +++ b/crates/busbar-kernel/src/tests/members/wal/journal_chain_tests.rs @@ -15,10 +15,10 @@ use busbar_kernel_wal::journal::{ decode_run, tail_of, verify, Entry, Journal, JournalBreakKind, JournalRecord, RecordClass, JOURNAL_HEADER_BYTES, JOURNAL_MAGIC, JOURNAL_VERSION, OVERFLOW_HISTORY_RECORDS, }; -use busbar_kernel_wal::ship::{BufferShipper, NullShipper, ShipError, Shipper}; +use busbar_kernel_wal::ship::{NullShipper, ShipError, Shipper}; use busbar_kernel_wal::wal::{Mode, Wal}; -use super::fixtures::{durability_token, TempDir}; +use super::fixtures::{durability_token, KeepingShipper, TempDir}; /// A store that will not take anything. What a memory-buffered node's durability failure looks like /// from the log's side, and the only way to make the buffer grow at all. @@ -48,7 +48,8 @@ fn the_record_header_is_fixed() { assert_eq!(JOURNAL_MAGIC, *b"BJRN"); assert_eq!(JOURNAL_VERSION, 1); - let mut journal = Journal::memory_buffered(7, super::fixtures::wall_ms); + let mut journal = + Journal::memory_buffered_to(7, Box::new(NullShipper::new()), super::fixtures::wall_ms); let token = durability_token(); let ack = journal .append( @@ -87,7 +88,8 @@ fn the_record_header_is_fixed() { /// Every unit's records go on ONE chain, in the order they were sealed, and the run verifies. #[test] fn one_chain_carries_every_unit() { - let mut journal = Journal::memory_buffered(1, super::fixtures::wall_ms); + let mut journal = + Journal::memory_buffered_to(1, Box::new(NullShipper::new()), super::fixtures::wall_ms); let token = durability_token(); for class in [ RecordClass::Transaction, @@ -118,7 +120,8 @@ fn one_chain_carries_every_unit() { /// The chain catches a record whose BODY was edited, at that record. #[test] fn chain_verification_catches_a_mutated_body() { - let mut journal = Journal::memory_buffered(2, super::fixtures::wall_ms); + let mut journal = + Journal::memory_buffered_to(2, Box::new(NullShipper::new()), super::fixtures::wall_ms); let token = durability_token(); let ack = journal .append( @@ -140,7 +143,8 @@ fn chain_verification_catches_a_mutated_body() { /// and not only the identity and the two digests. #[test] fn chain_verification_catches_a_mutated_header() { - let mut journal = Journal::memory_buffered(2, super::fixtures::wall_ms); + let mut journal = + Journal::memory_buffered_to(2, Box::new(NullShipper::new()), super::fixtures::wall_ms); let token = durability_token(); let mut run = journal .append( @@ -186,16 +190,17 @@ fn chain_verification_catches_a_mutated_header() { /// identity it already holds by passing the record over and reporting success, so the settlements in /// the gap are on no medium and every append said `Ok`. /// -/// So the fixture is built the way the defect is: a real journal, sealed properly, resumed from its -/// own head at a number two past where it stopped. Nothing is edited by hand, which is what makes +/// So the fixture is built the way the defect is: a real journal, sealed properly, continued from its +/// own head at a number two past where it stopped — each continuing record sealed exactly as the +/// journal seals one (body digest, link onto the head, chain hash over the header), so the ONLY +/// thing wrong with the run is the numbering. Nothing is edited after sealing, which is what makes /// this a different failure from the three above. #[test] fn chain_verification_catches_a_gap_in_the_numbering() { - let shipper = BufferShipper::new(); let token = durability_token(); let mut journal = - Journal::memory_buffered_to(5, Box::new(shipper.clone()), super::fixtures::wall_ms); + Journal::memory_buffered_to(5, Box::new(NullShipper::new()), super::fixtures::wall_ms); let first = journal .append( &token, @@ -207,20 +212,29 @@ fn chain_verification_catches_a_gap_in_the_numbering() { assert_eq!(journal.next_seq(), 4, "the log ended on three"); // A node coming back up from the right head but at the wrong number: it skips 4 and 5. - let mut resumed = Journal::resuming( - Wal::memory_buffered_to(Box::new(BufferShipper::new()), super::fixtures::wall_ms), - 5, - journal.head(), - 6, - ); - let after = resumed - .append( - &token, - StepName::Meter, - &entries(RecordClass::Transaction, 2, 2), - ) - .expect("the store takes it") - .sealed; + let mut head = journal.head(); + let after: Vec = entries(RecordClass::Transaction, 2, 2) + .into_iter() + .zip(6u64..) + .map(|(entry, node_seq)| { + let mut record = JournalRecord { + class: entry.class, + node: 5, + node_seq, + lease_epoch: entry.lease_epoch, + policy_epoch: entry.policy_epoch, + wall: entry.wall, + mono: entry.mono, + body_hash: busbar_kernel_wal::journal::body_digest(&entry.body), + body: entry.body, + prev_hash: head, + hash: [0u8; 32], + }; + record.hash = record.digest_of_chain(); + head = record.hash; + record + }) + .collect(); assert_eq!(after[0].node_seq, 6, "the fixture really did skip two"); // The run links and digests perfectly, and is still wrong. @@ -250,7 +264,8 @@ fn chain_verification_catches_a_gap_in_the_numbering() { /// A record REMOVED from the middle breaks the link rather than passing as a shorter history. #[test] fn chain_verification_catches_a_removed_record() { - let mut journal = Journal::memory_buffered(2, super::fixtures::wall_ms); + let mut journal = + Journal::memory_buffered_to(2, Box::new(NullShipper::new()), super::fixtures::wall_ms); let token = durability_token(); let mut run = journal .append( @@ -275,7 +290,7 @@ fn without_a_data_dir_the_journal_creates_no_file() { assert!(dir.walk().is_empty()); let cwd_before = super::fixtures::cwd_segment_names(); - let shipper = BufferShipper::new(); + let shipper = KeepingShipper::new(); let mut journal = Journal::memory_buffered_to(3, Box::new(shipper.clone()), super::fixtures::wall_ms); assert_eq!(journal.mode(), Mode::MemoryBuffered); @@ -306,11 +321,11 @@ fn without_a_data_dir_the_journal_creates_no_file() { assert_eq!(shipper.records().len(), 24); } -/// A restart without a data directory loses nothing that was SHIPPED. What the store acknowledged -/// is what exists, so the chain resumes from the store's own copy and the next record links onto it. +/// Without a data directory, what the store took IS the chain: the shipped batches decode, verify +/// end to end, and end at the node's own head and number. #[test] -fn a_restart_without_a_data_dir_loses_nothing_that_was_shipped() { - let shipper = BufferShipper::new(); +fn without_a_data_dir_what_the_store_took_is_the_chain_to_the_nodes_head() { + let shipper = KeepingShipper::new(); let token = durability_token(); let (head_before, next_before) = { @@ -335,27 +350,6 @@ fn a_restart_without_a_data_dir_loses_nothing_that_was_shipped() { let (head, next_seq) = tail_of(&shipped, 9).expect("this node's own tail"); assert_eq!(head, head_before, "the store's head is the node's head"); assert_eq!(next_seq, next_before); - - // And a node coming back up continues that chain rather than starting a second one. - let mut restarted = Journal::resuming( - Wal::memory_buffered_to(Box::new(shipper.clone()), super::fixtures::wall_ms), - 9, - head, - next_seq, - ); - let ack = restarted - .append( - &token, - StepName::Meter, - &entries(RecordClass::Checkpoint, 1, 9), - ) - .expect("the store takes it"); - assert_eq!(ack.sealed[0].prev_hash, head_before); - assert_eq!(ack.sealed[0].node_seq, 10); - - let whole = decode_run(&shipper.records()).expect("still journal records"); - verify(&whole).expect("the chain across the restart verifies end to end"); - assert_eq!(whole.len(), 10); } /// With a data directory the journal replays to the same head it had before the restart, off its own diff --git a/crates/busbar-kernel/src/tests/members/wal/kill_at_every_offset_tests.rs b/crates/busbar-kernel/src/tests/members/wal/kill_at_every_offset_tests.rs index b046a8f9e5..8584b7b0f1 100644 --- a/crates/busbar-kernel/src/tests/members/wal/kill_at_every_offset_tests.rs +++ b/crates/busbar-kernel/src/tests/members/wal/kill_at_every_offset_tests.rs @@ -7,6 +7,11 @@ //! battery writes a run of records, truncates the segment at EVERY offset from zero to the end, //! recovers, and asserts the recovered records are exactly the longest complete prefix of what was //! written. A special case in the scan shows up here as the one offset the loop fails on. +//! +//! A crash does not only shorten a file. The space a segment claims ahead of its writes is zeros, +//! so a write that stopped part way leaves the file its full length with zeros from the stopping +//! point on — and that is walked at every offset too, and must read as a torn tail, never as +//! corruption. use busbar_kernel_wal::backend::{MemoryFactory, SegmentFactory as _}; use busbar_kernel_wal::record::{Record, FRAME_BYTES}; @@ -60,6 +65,34 @@ fn recover_from(bytes: &[u8]) -> Vec { .records } +/// Write each of `commits` as its own group commit into a fresh memory segment and return exactly +/// the committed bytes. +fn lay_down_commits(commits: &[Vec]) -> Vec { + let mut factory = MemoryFactory::retaining(); + let mut wal = busbar_kernel_wal::wal::Wal::with_parts( + Box::new(factory.clone()), + Box::new(busbar_kernel_wal::ship::NullShipper::new()), + busbar_kernel_wal::wal::Mode::OnDisk, + TEST_CEILING, + super::fixtures::wall_ms, + ) + .unwrap(); + let token = durability_token(); + let mut end = 0; + for commit in commits { + end = wal + .append_batch(&token, busbar_contract::caps::StepName::Meter, commit) + .unwrap() + .durable_end; + } + drop(wal); + let backend = factory.open(0).unwrap(); + let mut bytes = vec![0u8; usize::try_from(end).unwrap()]; + let n = backend.read_at(0, &mut bytes).unwrap(); + assert_eq!(n, bytes.len()); + bytes +} + /// How many whole records survive a cut at `offset`, given the frame count of each record. fn expected_prefix(written: &[Record], offset: usize) -> usize { let mut end = 0usize; @@ -111,6 +144,65 @@ fn a_cut_at_every_byte_offset_recovers_the_longest_complete_prefix() { } } +/// **A WRITE THAT STOPPED AT ANY BYTE, WITH ZEROS AFTER IT, IS A TORN TAIL — NEVER CORRUPTION** +/// (Q128 kernel-wal 5). +/// +/// The shape a crash mid-commit really leaves on a segment that claimed its space ahead: the file +/// keeps its length, the bytes up to where the write stopped are there, and everything after is the +/// claimed zeros. Walked at every byte of two group commits. Nothing past the stopping point was +/// acknowledged — the commit it is in never synced, and nothing after it was written — so the cut is +/// silent: no quarantine, no identity taken. RED before the fix: a stop inside a frame's payload +/// (its header and digest written, its payload not) read as a whole frame altered afterwards, and +/// raised a corruption alarm over an ordinary crash. +#[test] +fn a_zero_padded_tear_at_every_byte_offset_is_torn_never_corrupt() { + let mut first = Vec::new(); + let mut second = Vec::new(); + for (i, body_len) in [0usize, 10, 415, 416, 417, 900, 1000] + .into_iter() + .enumerate() + { + let record = Record::new(7, 100 + i as u64, vec![(i + 1) as u8; body_len]); + if i < 3 { + first.push(record); + } else { + second.push(record); + } + } + let bytes = lay_down_commits(&[first.clone(), second.clone()]); + let mut written = first; + written.extend(second); + // Space claimed ahead of the writes, which a crash leaves as zeros too. + let claimed = bytes.len() + 4 * FRAME_BYTES; + + for stop in 0..=bytes.len() { + let mut torn = bytes[..stop].to_vec(); + torn.resize(claimed, 0); + let shared = busbar_kernel_wal::backend::SharedBytes::new(std::sync::Mutex::new(torn)); + let backend = Box::new(busbar_kernel_wal::backend::MemorySegment::over(shared)); + let mut segment = Segment::open_at(backend, 0, 0, TEST_CEILING).unwrap(); + let mut sink = MemoryFactory::new(); + let recovered = + recover_and_truncate(&mut segment, &mut sink, super::fixtures::wall_ms).unwrap(); + let expected = expected_prefix(&written, stop); + assert!( + !recovered.was_corrupt(), + "a write that stopped at byte {stop} of {} read as corruption: {:?}", + bytes.len(), + recovered.verdict + ); + assert!( + recovered.quarantined.is_none() && sink.quarantined().is_empty(), + "a write that stopped at byte {stop} was quarantined" + ); + assert_eq!( + recovered.records, + written[..expected].to_vec(), + "the records recovered after a stop at byte {stop} are not the longest complete prefix" + ); + } +} + #[test] fn a_cut_mid_record_drops_the_whole_record_not_a_short_one() { // One record that needs three frames. A cut after two of them must produce nothing: a truncated diff --git a/crates/busbar-kernel/src/tests/members/wal/no_disk_tests.rs b/crates/busbar-kernel/src/tests/members/wal/no_disk_tests.rs index a65ed2a617..7694fa93cd 100644 --- a/crates/busbar-kernel/src/tests/members/wal/no_disk_tests.rs +++ b/crates/busbar-kernel/src/tests/members/wal/no_disk_tests.rs @@ -27,7 +27,10 @@ fn a_memory_buffered_log_creates_no_file_anywhere() { ); let cwd_before = cwd_segment_names(); - let mut wal = Wal::memory_buffered(super::fixtures::wall_ms); + let mut wal = Wal::memory_buffered_to( + Box::new(busbar_kernel_wal::ship::NullShipper::new()), + super::fixtures::wall_ms, + ); assert_eq!(wal.mode(), Mode::MemoryBuffered); let token = durability_token(); // Enough records, and enough of them large enough to continue across frames, that any @@ -58,7 +61,7 @@ fn a_memory_buffered_log_creates_no_file_anywhere() { #[test] fn a_memory_buffered_log_ships_every_committed_batch_synchronously() { - let shipper = busbar_kernel_wal::ship::BufferShipper::new(); + let shipper = super::fixtures::KeepingShipper::new(); let mut wal = Wal::memory_buffered_to(Box::new(shipper.clone()), super::fixtures::wall_ms); let token = durability_token(); let first = records(4, 1, 3, 30); diff --git a/crates/busbar-kernel/src/tests/members/wal/poison_tests.rs b/crates/busbar-kernel/src/tests/members/wal/poison_tests.rs index c5c31f64fb..3d3a974814 100644 --- a/crates/busbar-kernel/src/tests/members/wal/poison_tests.rs +++ b/crates/busbar-kernel/src/tests/members/wal/poison_tests.rs @@ -183,17 +183,25 @@ fn a_group_commit_costs_one_sync_however_many_records_are_in_it() { fn a_poisoned_segment_never_takes_another_write() { let (mut wal, switch, memory) = wal_with_faults(); let token = durability_token(); - switch.arm(Fault::SyncEio); + // A batch that lands first, so the poisoned segment holds bytes once the failed batch is cut + // off it: the region the claim is about. wal.append_batch( &token, busbar_contract::caps::StepName::Meter, &records(1, 1, 1, 10), ) + .unwrap(); + switch.arm(Fault::SyncEio); + wal.append_batch( + &token, + busbar_contract::caps::StepName::Meter, + &records(1, 2, 1, 10), + ) .expect_err("armed"); // The poisoned segment's own bytes, held by a strong reference so they survive the roll that - // follows. A fresh segment NUMBER is not the claim: the claim is that the region of unknown - // state receives no further writes, and only the bytes can say that. Reading them back through + // follows. A fresh segment NUMBER is not the claim: the claim is that the segment that lost a + // sync receives no further writes, and only the bytes can say that. Reading them back through // `wal.read_back()` afterwards would read the NEW segment, which is why the factory is asked. let poisoned_bytes = memory.segment_bytes(0); let before = poisoned_bytes.lock().unwrap().clone(); @@ -203,12 +211,12 @@ fn a_poisoned_segment_never_takes_another_write() { ); // The switch is one-shot, so the disk is healthy again — but the segment stays closed and the - // log moves on rather than writing more bytes into a region of unknown state. + // log moves on rather than writing more bytes into a segment that lost a sync. let ack = wal .append_batch( &token, busbar_contract::caps::StepName::Meter, - &records(1, 2, 1, 10), + &records(1, 3, 1, 10), ) .unwrap(); assert!(ack.segment > 0, "the write went to a fresh segment"); @@ -219,6 +227,244 @@ fn a_poisoned_segment_never_takes_another_write() { ); } +/// A log over memory that keeps every segment, as a data directory keeps its files, so a second log +/// opened over the same memory is a restart. +fn restartable_wal_with_faults() -> ( + Wal, + super::fixtures::FaultSwitch, + busbar_kernel_wal::backend::MemoryFactory, +) { + let (factory, switch, memory) = FaultyFactory::retaining(); + let wal = Wal::with_parts( + Box::new(factory), + Box::new(busbar_kernel_wal::ship::NullShipper::new()), + Mode::OnDisk, + CEILING, + super::fixtures::wall_ms, + ) + .unwrap(); + (wal, switch, memory) +} + +fn restart(memory: &busbar_kernel_wal::backend::MemoryFactory) -> Wal { + Wal::with_parts( + Box::new(memory.clone()), + Box::new(busbar_kernel_wal::ship::NullShipper::new()), + Mode::OnDisk, + CEILING, + super::fixtures::wall_ms, + ) + .unwrap() +} + +/// **A BATCH THE CALLER WAS TOLD WAS LOST DOES NOT COME BACK AT A RESTART** (Q128 kernel-wal 3). +/// +/// After a write error the failed batch's pages can still be readable — here they are, in the +/// memory the failing disk wrote them to — without ever having reached the medium. A restart that +/// read them would put on the book records the caller was told were lost, and chain the next ones +/// onto records that are on no disk. So the failed batch is cut off its segment when it is lost. +/// RED before the fix: the poison was an in-memory flag, the bytes stayed, and the restart +/// recovered the lost batch as the tail of the log. +#[test] +fn a_batch_reported_lost_does_not_come_back_at_a_restart() { + let (mut wal, switch, memory) = restartable_wal_with_faults(); + let token = durability_token(); + + let landed = records(1, 1, 2, 40); + wal.append_batch(&token, busbar_contract::caps::StepName::Meter, &landed) + .unwrap(); + switch.arm(Fault::SyncEio); + let lost = records(1, 3, 2, 40); + wal.append_batch(&token, busbar_contract::caps::StepName::Meter, &lost) + .expect_err("the sync was armed to fail"); + drop(wal); + + let restarted = restart(&memory); + assert_eq!( + restarted.recovered(), + landed.as_slice(), + "the restart's tail is what was acknowledged, not the batch reported lost" + ); + assert!( + !restarted.holds(1, 3), + "the lost batch's identities are free" + ); +} + +/// **A RESTART NEVER APPENDS TO THE SEGMENT THAT LOST A SYNC** (Q128 kernel-wal 3). +/// +/// Poison was a flag in the process that saw the sync fail, and the roll away from the poisoned +/// segment waited for the next append; a restart before it opened the same segment unpoisoned and +/// appended to it. Now the next segment is opened the moment the sync fails, and a restart appends +/// to the newest segment even when its tail is read from the one below. RED before the fix: the +/// restart appended into segment zero. +#[test] +fn a_restart_never_appends_to_the_segment_that_lost_a_sync() { + let (mut wal, switch, memory) = restartable_wal_with_faults(); + let token = durability_token(); + let segment_zero = memory.segment_bytes(0); + + let landed = records(1, 1, 2, 40); + wal.append_batch(&token, busbar_contract::caps::StepName::Meter, &landed) + .unwrap(); + switch.arm(Fault::SyncEio); + wal.append_batch( + &token, + busbar_contract::caps::StepName::Meter, + &records(1, 3, 2, 40), + ) + .expect_err("the sync was armed to fail"); + drop(wal); + + let mut restarted = restart(&memory); + let before = segment_zero.lock().unwrap().clone(); + let after_restart = records(1, 3, 2, 40); + let ack = restarted + .append_batch( + &token, + busbar_contract::caps::StepName::Meter, + &after_restart, + ) + .expect("a healthy disk takes the append"); + assert_eq!(ack.segment, 1, "the append went to the segment after it"); + assert_eq!( + *segment_zero.lock().unwrap(), + before, + "the segment that lost a sync took a write after the restart" + ); + let mut expected = landed; + expected.extend(after_restart); + assert_eq!(restarted.read_back().unwrap().records, expected); +} + +/// **A ROLL THAT CANNOT OPEN THE NEXT SEGMENT KEEPS THE BATCH IT WAS HANDED** (Q128 kernel-wal 4). +/// +/// The journal chains a batch before the log takes it. When the segment that lost a sync cannot be +/// replaced — a full volume, no descriptors left — every other failure arm retains the batch for +/// the retry; this one returned before it, so the records were on no medium and in no queue, and +/// once the disk recovered the next record linked onto a head that was never written. RED before +/// the fix: `owed()` held only the earlier lost batch. +#[test] +fn a_roll_that_cannot_open_the_next_segment_keeps_the_batch_it_was_handed() { + let (mut wal, switch, _memory) = wal_with_faults(); + let token = durability_token(); + + // The volume is full: the sync fails, and no segment can be opened to move to. + switch.refuse_new_segments(true); + switch.arm(Fault::SyncEio); + let n = records(1, 1, 2, 40); + wal.append_batch(&token, busbar_contract::caps::StepName::Meter, &n) + .expect_err("the sync was armed to fail"); + + let n_plus_one = records(1, 3, 2, 40); + wal.append_batch(&token, busbar_contract::caps::StepName::Meter, &n_plus_one) + .expect_err("no segment can be opened to roll to"); + let mut owed = n.clone(); + owed.extend(n_plus_one.clone()); + assert_eq!( + wal.owed(), + owed.as_slice(), + "both batches are retained, in order" + ); + + switch.refuse_new_segments(false); + let n_plus_two = records(1, 5, 1, 40); + let ack = wal + .append_batch(&token, busbar_contract::caps::StepName::Meter, &n_plus_two) + .expect("the disk has room again"); + assert_eq!(ack.appended, 5, "all three batches, each once"); + assert!(wal.owed().is_empty()); + owed.extend(n_plus_two); + assert_eq!(wal.read_back().unwrap().records, owed); +} + +/// The same, at the journal: a failed roll leaves no hole in the chain. RED before the fix: the +/// record sealed into the refused batch was never written, and the chain read back broke at the +/// record after it. +#[test] +fn a_failed_roll_leaves_no_hole_in_the_chain() { + use busbar_kernel_wal::journal::{Entry, Journal, RecordClass}; + + let (wal, switch, memory) = wal_with_faults(); + // The first segment stays resident, the way a data directory keeps its file, so the chain read + // back is the whole log. + let _segment_zero = memory.segment_bytes(0); + let mut journal = Journal::over(wal, 4); + let token = durability_token(); + let entry = |tag: u8| [Entry::new(RecordClass::Transaction, vec![tag; 8])]; + + journal + .append(&token, busbar_contract::caps::StepName::Meter, &entry(1)) + .expect("a healthy disk"); + switch.refuse_new_segments(true); + switch.arm(Fault::SyncEio); + journal + .append(&token, busbar_contract::caps::StepName::Meter, &entry(2)) + .expect_err("the sync was armed to fail"); + journal + .append(&token, busbar_contract::caps::StepName::Meter, &entry(3)) + .expect_err("no segment can be opened to roll to"); + switch.refuse_new_segments(false); + journal + .append(&token, busbar_contract::caps::StepName::Meter, &entry(4)) + .expect("the disk has room again"); + + let chain = journal + .replay() + .expect("the log reads back") + .expect("and the chain verifies: nothing sealed was dropped"); + assert_eq!( + chain.iter().map(|r| r.node_seq).collect::>(), + vec![1, 2, 3, 4] + ); +} + +/// A journal that RETAINS at its bound keeps a failed roll's batch past that bound: the batch a +/// failed roll keeps and the records a full lane keeps are the one retained batch, and a retaining +/// journal never forgets from it to make room (H3's `retaining_at_bound`, Q128 kernel-wal 4). +#[test] +fn a_retaining_journal_keeps_a_failed_rolls_batch_past_its_bound() { + use busbar_kernel_wal::journal::{Entry, Journal, RecordClass}; + + let (wal, switch, memory) = wal_with_faults(); + let _segment_zero = memory.segment_bytes(0); + let mut journal = Journal::over(wal, 4).with_capacity(1).retaining_at_bound(); + let token = durability_token(); + let entry = |tag: u8| [Entry::new(RecordClass::Transaction, vec![tag; 8])]; + + journal + .append(&token, busbar_contract::caps::StepName::Meter, &entry(1)) + .expect("a healthy disk"); + switch.refuse_new_segments(true); + switch.arm(Fault::SyncEio); + journal + .append(&token, busbar_contract::caps::StepName::Meter, &entry(2)) + .expect_err("the sync was armed to fail"); + let ack = journal.append(&token, busbar_contract::caps::StepName::Meter, &entry(3)); + assert!(ack.is_err(), "no segment can be opened to roll to"); + assert!( + journal.at_bound(), + "two records owed against a bound of one" + ); + assert_eq!(journal.buffered(), 2, "both batches are retained"); + assert_eq!(journal.dropped_total(), 0, "nothing was forgotten"); + assert!(journal.overflows().is_empty(), "no break was sealed"); + + switch.refuse_new_segments(false); + journal + .append(&token, busbar_contract::caps::StepName::Meter, &entry(4)) + .expect("the disk has room again"); + assert_eq!(journal.buffered(), 0); + let chain = journal + .replay() + .expect("the log reads back") + .expect("and the chain verifies"); + assert_eq!( + chain.iter().map(|r| r.node_seq).collect::>(), + vec![1, 2, 3, 4] + ); +} + /// What the store is owed on disk is a QUEUE WITH A BOUND, not a list that grows for as long as the /// outage lasts. What it gives up on is counted, and those records are still in the segments. #[test] diff --git a/crates/busbar-kernel/src/tests/members/wal/vectors/legacy_segment.bin.provenance b/crates/busbar-kernel/src/tests/members/wal/vectors/legacy_segment.bin.provenance index be1b5f94c5..aea3d4273e 100644 --- a/crates/busbar-kernel/src/tests/members/wal/vectors/legacy_segment.bin.provenance +++ b/crates/busbar-kernel/src/tests/members/wal/vectors/legacy_segment.bin.provenance @@ -7,7 +7,7 @@ # b is (node_seq + b) % 251) framed under wal layout version 1 (the layout before the header check, never # released; no header check): 4 frames x 512 bytes = 2048 bytes, frame i at offset 512*i, no segment header, no padding. # Read by `a_version_1_segment_is_quarantined_never_read` in ../corruption_verdict_tests.rs: this build -# reads only version 2, so the segment is quarantined whole. +# reads only version 3, so the segment is quarantined whole. # # ENCODER REPRODUCED: busbar_kernel_wal::record::encode_legacy (crates/busbar-kernel-wal/src/record.rs, # `#[cfg(test)] pub(crate) fn encode_legacy`) at commit 4aa05a13d71ce79bf8e2320c6689e39cb6e03033: @@ -38,5 +38,6 @@ # 3 4257414c01000000030000000000000004000000000000000000000001000000c8000000000000000000000000000000000000000000000000000000000000008f436e6d36279edbc78363c382dd4bd546fdd4e085690f9bb05cd914bd470aa7 # # SELF-CHECK: the test asserts, before using the vector, that each frame is version 1, matches the -# current Record::encode byte for byte outside [4,6) (version) and [34,96) (header check zeroed, -# digest), has [34,64) zero, and is refused by decode_frame as an unknown layout (version 1). +# current Record::encode byte for byte outside [4,6) (version), [34,96) (header check zeroed, +# digest) and [511] (the current layout's end mark, which version 1 did not have), has [34,64) +# zero, and is refused by decode_frame as an unknown layout (version 1). diff --git a/crates/busbar/src/root/kernel.rs b/crates/busbar/src/root/kernel.rs index 50c6b842d2..bb2a761dc6 100644 --- a/crates/busbar/src/root/kernel.rs +++ b/crates/busbar/src/root/kernel.rs @@ -237,10 +237,10 @@ pub enum CardRefused { /// `Journal::append`, `Wal::append_batch`). So the record may yet land, and the holder put the /// card in force back on the journal behind it at the same instant ([`RootHistory::withdraw`]). Lost(busbar_contract::caps::DurabilityLost), - /// The log did not even retain the card: its segment was poisoned and a fresh one could not be - /// opened, so the batch was refused before it was held (`Wal::append_batch`). The record can - /// never land, so nothing is published and nothing is put back — the node fails closed, and - /// every later append on that log is refused at the same check until a segment opens. + /// The log neither holds nor owes the card, so the record can never land: nothing is published + /// and nothing is put back — the node fails closed. Every arm of `Wal::append_batch` retains + /// the batch it was handed, a failed roll onto a fresh segment included (kernel-wal finding 4), + /// so a log that keeps that rule never answers this; it is the answer should one not. Dropped(busbar_contract::caps::DurabilityLost), /// The boot's book has rebuilt the history but its handle is not bound yet: there is no journal /// the card could go on. @@ -1267,8 +1267,8 @@ fn journal_card( /// What a `DurabilityLost` from [`journal_card`] leaves the card as. The journal sealed the card /// last, so it is the number just below the journal's next; the log either holds it or owes it -/// (retained, offered again on the next append: IN DOUBT) — or neither, which is the one path that -/// refuses a batch before retaining it: a poisoned segment no fresh one could replace (DROPPED). +/// (retained, offered again on the next append: IN DOUBT, a failed roll onto a fresh segment +/// included) — or neither (DROPPED), which a log that retains on every arm never answers. fn refusal_of( durability: &crate::root::durability::Durability, lost: busbar_contract::caps::DurabilityLost, diff --git a/crates/busbar/src/root/tests/durability.rs b/crates/busbar/src/root/tests/durability.rs index 6c940b8d73..514998274c 100644 --- a/crates/busbar/src/root/tests/durability.rs +++ b/crates/busbar/src/root/tests/durability.rs @@ -789,11 +789,34 @@ fn journalling_carries_no_admin_row_and_the_root_holds_no_second_admin_ring() { } } +/// A store that keeps every record it was handed, in order, behind a shared handle. +#[derive(Default, Clone)] +struct KeepingShipper(std::sync::Arc>>); + +impl KeepingShipper { + fn records(&self) -> Vec { + self.0.lock().expect("the kept records").clone() + } +} + +impl busbar_kernel_wal::Shipper for KeepingShipper { + fn ship( + &mut self, + records: &[busbar_kernel_wal::Record], + ) -> Result<(), busbar_kernel_wal::ShipError> { + self.0 + .lock() + .expect("the kept records") + .extend_from_slice(records); + Ok(()) + } +} + /// The shipped batches ARE the chain, so a node with no data directory that shipped everything /// has lost nothing it shipped. #[test] fn what_the_store_took_is_the_chain() { - let shipper = busbar_kernel_wal::BufferShipper::new(); + let shipper = KeepingShipper::default(); let mut durability = build_for_node( &DurabilityConfig { data_dir: None }, 4, diff --git a/crates/busbar/src/root/tests/kernel.rs b/crates/busbar/src/root/tests/kernel.rs index 3c73a2c537..25f44da89d 100644 --- a/crates/busbar/src/root/tests/kernel.rs +++ b/crates/busbar/src/root/tests/kernel.rs @@ -1312,13 +1312,15 @@ impl busbar_kernel_wal::SegmentFactory for BreakableDisk { } } -/// **A CARD THE LOG DROPPED OUTRIGHT IS NEVER PUBLISHED** (REV-293 #1, REV-309 #1: the truly -/// dropped path fails closed). A poisoned segment that no fresh one can replace refuses the batch -/// BEFORE the log retains it (`Wal::append_batch`), so that record can never land: the apply is -/// refused as `Dropped`, nothing goes on the live history — neither the card nor a put-back — and -/// the card in force prices on. Answering it as in doubt would publish an entry no chain will hold. +/// **A CARD THE LOG COULD NOT ROLL FOR IS OWED, UNCONFIRMED, AND NEVER PUBLISHED WHILE IT IS** +/// (REV-293 #1, REV-309 #1; ARCHITECT 2026-10-07 ruling on kernel-wal finding 4). A poisoned +/// segment that no fresh one can replace keeps the batch it was handed (`Wal::append_batch`), as +/// every other arm of the log does, so the chain has no hole once the volume recovers: the card is +/// OWED. The root answers the apply as in doubt (`Lost`): the card is not confirmed, so it is never +/// put in force — the card in force prices every instant, the refused card's era included, and the +/// log owes the card for its retry. #[test] -fn a_rate_card_the_log_drops_outright_is_never_published() { +fn a_rate_card_the_log_owes_after_a_failed_roll_is_not_published_while_unconfirmed() { use busbar_contract::caps::DurableWrite; let dir = journal_dir("dropped-apply"); let holder = process_holder(); @@ -1339,7 +1341,7 @@ fn a_rate_card_the_log_drops_outright_is_never_published() { book.lock().expect("the book").journal = busbar_kernel_wal::Journal::over(log, 7); failing.store(true, std::sync::atomic::Ordering::SeqCst); // The disk goes: the write in flight poisons the segment (retained, owed), and from here no - // fresh segment opens. + // fresh segment opens; the log keeps every batch it is handed. let token = busbar_kernel::test_support::tokens::grant::(); assert!(book .lock() @@ -1357,12 +1359,20 @@ fn a_rate_card_the_log_drops_outright_is_never_published() { assert!( matches!( try_apply_at(holder, 5.0, APPLIED_B), - Err(CardRefused::Dropped(_)) + Err(CardRefused::Lost(_)) ), - "a card the log could not retain is answered as dropped" + "a card the log retained but could not write is answered as in doubt, not confirmed" + ); + assert_eq!( + flat_price_at(holder, APPLIED_B), + Some(3_000), + "the unconfirmed card is not in force from the instant it was applied at" + ); + assert_eq!( + flat_price_at(holder, EARNED_B), + Some(3_000), + "nor at any instant after it: the card in force prices on" ); - assert_eq!(holder.len(), 1, "nothing is published for a dropped card"); - assert_eq!(flat_price_at(holder, EARNED_B), Some(3_000)); let owed_cards = book .lock() .expect("the book") @@ -1373,7 +1383,11 @@ fn a_rate_card_the_log_drops_outright_is_never_published() { .filter_map(|r| busbar_kernel_wal::JournalRecord::decode(&r.body).ok()) .filter(|r| super::CardApplied::from_body(&r.body).is_some()) .count(); - assert_eq!(owed_cards, 0, "and the log owes no card"); + assert_eq!( + owed_cards, 2, + "the log owes the card, and the card in force put back behind it: both retained for their \ + retry, neither dropped" + ); drop(book); let _ = std::fs::remove_dir_all(&dir); } diff --git a/crates/plugin-loader/src/durable.rs b/crates/plugin-loader/src/durable.rs index 41ec95ff67..79af2023cd 100644 --- a/crates/plugin-loader/src/durable.rs +++ b/crates/plugin-loader/src/durable.rs @@ -3,11 +3,14 @@ //! The ONE durable-write choke point. Every durable file publish in busbar goes through here: //! `temp → write → flush → fsync(file) → rename → fsync(parent)`, with RAII tmp cleanup on EVERY -//! error path. There is no other public durable-write path in the crate (see the guard test in -//! `admin::structure` / the `structure-lint` gate): outside this module, the ephemeral -//! `plugin-loader::stage`, and `#[cfg(test)]` blocks, no source file may hand-roll a -//! `std::fs::rename` used to publish or a `sync_all` for durability. A 5th call-site that tries to -//! re-hand-roll the dance fails CI instead of compiling — the atomic-write bug class is made +//! error path. It is also the home of directory durability — the parent fsync that makes a +//! created, renamed or removed entry survive a power loss. The log's own segment files and +//! quarantine copies keep a copy of the same rule in `busbar-kernel-wal`'s backend, which takes no +//! edge onto this crate. There is no other durable-write path: the `A-persistence` +//! row of the `structure-lint` gate (`xtask/src/gates/structure_lint/choke_points.rs`) refuses a +//! hand-rolled `fs::rename(` publish, a hand-rolled `sync_all`/`sync_data`, or an +//! `fs::create_dir_all(` anywhere outside the files that row ledgers, so a call site that tries to +//! re-hand-roll the dance fails CI instead of merging — the atomic-write bug class is made //! structurally unrepresentable rather than "fixed at N sites". //! //! The primitive's contract makes it impossible to call it and skip a step: a caller supplies bytes @@ -54,76 +57,86 @@ pub struct DurableOpts { /// Atomically + durably publish `bytes` to `path` (default posture: overlay/state). /// -/// See [`write_with`] for the full contract. In short: -/// 1. create a sibling temp in the SAME directory as `path`, -/// 2. `create → write_all → flush → fsync(file)` the temp, -/// 3. `rename(temp, path)` — atomic for a concurrent reader, -/// 4. `fsync(parent dir)` — best-effort; makes the rename's directory entry durable. -/// -/// On ANY error in steps 1–3 the temp is removed by an RAII guard before returning the error, so a -/// failed write NEVER leaves a stale temp to accumulate or to wedge a retry. Steps 1–3 failing is a -/// hard error (`Err`); step 4 failing is swallowed (contents are already durable; not every FS -/// supports opening a directory for fsync). +/// See [`write_with`] for the contract. pub fn write(path: &Path, bytes: &[u8]) -> io::Result<()> { write_with(path, bytes, DurableOpts::default()) } -/// Atomically + durably publish `bytes` to `path` with `opts`. -/// -/// The primitive OWNS temp naming (a per-call-unique sibling in the target's directory) so no -/// call-site can pick a cross-directory temp (a cross-FS rename is not atomic and can `EXDEV`-fail) -/// or collide with a concurrent writer to the same target. Naming: the target file-name prefixed -/// `.` and suffixed `.-.tmp`, where `seq` is a process-monotonic counter. A leftover temp -/// from a crashed run has a DIFFERENT name and is simply ignored (it can't wedge us); the -/// `exclusive` posture additionally best-effort removes our own about-to-use name first. -/// -/// Post-condition on `Ok(())`: a concurrent reader observes either the old file or the fully-written -/// new file, never a torn/partial one (rename atomicity); and after a power loss the surfaced file -/// is the fully-written new contents (contents fsync before rename) with its directory entry durable -/// (best-effort parent fsync after). On `Err`: no temp is left behind and `path` is untouched (still -/// the prior contents, or still absent). /// The directory whose ENTRY a publish or an unlink of `path` mutates -- the one that has to be /// fsynced for the change to survive a power loss. A RELATIVE `path` has an empty parent /// (`Some("")`, which cannot be opened), so it resolves to "." -- the CWD, where the file actually /// lives. UNCONDITIONAL and in ONE place, so no caller can pass a relative path that dodges the /// parent fsync, and `write`/`remove` can never disagree about which directory it is. -fn holding_dir(path: &Path) -> &Path { +pub(crate) 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 a rename or an unlink of it is itself durable. -/// Best-effort by design: the file CONTENTS are already durable at every call site, and not every -/// platform/filesystem supports opening a directory for fsync. +/// fsync the directory that holds `path`, so a creation, rename or unlink of it is itself durable. /// -/// **ON WINDOWS THIS IS A COMPLETE NO-OP, and it is a silent one — worth naming because nothing in -/// the code shape shows it.** `File::open` on a DIRECTORY fails on Windows unless the handle is -/// opened with `FILE_FLAG_BACKUP_SEMANTICS`, so the `if let Ok(..)` below never binds and no fsync -/// is even attempted; and even with that flag `FlushFileBuffers` on a directory handle is not the -/// directory-entry barrier `fsync(dirfd)` is on unix. There is no Win32 equivalent to reach for, so -/// this is a platform gap rather than an omission: on Windows the durability of a publish rests on -/// NTFS's metadata journal ordering the rename, not on anything this function does. The FILE -/// CONTENTS half is unaffected and holds everywhere — `sync_all` on the temp runs before the rename -/// on every platform, so the failure this guards against on Windows is a lost RENAME after a power -/// loss (the old file survives, or the file is absent), never a torn or half-written one. -fn sync_holding_dir(path: &Path) { - let parent = holding_dir(path); +/// THE one home of directory durability in this crate: the publish below, [`remove`] and +/// [`create_dir_all`] all go through it (the log's segment files keep the same rule in +/// `busbar-kernel-wal`'s `backend::DirectoryFactory`). +/// +/// A failure is the caller's error. Opening the directory, or fsyncing it, failing with an I/O error +/// means the entry is NOT durable, and a caller that reported success would be reporting a +/// durability the medium did not give. The one exception is a filesystem that says the operation +/// is not supported on a directory — `EINVAL`, which `fsync(2)` names for exactly that, or an error +/// the platform reports as [`io::ErrorKind::Unsupported`]: it cannot be made to promise more than it +/// does, and refusing every publish on it would trade a weaker durability story for no durability +/// at all. +/// +/// **ON WINDOWS THIS IS A NO-OP, and it is worth naming because nothing in the code shape shows +/// it.** `File::open` on a DIRECTORY fails on Windows unless the handle is opened with +/// `FILE_FLAG_BACKUP_SEMANTICS`, and even with that flag `FlushFileBuffers` on a directory handle is +/// not the directory-entry barrier `fsync(dirfd)` is on unix. There is no Win32 equivalent to reach +/// for, so this is a platform gap rather than an omission: on Windows the durability of a publish +/// rests on NTFS's metadata journal ordering the rename, not on anything this function does. The +/// FILE CONTENTS half is unaffected and holds everywhere — `sync_all` on the temp runs before the +/// rename on every platform, so the failure this guards against on Windows is a lost RENAME after a +/// power loss (the old file survives, or the file is absent), never a torn or half-written one. +pub(crate) 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)] - fault_record_parent_fsync(parent); - if let Ok(dir) = std::fs::File::open(parent) { - let _ = dir.sync_all(); + crate::durable::hooks::parent_fsync(dir)?; + #[cfg(unix)] + { + std::fs::File::open(dir)?.sync_all() + } + // See [`sync_holding_dir`]: 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: `EINVAL` (the descriptor does not support synchronization), or an error the platform +/// reports as unsupported. +fn fsync_unsupported(e: &io::Error) -> bool { + matches!( + e.kind(), + io::ErrorKind::Unsupported | io::ErrorKind::InvalidInput + ) +} + /// DURABLY remove `path`: unlink it, then fsync the holding directory so the REMOVAL survives a /// power loss. The asymmetric sibling of [`write()`] -- installing a file fsynced the directory entry /// and removing one did not, so a crash after a plugin delete could resurrect the deleted artifact -/// on the next boot and load it. `Err` only if the unlink itself fails. +/// on the next boot and load it. `Err` if the unlink fails, or if the directory fsync after it does +/// (the file is gone, but its removal may not survive a power loss). pub fn remove(path: &Path) -> io::Result<()> { std::fs::remove_file(path)?; - sync_holding_dir(path); - Ok(()) + sync_holding_dir(path) } /// DURABLY create `path` and any missing ancestors: each directory that is actually created has its @@ -133,6 +146,7 @@ pub fn remove(path: &Path) -> io::Result<()> { /// entry were fsynced -- the same asymmetry class as an unlink that skips the parent fsync. /// /// Already-existing directories are left alone: their entries are durable by whoever created them. +/// A directory fsync that fails is this call's error, as `sync_holding_dir` says. pub fn create_dir_all(path: &Path) -> io::Result<()> { // Walk up collecting the missing ancestors (deepest first), then create shallowest first so each // `create_dir` finds its parent present. @@ -148,7 +162,7 @@ pub fn create_dir_all(path: &Path) -> io::Result<()> { for dir in missing.iter().rev() { match std::fs::create_dir(dir) { // Fsync the HOLDING directory, which is what makes `dir`'s own entry durable. - Ok(()) => sync_holding_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), @@ -157,7 +171,31 @@ pub fn create_dir_all(path: &Path) -> io::Result<()> { Ok(()) } -/// Atomically + durably publish `bytes` to `path` with `opts` (the contract is on [`write`]). +/// Atomically + durably publish `bytes` to `path` with `opts`. +/// +/// 1. create a sibling temp in the SAME directory as `path`, +/// 2. `create → write_all → flush → fsync(file)` the temp, +/// 3. `rename(temp, path)` — atomic for a concurrent reader, +/// 4. `fsync(parent dir)` — makes the rename's directory entry durable. +/// +/// The primitive OWNS temp naming (a per-call-unique sibling in the target's directory) so no +/// call-site can pick a cross-directory temp (a cross-FS rename is not atomic and can `EXDEV`-fail) +/// or collide with a concurrent writer to the same target. Naming: the target file-name prefixed +/// `.` and suffixed `.-.tmp`, where `seq` is a process-monotonic counter. A leftover temp +/// from a crashed run has a DIFFERENT name and is simply ignored (it can't wedge us); the +/// `exclusive` posture additionally best-effort removes our own about-to-use name first. +/// +/// Post-condition on `Ok(())`: a concurrent reader observes either the old file or the fully-written +/// new file, never a torn/partial one (rename atomicity); and after a power loss the surfaced file +/// is the fully-written new contents (contents fsync before rename) with its directory entry durable +/// (parent fsync after). +/// +/// On `Err` from steps 1–3: no temp is left behind and `path` is untouched (still the prior +/// contents, or still absent) — the RAII guard removes the temp on every early return. On `Err` +/// from step 4: the rename happened, so `path` holds the new contents, but its directory entry was +/// NOT made durable and a power loss may still surface the prior contents; the caller is told +/// rather than handed an `Ok` the medium did not give. A filesystem that reports a directory fsync +/// as unsupported is not an error (see `sync_holding_dir`). pub fn write_with(path: &Path, bytes: &[u8], opts: DurableOpts) -> io::Result<()> { use std::io::Write as _; use std::sync::atomic::{AtomicU64, Ordering}; @@ -187,11 +225,11 @@ pub fn write_with(path: &Path, bytes: &[u8], opts: DurableOpts) -> io::Result<() SEQ.fetch_add(1, Ordering::Relaxed) )); let tmp = parent.join(tmp_name); - // #[cfg(test)] seam: plant a decoy AT the exact about-to-use temp name. The name embeds a pid + - // an atomic sequence, so no test can predict it from the outside -- which is why the `exclusive` - // anti-wedge pre-removal below had no test at all until this existed. + // #[cfg(test)] hook: a test may plant a decoy AT the exact about-to-use temp name. The name + // embeds a pid + an atomic sequence, so no test can predict it from the outside -- which is why + // the `exclusive` anti-wedge pre-removal below had no test at all until this existed. #[cfg(test)] - plant_decoy_if_armed(&tmp); + crate::durable::hooks::plant_decoy_if_armed(&tmp); // RAII: the temp is removed on EVERY early return (every `?` below, and any future `?` an editor // adds) UNLESS we disarm after a successful rename. There is no manual cleanup to forget, so the @@ -245,20 +283,18 @@ pub fn write_with(path: &Path, bytes: &[u8], opts: DurableOpts) -> io::Result<() guard.armed = false; // published: the temp was consumed by the rename; disarm. // fsync the parent dir so the rename's directory entry is itself durable. - sync_holding_dir(path); - Ok(()) + sync_holding_dir(path) } -// ── `#[cfg(test)]`-only fault-injection seam ────────────────────────────────────────────────────── +// ── `#[cfg(test)]` hook points ───────────────────────────────────────────────────────────────────── // In a release build `fault_point!` expands to nothing, so the primitive's production path carries -// ZERO indirection — no trait object, no branch. Under test it consults a thread-local so a single -// harness can inject `ENOSPC`/`EIO` at any step and observe the resolved parent that was fsync'd. -// These are module-level `#[cfg(test)]` items (not a wrapping `mod`) so the file keeps exactly ONE -// inline test-BODY (`mod tests`), satisfying the test-locality structure invariant. +// ZERO indirection — no trait object, no branch. Under test it asks the test hooks +// (`src/tests/hooks.rs`, where the armed faults and the recorded fsyncs live) whether to fail this +// step. #[cfg(test)] macro_rules! fault_point { ($step:expr) => { - if let Some(err) = fault_take_if($step) { + if let Some(err) = crate::durable::hooks::fault_take_if($step) { return Err(err); } }; @@ -269,91 +305,14 @@ macro_rules! fault_point { } use fault_point; -/// The fallible steps of the primitive, in order (the fault-injection axis of the class test). #[cfg(test)] -#[derive(Clone, Copy, PartialEq, Eq, Debug)] -enum FaultStep { - Create, - Write, - Flush, - Fsync, - Rename, -} +use crate::durable::hooks::FaultStep; +/// The test side of the `#[cfg(test)]` hook points above: the armed faults, the recorded fsyncs and +/// the decoy switch. #[cfg(test)] -thread_local! { - /// `(step, raw_os_errno)` — the next `fault_point!(step)` on this thread returns that errno once. - static FAULT_INJECT: std::cell::RefCell> = - const { std::cell::RefCell::new(None) }; - /// Every parent path the primitive fsync'd on this thread, in order. A Vec, not a single slot, - /// because `create_dir_all` fsyncs one parent per directory it creates and the ancestor walk is - /// the thing under test. - static FAULT_PARENT_FSYNC: std::cell::RefCell> = - const { std::cell::RefCell::new(Vec::new()) }; - /// One-shot: plant a decoy file at the NEXT write's exact temp path on this thread. - static PLANT_DECOY: std::cell::Cell = const { std::cell::Cell::new(false) }; -} - -/// Arm a one-shot fault: the next matching step returns `io::Error::from_raw_os_error(errno)`. -#[cfg(test)] -fn fault_arm(step: FaultStep, errno: i32) { - FAULT_INJECT.with(|c| *c.borrow_mut() = Some((step, errno))); -} - -/// Clear any armed fault + recorded parent fsync (call at the start of each case). -#[cfg(test)] -fn fault_reset() { - FAULT_INJECT.with(|c| *c.borrow_mut() = None); - FAULT_PARENT_FSYNC.with(|c| c.borrow_mut().clear()); -} - -/// If a fault is armed for `step`, consume it and return the injected error. -#[cfg(test)] -fn fault_take_if(step: FaultStep) -> Option { - FAULT_INJECT.with(|c| { - let mut slot = c.borrow_mut(); - match *slot { - Some((s, errno)) if s == step => { - *slot = None; - Some(io::Error::from_raw_os_error(errno)) - } - _ => None, - } - }) -} - -/// Arm a one-shot decoy: the next write on this thread finds a file ALREADY SITTING at the exact -/// temp path it is about to create. That is the crashed-previous-run state the `exclusive` posture's -/// pre-removal exists for, and the only way to reach it deterministically -- the temp name embeds a -/// pid and an atomic counter, so a test cannot name it from outside. -#[cfg(test)] -fn plant_decoy_arm() { - PLANT_DECOY.with(|c| c.set(true)); -} - -#[cfg(test)] -fn plant_decoy_if_armed(tmp: &Path) { - if PLANT_DECOY.with(|c| c.replace(false)) { - let _ = std::fs::write(tmp, b"leftover from a crashed run"); - } -} - -#[cfg(test)] -fn fault_record_parent_fsync(parent: &Path) { - FAULT_PARENT_FSYNC.with(|c| c.borrow_mut().push(parent.to_path_buf())); -} - -/// The LAST parent fsync'd, for the single-publish assertions. -#[cfg(test)] -fn fault_parent_fsynced() -> Option { - FAULT_PARENT_FSYNC.with(|c| c.borrow().last().cloned()) -} - -/// Every parent fsync'd, in order, for the `create_dir_all` ancestor-walk assertion. -#[cfg(test)] -fn fault_parents_fsynced() -> Vec { - FAULT_PARENT_FSYNC.with(|c| c.borrow().clone()) -} +#[path = "tests/durable_hooks.rs"] +pub(crate) mod hooks; #[cfg(test)] #[path = "tests/durable_tests.rs"] diff --git a/crates/plugin-loader/src/tests/durable_hooks.rs b/crates/plugin-loader/src/tests/durable_hooks.rs new file mode 100644 index 0000000000..f0f49d1931 --- /dev/null +++ b/crates/plugin-loader/src/tests/durable_hooks.rs @@ -0,0 +1,97 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! The test side of `durable.rs`'s `#[cfg(test)]` hook points. +//! +//! `durable.rs` keeps only the calls: `fault_point!` at each fallible step of a publish, the decoy +//! plant before the temp is created, and the parent-fsync hook at the one directory fsync. What +//! those calls consult — the armed fault, the recorded fsyncs, the decoy switch — lives here, per +//! thread, so a test arms exactly the case it means and no other test sees it. + +use std::cell::{Cell, RefCell}; +use std::io; +use std::path::{Path, PathBuf}; + +/// The fallible steps of the durable primitive, in order (the fault-injection axis of the class +/// test), and the directory fsync every durable step ends with. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub(crate) enum FaultStep { + Create, + Write, + Flush, + Fsync, + Rename, + /// The fsync of the directory holding an entry that was created, renamed or removed. + DirSync, +} + +thread_local! { + /// `(step, raw_os_errno)` — the next matching step on this thread returns that errno once. + static FAULT_INJECT: RefCell> = const { RefCell::new(None) }; + /// Every parent path the primitive fsync'd on this thread, in order. A Vec, not a single slot, + /// because `create_dir_all` fsyncs one parent per directory it creates and the ancestor walk is + /// the thing under test. + static PARENT_FSYNCS: RefCell> = const { RefCell::new(Vec::new()) }; + /// One-shot: plant a decoy file at the NEXT write's exact temp path on this thread. + static PLANT_DECOY: Cell = const { Cell::new(false) }; +} + +/// Arm a one-shot fault: the next matching step returns `io::Error::from_raw_os_error(errno)`. +pub(crate) fn fault_arm(step: FaultStep, errno: i32) { + FAULT_INJECT.with(|c| *c.borrow_mut() = Some((step, errno))); +} + +/// Clear any armed fault and every recorded parent fsync (call at the start of each case). +pub(crate) fn fault_reset() { + FAULT_INJECT.with(|c| *c.borrow_mut() = None); + PARENT_FSYNCS.with(|c| c.borrow_mut().clear()); +} + +/// If a fault is armed for `step`, consume it and return the injected error. +pub(crate) fn fault_take_if(step: FaultStep) -> Option { + FAULT_INJECT.with(|c| { + let mut slot = c.borrow_mut(); + match *slot { + Some((s, errno)) if s == step => { + *slot = None; + Some(io::Error::from_raw_os_error(errno)) + } + _ => None, + } + }) +} + +/// The parent-fsync hook: record `parent` as fsynced on this thread, then fail if a +/// [`FaultStep::DirSync`] fault is armed. +pub(crate) fn parent_fsync(parent: &Path) -> io::Result<()> { + PARENT_FSYNCS.with(|c| c.borrow_mut().push(parent.to_path_buf())); + match fault_take_if(FaultStep::DirSync) { + Some(err) => Err(err), + None => Ok(()), + } +} + +/// The LAST parent fsync'd, for the single-publish assertions. +pub(crate) fn parent_fsynced() -> Option { + PARENT_FSYNCS.with(|c| c.borrow().last().cloned()) +} + +/// Every parent fsync'd, in order, for the ancestor-walk and segment-creation assertions. +pub(crate) fn parents_fsynced() -> Vec { + PARENT_FSYNCS.with(|c| c.borrow().clone()) +} + +/// Arm a one-shot decoy: the next write on this thread finds a file ALREADY SITTING at the exact +/// temp path it is about to create. That is the crashed-previous-run state the `exclusive` posture's +/// pre-removal exists for, and the only way to reach it deterministically -- the temp name embeds a +/// pid and an atomic counter, so a test cannot name it from outside. +pub(crate) fn plant_decoy_arm() { + PLANT_DECOY.with(|c| c.set(true)); +} + +/// The decoy hook: plant the decoy at `tmp` if one is armed. +pub(crate) fn plant_decoy_if_armed(tmp: &Path) { + if PLANT_DECOY.with(|c| c.replace(false)) { + let _ = std::fs::write(tmp, b"leftover from a crashed run"); + } +} diff --git a/crates/plugin-loader/src/tests/durable_tests.rs b/crates/plugin-loader/src/tests/durable_tests.rs index 71cbbad37d..5ac82a7a0f 100644 --- a/crates/plugin-loader/src/tests/durable_tests.rs +++ b/crates/plugin-loader/src/tests/durable_tests.rs @@ -3,15 +3,18 @@ //! Tests for `crates/plugin-loader/src/durable.rs`. -use super::{ - create_dir_all, fault_arm, fault_parent_fsynced, fault_parents_fsynced, fault_reset, - holding_dir, plant_decoy_arm, write, write_with, DurableOpts, FaultStep, +use crate::durable::hooks::{ + fault_arm, fault_reset, parent_fsynced as fault_parent_fsynced, + parents_fsynced as fault_parents_fsynced, plant_decoy_arm, FaultStep, }; +use crate::durable::{create_dir_all, holding_dir, write, write_with, DurableOpts}; use std::path::{Path, PathBuf}; // Linux ENOSPC=28, EIO=5; macOS shares these values. Injected via `from_raw_os_error`. const ENOSPC: i32 = 28; const EIO: i32 = 5; +/// `EINVAL`: what `fsync(2)` answers for a descriptor that does not support synchronization. +const EINVAL: i32 = 22; /// Serializes the ONE test in this file that mutates process-global CWD /// (`success_relative_path_fsyncs_dot`) against itself. MUST be module-level, not @@ -282,7 +285,7 @@ fn remove_unlinks_and_fsyncs_the_holding_dir() { // -- left `.last()` still `Some(sc.dir)` and the test still green. fault_reset(); - super::remove(&target).expect("the unlink succeeds"); + crate::durable::remove(&target).expect("the unlink succeeds"); assert!(!target.exists(), "the artifact is gone"); assert_eq!( fault_parent_fsynced().as_deref(), @@ -293,7 +296,7 @@ fn remove_unlinks_and_fsyncs_the_holding_dir() { // A missing target is a real error, not a silent success -- the caller's 404 check is a // separate concern and must not be papered over here. assert!( - super::remove(&target).is_err(), + crate::durable::remove(&target).is_err(), "removing nothing is an error" ); } @@ -417,3 +420,60 @@ fn create_dir_all_concurrent_racers_all_succeed_on_the_same_new_path() { } assert!(target.is_dir()); } + +/// **A PUBLISH WHOSE DIRECTORY FSYNC FAILS IS AN ERROR.** RED before the fix: the failure was +/// swallowed and `write` answered `Ok` over a rename a power loss could still undo. +#[test] +fn a_publish_whose_directory_fsync_fails_is_an_error() { + let scratch = Scratch::new("dirfsync-publish"); + let target = scratch.path("overlay.json"); + fault_reset(); + fault_arm(FaultStep::DirSync, EIO); + let err = write(&target, b"new contents") + .expect_err("a directory fsync that failed is not a durable publish"); + assert_eq!( + err.raw_os_error(), + Some(EIO), + "the fsync's own error surfaces" + ); + // The contract says what an error from the last step means: the rename happened, and it is + // the entry's durability that was not given. + assert_eq!(std::fs::read(&target).unwrap(), b"new contents"); +} + +/// **A REMOVAL WHOSE DIRECTORY FSYNC FAILS IS AN ERROR.** RED before the fix: `remove` answered +/// `Ok` once the unlink succeeded, whatever the fsync after it said. +#[test] +fn a_removal_whose_directory_fsync_fails_is_an_error() { + let scratch = Scratch::new("dirfsync-remove"); + let target = scratch.path("plugin.tar.gz"); + std::fs::write(&target, b"artifact").unwrap(); + fault_reset(); + fault_arm(FaultStep::DirSync, EIO); + let err = crate::durable::remove(&target).expect_err("the removal was not made durable"); + assert_eq!(err.raw_os_error(), Some(EIO)); +} + +/// **A DIRECTORY CREATED WHOSE PARENT'S FSYNC FAILS IS AN ERROR.** RED before the fix: the walk +/// fsynced each new directory's parent and ignored the answer. +#[test] +fn a_created_directory_whose_parent_fsync_fails_is_an_error() { + let scratch = Scratch::new("dirfsync-mkdir"); + fault_reset(); + fault_arm(FaultStep::DirSync, EIO); + let err = create_dir_all(&scratch.path("plugins")) + .expect_err("the new directory's entry was not made durable"); + assert_eq!(err.raw_os_error(), Some(EIO)); +} + +/// A filesystem that does not support fsyncing a directory is not refused: there is nothing more +/// it can promise, and refusing every publish on it would leave no durability at all. +#[test] +fn a_directory_fsync_the_filesystem_does_not_support_is_not_an_error() { + let scratch = Scratch::new("dirfsync-unsupported"); + let target = scratch.path("state.json"); + fault_reset(); + fault_arm(FaultStep::DirSync, EINVAL); + write(&target, b"contents").expect("an unsupported directory fsync is not a failure"); + assert_eq!(std::fs::read(&target).unwrap(), b"contents"); +} diff --git a/xtask/src/gates/structure_lint/choke_points.rs b/xtask/src/gates/structure_lint/choke_points.rs index 30d9e5c979..377185b023 100644 --- a/xtask/src/gates/structure_lint/choke_points.rs +++ b/xtask/src/gates/structure_lint/choke_points.rs @@ -180,7 +180,6 @@ pub fn table(a: &Addresses) -> Vec { "crates/plugin-loader/src/durable.rs".into(), format!("{core}/test_support/mod.rs"), format!("{core}/test_support/export_axis.rs"), - "crates/busbar-kernel-wal/src/backend.rs".into(), ], ) .core_tier(),