gunnar_core/

txlog.rs

//! **The transaction log** — every mutation of durable state on the primary,
//! in one strictly increasing sequence, appended before the state changes,
//! fsynced per commit, torn tail detected and cut on open.
//!
//! This is what the twin holds. The owner's ruling (2026-09-14): *"the twin
//! must hold each transaction sent to the master; if it has missed something
//! it must ask the master for the missing part before continuing to append."*
//! The design note is `.nornir/twin-transaction-log.md`; this file is the
//! storage half of it and knows nothing about the wire.
//!
//! # What a transaction is
//!
//! One frame: a [`Kind`], a small JSON **meta** (which repository, which ref,
//! which document — [`Meta`]), and a **blob** that is the content itself —
//! the pack bytes a client sent, the bytes of a control-store document, an
//! LFS object. The blob is written from the caller's buffer with one
//! `write_vectored`, never copied into a frame first. Nothing in the log is a
//! reference to something the twin would have to fetch elsewhere: a twin
//! holding entries `1..=n` holds everything the primary was told in that
//! window, and a fresh primary is rebuilt from the twin alone.
//!
//! # On disk
//!
//! `<data-root>/txlog/` holds segments named by their first sequence number,
//! `seg-<first>.log`, each a plain run of frames:
//!
//! ```text
//! magic   "GTX1"      4     header crc32c covers 0..28
//! seq     u64 LE      8
//! kind    u16 LE      2
//! flags   u16 LE      2     reserved, 0
//! meta_len u32 LE     4
//! blob_len u64 LE     8
//! hcrc    u32 LE      4
//! meta    meta_len
//! blob    blob_len
//! pcrc    u32 LE      4     crc32c over meta ‖ blob
//! ```
//!
//! Beside each segment, `seg-<first>.idx` is a run of `u64 LE` byte offsets,
//! one per frame, so `seq → (segment, offset)` is one seek. The index is a
//! cache: it is appended **after** the frame's fsync and is not fsynced
//! itself. On open the last index entry is checked against the data and the
//! index is rebuilt from the data if it disagrees. The data is the truth.
//!
//! # Durability and the torn tail
//!
//! `append` = write the frame, `fdatasync` the segment, then return the
//! sequence number. **Group commit** (perf twin-apply-fsync, 2026-10-06): the
//! frame is staged under the writer lock and the fdatasync is
//! [`Log::sync_through`]'s — one leader syncs every frame staged so far, outside
//! the writer lock, so concurrent appenders share it; a twin writes a whole
//! fetched group with [`Pending::commit_unsynced`] and makes it durable with ONE
//! [`Log::sync`]. Either way a frame is visible to readers, indexed and counted in
//! the head only once it is durable — the head is what a twin acknowledges.
//! A crash before the fsync returns can leave a partial
//! frame at the end of the last segment and nowhere else; `open` walks from
//! the last indexed frame, verifies every frame it finds (header crc, both
//! lengths against the file, payload crc, expected seq), and truncates the
//! file at the first bad byte. Anything after a torn frame is unreachable
//! anyway, and an append after garbage would never be read back.
//!
//! # Why not `arrow_log` / `history`
//!
//! `arrow_log` has the durability machinery (group commit, torn-tail cut) but
//! is retired, per-repository, keeps no persisted sequence and no streaming
//! reader, and its own manifest says it goes when the legacy secret log is
//! deleted. `history` derives a `push_seq` at read time per repository. The
//! bus in `gunnar-grpc` has a global `seq` but *"there is no durable log and
//! this is not one"*. A twin that must hold every transaction needs one
//! persisted, global, gap-free number, and that is this module's whole job.
//!
//! # Sequential by design
//!
//! Every write here is an append and every read is a forward scan or one
//! seek — the twin sits on an HDD-tier volume and the log is laid out for
//! it. No B-tree, no random rewrite, no compaction that copies. Retention is
//! whole-segment deletion below a floor ([`Log::truncate_before`]).

/// The tee: the decorators that make every store write a frame first.
#[path = "txlog_tee.rs"]
pub mod tee;

use std::fs::{self, File, OpenOptions};
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, Weak};

use serde::{Deserialize, Serialize};

const MAGIC: [u8; 4] = *b"GTX1";
const HEADER_LEN: usize = 32;
const TRAILER_LEN: usize = 4;
const SEG_PREFIX: &str = "seg-";
const SEG_EXT: &str = "log";
const IDX_EXT: &str = "idx";

/// Frames are appended to the current segment until it passes this size;
/// the next append starts a new one. Retention deletes whole segments, so
/// this is the granularity of "what may be truncated".
pub const DEFAULT_SEGMENT_BYTES: u64 = 1 << 30;

/// The kinds of transaction, one per mutation seam. The number is the wire
/// and disk value; **never renumber, only add**.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[repr(u16)]
pub enum Kind {
    /// A pack landed in a repository: blob = the pack bytes as the client
    /// sent them. Meta: [`Meta::Pack`].
    Pack = 1,
    /// One or more reference edits in a repository, every ref kind — heads,
    /// tags, notes, remotes, salvage, superseded, and `HEAD`. No blob.
    /// Meta: [`Meta::Refs`].
    Refs = 2,
    /// A content-addressed blob of the control store (a document shard or a
    /// document root). Blob = the bytes; meta carries the sha-256 id.
    ControlBlob = 3,
    /// A control-store document pointer moved (`commit_doc`): the catalog
    /// (repositories, visibility, tombstones), the identity table
    /// (principals, keys, grants, tokens, memberships, gates), PKI, mirrors,
    /// flow, pulls, policy, entitlements. No blob; the bytes arrived as
    /// [`Kind::ControlBlob`] entries before it.
    ControlDoc = 4,
    /// A sealed secret was written or tombstoned. Blob = the sealed bytes
    /// exactly as the secret store holds them (sealed with the box's KEK).
    Secret = 5,
    /// An LFS object landed. Blob = the object bytes. Meta: [`Meta::Lfs`].
    LfsPut = 6,
    /// An LFS object was removed. No blob.
    LfsDelete = 7,
    /// A repository's object store was created or purged on disk (the
    /// catalog row is a `ControlDoc`; this is the store beside it).
    RepoStore = 8,
    /// **The box's key-encrypting key** — the first frame of a primary's log,
    /// so a restore into a fresh box can read every `Secret` frame after it.
    /// Blob = the 32 KEK bytes exactly as `secrets.kek` holds them. The twin
    /// holds it: a twin that held every sealed secret and not the key would be
    /// a backup of nothing.
    Kek = 9,
    /// One line appended to an append-only signed log under the data root
    /// (attestations, transparency heads, forge events, build records,
    /// membership). Blob = the line without its newline. Meta: [`Meta::Append`].
    Append = 10,
    /// A whole document under the data root replaced or removed
    /// (`replication.json`, `webhooks.json`). Blob = the bytes. Meta: [`Meta::File`].
    File = 11,
    /// **Never in a log**: the terminator of a twin SNAPSHOT stream
    /// (`Mirror.SnapshotFetch`, `Bootstrap.Restore`). Every item before it is an
    /// entry of the kinds above with `seq` 0; this one says at which sequence the
    /// state they carry was taken. Meta: [`Meta::SnapshotAt`].
    SnapshotAt = 12,
    /// **A frame whose content was ERASED in place** ([`Log::erase`]): it was a
    /// `Pack`, `Refs`, `LfsPut` or `LfsDelete` of a repository whose account was
    /// deleted. The meta is the original one (it still names the repository);
    /// the blob reads as empty and its bytes are gone from the file. Never
    /// appended — only ever written over a frame that is already there, so the
    /// sequence, the offsets and the twin's gap rule stay exactly as they were.
    /// Applying one does nothing: the data it carried is what was deleted.
    Redacted = 13,
    /// **An account's data was deleted** (owner ruling g-nonpay, 2026-09-29):
    /// every frame BEFORE this one of the repositories it names is erased
    /// ([`Log::erase`]) — on the primary when it appends it, on the twin when it
    /// stores it, on a box being restored when it replays it. No blob.
    /// Meta: [`Meta::Erase`].
    Erase = 14,
    /// **Last-seen stamps** (owner, 2026-10-21): the principals whose last
    /// sign-in or last operation moved since the previous flush, ONE frame per
    /// flush ([`crate::activity`]). Blob = JSON, principal id → stamps. Applied
    /// on the twin into the same control-database table; a snapshot carries
    /// the whole table as one item of this kind. Meta: [`Meta::Activity`].
    Activity = 15,
}

impl Kind {
    /// The disk value back to a kind. `None` for a value this build does
    /// not know — a newer primary's frame; the twin still stores it.
    pub fn from_u16(v: u16) -> Option<Self> {
        Some(match v {
            1 => Kind::Pack,
            2 => Kind::Refs,
            3 => Kind::ControlBlob,
            4 => Kind::ControlDoc,
            5 => Kind::Secret,
            6 => Kind::LfsPut,
            7 => Kind::LfsDelete,
            8 => Kind::RepoStore,
            9 => Kind::Kek,
            10 => Kind::Append,
            11 => Kind::File,
            12 => Kind::SnapshotAt,
            13 => Kind::Redacted,
            14 => Kind::Erase,
            15 => Kind::Activity,
            _ => return None,
        })
    }
}

/// One reference edit as the log records it: name, the value expected before
/// (`None` = must not exist, `Some("")` = no claim), the value after
/// (`None` = delete). Oids are hex; a symbolic target is `ref:<name>`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RefEdit {
    /// Full ref name, or `HEAD`.
    pub name: String,
    /// `None` = no claim about the previous value (a plain `put_refs`);
    /// `Some(None)` = must not have existed; `Some(Some(oid))` = must have
    /// been `oid`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub old: Option<Option<String>>,
    /// `None` deletes.
    pub new: Option<String>,
    /// Peeled target of an annotated tag, hex.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub peeled: Option<String>,
    /// Symbolic target, full ref name, when `new` is a symref.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub symref: Option<String>,
}

/// The typed meta of a frame. Serialized as JSON; every variant is additive.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "k", rename_all = "snake_case")]
pub enum Meta {
    /// [`Kind::Pack`].
    Pack {
        /// Repository key, `namespace/name`.
        repo: String,
        /// Hash kind of the repository, `sha1` or `sha256`.
        hash: String,
    },
    /// [`Kind::Refs`].
    Refs {
        /// Repository key.
        repo: String,
        /// Whether the batch is all-or-nothing (`put_refs_cas`).
        atomic: bool,
        /// The edits, in order.
        edits: Vec<RefEdit>,
    },
    /// [`Kind::ControlBlob`].
    ControlBlob {
        /// sha-256 of the blob, hex.
        id: String,
    },
    /// [`Kind::ControlDoc`].
    ControlDoc {
        /// Document name, e.g. `refs/gunnar/auth`.
        name: String,
        /// Blob id the pointer held before, hex, `None` = absent.
        expected: Option<String>,
        /// Blob id it holds now, hex.
        new: String,
    },
    /// [`Kind::Secret`].
    Secret {
        /// Secret key.
        key: String,
        /// `true` = tombstone (the blob is empty).
        deleted: bool,
    },
    /// [`Kind::LfsPut`] and [`Kind::LfsDelete`].
    Lfs {
        /// Repository key.
        repo: String,
        /// Object id, hex.
        oid: String,
        /// Size in bytes.
        size: u64,
    },
    /// [`Kind::RepoStore`].
    RepoStore {
        /// Repository key.
        repo: String,
        /// `created` or `purged`.
        op: String,
        /// Hash kind.
        hash: String,
    },
    /// [`Kind::Kek`]. No fields: the blob is the key.
    Kek {},
    /// [`Kind::Append`].
    Append {
        /// The log's path relative to the data root, e.g. `attestations.jsonl`.
        log: String,
    },
    /// [`Kind::File`].
    File {
        /// The document's path relative to the data root.
        path: String,
        /// `true` = removed (the blob is empty).
        deleted: bool,
    },
    /// [`Kind::Activity`]. No fields: the blob is the batch.
    Activity {},
    /// [`Kind::SnapshotAt`].
    SnapshotAt {
        /// The state the snapshot's items carry is the log's state through this sequence.
        at: u64,
    },
    /// [`Kind::Erase`].
    Erase {
        /// The account whose data went.
        account: String,
        /// The repository keys (`namespace/name`) whose frames before this one are erased —
        /// exactly the stores the purge removed, never a prefix of them.
        repos: Vec<String>,
        /// Why, for the audit trail.
        reason: String,
    },
}

impl Meta {
    /// The repository a data frame's meta names — `Pack`, `Refs`, `Lfs`, `RepoStore`.
    pub fn repo(&self) -> Option<&str> {
        match self {
            Meta::Pack { repo, .. }
            | Meta::Refs { repo, .. }
            | Meta::Lfs { repo, .. }
            | Meta::RepoStore { repo, .. } => Some(repo),
            _ => None,
        }
    }
}

/// What [`Log::erase`] did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Erased {
    /// Frames rewritten as [`Kind::Redacted`] by this call.
    pub frames: u64,
    /// Blob bytes gone from the files.
    pub bytes: u64,
}

impl Meta {
    /// Serialize to the bytes the frame carries.
    pub fn to_bytes(&self) -> Vec<u8> {
        serde_json::to_vec(self).expect("Meta serializes")
    }

    /// Parse a frame's meta.
    pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
        serde_json::from_slice(bytes).map_err(|e| Error::BadMeta(e.to_string()))
    }
}

/// A frame read back.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Entry {
    /// Its sequence number.
    pub seq: u64,
    /// Kind as stored (a twin keeps kinds it does not know).
    pub kind: u16,
    /// Meta bytes (JSON).
    pub meta: Vec<u8>,
    /// Blob bytes.
    pub blob: Vec<u8>,
}

impl Entry {
    /// The typed kind, if this build knows it.
    pub fn kind(&self) -> Option<Kind> {
        Kind::from_u16(self.kind)
    }
    /// The typed meta.
    pub fn meta(&self) -> Result<Meta> {
        Meta::from_bytes(&self.meta)
    }
}

/// Where a frame's blob lies, for a reader that streams it in chunks rather
/// than loading it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Located {
    /// Sequence number.
    pub seq: u64,
    /// Kind as stored.
    pub kind: u16,
    /// Meta bytes.
    pub meta: Vec<u8>,
    /// The segment file the blob is in.
    pub path: PathBuf,
    /// Byte offset of the blob inside that file.
    pub blob_offset: u64,
    /// Blob length.
    pub blob_len: u64,
    /// **The frame's payload crc32c (meta ‖ blob)** — `Some` only on the LAST piece (restore-large-
    /// entry, 2026-10-07): what a sender stamps on the entry's last chunk so a box that put the
    /// entry together across several streams applies it only if it IS the entry. From
    /// [`Log::read_pieces`] it is CHECKED: the crc over the bytes read matched the trailer. From
    /// [`Log::read_pieces_from`] at an offset it is the trailer AS STORED, unchecked here — the
    /// bytes below the offset were not read — and the receiver, which holds the whole entry,
    /// checks it. `None` from [`Log::locate_blob`] and on an erased frame (its crc was over bytes
    /// now gone).
    pub payload_crc: Option<u32>,
}

/// Errors of the log.
#[derive(Debug, thiserror::Error)]
pub enum Error {
    /// I/O.
    #[error("txlog: {0}")]
    Io(#[from] io::Error),
    /// A frame's meta did not parse.
    #[error("txlog: bad meta: {0}")]
    BadMeta(String),
    /// The requested sequence is below what the log retains.
    #[error("txlog: seq {seq} not retained (first retained is {first})")]
    NotRetained {
        /// What was asked for.
        seq: u64,
        /// The oldest sequence still on disk.
        first: u64,
    },
    /// The requested sequence is above the head.
    #[error("txlog: seq {seq} is beyond head {head}")]
    BeyondHead {
        /// What was asked for.
        seq: u64,
        /// The newest committed sequence.
        head: u64,
    },
    /// A frame in the middle of a segment (not the tail) is corrupt.
    #[error("txlog: corrupt frame at {path}:{offset}: {what}")]
    Corrupt {
        /// Segment.
        path: PathBuf,
        /// Offset of the frame.
        offset: u64,
        /// What was wrong.
        what: &'static str,
    },
    /// The log's files are RELEASED ([`Log::release`]): the data set under it is being grown
    /// and unmounted, so nothing may write until [`Log::reacquire`].
    #[error("txlog: released — the files under {0} are given up while the data set is grown; writes resume after the remount")]
    Released(PathBuf),
    /// An append was asked to start at a sequence that is not `head + 1`.
    #[error("txlog: append out of order: got {got}, expected {expected}")]
    OutOfOrder {
        /// The sequence offered.
        got: u64,
        /// The one the log would accept.
        expected: u64,
    },
    /// This log belongs to a TWIN: it holds the primary's numbers and
    /// originates none. A local mutation on a twin is refused here, loudly,
    /// rather than stealing a sequence number from the primary.
    #[error("txlog: this box is a twin (or restoring); it originates no transactions")]
    Follower,
    /// The segment directory holds a file the layout does not describe.
    #[error("txlog: unexpected file {0}")]
    Unexpected(PathBuf),
    /// **The log is POISONED** (stinky rows 3 and 6, 2026-10-10): a frame's `fdatasync` failed,
    /// or the cut of a half-written frame did. After a failed fsync the kernel may have dropped
    /// the dirty pages and the NEXT fsync succeeds without them (fsyncgate) — a frame its writer
    /// was told had failed would then be published by the next leader and applied by the twin.
    /// After a failed cut the next frame would land behind garbage. So nothing more is written:
    /// every append refuses with this until the log is reopened, which recovers from the last
    /// durable frame.
    #[error("txlog: poisoned under {dir} — {cause}; every write refuses until the log is reopened from its last durable frame")]
    Poisoned {
        /// The log's directory.
        dir: PathBuf,
        /// What failed.
        cause: String,
    },
}

/// Result alias.
pub type Result<T> = std::result::Result<T, Error>;

// ── crc32c (Castagnoli), table-driven, no dependency ─────────────────────────

const CRC32C_POLY: u32 = 0x82F6_3B78;

const fn crc32c_table() -> [u32; 256] {
    let mut table = [0u32; 256];
    let mut i = 0;
    while i < 256 {
        let mut c = i as u32;
        let mut k = 0;
        while k < 8 {
            c = if c & 1 != 0 {
                CRC32C_POLY ^ (c >> 1)
            } else {
                c >> 1
            };
            k += 1;
        }
        table[i] = c;
        i += 1;
    }
    table
}

static CRC32C_TABLE: [u32; 256] = crc32c_table();

/// Streaming crc32c.
#[derive(Debug, Clone, Copy)]
pub struct Crc32c(u32);

impl Default for Crc32c {
    fn default() -> Self {
        Self::new()
    }
}

impl Crc32c {
    /// Fresh.
    pub const fn new() -> Self {
        Crc32c(0xFFFF_FFFF)
    }
    /// Feed bytes.
    pub fn update(&mut self, bytes: &[u8]) {
        let mut c = self.0;
        for &b in bytes {
            c = CRC32C_TABLE[((c ^ b as u32) & 0xFF) as usize] ^ (c >> 8);
        }
        self.0 = c;
    }
    /// The checksum.
    pub fn finish(self) -> u32 {
        self.0 ^ 0xFFFF_FFFF
    }
    /// The running state, to be kept with the bytes it covers and [`Crc32c::resume`]d — a
    /// checksum carried across a restart (restore-after-snapshot: a snapshot item staged over
    /// several streams).
    pub fn state(&self) -> u32 {
        self.0
    }
    /// Go on from a kept [`Crc32c::state`].
    pub const fn resume(state: u32) -> Self {
        Crc32c(state)
    }
}

/// crc32c of one buffer.
pub fn crc32c(bytes: &[u8]) -> u32 {
    let mut c = Crc32c::new();
    c.update(bytes);
    c.finish()
}

// ── frame header ─────────────────────────────────────────────────────────────

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Header {
    seq: u64,
    kind: u16,
    meta_len: u32,
    blob_len: u64,
}

impl Header {
    fn encode(&self) -> [u8; HEADER_LEN] {
        let mut h = [0u8; HEADER_LEN];
        h[0..4].copy_from_slice(&MAGIC);
        h[4..12].copy_from_slice(&self.seq.to_le_bytes());
        h[12..14].copy_from_slice(&self.kind.to_le_bytes());
        h[14..16].copy_from_slice(&0u16.to_le_bytes());
        h[16..20].copy_from_slice(&self.meta_len.to_le_bytes());
        h[20..28].copy_from_slice(&self.blob_len.to_le_bytes());
        let crc = crc32c(&h[0..28]);
        h[28..32].copy_from_slice(&crc.to_le_bytes());
        h
    }

    fn decode(h: &[u8]) -> Option<Self> {
        if h.len() < HEADER_LEN || h[0..4] != MAGIC {
            return None;
        }
        let crc = u32::from_le_bytes(h[28..32].try_into().ok()?);
        if crc32c(&h[0..28]) != crc {
            return None;
        }
        Some(Header {
            seq: u64::from_le_bytes(h[4..12].try_into().ok()?),
            kind: u16::from_le_bytes(h[12..14].try_into().ok()?),
            meta_len: u32::from_le_bytes(h[16..20].try_into().ok()?),
            blob_len: u64::from_le_bytes(h[20..28].try_into().ok()?),
        })
    }

    fn frame_len(&self) -> u64 {
        HEADER_LEN as u64 + self.meta_len as u64 + self.blob_len + TRAILER_LEN as u64
    }
}

// ── segments ─────────────────────────────────────────────────────────────────

fn seg_path(dir: &Path, first: u64) -> PathBuf {
    dir.join(format!("{SEG_PREFIX}{first:020}.{SEG_EXT}"))
}

fn idx_path(dir: &Path, first: u64) -> PathBuf {
    dir.join(format!("{SEG_PREFIX}{first:020}.{IDX_EXT}"))
}

fn parse_seg_name(name: &str) -> Option<(u64, bool)> {
    let rest = name.strip_prefix(SEG_PREFIX)?;
    let (num, ext) = rest.split_once('.')?;
    let first: u64 = num.parse().ok()?;
    match ext {
        SEG_EXT => Some((first, true)),
        IDX_EXT => Some((first, false)),
        _ => None,
    }
}

fn sync_dir(dir: &Path) -> io::Result<()> {
    File::open(dir)?.sync_all()
}

/// **Make `len` bytes at `at` gone**: a punched hole where the filesystem can (the
/// space is freed and the range reads as zeros), zeros written over it where it cannot.
fn zero_range(file: &mut File, at: u64, len: u64) -> io::Result<()> {
    if len == 0 {
        return Ok(());
    }
    #[cfg(target_os = "linux")]
    {
        use rustix::fs::{fallocate, FallocateFlags};
        if fallocate(
            &*file,
            FallocateFlags::PUNCH_HOLE | FallocateFlags::KEEP_SIZE,
            at,
            len,
        )
        .is_ok()
        {
            return Ok(());
        }
    }
    let zeros = vec![0u8; (1usize << 20).min(len as usize)];
    file.seek(SeekFrom::Start(at))?;
    let mut left = len;
    while left > 0 {
        let n = left.min(zeros.len() as u64) as usize;
        file.write_all(&zeros[..n])?;
        left -= n as u64;
    }
    Ok(())
}

/// One segment as the log knows it: its first sequence, its frame offsets.
#[derive(Debug)]
struct Segment {
    first: u64,
    /// Byte offset of every frame, in order. `offsets.len()` frames.
    offsets: Vec<u64>,
    /// Byte length of the valid prefix.
    len: u64,
}

impl Segment {
    fn last_seq(&self) -> Option<u64> {
        if self.offsets.is_empty() {
            None
        } else {
            Some(self.first + self.offsets.len() as u64 - 1)
        }
    }
}

/// Verify frames from `from` to the end of `file`, appending offsets.
/// Returns the length of the valid prefix and whether the tail was torn.
fn scan_frames(
    path: &Path,
    file: &mut File,
    first: u64,
    offsets: &mut Vec<u64>,
    from: u64,
) -> Result<(u64, bool)> {
    let total = file.metadata()?.len();
    let mut pos = from;
    let mut hdr = [0u8; HEADER_LEN];
    let mut buf = Vec::new();
    while pos < total {
        if total - pos < HEADER_LEN as u64 {
            return Ok((pos, true));
        }
        file.seek(SeekFrom::Start(pos))?;
        file.read_exact(&mut hdr)?;
        let Some(h) = Header::decode(&hdr) else {
            return Ok((pos, true));
        };
        let expect = first + offsets.len() as u64;
        if h.seq != expect {
            return Ok((pos, true));
        }
        let payload = h.meta_len as u64 + h.blob_len;
        if total - pos < h.frame_len() {
            return Ok((pos, true));
        }
        // An ERASED frame's payload no longer matches its checksum — by design,
        // its bytes are gone ([`Log::erase`]). Its header crc vouched for it.
        if h.kind == Kind::Redacted as u16 {
            offsets.push(pos);
            pos += h.frame_len();
            continue;
        }
        // Verify the payload crc with a bounded buffer: the blob may be large.
        let mut crc = Crc32c::new();
        let mut remaining = payload;
        buf.resize(1 << 20, 0);
        while remaining > 0 {
            let n = remaining.min(buf.len() as u64) as usize;
            file.read_exact(&mut buf[..n])?;
            crc.update(&buf[..n]);
            remaining -= n as u64;
        }
        let mut tr = [0u8; TRAILER_LEN];
        file.read_exact(&mut tr)?;
        if u32::from_le_bytes(tr) != crc.finish() {
            return Ok((pos, true));
        }
        let _ = path;
        offsets.push(pos);
        pos += h.frame_len();
    }
    Ok((pos, false))
}

fn open_segment(dir: &Path, first: u64, repair: bool) -> Result<Segment> {
    let path = seg_path(dir, first);
    let mut file = OpenOptions::new().read(true).write(repair).open(&path)?;
    let total = file.metadata()?.len();

    // Trust the index up to the last entry that verifies against the data.
    let mut offsets: Vec<u64> = Vec::new();
    let ipath = idx_path(dir, first);
    if let Ok(bytes) = fs::read(&ipath) {
        let n = bytes.len() / 8;
        offsets.reserve(n);
        for i in 0..n {
            let off = u64::from_le_bytes(bytes[i * 8..i * 8 + 8].try_into().expect("8 bytes"));
            offsets.push(off);
        }
        // Offsets must be strictly increasing and inside the file.
        let monotone = offsets.windows(2).all(|w| w[0] < w[1]);
        if !monotone || offsets.first().is_some_and(|&o| o != 0) {
            offsets.clear();
        }
        while let Some(&last) = offsets.last() {
            if last + HEADER_LEN as u64 > total {
                offsets.pop();
                continue;
            }
            file.seek(SeekFrom::Start(last))?;
            let mut hdr = [0u8; HEADER_LEN];
            file.read_exact(&mut hdr)?;
            match Header::decode(&hdr) {
                Some(h)
                    if h.seq == first + offsets.len() as u64 - 1
                        && last + h.frame_len() <= total =>
                {
                    break;
                }
                _ => {
                    offsets.pop();
                }
            }
        }
    }

    // Resume AT the last indexed frame, not after it: the index vouched for
    // its header only, and the torn tail is exactly that frame — a crash
    // between the payload write and the fsync leaves a header that decodes
    // over a payload that does not. `scan_frames` re-verifies it in full and
    // puts it back when it is whole.
    let trusted = offsets.len();
    let resume = offsets.pop().unwrap_or(0);
    let (valid_len, torn) = scan_frames(&path, &mut file, first, &mut offsets, resume)?;

    if torn && repair {
        tracing::warn!(
            segment = %path.display(),
            kept = valid_len,
            discarded = total - valid_len,
            "txlog: torn tail cut"
        );
        file.set_len(valid_len)?;
        file.sync_all()?;
    }
    if (torn && repair) || offsets.len() != trusted {
        // Rewrite the index to exactly the verified frames.
        if repair {
            let mut bytes = Vec::with_capacity(offsets.len() * 8);
            for off in &offsets {
                bytes.extend_from_slice(&off.to_le_bytes());
            }
            fs::write(&ipath, bytes)?;
        }
    }
    Ok(Segment {
        first,
        offsets,
        len: valid_len,
    })
}

// ── the log ──────────────────────────────────────────────────────────────────

struct Writer {
    /// The open segment's data file, positioned at its end.
    file: File,
    /// Its index file.
    idx: File,
    /// The open segment as the WRITER knows it: every frame committed into it, including the
    /// ones of a group commit not yet made durable ([`Pending::commit_unsynced`]). Readers go
    /// by [`Log::segments`] and [`Log::head_seq`], which hold only what is durable.
    seg: Segment,
    /// Bytes of the open segment already handed to the kernel; `buf` follows them.
    disk_len: u64,
    /// **Frame bytes staged in memory** (perf twin-apply-fsync, 2026-10-06): a header, its meta,
    /// small pieces and the trailer, of one frame or of a whole group, go to the kernel as ONE
    /// write — the twin used to spend four writes on every frame. Flushed before every fsync,
    /// roll, release and cut, and whenever it passes [`STAGE_BYTES`].
    buf: Vec<u8>,
    /// How many of `seg.offsets` are durable and published (in [`Log::segments`], indexed, at or
    /// below the head). The rest are a group commit in progress.
    published: usize,
    /// Test-only: injected write and truncate failures.
    #[cfg(test)]
    faults: WriterFaults,
}

/// Test-only failures a [`Writer`] can be told to meet.
#[cfg(test)]
#[derive(Default)]
struct WriterFaults {
    /// Fail the put after this many more succeed.
    fail_put_after: Option<usize>,
    /// Fail the next [`Writer::cut_to_committed`] before it touches anything.
    fail_cut: bool,
}

/// The most frame bytes [`Writer::buf`] stages before it writes them out.
const STAGE_BYTES: usize = 1 << 20;
/// A piece at least this long is written straight from the caller's buffer, never copied into
/// the stage (a 1 MiB piece of a pack has nothing to gain from being coalesced).
const DIRECT_BYTES: usize = 64 << 10;

impl Writer {
    /// Hand every staged byte to the kernel.
    fn flush_buf(&mut self) -> io::Result<()> {
        if !self.buf.is_empty() {
            self.file.write_all(&self.buf)?;
            self.disk_len += self.buf.len() as u64;
            self.buf.clear();
        }
        Ok(())
    }

    /// Stage `bytes` at the end of the segment (written straight out when large).
    fn put(&mut self, bytes: &[u8]) -> io::Result<()> {
        #[cfg(test)]
        if let Some(n) = self.faults.fail_put_after {
            if n == 0 {
                self.faults.fail_put_after = None;
                return Err(io::Error::other("txlog test: injected write failure"));
            }
            self.faults.fail_put_after = Some(n - 1);
        }
        if bytes.len() >= DIRECT_BYTES {
            self.flush_buf()?;
            self.file.write_all(bytes)?;
            self.disk_len += bytes.len() as u64;
            return Ok(());
        }
        self.buf.extend_from_slice(bytes);
        if self.buf.len() >= STAGE_BYTES {
            self.flush_buf()?;
        }
        Ok(())
    }

    /// Frames committed into the open segment and not yet durable.
    fn unpublished(&self) -> usize {
        self.seg.offsets.len() - self.published
    }

    /// Drop a partial frame: everything after the last committed frame goes, staged or on disk.
    fn cut_to_committed(&mut self) -> io::Result<()> {
        #[cfg(test)]
        if std::mem::take(&mut self.faults.fail_cut) {
            return Err(io::Error::other("txlog test: injected truncate failure"));
        }
        let keep = self.seg.len;
        if keep >= self.disk_len {
            // Everything past `disk_len` is staged: keep the committed part of the stage.
            self.buf.truncate((keep - self.disk_len) as usize);
            return Ok(());
        }
        // The partial frame reached the file (a large piece goes straight out); nothing is
        // staged after it, so the stage is all partial frame.
        self.buf.clear();
        self.file.set_len(keep)?;
        self.file.seek(SeekFrom::Start(keep))?;
        self.disk_len = keep;
        self.file.sync_data()
    }
}

/// **The test hold of a group commit, by environment** (round 5.2, 2026-10-07): set to `n`, a
/// leader of [`Log::sync_through`] waits until `n` frames are staged before its fsync (at most
/// [`TEST_GROUP_WAIT`]). A test that counts the fsyncs of `n` concurrent writers (gunnar-server
/// `tests/audit_read_io.rs`, whose writers run in a child under `strace`) sets it, so the count is
/// of what the group commit shares, not of scheduling luck. Unset on every box: no wait at all.
pub const TEST_GROUP_ENV: &str = "GUNNAR_TXLOG_TEST_GROUP";

/// **The test hold, read only by a build with `test-hooks`** (wiring tracer B 2026-10-08): a
/// shipped box never reads [`TEST_GROUP_ENV`], so no stray variable holds its writes.
#[cfg(feature = "test-hooks")]
fn test_group_from_env() -> Option<usize> {
    std::env::var(TEST_GROUP_ENV)
        .ok()
        .and_then(|v| v.trim().parse::<usize>().ok())
        .filter(|n| *n > 1)
}

#[cfg(not(feature = "test-hooks"))]
fn test_group_from_env() -> Option<usize> {
    None
}

/// The longest a [`TEST_GROUP_ENV`] hold waits for its group: the last writers of a test finish
/// one by one, and a group that will never fill goes on without them.
pub const TEST_GROUP_WAIT: std::time::Duration = std::time::Duration::from_millis(100);

/// The transaction log: one writer, any number of readers, one directory.
///
/// `append` is O(1) in the size of the log: a positioned write at the known
/// end of the open segment, one fsync, one 8-byte index append.
pub struct Log {
    dir: PathBuf,
    /// Closed segments, oldest first, plus the open one at the end.
    segments: Mutex<Vec<Segment>>,
    writer: Mutex<Writer>,
    head: AtomicU64,
    first: AtomicU64,
    segment_bytes: u64,
    /// Only `append_at` / `begin(Some(_))` may write (a twin, a restore).
    follower: AtomicBool,
    /// The mutation gate ([`Log::mutation`] / [`Log::freeze`]).
    gate: Arc<Gate>,
    /// [`Log::release`] gave up the writer's files; every write refuses until [`Log::reacquire`].
    released: AtomicBool,
    /// [`Error::Poisoned`]: what failed. Set once, never cleared — only a reopen recovers.
    poisoned: Mutex<Option<String>>,
    /// **Is a group-commit leader in its fsync?** ([`Log::sync_through`]). A flag under a mutex
    /// that is NEVER held across I/O, and a condvar the leader wakes everybody on when its fsync
    /// is published (r5 stall review, 2026-10-07). It used to be a `Mutex<()>` HELD ACROSS the
    /// `fdatasync`: a waiter whose frame the leader's fsync had already made durable still had to
    /// ACQUIRE that mutex before it could see so — and an unfair mutex hands it to the next
    /// leader first, so the waiter sat behind an unrelated fsync, then the next, for as long as
    /// writes kept arriving, every request thread in `futex_wait`.
    sync_state: Mutex<bool>,
    /// Notified (all) whenever a leader ends.
    synced: std::sync::Condvar,
    /// Test-only: each leader takes the next gate, if any, and waits on it before its fsync.
    #[cfg(test)]
    fsync_gates: Mutex<std::collections::VecDeque<std::sync::mpsc::Receiver<()>>>,
    /// Test-only: this many of the next frame fsyncs fail.
    #[cfg(test)]
    fail_fsyncs: std::sync::atomic::AtomicU32,
    /// **The test hold of a group commit** ([`TEST_GROUP_ENV`]): a leader waits, at most
    /// [`TEST_GROUP_WAIT`], until this many frames are staged before its fsync. `None` — every
    /// box, every ordinary test — does not wait at all.
    test_group: Option<usize>,
    /// `fdatasync`s that made frames durable ([`Log::frame_syncs`]).
    frame_syncs: AtomicU64,
    /// fsyncs an erase made ([`Log::erase_syncs`]).
    erase_syncs: AtomicU64,
    /// One erase at a time ([`Log::erase`]); the writer lock is NOT held for it.
    erase_lock: Mutex<()>,
    /// Test-only: an erase takes the next gate, if any, and waits on it after its first header.
    #[cfg(test)]
    erase_gates: Mutex<std::collections::VecDeque<std::sync::mpsc::Receiver<()>>>,
}

/// **The mutation gate**: every framed mutation holds a [`MutationGuard`] from before its frame
/// is appended until the store has taken it; [`Log::freeze`] stops new ones and waits for the
/// ones in flight. While frozen, the log's head IS the stores' state — what a consistent
/// snapshot needs (the log is written BEFORE the store, so without the wait a frame at the head
/// could still be on its way into the store).
#[derive(Default)]
struct Gate {
    state: Mutex<GateState>,
    cv: std::sync::Condvar,
}

#[derive(Default)]
struct GateState {
    frozen_by: Option<std::thread::ThreadId>,
    inflight: u64,
}

thread_local! {
    /// Nested guards on one thread count once (a tee write inside another framed write).
    static MUTATION_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
}

/// Held by a framed mutation, frame through store write. See [`Log::mutation`].
#[must_use = "a mutation guard must be held until the store has taken the write"]
pub struct MutationGuard {
    gate: Option<Arc<Gate>>,
}

impl Drop for MutationGuard {
    fn drop(&mut self) {
        MUTATION_DEPTH.with(|d| d.set(d.get().saturating_sub(1)));
        if let Some(gate) = self.gate.take() {
            let mut st = gate.state.lock().unwrap_or_else(|p| p.into_inner());
            st.inflight = st.inflight.saturating_sub(1);
            gate.cv.notify_all();
        }
    }
}

/// Held while the log is frozen. See [`Log::freeze`].
pub struct FreezeGuard {
    gate: Arc<Gate>,
}

impl Drop for FreezeGuard {
    fn drop(&mut self) {
        let mut st = self.gate.state.lock().unwrap_or_else(|p| p.into_inner());
        st.frozen_by = None;
        self.gate.cv.notify_all();
    }
}

impl std::fmt::Debug for Log {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Log")
            .field("dir", &self.dir)
            .field("first", &self.first_seq())
            .field("head", &self.head_seq())
            .finish()
    }
}

/// **The one entry in the log's directory that is not a segment.**
///
/// `gunnar-server`'s `txlog_lfs` spools an arriving LFS object beside the log
/// — `<txlog>/spool/<oid>.part` — because the object has to be on disk before
/// the frame that names it is appended. It removes the FILE when the object
/// lands; the DIRECTORY stays, because `create_dir_all` made it and nothing
/// unmakes it.
///
/// It is named HERE and not there for the reason the name is load-bearing:
/// this module describes what may live in that directory, and [`Log::open_with`]
/// refuses everything else. A writer choosing a name the reader has to guess is
/// how one LFS push plus one restart became
/// `Kernel panic - not syncing: Attempted to kill init!` on the live appliance,
/// MEASURED 2026-09-17.
pub const SPOOL_DIR: &str = "spool";

impl Log {
    /// Open or create the log at `dir`, cutting a torn tail if there is one.
    pub fn open(dir: impl AsRef<Path>) -> Result<Self> {
        Self::open_with(dir, DEFAULT_SEGMENT_BYTES)
    }

    /// [`Log::open`] with a segment size (tests roll segments small).
    pub fn open_with(dir: impl AsRef<Path>, segment_bytes: u64) -> Result<Self> {
        let dir = dir.as_ref().to_path_buf();
        fs::create_dir_all(&dir)?;
        let mut firsts: Vec<u64> = Vec::new();
        for entry in fs::read_dir(&dir)? {
            let entry = entry?;
            let name = entry.file_name();
            let name = name.to_string_lossy();
            match parse_seg_name(&name) {
                Some((first, true)) => firsts.push(first),
                Some((_, false)) => {}
                // The LFS spool is part of this directory's layout — see
                // [`SPOOL_DIR`]. Only that exact name, and only as a directory:
                // a stray FILE called `spool` is still something nobody wrote
                // on purpose, and a reader that shrugs at one shrugs at
                // corruption.
                None if name == SPOOL_DIR && entry.file_type().is_ok_and(|t| t.is_dir()) => {}
                None => return Err(Error::Unexpected(entry.path())),
            }
        }
        firsts.sort_unstable();
        let mut segments = Vec::with_capacity(firsts.len().max(1));
        let n = firsts.len();
        for (i, first) in firsts.iter().enumerate() {
            let seg = open_segment(&dir, *first, true)?;
            // A closed segment must be intact end to end; only the last may
            // have been torn.
            if i + 1 < n && seg.offsets.is_empty() {
                return Err(Error::Corrupt {
                    path: seg_path(&dir, *first),
                    offset: 0,
                    what: "closed segment holds no frame",
                });
            }
            // Adjacent segments must be contiguous.
            if let Some(next) = firsts.get(i + 1) {
                match seg.last_seq() {
                    Some(last) if last + 1 == *next => {}
                    _ => {
                        return Err(Error::Corrupt {
                            path: seg_path(&dir, *first),
                            offset: seg.len,
                            what: "segment does not run up to the next one",
                        })
                    }
                }
            }
            segments.push(seg);
        }
        if segments.is_empty() {
            let first = 1;
            File::create(seg_path(&dir, first))?;
            File::create(idx_path(&dir, first))?;
            sync_dir(&dir)?;
            segments.push(Segment {
                first,
                offsets: Vec::new(),
                len: 0,
            });
        }
        let open = segments.last().expect("one segment");
        let head = open.last_seq().unwrap_or(open.first - 1);
        let first = segments[0].first;
        let mut file = OpenOptions::new()
            .append(true)
            .open(seg_path(&dir, open.first))?;
        file.seek(SeekFrom::Start(open.len))?;
        let idx = OpenOptions::new()
            .append(true)
            .open(idx_path(&dir, open.first))?;
        let seg = Segment {
            first: open.first,
            offsets: open.offsets.clone(),
            len: open.len,
        };
        let published = seg.offsets.len();
        Ok(Log {
            dir,
            segments: Mutex::new(segments),
            writer: Mutex::new(Writer {
                file,
                idx,
                disk_len: seg.len,
                seg,
                buf: Vec::new(),
                published,
                #[cfg(test)]
                faults: WriterFaults::default(),
            }),
            head: AtomicU64::new(head),
            first: AtomicU64::new(first),
            segment_bytes,
            follower: AtomicBool::new(false),
            gate: Arc::new(Gate::default()),
            released: AtomicBool::new(false),
            poisoned: Mutex::new(None),
            sync_state: Mutex::new(false),
            synced: std::sync::Condvar::new(),
            #[cfg(test)]
            fsync_gates: Mutex::new(Default::default()),
            #[cfg(test)]
            fail_fsyncs: std::sync::atomic::AtomicU32::new(0),
            test_group: test_group_from_env(),
            frame_syncs: AtomicU64::new(0),
            erase_syncs: AtomicU64::new(0),
            erase_lock: Mutex::new(()),
            #[cfg(test)]
            erase_gates: Mutex::new(Default::default()),
        })
    }

    /// **Enter a framed mutation.** Hold the guard from before the frame is appended until the
    /// store has the write. Waits while another thread holds [`Log::freeze`]; the freezing
    /// thread itself passes (a mutation it causes lands after the snapshot's sequence and is
    /// replayed, which every apply tolerates). Nested guards on one thread count once.
    pub fn mutation(&self) -> MutationGuard {
        let depth = MUTATION_DEPTH.with(|d| {
            let v = d.get();
            d.set(v + 1);
            v
        });
        if depth > 0 {
            return MutationGuard { gate: None };
        }
        let me = std::thread::current().id();
        let mut st = self.gate.state.lock().unwrap_or_else(|p| p.into_inner());
        while matches!(st.frozen_by, Some(t) if t != me) {
            st = self.gate.cv.wait(st).unwrap_or_else(|p| p.into_inner());
        }
        st.inflight += 1;
        MutationGuard {
            gate: Some(Arc::clone(&self.gate)),
        }
    }

    /// **Freeze**: no new framed mutation starts on another thread, and this returns once every
    /// one in flight has reached its store. Reads are untouched. One freezer at a time. Must not
    /// be called from inside a mutation (it would wait for itself).
    pub fn freeze(&self) -> FreezeGuard {
        debug_assert_eq!(
            MUTATION_DEPTH.with(|d| d.get()),
            0,
            "freeze from inside a mutation waits for itself"
        );
        let me = std::thread::current().id();
        let mut st = self.gate.state.lock().unwrap_or_else(|p| p.into_inner());
        while st.frozen_by.is_some_and(|t| t != me) {
            st = self.gate.cv.wait(st).unwrap_or_else(|p| p.into_inner());
        }
        st.frozen_by = Some(me);
        while st.inflight > 0 {
            st = self.gate.cv.wait(st).unwrap_or_else(|p| p.into_inner());
        }
        FreezeGuard {
            gate: Arc::clone(&self.gate),
        }
    }

    /// The directory.
    pub fn dir(&self) -> &Path {
        &self.dir
    }

    /// Make this log follower-only (a twin; a box being restored): `append`
    /// and `begin(None)` refuse with [`Error::Follower`], `append_at` works.
    pub fn set_follower(&self, on: bool) {
        self.follower.store(on, Ordering::Release);
    }

    /// See [`Log::set_follower`].
    pub fn is_follower(&self) -> bool {
        self.follower.load(Ordering::Acquire)
    }

    /// The test hold of a group commit this log honours ([`TEST_GROUP_ENV`]); `None` on every
    /// box.
    pub fn group_hold(&self) -> Option<usize> {
        self.test_group
    }

    /// The newest committed sequence; 0 when nothing has been committed.
    pub fn head_seq(&self) -> u64 {
        self.head.load(Ordering::Acquire)
    }

    /// The oldest sequence still retained. `head + 1` when the log is empty.
    pub fn first_seq(&self) -> u64 {
        self.first.load(Ordering::Acquire)
    }

    /// Append one transaction with the next sequence number. Durable when
    /// this returns.
    pub fn append(&self, kind: Kind, meta: &Meta, blob: &[u8]) -> Result<u64> {
        if self.is_follower() {
            return Err(Error::Follower);
        }
        let meta = meta.to_bytes();
        self.append_raw(kind as u16, &meta, blob)
    }

    /// [`Log::append`], but the caller says which sequence number the entry
    /// must get — the twin's path: it holds the primary's numbers, not its
    /// own. Refuses anything but `head + 1`.
    pub fn append_at(&self, seq: u64, kind: u16, meta: &[u8], blob: &[u8]) -> Result<u64> {
        {
            let mut w = self.lock_writer()?;
            let expected = w.seg.first + w.seg.offsets.len() as u64;
            if seq != expected {
                return Err(Error::OutOfOrder { got: seq, expected });
            }
            self.stage_locked(&mut w, kind, meta, blob)?;
        }
        self.sync_through(seq)?;
        Ok(seq)
    }

    /// Append with raw kind and meta bytes. Durable when this returns.
    ///
    /// **Group commit for concurrent appenders** (perf twin-apply-fsync, 2026-10-06): the frame
    /// is staged under the writer lock, and its fsync is [`Log::sync_through`]'s — taken OUTSIDE
    /// the writer lock by one leader for every frame staged so far. Eight request threads that
    /// append together pay about one `fdatasync` between them instead of eight; one appender alone
    /// pays exactly what it paid before (one write, one fdatasync, one index write).
    pub fn append_raw(&self, kind: u16, meta: &[u8], blob: &[u8]) -> Result<u64> {
        let seq = {
            let mut w = self.lock_writer()?;
            self.stage_locked(&mut w, kind, meta, blob)?
        };
        self.sync_through(seq)?;
        Ok(seq)
    }

    /// **[`Log::append`] without waiting for durability** — for a frame whose mutation is
    /// invisible until a LATER frame publishes it (a control document's shard blob, published by
    /// the `ControlDoc` frame that moves the pointer): that frame's `fdatasync` covers this one,
    /// which is earlier in the same file. Refused on a follower, as `append` is.
    pub fn stage(&self, kind: Kind, meta: &Meta, blob: &[u8]) -> Result<u64> {
        if self.is_follower() {
            return Err(Error::Follower);
        }
        self.stage_raw(kind as u16, &meta.to_bytes(), blob)
    }

    /// [`Log::append_raw`] without waiting for durability: the frame is whole in the segment and
    /// holds its number; nobody reads it and it is not acknowledged until [`Log::sync_through`]
    /// (or any later sync) makes it durable. The caller MUST NOT change any state the frame
    /// describes before then — the log is written before the state, always.
    pub fn stage_raw(&self, kind: u16, meta: &[u8], blob: &[u8]) -> Result<u64> {
        let mut w = self.lock_writer()?;
        self.stage_locked(&mut w, kind, meta, blob)
    }

    fn stage_locked(&self, w: &mut Writer, kind: u16, meta: &[u8], blob: &[u8]) -> Result<u64> {
        self.roll_if_full(w)?;
        let seq = w.seg.first + w.seg.offsets.len() as u64;
        let header = Header {
            seq,
            kind,
            meta_len: meta.len() as u32,
            blob_len: blob.len() as u64,
        };
        let h = header.encode();
        let mut crc = Crc32c::new();
        crc.update(meta);
        crc.update(blob);
        let tr = crc.finish().to_le_bytes();
        let put = (|| {
            w.put(&h)?;
            w.put(meta)?;
            w.put(blob)?;
            w.put(&tr)
        })();
        if let Err(e) = put {
            // Half a frame is never left behind for the next one to land after — and when it
            // cannot be cut, nothing more is written behind it.
            if let Err(cut) = w.cut_to_committed() {
                self.poison(format!("a half-written frame could not be cut off the segment ({cut}) after its write failed ({e})"));
            }
            return Err(e.into());
        }
        let off = w.seg.len;
        w.seg.offsets.push(off);
        w.seg.len += header.frame_len();
        Ok(seq)
    }

    /// **Return once frame `seq` is durable** — making it so if nobody is: a waiter becomes the
    /// leader when no leader is in its fsync, writes out the stage, `fdatasync`s a duplicate of
    /// the segment's descriptor WITHOUT holding the writer lock (appenders keep staging behind it),
    /// then indexes and publishes everything that fsync covered and wakes every waiter. A waiter
    /// re-checks the head on EVERY wake-up, so the moment a leader's fsync covers its frame it
    /// returns — it never waits for a later leader's fsync, and no lock is held across I/O.
    pub fn sync_through(&self, seq: u64) -> Result<()> {
        {
            let mut leading = self.sync_state.lock().unwrap_or_else(|p| p.into_inner());
            loop {
                if self.head_seq() >= seq {
                    return Ok(());
                }
                if !*leading {
                    *leading = true;
                    break;
                }
                leading = self.synced.wait(leading).unwrap_or_else(|p| p.into_inner());
            }
        }
        // This thread leads until `_lead` drops — on every path out, an error and a panic
        // included — and then every waiter is woken to look at the head again.
        struct Lead<'a>(&'a Log);
        impl Drop for Lead<'_> {
            fn drop(&mut self) {
                *self.0.sync_state.lock().unwrap_or_else(|p| p.into_inner()) = false;
                self.0.synced.notify_all();
            }
        }
        let _lead = Lead(self);
        if let Some(n) = self.test_group {
            self.hold_for_the_group(n);
        }
        let snap = (|| -> Result<(File, u64, usize)> {
            let mut w = self.lock_writer()?;
            w.flush_buf()?;
            Ok((w.file.try_clone()?, w.seg.first, w.seg.offsets.len()))
        })();
        let (file, first, upto) = match snap {
            Ok(s) => s,
            // A release (the data set being grown) publishes everything first.
            Err(_) if self.head_seq() >= seq => return Ok(()),
            Err(e) => return Err(e),
        };
        #[cfg(test)]
        {
            let gate = self
                .fsync_gates
                .lock()
                .unwrap_or_else(|p| p.into_inner())
                .pop_front();
            if let Some(gate) = gate {
                let _ = gate.recv();
            }
        }
        if let Err(e) = self.sync_frames(&file) {
            let mut w = self.writer.lock().expect("txlog writer");
            if w.seg.first == first {
                return Err(self.poison_and_cut(&mut w, e));
            }
            // A roll since then published (and synced) that segment whole — the frames this
            // fsync was for are durable.
            drop(w);
            if self.head_seq() < seq {
                return Err(self.poison_and_cut(&mut self.writer.lock().expect("txlog writer"), e));
            }
            return Ok(());
        }
        self.frame_syncs.fetch_add(1, Ordering::Relaxed);
        let mut w = match self.lock_writer() {
            Ok(w) => w,
            Err(_) if self.head_seq() >= seq => return Ok(()),
            Err(e) => return Err(e),
        };
        // A roll since then published (and synced) that segment whole.
        if w.seg.first == first && upto > w.published {
            self.publish_upto(&mut w, upto)?;
        }
        Ok(())
    }

    /// **The test hold** ([`TEST_GROUP_ENV`]): the leader waits until `n` frames are staged and
    /// not yet durable — or [`TEST_GROUP_WAIT`] passes — so a count of fsyncs measures what the
    /// group commit shares when the writers ARE there, not how the scheduler happened to
    /// interleave them. The writer lock is taken only to read the count.
    fn hold_for_the_group(&self, n: usize) {
        let until = std::time::Instant::now() + TEST_GROUP_WAIT;
        loop {
            let staged = self.writer.lock().map(|w| w.unpublished()).unwrap_or(n);
            if staged >= n || std::time::Instant::now() >= until {
                return;
            }
            std::thread::sleep(std::time::Duration::from_micros(200));
        }
    }

    /// The `fdatasync` that makes frames durable (and, in a test, the one an injected failure
    /// replaces).
    fn sync_frames(&self, file: &File) -> io::Result<()> {
        #[cfg(test)]
        loop {
            let n = self.fail_fsyncs.load(Ordering::SeqCst);
            if n == 0 {
                break;
            }
            if self
                .fail_fsyncs
                .compare_exchange(n, n - 1, Ordering::SeqCst, Ordering::SeqCst)
                .is_ok()
            {
                return Err(io::Error::other("txlog test: injected fsync failure (EIO)"));
            }
        }
        file.sync_data()
    }

    /// Test-only: the next `n` frame fsyncs fail.
    #[cfg(test)]
    fn fail_next_fsyncs(&self, n: u32) {
        self.fail_fsyncs.store(n, Ordering::SeqCst);
    }

    /// Test-only: the next leader waits on `gate` before its fsync.
    #[cfg(test)]
    fn hold_next_fsync(&self, gate: std::sync::mpsc::Receiver<()>) {
        self.fsync_gates.lock().unwrap().push_back(gate);
    }

    /// **Make a group commit durable and visible** (perf twin-apply-fsync, 2026-10-06): every
    /// frame committed with [`Pending::commit_unsynced`] since the last call is written out,
    /// `fdatasync`ed ONCE, indexed with ONE write, and published — readers see it and the head
    /// moves to the last of them only now. A twin acknowledges its head to the primary, so what it
    /// acknowledges is what this made durable. Answers the head. Nothing pending = no I/O.
    pub fn sync(&self) -> Result<u64> {
        let mut w = self.lock_writer()?;
        self.publish(&mut w)?;
        Ok(self.head_seq())
    }

    /// Frames committed and not yet durable ([`Log::sync`] makes them so).
    pub fn unsynced(&self) -> u64 {
        self.writer.lock().expect("txlog writer").unpublished() as u64
    }

    /// The body of [`Log::sync`], under the writer lock.
    fn publish(&self, w: &mut Writer) -> Result<()> {
        w.flush_buf()?;
        let n = w.unpublished();
        if n == 0 {
            return Ok(());
        }
        if let Some(cause) = self.poisoned_by() {
            return Err(Error::Poisoned {
                dir: self.dir.clone(),
                cause,
            });
        }
        if let Err(e) = self.sync_frames(&w.file) {
            return Err(self.poison_and_cut(w, e));
        }
        self.frame_syncs.fetch_add(1, Ordering::Relaxed);
        self.publish_upto(w, w.seg.offsets.len())
    }

    /// What poisoned the log, if anything did ([`Error::Poisoned`]).
    pub fn poisoned_by(&self) -> Option<String> {
        self.poisoned
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .clone()
    }

    /// **Poison the log** ([`Error::Poisoned`]): the first cause is kept, said once at ERROR.
    fn poison(&self, cause: String) {
        let mut p = self.poisoned.lock().unwrap_or_else(|p| p.into_inner());
        if p.is_none() {
            tracing::error!(
                target: crate::observe::TARGET_STORE,
                event = "txlog.poisoned",
                dir = %self.dir.display(),
                cause = %cause,
                "the transaction log is poisoned: every write refuses until it is reopened"
            );
            *p = Some(cause);
        }
    }

    /// **A frame fsync failed**: poison the log, then cut every frame it did not make durable off
    /// the segment — staged, written, or in the page cache — so neither a later fsync nor a
    /// reopen can publish what its writer was told had failed. Answers the error to return.
    fn poison_and_cut(&self, w: &mut Writer, e: io::Error) -> Error {
        let mut cause = format!("a frame fsync failed ({e})");
        let keep = w.seg.offsets.get(w.published).copied().unwrap_or(w.seg.len);
        w.seg.offsets.truncate(w.published);
        w.seg.len = keep;
        if keep >= w.disk_len {
            w.buf.truncate((keep - w.disk_len) as usize);
        } else {
            w.buf.clear();
            let cut = (|| {
                w.file.set_len(keep)?;
                w.file.seek(SeekFrom::Start(keep))?;
                w.disk_len = keep;
                w.file.sync_data()
            })();
            if let Err(c) = cut {
                cause.push_str(&format!(
                    ", and the frames it did not make durable could not be cut ({c})"
                ));
            }
        }
        self.poison(cause.clone());
        Error::Poisoned {
            dir: self.dir.clone(),
            cause,
        }
    }

    /// **How many `fdatasync`s made frames durable** since this log was opened — the leader's in
    /// [`Log::sync_through`] and a group's in [`Log::sync`] (a roll, a release and a cut sync too,
    /// rarely, and are not counted). What the I/O tests hold to a bound per entry.
    pub fn frame_syncs(&self) -> u64 {
        self.frame_syncs.load(Ordering::Relaxed)
    }

    /// Index and publish the open segment's frames up to `upto` — already durable.
    fn publish_upto(&self, w: &mut Writer, upto: usize) -> Result<()> {
        if upto <= w.published {
            return Ok(());
        }
        let fresh: Vec<u64> = w.seg.offsets[w.published..upto].to_vec();
        let mut bytes = Vec::with_capacity(fresh.len() * 8);
        for off in &fresh {
            bytes.extend_from_slice(&off.to_le_bytes());
        }
        // The index is a cache appended AFTER the data's fsync, never fsynced itself (the
        // module's rule): one write for the whole group.
        w.idx.write_all(&bytes)?;
        w.published = upto;
        let end = match w.seg.offsets.get(upto) {
            Some(&next) => next,
            None => w.seg.len,
        };
        {
            let mut segs = self.segments.lock().expect("txlog segments");
            let open = segs.last_mut().expect("open segment");
            debug_assert_eq!(
                open.first, w.seg.first,
                "the published open segment is the writer's"
            );
            open.offsets.extend_from_slice(&fresh);
            open.len = end;
        }
        let last = w.seg.first + upto as u64 - 1;
        self.head.store(last, Ordering::Release);
        Ok(())
    }

    /// Begin a streaming append for a blob whose bytes arrive in chunks —
    /// the twin receiving a large pack over the wire without holding it in
    /// memory. The writer lock is held until [`Pending::commit`] or
    /// [`Pending::abort`]; the frame does not exist until `commit`.
    pub fn begin(
        &self,
        seq: Option<u64>,
        kind: u16,
        meta: &[u8],
        blob_len: u64,
    ) -> Result<Pending<'_>> {
        if seq.is_none() && self.is_follower() {
            return Err(Error::Follower);
        }
        let mut w = self.lock_writer()?;
        self.roll_if_full(&mut w)?;
        let next = w.seg.first + w.seg.offsets.len() as u64;
        if let Some(seq) = seq {
            if seq != next {
                return Err(Error::OutOfOrder {
                    got: seq,
                    expected: next,
                });
            }
        }
        let header = Header {
            seq: next,
            kind,
            meta_len: meta.len() as u32,
            blob_len,
        };
        let h = header.encode();
        w.put(&h)?;
        w.put(meta)?;
        let mut crc = Crc32c::new();
        crc.update(meta);
        Ok(Pending {
            log: self,
            w,
            header,
            crc,
            written: 0,
            finished: false,
        })
    }

    /// The writer, or [`Error::Released`] while the files are given up.
    fn lock_writer(&self) -> Result<std::sync::MutexGuard<'_, Writer>> {
        let w = self.writer.lock().expect("txlog writer");
        if let Some(cause) = self.poisoned_by() {
            return Err(Error::Poisoned {
                dir: self.dir.clone(),
                cause,
            });
        }
        if self.released.load(Ordering::Acquire) {
            return Err(Error::Released(self.dir.clone()));
        }
        Ok(w)
    }

    /// **Give up the writer's files** so the filesystem under the log can be unmounted — the
    /// runtime data-set growth (gunnar-server `grow`), where PID 1 unmounts `/data` itself.
    ///
    /// MEASURED 2026-09-15 under KVM (growbench R5): the open segment's `.log` and `.idx` were
    /// the only descriptors still under `/data` after the stores closed, and `umount` answered
    /// EBUSY — so every runtime growth on a box carrying the log aborted.
    ///
    /// Waits for an in-flight [`Pending`] (it holds the writer lock), syncs both files, and
    /// parks the handles on `/dev/null`; readers open per call and are unaffected until the
    /// unmount itself. Every write refuses with [`Error::Released`] until [`Log::reacquire`].
    /// Idempotent.
    pub fn release(&self) -> Result<()> {
        let mut w = self.writer.lock().expect("txlog writer");
        if self.released.load(Ordering::Acquire) {
            return Ok(());
        }
        self.publish(&mut w)?;
        w.file.sync_all()?;
        w.idx.sync_all()?;
        w.file = OpenOptions::new().append(true).open("/dev/null")?;
        w.idx = OpenOptions::new().append(true).open("/dev/null")?;
        self.released.store(true, Ordering::Release);
        Ok(())
    }

    /// **Take the files back** after the remount. The open segment must be exactly as long as
    /// the writer left it — anything else is a different disk, and is refused by name rather
    /// than appended to. Idempotent.
    pub fn reacquire(&self) -> Result<()> {
        let mut w = self.writer.lock().expect("txlog writer");
        if !self.released.load(Ordering::Acquire) {
            return Ok(());
        }
        let path = seg_path(&self.dir, w.seg.first);
        let mut file = OpenOptions::new().append(true).open(&path)?;
        let on_disk = file.metadata()?.len();
        if on_disk != w.seg.len {
            return Err(Error::Corrupt {
                path,
                offset: on_disk,
                what: "the open segment changed length while the log was released",
            });
        }
        file.seek(SeekFrom::Start(w.seg.len))?;
        let idx = OpenOptions::new()
            .append(true)
            .open(idx_path(&self.dir, w.seg.first))?;
        w.file = file;
        w.idx = idx;
        w.disk_len = w.seg.len;
        self.released.store(false, Ordering::Release);
        Ok(())
    }

    /// The files are given up ([`Log::release`]).
    pub fn is_released(&self) -> bool {
        self.released.load(Ordering::Acquire)
    }

    fn roll_if_full(&self, w: &mut Writer) -> Result<()> {
        if w.seg.len < self.segment_bytes || w.seg.offsets.is_empty() {
            return Ok(());
        }
        let next_first = w.seg.first + w.seg.offsets.len() as u64;
        // A group in progress ends with the segment it was written into.
        self.publish(w)?;
        w.file.sync_all()?;
        w.idx.sync_all()?;
        let file = File::create(seg_path(&self.dir, next_first))?;
        let idx = File::create(idx_path(&self.dir, next_first))?;
        sync_dir(&self.dir)?;
        w.file = file;
        w.idx = idx;
        w.disk_len = 0;
        w.published = 0;
        w.seg = Segment {
            first: next_first,
            offsets: Vec::new(),
            len: 0,
        };
        self.segments.lock().expect("txlog segments").push(Segment {
            first: next_first,
            offsets: Vec::new(),
            len: 0,
        });
        Ok(())
    }

    /// Locate a frame: which segment, which offset, how long.
    fn locate(&self, seq: u64) -> Result<(PathBuf, u64)> {
        let head = self.head_seq();
        if seq > head {
            return Err(Error::BeyondHead { seq, head });
        }
        let first = self.first_seq();
        if seq < first {
            return Err(Error::NotRetained { seq, first });
        }
        let segs = self.segments.lock().expect("txlog segments");
        let i = match segs.binary_search_by(|s| s.first.cmp(&seq)) {
            Ok(i) => i,
            Err(0) => return Err(Error::NotRetained { seq, first }),
            Err(i) => i - 1,
        };
        let seg = &segs[i];
        let k = (seq - seg.first) as usize;
        let off = *seg.offsets.get(k).ok_or(Error::BeyondHead { seq, head })?;
        Ok((seg_path(&self.dir, seg.first), off))
    }

    /// Read one frame whole.
    pub fn read(&self, seq: u64) -> Result<Entry> {
        let (path, off) = self.locate(seq)?;
        let mut file = File::open(&path)?;
        file.seek(SeekFrom::Start(off))?;
        let mut hdr = [0u8; HEADER_LEN];
        file.read_exact(&mut hdr)?;
        let h = Header::decode(&hdr).ok_or(Error::Corrupt {
            path: path.clone(),
            offset: off,
            what: "header",
        })?;
        let mut meta = vec![0u8; h.meta_len as usize];
        file.read_exact(&mut meta)?;
        if h.kind == Kind::Redacted as u16 {
            // Erased: the meta is whole, the blob is gone. Nothing to check it against.
            return Ok(Entry {
                seq: h.seq,
                kind: h.kind,
                meta,
                blob: Vec::new(),
            });
        }
        let mut blob = vec![0u8; h.blob_len as usize];
        file.read_exact(&mut blob)?;
        let mut tr = [0u8; TRAILER_LEN];
        file.read_exact(&mut tr)?;
        let mut crc = Crc32c::new();
        crc.update(&meta);
        crc.update(&blob);
        if crc.finish() != u32::from_le_bytes(tr) {
            return Err(Error::Corrupt {
                path,
                offset: off,
                what: "payload crc",
            });
        }
        Ok(Entry {
            seq: h.seq,
            kind: h.kind,
            meta,
            blob,
        })
    }

    /// Read a frame's header and meta, and say where its blob is, so a
    /// caller can stream the blob in chunks.
    pub fn locate_blob(&self, seq: u64) -> Result<Located> {
        let (path, off) = self.locate(seq)?;
        let mut file = File::open(&path)?;
        file.seek(SeekFrom::Start(off))?;
        let mut hdr = [0u8; HEADER_LEN];
        file.read_exact(&mut hdr)?;
        let h = Header::decode(&hdr).ok_or(Error::Corrupt {
            path: path.clone(),
            offset: off,
            what: "header",
        })?;
        let mut meta = vec![0u8; h.meta_len as usize];
        file.read_exact(&mut meta)?;
        Ok(Located {
            seq: h.seq,
            kind: h.kind,
            meta,
            path,
            blob_offset: off + HEADER_LEN as u64 + h.meta_len as u64,
            // An erased frame streams nothing: its bytes are gone.
            blob_len: if h.kind == Kind::Redacted as u16 {
                0
            } else {
                h.blob_len
            },
            payload_crc: None,
        })
    }

    /// **One frame's blob in pieces of at most `piece` bytes, never whole** — the primary's half
    /// of a twin's fetch (twin-streaming, 2026-10-06). `each` gets the frame's header and meta
    /// ([`Located`]), one piece, and whether it is the LAST.
    ///
    /// **The last piece is handed over only after the payload crc checked out**, the same check
    /// [`Log::read`] makes before it answers. A reader that makes the entry visible on `last` (a
    /// twin's commit) therefore never makes a corrupt one visible: a frame that fails its crc
    /// ends in [`Error::Corrupt`] with every piece but the last delivered, and the reader drops
    /// what it was given. Memory is one piece, whatever the blob's size.
    ///
    /// An erased frame ([`Kind::Redacted`]) is one empty last piece and is not checked: its bytes
    /// are gone and its crc was over them. A frame with an empty blob checks the crc over its
    /// meta and is one empty last piece.
    pub fn read_pieces<E: From<Error>>(
        &self,
        seq: u64,
        piece: usize,
        mut each: impl FnMut(&Located, &[u8], bool) -> std::result::Result<(), E>,
    ) -> std::result::Result<Located, E> {
        let mut at = self.locate_blob(seq)?;
        let frame_at = at.blob_offset - HEADER_LEN as u64 - at.meta.len() as u64;
        let corrupt = |path: &Path| Error::Corrupt {
            path: path.to_path_buf(),
            offset: frame_at,
            what: "payload crc",
        };
        if at.kind == Kind::Redacted as u16 {
            each(&at, &[], true)?;
            return Ok(at);
        }
        let mut file = File::open(&at.path).map_err(Error::from)?;
        file.seek(SeekFrom::Start(at.blob_offset))
            .map_err(Error::from)?;
        let mut crc = Crc32c::new();
        crc.update(&at.meta);
        let mut buf = vec![0u8; (piece.max(1) as u64).min(at.blob_len) as usize];
        let mut left = at.blob_len;
        loop {
            let n = (buf.len() as u64).min(left) as usize;
            file.read_exact(&mut buf[..n]).map_err(Error::from)?;
            crc.update(&buf[..n]);
            left -= n as u64;
            if left > 0 {
                each(&at, &buf[..n], false)?;
                continue;
            }
            let mut tr = [0u8; TRAILER_LEN];
            file.read_exact(&mut tr).map_err(Error::from)?;
            if crc.finish() != u32::from_le_bytes(tr) {
                return Err(corrupt(&at.path).into());
            }
            at.payload_crc = Some(u32::from_le_bytes(tr));
            each(&at, &buf[..n], true)?;
            return Ok(at);
        }
    }

    /// **[`Log::read_pieces`] from byte `from` of the blob on** (restore-large-entry, 2026-10-07):
    /// the twin's half of a restore resuming an entry the box being filled holds part of. 0 is
    /// [`Log::read_pieces`] exactly. Otherwise only the blob from `from` is read — a resumed entry
    /// of gigabytes costs the twin its REST on every call, never a re-read of what was sent, which
    /// on a slow volume would eat the call's budget and stall the restore again near the end —
    /// and the crc cannot be checked here (it is over all of it): the last piece carries the
    /// frame's stored crc as [`Located::payload_crc`], unchecked, and the receiver checks it over
    /// the entry it put together. `from` at the blob's end is one empty last piece; `from` past
    /// it, or on an erased frame, is refused: an `InvalidInput` I/O error naming both numbers.
    pub fn read_pieces_from<E: From<Error>>(
        &self,
        seq: u64,
        from: u64,
        piece: usize,
        mut each: impl FnMut(&Located, &[u8], bool) -> std::result::Result<(), E>,
    ) -> std::result::Result<Located, E> {
        if from == 0 {
            return self.read_pieces(seq, piece, each);
        }
        let mut at = self.locate_blob(seq)?;
        if at.kind == Kind::Redacted as u16 || from > at.blob_len {
            return Err(Error::Io(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!(
                    "txlog: seq {seq} read from byte {from} of a {}-byte blob",
                    at.blob_len
                ),
            ))
            .into());
        }
        let mut file = File::open(&at.path).map_err(Error::from)?;
        file.seek(SeekFrom::Start(at.blob_offset + from))
            .map_err(Error::from)?;
        let mut left = at.blob_len - from;
        let mut buf = vec![0u8; (piece.max(1) as u64).min(left.max(1)) as usize];
        loop {
            let n = (buf.len() as u64).min(left) as usize;
            file.read_exact(&mut buf[..n]).map_err(Error::from)?;
            left -= n as u64;
            if left > 0 {
                each(&at, &buf[..n], false)?;
                continue;
            }
            let mut tr = [0u8; TRAILER_LEN];
            file.read_exact(&mut tr).map_err(Error::from)?;
            at.payload_crc = Some(u32::from_le_bytes(tr));
            each(&at, &buf[..n], true)?;
            return Ok(at);
        }
    }

    /// **Erase every frame before `before` that carries data of one of `repos`** — the
    /// `Pack`, `Refs`, `LfsPut` and `LfsDelete` frames whose meta names one of them
    /// exactly — in place (owner ruling g-nonpay: an account deleted for non-payment
    /// is deleted from the log too, on main and on the twin).
    ///
    /// Per segment, every matching frame is rewritten as [`Kind::Redacted`]: the HEADERS
    /// first, made durable with ONE `fdatasync` (from then on every reader reads those frames
    /// as erased, and a crash leaves erased frames, never a frame whose checksum fails), then
    /// the blobs' ranges are punched out of the file (zeros where the filesystem cannot punch),
    /// then ONE `fsync` — two per segment, where it used to be one per frame (R5.1-B). The sequence, the offsets and the index are untouched: a twin's gap
    /// rule, a reader's seek and the retention floor mean exactly what they meant.
    ///
    /// `RepoStore` frames stay: a replay needs the store's create and purge to land
    /// in order. Frames AT or after `before` are never touched — a repository created
    /// again under the same name after the erase is somebody's new data. Idempotent:
    /// an erased frame is punched again (cheap), so an erase a crash cut between its
    /// headers and its punches is finished by the next. **Appends do not wait for it**
    /// (R5.1-B, 2026-10-07): the writer lock is taken only to publish what is staged and
    /// to read the plan, and an erase only touches frames below `before`, all of them
    /// published; one erase runs at a time.
    pub fn erase(&self, before: u64, repos: &[String]) -> Result<Erased> {
        let _one = self.erase_lock.lock().unwrap_or_else(|p| p.into_inner());
        // **The writer lock only to publish and to read the plan** (R5.1-B, 2026-10-07). Every
        // frame below `before` is published (durable, indexed) once this block ends, and the erase
        // touches nothing else: appends go on at the tail beside it.
        let plan: Vec<(u64, Vec<u64>)> = {
            let mut w = self.lock_writer()?;
            self.publish(&mut w)?;
            self.segments
                .lock()
                .expect("txlog segments")
                .iter()
                .filter(|s| !s.offsets.is_empty() && s.first < before)
                .map(|s| (s.first, s.offsets.clone()))
                .collect()
        };
        let data = [
            Kind::Pack as u16,
            Kind::Refs as u16,
            Kind::LfsPut as u16,
            Kind::LfsDelete as u16,
        ];
        let mut out = Erased::default();
        let mut hdr = [0u8; HEADER_LEN];
        for (first, offsets) in plan {
            let path = seg_path(&self.dir, first);
            let mut file = match OpenOptions::new().read(true).write(true).open(&path) {
                Ok(f) => f,
                // Retention dropped the whole segment meanwhile: nothing of it is left to erase.
                Err(e) if e.kind() == io::ErrorKind::NotFound => continue,
                Err(e) => return Err(e.into()),
            };
            // Pass 1: every matching frame's header becomes Redacted; ONE fdatasync makes them all
            // durable before a single byte goes — from then on each frame IS erased, and a crash
            // leaves erased frames, never a frame whose checksum fails.
            let mut punch: Vec<(u64, u64)> = Vec::new();
            for (i, &off) in offsets.iter().enumerate() {
                let seq = first + i as u64;
                if seq >= before {
                    break;
                }
                file.seek(SeekFrom::Start(off))?;
                file.read_exact(&mut hdr)?;
                let h = Header::decode(&hdr).ok_or(Error::Corrupt {
                    path: path.clone(),
                    offset: off,
                    what: "header",
                })?;
                let redacted_already = h.kind == Kind::Redacted as u16;
                if !redacted_already && !data.contains(&h.kind) {
                    continue;
                }
                let mut meta = vec![0u8; h.meta_len as usize];
                file.read_exact(&mut meta)?;
                let named = Meta::from_bytes(&meta)
                    .ok()
                    .and_then(|m| m.repo().map(str::to_owned));
                if !named.is_some_and(|r| repos.contains(&r)) {
                    continue;
                }
                let blob_at = off + HEADER_LEN as u64 + h.meta_len as u64;
                if redacted_already {
                    // Erased before — punched again (idempotent, cheap), so an erase a crash cut
                    // between its headers and its punches is finished by the next one.
                    punch.push((blob_at, h.blob_len));
                    continue;
                }
                let redacted = Header {
                    kind: Kind::Redacted as u16,
                    ..h
                };
                file.seek(SeekFrom::Start(off))?;
                file.write_all(&redacted.encode())?;
                punch.push((blob_at, h.blob_len));
                out.frames += 1;
                out.bytes += h.blob_len;
            }
            if punch.is_empty() {
                continue;
            }
            file.sync_data()?;
            self.erase_syncs.fetch_add(1, Ordering::Relaxed);
            #[cfg(test)]
            self.erase_gate();
            // Pass 2: the bytes, then ONE fsync.
            for (at, len) in punch {
                zero_range(&mut file, at, len)?;
            }
            file.sync_all()?;
            self.erase_syncs.fetch_add(1, Ordering::Relaxed);
        }
        Ok(out)
    }

    /// **How many fsyncs [`Log::erase`] made** since this log was opened.
    pub fn erase_syncs(&self) -> u64 {
        self.erase_syncs.load(Ordering::Relaxed)
    }

    /// Test-only: the next erase waits on `gate` once its first header is down.
    #[cfg(test)]
    fn hold_next_erase(&self, gate: std::sync::mpsc::Receiver<()>) {
        self.erase_gates.lock().unwrap().push_back(gate);
    }

    #[cfg(test)]
    fn erase_gate(&self) {
        let gate = self
            .erase_gates
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .pop_front();
        if let Some(gate) = gate {
            let _ = gate.recv();
        }
    }

    /// **Walk every retained frame's header and meta, oldest first**, without its blob: one
    /// open and one sequential buffered read per SEGMENT, each blob skipped by a relative seek
    /// inside the buffer, one header array and one meta buffer reused for every frame (T17).
    ///
    /// [`Log::locate_blob`] opens the segment file once per frame; a scan that runs on every
    /// start (`gunnar_server::twin::rebuild_audit_trail`) cannot pay an `open(2)` per frame of a
    /// long log. Bounded by the head at call time; a frame whose header does not decode, or
    /// whose sequence is not the one its position says, is [`Error::Corrupt`].
    pub fn scan_meta(&self, mut each: impl FnMut(u64, u16, &[u8])) -> Result<()> {
        let head = self.head_seq();
        let plan: Vec<(u64, u64, u64)> = self
            .segments
            .lock()
            .expect("txlog segments")
            .iter()
            .filter(|s| !s.offsets.is_empty())
            .map(|s| (s.first, s.offsets[0], s.offsets.len() as u64))
            .collect();
        let mut hdr = [0u8; HEADER_LEN];
        let mut meta: Vec<u8> = Vec::with_capacity(256);
        for (first, start, n) in plan {
            let path = seg_path(&self.dir, first);
            let mut r = io::BufReader::with_capacity(1 << 16, File::open(&path)?);
            r.seek(SeekFrom::Start(start))?;
            let mut offset = start;
            for seq in first..first + n {
                if seq > head {
                    return Ok(());
                }
                r.read_exact(&mut hdr)?;
                let h = match Header::decode(&hdr) {
                    Some(h) if h.seq == seq => h,
                    Some(_) => {
                        return Err(Error::Corrupt {
                            path,
                            offset,
                            what: "sequence out of place",
                        })
                    }
                    None => {
                        return Err(Error::Corrupt {
                            path,
                            offset,
                            what: "header",
                        })
                    }
                };
                meta.resize(h.meta_len as usize, 0);
                r.read_exact(&mut meta)?;
                each(seq, h.kind, &meta);
                r.seek_relative((h.blob_len + TRAILER_LEN as u64) as i64)?;
                offset += h.frame_len();
            }
        }
        Ok(())
    }

    /// Read `from..=to` whole (bounded by the head).
    pub fn range(&self, from: u64, to: u64) -> Result<Vec<Entry>> {
        let to = to.min(self.head_seq());
        let mut out = Vec::new();
        let mut seq = from;
        while seq <= to {
            out.push(self.read(seq)?);
            seq += 1;
        }
        Ok(out)
    }

    /// Drop every whole segment whose last sequence is below `seq`. The
    /// open segment is never dropped. Returns the new first sequence.
    pub fn truncate_before(&self, seq: u64) -> Result<u64> {
        let _w = self.lock_writer()?;
        let mut segs = self.segments.lock().expect("txlog segments");
        while segs.len() > 1 {
            let drop = match segs[0].last_seq() {
                Some(last) => last < seq,
                None => true,
            };
            if !drop {
                break;
            }
            let first = segs[0].first;
            fs::remove_file(seg_path(&self.dir, first))?;
            let _ = fs::remove_file(idx_path(&self.dir, first));
            segs.remove(0);
        }
        sync_dir(&self.dir)?;
        let first = segs[0].first;
        self.first.store(first, Ordering::Release);
        Ok(first)
    }

    /// **Retain from `seq`**: close the open segment (when it holds anything) so every frame
    /// up to the head sits in a closed segment, then drop every whole segment below `seq`.
    /// Returns the new first sequence. `seq` past the head retains nothing old: the log keeps
    /// its numbers and the next append takes `head + 1`. The operator's retention door — and
    /// the drill's way to open a retention hole without writing a gigabyte first.
    pub fn retain_from(&self, seq: u64) -> Result<u64> {
        {
            let mut w = self.writer.lock().expect("txlog writer");
            self.publish(&mut w)?;
            if !w.seg.offsets.is_empty() {
                let next_first = w.seg.first + w.seg.offsets.len() as u64;
                w.file.sync_all()?;
                w.idx.sync_all()?;
                let file = File::create(seg_path(&self.dir, next_first))?;
                let idx = File::create(idx_path(&self.dir, next_first))?;
                sync_dir(&self.dir)?;
                w.file = file;
                w.idx = idx;
                w.disk_len = 0;
                w.published = 0;
                w.seg = Segment {
                    first: next_first,
                    offsets: Vec::new(),
                    len: 0,
                };
                self.segments.lock().expect("txlog segments").push(Segment {
                    first: next_first,
                    offsets: Vec::new(),
                    len: 0,
                });
            }
        }
        self.truncate_before(seq)
    }

    /// **Start over**: drop every segment and begin again at sequence 1 with
    /// nothing in the log. The restore door on a fresh box: its own first
    /// frames (its minted KEK) make way for the twin's numbers. Refused when
    /// the log holds more than `keep_at_most` frames — a box with history is
    /// not fresh, and a reset would be data loss with a verb on it.
    pub fn reset(&self, keep_at_most: u64) -> Result<()> {
        let head = self.head_seq();
        if head > keep_at_most {
            return Err(Error::OutOfOrder {
                got: head,
                expected: keep_at_most,
            });
        }
        self.restart_after(0)
    }

    /// **Start over after `at`**: drop every segment; the log is empty with its head at `at`
    /// and the next append taking `at + 1`. What a twin does once it has installed a snapshot
    /// taken at `at` — it holds that state, and the log resumes from the next number.
    pub fn restart_after(&self, at: u64) -> Result<()> {
        let mut w = self.lock_writer()?;
        let mut segs = self.segments.lock().expect("txlog segments");
        for s in segs.iter() {
            fs::remove_file(seg_path(&self.dir, s.first))?;
            let _ = fs::remove_file(idx_path(&self.dir, s.first));
        }
        segs.clear();
        let first = at + 1;
        File::create(seg_path(&self.dir, first))?;
        File::create(idx_path(&self.dir, first))?;
        sync_dir(&self.dir)?;
        segs.push(Segment {
            first,
            offsets: Vec::new(),
            len: 0,
        });
        w.file = OpenOptions::new()
            .append(true)
            .open(seg_path(&self.dir, first))?;
        w.idx = OpenOptions::new()
            .append(true)
            .open(idx_path(&self.dir, first))?;
        w.seg = Segment {
            first,
            offsets: Vec::new(),
            len: 0,
        };
        w.disk_len = 0;
        w.buf.clear();
        w.published = 0;
        self.head.store(at, Ordering::Release);
        self.first.store(first, Ordering::Release);
        Ok(())
    }

    /// **The segments as retention sees them**, oldest first, the open one last:
    /// `(first, last, bytes)` — `last` is `first - 1` for a segment that holds no
    /// frame yet. What a byte-bounded retention decides from (2026-10-06, cov-13):
    /// only whole segments are ever dropped, so the bytes a `from` keeps are the
    /// sum of the segments at and after the one holding it.
    pub fn layout(&self) -> Vec<(u64, u64, u64)> {
        self.segments
            .lock()
            .expect("txlog segments")
            .iter()
            .map(|s| (s.first, s.first + s.offsets.len() as u64 - 1, s.len))
            .collect()
    }

    /// Bytes on disk across all segments.
    pub fn bytes(&self) -> u64 {
        self.segments
            .lock()
            .expect("txlog segments")
            .iter()
            .map(|s| s.len)
            .sum()
    }
}

/// A streaming append in progress. See [`Log::begin`].
///
/// **Dropped without [`Pending::commit`] or [`Pending::abort`], it aborts** (twin-streaming,
/// 2026-10-06). The partial frame is already in the segment file, and the writer appends with
/// `O_APPEND`: before this, a `?` between `begin` and `commit` (a write that failed, a wire that
/// broke under a twin's fetch) left the half frame where it was, the next append landed AFTER
/// it while the index recorded the offset BEFORE it, and every read of that next frame was a
/// corrupt header — a log broken by one failed transfer. Now the half frame is cut off the
/// segment on every path out, and a partial transfer leaves exactly nothing.
pub struct Pending<'a> {
    log: &'a Log,
    w: std::sync::MutexGuard<'a, Writer>,
    header: Header,
    crc: Crc32c,
    written: u64,
    /// Committed, or the partial frame already cut: nothing for `Drop` to do.
    finished: bool,
}

impl Pending<'_> {
    /// The sequence number this frame will commit as.
    pub fn seq(&self) -> u64 {
        self.header.seq
    }

    /// Bytes of the blob still owed.
    pub fn remaining(&self) -> u64 {
        self.header.blob_len - self.written
    }

    /// Write a chunk of the blob.
    pub fn write(&mut self, chunk: &[u8]) -> Result<()> {
        if chunk.len() as u64 > self.remaining() {
            return Err(Error::Io(io::Error::new(
                io::ErrorKind::InvalidInput,
                "txlog: chunk exceeds the declared blob length",
            )));
        }
        self.w.put(chunk)?;
        self.crc.update(chunk);
        self.written += chunk.len() as u64;
        Ok(())
    }

    /// Finish: trailer, fsync, index. Durable when this returns — and so is every frame of a
    /// group committed before it ([`Pending::commit_unsynced`]).
    pub fn commit(mut self) -> Result<u64> {
        let seq = self.close()?;
        self.log.publish(&mut self.w)?;
        Ok(seq)
    }

    /// **Finish WITHOUT making it durable — one frame of a group commit** (perf twin-apply-fsync,
    /// 2026-10-06). The frame is whole in the segment (staged or written, trailer and all), it
    /// holds its sequence number — the next `begin` takes the one after — but it is not fsynced,
    /// not indexed, not visible to readers and not the head until [`Log::sync`] (or any synced
    /// append, roll or release) makes the whole group durable with ONE `fdatasync`.
    ///
    /// A crash before that loses the group's tail, never anything acknowledged: `open` keeps
    /// every whole frame that reached the disk (each checks its own crc) and cuts the rest, and a
    /// twin re-fetches from its durable head.
    pub fn commit_unsynced(mut self) -> Result<u64> {
        self.close()
    }

    /// The trailer and the writer's own bookkeeping: the frame is committed, not yet durable.
    fn close(&mut self) -> Result<u64> {
        if self.remaining() != 0 {
            let owed = self.remaining();
            self.abort_inner()?;
            return Err(Error::Io(io::Error::new(
                io::ErrorKind::UnexpectedEof,
                format!("txlog: blob short by {owed} bytes at commit"),
            )));
        }
        let tr = self.crc.finish().to_le_bytes();
        self.w.put(&tr)?;
        // From here the frame is whole: it is the writer's, never cut by a drop.
        self.finished = true;
        let off = self.w.seg.len;
        self.w.seg.offsets.push(off);
        self.w.seg.len += self.header.frame_len();
        Ok(self.header.seq)
    }

    /// Give up: the partial frame is cut off the segment.
    pub fn abort(mut self) -> Result<()> {
        self.abort_inner()
    }

    fn abort_inner(&mut self) -> Result<()> {
        self.finished = true;
        if let Err(e) = self.w.cut_to_committed() {
            // The half frame stays where the next append would land after it: nothing more is.
            self.log.poison(format!(
                "a partial frame could not be cut off the segment ({e})"
            ));
            return Err(e.into());
        }
        Ok(())
    }
}

impl Drop for Pending<'_> {
    fn drop(&mut self) {
        if self.finished {
            return;
        }
        // A cut that fails here poisons the log; the half frame is the torn tail `open` cuts at
        // the next start (it was never indexed and its trailer never written).
        let _ = self.abort_inner();
    }
}

/// **Is this path (relative to a data root) a CACHE** — something rebuilt from packs,
/// derived from another store, or scratch, that may be lost and must never exist on the
/// twin? The exploded read cache, the start-up disk probe, repack and quarantine scratch,
/// the metadata mirror (a derived copy of the identity table), the grow journal, and every
/// temporary. The drill scans the twin's disk for these.
pub fn is_cache_path(rel: &str) -> bool {
    let rel = rel.trim_start_matches("./");
    // A box with no /serialdat partition has that tree at `<data-root>/.serialdat`, with
    // `crash`, `log`, `gc`, `quarantine` linked to it: judge a path by its place in the tree.
    let rel = rel
        .strip_prefix(".serialdat/")
        .or_else(|| rel.strip_prefix("serialdat/"))
        .unwrap_or(rel);
    let first = rel.split('/').next().unwrap_or("");
    let name = rel.rsplit('/').next().unwrap_or("");
    matches!(first, ".korp-disktest" | "gc" | "quarantine" | "meta" | "serialdat")
        || rel.split('/').any(|c| c.starts_with("objects.exploded"))
        // The commit-graph memo the engine folds from the packs (C2): derived, and off on a twin.
        || name == "objects.graph"
        || name.ends_with(".tmp")
        || name.starts_with("incoming-")
        || name == "grow-journal"
}

/// **The control plane's audit trail, as the LOG names it** — one canonical relative path,
/// whatever the layout config put the `log` directory on.
///
/// It is a frame like `attestations.jsonl` is a frame, and it is spelled here rather than
/// derived from a physical path because the two boxes in a restore need not agree on where
/// `log/` physically lives: the frame carries the name, the applying box resolves the name.
pub const AUDIT_REL: &str = "log/audit-chain.jsonl";

/// **A twin's — or a restoring box's — OWN audit trail.** Per box, never framed, never sent.
///
/// A twin serves a control plane of its own (`Mirror.*`, browse), and every decision it makes
/// is ITS decision, not the primary's. Writing those into [`AUDIT_REL`] would interleave a
/// second writer into a hash chain whose whole value is that it has exactly one — the `prev`
/// links would not hold and the restored copy would be refused at open. So the twin's own
/// decisions go here, this file stays [`is_per_box_path`], and the twin **originates nothing**
/// into the trail it is holding for the primary.
///
/// The consequence is the one worth having: a restored box opens [`AUDIT_REL`], finds the dead
/// primary's tip, and appends its own first decision with `prev` = that tip. **The trail is one
/// unbroken chain across the disaster.**
pub const AUDIT_LOCAL_REL: &str = "log/audit-chain.local.jsonl";

/// **The site's WireGuard hub declaration** — `<data root>/wg/hub.conf`, laid there by
/// `korp-installer` through the ESP data stage on a fresh install
/// (`korp_installer::esp::DATA_STAGE_DIR`), then never written again.
///
/// It is **public site data**: the hub's public key, endpoint, allowed IPs and the
/// per-box `[Interface] Address` gunnar requires (`gunnar_server::appliance`). It is the
/// answer to "how is this box reached", and a restored main that does not have it is a box
/// nobody can reach — the worst thing to discover on the day the twin is used.
///
/// Framed since 2026-09-21 (lane T11): the twin holds it and a restore brings it back.
pub const HUB_CONF_REL: &str = "wg/hub.conf";

/// **The digest of the hub declaration this box last framed**, so a boot that changes
/// nothing frames nothing. Per box, never framed — it is a record of what THIS log already
/// carries, and it is meaningless on any other box.
pub const HUB_CONF_FRAMED_REL: &str = "wg/hub.conf.framed";

/// **The machine's own WireGuard identity** — the private key minted on the machine and
/// wrapped to its own TPM, and its public half.
///
/// **Never replicated, and now refused by name.** Two reasons, either sufficient: it is a
/// PRIVATE KEY, and a twin that held it would be a second place to steal the site's network
/// identity from; and it names THIS box to the hub, so a restored main carrying it would come
/// up claiming the dead machine's identity on a tunnel the hub has registered to a key that no
/// longer exists anywhere. The same ruling holger makes about its admin token and its role
/// record. Before this they were merely un-framed by accident — nothing wrote them through the
/// sink — which is not the same as being refused.
pub const WG_IDENTITY_KEY_REL: &str = "wg/identity.key";
/// See [`WG_IDENTITY_KEY_REL`]. The public half is equally per box: it names this machine.
pub const WG_IDENTITY_PUB_REL: &str = "wg/identity.pub";

/// **Is this path a record of THIS box** — not cache and not a transaction: its own crash
/// records and console log (`crash/`, `log/`), a twin's own `twin.json` and
/// `twin.key`, and the restore markers. A twin keeps its own; the primary's are never sent.
///
/// **[`AUDIT_REL`] is the one exception, since 2026-09-21 (lane T11).** It used to be swept up
/// by `log/` and was therefore the one class of state a restore silently destroyed: the record
/// of who did what, gone on exactly the day somebody needs it. It is a signed-shaped
/// append-only log like the other nine in [`FRAMED_FILES`] and it is replicated like them.
/// [`AUDIT_LOCAL_REL`] stays per box — see its docs for why the two must not be one file.
pub fn is_per_box_path(rel: &str) -> bool {
    let rel = rel.trim_start_matches("./");
    // A box with no /serialdat partition has that tree at `<data-root>/.serialdat`, with
    // `crash`, `log`, `gc`, `quarantine` linked to it: judge a path by its place in the tree.
    let rel = rel
        .strip_prefix(".serialdat/")
        .or_else(|| rel.strip_prefix("serialdat/"))
        .unwrap_or(rel);
    if rel == AUDIT_REL || rel == HUB_CONF_REL {
        return false;
    }
    // The machine's own WireGuard identity and the digest of what this log already carries:
    // see [`WG_IDENTITY_KEY_REL`] and [`HUB_CONF_FRAMED_REL`].
    if matches!(
        rel,
        WG_IDENTITY_KEY_REL | WG_IDENTITY_PUB_REL | HUB_CONF_FRAMED_REL
    ) {
        return true;
    }
    let first = rel.split('/').next().unwrap_or("");
    // `txlog-snapshot-base` is the snapshot a store-only twin KEEPS as bytes (ruling 2026-09-22);
    // `-dropped` / `-too-small` are its resync's room markers (twin-resync-space, 2026-10-06).
    matches!(
        first,
        "crash" | "log" | "twin.json" | "twin.key" | "txlog-restoring" | "txlog-fresh-docs.json" | "txlog-reseal" | "txlog-snapshot-installing" | "txlog-snapshot-base" | "txlog-snapshot-base.old"
            | "txlog-snapshot-dropped" | "txlog-snapshot-too-small"
            // The entry a restoring box holds part of (restore-large-entry, 2026-10-07).
            | "txlog-restore-partial"
            // The twin's base a restoring box holds part of (restore-after-snapshot, 2026-10-07).
            | "txlog-restore-snapshot"
    )
}

/// **The documents under a data root that are frames** (`Append` and `File` kinds): the signed
/// append-only logs and their tree heads, and the whole documents. A snapshot carries each that
/// exists; installing one removes each that the snapshot does not carry. A new sink writer adds
/// its file here (the twin test fixture checks every sink-framed path is listed).
pub const FRAMED_FILES: &[&str] = &[
    "attestations.jsonl",
    "transparency-heads.jsonl",
    "builds.jsonl",
    "build-heads.jsonl",
    "forge.jsonl",
    "forge-heads.jsonl",
    "membership.jsonl",
    "membership-heads.jsonl",
    "replication.json",
    "webhooks.json",
    // The control plane's audit trail (lane T11, 2026-09-21). Listed LAST so the ordering of
    // the nine above — which several fixtures index positionally — does not move.
    AUDIT_REL,
    // The site's WireGuard hub declaration (lane T11): a restored main that does not have it
    // is a box nobody can reach.
    HUB_CONF_REL,
    // The identity core's records (sessions and invites as hashes, the people, the floors;
    // IDENTITY-HOST): a twin and a snapshot carry them, so a main rebuilt from its twin has the
    // main's identity state (RA5-10).
    "identity.json",
];

/// **Never a transaction**: cache ([`is_cache_path`]) or per box ([`is_per_box_path`]).
/// The sink refuses to frame one and the twin refuses to apply one.
pub fn never_framed(rel: &str) -> bool {
    is_cache_path(rel) || is_per_box_path(rel)
}

/// **The sink**: how a writer that holds only a PATH (an append-only signed
/// log under the data root, an atomically replaced document) frames its write
/// in the log of the data root that path is under — without threading an
/// `Arc<Log>` through every constructor that names a file.
///
/// `ServerStores::open` registers `(data root, log)` on a primary. A path
/// under no registered root (a test's scratch file, a box with no log, a
/// twin — whose writes are applies, never originations) frames nothing. The
/// registry holds `Weak`s, so a dropped store unregisters itself.
pub mod sink {
    use super::*;

    static ROOTS: Mutex<Vec<(PathBuf, Weak<Log>)>> = Mutex::new(Vec::new());

    fn refuse_cache(rel: &str) -> Result<()> {
        if never_framed(rel) {
            return Err(Error::BadMeta(format!(
                "{rel} is a cache or per-box path; it is never a transaction"
            )));
        }
        Ok(())
    }

    /// Frame writes under `root` into `log`.
    pub fn register(root: &Path, log: &Arc<Log>) {
        let mut roots = ROOTS.lock().expect("txlog sink");
        roots.retain(|(r, w)| w.strong_count() > 0 && r != root);
        roots.push((root.to_path_buf(), Arc::downgrade(log)));
    }

    /// The log `path` is framed into and `path` relative to its root.
    pub fn log_for(path: &Path) -> Option<(Arc<Log>, String)> {
        let roots = ROOTS.lock().expect("txlog sink");
        roots
            .iter()
            .filter(|(r, _)| path.starts_with(r))
            .max_by_key(|(r, _)| r.as_os_str().len())
            .and_then(|(r, w)| {
                let log = w.upgrade()?;
                let rel = path.strip_prefix(r).ok()?.to_string_lossy().into_owned();
                Some((log, rel))
            })
    }

    /// A framed write in progress: hold it until the file has the write (the mutation gate,
    /// [`Log::mutation`]). Empty when the path is under no registered root.
    #[must_use = "hold the frame until the file has the write"]
    pub struct Framed(#[allow(dead_code)] Option<MutationGuard>);

    /// Before `line` (no newline) is appended to the file at `path`.
    pub fn append_line(path: &Path, line: &str) -> Result<Framed> {
        if let Some((log, rel)) = log_for(path) {
            refuse_cache(&rel)?;
            let guard = log.mutation();
            log.append(Kind::Append, &Meta::Append { log: rel }, line.as_bytes())?;
            return Ok(Framed(Some(guard)));
        }
        Ok(Framed(None))
    }

    /// **A line framed and NOT yet durable** ([`stage_line`]): hold it — it is the mutation
    /// guard — until the file has the line, and write the line only after [`StagedLine::durable`].
    #[must_use = "wait for the frame to be durable, then write the line, then drop this"]
    pub struct StagedLine {
        _guard: MutationGuard,
        log: Arc<Log>,
        seq: u64,
    }

    impl StagedLine {
        /// The frame's sequence number.
        pub fn seq(&self) -> u64 {
            self.seq
        }
        /// Return once the frame is durable — group-committed with every frame staged with it
        /// ([`Log::sync_through`]).
        pub fn durable(&self) -> Result<()> {
            self.log.sync_through(self.seq)
        }
        /// The log the frame is in.
        pub fn log(&self) -> &Arc<Log> {
            &self.log
        }
    }

    /// **[`append_line`], group-committed** (perf, 2026-10-06): the frame is staged and its
    /// fsync is shared with every other frame staged before the leader's ([`Log::sync_through`]).
    /// `None` = the path is under no registered root (nothing is framed).
    pub fn stage_line(path: &Path, line: &str) -> Result<Option<StagedLine>> {
        if let Some((log, rel)) = log_for(path) {
            refuse_cache(&rel)?;
            if log.is_follower() {
                return Err(Error::Follower);
            }
            let guard = log.mutation();
            let seq = log.stage_raw(
                Kind::Append as u16,
                &Meta::Append { log: rel }.to_bytes(),
                line.as_bytes(),
            )?;
            return Ok(Some(StagedLine {
                _guard: guard,
                log,
                seq,
            }));
        }
        Ok(None)
    }

    /// Before the document at `path` is replaced by `bytes`, or removed.
    pub fn replace_file(path: &Path, bytes: &[u8], deleted: bool) -> Result<Framed> {
        if let Some((log, rel)) = log_for(path) {
            refuse_cache(&rel)?;
            let guard = log.mutation();
            log.append(Kind::File, &Meta::File { path: rel, deleted }, bytes)?;
            return Ok(Framed(Some(guard)));
        }
        Ok(Framed(None))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn meta(repo: &str) -> Meta {
        Meta::Pack {
            repo: repo.into(),
            hash: "sha256".into(),
        }
    }

    /// ★ **A failed fsync poisons the log** (stinky row 3, fsyncgate). The caller of the append
    /// whose `fdatasync` failed was told so, but its frame stayed staged; the NEXT leader's
    /// fsync — which Linux answers with success once the error has been reported — published
    /// it, and a twin applied a transaction main had reported as failed. After a failed fsync
    /// nothing more is written or published until the log is reopened from its committed point.
    #[test]
    fn a_failed_fsync_poisons_the_log_until_it_is_reopened() {
        let dir = tempfile::tempdir().unwrap();
        let m = meta("g/r").to_bytes();
        {
            let log = Log::open(dir.path()).unwrap();
            assert_eq!(log.append_raw(Kind::Pack as u16, &m, b"one").unwrap(), 1);
            log.fail_next_fsyncs(1);
            log.append_raw(Kind::Pack as u16, &m, b"two")
                .expect_err("its fsync failed");
            let next = log.append_raw(Kind::Pack as u16, &m, b"three");
            assert!(next.is_err(), "an append after a failed fsync was taken (and published the failed frame with it): {next:?}");
            assert!(
                log.sync().is_err() || log.head_seq() == 1,
                "a sync published past the failed fsync"
            );
            assert_eq!(
                log.head_seq(),
                1,
                "the frame whose fsync failed was published"
            );
            assert!(
                log.read(2).is_err(),
                "the frame whose fsync failed is readable"
            );
        }
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(
            log.head_seq(),
            1,
            "a reopen resurrected the frame whose fsync failed"
        );
        assert_eq!(log.append_raw(Kind::Pack as u16, &m, b"again").unwrap(), 2);
        assert_eq!(log.read(2).unwrap().blob, b"again");
    }

    /// ★ **A failed cut poisons the log** (stinky row 6). A frame whose write failed half way is
    /// cut back off the segment; when that truncate ALSO failed it was ignored (`let _ =`), the
    /// half frame stayed on disk and the next append landed after it while the index recorded
    /// the offset before it — every later read a corrupt header, replication broken from there.
    #[test]
    fn a_failed_cut_of_a_half_frame_poisons_the_log() {
        let dir = tempfile::tempdir().unwrap();
        let m = meta("g/r").to_bytes();
        {
            let log = Log::open(dir.path()).unwrap();
            assert_eq!(log.append_raw(Kind::Pack as u16, &m, b"one").unwrap(), 1);
            {
                let mut w = log.writer.lock().unwrap();
                // header, meta and the (directly written) blob go out; the trailer fails.
                w.faults.fail_put_after = Some(3);
                w.faults.fail_cut = true;
            }
            let big = vec![7u8; DIRECT_BYTES * 2];
            log.append_raw(Kind::Pack as u16, &m, &big)
                .expect_err("its trailer failed");
            let next = log.append_raw(Kind::Pack as u16, &m, b"three");
            assert!(
                next.is_err(),
                "an append after a failed cut landed after the half frame: {next:?}"
            );
            assert_eq!(log.head_seq(), 1);
        }
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 1);
        assert_eq!(log.append_raw(Kind::Pack as u16, &m, b"three").unwrap(), 2);
        assert_eq!(log.read(2).unwrap().blob, b"three");
    }

    #[test]
    fn crc32c_known_answer() {
        // RFC 3720 / standard check value for "123456789".
        assert_eq!(crc32c(b"123456789"), 0xE306_9283);
        // Carried across a restart: the state kept after "1234" goes on to the same answer.
        let mut a = Crc32c::new();
        a.update(b"1234");
        let mut b = Crc32c::resume(a.state());
        b.update(b"56789");
        assert_eq!(b.finish(), 0xE306_9283);
    }

    #[test]
    fn appends_are_dense_and_read_back() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 0);
        assert_eq!(log.first_seq(), 1);
        for i in 1..=10u64 {
            let s = log
                .append(Kind::Pack, &meta("a/b"), &[i as u8; 100])
                .unwrap();
            assert_eq!(s, i);
        }
        assert_eq!(log.head_seq(), 10);
        let e = log.read(7).unwrap();
        assert_eq!(e.seq, 7);
        assert_eq!(e.kind(), Some(Kind::Pack));
        assert_eq!(e.blob, vec![7u8; 100]);
        assert_eq!(e.meta().unwrap(), meta("a/b"));
        assert!(matches!(log.read(11), Err(Error::BeyondHead { .. })));
        assert!(matches!(log.read(0), Err(Error::NotRetained { .. })));
    }

    #[test]
    fn reopen_keeps_head_and_trusts_index() {
        let dir = tempfile::tempdir().unwrap();
        {
            let log = Log::open(dir.path()).unwrap();
            for i in 0..5u64 {
                log.append(Kind::Refs, &meta("x/y"), &[i as u8]).unwrap();
            }
        }
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 5);
        assert_eq!(log.read(5).unwrap().blob, vec![4u8]);
        assert_eq!(log.append(Kind::Refs, &meta("x/y"), b"6").unwrap(), 6);
    }

    #[test]
    fn torn_tail_is_cut_and_the_log_goes_on() {
        let dir = tempfile::tempdir().unwrap();
        {
            let log = Log::open(dir.path()).unwrap();
            log.append(Kind::Pack, &meta("a/b"), &[1u8; 1000]).unwrap();
            log.append(Kind::Pack, &meta("a/b"), &[2u8; 1000]).unwrap();
        }
        let path = seg_path(dir.path(), 1);
        let len = fs::metadata(&path).unwrap().len();
        // Tear the last frame: cut 100 bytes off it.
        let f = OpenOptions::new().write(true).open(&path).unwrap();
        f.set_len(len - 100).unwrap();
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 1, "the torn second frame is gone");
        assert_eq!(
            fs::metadata(&path).unwrap().len(),
            len - 1000 - 32 - 4 - meta("a/b").to_bytes().len() as u64
        );
        assert_eq!(log.append(Kind::Pack, &meta("a/b"), &[3u8; 10]).unwrap(), 2);
        drop(log);
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.read(2).unwrap().blob, vec![3u8; 10]);
    }

    /// One frame through `begin` → `write` → `commit_unsynced`, at `seq`.
    fn unsynced(log: &Log, seq: u64, blob: &[u8]) -> u64 {
        let m = meta("g/r").to_bytes();
        let mut p = log
            .begin(Some(seq), Kind::Pack as u16, &m, blob.len() as u64)
            .unwrap();
        p.write(blob).unwrap();
        p.commit_unsynced().unwrap()
    }

    /// **A group commit is invisible until `sync`, then durable as one** (perf twin-apply-fsync):
    /// the frames hold their numbers (the next `begin` takes the one after), nobody reads them and
    /// the head does not move until `sync`, which writes the index once and moves the head to the
    /// group's last; a reopen finds them all.
    #[test]
    fn a_group_is_invisible_until_synced_and_then_durable_as_one() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        log.set_follower(true);
        for seq in 1..=5u64 {
            assert_eq!(unsynced(&log, seq, &[seq as u8; 100]), seq);
        }
        assert_eq!(log.head_seq(), 0, "nothing acknowledged before the sync");
        assert_eq!(log.unsynced(), 5);
        assert!(
            matches!(log.read(1), Err(Error::BeyondHead { .. })),
            "a reader does not see a frame that is not durable"
        );
        assert!(
            matches!(
                log.begin(Some(5), Kind::Pack as u16, b"{}", 0),
                Err(Error::OutOfOrder {
                    got: 5,
                    expected: 6
                })
            ),
            "the gap rule counts the group"
        );
        assert_eq!(
            fs::metadata(idx_path(dir.path(), 1)).unwrap().len(),
            0,
            "the index is written after the data's fsync"
        );
        assert_eq!(log.sync().unwrap(), 5);
        assert_eq!((log.head_seq(), log.unsynced()), (5, 0));
        assert_eq!(log.read(3).unwrap().blob, vec![3u8; 100]);
        assert_eq!(
            fs::metadata(idx_path(dir.path(), 1)).unwrap().len(),
            5 * 8,
            "one index entry per frame"
        );
        assert_eq!(log.sync().unwrap(), 5, "nothing pending: nothing done");
        drop(log);
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 5);
        assert_eq!(log.read(5).unwrap().blob, vec![5u8; 100]);
    }

    /// **A crash before the sync loses only what was never acknowledged.** A synced group stays;
    /// a group still staged in memory is simply gone (it never reached the file); a group whose
    /// bytes reached the file but were never fsynced is kept frame by frame where whole (each frame
    /// checks its own crc), and a half frame is cut. The log takes the next number either way.
    #[test]
    fn a_crash_before_the_sync_loses_only_what_was_never_acknowledged() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        for seq in 1..=3u64 {
            unsynced(&log, seq, b"acked");
        }
        assert_eq!(log.sync().unwrap(), 3);
        // Staged, never written: a crash (here: the process forgets the log) loses them.
        unsynced(&log, 4, b"staged");
        unsynced(&log, 5, b"staged");
        drop(log);
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(
            log.head_seq(),
            3,
            "the acknowledged group is whole; the staged one never reached the file"
        );
        // Written (a large piece goes straight to the file), never fsynced, then a half frame.
        let big = vec![7u8; DIRECT_BYTES * 2];
        unsynced(&log, 4, &big);
        let m = meta("g/r").to_bytes();
        let mut half = log
            .begin(Some(5), Kind::Pack as u16, &m, (DIRECT_BYTES * 4) as u64)
            .unwrap();
        half.write(&big).unwrap();
        std::mem::forget(half); // the process dies with the frame half-written: no cut, no sync
        let path = seg_path(dir.path(), 1);
        let on_disk = fs::metadata(&path).unwrap().len();
        drop(log);
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(
            log.head_seq(),
            4,
            "the whole frame that reached the file is kept, the half one is cut"
        );
        assert_eq!(log.read(4).unwrap().blob, big);
        assert!(
            fs::metadata(&path).unwrap().len() < on_disk,
            "the half frame is gone from the segment"
        );
        assert_eq!(log.append(Kind::Pack, &meta("g/r"), b"next").unwrap(), 5);
    }

    /// **A frame given up inside a group takes only itself** — the frames committed before it in
    /// the same group stay, staged or written, and the sync makes them durable.
    #[test]
    fn an_abort_inside_a_group_keeps_the_frames_committed_before_it() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        unsynced(&log, 1, b"one");
        unsynced(&log, 2, &vec![2u8; DIRECT_BYTES]);
        let m = meta("g/r").to_bytes();
        {
            let mut p = log.begin(Some(3), Kind::Pack as u16, &m, 10).unwrap();
            p.write(b"abc").unwrap();
            // dropped without commit: cut
        }
        {
            let mut p = log
                .begin(Some(3), Kind::Pack as u16, &m, (DIRECT_BYTES * 2) as u64)
                .unwrap();
            p.write(&vec![9u8; DIRECT_BYTES]).unwrap();
            p.abort().unwrap();
        }
        assert_eq!(log.sync().unwrap(), 2);
        assert_eq!(log.read(2).unwrap().blob, vec![2u8; DIRECT_BYTES]);
        assert_eq!(unsynced(&log, 3, b"three"), 3);
        assert_eq!(log.sync().unwrap(), 3);
        drop(log);
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 3);
        assert_eq!(log.read(3).unwrap().blob, b"three".to_vec());
    }

    /// **A waiter whose frame is already durable returns at once — it never waits for the NEXT
    /// leader's fsync** (r5 stall review, 2026-10-07: the UI rig's control writes took 10–36 s
    /// with every gunnar-server thread in `futex_wait`).
    ///
    /// The leader used to hold a `Mutex<()>` across its `fdatasync`, and a waiter could only see
    /// that its frame was durable after ACQUIRING that mutex. Here leader A's fsync covers frames
    /// 1 and 2; B (frame 3, staged after A took its snapshot) queues for the leader BEFORE X
    /// (frame 2). When A ends, the mutex goes to B, which leads frame 3's fsync — held here by a
    /// gate the test never opens until the end — and X, whose frame has been durable since A
    /// published, stays parked behind it. On a loaded disk that is one unrelated fsync after
    /// another for as long as writes keep coming.
    ///
    /// RED with the mutex (X does not return while B's fsync is held); GREEN with the flag and
    /// condvar, where every waiter re-reads the head on every wake-up.
    #[test]
    fn a_waiter_whose_frame_is_durable_returns_without_waiting_for_the_next_leaders_fsync() {
        use std::sync::mpsc::channel;
        use std::time::Duration;
        let dir = tempfile::tempdir().unwrap();
        let log = Arc::new(Log::open(dir.path()).unwrap());
        let m = meta("g/r").to_bytes();
        assert_eq!(log.stage_raw(Kind::Pack as u16, &m, b"one").unwrap(), 1);
        assert_eq!(log.stage_raw(Kind::Pack as u16, &m, b"x").unwrap(), 2);
        let (open_a, gate_a) = channel::<()>();
        let (open_b, gate_b) = channel::<()>();
        log.hold_next_fsync(gate_a);
        log.hold_next_fsync(gate_b);
        let gates_left = |log: &Log| log.fsync_gates.lock().unwrap().len();

        // A leads frames 1..=2 and is held before its fsync.
        let a = {
            let log = Arc::clone(&log);
            std::thread::spawn(move || log.sync_through(1).unwrap())
        };
        while gates_left(&log) != 1 {
            std::thread::sleep(Duration::from_millis(5));
        }
        // Frame 3 is staged after A's snapshot; B waits for it first, then X for frame 2.
        assert_eq!(log.stage_raw(Kind::Pack as u16, &m, b"three").unwrap(), 3);
        let b = {
            let log = Arc::clone(&log);
            std::thread::spawn(move || log.sync_through(3).unwrap())
        };
        std::thread::sleep(Duration::from_millis(200));
        let (x_done, x_said) = channel::<()>();
        let x = {
            let log = Arc::clone(&log);
            std::thread::spawn(move || {
                log.sync_through(2).unwrap();
                let _ = x_done.send(());
            })
        };
        std::thread::sleep(Duration::from_millis(200));

        // A's fsync goes through: frames 1 and 2 are durable and published.
        open_a.send(()).unwrap();
        a.join().unwrap();
        assert!(log.head_seq() >= 2, "A published frames 1 and 2");
        // B now leads frame 3 and is held before ITS fsync. X's frame is durable: X must be back.
        let t = std::time::Instant::now();
        while gates_left(&log) != 0 && t.elapsed() < Duration::from_secs(5) {
            std::thread::sleep(Duration::from_millis(5));
        }
        let held = gates_left(&log) == 0;
        let back = x_said.recv_timeout(Duration::from_secs(5));
        open_b.send(()).unwrap();
        b.join().unwrap();
        x.join().unwrap();
        assert!(held, "B led frame 3's fsync (its gate was taken)");
        assert!(
            back.is_ok(),
            "X's frame was durable since A published it, yet X waited behind B's fsync of frame 3"
        );
        assert_eq!(log.head_seq(), 3);
    }

    /// **An account's erase costs about two fsyncs per segment, and appends do not wait for it**
    /// (R5.1-B, 2026-10-07). `Log::erase` held the writer lock for the whole erase and made an
    /// `fdatasync` per erased frame (the redacted header) plus one per segment: an account with
    /// 200 frames was ~200 fsyncs with every append on the box — every control write, every
    /// audited call — waiting behind them. Per segment, ONE fdatasync makes every redacted header
    /// durable before any byte is punched, and ONE fsync follows the punches; the writer lock is
    /// taken only to publish and to read the plan.
    #[test]
    fn an_erase_is_two_fsyncs_per_segment_and_appends_do_not_wait_for_it() {
        use std::sync::mpsc::channel;
        use std::time::Duration;
        let dir = tempfile::tempdir().unwrap();
        let log = Arc::new(Log::open_with(dir.path(), 64 << 10).unwrap());
        for i in 0..200u64 {
            let repo = if i % 2 == 0 { "gone/a" } else { "stays/b" };
            log.append(Kind::Pack, &meta(repo), &vec![i as u8; 1500])
                .unwrap();
            log.append(Kind::Pack, &meta("gone/a"), &vec![7u8; 600])
                .unwrap();
        }
        let before = log.append(Kind::Refs, &meta("stays/b"), b"marker").unwrap();
        let segments = log.segments.lock().unwrap().len() as u64;
        let syncs0 = log.erase_syncs();

        // The erase is held once its first header is down; an append meanwhile must not wait.
        let (open, gate) = channel::<()>();
        log.hold_next_erase(gate);
        let eraser = {
            let log = Arc::clone(&log);
            std::thread::spawn(move || log.erase(before, &["gone/a".to_string()]).unwrap())
        };
        while !log.erase_gates.lock().unwrap().is_empty() {
            std::thread::sleep(Duration::from_millis(5));
        }
        let (done, said) = channel::<u64>();
        let appender = {
            let log = Arc::clone(&log);
            std::thread::spawn(move || {
                let s = log.append(Kind::Refs, &meta("stays/b"), b"during").unwrap();
                let _ = done.send(s);
            })
        };
        let appended = said.recv_timeout(Duration::from_secs(3));
        open.send(()).unwrap();
        let erased = eraser.join().unwrap();
        appender.join().unwrap();
        let syncs = log.erase_syncs() - syncs0;
        let line = format!(
            "erased {} frames in {segments} segments with {syncs} fsyncs",
            erased.frames
        );
        eprintln!("{line}");
        assert_eq!(erased.frames, 300, "{line}");
        assert!(appended.is_ok(), "an append waited for the erase: {line}");
        assert!(
            syncs <= 2 * segments,
            "about two fsyncs per segment: {line}"
        );
        // Erased frames read as erased, the others as they were.
        for seq in 1..before {
            let e = log.read(seq).unwrap();
            let gone = Meta::from_bytes(&e.meta).unwrap().repo() == Some("gone/a");
            assert_eq!(e.kind == Kind::Redacted as u16, gone, "seq {seq}");
            assert_eq!(e.blob.is_empty(), gone, "seq {seq}");
        }
    }

    /// **A group that crosses a segment roll** is published at the roll (the old segment is synced
    /// whole before the new one starts) and the rest at the sync: every frame reads back, across
    /// segments, before and after a reopen. A synced `append` after a group makes the group durable
    /// first, in order.
    #[test]
    fn a_group_across_a_segment_roll_and_a_synced_append_after_it() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open_with(dir.path(), 4096).unwrap();
        for seq in 1..=40u64 {
            unsynced(&log, seq, &vec![seq as u8; 300]);
        }
        assert!(
            log.head_seq() > 0,
            "the rolls published the groups in the closed segments"
        );
        assert_eq!(log.sync().unwrap(), 40);
        unsynced(&log, 41, b"group");
        assert_eq!(
            log.append(Kind::Pack, &meta("g/r"), b"synced").unwrap(),
            42,
            "the append comes after the group"
        );
        assert_eq!(log.head_seq(), 42);
        for seq in 1..=40u64 {
            assert_eq!(
                log.read(seq).unwrap().blob,
                vec![seq as u8; 300],
                "seq {seq}"
            );
        }
        drop(log);
        let log = Log::open_with(dir.path(), 4096).unwrap();
        assert_eq!(log.head_seq(), 42);
        assert_eq!(log.read(41).unwrap().blob, b"group".to_vec());
        assert_eq!(log.read(42).unwrap().blob, b"synced".to_vec());
    }

    #[test]
    fn torn_header_is_cut_too() {
        let dir = tempfile::tempdir().unwrap();
        {
            let log = Log::open(dir.path()).unwrap();
            log.append(Kind::Pack, &meta("a/b"), &[1u8; 10]).unwrap();
        }
        let path = seg_path(dir.path(), 1);
        let mut f = OpenOptions::new().append(true).open(&path).unwrap();
        f.write_all(b"GTX1\x05\x00\x00").unwrap(); // seven bytes of a header
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 1);
        assert_eq!(log.append(Kind::Pack, &meta("a/b"), &[2u8; 10]).unwrap(), 2);
    }

    /// `scan_meta` answers exactly what `locate_blob` answers, frame by frame, across segment
    /// boundaries and a retention hole, with blobs of every size skipped (T17).
    #[test]
    fn scan_meta_agrees_with_locate_blob_across_segments_and_retention() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open_with(dir.path(), 4096).unwrap();
        for i in 0..200u64 {
            let blob = vec![i as u8; (i as usize * 37) % 9000];
            log.append(Kind::Pack, &meta(&format!("r/{i}")), &blob)
                .unwrap();
        }
        log.retain_from(50).unwrap();
        let mut seen = Vec::new();
        log.scan_meta(|seq, kind, m| seen.push((seq, kind, m.to_vec())))
            .unwrap();
        let want: Vec<_> = (log.first_seq()..=log.head_seq())
            .map(|s| {
                let at = log.locate_blob(s).unwrap();
                (at.seq, at.kind, at.meta)
            })
            .collect();
        assert!(log.first_seq() > 1 && !want.is_empty());
        assert_eq!(seen, want);
    }

    #[test]
    fn corrupted_payload_in_tail_is_cut() {
        let dir = tempfile::tempdir().unwrap();
        let first_len;
        {
            let log = Log::open(dir.path()).unwrap();
            log.append(Kind::Pack, &meta("a/b"), &[1u8; 10]).unwrap();
            first_len = log.bytes();
            log.append(Kind::Pack, &meta("a/b"), &[2u8; 10]).unwrap();
        }
        let path = seg_path(dir.path(), 1);
        let mut f = OpenOptions::new().write(true).open(&path).unwrap();
        // Flip one byte of the second frame's blob.
        f.seek(SeekFrom::Start(
            first_len + 32 + meta("a/b").to_bytes().len() as u64 + 3,
        ))
        .unwrap();
        f.write_all(&[0xEE]).unwrap();
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 1);
    }

    #[test]
    fn stale_index_is_rebuilt_from_data() {
        let dir = tempfile::tempdir().unwrap();
        {
            let log = Log::open(dir.path()).unwrap();
            for _ in 0..4 {
                log.append(Kind::Pack, &meta("a/b"), &[9u8; 10]).unwrap();
            }
        }
        // Lose the index entirely, and separately make it lie.
        fs::remove_file(idx_path(dir.path(), 1)).unwrap();
        fs::write(idx_path(dir.path(), 1), []).unwrap();
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 4);
        drop(log);
        fs::write(idx_path(dir.path(), 1), [0u8; 8 * 9]).unwrap();
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 4);
        assert_eq!(log.read(3).unwrap().seq, 3);
    }

    #[test]
    fn segments_roll_and_retention_drops_whole_segments() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open_with(dir.path(), 300).unwrap();
        for i in 1..=20u64 {
            log.append(Kind::Pack, &meta("a/b"), &[i as u8; 100])
                .unwrap();
        }
        assert!(log.segments.lock().unwrap().len() > 3);
        assert_eq!(log.read(1).unwrap().blob[0], 1);
        let first = log.truncate_before(10).unwrap();
        assert!(first <= 10 && first > 1, "first={first}");
        assert!(matches!(log.read(1), Err(Error::NotRetained { .. })));
        assert_eq!(log.read(10).unwrap().blob[0], 10);
        assert_eq!(log.read(20).unwrap().blob[0], 20);
        drop(log);
        let log = Log::open_with(dir.path(), 300).unwrap();
        assert_eq!(log.first_seq(), first);
        assert_eq!(log.head_seq(), 20);
        assert_eq!(
            log.append(Kind::Pack, &meta("a/b"), &[21u8; 10]).unwrap(),
            21
        );
    }

    #[test]
    fn append_at_refuses_out_of_order() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        let m = meta("a/b").to_bytes();
        assert_eq!(log.append_at(1, 1, &m, b"x").unwrap(), 1);
        assert!(matches!(
            log.append_at(3, 1, &m, b"x"),
            Err(Error::OutOfOrder {
                got: 3,
                expected: 2
            })
        ));
        assert!(matches!(
            log.append_at(1, 1, &m, b"x"),
            Err(Error::OutOfOrder { .. })
        ));
        assert_eq!(log.append_at(2, 1, &m, b"x").unwrap(), 2);
    }

    #[test]
    fn streaming_append_commits_and_aborts() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        let m = meta("a/b").to_bytes();
        {
            let mut p = log.begin(Some(1), 1, &m, 6).unwrap();
            p.write(b"abc").unwrap();
            p.write(b"def").unwrap();
            assert_eq!(p.commit().unwrap(), 1);
        }
        {
            let mut p = log.begin(None, 1, &m, 6).unwrap();
            p.write(b"abc").unwrap();
            p.abort().unwrap();
        }
        assert_eq!(log.head_seq(), 1);
        {
            let mut p = log.begin(None, 1, &m, 3).unwrap();
            p.write(b"xy").unwrap();
            assert!(p.commit().is_err(), "short blob is refused");
        }
        assert_eq!(log.head_seq(), 1);
        assert_eq!(log.append_at(2, 1, &m, b"ok").unwrap(), 2);
        drop(log);
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 2);
        assert_eq!(log.read(1).unwrap().blob, b"abcdef");
        assert_eq!(log.read(2).unwrap().blob, b"ok");
        let l = log.locate_blob(1).unwrap();
        assert_eq!(l.blob_len, 6);
        let mut f = File::open(&l.path).unwrap();
        f.seek(SeekFrom::Start(l.blob_offset)).unwrap();
        let mut b = vec![0u8; 6];
        f.read_exact(&mut b).unwrap();
        assert_eq!(b, b"abcdef");
    }

    /// **A streaming append DROPPED half-way leaves nothing** (twin-streaming, 2026-10-06): the
    /// shape of a twin's fetch whose wire broke inside an entry, or any `?` between `begin` and
    /// `commit`. Without `Drop` cutting it, the half frame stayed in the segment, the next append
    /// went after it (`O_APPEND`) while the index recorded the offset before it, and that next
    /// frame read back as a corrupt header — in this process AND after a restart.
    #[test]
    fn a_streaming_append_dropped_half_way_leaves_nothing_and_the_next_append_reads_back() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        let m = meta("a/b").to_bytes();
        assert_eq!(log.append_at(1, 1, &m, b"first").unwrap(), 1);
        let before = log.bytes();
        {
            let mut p = log.begin(Some(2), 1, &m, 1 << 20).unwrap();
            p.write(&[7u8; 4096]).unwrap();
            // dropped: no commit, no abort
        }
        assert_eq!(log.head_seq(), 1, "the dropped frame is not in the log");
        let seg = log.locate_blob(1).unwrap().path;
        assert_eq!(
            std::fs::metadata(&seg).unwrap().len(),
            before,
            "and not in the segment file either"
        );
        assert_eq!(log.append_at(2, 1, &m, b"second").unwrap(), 2);
        assert_eq!(
            log.read(2).unwrap().blob,
            b"second",
            "the next append reads back in this process"
        );
        drop(log);
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 2);
        assert_eq!(log.read(1).unwrap().blob, b"first");
        assert_eq!(log.read(2).unwrap().blob, b"second", "and after a restart");
    }

    /// **`read_pieces` hands a frame over in pieces and the last one only after the crc** — the
    /// primary's half of a twin's streamed fetch: the pieces are the blob, byte for byte; an
    /// empty blob and an erased frame are one empty last piece; and a frame whose bytes rotted
    /// on disk delivers every piece BUT the last and then refuses, so a reader that commits on
    /// `last` never commits it.
    #[test]
    fn read_pieces_streams_the_blob_and_withholds_the_last_piece_of_a_corrupt_frame() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        let blob: Vec<u8> = (0..10_000u32).map(|i| (i * 7 % 251) as u8).collect();
        let s = log.append(Kind::Pack, &meta("a/b"), &blob).unwrap();
        let empty = log.append(Kind::Refs, &meta("a/b"), b"").unwrap();
        let rot = log.append(Kind::Pack, &meta("a/c"), &blob).unwrap();
        let mut got = Vec::new();
        let mut lasts = 0;
        // The checked payload crc rides on the LAST piece only (restore-large-entry).
        let mut want = Crc32c::new();
        want.update(&meta("a/b").to_bytes());
        want.update(&blob);
        let want = want.finish();
        let at = log
            .read_pieces::<Error>(s, 3000, |l, p, last| {
                assert_eq!(l.seq, s);
                assert!(p.len() <= 3000);
                assert_eq!(
                    l.payload_crc,
                    last.then_some(want),
                    "the crc is handed over with the last piece and only then"
                );
                got.extend_from_slice(p);
                lasts += usize::from(last);
                Ok(())
            })
            .unwrap();
        assert_eq!(
            (got, lasts, at.blob_len),
            (blob.clone(), 1, blob.len() as u64)
        );
        assert_eq!(at.payload_crc, Some(want));
        // From an offset (a resumed restore): only the bytes from it, the stored crc on the last
        // piece; at the blob's end one empty last piece; past it, or on nothing, refused.
        let mut tail = Vec::new();
        let at = log
            .read_pieces_from::<Error>(s, 4321, 3000, |l, p, last| {
                assert!(p.len() <= 3000);
                assert_eq!(l.payload_crc, last.then_some(want));
                tail.extend_from_slice(p);
                Ok(())
            })
            .unwrap();
        assert_eq!(
            (tail.as_slice(), at.payload_crc),
            (&blob[4321..], Some(want))
        );
        let mut end = Vec::new();
        log.read_pieces_from::<Error>(s, blob.len() as u64, 3000, |l, p, last| {
            end.push((p.len(), last, l.payload_crc));
            Ok(())
        })
        .unwrap();
        assert_eq!(end, vec![(0, true, Some(want))]);
        assert!(log
            .read_pieces_from::<Error>(s, blob.len() as u64 + 1, 3000, |_, _, _| Ok(()))
            .is_err());
        let mut pieces = Vec::new();
        log.read_pieces::<Error>(empty, 3000, |_, p, last| {
            pieces.push((p.len(), last));
            Ok(())
        })
        .unwrap();
        assert_eq!(pieces, vec![(0, true)]);
        // One byte in the middle of the third frame's blob flips on disk.
        let l = log.locate_blob(rot).unwrap();
        let mut f = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&l.path)
            .unwrap();
        f.seek(SeekFrom::Start(l.blob_offset + 5000)).unwrap();
        let mut b = [0u8; 1];
        f.read_exact(&mut b).unwrap();
        f.seek(SeekFrom::Start(l.blob_offset + 5000)).unwrap();
        f.write_all(&[b[0] ^ 0xFF]).unwrap();
        drop(f);
        let mut seen = 0usize;
        let mut last_seen = false;
        let r = log.read_pieces::<Error>(rot, 3000, |_, p, last| {
            seen += p.len();
            last_seen |= last;
            Ok(())
        });
        assert!(
            matches!(
                r,
                Err(Error::Corrupt {
                    what: "payload crc",
                    ..
                })
            ),
            "{r:?}"
        );
        assert!(
            !last_seen,
            "the last piece of a corrupt frame is never handed over"
        );
        assert!(
            seen < blob.len(),
            "{seen} of {} bytes went out before the refusal",
            blob.len()
        );
        assert!(
            matches!(log.read(rot), Err(Error::Corrupt { .. })),
            "read refuses the same frame"
        );
        // An erased frame: one empty last piece, unchecked.
        log.erase(rot + 1, &["a/b".to_string()]).unwrap();
        let mut erased = Vec::new();
        log.read_pieces::<Error>(s, 3000, |l, p, last| {
            erased.push((l.kind, p.len(), last));
            Ok(())
        })
        .unwrap();
        assert_eq!(erased, vec![(Kind::Redacted as u16, 0, true)]);
    }

    #[test]
    fn unknown_kind_is_stored_and_read_back() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        log.append_raw(999, b"{}", b"future").unwrap();
        let e = log.read(1).unwrap();
        assert_eq!(e.kind(), None);
        assert_eq!(e.blob, b"future");
    }

    #[test]
    fn reset_starts_over_at_one_and_refuses_a_box_with_history() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        log.append(Kind::Kek, &Meta::Kek {}, &[7u8; 32]).unwrap();
        // A box with more frames than the door allows is not fresh.
        assert!(log.reset(0).is_err());
        assert_eq!(log.head_seq(), 1);
        log.reset(1).unwrap();
        assert_eq!(log.head_seq(), 0);
        assert_eq!(log.first_seq(), 1);
        // The twin's numbers go in from 1, and survive a reopen.
        log.append_at(1, Kind::Kek as u16, &Meta::Kek {}.to_bytes(), &[9u8; 32])
            .unwrap();
        drop(log);
        let log = Log::open(dir.path()).unwrap();
        assert_eq!(log.head_seq(), 1);
        assert_eq!(log.read(1).unwrap().blob, vec![9u8; 32]);
    }

    #[test]
    fn kek_append_and_file_kinds_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        let metas = [
            (Kind::Kek, Meta::Kek {}),
            (
                Kind::Append,
                Meta::Append {
                    log: "attestations.jsonl".into(),
                },
            ),
            (
                Kind::File,
                Meta::File {
                    path: "replication.json".into(),
                    deleted: false,
                },
            ),
        ];
        for (k, m) in &metas {
            log.append(*k, m, b"x").unwrap();
        }
        for (i, (k, m)) in metas.iter().enumerate() {
            let e = log.read(i as u64 + 1).unwrap();
            assert_eq!(e.kind(), Some(*k));
            assert_eq!(&e.meta().unwrap(), m);
        }
    }

    #[test]
    fn a_follower_log_refuses_originations_and_takes_the_primarys_numbers() {
        let dir = tempfile::tempdir().unwrap();
        let log = Log::open(dir.path()).unwrap();
        log.set_follower(true);
        assert!(matches!(
            log.append(Kind::Pack, &meta("a/b"), b"x"),
            Err(Error::Follower)
        ));
        assert!(matches!(
            log.begin(None, Kind::Pack as u16, b"{}", 1),
            Err(Error::Follower)
        ));
        log.append_at(1, Kind::Pack as u16, &meta("a/b").to_bytes(), b"x")
            .unwrap();
        assert_eq!(log.head_seq(), 1);
        log.set_follower(false);
        assert_eq!(log.append(Kind::Pack, &meta("a/b"), b"y").unwrap(), 2);
    }

    #[test]
    fn cache_and_per_box_paths_are_judged_by_their_place_in_the_serialdat_tree() {
        for p in [
            "gc/x",
            ".serialdat/gc/x",
            "serialdat/quarantine/p.pack",
            ".serialdat/meta/identity.json",
            "repos/a/b.git/objects.exploded/1",
            ".korp-disktest/probe",
            "webhooks.json.tmp",
        ] {
            assert!(is_cache_path(p) && never_framed(p), "{p} is cache");
        }
        for p in [
            "crash/panic-1",
            ".serialdat/log/console.txt",
            "log/audit-chain.local.jsonl",
            ".serialdat/log/audit-chain.local.jsonl",
            "twin.key",
        ] {
            assert!(
                !is_cache_path(p) && is_per_box_path(p) && never_framed(p),
                "{p} is per box"
            );
        }
        for p in [
            "attestations.jsonl",
            "replication.json",
            "repos/a/b.git/pack",
            "control.redb",
        ] {
            assert!(!never_framed(p), "{p} is a transaction's path");
        }
    }

    /// **The audit trail is a transaction, not a per-box record** (lane T11, 2026-09-21) — the
    /// D6 finding: a restore used to bring a box back with no audit history and say nothing.
    ///
    /// Both spellings the layout can produce, and the twin's OWN trail beside it, which must
    /// stay per box or two writers would interleave into one hash chain.
    ///
    /// **Fails before the fix**: delete the `if rel == AUDIT_REL { return false; }` early
    /// return from [`is_per_box_path`] and every assertion in the first two blocks reverses.
    #[test]
    fn the_audit_trail_is_framed_and_the_twins_own_trail_is_not() {
        for p in [
            "log/audit-chain.jsonl",
            ".serialdat/log/audit-chain.jsonl",
            "serialdat/log/audit-chain.jsonl",
            "./log/audit-chain.jsonl",
        ] {
            assert!(
                !is_per_box_path(p),
                "{p} is the audit trail and travels with the log"
            );
            assert!(
                !never_framed(p),
                "{p} must be framable, or a restore loses the audit trail"
            );
        }
        assert!(
            FRAMED_FILES.contains(&AUDIT_REL),
            "a snapshot must carry the audit trail"
        );
        // The twin's own decisions are its own: per box, never framed, never sent.
        assert!(is_per_box_path(AUDIT_LOCAL_REL) && never_framed(AUDIT_LOCAL_REL));
        assert!(!FRAMED_FILES.contains(&AUDIT_LOCAL_REL));
    }

    /// **The site's hub declaration travels; the machine's own identity never does**
    /// (lane T11, 2026-09-21 — the owner's "config must be saved in twin too").
    ///
    /// `wg/hub.conf` is public site data the installer lays into the data root and that
    /// nothing in gunnar writes, so it had no sink call site and the twin never held it. A
    /// restored main without it is a box nobody can reach. `wg/identity.key` is the opposite
    /// case in every respect: a private key, minted on the machine, naming THIS box to the
    /// hub — it must never reach the twin and must never come back on a restore.
    ///
    /// **Fails before the fix**: drop `|| rel == HUB_CONF_REL` from the early return in
    /// [`is_per_box_path`] and the first block reverses; drop the `matches!(rel,
    /// WG_IDENTITY_KEY_REL | …)` block and the second does.
    #[test]
    fn the_hub_declaration_travels_and_the_machines_own_identity_never_does() {
        for p in [HUB_CONF_REL, ".serialdat/wg/hub.conf", "./wg/hub.conf"] {
            assert!(
                !never_framed(p),
                "{p} must travel: a restored main without it is unreachable"
            );
        }
        assert!(
            FRAMED_FILES.contains(&HUB_CONF_REL),
            "a snapshot must carry the hub declaration"
        );
        for p in [
            WG_IDENTITY_KEY_REL,
            WG_IDENTITY_PUB_REL,
            HUB_CONF_FRAMED_REL,
        ] {
            assert!(
                is_per_box_path(p) && never_framed(p),
                "{p} is this machine's own and never travels"
            );
            assert!(!FRAMED_FILES.contains(&p), "{p} must not be in a snapshot");
        }
    }

    /// **A released log holds no descriptor under its directory** (growbench R5, 2026-09-15:
    /// the open segment kept `/data` busy and every runtime growth aborted with EBUSY), refuses
    /// every write by name, and after `reacquire` appends where it left off — on the live
    /// handle and across a reopen.
    #[test]
    #[cfg(target_os = "linux")]
    fn a_released_log_holds_nothing_under_its_dir_refuses_writes_and_resumes_after_reacquire() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("txlog");
        let log = Log::open(&dir).unwrap();
        let holds = |dir: &Path| -> Vec<PathBuf> {
            fs::read_dir("/proc/self/fd")
                .unwrap()
                .filter_map(|e| fs::read_link(e.ok()?.path()).ok())
                .filter(|p| p.starts_with(dir))
                .collect()
        };
        log.append(Kind::Kek, &Meta::Kek {}, b"one").unwrap();
        log.append(Kind::Kek, &Meta::Kek {}, b"two").unwrap();
        assert!(!holds(&dir).is_empty(), "the writer holds the open segment");
        log.release().unwrap();
        log.release().unwrap();
        assert!(log.is_released());
        assert_eq!(
            holds(&dir),
            Vec::<PathBuf>::new(),
            "released: nothing under the dir"
        );
        assert!(matches!(
            log.append(Kind::Kek, &Meta::Kek {}, b"no"),
            Err(Error::Released(_))
        ));
        assert!(matches!(
            log.begin(None, 1, b"", 1),
            Err(Error::Released(_))
        ));
        assert_eq!(log.read(2).unwrap().blob, b"two", "readers open per call");
        log.reacquire().unwrap();
        log.reacquire().unwrap();
        assert_eq!(log.append(Kind::Kek, &Meta::Kek {}, b"three").unwrap(), 3);
        drop(log);
        let back = Log::open(&dir).unwrap();
        assert_eq!(back.head_seq(), 3);
        assert_eq!(back.read(3).unwrap().blob, b"three");
    }

    /// A segment that moved while released is a different disk: refused, not appended to.
    #[test]
    fn reacquire_refuses_a_segment_that_changed_length() {
        let root = tempfile::tempdir().unwrap();
        let dir = root.path().join("txlog");
        let log = Log::open(&dir).unwrap();
        log.append(Kind::Kek, &Meta::Kek {}, b"one").unwrap();
        log.release().unwrap();
        OpenOptions::new()
            .append(true)
            .open(seg_path(&dir, 1))
            .unwrap()
            .write_all(b"junk")
            .unwrap();
        assert!(matches!(log.reacquire(), Err(Error::Corrupt { .. })));
        assert!(log.is_released(), "still refusing writes");
    }

    #[test]
    fn the_sink_frames_writes_under_a_registered_root_and_nothing_elsewhere() {
        let root = tempfile::tempdir().unwrap();
        let other = tempfile::tempdir().unwrap();
        let log = Arc::new(Log::open(root.path().join("txlog")).unwrap());
        sink::register(root.path(), &log);
        drop(sink::append_line(&root.path().join("attestations.jsonl"), "{\"a\":1}").unwrap());
        drop(sink::replace_file(&root.path().join("webhooks.json"), b"[]", false).unwrap());
        drop(sink::append_line(&other.path().join("attestations.jsonl"), "{}").unwrap());
        assert_eq!(log.head_seq(), 2);
        assert_eq!(
            log.read(1).unwrap().meta().unwrap(),
            Meta::Append {
                log: "attestations.jsonl".into()
            }
        );
        assert_eq!(
            log.read(2).unwrap().meta().unwrap(),
            Meta::File {
                path: "webhooks.json".into(),
                deleted: false
            }
        );
        drop(log);
        // A dropped log unregisters itself: nothing to frame into, no error.
        drop(sink::append_line(&root.path().join("attestations.jsonl"), "{}").unwrap());
    }

    #[test]
    fn restart_after_leaves_an_empty_log_whose_next_number_follows_the_snapshot_and_survives_a_reopen(
    ) {
        let dir = tempfile::tempdir().unwrap();
        {
            let log = Log::open(dir.path()).unwrap();
            log.append(Kind::Pack, &meta("a/b"), b"x").unwrap();
            log.restart_after(41).unwrap();
            assert_eq!((log.first_seq(), log.head_seq()), (42, 41));
            assert!(matches!(
                log.append_at(43, Kind::Pack as u16, b"{}", b""),
                Err(Error::OutOfOrder { .. })
            ));
            assert_eq!(
                log.append_at(42, Kind::Pack as u16, &meta("a/b").to_bytes(), b"y")
                    .unwrap(),
                42
            );
        }
        let log = Log::open(dir.path()).unwrap();
        assert_eq!((log.first_seq(), log.head_seq()), (42, 42));
        assert_eq!(log.read(42).unwrap().blob, b"y");
        assert!(matches!(log.read(1), Err(Error::NotRetained { .. })));
    }

    #[test]
    fn a_freeze_waits_for_a_mutation_in_flight_and_holds_new_ones_until_it_ends() {
        use std::sync::atomic::AtomicU32;
        let dir = tempfile::tempdir().unwrap();
        let log = Arc::new(Log::open(dir.path()).unwrap());
        let stage = Arc::new(AtomicU32::new(0));
        // A mutation in flight on another thread: frame appended, store write not yet done.
        let (l, s) = (Arc::clone(&log), Arc::clone(&stage));
        let (go_tx, go_rx) = std::sync::mpsc::channel::<()>();
        let (in_tx, in_rx) = std::sync::mpsc::channel::<()>();
        let writer = std::thread::spawn(move || {
            let g = l.mutation();
            l.append(Kind::Pack, &meta("a/b"), b"1").unwrap();
            in_tx.send(()).unwrap();
            go_rx.recv().unwrap();
            s.store(1, Ordering::SeqCst); // "the store has the write"
            drop(g);
        });
        in_rx.recv().unwrap();
        let (l, s) = (Arc::clone(&log), Arc::clone(&stage));
        let (frozen_tx, frozen_rx) = std::sync::mpsc::channel::<()>();
        let freezer = std::thread::spawn(move || {
            let f = l.freeze();
            frozen_tx.send(()).unwrap();
            // The freeze returned: the write in flight reached its store first.
            assert_eq!(
                s.load(Ordering::SeqCst),
                1,
                "freeze returned before the mutation in flight ended"
            );
            let head = l.head_seq();
            std::thread::sleep(std::time::Duration::from_millis(50));
            assert_eq!(l.head_seq(), head, "no mutation started while frozen");
            s.store(2, Ordering::SeqCst);
            drop(f);
        });
        std::thread::sleep(std::time::Duration::from_millis(30));
        go_tx.send(()).unwrap();
        writer.join().unwrap();
        // A new mutation on this thread, started while frozen, waits for the freeze to end.
        frozen_rx.recv().unwrap();
        let g = log.mutation();
        assert_eq!(
            stage.load(Ordering::SeqCst),
            2,
            "a mutation started while the log was frozen"
        );
        log.append(Kind::Pack, &meta("a/b"), b"2").unwrap();
        drop(g);
        freezer.join().unwrap();
        // Nested guards on one thread count once, and the freezing thread passes its own gate.
        let outer = log.mutation();
        let inner = log.mutation();
        drop(inner);
        drop(outer);
        let f = log.freeze();
        let own = log.mutation();
        drop(own);
        drop(f);
    }
}

#[cfg(test)]
mod spool_tests {
    //! **The log's directory holds one thing besides segments, and the log has
    //! to know it.**

    use super::*;

    /// **MEASURED 2026-09-17 on the live appliance, as a kernel panic.**
    ///
    /// `gunnar-server`'s `txlog_lfs` spools an arriving LFS object beside the
    /// log — `<txlog>/spool/<oid>.part` — and removes the FILE when the object
    /// lands. Nothing ever removes the DIRECTORY, so `create_dir_all` leaves it
    /// there for ever after the first LFS upload the box ever takes.
    ///
    /// [`Log::open_with`] then refused it: every entry that is not a segment
    /// name was `Error::Unexpected`. So one LFS push plus one restart was a
    /// box that could never open its transaction log again, and since gunnar is
    /// PID 1 the exit came back as
    /// `Kernel panic - not syncing: Attempted to kill init!`.
    ///
    /// It hid for as long as it did because it needs BOTH halves: the appliance
    /// had taken LFS objects and simply had not restarted since.
    ///
    /// The spool is part of this directory's layout, so the layout is what
    /// names it — one constant, used by the writer and skipped by the reader,
    /// rather than the reader guessing at a name the writer chose.
    #[test]
    fn the_lfs_spool_directory_is_part_of_the_layout_and_does_not_refuse_the_log() {
        let dir = tempfile::tempdir().expect("temp dir");
        let log = Log::open(dir.path()).expect("a fresh log opens");
        drop(log);
        std::fs::create_dir_all(dir.path().join(SPOOL_DIR)).expect("the spool dir");
        Log::open(dir.path()).expect("a log whose spool directory exists still opens");
    }

    /// The skip is for the ONE name the layout describes, and nothing else: a
    /// stray file in the log's directory is still a refusal, because the
    /// alternative is a reader that shrugs at corruption.
    #[test]
    fn a_stray_entry_is_still_refused_by_name() {
        let dir = tempfile::tempdir().expect("temp dir");
        Log::open(dir.path()).expect("a fresh log opens");
        std::fs::write(dir.path().join("stray"), b"x").expect("write");
        let err = Log::open(dir.path()).expect_err("a stray file is not part of the layout");
        assert!(matches!(err, Error::Unexpected(_)), "{err}");
    }
}