crates/gunnar-core/src/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,
/// 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}");
}
}