nordisk/gunnar
crates/gunnar-server/src/twin.rs
//! **The twin**: the box that holds every transaction the primary was told, in
//! order, durably, and is never promoted.
//! `.nornir/twin-transaction-log.md` §"The twin's loop".
//!
//! **The twin only stores** (OWNER RULING 2026-09-22: "the twin shouldnt do
//! indexing or processing, only receive and save data … if the main is
//! recharged from the twin it could do the indexing then, at recharge time").
//! A twin appends every frame to its own log — the primary's numbers, the raw
//! pack bytes as pushed — and applies ONLY the control plane it itself runs on
//! ([`twin_applies`]: the KEK, control blobs and documents, sealed secrets —
//! the identity table that gates who may read it, and its host key). Packs,
//! refs, LFS, repository stores, append-only lines and framed files stay in
//! the log as bytes: no pack index, no oid tail, no commit graph, no explode
//! cache, no LFS store is ever built on a twin. They are built on the NEW main
//! when it is recharged from the twin ([`restore`], `gunnar control twin
//! replay`), by the same [`apply`] a primary's restore always ran.
//!
//! Three things live here:
//!
//! * [`stream_range`] — the primary's half of `Mirror.LogFetch`: entries
//! `from..=to`, bounded by bytes, never fewer than one, each read off the
//! segment in pieces and handed on as it is read (twin-streaming,
//! 2026-10-06); [`fetch_range`] is the same range gathered, for callers that
//! want whole entries.
//! * **One entry is never held whole on either box.** The batch budget
//! ([`DEFAULT_FETCH_BYTES`]) bounds how many entries one fetch carries, never
//! the size of one: a 512 MiB pack used to be read whole on the primary, cut
//! into a second whole copy of 1 MiB messages, and glued back into one `Vec`
//! on the twin before it was appended. Now the primary reads it a piece at a
//! time ([`Log::read_pieces`], the payload crc checked before the LAST piece
//! leaves), and the twin writes each piece into its own log as it arrives
//! ([`FrameSink`] into a [`txlog::Pending`]): the frame exists only once its
//! last byte is in, and a cut, short or refused transfer leaves nothing.
//! * [`Follower`] — the loop over a [`LogSource`]: head, the gap rule, fetch,
//! `append_at` into this box's OWN log (the primary's numbers, durable) and
//! only then [`take`]: on a twin that keeps the entry (store-only), on a box
//! being restored it is the full [`apply`]. A restart resumes at the own
//! head; every apply is idempotent or a replayed CAS, so re-applying the
//! tail after a crash is the same as applying it.
//! * [`restore`] — replay a twin's log into a fresh data root: the recharge,
//! where every derived index is built.
//!
//! **The gap rule is the loop's shape.** The twin never appends `seq` unless
//! it holds `seq - 1`; `Log::append_at` refuses anything else. When the
//! primary no longer retains `own_head + 1`, the twin stops and says
//! `snapshot-needed` — a twin that kept going past a hole would be a copy that
//! looks complete and is not.
//!
//! **Through the unlogged stores.** The twin's `ServerStores` carry the same
//! tee a primary's do; applying through it would write a SECOND log with the
//! twin's own numbers. So `apply` goes through `raw_control_store`,
//! `raw_secrets` and `OpenRepo::unlogged`, and the twin's `<data-root>/txlog`
//! is exactly the primary's log, number for number.
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use anyhow::{anyhow, Context};
use gunnar_core::control::BlobId;
use gunnar_core::txlog::{self, Entry, Kind, Log, Meta, RefEdit};
use gunnar_core::RepoRef;
use gunnar_store::git_storage_trait::{RefCas, RefRejection, RefUpdate};
use crate::stores::ServerStores;
/// `Mirror.LogFetch`'s ceiling when the request says 0.
pub const DEFAULT_FETCH_BYTES: u64 = 64 << 20;
/// **Entries one [`fetch_range`] may answer with, whatever the byte ceiling buys.**
/// The byte ceiling alone is not a bound on the WORK: a range of tiny frames costs one
/// `log.read` and one `Vec` slot each on the primary, and the gRPC layer turns each into
/// at least one `LogChunk`. The twin's loop asks again the moment it is level, so a cap
/// here costs a round trip and nothing else.
pub const MAX_FETCH_ENTRIES: u64 = 10_000;
/// **A twin's group commit** (perf twin-apply-fsync, 2026-10-06): at most this many entries are
/// written into its log between two `fdatasync`s. MEASURED before (oden, `tests/twin_apply_io.rs`):
/// 9.6 writes and 1.85 fsyncs per entry — one fsync per frame in the log and one per redb commit
/// — which on the UpCloud mockup's HDD-tier twin was ~2,800 write IOPS and a lag of ~93,000
/// entries under a storm of small pushes.
pub const GROUP_ENTRIES: usize = 2048;
/// …or this many bytes of frame.
pub const GROUP_BYTES: u64 = 32 << 20;
/// …or this long since the group's first entry: what a twin acknowledges (its durable head) is
/// never older than this while entries keep arriving.
pub const GROUP_AGE: Duration = Duration::from_millis(500);
/// The applied marker is moved at least this often even over a run with no control-plane write,
/// so a restart re-takes at most this many entries.
pub const MARK_EVERY: u64 = 4096;
/// What `Mirror.LogHead` says about a twin.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TwinStatus {
/// The twin's own head — every entry up to here is in its log AND its
/// stores.
pub applied_seq: u64,
/// The primary's head as of the last successful `LogHead`.
pub primary_head_seq: u64,
/// One word.
pub state: State,
/// Why, when not `Following`.
pub why: String,
/// Wall clock of the last `LogHead` the primary answered; 0 = never.
pub last_answer_unix_ms: i64,
/// **Wall clock of the last answer at which this twin was LEVEL** with its
/// primary — `following` and `applied >= primary head` (2026-09-21, lane T4,
/// the restore guard). 0 = never.
///
/// Owned by [`set_status`]: whatever a caller passes is replaced, because
/// this is a fact about the HISTORY of statuses — carried forward through
/// every `unreachable`, `stalled` and `diverged` — and not something any
/// one status knows by itself. Equal to `last_answer_unix_ms` exactly when
/// the twin was level at its last word from the primary, which is the one
/// fact a restore from a twin whose primary is dead has to be able to see.
pub last_level_unix_ms: i64,
}
impl TwinStatus {
/// The watermark: everything the primary has, the twin has applied.
pub fn caught_up(&self) -> bool {
self.state == State::Following && self.applied_seq == self.primary_head_seq
}
}
/// The twin's state word.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum State {
/// Polling, fetching, applying.
Following,
/// The primary no longer retains what the twin needs next.
SnapshotNeeded,
/// The last poll could not reach the primary.
Unreachable,
/// **The poll got as far as the data and then failed** — the fetch died
/// mid-range, the append would not go on the volume, an entry would not
/// apply. The primary is not the problem; this copy stops where it stopped.
Stalled,
/// **The box at the other end is not the log this twin has been following.**
/// Its head is BELOW what this twin has already applied, and a log's head
/// only ever goes forward — so it is a re-imaged primary, a restore from
/// somewhere else, or an address that now points at a different set.
/// Stopped: an operator decides.
Diverged,
/// **This twin is growing its own data set** (T21, 2026-09-22, owner ruling:
/// the twin grows EXACTLY like main — no restart). The grow machine has
/// drained it: the follower applies nothing and the set may be unmounted.
/// It still asks its primary for the head, so both ends see the word and
/// the lag; it follows again the moment the set is mounted back.
Maintenance,
/// ★ **This twin is owed a growth before it may follow** (cov-13, 2026-10-06):
/// main's data set grew while it was silent, and its volume cannot hold it —
/// the primary refuses its fetches by name (`gunnar_grpc::TWIN_TOO_SMALL`)
/// rather than let it fill part-way through main's data. It keeps polling
/// (so it reads heard, and monetize's twin lane grows it) and follows the
/// moment its report holds main's set. Nothing is appended meanwhile.
Recovering,
/// ★ **This twin is taking a full copy (a snapshot) from its primary** (twin-resync-space,
/// 2026-10-06): it fell behind a hole main's retention trimmed. Its old copy is kept while
/// the volume holds both; when it does not, the twin GIVES IT UP (`snapshot::DROPPED`) and the
/// `why` says it holds NO copy — it serves nothing then, and a crash restarts the resync.
Resyncing,
}
impl std::fmt::Display for State {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(match self {
State::Following => "following",
State::SnapshotNeeded => "snapshot-needed",
State::Unreachable => "unreachable",
State::Stalled => "stalled",
State::Diverged => "diverged",
State::Maintenance => "maintenance",
State::Recovering => "recovering",
State::Resyncing => "resyncing",
})
}
}
/// **The twin status, per DATA ROOT** (T17, 2026-09-21). It was one process-wide slot: a box
/// runs one follower, so one slot was enough in production — but every test that ran a
/// follower wrote the same slot, and `set_status` carries `last_level_unix_ms` forward from
/// whatever was there before. Tests in parallel read each other's state and inherited each
/// other's level time (T15: `a_twin_keeps_the_primarys_last_head_and_time_when_the_primary_stops_answering`
/// failed once in a full run, 5/5 alone). Keyed by the root the follower's stores were opened
/// at, each box — and each test — has its own.
static STATUS: OnceLock<Mutex<HashMap<PathBuf, TwinStatus>>> = OnceLock::new();
fn status_slots() -> std::sync::MutexGuard<'static, HashMap<PathBuf, TwinStatus>> {
STATUS
.get_or_init(|| Mutex::new(HashMap::new()))
.lock()
.unwrap_or_else(|p| p.into_inner())
}
/// The twin status of the box whose data root is `root`, once its follower has polled.
pub fn status(root: &Path) -> Option<TwinStatus> {
status_slots().get(root).cloned()
}
/// Record a status, carrying [`TwinStatus::last_level_unix_ms`] forward: it moves
/// only on a `following` answer at which the twin was level, and survives every
/// other state untouched.
pub(crate) fn set_status(root: &Path, mut s: TwinStatus) {
let mut slots = status_slots();
let before = slots.get(root).map(|p| p.last_level_unix_ms).unwrap_or(0);
s.last_level_unix_ms = if s.state == State::Following
&& s.last_answer_unix_ms > 0
&& s.applied_seq >= s.primary_head_seq
{
s.last_answer_unix_ms
} else {
before
};
slots.insert(root.to_path_buf(), s);
}
// ── the primary's half ──────────────────────────────────────────────────────
/// Entries `from..=to` (`to == 0` = the head now), whole, stopping after the
/// entry that crosses `max_bytes` of TRANSACTION (`0` = [`DEFAULT_FETCH_BYTES`])
/// — never fewer than one. `from` below the oldest retained sequence is
/// [`txlog::Error::NotRetained`]; above the head, [`txlog::Error::BeyondHead`].
///
/// **The ceiling counts meta as well as blob, and the count of entries is capped
/// too** (2026-09-16, lane AUDIT-TWIN). It used to count `blob` alone, and most of
/// a forge's frames carry no blob at all: every `Refs`, `ControlDoc`, `LfsDelete`
/// and `RepoStore` is meta only. `bytes >= cap` was then never reached over such a
/// range, so this read the WHOLE range into one `Vec<Entry>` however small a ceiling
/// the caller asked for — and `service::mirror::chunks_of` built a second whole copy
/// of it beside this one. A twin catching up over a ref-heavy history, or any member
/// of group `twin` asking `LogFetch{from: 1, to: 0, max_bytes: 1}`, was an
/// out-of-memory on the PRIMARY. A `Refs` meta IS the transaction when the blob is
/// empty, so charging for it is what the word "bounded" was always supposed to mean.
pub fn fetch_range(log: &Log, from: u64, to: u64, max_bytes: u64) -> txlog::Result<Vec<Entry>> {
let head = log.head_seq();
let to = if to == 0 { head } else { to.min(head) };
let cap = if max_bytes == 0 {
DEFAULT_FETCH_BYTES
} else {
max_bytes
};
let mut out = Vec::new();
let mut bytes = 0u64;
let mut seq = from;
while seq <= to {
let e = log.read(seq)?;
bytes += (e.blob.len() + e.meta.len()) as u64;
out.push(e);
seq += 1;
if bytes >= cap || out.len() as u64 >= MAX_FETCH_ENTRIES {
break;
}
}
if out.is_empty() && from > head {
// Nothing to give, and not because of retention: say so by the
// same word the reader would.
return Err(txlog::Error::BeyondHead { seq: from, head });
}
Ok(out)
}
/// **The most blob bytes one piece of a streamed entry carries** — the wire's `LogChunk` size,
/// and all a twin or a primary holds of an entry at once.
pub const PIECE_BYTES: usize = 1 << 20;
/// **The refusals [`fetch_range`] makes before it reads anything**, made up front: `from` above
/// the head is [`txlog::Error::BeyondHead`]; a non-empty range starting below the oldest
/// retained sequence is [`txlog::Error::NotRetained`]. What a streamed `Mirror.LogFetch` answers
/// as the call's status, before its first piece, exactly as the gathered one did.
pub fn check_range(log: &Log, from: u64, to: u64) -> txlog::Result<()> {
let head = log.head_seq();
if from > head {
return Err(txlog::Error::BeyondHead { seq: from, head });
}
let to = if to == 0 { head } else { to.min(head) };
let first = log.first_seq();
if from <= to && from < first {
return Err(txlog::Error::NotRetained { seq: from, first });
}
Ok(())
}
/// **[`fetch_range`], streamed** (twin-streaming, 2026-10-06): the same entries under the same
/// bounds — stopping after the entry that crosses `max_bytes` of transaction, or at
/// [`MAX_FETCH_ENTRIES`], never fewer than one — each handed to `into` as it is read off the
/// segment, [`PIECE_BYTES`] at a time ([`Log::read_pieces`]). `into.end()` for an entry is called
/// only after its payload crc checked out, so a reader that makes an entry visible at its end
/// never makes a corrupt one visible. Answers how many entries went out whole.
///
/// What the primary holds is ONE piece, whatever the entry's size; `fetch_range` held the whole
/// range and the gRPC layer a second copy of it.
pub fn stream_range(
log: &Log,
from: u64,
to: u64,
max_bytes: u64,
into: &mut dyn FrameSink,
) -> anyhow::Result<u64> {
stream_range_at(log, from, 0, to, max_bytes, into)
}
/// **[`stream_range`] with the FIRST entry started at byte `from_offset` of its blob**
/// (restore-large-entry, 2026-10-07): a restore resuming an entry the box being filled already
/// holds part of (`LogFetchRequest.from_offset`). Only the blob from the offset is read
/// ([`Log::read_pieces_from`]): a resumed entry of gigabytes costs this box its rest per call,
/// never a re-read of what went out before. Its stored payload crc goes out with the last piece
/// ([`FrameSink::checked`]) and the box that put the entry together checks it — the bytes below
/// the offset are not here to check it over. `into` is told where the entry starts
/// ([`FrameSink::begin_at`]). An offset at the blob's end is one empty last piece. An offset PAST
/// it (or on an erased frame) is not the entry the asker holds part of (one erased since, say):
/// the entry goes out whole from 0, checked, and the asker sees its length differ.
pub fn stream_range_at(
log: &Log,
from: u64,
from_offset: u64,
to: u64,
max_bytes: u64,
into: &mut dyn FrameSink,
) -> anyhow::Result<u64> {
check_range(log, from, to)?;
let head = log.head_seq();
let to = if to == 0 { head } else { to.min(head) };
let cap = if max_bytes == 0 {
DEFAULT_FETCH_BYTES
} else {
max_bytes
};
let (mut bytes, mut sent, mut seq) = (0u64, 0u64, from);
while seq <= to {
let mut begun = false;
let start = if seq == from && from_offset > 0 {
let l = log.locate_blob(seq)?;
if l.kind == Kind::Redacted as u16 || from_offset > l.blob_len {
0
} else {
from_offset
}
} else {
0
};
let at =
log.read_pieces_from::<anyhow::Error>(seq, start, PIECE_BYTES, |l, piece, last| {
if !begun {
into.begin_at(l.seq, l.kind, &l.meta, l.blob_len, start)?;
begun = true;
}
if last {
if let Some(crc) = l.payload_crc {
into.checked(crc)?;
}
}
if !piece.is_empty() {
into.piece(piece)?;
}
if last {
into.end()?;
}
Ok(())
})?;
bytes += at.blob_len + at.meta.len() as u64;
sent += 1;
seq += 1;
if bytes >= cap || sent >= MAX_FETCH_ENTRIES {
break;
}
}
Ok(sent)
}
// ── streamed entries ────────────────────────────────────────────────────────
/// **Where a streamed range goes: entry by entry, piece by piece** (twin-streaming,
/// 2026-10-06). `begin` names an entry and the length its blob will have, `piece` carries the
/// next bytes of it in order, `end` says its last byte arrived and the source vouched for it.
/// An entry that never reaches `end` — the wire broke, the source refused, a check failed — was
/// never an entry: an implementation keeps nothing of it.
pub trait FrameSink {
/// An entry begins.
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()>;
/// The next piece of the open entry's blob.
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()>;
/// The open entry is whole.
fn end(&mut self) -> anyhow::Result<()>;
/// **An entry begins part-way** — its pieces start at byte `offset` of the blob
/// (restore-large-entry, 2026-10-07: [`stream_range_at`]). A sink that can only take whole
/// entries refuses any offset but 0.
fn begin_at(
&mut self,
seq: u64,
kind: u16,
meta: &[u8],
blob_len: u64,
offset: u64,
) -> anyhow::Result<()> {
anyhow::ensure!(
offset == 0,
"seq {seq}: this sink takes whole entries, not one from byte {offset}"
);
self.begin(seq, kind, meta, blob_len)
}
/// The open entry's payload crc32c (meta ‖ blob), checked by the source against what it
/// read, before its last piece: a sink that forwards entries vouches with it.
fn checked(&mut self, _payload_crc: u32) -> anyhow::Result<()> {
Ok(())
}
}
/// **Whole entries out of a stream** — [`LogSource::fetch`] spelled over
/// [`LogSource::fetch_into`], for the callers that want them (tests, small ranges). Holds what it
/// gathers, which is the point of using it only where that is small — and
/// BOUNDED (stinky row 20, 2026-10-10): no entry over [`GATHER_MAX_ENTRY`], no
/// more than [`GATHER_MAX_TOTAL`] of blob in all, and no entry past the length
/// it declared. `reserve_for` bounds only the first allocation; these bound
/// what a peer that keeps sending can make this box hold.
#[derive(Debug)]
pub struct Gather {
/// The entries, in order, once ended.
pub out: Vec<Entry>,
open: Option<Entry>,
/// The open entry's declared blob length.
declared: u64,
/// Blob bytes held, ended entries and the open one.
held: u64,
max_entry: u64,
max_total: u64,
}
impl Default for Gather {
fn default() -> Self {
Gather::bounded(GATHER_MAX_ENTRY, GATHER_MAX_TOTAL)
}
}
/// The largest entry a [`Gather`] holds whole.
pub const GATHER_MAX_ENTRY: u64 = 256 << 20;
/// The most blob bytes a [`Gather`] holds in all.
pub const GATHER_MAX_TOTAL: u64 = 512 << 20;
impl Gather {
/// A gather that holds no entry over `max_entry` bytes and no more than
/// `max_total` bytes of blob in all.
pub fn bounded(max_entry: u64, max_total: u64) -> Gather {
Gather {
out: Vec::new(),
open: None,
declared: 0,
held: 0,
max_entry,
max_total,
}
}
}
impl FrameSink for Gather {
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
anyhow::ensure!(self.open.is_none(), "seq {seq} began inside another entry");
anyhow::ensure!(
blob_len <= self.max_entry,
"seq {seq} declares {blob_len} bytes, more than the {} one gathered entry may hold; it is streamed, not gathered",
self.max_entry
);
anyhow::ensure!(
self.held.saturating_add(blob_len) <= self.max_total,
"seq {seq} ({blob_len} bytes) would take this gather past the {} bytes it may hold in all ({} held)",
self.max_total,
self.held
);
self.declared = blob_len;
self.open = Some(Entry {
seq,
kind,
meta: meta.to_vec(),
blob: Vec::with_capacity(gunnar_proto::reserve_for(blob_len)),
});
Ok(())
}
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
let open = self
.open
.as_mut()
.ok_or_else(|| anyhow!("a piece arrived outside an entry"))?;
let after = open.blob.len() as u64 + bytes.len() as u64;
anyhow::ensure!(
after <= self.declared,
"seq {}: {after} bytes arrived for an entry that declared {}",
open.seq,
self.declared
);
open.blob.extend_from_slice(bytes);
self.held += bytes.len() as u64;
Ok(())
}
fn end(&mut self) -> anyhow::Result<()> {
let e = self
.open
.take()
.ok_or_else(|| anyhow!("an entry ended that never began"))?;
self.out.push(e);
Ok(())
}
}
/// **The `LogChunk` protocol, checked, for every reader of `Mirror.LogFetch`** — the twin's
/// follower and `gunnar control twin follow` read the same stream the same way. One entry's
/// pieces arrive in order with the same `seq`, `kind` and `blob_len`, `meta` on the first; its
/// bytes add up to exactly `blob_len` when `last` ends it (more is refused at the piece that
/// overflows, less at the end); a new `seq` before `last` and a stream that ends inside an
/// entry are both refused. Each piece goes to the [`FrameSink`] as it arrives and is dropped.
#[derive(Debug, Default)]
pub struct ChunkFeed {
/// `(seq, kind, blob_len, bytes so far)` of the entry being received.
open: Option<(u64, u16, u64, u64)>,
}
impl ChunkFeed {
/// One chunk off the wire.
pub fn take(
&mut self,
chunk: &gunnar_proto::LogChunk,
into: &mut dyn FrameSink,
) -> anyhow::Result<()> {
let kind = u16::try_from(chunk.kind)
.map_err(|_| anyhow!("seq {}: kind {} is not a frame kind", chunk.seq, chunk.kind))?;
match self.open {
Some((seq, k, len, _)) if seq == chunk.seq => anyhow::ensure!(
k == kind && len == chunk.blob_len,
"seq {seq}: a later piece says kind {kind} of {} bytes, the first said kind {k} of {len}",
chunk.blob_len
),
Some((seq, ..)) => anyhow::bail!("the primary moved to seq {} with seq {seq} unfinished", chunk.seq),
None => {
into.begin(chunk.seq, kind, &chunk.meta, chunk.blob_len)?;
self.open = Some((chunk.seq, kind, chunk.blob_len, 0));
}
}
let Some((seq, _, len, got)) = self.open.as_mut() else {
unreachable!("an entry is open")
};
*got += chunk.blob.len() as u64;
anyhow::ensure!(
*got <= *len,
"seq {seq}: more than the {len} blob bytes it declared arrived"
);
if !chunk.blob.is_empty() {
into.piece(&chunk.blob)?;
}
if chunk.last {
anyhow::ensure!(*got == *len, "seq {seq}: {got} of {len} blob bytes arrived");
self.open = None;
into.end()?;
}
Ok(())
}
/// The stream ended: refused if it ended inside an entry.
pub fn finish(&self) -> anyhow::Result<()> {
match self.open {
Some((seq, _, len, got)) => Err(anyhow!(
"the stream ended inside seq {seq}, {got} of {len} blob bytes in"
)),
None => Ok(()),
}
}
}
/// **A `Mirror.LogFetch` stream into a [`FrameSink`]**, each chunk handed on as it arrives,
/// none gathered: what both `GrpcSource` and `gunnar control twin follow` run (any stream of
/// chunks — tonic's `Streaming` is one). `idle` bounds the wait for every next chunk (the
/// follower's [`FETCH_IDLE_WITHIN`]); `said` turns the stream's error into the caller's.
pub async fn feed_log_stream<S, E>(
stream: &mut S,
idle: Option<(Duration, &(dyn Fn() -> anyhow::Error + Sync))>,
said: &(dyn Fn(E) -> anyhow::Error + Sync),
into: &mut dyn FrameSink,
) -> anyhow::Result<()>
where
S: tokio_stream::Stream<Item = Result<gunnar_proto::LogChunk, E>> + Unpin,
{
use tokio_stream::StreamExt as _;
let mut feed = ChunkFeed::default();
loop {
let next = match idle {
Some((within, quiet)) => tokio::time::timeout(within, stream.next())
.await
.map_err(|_| quiet())?,
None => stream.next().await,
};
match next.transpose().map_err(said)? {
Some(chunk) => feed.take(&chunk, into)?,
None => return feed.finish(),
}
}
}
// ── the source ──────────────────────────────────────────────────────────────
/// What the twin reads from: `LogHead` and `LogFetch`, over gRPC in the
/// appliance and over a shared `Log` in tests and in `restore`.
pub trait LogSource: Send + Sync {
/// `(first_seq, head_seq)` on the primary.
fn head(&self) -> anyhow::Result<(u64, u64)>;
/// Does the primary have another address to re-dial when this one keeps
/// failing? (The tunnel's and the utility network's.) In-process sources
/// have none.
fn alternatives(&self) -> bool {
false
}
/// **`(first_seq, head_seq, the source's log identity)`** — one call, because
/// the identity has to come from the SAME answer as the numbers it
/// qualifies. Asking twice would leave a window in which the numbers are
/// one box's and the name is another's, which is the bug in a nicer shape.
///
/// The default carries no identity, and that is the truthful answer for
/// every in-process source: a bare `Arc<Log>`, a `LocalSource` over another
/// data root in this process, a test's script. Nothing was fetched over a
/// wire, so there is nothing to have been lied to about. Only
/// [`GrpcSource`] overrides it. See [`LOG_ID`].
fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
let (first, head) = self.head()?;
Ok((first, head, None))
}
/// **[`LogSource::head_and_identity`] and the source's GENERATION**
/// ([`LOG_GENERATION`]), out of the same answer. 0 = none claimed: every
/// in-process source, and a primary never restored or older than the field.
fn head_identity_generation(&self) -> anyhow::Result<(u64, u64, Option<String>, u64)> {
let (first, head, id) = self.head_and_identity()?;
Ok((first, head, id, 0))
}
/// Entries `from..=to`, at most about `max_bytes` of blob, never fewer
/// than one when `from <= to`.
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>>;
/// **[`LogSource::fetch`], streamed into `into`** — what the follower calls (twin-streaming,
/// 2026-10-06). Every source that reads a real log overrides it ([`GrpcSource`] off the wire,
/// a bare [`Arc<Log>`] and [`LocalSource`] off the segments), so an entry is never whole in
/// memory however large it is. The default gathers through `fetch` and hands each blob over
/// as one piece: right for a scripted source, whose entries are already in its hands.
fn fetch_into(
&self,
from: u64,
to: u64,
max_bytes: u64,
into: &mut dyn FrameSink,
) -> anyhow::Result<()> {
for e in self.fetch(from, to, max_bytes)? {
into.begin(e.seq, e.kind, &e.meta, e.blob.len() as u64)?;
if !e.blob.is_empty() {
into.piece(&e.blob)?;
}
into.end()?;
}
Ok(())
}
/// A consistent snapshot of the source's stores, for a twin behind a retention hole, fed
/// piece by piece `into` the twin's staging area as it arrives. `false` = this source
/// cannot make one (a bare log): the twin stops with snapshot-needed.
fn snapshot(&self, _into: &mut snapshot::Stager) -> anyhow::Result<bool> {
Ok(false)
}
}
impl LogSource for Arc<Log> {
fn head(&self) -> anyhow::Result<(u64, u64)> {
Ok((self.first_seq(), self.head_seq()))
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
Ok(fetch_range(self, from, to, max_bytes)?)
}
fn fetch_into(
&self,
from: u64,
to: u64,
max_bytes: u64,
into: &mut dyn FrameSink,
) -> anyhow::Result<()> {
stream_range(self, from, to, max_bytes, into).map(drop)
}
}
// ── apply ───────────────────────────────────────────────────────────────────
/// What one entry did on this box.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Applied {
/// Stored, or already there.
Done,
/// A CAS the primary lost, lost here too — the design's replayed race.
Lost(String),
/// A kind this build does not know: kept in the log, not applied.
Unknown(u16),
/// **Kept in a twin's log and nothing else** (ruling 2026-09-22, "the twin only
/// stores"): a data frame a twin does not apply. The recharge applies it on the new main.
Stored,
}
/// **The frames a twin applies** — the control plane the twin itself runs on, and nothing
/// else (ruling 2026-09-22, "the twin only stores").
///
/// * `Kek` — without it the replicated sealed secrets are unreadable on this box.
/// * `ControlBlob` / `ControlDoc` — the identity table: who may read this twin's log
/// (`Mirror.LogFetch`/`SnapshotFetch`, `twin status`, `twin replay`) is decided against the
/// primary's principals and groups, so the twin must hold them to gate its own door.
/// * `Secret` — the host key the twin answers with, and the rest of the sealed rows (a row
/// write, no derived structure).
///
/// Everything else — `Pack`, `Refs`, `LfsPut`, `LfsDelete`, `RepoStore`, `Append`, `File` —
/// is the DATA a twin saves and does not process: the frame in the log IS the copy. Applying
/// a `Pack` is what builds the archive's oid tail, the pack index and (on read) the commit
/// graph and the explode pool; applying `Refs` needs that tail. None of it exists on a twin.
pub fn twin_applies(kind: u16) -> bool {
// `Erase` too (owner ruling g-nonpay): the twin's log IS the copy, so an account
// deleted on main is erased from it here — the one thing a twin does to data.
// `Activity` too (2026-10-21): the last-seen stamps are control-plane rows, and a twin
// promoted to main answers the admin's user list with them.
matches!(
Kind::from_u16(kind),
Some(
Kind::Kek
| Kind::ControlBlob
| Kind::ControlDoc
| Kind::Secret
| Kind::Erase
| Kind::Activity
)
)
}
/// **What a twin applies**: [`twin_applies`]' kinds, and the identity core's
/// document (`identity.json`, a `File` frame): the core answers from those tables,
/// so a twin that only stored them would answer from the ones it loaded at its
/// boot and drift from its main (RA5-10).
pub(crate) fn twin_follows(kind: u16, meta: &[u8]) -> bool {
twin_applies(kind)
|| (matches!(Kind::from_u16(kind), Some(Kind::File))
&& matches!(Meta::from_bytes(meta), Ok(Meta::File { ref path, .. }) if path == crate::identity::FILE))
}
/// **What a follower does with an entry that is durable in its log.** On a twin: the
/// [`twin_applies`] kinds are applied, every other kind is [`Applied::Stored`] (the log holds
/// it) — an unknown kind is still [`Applied::Unknown`]. On any other box (a fresh box being
/// restored — the recharge): the full [`apply`], which is where the indexes get built.
pub fn take(stores: &ServerStores, e: &Entry) -> anyhow::Result<Applied> {
take_raw(stores, e.seq, e.kind, &e.meta, &e.blob)
}
/// [`take`] over borrowed parts.
pub fn take_raw(
stores: &ServerStores,
seq: u64,
kind: u16,
meta: &[u8],
blob: &[u8],
) -> anyhow::Result<Applied> {
if stores.is_twin() && Kind::from_u16(kind).is_some() && !twin_follows(kind, meta) {
// The meta is parsed — a frame that could never be applied is refused NOW, while the
// primary is alive to be asked again, not discovered at the recharge. The bytes are
// not touched.
Meta::from_bytes(meta)
.with_context(|| format!("seq {seq}: the frame's meta does not parse"))?;
return Ok(Applied::Stored);
}
apply_raw(stores, seq, kind, meta, blob)
}
/// **[`take`] for an entry already durable in this box's log**, reading its blob back only when
/// the take needs it (twin-streaming, 2026-10-06). A twin keeps its data frames without touching
/// their bytes ([`take_raw`] parses the meta and stops), so a 512 MiB pack it just streamed to
/// disk is never read back into memory; the control plane it applies is small, and a box being
/// restored reads each frame whole because [`apply`] needs it whole.
pub fn take_logged(
stores: &ServerStores,
seq: u64,
kind: u16,
meta: &[u8],
) -> anyhow::Result<Applied> {
if stores.is_twin() && !twin_follows(kind, meta) {
return take_raw(stores, seq, kind, meta, &[]);
}
let e = stores.txlog().read(seq)?;
take(stores, &e)
}
fn oid_bytes(hex_text: &str) -> anyhow::Result<Vec<u8>> {
hex::decode(hex_text).map_err(|e| anyhow!("oid {hex_text}: {e}"))
}
fn repo_of(stores: &ServerStores, key: &str) -> anyhow::Result<gunnar_core::Repo> {
let (ns, name) = key
.split_once('/')
.ok_or_else(|| anyhow!("repository key {key:?} is not namespace/name"))?;
let path = RepoRef::new(ns, name).with_context(|| format!("repository key {key:?}"))?;
stores
.catalog()
.get(&path)?
.ok_or_else(|| anyhow!("the frame names {key}, which the catalog does not hold yet — the catalog row must precede the first pack"))
}
/// Apply one entry to `stores` through their UNLOGGED halves.
pub fn apply(stores: &ServerStores, e: &Entry) -> anyhow::Result<Applied> {
apply_raw(stores, e.seq, e.kind, &e.meta, &e.blob)
}
/// [`apply`] over borrowed parts — a snapshot item read back from the twin's staging file
/// (`seq` 0) is never an owned [`Entry`].
pub fn apply_raw(
stores: &ServerStores,
seq: u64,
kind: u16,
meta: &[u8],
blob: &[u8],
) -> anyhow::Result<Applied> {
struct E<'a> {
seq: u64,
blob: &'a [u8],
}
let e = E { seq, blob };
let Some(kind) = Kind::from_u16(kind) else {
// **A FOLLOWING twin keeps it and goes on; a box being RESTORED stops here**
// (2026-09-16, lane AUDIT-TWIN). The two are not the same box:
//
// * a twin's log holds the frame byte for byte and the twin is never promoted, so a
// kind a newer primary invented costs it nothing — the restore that later reads this
// log gets the frame whole;
// * a box being restored stops being a follower at [`restore_finish`] and starts
// ORIGINATING from that head. Skipping the frame there was SILENT: the replay
// answered "restored to seq N (N applied, 0 lost)" and the box came up as a primary
// whose stores were missing whatever those frames carried. Measured 2026-09-16: a log
// with one kind-99 frame restored green.
//
// Refusing it as an apply failure is the resumable shape the restore already has: the
// entry is durable at the head, the stream stops, the box stays restoring, and
// [`restore_begin`] re-applies the head — so re-running the replay under a build that
// knows the kind finishes it instead of stepping over it.
if is_restoring(stores.root()) {
anyhow::bail!(
"seq {seq}: frame kind {kind} is not one this build can apply — the log that is being \
restored was written by a NEWER gunnar. The frame is durable at this box's head and \
nothing is lost; restore again from a build that knows kind {kind}. This box stays \
restoring and originates nothing."
);
}
return Ok(Applied::Unknown(kind));
};
// An erased frame's data is what was deleted: nothing to apply. Its meta is the
// original frame's, so it is not matched below.
if kind == Kind::Redacted {
return Ok(Applied::Done);
}
let meta = Meta::from_bytes(meta).with_context(|| format!("seq {seq}"))?;
match (kind, meta) {
// **An account's data was deleted** (g-nonpay): every frame of these repositories
// before this one is erased from THIS box's log — the twin's copy, or the log a
// restore is rebuilding. The stores themselves went with their `RepoStore` frames.
(Kind::Erase, Meta::Erase { repos, .. }) => {
stores.txlog().erase(seq, &repos).with_context(|| {
format!(
"seq {seq}: erasing {} repositor(ies) from the log",
repos.len()
)
})?;
Ok(Applied::Done)
}
(Kind::Pack, Meta::Pack { repo, .. }) => {
let r = repo_of(stores, &repo)?;
let store = stores.repo_store(&r)?;
store
.unlogged()
.put_pack(e.blob)
.with_context(|| format!("seq {}: pack into {repo}", e.seq))?;
Ok(Applied::Done)
}
(
Kind::Refs,
Meta::Refs {
repo,
atomic,
edits,
},
) => {
let r = repo_of(stores, &repo)?;
let store = stores.repo_store(&r)?;
let git = store.unlogged();
if atomic {
let olds: Vec<Option<Vec<u8>>> = edits
.iter()
.map(|ed| ed.old.clone().flatten().map(|h| oid_bytes(&h)).transpose())
.collect::<anyhow::Result<_>>()?;
let news: Vec<Option<Vec<u8>>> = edits
.iter()
.map(|ed| ed.new.as_deref().map(oid_bytes).transpose())
.collect::<anyhow::Result<_>>()?;
let peeled: Vec<Option<Vec<u8>>> = edits
.iter()
.map(|ed| ed.peeled.as_deref().map(oid_bytes).transpose())
.collect::<anyhow::Result<_>>()?;
let cas: Vec<RefCas<'_>> = edits
.iter()
.enumerate()
.map(|(i, ed)| RefCas {
name: ed.name.clone(),
old: olds[i].as_deref(),
new: news[i].as_deref(),
peeled: peeled[i].clone(),
})
.collect();
match git.put_refs_cas(&cas) {
Ok(_) => Ok(Applied::Done),
Err(err) => match err.downcast_ref::<RefRejection>() {
Some(rej) => Ok(Applied::Lost(format!("{rej:?}"))),
None => {
Err(err).with_context(|| format!("seq {}: refs into {repo}", e.seq))
}
},
}
} else {
let mut plain: Vec<RefUpdate> = Vec::new();
for ed in &edits {
if ed.name == "HEAD" {
if let Some(target) = &ed.symref {
git.set_head(target)
.with_context(|| format!("seq {}: HEAD of {repo}", e.seq))?;
continue;
}
}
plain.push(update_of_edit(ed));
}
if !plain.is_empty() {
git.put_refs(&plain)
.with_context(|| format!("seq {}: refs into {repo}", e.seq))?;
}
Ok(Applied::Done)
}
}
(Kind::ControlBlob, Meta::ControlBlob { id }) => {
let got = stores.raw_control_store().put_blob(e.blob)?;
let want = BlobId::from_hex(&id)
.ok_or_else(|| anyhow!("seq {}: blob id {id:?} is not hex", e.seq))?;
if got != want {
return Err(anyhow!(
"seq {}: the blob hashes to {} but the frame says {id}",
e.seq,
hex::encode(got.as_bytes())
));
}
Ok(Applied::Done)
}
(
Kind::ControlDoc,
Meta::ControlDoc {
name,
expected,
new,
},
) => {
let mut expected = expected
.as_deref()
.map(|h| {
BlobId::from_hex(h)
.ok_or_else(|| anyhow!("seq {}: expected {h:?} is not hex", e.seq))
})
.transpose()?;
let new = BlobId::from_hex(&new)
.ok_or_else(|| anyhow!("seq {}: new {new:?} is not hex", e.seq))?;
if expected.is_none() && fresh_take(stores.root(), &name)? {
// The fresh-box rule: this box wrote `name` itself before it
// was reset; the primary's create is a create into empty.
expected = stores.raw_control_store().read_doc(&name)?;
}
if stores
.raw_control_store()
.commit_doc(&name, expected, new)?
{
Ok(Applied::Done)
} else {
Ok(Applied::Lost(format!("document {name}")))
}
}
(Kind::Secret, Meta::Secret { key, deleted }) => {
use gunnar_store::SecretStore as _;
if deleted {
stores.raw_secrets().delete(&key)?;
} else {
stores.raw_secrets().put(&key, e.blob)?;
}
Ok(Applied::Done)
}
(Kind::LfsPut, Meta::Lfs { repo, oid, size }) => {
use gunnar_lfs::LfsStore as _;
use std::io::Write as _;
let r = repo_of(stores, &repo)?;
let store = stores.repo_store(&r)?;
let lfs = gunnar_lfs::FsLfsStore::open_or_init(store.git_dir())?;
let id = gunnar_lfs::Oid::parse(&oid)
.map_err(|err| anyhow!("seq {}: {oid}: {err}", e.seq))?;
if lfs.size_of(&id)?.is_some() {
return Ok(Applied::Done);
}
let mut w = lfs.begin_write(&id, size)?;
w.write_all(e.blob)?;
w.commit()?;
Ok(Applied::Done)
}
(Kind::Activity, Meta::Activity {}) => {
let rows = gunnar_core::activity::decode_batch(e.blob)
.with_context(|| format!("seq {}: last-seen stamps", e.seq))?;
stores
.activity()
.absorb(rows)
.with_context(|| format!("seq {}: storing last-seen stamps", e.seq))?;
Ok(Applied::Done)
}
(Kind::Kek, Meta::Kek {}) => {
write_kek(stores.control_dir(), e.blob)
.with_context(|| format!("seq {}: the KEK", e.seq))?;
stores
.adopt_kek(e.blob)
.with_context(|| format!("seq {}: holding the KEK", e.seq))?;
Ok(Applied::Done)
}
(Kind::Append, Meta::Append { log }) => {
let path = under_root(stores.root(), &log)?;
append_line_once(&path, e.blob)
.with_context(|| format!("seq {}: a line of {log}", e.seq))?;
Ok(Applied::Done)
}
(Kind::File, Meta::File { path, deleted }) => {
let full = under_root(stores.root(), &path)?;
if deleted {
match std::fs::remove_file(&full) {
Ok(()) => {}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
Err(err) => {
return Err(err).with_context(|| format!("seq {}: removing {path}", e.seq))
}
}
} else {
let dir = full.parent().expect("under the root");
std::fs::create_dir_all(dir)?;
let name = full
.file_name()
.expect("a file")
.to_string_lossy()
.into_owned();
crate::durable::write_atomic_unframed(dir, &name, e.blob)
.with_context(|| format!("seq {}: {path}", e.seq))?;
}
// The identity core holds its tables in memory: the main's document is read
// again, so the twin's identity state follows the main (RA5-10).
if path == crate::identity::FILE {
if let Some(identity) = stores.identity() {
crate::identity::follow(stores.root(), &identity)
.map_err(|why| anyhow!("seq {}: {why}", e.seq))?;
}
}
Ok(Applied::Done)
}
(Kind::LfsDelete, Meta::Lfs { repo, oid, .. }) => {
use gunnar_lfs::LfsStore as _;
let (ns, name) = repo.split_once('/').ok_or_else(|| {
anyhow!(
"seq {}: repository key {repo:?} is not namespace/name",
e.seq
)
})?;
let dir = stores.repo_dir(&RepoRef::new(ns, name)?)?;
// No store, no object: the delete is already true here.
if !dir.exists() {
return Ok(Applied::Done);
}
let id = gunnar_lfs::Oid::parse(&oid)
.map_err(|err| anyhow!("seq {}: {oid}: {err}", e.seq))?;
gunnar_lfs::FsLfsStore::open_or_init(&dir)?
.delete(&id)
.with_context(|| format!("seq {}: deleting LFS {oid} of {repo}", e.seq))?;
Ok(Applied::Done)
}
(Kind::RepoStore, Meta::RepoStore { repo, op, .. }) => {
let (ns, name) = repo.split_once('/').ok_or_else(|| {
anyhow!(
"seq {}: repository key {repo:?} is not namespace/name",
e.seq
)
})?;
let path = RepoRef::new(ns, name)?;
match op.as_str() {
"purged" => {
stores
.reclaim_repo(&path)
.with_context(|| format!("seq {}: purging the store of {repo}", e.seq))?;
}
// The store opens on its first pack; a create frame, when one exists, opens it now.
"created" => {
if let Some(r) = stores.catalog().get(&path)? {
stores.repo_store(&r)?;
}
}
other => {
return Err(anyhow!(
"seq {}: RepoStore op {other:?} is neither created nor purged",
e.seq
))
}
}
Ok(Applied::Done)
}
(kind, meta) => Err(anyhow!(
"seq {}: kind {kind:?} with meta {meta:?} do not belong together",
e.seq
)),
}
}
fn update_of_edit(ed: &RefEdit) -> RefUpdate {
RefUpdate {
name: ed.name.clone(),
target: ed.new.clone(),
peeled: ed.peeled.clone(),
symref_target: ed.symref.clone(),
}
}
// ── the box's role and the fresh-box rule ───────────────────────────────────
/// `<data-root>/twin.json`: this box is a twin of `primary`. Per box, never
/// framed (a twin's own facts are not the primary's transactions).
pub const TWIN_CONFIG: &str = "twin.json";
/// `<data-root>/twin.key`: the twin's own ed25519 key, minted at join — the
/// credential it presents to the primary's `Mirror.LogFetch`. Per box, never
/// framed; its public half is enrolled on the primary in group [`TWIN_GROUP`].
pub const TWIN_KEY: &str = "twin.key";
/// The group whose members may read the primary's log without being admins.
pub const TWIN_GROUP: &str = "twin";
/// `<data-root>/twin-owed.json`, on a PRIMARY: what each twin is owed — a full resync
/// after retention trimmed past it while it was silent, a growth after main's set grew
/// while it was silent (`gunnar_grpc::twin_owed`, cov-13). Per box, never framed.
pub const TWIN_OWED_FILE: &str = "twin-owed.json";
/// `<data-root>/txlog-fresh-docs.json`: the documents a fresh box wrote itself
/// before it was reset to take another log's numbers (see [`prepare_replay`]).
pub const FRESH_DOCS: &str = "txlog-fresh-docs.json";
/// `<data-root>/txlog-restoring`: a restore is in progress; the log stays
/// follower-only across a restart until [`restore_finish`].
pub const RESTORING: &str = "txlog-restoring";
/// What `twin.json` says.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TwinConfig {
/// The primary's control plane, `host:port`.
pub primary: String,
/// When this box became a twin.
pub joined_at_unix_ms: i64,
}
impl TwinConfig {
/// `Ok(None)` = not a twin. A `twin.json` that does not parse is an error,
/// never "not a twin": a twin that booted as a primary would frame its own
/// KEK and fork the numbers.
pub fn load(root: &Path) -> anyhow::Result<Option<Self>> {
match std::fs::read(root.join(TWIN_CONFIG)) {
Ok(bytes) => Ok(Some(serde_json::from_slice(&bytes).with_context(|| {
format!("{} does not parse", root.join(TWIN_CONFIG).display())
})?)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => {
Err(e).with_context(|| format!("reading {}", root.join(TWIN_CONFIG).display()))
}
}
}
/// Write it (unframed, atomic).
pub fn save(&self, root: &Path) -> anyhow::Result<()> {
crate::durable::write_atomic_unframed(root, TWIN_CONFIG, &serde_json::to_vec_pretty(self)?)
}
}
/// Does `root` belong to a twin? Existence decides; [`TwinConfig::load`] parses.
pub fn is_twin_root(root: &Path) -> bool {
root.join(TWIN_CONFIG).exists()
}
/// `<data>/pairing/recharge_from`: this box was re-admitted as a recharged
/// main and has not been restored yet — what `Bootstrap.Status.recharge_from`
/// answers, so the restore lane (monetize's) knows the box awaits its twin.
///
/// Declared HERE and re-exported by `pairing`: `stores` reads it in every build, and
/// `pairing` is compiled only with `wg-client` (2026-10-14: `--features
/// server,engine-znippy,metered` did not compile — "cannot find `pairing` in `crate`").
pub const RECHARGE_REL: &str = "pairing/recharge_from";
/// Is a restore in progress under `root`?
pub fn is_restoring(root: &Path) -> bool {
root.join(RESTORING).exists()
}
// ── the log's identity ──────────────────────────────────────────────────────
/// `<data-root>/txlog-id`: **which log this is.** 128 random bits, hex, minted
/// once at a primary's first open and never derived from anything.
///
/// # Why a log needed a name at all
///
/// Until 2026-09-21 (lane T4) NOTHING on the wire said which log a
/// `Mirror.LogHead` answer was about. `LogHeadResponse` carried nine fields —
/// head, first, bytes, twin, applied, primary head, caught up, state, why —
/// and every one of them is a NUMBER ABOUT a log, not a name FOR one. A twin
/// compared sequence numbers against an address and called that following.
///
/// The lineage check added with `State::Diverged` catches a primary whose head
/// is BELOW the twin's, which is a re-imaged main caught early. It cannot catch
/// the same main once it has written past the twin's head: a fresh install at a
/// familiar address, left alone for an afternoon, reaches seq 20 000 and the
/// twin — sitting at 19 264 of a history that no longer exists anywhere else —
/// SPLICES ONE HISTORY ONTO ANOTHER, cleanly and silently, and the only copy of
/// the real one is gone. That is the single worst thing this repository can do,
/// and no number can catch it. Only a name can.
///
/// # Not the host key, and not anything else already on the box
///
/// The host key is replayed by a restore ON PURPOSE (T3, `T1h host key
/// identical`), and so are the KEK and every document — which is exactly right
/// for a restore and exactly wrong here, because a value that survives a restore
/// cannot by itself distinguish the two boxes we care about. The id is minted
/// from the system CSPRNG and from nothing else, so a fresh box CANNOT produce
/// one it has not been given.
///
/// # A restored main KEEPS the id, and that is the whole design
///
/// The id names the HISTORY, not the box. A restore replays the twin's log into
/// a fresh root at the same sequence numbers, frame for frame; if that is not
/// the same history then the word restore means nothing. So `restore_finish`
/// writes the id that came down the stream, and the twin the new main was built
/// FROM keeps following it without being touched.
///
/// The alternative — a restored main mints a new id — was rejected on what it
/// costs at the one moment that matters. It would put the twin into `Diverged`
/// against the box that had just been built out of it, so the estate would have
/// NO twin in the hour after it has just survived losing its main; and getting
/// one back is not cheap, because `become_twin` takes a FRESH box and the twin
/// is not fresh — it holds the whole history — so the twin would have to be
/// wiped and re-seeded from the new main before the estate was covered again.
/// Paying that, every restore, to detect a case the id is not the mechanism for
/// is the wrong trade.
///
/// **What it therefore does NOT catch, said plainly:** two boxes can hold the
/// same id — the restored main, and the old main if it ever comes back. That
/// split-brain is created by the restore itself, not by this file, and it is
/// the sequence rule that meets it: the old main returns BEHIND the restored
/// one, so its head is below the twin's and [`State::Diverged`] already stops
/// there. An id cannot order two boxes in time; it can only say whether they
/// are the same history, which is the question it is here to answer.
///
/// The old main can also come back AHEAD: it took writes the twin never saw
/// before it went dark, or it keeps taking them after it returns, until its
/// head passes the twin's and every number agrees again. That ordering is what
/// [`LOG_GENERATION`] gives (stinky row 18, 2026-10-10): a restore stamps a newer
/// generation, the twin pins the highest it has followed, and an answer from a
/// lower one is [`State::Diverged`] by name.
pub const LOG_ID: &str = "txlog-id";
/// `<data-root>/txlog-generation`: **which restore of its log this box is**
/// (stinky row 18, 2026-10-10). Absent = 0: never restored.
///
/// `restore_finish` writes the unix ms of the restore, kept strictly above the
/// generation the root held before, so a later restore is always a higher
/// number even across a clock that stepped back. It is time and not a counter
/// carried down the restore stream on purpose: the stream (and the pumps that
/// relay it) stay exactly as they are, and two restores of the same log are
/// still ordered. A primary names it in every `Mirror.LogHead` answer, inside
/// its signature.
///
/// What it fences: the TWIN. An old main that comes back after its replacement
/// was restored carries the lower generation, and the twin refuses it however
/// good its id, key and numbers are. And the OLD MAIN, from the first poll a twin
/// sends it: the twin relays the restored main's signed answer, and the old
/// main fences itself ([`LOG_FENCED`]). Until a twin reaches it, an old main
/// cut off from every twin still takes writes — knowing sooner would take a
/// lease or a quorum, which is a different design.
pub const LOG_GENERATION: &str = "txlog-generation";
/// `<data-root>/twin-primary-generation`: the highest [`LOG_GENERATION`] this
/// twin has followed. Absent = 0. It only ever moves up.
pub const PINNED_GENERATION: &str = "twin-primary-generation";
/// One generation file. NotFound = 0; anything else that cannot be read, or is
/// not a number, is an error by name — a corrupt fence must never read as "no
/// fence", which is exactly the hole it is here to close.
fn read_generation(root: &Path, name: &str) -> anyhow::Result<u64> {
let path = root.join(name);
match std::fs::read_to_string(&path) {
Ok(raw) => raw.trim().parse::<u64>().map_err(|e| {
anyhow::anyhow!(
"{} holds {:?}, which is not a generation: {e}",
path.display(),
raw.trim()
)
}),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
Err(e) => Err(anyhow::Error::from(e).context(format!("reading {}", path.display()))),
}
}
fn forget_generation(root: &Path, name: &str) -> anyhow::Result<()> {
match std::fs::remove_file(root.join(name)) {
Ok(()) => crate::durable::sync_dir(root),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => {
Err(anyhow::Error::from(e).context(format!("removing {}", root.join(name).display())))
}
}
}
/// This box's [`LOG_GENERATION`].
pub fn log_generation(root: &Path) -> anyhow::Result<u64> {
read_generation(root, LOG_GENERATION)
}
/// The generation this twin has pinned ([`PINNED_GENERATION`]).
pub fn pinned_generation(root: &Path) -> anyhow::Result<u64> {
read_generation(root, PINNED_GENERATION)
}
/// **Stamp a restore**: a generation strictly above the one this root held.
fn stamp_generation(root: &Path) -> anyhow::Result<u64> {
let before = log_generation(root)?;
let generation = u64::try_from(now_ms())
.unwrap_or(0)
.max(before.saturating_add(1));
crate::durable::write_atomic_unframed(
root,
LOG_GENERATION,
format!("{generation}\n").as_bytes(),
)?;
Ok(generation)
}
/// Pin a primary's generation. Never moves the pin down.
fn pin_generation(root: &Path, generation: u64) -> anyhow::Result<()> {
if generation <= pinned_generation(root)? {
return Ok(());
}
crate::durable::write_atomic_unframed(
root,
PINNED_GENERATION,
format!("{generation}\n").as_bytes(),
)
}
/// `<data-root>/txlog-fenced`: **this box is an OLD generation of its log**
/// (stinky row 18, the main's half, 2026-10-10).
///
/// [`LOG_GENERATION`] fences the twin off an old main come back after a
/// restore. This fences the old main itself: the twin relays the restored
/// main's signed answer ([`GENERATION_PROOF`]) on every poll, and a box that
/// finds in it a NEWER generation of its OWN log, signed by its OWN host key, is
/// that old main — nobody else holds the key, and the restore replayed it onto
/// the new main. It writes this file (durably, before anything else), makes its
/// log follower-only, and from then on originates nothing and serves no
/// `Mirror.LogFetch`; [`ServerStores::open`] reads the file at every start, so a
/// restart does not undo it. It still answers `Mirror.LogHead`, with its own
/// (older) generation, so a twin that reaches it says why by name.
///
/// Only an operator removes it, deliberately: the box's writes since the
/// restore belong to no history, and the way back is to retire it or to re-join
/// it as a twin (`become_twin` forgets the fence along with the rest of its
/// own state).
pub const LOG_FENCED: &str = "txlog-fenced";
/// `<data-root>/twin-generation-proof`: the newest generation of its log this
/// twin has followed, **as its main signed it** — the fields of one verified
/// `Mirror.LogHead` answer that its `gunnar-log-head-v2` signature covers, and
/// the key that made it, one per line. Relayed as
/// `LogHeadRequest.generation_proof` on every poll; see [`LOG_FENCED`].
pub const GENERATION_PROOF: &str = "twin-generation-proof";
/// **Is this box fenced off as an old generation of its log?** ([`LOG_FENCED`])
/// A marker that cannot be looked at is a fence: the hole it closes is worse
/// than a box refusing writes it could have taken.
pub fn is_fenced(root: &Path) -> bool {
match std::fs::symlink_metadata(root.join(LOG_FENCED)) {
Ok(_) => true,
Err(e) => e.kind() != std::io::ErrorKind::NotFound,
}
}
/// What [`LOG_FENCED`] says, for the alarm and the refusals.
pub fn fenced_why(root: &Path) -> Option<String> {
is_fenced(root).then(|| {
let said = std::fs::read_to_string(root.join(LOG_FENCED)).unwrap_or_default();
let said = said.trim();
if said.is_empty() {
format!("this box is fenced off as an OLD generation of its log ({LOG_FENCED} is on its data root)")
} else {
format!("{said} ({LOG_FENCED} is on its data root)")
}
})
}
/// The proof this twin relays on every poll ([`GENERATION_PROOF`]). `None` =
/// it has never followed a restored main. A file that is there and does not
/// read as one is an error by name.
pub fn generation_proof(root: &Path) -> anyhow::Result<Option<gunnar_proto::GenerationProof>> {
let path = root.join(GENERATION_PROOF);
let raw = match std::fs::read_to_string(&path) {
Ok(raw) => raw,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => return Err(anyhow::Error::from(e).context(format!("reading {}", path.display()))),
};
let lines: Vec<&str> = raw.lines().collect();
let [answer_key, answer_sig, nonce, log_id, first_seq, head_seq, generation] = lines[..] else {
anyhow::bail!(
"{} holds {} lines, not the 7 of one signed answer",
path.display(),
lines.len()
);
};
let number = |what: &str, v: &str| {
v.trim()
.parse::<u64>()
.map_err(|e| anyhow::anyhow!("{}: {what} {v:?} is not a number: {e}", path.display()))
};
Ok(Some(gunnar_proto::GenerationProof {
answer_key: answer_key.to_owned(),
answer_sig: answer_sig.to_owned(),
nonce: nonce.to_owned(),
log_id: log_id.to_owned(),
first_seq: number("first_seq", first_seq)?,
head_seq: number("head_seq", head_seq)?,
generation: number("generation", generation)?,
}))
}
/// **Keep a verified answer as this twin's proof** ([`GENERATION_PROOF`]) — called
/// only after the answer passed [`GrpcSource`]'s signature check against `nonce`.
/// Kept when it is signed, names a generation, and that generation is above the
/// proof already kept; otherwise nothing changes.
pub fn remember_generation_proof(
root: &Path,
nonce: &str,
head: &gunnar_proto::LogHeadResponse,
) -> anyhow::Result<()> {
if head.generation == 0
|| head.answer_sig.trim().is_empty()
|| head.answer_key.trim().is_empty()
{
return Ok(());
}
if generation_proof(root)
.ok()
.flatten()
.is_some_and(|kept| kept.generation >= head.generation)
{
return Ok(());
}
let one_line = |v: &str| !v.contains('\n') && !v.contains('\r');
anyhow::ensure!(
[
head.answer_key.as_str(),
head.answer_sig.as_str(),
nonce,
head.log_id.as_str()
]
.into_iter()
.all(one_line),
"a signed answer whose key, signature, nonce or log id spans lines is not kept as a proof"
);
let text = format!(
"{}\n{}\n{nonce}\n{}\n{}\n{}\n{}\n",
head.answer_key.trim(),
head.answer_sig.trim(),
head.log_id,
head.first_seq,
head.head_seq,
head.generation
);
crate::durable::write_atomic_unframed(root, GENERATION_PROOF, text.as_bytes())
}
/// **Fence this box when `proof` shows a newer generation of its own log**
/// ([`LOG_FENCED`]). `Some(generation)` = fenced by this call; `None` = not a
/// proof about this box (its own or an older generation, another log, another
/// key, a signature that does not verify), or nothing to fence (a twin, a box
/// mid-restore, a box already fenced). Only a failure to WRITE the fence is an
/// error, and the log is made follower-only even then.
pub fn fence_by_proof(
stores: &ServerStores,
proof: &gunnar_proto::GenerationProof,
) -> anyhow::Result<Option<u64>> {
let root = stores.root();
if is_twin_root(root) || stores.is_twin() || is_restoring(root) || is_fenced(root) {
return Ok(None);
}
let own = log_generation(root)?;
if proof.generation <= own {
return Ok(None);
}
let Some(id) = log_id(root) else {
return Ok(None);
};
if proof.log_id.trim() != id {
tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, theirs = %proof.log_id, mine = %id, "a twin relayed a generation of another log; not this box's business");
return Ok(None);
}
let Some(key) = answering_key(stores) else {
return Ok(None);
};
let Ok(line) = key.public_key().to_openssh() else {
return Ok(None);
};
if proof.answer_key.trim() != line.trim()
|| !gunnar_grpc::auth::verify_twin_answer_at(
&line,
&proof.answer_sig,
&proof.nonce,
&proof.log_id,
proof.first_seq,
proof.head_seq,
proof.generation,
)
{
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, generation = proof.generation, "a twin relayed a generation of this log that this box's host key did not sign: ignored");
return Ok(None);
}
let why = format!(
"this box is generation {own} of log {id}; generation {} of the same log exists, restored at {} (unix ms) and signed by this \
box's own host key. This is the OLD main come back after its replacement was restored: it originates nothing and serves no \
log from here, because its writes since that restore belong to no history. Retire it, or re-join it as a twin.",
proof.generation, proof.generation
);
stores.txlog().set_follower(true);
crate::durable::write_atomic_unframed(root, LOG_FENCED, format!("{why}\n").as_bytes())
.context("writing the fence")?;
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, event = "twin.fenced_by_generation", own, newer = proof.generation, "{why}");
Ok(Some(proof.generation))
}
fn forget_fence(root: &Path) -> anyhow::Result<()> {
for name in [LOG_FENCED, GENERATION_PROOF] {
match std::fs::remove_file(root.join(name)) {
Ok(()) => crate::durable::sync_dir(root)?,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => {
return Err(anyhow::Error::from(e)
.context(format!("removing {}", root.join(name).display())))
}
}
}
Ok(())
}
/// **The key this box signs its answers to a twin with** — its SSH host key,
/// read once per process out of the sealed store.
///
/// The host key and not a new one, deliberately. It is the key the box already
/// proves itself with on :2222, so an operator who has checked one has checked
/// the other; and it is REPLAYED BY A RESTORE, which is exactly the behaviour
/// this pairing wants — a main restored from the twin keeps signing with the
/// same key, so the twin keeps following it without being re-joined, matching
/// what [`LOG_ID`] does for the history. A fresh box at the same address mints
/// its own and is refused. (That same survive-a-restore property is why the
/// host key could NOT be the log identity: there it would have had to
/// distinguish things a restore is supposed to make identical.)
///
/// Cached because a twin polls every two seconds and a sealed-store read per
/// poll is a cost for nothing: a host key does not change under a running
/// process.
fn answering_key(stores: &ServerStores) -> Option<gunnar_core::ssh_key::PrivateKey> {
static KEY: OnceLock<Option<gunnar_core::ssh_key::PrivateKey>> = OnceLock::new();
KEY.get_or_init(|| {
// A twin reads its (replicated) host key through the unframed store; a
// primary through its own. Both are "the key this box answers as".
let secrets = if stores.is_twin() { stores.secrets_unframed() } else { stores.secrets() };
match secrets.and_then(|s| gunnar_core::ensure_host_key(&s).map_err(anyhow::Error::from)) {
Ok(host) => Some(host.private_key().clone()),
Err(err) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "this box cannot read its host key, so it cannot sign its answers to a twin; a twin that has pinned a key will refuse them");
None
}
}
})
.clone()
}
/// The `authorized_keys` line and one signature over this answer, for
/// `Mirror.LogHead`. `("", "")` when the caller sent no nonce (nothing to be
/// fresh about) or this box cannot read its key.
pub fn sign_head_answer(
stores: &ServerStores,
nonce: &str,
log_id: &str,
first_seq: u64,
head_seq: u64,
generation: u64,
) -> (String, String) {
if nonce.trim().is_empty() {
return (String::new(), String::new());
}
let Some(key) = answering_key(stores) else {
return (String::new(), String::new());
};
let line = match key.public_key().to_openssh() {
Ok(line) => line,
Err(err) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "this box's host key does not re-encode");
return (String::new(), String::new());
}
};
match gunnar_grpc::auth::sign_twin_answer_at(
&key, nonce, log_id, first_seq, head_seq, generation,
) {
Ok(sig) => (line, sig),
Err(err) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "signing this answer to a twin");
(String::new(), String::new())
}
}
}
/// This data root's log identity, when it has one. `None` = never minted (a box
/// older than this change, or a twin that has not yet heard from its primary).
pub fn log_id(root: &Path) -> Option<String> {
let raw = std::fs::read_to_string(root.join(LOG_ID)).ok()?;
let id = raw.trim().to_owned();
if id.is_empty() {
None
} else {
Some(id)
}
}
/// **Mint this log's identity, once.** Returns the existing one when there is
/// one, so it is safe to call on every open. 128 bits of CSPRNG, hex.
pub fn mint_log_id(root: &Path) -> anyhow::Result<String> {
if let Some(id) = log_id(root) {
return Ok(id);
}
use gunnar_core::ssh_key::rand_core::Rng as _;
let mut bytes = [0u8; 16];
gunnar_core::os_rng().fill_bytes(&mut bytes);
let id: String = bytes.iter().map(|b| format!("{b:02x}")).collect();
crate::durable::write_atomic_unframed(root, LOG_ID, format!("{id}\n").as_bytes())?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, log_id = %id, "this box minted its log's identity: from here a twin can tell this history from any other");
Ok(id)
}
/// **Write the identity of the log this root now holds** — a twin learning its
/// primary's on the first poll, or a fresh box taking one off a restore stream.
/// Refuses to overwrite a DIFFERENT id: that is the splice this whole file
/// exists to prevent, and it must never happen quietly.
pub fn pin_log_id(root: &Path, id: &str) -> anyhow::Result<()> {
let id = id.trim();
anyhow::ensure!(!id.is_empty(), "a log identity cannot be empty");
if let Some(mine) = log_id(root) {
anyhow::ensure!(
mine == id,
"this data root already holds log {mine} and was asked to hold log {id}; two histories do not become \
one by overwriting a file"
);
return Ok(());
}
crate::durable::write_atomic_unframed(root, LOG_ID, format!("{id}\n").as_bytes())
}
/// `<data-root>/twin-primary.pub`: **the key the box this twin follows answers
/// with** — one OpenSSH `authorized_keys` line, pinned on the first poll and
/// required on every one after (2026-09-21, lane T4).
///
/// # What was there before
///
/// Nothing. `GrpcSource::connect` dialled `client_signer::plain_channel` — a
/// bare `http://` endpoint, no TLS — and the twin verified NOTHING about who
/// answered. The signed credential on every call proved the TWIN to the
/// primary; not one byte ever went the other way. The trust was one-directional,
/// so whatever won the address won the twin: a DNS answer, a recycled lease, a
/// cable in the wrong port, a box rebuilt at the same IP.
///
/// # Trust on first use, and why there is no earlier moment
///
/// `Bootstrap.BecomeTwin` is answered by the TWIN. The primary is not in that
/// conversation at all — it does not learn it has a twin until that twin's key
/// is enrolled on it by hand, afterwards — so at join there is nothing to pin
/// except an address the operator typed. The first poll is the first time this
/// box has ever heard from the other one. That is exactly the position `ssh`
/// is in the first time it connects to a host, and `known_hosts` is the answer
/// there for the same reason it is the answer here: pinning at first sight
/// converts an open-ended exposure into a single one, and every later change is
/// loud.
///
/// # Its own file, not a field of `twin.json`
///
/// `twin.json` is what the operator wrote at join: an address and a time.
/// This is what the twin MEASURED afterwards. Keeping them apart means a hand
/// edit of the join record cannot silently re-pin the key, and it matches
/// [`LOG_ID`], which is a measured fact about the log in the same way. (It also
/// keeps a struct that four other test files construct by literal from growing
/// a field, one of which this lane must not touch.)
pub const PRIMARY_KEY: &str = "twin-primary.pub";
/// The pinned `authorized_keys` line, when this twin has one.
pub fn pinned_primary_key(root: &Path) -> Option<String> {
let raw = std::fs::read_to_string(root.join(PRIMARY_KEY)).ok()?;
let line = raw.trim().to_owned();
if line.is_empty() {
None
} else {
Some(line)
}
}
/// **Pin the key the primary answers with**, on the first poll. Refuses to
/// replace a different one: that is the substitution this exists to catch.
pub fn pin_primary_key(root: &Path, line: &str) -> anyhow::Result<()> {
let line = line.trim();
anyhow::ensure!(!line.is_empty(), "a primary's key cannot be empty");
if let Some(mine) = pinned_primary_key(root) {
anyhow::ensure!(
mine == line,
"this twin already follows a box that answers with {mine}; it was offered {line}"
);
return Ok(());
}
crate::durable::write_atomic_unframed(root, PRIMARY_KEY, format!("{line}\n").as_bytes())?;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, key = %line, "this twin has pinned the key its primary answers with; every later answer must be signed by it");
Ok(())
}
/// **Give up this box's own log identity**, with its numbers and its secrets: a
/// fresh box becoming a twin, or taking another box's log in a restore. Without
/// this it would compare the identity it minted at its own first open against
/// the one it is about to start following and refuse itself at the first poll.
/// **Drop this box's own control-plane audit trail** before it takes another
/// box's log (lane T11).
///
/// Not "best effort": a leftover line would make the restored chain fail
/// `AuditChain::open`, and a box that cannot open its trail serves without one.
/// The restore would then have destroyed the audit history it exists to carry —
/// the exact D6 failure, one layer in.
///
/// It is the replicated path that is removed, and only that: `prepare_replay`
/// runs before the `txlog-restoring` marker exists, so `audit_path()` still
/// names `log/audit-chain.jsonl`, which is precisely the file about to be
/// rebuilt from the stream.
fn forget_own_audit_trail(stores: &ServerStores) -> anyhow::Result<()> {
let path = stores.audit_path();
match std::fs::remove_file(&path) {
Ok(()) => {
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
"dropped this fresh box's own audit trail: the one being restored starts at its own seq 0"
);
Ok(())
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e)
.with_context(|| format!("dropping this box's own audit trail {}", path.display())),
}
}
/// **Check the control-plane audit trail against this box's own log, on EVERY start**, and
/// rebuild what the file lost — before the gate opens it (lane T11, 2026-09-21; every start
/// since T17, owner's decision the same day).
///
/// **What this is NOT: the fix for T3's KVM `T1v`/`T2v` failure** (medium `ff2c7c9`). This
/// function was first written on the belief that the restored box's copy of the trail, on the
/// boot disk's P3, was lost in the hard stop between the replay and the restart. That diagnosis
/// was WRONG, and it is corrected here rather than left standing. T3 found the real cause:
/// `restore_finish` unlinked `txlog-restoring` with no directory fsync, the hard stop dropped
/// the unlink, and the box reopened STILL RESTORING — a follower that re-sealed nothing,
/// originated nothing, and answered `Admin.AuditTrail` from its own per-box trail. The log
/// says so: the twin polled the restored box every 2 s and saw its head stay at 262, where a
/// box that had re-sealed and served 16 audited calls would have framed far past it. And the
/// guard that decides whether THIS function runs is "not restoring", so it never ran. T3 fixed
/// the marker (`durable::remove_durable`, alarm `twin.restoring_at_open`).
///
/// **The window on a running primary** (T11 named it, T17 closes it): `AuditChain::append`
/// syncs the FRAME, then writes the line to the file. A crash between the two leaves the log one
/// entry ahead of the file; XFS log recovery on P3 after a hard stop can also bring the file
/// back shorter, or with a torn last line. The gate would then append to the SHORTER chain — a
/// second entry at an existing `seq`, a fork of the chain the twin
/// holds, whose copy stops verifying from that entry on. So the check is not a restore's: it
/// runs on every open of a box that originates its log (not a twin, not mid-restore), in
/// `ServerStores::open`, before `serve` opens the chain for the gate. A box that reopens still
/// RESTORING (T3: the marker's unlink lost, fixed by `durable::remove_durable` on
/// `agent/twin-restore-proof`) originates nothing and writes its per-box trail; it is not
/// checked here, and T3's alarm `twin.restoring_at_open` is what names it.
///
/// The trail is a materialised view of the log's frames for [`gunnar_core::txlog::AUDIT_REL`]
/// — `Append` lines, and a `File` frame where a snapshot carried the whole document (which
/// supersedes every earlier line). What the file lacks past its own tip is APPENDED, linked by
/// `prev`, checked in memory to verify, written atomically (temporary, fsync, rename, fsync of
/// the directory), and checked to open.
///
/// **It never discards a line the log does not hold.** The log may carry only a WINDOW of the
/// trail (the operator's retention door, [`Log::retain_from`]; a box upgraded from before the
/// trail was framed), so "the log's view" is never written over the file. A copy LONGER than
/// the log is left as found, and so is one whose tip the log contradicts, or one the log's
/// window cannot be joined to — each said out loud. The only bytes ever dropped are a torn
/// last line that does not parse as an entry.
///
/// Best effort in the sense the gate is: a failure is one `error!` naming the path and the box
/// still serves. But it is never silent — the one thing D6 was.
///
/// Cost: one header-and-meta pass over the log ([`Log::scan_meta`], one open per segment), then
/// blobs read newest-first only until the file's tip is met — ONE blob when the file is whole.
pub fn rematerialise_audit_trail(stores: &ServerStores, just_restored: bool) {
let path = stores.audit_path();
let when = if just_restored {
"the first start after a restore"
} else {
"this start"
};
let t0 = std::time::Instant::now();
let outcome = rebuild_audit_trail(stores.txlog(), &path);
let micros = t0.elapsed().as_micros() as u64;
match outcome {
Ok(Rebuilt::Nothing) => {}
Ok(Rebuilt::Intact(n)) => tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
entries = n,
micros,
"the audit trail is whole at {when}: its tip is the log's newest audit entry"
),
Ok(Rebuilt::Rebuilt { had, now }) => tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
had,
now,
micros,
"the audit trail held {had} of the {now} entries its log carries at {when} — the \
off-stripe copy lost its tail (a hard stop; XFS recovery on the boot disk's P3). \
REBUILT from the log, which did not: the trail is whole again and the next entry \
continues the chain the twin holds"
),
Ok(Rebuilt::Longer { file, log }) => tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
file,
log,
"the audit trail holds MORE entries ({file}) than its log carries ({log}) at {when}; \
left exactly as found — nothing is deleted from an audit trail"
),
Ok(Rebuilt::Diverged { why }) => tracing::error!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
why = %why,
"the audit trail and its log DISAGREE at {when}; left exactly as found — nothing is \
guessed into an audit trail. The twin's copy is the one to compare against"
),
Err(err) => tracing::error!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
error = %format!("{err:#}"),
"the audit trail could not be checked against its log at {when}; if the off-stripe \
copy lost its tail, the gate appends to a shorter chain than the twin holds"
),
}
}
/// What [`rebuild_audit_trail`] did.
#[derive(Debug, PartialEq, Eq)]
pub enum Rebuilt {
/// The log carries no audit frame at all (a primary from before lane T11, or a box that has
/// made no audited decision).
Nothing,
/// The file's tip is the log's newest audit entry: this many entries in the file.
Intact(usize),
/// The file held `had` whole entries; `now - had` were appended from the log.
Rebuilt {
/// Whole lines in the file before.
had: usize,
/// Lines in the file now.
now: usize,
},
/// The file's tip is past the log's newest audit entry; left alone.
Longer {
/// Lines in the file.
file: usize,
/// Audit entries the log carries.
log: usize,
},
/// The file and the log cannot be joined without guessing; left alone.
Diverged {
/// Which fact disagreed.
why: String,
},
}
/// One audit line of the log's view: where it came from, for a lazy read newest-first.
enum ViewLine {
/// An `Append` frame's blob.
Frame(u64),
/// Line `i` of the snapshot's document.
Snap(usize),
}
/// The work of [`rematerialise_audit_trail`], over a log and a path, so it can be proved without
/// a whole box.
pub fn rebuild_audit_trail(log: &Log, path: &Path) -> anyhow::Result<Rebuilt> {
// 1. Which frames carry the trail: the newest snapshot of it, then every Append after it.
let (append, file) = (Kind::Append as u16, Kind::File as u16);
// The exact bytes this code writes for an audit line's meta: compared, not parsed — parsing the JSON
// of every audit frame's meta made 90 000 more audit frames cost 263 ms more (MEASURED,
// `audit_check_boot_cost`). Any other meta naming the trail is parsed in full.
let audit_meta = Meta::Append {
log: txlog::AUDIT_REL.into(),
}
.to_bytes();
let names_trail = |meta: &[u8]| {
meta.windows(txlog::AUDIT_REL.len())
.any(|w| w == txlog::AUDIT_REL.as_bytes())
};
let mut frames: Vec<u64> = Vec::new();
let mut snapshot: Option<(u64, bool)> = None;
let mut bad_meta: Option<(u64, String)> = None;
log.scan_meta(|seq, kind, meta| {
if kind == append && meta == audit_meta.as_slice() {
frames.push(seq);
return;
}
if (kind != append && kind != file) || !names_trail(meta) {
return;
}
match Meta::from_bytes(meta) {
Ok(Meta::Append { log: rel }) if kind == append && rel == txlog::AUDIT_REL => {
frames.push(seq)
}
Ok(Meta::File { path: rel, deleted }) if kind == file && rel == txlog::AUDIT_REL => {
frames.clear();
snapshot = Some((seq, deleted));
}
Ok(_) => {}
Err(e) => bad_meta = bad_meta.take().or(Some((seq, e.to_string()))),
}
})
.context("scanning the log's frame headers")?;
if let Some((seq, e)) = bad_meta {
return Err(anyhow!("frame {seq}'s meta does not parse: {e}"));
}
if frames.is_empty() && snapshot.is_none() {
return Ok(Rebuilt::Nothing);
}
let snap_lines: Vec<Vec<u8>> = match snapshot {
Some((seq, false)) => split_lines(&log.read(seq)?.blob).0,
_ => Vec::new(),
};
let mut view: Vec<ViewLine> = (0..snap_lines.len()).map(ViewLine::Snap).collect();
view.extend(frames.iter().map(|s| ViewLine::Frame(*s)));
// 2. The file: whole lines, and a torn last one (no newline, and not an entry) set aside.
let bytes = match std::fs::read(path) {
Ok(b) => b,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),
Err(e) => return Err(e).with_context(|| format!("reading {}", path.display())),
};
let (mut lines, rest) = split_lines(&bytes);
let mut torn = false;
if let Some(rest) = rest {
if serde_json::from_slice::<gunnar_core::AuditEntry>(&rest).is_ok() {
lines.push(rest);
} else {
torn = true;
}
}
let had = lines.len();
let tip: Option<gunnar_core::AuditEntry> = match lines.last() {
Some(l) => Some(serde_json::from_slice(l).with_context(|| {
format!(
"the last whole line of {} is not an audit entry",
path.display()
)
})?),
None => None,
};
// 3. The log's view, newest first, until the file's tip is met: ONE blob when it is whole.
let line_at = |v: &ViewLine| -> anyhow::Result<Vec<u8>> {
Ok(match v {
ViewLine::Frame(seq) => log.read(*seq)?.blob,
ViewLine::Snap(i) => snap_lines[*i].clone(),
})
};
let mut missing: Vec<Vec<u8>> = Vec::new();
let mut met = false;
for (k, v) in view.iter().enumerate().rev() {
let line = line_at(v)?;
let entry: gunnar_core::AuditEntry =
serde_json::from_slice(&line).context("an audit line in the log is not an entry")?;
match &tip {
Some(t) if entry.seq == t.seq => {
if entry.hash != t.hash {
return Ok(Rebuilt::Diverged {
why: format!(
"seq {}: the file's hash is {}, the log's {}",
t.seq, t.hash, entry.hash
),
});
}
met = true;
break;
}
Some(t) if entry.seq < t.seq => {
if k + 1 == view.len() {
return Ok(Rebuilt::Longer {
file: lines.len(),
log: view.len(),
});
}
return Ok(Rebuilt::Diverged {
why: format!(
"the log skips from seq {} to a newer entry, past the file's tip {}",
entry.seq, t.seq
),
});
}
_ => missing.push(line),
}
}
if missing.is_empty() && !torn {
return Ok(Rebuilt::Intact(had));
}
missing.reverse();
if !met {
// The log's window starts after the file's tip, or the file is empty: only a window that
// begins exactly one past the tip (or at seq 0 for an empty file) joins without a gap.
let first: gunnar_core::AuditEntry = serde_json::from_slice(&missing[0])?;
let want = tip.as_ref().map_or(0, |t| t.seq + 1);
if first.seq != want {
return Ok(Rebuilt::Diverged { why: format!("the log's audit entries begin at seq {}, the file needs {want} next — a retention hole nothing can bridge", first.seq) });
}
}
// 4. The joined trail — the exact bytes about to be written — must verify in memory before
// one of them lands: a file that was already broken is not made to look whole by an
// append, and nothing is written that the gate would then refuse to open.
let mut out =
Vec::with_capacity(bytes.len() + missing.iter().map(|l| l.len() + 1).sum::<usize>());
for l in lines.iter().chain(missing.iter()) {
out.extend_from_slice(l);
out.push(b'\n');
}
let entries: Vec<gunnar_core::AuditEntry> = out
.split(|b| *b == b'\n')
.filter(|l| !l.is_empty())
.map(serde_json::from_slice)
.collect::<Result<_, _>>()
.context("parsing the joined trail")?;
if let Err(e) = gunnar_core::audit::verify_entries(&entries) {
return Ok(Rebuilt::Diverged {
why: format!("the file joined to the log's tail does not verify: {e}"),
});
}
let dir = path
.parent()
.ok_or_else(|| anyhow!("{} has no directory", path.display()))?;
std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
let name = path
.file_name()
.ok_or_else(|| anyhow!("{} names no file", path.display()))?
.to_string_lossy()
.into_owned();
// Unframed: these lines ARE frames already — this box's log holds every one of them.
crate::durable::write_atomic_unframed(dir, &name, &out)
.with_context(|| format!("rewriting {}", path.display()))?;
gunnar_core::AuditChain::open(path)
.map_err(|e| anyhow!("the rebuilt trail does not open: {e}"))?;
Ok(Rebuilt::Rebuilt {
had,
now: had + missing.len(),
})
}
/// Whole `\n`-terminated non-empty lines, and what follows the last newline (a torn line), if any.
fn split_lines(bytes: &[u8]) -> (Vec<Vec<u8>>, Option<Vec<u8>>) {
let (whole, rest) = match bytes.iter().rposition(|b| *b == b'\n') {
Some(i) => (&bytes[..i], &bytes[i + 1..]),
None => (&bytes[..0], bytes),
};
let lines = whole
.split(|b| *b == b'\n')
.filter(|l| !l.iter().all(u8::is_ascii_whitespace))
.map(<[u8]>::to_vec)
.collect();
let rest = (!rest.iter().all(|b| b.is_ascii_whitespace() || *b == 0)).then(|| rest.to_vec());
(lines, rest)
}
fn forget_log_id(root: &Path) -> anyhow::Result<()> {
match std::fs::remove_file(root.join(LOG_ID)) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e).with_context(|| format!("dropping this box's own {LOG_ID}")),
}
}
/// **Was the git listener bound in a twin's shape?** — set once by
/// `serve::start`, when it chose the bind (2026-09-21, lane T4).
static GIT_BOUND_AS_TWIN: AtomicBool = AtomicBool::new(false);
/// `serve::start` says which shape it bound the git listener in.
pub fn note_git_bind_is_twin_shaped(yes: bool) {
GIT_BOUND_AS_TWIN.store(yes, Ordering::SeqCst);
}
/// **Does this box need a restart before its git listener is where a twin's
/// belongs?**
///
/// `serve::start` picks the git bind ONCE, from `stores.is_twin()` at that
/// instant: a twin gets `127.0.0.1:0`, everything else gets the operator's
/// address. A box conferred the twin role over the wire by
/// `Bootstrap.BecomeTwin` is past that point — and under the owner's
/// one-medium ruling that is the ONLY way a twin is ever made, because a twin
/// is installed from the same image as a main with no per-box parameter to say
/// which it is. MEASURED under KVM by lane T3, 2026-09-21: such a box kept
/// `0.0.0.0:2222` bound for the rest of its life and the
/// `this box is a TWIN: the git listener is bound to loopback only` warning
/// never printed.
///
/// The door refuses every git service on a twin from the instant the role lands
/// (`gunnar_ssh::Control::is_twin`), so nothing is SERVED through that socket —
/// but it is still bound on an address a twin's socket has no business being
/// on, and only a restart moves it. This is how that fact is told rather than
/// left for somebody to find with `ss`.
pub fn git_listener_needs_restart(stores: &ServerStores) -> bool {
stores.is_twin() && !GIT_BOUND_AS_TWIN.load(Ordering::SeqCst)
}
/// Wall clock, ms since the epoch.
pub fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
/// **A fresh box gives up its own numbers.** Refused unless the box is
/// fresh: no key enrolled, the bootstrap window open, no repository. The
/// documents it DID write (its first-start issuer, its catalog seed — every
/// `ControlDoc` in its own log) are remembered in [`FRESH_DOCS`]; its log is
/// made follower-only and reset to empty. The first replayed frame for a
/// remembered document is then applied as if the document were absent — the
/// replay is into an empty store, and the fresh box's own value is not a
/// claim the primary ever made.
pub fn prepare_replay(stores: &ServerStores) -> anyhow::Result<()> {
if !stores.auth().is_empty()? || !stores.auth().bootstrap_window_open()? {
anyhow::bail!("this box has enrolled identities; only a fresh box (bootstrap window open, no key) can take another box's log");
}
if !stores.catalog().list(None)?.is_empty() {
anyhow::bail!("this box holds repositories; only a fresh box can take another box's log");
}
let log = stores.txlog();
let mut names = std::collections::BTreeSet::new();
if log.head_seq() >= log.first_seq() {
for e in log.range(log.first_seq(), log.head_seq())? {
if let Ok(Meta::ControlDoc { name, .. }) = e.meta() {
names.insert(name);
}
}
}
crate::durable::write_atomic_unframed(stores.root(), FRESH_DOCS, &serde_json::to_vec(&names)?)?;
forget_own_secrets(stores)?;
// **And its own log's NAME.** This box minted one at its first open, for a
// history it is about to give up entirely. Keeping it would make the twin
// refuse itself at its very first poll — its own id against the primary's —
// and would make a restored box claim to be a history it only holds a copy
// of. The id it ends up with is the one it is told: the primary's, on the
// first poll, or the stream's, at `restore_finish`.
forget_log_id(stores.root())?;
// And any primary generation it pinned: the generation it ends up with is
// the one `restore_finish` stamps (row 18). Its OWN `txlog-generation` stays,
// so the stamp only ever moves up.
forget_generation(stores.root(), PINNED_GENERATION)?;
// And a fence from a life as an old main, and any proof it relayed (row 18).
forget_fence(stores.root())?;
// **And its own AUDIT TRAIL** (lane T11, 2026-09-21). The trail is a
// replicated, hash-chained document now, so the frames arriving from the
// twin are `Append`s of the DEAD primary's chain from its seq 0. A fresh
// box that had written even one line of its own would take them on the end
// of it, and the result would not verify: entry 0's `prev` is GENESIS, and
// here it would follow somebody else's tip. The same reasoning as
// `forget_own_secrets` and `forget_log_id` above — this box is giving up
// its own short history entirely, not merging it with the one it is being
// handed.
forget_own_audit_trail(stores)?;
log.set_follower(true);
log.reset(u64::MAX)?;
Ok(())
}
/// **A fresh box gives up its own SECRETS with its own numbers** (2026-09-17, lane
/// TWIN-COMPLETENESS).
///
/// Every secret a fresh box holds is sealed under the KEK it minted at first open, and the
/// replay's `Kek` frame — seq 1 of every primary's log — overwrites that key on disk before any
/// `Secret` frame lands. A row the replay then OVERWRITES is fine; a row it has no frame for is
/// left sealed under a key that no longer exists, and [`reseal_after_restore`] meets it with
/// neither key able to open it.
///
/// MEASURED: a fresh box holding one secret of its own refused to open after an otherwise
/// successful restore, every time, for ever. [`reseal_after_restore`] now drops such a row
/// loudly rather than bricking the box; this is the other half — the row never reaches it,
/// because a box that gives up its numbers and its documents has no business keeping the keys
/// it sealed under a KEK it is about to lose. The box is FRESH by the checks above (bootstrap
/// window open, no enrolled identity, no repository), so nothing here is anybody's data: a host
/// key it minted at first boot, an issuer it minted at first start. A twin re-mints what it
/// needs through its unlogged store at the next open, and the primary's replayed value
/// overwrites that in turn.
fn forget_own_secrets(stores: &ServerStores) -> anyhow::Result<()> {
use gunnar_store::SecretStore as _;
let rows = stores
.control_db()
.export()
.context("reading this box's own secrets before it gives up its numbers")?
.secrets;
let mut gone = Vec::new();
for (key, value) in &rows {
if value.is_none() {
continue;
}
stores
.raw_secrets()
.delete(key)
.with_context(|| format!("dropping this box's own secret {key}"))?;
gone.push(key.clone());
}
if !gone.is_empty() {
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
secrets = ?gone,
"this box gave up the secrets it minted for itself along with its own log numbers: they are sealed under a KEK the replay is about to replace, and what the log carries is the only identity this box keeps"
);
}
Ok(())
}
/// The fresh-box rule, one document: `true` (and forgotten) when `name` was
/// written by this box before it was reset.
fn fresh_take(root: &Path, name: &str) -> anyhow::Result<bool> {
let path = root.join(FRESH_DOCS);
let bytes = match std::fs::read(&path) {
Ok(b) => b,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(false),
Err(e) => return Err(e.into()),
};
let mut names: std::collections::BTreeSet<String> = serde_json::from_slice(&bytes)?;
if !names.remove(name) {
return Ok(false);
}
crate::durable::write_atomic_unframed(root, FRESH_DOCS, &serde_json::to_vec(&names)?)?;
Ok(true)
}
/// A path from a frame, joined under `root` — never absolute, never `..`,
/// never a cache.
fn under_root(root: &Path, rel: &str) -> anyhow::Result<std::path::PathBuf> {
let p = Path::new(rel);
if p.is_absolute()
|| p.components()
.any(|c| !matches!(c, std::path::Component::Normal(_)))
{
anyhow::bail!("the frame names {rel:?}, which is not a plain path under the data root");
}
if txlog::never_framed(rel) {
anyhow::bail!("the frame names {rel:?}, a cache or per-box path — never applied on a twin");
}
Ok(root.join(p))
}
/// Append `line` + newline to `path` unless the file already ends with
/// exactly that line (a re-apply of the tail after a restart).
fn append_line_once(path: &Path, line: &[u8]) -> anyhow::Result<()> {
use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
let mut want = line.to_vec();
want.push(b'\n');
if let Ok(mut f) = std::fs::File::open(path) {
let len = f.metadata()?.len();
if len >= want.len() as u64 {
f.seek(SeekFrom::Start(len - want.len() as u64))?;
let mut tail = vec![0u8; want.len()];
f.read_exact(&mut tail)?;
let boundary = len == want.len() as u64 || {
f.seek(SeekFrom::Start(len - want.len() as u64 - 1))?;
let mut b = [0u8; 1];
f.read_exact(&mut b)?;
b[0] == b'\n'
};
if tail == want && boundary {
return Ok(());
}
}
}
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let mut f = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(path)?;
f.write_all(&want)?;
f.sync_data()?;
Ok(())
}
/// The primary's KEK, written where `Kek::for_control_dir` reads it (0600,
/// atomic). Takes effect at the next open — a restored box restarts.
fn write_kek(control_dir: &Path, bytes: &[u8]) -> anyhow::Result<()> {
use std::io::Write as _;
anyhow::ensure!(
bytes.len() == 32,
"a KEK frame carries {} bytes, not 32",
bytes.len()
);
let path = gunnar_core::seal::kek_path(control_dir);
if std::fs::read(&path).ok().as_deref() == Some(bytes) {
return Ok(());
}
std::fs::create_dir_all(control_dir)?;
let tmp = control_dir.join("secrets.kek.tmp");
let mut options = std::fs::OpenOptions::new();
options.write(true).create(true).truncate(true);
{
use std::os::unix::fs::OpenOptionsExt as _;
options.mode(0o600);
}
let mut f = options.open(&tmp)?;
f.write_all(bytes)?;
f.sync_all()?;
drop(f);
std::fs::rename(&tmp, &path)?;
// The KEK is written once its name is durable (stinky row 9).
crate::durable::sync_dir(control_dir)
}
// ── the doors: join as a twin, restore a fresh box ──────────────────────────
/// **`Bootstrap.BecomeTwin`**: a fresh box becomes a twin of `primary`. Mints
/// the twin's key (per box, `twin.key`), writes `twin.json`, gives up its own
/// numbers ([`prepare_replay`]) and freezes its control plane. Returns the
/// public key line to enrol on the primary in group [`TWIN_GROUP`]. Idempotent
/// for the same primary (the same key comes back); another primary is refused.
pub fn become_twin(stores: &ServerStores, primary: &str) -> anyhow::Result<String> {
let root = stores.root();
if let Some(existing) = TwinConfig::load(root)? {
anyhow::ensure!(
existing.primary == primary,
"this box is already the twin of {}; a twin is never re-pointed",
existing.primary
);
return twin_public_key(root);
}
anyhow::ensure!(
!primary.trim().is_empty(),
"the primary's address is required"
);
prepare_replay(stores)?;
if !root.join(TWIN_KEY).exists() {
let key = gunnar_core::mint_ed25519()?;
crate::durable::write_atomic_unframed(root, TWIN_KEY, &key)?;
use std::os::unix::fs::PermissionsExt as _;
std::fs::set_permissions(root.join(TWIN_KEY), std::fs::Permissions::from_mode(0o600))?;
}
TwinConfig {
primary: primary.trim().to_string(),
joined_at_unix_ms: now_ms(),
}
.save(root)?;
stores.mark_twin();
forget_mirrors(stores);
stores.browse().store(true, Ordering::SeqCst);
stores.writes_frozen().store(true, Ordering::SeqCst);
twin_public_key(root)
}
/// **A twin keeps no per-box mirror.** The metadata mirror (`slow`'s `meta` dir) holds DERIVED
/// copies — the identity table, `replication.json`, `webhooks.json` — and a fresh box writes its
/// own at first open, before anybody can tell it to become a twin. Those copies are cache on a
/// twin (C7): removed here, and never written again once the box is marked. Best effort: a file
/// that is not there is the goal, and a removal that fails is one WARN naming the path.
pub fn forget_mirrors(stores: &ServerStores) {
let meta = stores.slow().path_of(crate::slow::META_DIR);
#[allow(unused_mut)]
let mut names = vec![crate::stores::IDENTITY_MIRROR];
// A build without the feature never wrote that mirror, so it has nothing to forget.
#[cfg(feature = "replication")]
names.push(crate::mirror::REPLICATION_CONFIG);
#[cfg(feature = "webhooks")]
names.push(crate::webhook::WEBHOOK_CONFIG);
for name in names {
let path = meta.join(name);
match std::fs::remove_file(&path) {
Ok(()) => {
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, path = %path.display(), "removed a per-box mirror: a twin keeps none")
}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
Err(err) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, path = %path.display(), %err, "could not remove a per-box mirror on a twin")
}
}
}
}
/// The twin's key.
pub fn twin_private_key(root: &Path) -> anyhow::Result<gunnar_core::ssh_key::PrivateKey> {
let bytes = std::fs::read(root.join(TWIN_KEY))
.with_context(|| format!("reading {}", root.join(TWIN_KEY).display()))?;
gunnar_core::ssh_key::PrivateKey::from_openssh(&bytes).map_err(|e| anyhow!("{TWIN_KEY}: {e}"))
}
/// The twin's public key line.
pub fn twin_public_key(root: &Path) -> anyhow::Result<String> {
let key = twin_private_key(root)?;
let mut line = key.public_key().to_openssh().map_err(|e| anyhow!("{e}"))?;
line.push_str(" gunnar-twin");
Ok(line)
}
/// **The owner of an in-process restore** ([`restore`], [`restore_from`]): a caller that
/// holds the data root itself, and so needs no key to prove anything to this box.
pub const RESTORE_OWNER_LOCAL: &str = "local";
/// The refusal word for a resume by somebody who did not start the restore.
pub const RESTORE_NOT_YOURS: &str = "restore-not-yours";
/// **A resume by somebody else, refused** (h-v1-6a, 2026-09-28). Its own type so the
/// control plane answers it `PERMISSION_DENIED` — the caller is known and the answer is no —
/// rather than the `FAILED_PRECONDITION` of "this box is not fresh".
#[derive(Debug)]
pub struct RestoreNotYours {
/// Who the [`RESTORING`] marker says started it.
pub started_by: String,
/// Who asked to resume it.
pub asked_by: String,
}
impl std::fmt::Display for RestoreNotYours {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"REFUSED [{RESTORE_NOT_YOURS}] this box is mid-restore and the restore was started by {}; {} did not start it \
and cannot resume it. Only the key that started a restore resumes it — run the same replay again with the \
same key (monetize's restore lane, or `gunnar control twin replay -i <that key>`). A box whose starter is \
gone for good is re-imaged, which makes it fresh again; nothing of the stream so far was touched",
self.started_by, self.asked_by
)
}
}
impl std::error::Error for RestoreNotYours {}
/// The [`RESTORING`] marker's content: the word, then who started it.
fn restoring_marker(owner: &str) -> String {
format!("restoring\nowner {owner}\n")
}
/// Who started the restore in progress, out of the [`RESTORING`] marker. `None` for a marker
/// with no owner line — one written before 2026-09-28, when the marker was the bare word.
fn restoring_owner(root: &Path) -> anyhow::Result<Option<String>> {
let text = std::fs::read_to_string(root.join(RESTORING))
.with_context(|| format!("reading {}", root.join(RESTORING).display()))?;
Ok(text
.lines()
.find_map(|l| l.strip_prefix("owner "))
.map(|o| o.trim().to_owned())
.filter(|o| !o.is_empty()))
}
/// **One lock per data root around `restore_begin`'s check-then-mark** — so two streams
/// arriving together cannot both find the box fresh (or both find no owner) and both mark
/// it. Per ROOT rather than one for the process because a test process holds many boxes.
fn restore_begin_lock(root: &Path) -> Arc<std::sync::Mutex<()>> {
static LOCKS: std::sync::OnceLock<
std::sync::Mutex<std::collections::HashMap<PathBuf, Arc<std::sync::Mutex<()>>>>,
> = std::sync::OnceLock::new();
let mut locks = LOCKS
.get_or_init(Default::default)
.lock()
.unwrap_or_else(|p| p.into_inner());
Arc::clone(locks.entry(root.to_path_buf()).or_default())
}
/// **`Bootstrap.Restore`, the start**, for a caller holding the data root itself
/// ([`RESTORE_OWNER_LOCAL`]). See [`restore_begin_by`].
pub fn restore_begin(stores: &ServerStores) -> anyhow::Result<u64> {
restore_begin_by(stores, RESTORE_OWNER_LOCAL)
}
/// **`Bootstrap.Restore`, the start**: the next sequence this box takes. A
/// box already restoring continues at its head + 1 — **only for the `owner`
/// that started it**; otherwise it must be fresh ([`prepare_replay`]), starts at
/// 1, and the marker records `owner`.
///
/// `owner` is who the caller PROVED to be — `Bootstrap.Restore` passes the
/// fingerprint of the key its credential verified under (h-v1-6a, 2026-09-28).
/// Until then the resume branch below asked nothing at all: once
/// `txlog-restoring` existed, anybody who reached :50051 could stream the rest
/// of the box's history into it, KEK and identity table included. A marker
/// written before that day names no owner; the first verified caller adopts it,
/// said out loud — the credential it presented is what keeps a stranger out.
///
/// The check and the mark are one step under [`restore_begin_lock`].
pub fn restore_begin_by(stores: &ServerStores, owner: &str) -> anyhow::Result<u64> {
anyhow::ensure!(
!owner.trim().is_empty() && !owner.contains('\n'),
"a restore needs the name of who is restoring; {owner:?} is not one"
);
let lock = restore_begin_lock(stores.root());
let _held = lock.lock().unwrap_or_else(|p| p.into_inner());
if is_restoring(stores.root()) {
match restoring_owner(stores.root())? {
Some(started_by) if started_by != owner => {
return Err(RestoreNotYours {
started_by,
asked_by: owner.to_owned(),
}
.into());
}
Some(_) => {}
None => {
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
owner,
"this box is mid-restore under a marker that names no owner (written before 2026-09-28); {owner} resumes it and owns it from now on"
);
crate::durable::write_atomic_unframed(
stores.root(),
RESTORING,
restoring_marker(owner).as_bytes(),
)?;
}
}
}
// One stream, one staging area: whatever a stream that BROKE left staged is not part of
// this one ([`snapshot::forget_restore_stager`]). After the owner check, so a refused
// caller throws away nothing of the starter's.
snapshot::forget_restore_stager(stores.root());
if is_restoring(stores.root()) {
stores.txlog().set_follower(true);
// The stream that broke may have appended its last entry without applying it: apply
// the head again before taking more (the follower's `reapply` rule).
let head = stores.txlog().head_seq();
if head > 0 && head >= stores.txlog().first_seq() && !snapshot::installing(stores.root()) {
let e = stores.txlog().read(head)?;
apply(stores, &e)
.with_context(|| format!("re-applying seq {head} before the restore resumes"))?;
}
return Ok(stores.txlog().head_seq() + 1);
}
anyhow::ensure!(
!stores.is_twin(),
"a twin is never restored into; restore into a fresh primary"
);
prepare_replay(stores)?;
// A fresh box holds no part of anybody's entry, nor of anybody's snapshot.
intake::forget(stores.root())?;
snapshot::forget_restore_snapshot(stores.root())?;
crate::durable::write_atomic_unframed(
stores.root(),
RESTORING,
restoring_marker(owner).as_bytes(),
)?;
Ok(1)
}
/// **What a restore costs on the box taking it, per byte the sender streams.**
///
/// A restore writes each frame TWICE: once into this box's own log ([`restore_apply`] does
/// `append_at` before `apply`, durable first) and once into the store the frame builds — a pack
/// into the archive, an LFS object into its file, a document into `control.redb`. So the floor
/// is 2. The margin above it is the engine's own overhead on ingest: the archive's oid index
/// (`objects.tail`, measured at ~0.8 % of a 200 MB pack in the note's twin-image table), the
/// ref log, redb's B-tree pages, and the filesystem's own metadata on tens of thousands of
/// small files.
///
/// **Deliberately an ESTIMATE, and it is named as one in the refusal.** A tight number would
/// need the sender to say what its stores measure rather than what its log measures, which is a
/// wire change this rule does not need in order to catch the case it exists for: a set that is
/// half the size it left with. It is a floor, not a promise — a restore that passes this can
/// still run out on a set that only just fits, and that failure is the ordinary mid-stream one
/// (the box stays `restoring`, resumable), not a silent short restore.
pub const RESTORE_BYTES_PER_LOG_BYTE: u64 = 2;
/// Room the box keeps for itself on top of the restore: the exploded read cache the engine
/// grows on the first clone, the crash and audit records, and the slack a full xfs needs to
/// keep allocating. 5 % of the volume, never less than 256 MiB.
fn reserve_bytes(total: u64) -> u64 {
(total / 20).max(256 << 20)
}
/// **Does this box have room for the restore, asked BEFORE the first byte of it lands?**
///
/// The case this exists for is a restore onto DIFFERENT hardware: the data set left a machine
/// with four members and arrives at one with two, which may hold less total space than the set
/// being restored. Without this the stream fills the volume and the apply fails somewhere in
/// the middle — recoverable (the box stays `restoring` and nothing says "restored"), but the
/// operator learns it after an hour of streaming and learns it as an ENOSPC rather than as a
/// number. With it the box refuses at the first chunk and says how much is short.
///
/// `announced_bytes` of 0 means the sender did not say (an older `gunnar control twin replay`,
/// or a caller streaming a snapshot whose size it cannot know in advance): no refusal, because
/// a guess is not a measurement. The whole check is one `statvfs`.
pub fn room_for_restore(root: &Path, announced_bytes: u64) -> anyhow::Result<()> {
if announced_bytes == 0 {
return Ok(());
}
let (total, used) = volume_bytes(root);
if total == 0 {
// Unmeasurable is not "full": a box whose statvfs failed is one this must not refuse.
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, root = %root.display(), "the data volume could not be measured; the restore's room is not checked");
return Ok(());
}
let free = total.saturating_sub(used);
let need = announced_bytes
.saturating_mul(RESTORE_BYTES_PER_LOG_BYTE)
.saturating_add(reserve_bytes(total));
if free >= need {
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
announced_bytes, need, free, total,
"room for the restore: the data set on this machine holds it"
);
return Ok(());
}
anyhow::bail!(
"NOT ENOUGH ROOM for this restore, and nothing has been written: the sender streams \
{announced} of log, which needs about {need} on this box ({factor}× the log — a restore \
writes every frame into this box's own log AND into the store it builds — plus {reserve} \
the box keeps for its caches and slack), and this data set has {free} free of {total}. \
SHORT BY {short}. This box is being restored onto a SMALLER data set than the one the \
twin's log came from: give it more or larger members and restore again. NOTHING of the \
twin's log has landed — this box's log is still empty and it stays `restoring`, so the \
same replay into a bigger set starts at seq 1 and loses nothing. The estimate is \
DELIBERATELY conservative and a long-lived box's log is larger than the stores it \
builds; if you have read these numbers and this set does hold it, re-run the replay \
with `--ignore-room`.",
announced = human(announced_bytes),
need = human(need),
factor = RESTORE_BYTES_PER_LOG_BYTE,
reserve = human(reserve_bytes(total)),
free = human(free),
total = human(total),
short = human(need - free),
)
}
/// Bytes as an operator reads them. Small enough to live here rather than pull a crate in.
fn human(n: u64) -> String {
const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"];
let mut v = n as f64;
let mut u = 0;
while v >= 1024.0 && u + 1 < UNITS.len() {
v /= 1024.0;
u += 1;
}
if u == 0 {
format!("{n} B")
} else {
format!("{v:.1} {}", UNITS[u])
}
}
/// One entry of a restore: durable in the log at the twin's number, then applied.
pub fn restore_apply(stores: &ServerStores, e: &Entry) -> anyhow::Result<Applied> {
stores.txlog().append_at(e.seq, e.kind, &e.meta, &e.blob)?;
apply(stores, e)
}
/// The end of a restore: the marker and the fresh-docs list go. The log stays
/// follower-only in THIS process — it holds the fresh box's KEK in memory and
/// the replayed secrets are sealed under the primary's — so the box restarts,
/// and opens as the primary it was.
/// `announced_head` is the head the sender said it streams through (`LogChunk.restore_head`,
/// the largest seen; 0 = not announced): a box below it has taken a SHORT stream — one that
/// ended cleanly although the sender's fetch broke — and stays restoring, resumable by
/// resending from its head + 1.
pub fn restore_finish(
stores: &ServerStores,
announced_head: u64,
log_id_of_source: &str,
) -> anyhow::Result<u64> {
finish_restore(stores, announced_head, log_id_of_source, false)
}
/// **[`restore_finish`] of a restore streamed over the network** (`Bootstrap.Restore`): the stream
/// MUST name the log it restores, and so must every earlier stream of the same restore (stinky
/// row 7, 2026-10-10). The lineage check lets an empty id through on either side — right for a
/// local replay, which reads the id off the twin's disk — so a sender that named nothing could
/// finish a box on another history whenever the sequence numbers happened to fit. Every sender
/// in the tree names it (`LogHead.log_id`, copied onto every chunk).
pub fn restore_finish_streamed(
stores: &ServerStores,
announced_head: u64,
log_id_of_source: &str,
) -> anyhow::Result<u64> {
finish_restore(stores, announced_head, log_id_of_source, true)
}
fn finish_restore(
stores: &ServerStores,
announced_head: u64,
log_id_of_source: &str,
streamed: bool,
) -> anyhow::Result<u64> {
// **What this restore was told it restores holds for every resend** (estate-box3
// plain6): a resend whose sender's fetch failed before its first entry announces
// nothing, and judged alone it would finish a box that an EARLIER stream of the
// same restore announced to reach further. The largest head and the source's log
// id ever announced are kept beside the marker until the restore finishes.
let kept = restore_announced(stores.root())?;
let now_id = log_id_of_source.trim();
if streamed {
anyhow::ensure!(
!now_id.is_empty(),
"this restore stream named no log (LogChunk.log_id is empty): a box is restored from one named history, never from whatever numbers fit — this box stays restoring; resend the twin's log with its log id"
);
if let Some(kept) = &kept {
anyhow::ensure!(
!kept.log_id.is_empty(),
"an earlier stream of this restore named no log ({RESTORE_ANNOUNCED} says head {} and no log id): which history this box holds is unknown — this box stays restoring; start the restore again on a fresh box",
kept.head
);
}
}
let Announced {
head: kept_head,
log_id: kept_id,
} = kept.unwrap_or_default();
anyhow::ensure!(
kept_id.is_empty() || now_id.is_empty() || kept_id == now_id,
"this restore began as log {kept_id} and a resend is log {now_id}: one restore is one history — this box stays restoring"
);
let announced_head = announced_head.max(kept_head);
let log_id_of_source = if now_id.is_empty() {
kept_id.as_str()
} else {
now_id
};
if announced_head > 0 || !log_id_of_source.is_empty() {
crate::durable::write_atomic_unframed(
stores.root(),
RESTORE_ANNOUNCED,
format!("{announced_head} {log_id_of_source}\n").as_bytes(),
)?;
}
// A stream that ended with nothing in it is not a restore: a fetch that failed before its
// first entry looks exactly like this from here. The box stays restoring, so the operator
// resends the twin's log rather than restarting into an empty primary.
anyhow::ensure!(
stores.txlog().head_seq() > 0,
"nothing was restored (the stream carried no entry); this box stays restoring — resend the twin's log"
);
let head = stores.txlog().head_seq();
anyhow::ensure!(
head >= announced_head,
"the stream ended at seq {head} but the sender announced the twin's head {announced_head}; a short stream is not a restore — this box stays restoring, resend the twin's log"
);
// **This box IS the history it was just handed** (2026-09-21, lane T4).
// `prepare_replay` dropped whatever id this fresh box had minted for itself;
// the stream's id goes on now, BEFORE the restoring marker comes off, so no
// open can ever see a restored box without the name of what it restored. A
// sender older than this change announces nothing, and then the box mints
// its own at the next open — the pre-2026-09-21 behaviour exactly.
let announced_id = log_id_of_source.trim();
if !announced_id.is_empty() {
pin_log_id(stores.root(), announced_id)
.context("taking the identity of the log this box was restored from")?;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, log_id = %announced_id, "this box IS log {announced_id} now: the twin it was restored from keeps following it without being re-joined");
}
// **And a NEWER generation of it** (stinky row 18): the same name as the
// main it replaces, ordered after it, so the twin can refuse that old main
// if it ever comes back. Stamped before the restoring marker comes off, like
// the id.
let generation =
stamp_generation(stores.root()).context("stamping this restore's generation of the log")?;
forget_fence(stores.root())
.context("a restored box is the newest generation: no fence from an older life")?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, generation, "this restore is generation {generation} of its log: a twin refuses any older one from here");
let _ = std::fs::remove_file(stores.root().join(FRESH_DOCS));
// The next open re-seals every secret under a KEK this box owns ([`reseal_after_restore`]).
// Written BEFORE the restoring marker goes, so no open can see "restored" without "re-seal".
crate::durable::write_atomic_unframed(stores.root(), RESEAL, b"reseal\n")?;
// DURABLY: the box is "restarted" next, and on the appliance that can be a stop that
// loses the guest's page cache. An unlink that did not reach the disk reopens the box
// mid-restore — read-only, forever (`crate::durable::remove_durable`, MEASURED).
crate::durable::remove_durable(stores.root(), RESTORING)?;
let _ = std::fs::remove_file(stores.root().join(RESTORE_ANNOUNCED));
// Nothing staged outlives the restore (restore-large-entry): an entry staged is applied
// before its staging goes, so whatever is left here is no part of this box's history.
let _ = intake::forget(stores.root());
snapshot::forget_restore_snapshot_or_warn(stores.root());
Ok(stores.txlog().head_seq())
}
/// `<data-root>/txlog-restore-announced`: `<head> <log id>` — the largest head
/// and the source's log id any stream of the restore under way announced.
pub const RESTORE_ANNOUNCED: &str = "txlog-restore-announced";
/// What a restore under way was announced to be ([`RESTORE_ANNOUNCED`]).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Announced {
/// The largest head any stream announced; 0 = none.
pub head: u64,
/// The source's log id; empty = none named.
pub log_id: String,
}
/// **[`RESTORE_ANNOUNCED`] could not be trusted** (stinky row 4, 2026-10-10): it used to be read
/// with `unwrap_or_default()` and its head parsed with `.ok().unwrap_or(0)`, so a marker that
/// could not be read, or did not parse, meant "nothing announced" — and a restore could finish
/// on a shorter head, as another log. Now the restore refuses by name and the box stays restoring.
#[derive(Debug, thiserror::Error)]
pub enum RestoreMarkerError {
/// The marker is there and could not be read.
#[error("the restore marker {path} could not be read ({source}): what this restore was announced to reach is unknown — this box stays restoring")]
Unreadable {
/// The marker.
path: std::path::PathBuf,
/// Why.
source: std::io::Error,
},
/// The marker does not say `<head> [<log id>]`.
#[error("the restore marker {path} is corrupt ({what}: {text:?}): what this restore was announced to reach is unknown — this box stays restoring")]
Corrupt {
/// The marker.
path: std::path::PathBuf,
/// What is wrong with it.
what: &'static str,
/// What it holds.
text: String,
},
}
/// What the restore under way was announced to be; `None` when no stream announced anything (no
/// marker). A marker that cannot be read or parsed is a [`RestoreMarkerError`], never "nothing".
pub fn restore_announced(root: &Path) -> Result<Option<Announced>, RestoreMarkerError> {
let path = root.join(RESTORE_ANNOUNCED);
let text = match std::fs::read_to_string(&path) {
Ok(t) => t,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(source) => return Err(RestoreMarkerError::Unreadable { path, source }),
};
let corrupt = |what| RestoreMarkerError::Corrupt {
path: path.clone(),
what,
text: text.clone(),
};
let mut words = text.split_whitespace();
let head = words
.next()
.ok_or_else(|| corrupt("it is empty"))?
.parse::<u64>()
.map_err(|_| corrupt("its head is not a number"))?;
let log_id = words.next().unwrap_or("").to_string();
if words.next().is_some() {
return Err(corrupt("it holds more than a head and a log id"));
}
Ok(Some(Announced { head, log_id }))
}
/// **What a box does once a restore FINISHED** (c-no-restart): the restored
/// box must restart to open as the primary it was (the replayed secrets are
/// sealed under the dead main's KEK, the log is follower-only in this
/// process). Nobody outside restarted it — monetize only logged
/// `needs_restart` — so the box stayed half-open for ever.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AfterRestore {
/// The appliance's PID 1: sync and reboot itself, a few seconds after the
/// answer went out.
Reboot,
/// Anything else (a container, a developer's process): say so loudly; the
/// operator restarts it.
Operator,
}
/// Pure: which [`AfterRestore`], for a process that is (or is not) PID 1 of
/// the appliance's own kernel.
pub fn after_restore(is_init: bool, appliance: bool) -> AfterRestore {
if is_init && appliance {
AfterRestore::Reboot
} else {
AfterRestore::Operator
}
}
/// How long the restored box waits before it reboots: the `Bootstrap.Restore`
/// answer must reach the lane first.
pub const REBOOT_AFTER_RESTORE: std::time::Duration = std::time::Duration::from_secs(5);
/// **Restart the restored box** ([`after_restore`]): on the appliance, sync
/// and reboot on a thread of its own after [`REBOOT_AFTER_RESTORE`]; anywhere
/// else, say that the operator must. Returns what was decided.
pub fn restart_after_restore() -> AfterRestore {
let how = after_restore(
appliance_pid1::net::is_init(),
appliance_pid1::net::booted_by_korp_start(),
);
match how {
AfterRestore::Reboot => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, after_s = REBOOT_AFTER_RESTORE.as_secs(), "RESTORED from the twin: this box reboots itself to open as the primary it was");
// A28 M18, stays a thread: one wait, then the reboot; a start that fails is said.
let spawned = std::thread::Builder::new()
.name("restore-reboot".into())
.spawn(|| {
std::thread::sleep(REBOOT_AFTER_RESTORE);
crate::narrate::reboot_now();
});
if let Err(e) = spawned {
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, error = %e, "the reboot after the restore could not be scheduled; restart this box");
}
}
AfterRestore::Operator => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, "RESTORED from the twin: RESTART this process — it opens as the primary it was (not the appliance's PID 1, so it does not reboot itself)");
}
}
how
}
#[cfg(test)]
mod after_restore_tests {
use super::*;
#[test]
fn only_the_appliances_pid_1_reboots_itself_after_a_restore() {
assert_eq!(after_restore(true, true), AfterRestore::Reboot);
assert_eq!(
after_restore(true, false),
AfterRestore::Operator,
"a container's PID 1 never reboots the host"
);
assert_eq!(after_restore(false, true), AfterRestore::Operator);
assert_eq!(after_restore(false, false), AfterRestore::Operator);
}
}
/// `<data-root>/txlog-reseal`: a restore finished; the next open re-seals under a KEK of this
/// box's own. Per box, never framed.
pub const RESEAL: &str = "txlog-reseal";
/// `<control-dir>/secrets.kek.new`: the KEK being re-sealed to, durable before the first secret
/// moves, so a crash mid-way resumes with the same key.
const KEK_NEXT: &str = "secrets.kek.new";
/// Is a re-seal owed under `root`?
pub fn reseal_pending(root: &Path) -> bool {
root.join(RESEAL).exists()
}
/// **A restored box takes a KEK of its own** (`.nornir/twin-transaction-log.md` §"The KEK after
/// a restore"). The replayed `secrets.kek` is the dead primary's; every sealed secret is opened
/// with it and sealed again under a new key (or `GUNNAR_SECRET_KEK` when the box is handed one),
/// and the old key leaves the disk. Each step is a frame first — a `Kek` frame with the new key,
/// then one `Secret` frame per re-sealed value — so the twin following this box holds the new
/// key and the re-sealed bytes, and a later restore from it opens them.
///
/// Crash-safe by construction: the new key is durable in `secrets.kek.new` before any secret
/// moves; a value that already opens under the new key is skipped; the rename over
/// `secrets.kek` and the marker's removal come last. Run once at open, before any sealed store
/// is built. Returns the KEK the box now uses.
pub fn reseal_after_restore(
root: &Path,
control_dir: &Path,
log: &Log,
rows: &std::collections::BTreeMap<String, Option<Vec<u8>>>,
raw: &dyn gunnar_store::SecretStore,
) -> anyhow::Result<gunnar_core::Kek> {
use gunnar_core::Kek;
let path = gunnar_core::seal::kek_path(control_dir);
let key32 = |bytes: Vec<u8>, what: &Path| -> anyhow::Result<Kek> {
let arr: [u8; 32] = bytes
.as_slice()
.try_into()
.map_err(|_| anyhow!("{} is {} bytes, not 32", what.display(), bytes.len()))?;
Ok(Kek::from_bytes(arr))
};
let old = key32(
std::fs::read(&path)
.with_context(|| format!("reading the replayed KEK {}", path.display()))?,
&path,
)?;
let next_path = control_dir.join(KEK_NEXT);
let env = Kek::from_env()?;
let new = match &env {
Some(k) => k.clone(),
None => match std::fs::read(&next_path) {
Ok(bytes) => key32(bytes, &next_path)?,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
let k = Kek::generate()?;
write_key_file(control_dir, KEK_NEXT, &hex::decode(k.to_hex())?)?;
k
}
Err(e) => return Err(e).with_context(|| format!("reading {}", next_path.display())),
},
};
let mut moved = 0usize;
let mut dropped = 0usize;
if new.to_hex() != old.to_hex() {
log.append(Kind::Kek, &Meta::Kek {}, &hex::decode(new.to_hex())?)
.context("framing the restored box's own KEK")?;
for (key, row) in rows {
let Some(sealed) = row else { continue };
// A legacy plaintext row is the migrating store's business, not a sealed value.
if !gunnar_core::is_sealed(sealed) || gunnar_core::unseal(&new, key, sealed).is_ok() {
continue;
}
// **A row this box wrote ITSELF, under a KEK that no longer exists** (MEASURED
// 2026-09-17, lane TWIN-COMPLETENESS). A fresh box mints secrets of its own before
// it is restored into — its host key at first boot, its issuer at first start, and
// any key a request minted on first use. `prepare_replay` gives up its DOCUMENTS
// (`FRESH_DOCS`) and its numbers, and since the same commit its secrets too; but a
// box restored by an older build, or one whose fresh row the replay had no frame
// for, reaches here with a row sealed under a KEK the `Kek` frame has already
// overwritten on disk. That row opens under NEITHER key, and this used to be a `?`:
// the restored box then REFUSED TO OPEN, every time, for ever —
//
// re-sealing a restored box's secrets under its own KEK: opening fresh-only/key
// with the replayed KEK: … wrong key-encrypting key …
//
// — a box that had been restored successfully and could not start. The value is
// unrecoverable by construction (its KEK is gone) and it is not the primary's, so
// it is DROPPED, loudly and by name, rather than kept as the thing that bricks the
// box. Nothing of the restore is lost: every row the twin's log carried opens under
// the replayed KEK, which is the branch above.
let plain = match gunnar_core::unseal(&old, key, sealed) {
Ok(plain) => plain,
Err(err) => {
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
key = %key,
%err,
"a secret that opens under NEITHER the replayed KEK nor this box's own: it \
was written by this box before the restore, under a key that is gone. \
DROPPED — it is not the restored primary's and it cannot be read by \
anyone. If something needs it, it will be minted again on first use"
);
raw.delete(key)
.with_context(|| format!("dropping the unopenable {key}"))?;
dropped += 1;
continue;
}
};
let resealed = gunnar_core::seal(&new, key, &plain)?;
log.append(
Kind::Secret,
&Meta::Secret {
key: key.clone(),
deleted: false,
},
&resealed,
)
.with_context(|| format!("framing the re-sealed {key}"))?;
raw.put(key, &resealed)
.with_context(|| format!("storing the re-sealed {key}"))?;
moved += 1;
}
}
if env.is_some() {
// The box is handed its key; the dead primary's must not stay on the disk.
match std::fs::remove_file(&path) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => {
return Err(e)
.with_context(|| format!("removing the replayed KEK {}", path.display()))
}
}
} else {
std::fs::rename(&next_path, &path)
.with_context(|| format!("installing the new KEK at {}", path.display()))?;
}
// The KEK's rename (or removal) is durable BEFORE the marker that owes it goes (stinky
// row 9): a power cut between the two must find the re-seal still owed, never a box that
// thinks it re-sealed with the replayed KEK back on its disk.
crate::durable::sync_dir(control_dir).context("flushing the re-sealed KEK into place")?;
crate::durable::remove_durable(root, RESEAL)?;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, moved, dropped, "a restored box re-sealed its secrets under a KEK of its own; the replayed KEK is gone from this disk");
Ok(new)
}
/// A 0600 key file under `dir`, atomic and durable.
fn write_key_file(dir: &Path, name: &str, bytes: &[u8]) -> anyhow::Result<()> {
use std::io::Write as _;
std::fs::create_dir_all(dir)?;
let tmp = dir.join(format!("{name}.tmp"));
let mut options = std::fs::OpenOptions::new();
options.write(true).create(true).truncate(true);
{
use std::os::unix::fs::OpenOptionsExt as _;
options.mode(0o600);
}
let mut f = options.open(&tmp)?;
f.write_all(bytes)?;
f.sync_all()?;
drop(f);
std::fs::rename(&tmp, dir.join(name))?;
crate::durable::sync_dir(dir)
}
// ── the follower ────────────────────────────────────────────────────────────
/// What one [`Follower::step`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Progress {
/// Applied this many entries; the twin's head is now `head`.
Applied { count: u64, head: u64 },
/// Nothing new on the primary.
CaughtUp { head: u64 },
/// The primary retains from `first`; the twin needs `wanted`. Stopped.
SnapshotNeeded { wanted: u64, first: u64 },
/// The hole was closed by a snapshot taken at `at`; the twin follows from `at + 1`.
SnapshotInstalled { at: u64 },
/// The primary's head is BELOW this twin's own — a different log. Stopped.
Diverged {
/// What this twin has applied.
own: u64,
/// What the box at the other end says its head is.
primary_head: u64,
},
}
/// Consecutive failed polls after which a follower whose primary has another
/// address hands back to the supervisor to re-dial.
pub const FAILOVER_AFTER: u32 = 5;
/// The twin's loop over one source into one set of stores.
pub struct Follower {
stores: Arc<ServerStores>,
source: Box<dyn LogSource>,
/// Blob bytes per fetch.
fetch_bytes: u64,
/// **The log is one entry ahead of the stores.** The twin appends BEFORE it applies
/// (durable first), so an apply that failed — or a process that died between the two —
/// leaves the head entry unapplied. Set at construction (a restart does not know) and on
/// every append; cleared once that entry applied. Measured 2026-09-15 under KVM: without
/// it the next poll resumed at head + 1 and the twin reported `applied == head` with the
/// failed entries' refs missing from its stores.
reapply: AtomicBool,
/// The run of failed polls being lived through — kept across the supervisor's re-dials
/// ([`Follower::sharing_failures`]), so a primary that stays unreachable is said once.
failures: Arc<Mutex<Option<Failing>>>,
/// The peak resident set this follower last said ([`Follower::say_memory`]), bytes; 0 = never.
said_peak: std::sync::atomic::AtomicU64,
/// The applied marker ([`gunnar_core::TWIN_APPLIED`]) this follower last made durable.
marked: std::sync::atomic::AtomicU64,
/// The volume a snapshot is staged on ([`snapshot::Disk`]; a test caps one).
volume: Arc<dyn snapshot::Volume>,
}
/// **This process's memory, from the kernel**: `(VmHWM, VmRSS)` of `/proc/self/status` in bytes —
/// the peak resident set since the process started, and the resident set now. `None` where the
/// file cannot be read (not Linux).
pub fn own_memory() -> Option<(u64, u64)> {
let s = std::fs::read_to_string("/proc/self/status").ok()?;
let kib = |key: &str| {
s.lines()
.find_map(|l| l.strip_prefix(key))
.and_then(|v| v.split_whitespace().next())
.and_then(|v| v.parse::<u64>().ok())
};
Some((kib("VmHWM:")? << 10, kib("VmRSS:")? << 10))
}
/// A following twin says its memory again once its PEAK rose this much above the figure it last
/// said — so a twin whose memory is bounded says it once, and one that keeps climbing says so in
/// steps, never more often than its memory grows.
pub const SAY_MEMORY_STEP: u64 = 32 << 20;
/// Whether a peak of `peak` bytes is news after `said` was said (0 = nothing said yet).
pub fn memory_is_news(said: u64, peak: u64) -> bool {
said == 0 || peak >= said.saturating_add(SAY_MEMORY_STEP)
}
/// **A run of failed polls** (L1, estate-box check 3): what failed (its shape), since when,
/// how many, and when it was last said. A recharge makes a twin's primary unreachable, then
/// a blank box that does not prove itself (`BadSignature`) for minutes: every poll said
/// `poll failed; retrying` at WARN (30/min) and every hand-back `re-dialling` at WARN. The
/// first failure of a run is said at WARN; the same failure again only every [`RESAY_FAILING`];
/// the end of the run once, at INFO.
#[derive(Debug, Clone)]
pub struct Failing {
shape: String,
since: std::time::Instant,
said: std::time::Instant,
polls: u64,
}
/// While the same failure lasts, it is said again this often.
pub const RESAY_FAILING: Duration = Duration::from_secs(300);
/// An error's shape: its words without its numbers, so the same failure on another port or
/// with another nonce is the same failure.
/// Whether a poll failure is the primary refusing this twin's credential for clock SKEW —
/// [`gunnar_grpc::auth::CLOCK_SKEW_PHRASE`] in the refusal, which the gate's `Stale` status
/// carries — and not any other refusal or a transport failure.
pub(crate) fn refused_for_skew(err: &anyhow::Error) -> bool {
format!("{err:#}").contains(gunnar_grpc::auth::CLOCK_SKEW_PHRASE)
}
fn failure_shape(err: &anyhow::Error) -> String {
let text = format!("{err:#}");
let mut out = String::with_capacity(text.len().min(200));
let mut in_num = false;
for c in text.chars().take(200) {
if c.is_ascii_digit() {
if !in_num {
out.push('#');
}
in_num = true;
} else {
in_num = false;
out.push(c);
}
}
out
}
/// **The follower's end of a streamed range** (twin-streaming, 2026-10-06): each entry straight
/// into this box's OWN log as it arrives — a [`txlog::Pending`] at the primary's number, so the
/// gap rule holds piece by piece — and, once whole, [`take_logged`]. The frame does not exist
/// until its last byte is in: the length is held to the declared one in both directions
/// (`Pending::write` refuses more, `Pending::commit` refuses less), the primary hands over an
/// entry's last piece only after its payload crc checked, and an entry that never ends is cut off
/// the segment when the `Pending` drops. What this box holds of an entry in flight is the piece
/// in hand.
struct Appender<'a> {
stores: &'a ServerStores,
reapply: &'a AtomicBool,
/// The entry being received: its frame in progress, its kind, meta and blob length.
open: Option<(txlog::Pending<'a>, u16, Vec<u8>, u64)>,
/// The sequence due next.
next: u64,
/// Entries appended and taken.
count: u64,
/// **The group commit of a store-only twin** (perf twin-apply-fsync, 2026-10-06); `None` on a
/// box being restored, which keeps one durable frame and one full apply per entry.
group: Option<Group>,
/// The applied marker this follower last made durable.
marked: &'a std::sync::atomic::AtomicU64,
}
/// Entries written into the twin's log and not yet durable, in order, with what their take needs.
struct Group {
entries: Vec<(u64, u16, Vec<u8>)>,
bytes: u64,
since: std::time::Instant,
}
impl Group {
fn new() -> Self {
Group {
entries: Vec::new(),
bytes: 0,
since: std::time::Instant::now(),
}
}
fn full(&self) -> bool {
self.entries.len() >= GROUP_ENTRIES
|| self.bytes >= GROUP_BYTES
|| self.since.elapsed() >= GROUP_AGE
}
}
impl Appender<'_> {
/// **End the group**: ONE `fdatasync` makes every entry of it durable in this twin's log —
/// from here the head says so, and the head is what this twin acknowledges — then the
/// control-plane entries among them are taken, in order, into ONE durable redb commit
/// ([`take_group`]). A half-received entry is cut first. Nothing pending = nothing done.
fn flush(&mut self) -> anyhow::Result<()> {
drop(self.open.take());
let Some(g) = self.group.as_mut() else {
return Ok(());
};
if g.entries.is_empty() {
return Ok(());
}
let entries = std::mem::take(&mut g.entries);
g.bytes = 0;
g.since = std::time::Instant::now();
self.stores
.txlog()
.sync()
.context("making a group of entries durable in this twin's log")?;
self.reapply.store(true, Ordering::SeqCst);
take_group(self.stores, &entries, self.marked)?;
self.reapply.store(false, Ordering::SeqCst);
Ok(())
}
}
/// The name a `ControlDoc` frame's meta moves, if it is one.
fn doc_name(kind: u16, meta: &[u8]) -> Option<String> {
if kind != Kind::ControlDoc as u16 {
return None;
}
match Meta::from_bytes(meta) {
Ok(Meta::ControlDoc { name, .. }) => Some(name),
_ => None,
}
}
/// The document names the fresh-box rule still holds ([`FRESH_DOCS`]); empty when none.
fn fresh_names(root: &Path) -> anyhow::Result<std::collections::BTreeSet<String>> {
match std::fs::read(root.join(FRESH_DOCS)) {
Ok(b) => Ok(serde_json::from_slice(&b)
.with_context(|| format!("{} does not parse", root.join(FRESH_DOCS).display()))?),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Default::default()),
Err(e) => Err(e.into()),
}
}
/// **Take a group of entries that are durable in a twin's log, with ONE durable redb commit**
/// (perf twin-apply-fsync, 2026-10-06).
///
/// The control-plane writes of the whole group go into ONE redb write transaction
/// ([`gunnar_core::ControlDb::begin_group`]) and are committed once, with the applied marker
/// ([`gunnar_core::TWIN_APPLIED`]) at the group's last entry inside the same commit. A crash
/// anywhere leaves redb EXACTLY at its last durable marker — the group is in whole or not at all —
/// so the entries after the marker are re-taken in order onto exactly the state they were first
/// taken onto. That is what makes a re-take of more than one entry safe: a compare-and-swap the
/// primary LOST can only lose again when it is replayed onto the state it lost against. A write
/// that fails rolls the whole group back; the twin stops there and re-takes it next poll.
///
/// A `ControlDoc` the fresh-box rule still names is taken on its own (its `fresh_take` reads the
/// committed document and writes its own file durably): the group so far is committed first. A
/// run of data frames moves no redb page and is marked only every [`MARK_EVERY`] entries.
fn take_group(
stores: &ServerStores,
entries: &[(u64, u16, Vec<u8>)],
marked: &std::sync::atomic::AtomicU64,
) -> anyhow::Result<()> {
let db = stores.control_db();
let mut fresh = fresh_names(stores.root())?;
let mut done: Option<u64> = None;
db.begin_group()?;
// A panic inside the take must not leave the group's redb write transaction open for ever
// (every other writer of this box's control database would wait on it, and the next take
// would be refused "already open"): unwinding drops this, which rolls the group back. On
// every ordinary path the group is already committed or aborted and this finds none.
struct AbortOnDrop<'a>(&'a gunnar_core::ControlDb);
impl Drop for AbortOnDrop<'_> {
fn drop(&mut self) {
self.0.abort_group();
}
}
let _abort = AbortOnDrop(&db);
let taken = (|| -> anyhow::Result<()> {
for (seq, kind, meta) in entries {
let (seq, kind) = (*seq, *kind);
let fresh_doc = if fresh.is_empty() {
None
} else {
doc_name(kind, meta).filter(|n| fresh.contains(n))
};
let applied = if let Some(name) = fresh_doc {
if let Some(prev) = done {
db.commit_group(gunnar_core::TWIN_APPLIED, prev, false)?;
} else {
db.abort_group();
}
let a = take_logged(stores, seq, kind, meta)?;
db.set_mark(gunnar_core::TWIN_APPLIED, seq)?;
marked.store(seq, Ordering::SeqCst);
fresh.remove(&name);
db.begin_group()?;
a
} else {
take_logged(stores, seq, kind, meta)?
};
match applied {
Applied::Done | Applied::Stored => {}
Applied::Lost(what) => {
tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, seq, %what, "replayed a lost CAS")
}
Applied::Unknown(kind) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, seq, kind, "kept a frame of a kind this build does not apply")
}
}
done = Some(seq);
}
Ok(())
})();
match (taken, done) {
(Ok(()), Some(last)) => {
let force = last >= marked.load(Ordering::SeqCst).saturating_add(MARK_EVERY);
// The group's one durable redb commit.
if db
.commit_group(gunnar_core::TWIN_APPLIED, last, force)
.context("committing the group's control-plane writes")?
{
marked.store(last, Ordering::SeqCst);
}
Ok(())
}
(Ok(()), None) => {
db.abort_group();
Ok(())
}
(Err(err), _) => {
db.abort_group();
Err(err)
}
}
}
impl FrameSink for Appender<'_> {
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
if let Some((p, ..)) = &self.open {
anyhow::bail!(
"the primary began seq {seq} with seq {} unfinished",
p.seq()
);
}
if seq != self.next {
return Err(anyhow!(
"the primary sent seq {seq} where {} was due",
self.next
));
}
// Into the twin's own log first — the primary's number.
let pending = self.stores.txlog().begin(Some(seq), kind, meta, blob_len)?;
self.open = Some((pending, kind, meta.to_vec(), blob_len));
Ok(())
}
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
let (p, ..) = self
.open
.as_mut()
.ok_or_else(|| anyhow!("the primary sent blob bytes outside an entry"))?;
let seq = p.seq();
p.write(bytes)
.with_context(|| format!("seq {seq}: writing its blob into this box's log"))?;
Ok(())
}
fn end(&mut self) -> anyhow::Result<()> {
let (p, kind, meta, blob_len) = self
.open
.take()
.ok_or_else(|| anyhow!("the primary ended an entry it never began"))?;
if let Some(g) = self.group.as_mut() {
// Whole in the log, durable with the rest of its group.
let seq = p.commit_unsynced()?;
if g.entries.is_empty() {
g.since = std::time::Instant::now();
}
g.bytes += blob_len + meta.len() as u64;
g.entries.push((seq, kind, meta));
let full = g.full();
self.next = seq + 1;
self.count += 1;
if full {
self.flush()?;
}
return Ok(());
}
let seq = p.commit()?;
self.reapply.store(true, Ordering::SeqCst);
let applied = take_logged(self.stores, seq, kind, &meta)?;
self.reapply.store(false, Ordering::SeqCst);
match applied {
Applied::Done => {}
Applied::Stored => {}
Applied::Lost(what) => {
tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, seq, %what, "replayed a lost CAS")
}
Applied::Unknown(kind) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, seq, kind, "kept a frame of a kind this build does not apply")
}
}
self.next = seq + 1;
self.count += 1;
Ok(())
}
}
/// **The resync's own words** for the status (twin-resync-space, 2026-10-06).
fn resyncing_why(first: u64, own: u64, dropped: Option<&snapshot::Dropped>) -> String {
match dropped {
Some(d) => format!("taking a full copy from its primary (it retains from {first}) — {}", d.sentence()),
None => format!(
"taking a full copy from its primary (it retains from {first}; this twin holds up to {own}) — its previous \
copy is kept until the new one is whole, and it is NOT a current copy until then"
),
}
}
/// **Give this twin's stale copy up for room** (twin-resync-space, 2026-10-06), called by the
/// stager ONCE when the volume runs short mid-resync. The [`snapshot::DROPPED`] marker is written
/// FIRST (durable): from then on the twin serves nothing and a crash restarts the resync. Then the
/// kept base and the log go ([`finish_giving_up`]); the head stays the twin's own. Said at WARN
/// as an ALERT, alarmed (`twin.resync_without_copy`), and written in the status. Answers the
/// bytes freed.
fn give_up_stale_copy(
stores: &ServerStores,
short: &snapshot::Short,
first: u64,
primary_head: u64,
answered: i64,
) -> anyhow::Result<u64> {
let root = stores.root();
let log = stores.txlog();
let head = log.head_seq();
let base_at = snapshot::base_at(root).ok().flatten();
let freed = snapshot::bytes_under(&root.join(snapshot::BASE))
+ snapshot::bytes_under(&root.join(format!("{}.old", snapshot::BASE)))
+ log.bytes();
let d = snapshot::Dropped {
at_unix_ms: now_ms(),
head,
base_at,
freed,
};
snapshot::mark_dropped(root, &d)?;
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
event = "twin.resync_without_copy",
head,
freed,
staged = short.staged,
free = short.free,
total = short.total,
"ALERT the full resync does not fit beside this twin's stale copy ({} staged, {} free of {}{}): {}",
human(short.staged),
human(short.free),
human(short.total),
if short.enospc { ", the volume said ENOSPC" } else { "" },
d.sentence()
);
finish_giving_up(stores)?;
set_status(
root,
TwinStatus {
applied_seq: head,
primary_head_seq: primary_head,
state: State::Resyncing,
why: resyncing_why(first, head, Some(&d)),
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
},
);
Ok(freed)
}
/// **The stale copy goes**: the kept base (and an `.old` one a crash left) and every entry of
/// the log — emptied at the twin's own head, so the numbers do not move. Idempotent: a give-up
/// a crash cut short is finished by calling it again.
fn finish_giving_up(stores: &ServerStores) -> anyhow::Result<()> {
let root = stores.root();
for dir in [
root.join(snapshot::BASE),
root.join(format!("{}.old", snapshot::BASE)),
] {
match std::fs::remove_dir_all(&dir) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(e).with_context(|| format!("giving up {}", dir.display())),
}
}
let log = stores.txlog();
let head = log.head_seq();
if log.first_seq() <= head || log.bytes() > 0 {
log.restart_after(head).context("giving up the stale log")?;
}
Ok(())
}
impl Follower {
/// A follower of `source` into `stores`.
pub fn new(stores: Arc<ServerStores>, source: Box<dyn LogSource>) -> Self {
Self {
stores,
source,
fetch_bytes: DEFAULT_FETCH_BYTES,
reapply: AtomicBool::new(true),
failures: Arc::new(Mutex::new(None)),
said_peak: std::sync::atomic::AtomicU64::new(0),
marked: std::sync::atomic::AtomicU64::new(0),
volume: Arc::new(snapshot::Disk),
}
}
/// Stage snapshots on `volume` instead of the box's own filesystem (the tests' capped
/// volume, twin-resync-space 2026-10-06).
pub fn with_volume(mut self, volume: Arc<dyn snapshot::Volume>) -> Self {
self.volume = volume;
self
}
/// Share the run of failures with the follower before this one (the supervisor's
/// re-dial starts a new follower on the same primary: the run goes on).
pub fn sharing_failures(mut self, failures: Arc<Mutex<Option<Failing>>>) -> Self {
self.failures = failures;
self
}
/// **The twin says its own memory** (twin-streaming, 2026-10-06), at INFO, after a poll that
/// applied something — the first time, and again whenever the peak rose
/// [`SAY_MEMORY_STEP`] above what it last said. A twin has no shell: this line, on its log
/// door, is how an operator (and the UpCloud mockup) sees whether following a fill of large
/// packs costs it the packs' size. The test2 twin that went silent on 1 GB under 256 MiB
/// packs said nothing of the kind.
fn say_memory(&self, head: u64) {
let Some((peak, now)) = own_memory() else {
return;
};
let said = self.said_peak.load(Ordering::Relaxed);
if !memory_is_news(said, peak) {
return;
}
self.said_peak.store(peak, Ordering::Relaxed);
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
peak_mib = peak >> 20,
rss_mib = now >> 20,
head,
"twin memory: peak resident set {} MiB, {} MiB now, following through seq {head}",
peak >> 20,
now >> 20
);
}
/// A poll answered: the end of a run of failures, said once.
fn failures_ended(&self) {
if let Some(f) = self
.failures
.lock()
.unwrap_or_else(|p| p.into_inner())
.take()
{
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, polls = f.polls, seconds = f.since.elapsed().as_secs(), "the primary answers again: following on after {} failed poll(s) over {} s", f.polls, f.since.elapsed().as_secs());
}
}
/// A poll failed: said at WARN when it starts a run (or every [`RESAY_FAILING`] of the
/// same run), at DEBUG otherwise.
fn failed(&self, err: &anyhow::Error) {
// ★ This twin's OWN credential refused for skew: its clock or the primary's is wrong, and
// the front is the judge — the same trigger a refusal on this box's own gate pulls,
// rate-limited, carrying nothing from the refusal (test5's in-place upgrade, 2026-10-09:
// the re-laid twin's clock stayed two hours off while every poll was refused).
if refused_for_skew(err) {
estate_clock::trigger::skew_refused(0);
}
let shape = failure_shape(err);
let now = std::time::Instant::now();
let mut g = self.failures.lock().unwrap_or_else(|p| p.into_inner());
match g.as_mut() {
Some(f) if f.shape == shape => {
f.polls += 1;
if now.duration_since(f.said) >= RESAY_FAILING {
f.said = now;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, polls = f.polls, seconds = now.duration_since(f.since).as_secs(), "poll still failing the same way; this twin keeps asking");
} else {
tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, %err, polls = f.polls, "poll failed (said already)");
}
}
_ => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "poll failed; retrying (said once while it fails this way)");
*g = Some(Failing {
shape,
since: now,
said: now,
polls: 1,
});
}
}
}
/// How many blob bytes one fetch asks for (tests ask for tiny fetches
/// to exercise the restart-mid-range path).
pub fn with_fetch_bytes(mut self, bytes: u64) -> Self {
self.fetch_bytes = bytes.max(1);
self
}
/// The twin's own head.
pub fn applied_seq(&self) -> u64 {
self.stores.txlog().head_seq()
}
/// Apply the log's head entry again when it may not have landed (see `reapply`).
/// Harmless when it had: every apply is idempotent or a replayed CAS that loses the same
/// way. An entry that still does not apply stops the twin HERE, loudly, instead of
/// being skipped.
fn reapply_head(&self) -> anyhow::Result<()> {
if !self.reapply.load(Ordering::SeqCst) {
return Ok(());
}
if self.stores.is_twin() {
return self.retake_from_marker();
}
let log = self.stores.txlog();
let head = log.head_seq();
if head > 0 && head >= log.first_seq() {
// The header and meta only: a twin re-takes a data frame without reading its bytes.
let at = log.locate_blob(head)?;
take_logged(&self.stores, head, at.kind, &at.meta).with_context(|| format!("re-applying seq {head}, the head of this twin's log, before fetching anything newer"))?;
}
self.reapply.store(false, Ordering::SeqCst);
Ok(())
}
/// **A store-only twin re-takes what its log holds past its applied marker** (perf
/// twin-apply-fsync, 2026-10-06) — in order, from exactly the state the marker vouches for
/// ([`take_group`]): the log is durable before anything is taken, so after a crash the log
/// may be a whole group ahead of redb, never behind. A twin from before the marker has none:
/// it keeps the old rule (its head was the only entry that could be missing) and gets one.
fn retake_from_marker(&self) -> anyhow::Result<()> {
let log = self.stores.txlog();
let db = self.stores.control_db();
let head = log.head_seq();
match db.mark(gunnar_core::TWIN_APPLIED)? {
Some(m) if m >= head => {
if m > head {
// The log was cut below what redb holds (it cannot be by this code: redb is
// only ever written after the log's fsync). Say so; follow from the log.
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, marker = m, head, "the applied marker is past this twin's log head; following on from the head");
db.set_mark(gunnar_core::TWIN_APPLIED, head)?;
}
self.marked.store(head, Ordering::SeqCst);
}
Some(m) => {
self.marked.store(m, Ordering::SeqCst);
let from = (m + 1).max(log.first_seq());
let mut entries =
Vec::with_capacity((head + 1).saturating_sub(from).min(1 << 16) as usize);
for seq in from..=head {
let at = log.locate_blob(seq)?;
entries.push((seq, at.kind, at.meta));
}
take_group(&self.stores, &entries, &self.marked).with_context(|| format!("re-taking seq {from}..={head}, past this twin's applied marker {m}, before fetching anything newer"))?;
if self.marked.load(Ordering::SeqCst) < head {
db.set_mark(gunnar_core::TWIN_APPLIED, head)?;
self.marked.store(head, Ordering::SeqCst);
}
}
None => {
if head > 0 && head >= log.first_seq() {
let at = log.locate_blob(head)?;
take_logged(&self.stores, head, at.kind, &at.meta).with_context(|| format!("re-applying seq {head}, the head of this twin's log, before fetching anything newer"))?;
}
db.set_mark(gunnar_core::TWIN_APPLIED, head)?;
self.marked.store(head, Ordering::SeqCst);
}
}
self.reapply.store(false, Ordering::SeqCst);
Ok(())
}
/// One poll: head, the gap rule, fetch-append-apply until level.
///
/// **A poll that reached the primary and then failed says so in the status**
/// (2026-09-21, lane T4). Only the `head()` call used to set a status on
/// failure. Everything after it — `append_at` on a full volume, an `apply`
/// that will not go in, a fetch that dies mid-range — propagated with `?`,
/// and the STATUS WAS LEFT EXACTLY AS THE LAST GOOD POLL SET IT: `following`,
/// often `caught_up`, with an empty `why`. `run` logged one `warn` and
/// polled again. So the shape a twin's life actually ends in — its data
/// volume filling — read as a healthy, current copy on both `twin status`
/// and the primary's heartbeat, for as long as the box stayed up. A twin's
/// whole job is to be trustworthy at the moment nobody has time to check.
pub fn step(&self) -> anyhow::Result<Progress> {
// `step_inner` sets this when it has already written a status of its
// own for the failure it is returning (`unreachable` from `head()`,
// `snapshot-needed` from a half-installed snapshot). Every OTHER error
// path is one this twin used to keep quiet about.
let mut said_why = false;
// What THIS poll's answer said, when it got one: the primary's head and
// when. A poll that reached the primary and then failed must record the
// answer it got — it used to record the PREVIOUS poll's, so a twin that
// stalled behind a primary still taking writes reported an older head
// and an older time than the primary had actually just given it.
let mut heard: Option<(u64, i64)> = None;
let r = self.step_inner(&mut said_why, &mut heard);
if let (Err(err), false) = (&r, said_why) {
let last = status(self.stores.root());
let (primary_head_seq, last_answer_unix_ms) = heard.unwrap_or((
last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
));
// ★ cov-13: the primary refused to let this twin follow into a volume smaller than
// main's grown set. Not a stall: a debt, said in the primary's own words.
let said = format!("{err:#}");
// ★ twin-resync-space: a twin that gave its stale copy up says so on every failure
// after it, until the resync lands.
let no_copy = snapshot::dropped(self.stores.root())
.map(|d| format!(" — {}", d.sentence()))
.unwrap_or_default();
if let Some(at) = said.find(gunnar_grpc::TWIN_TOO_SMALL) {
set_status(
self.stores.root(),
TwinStatus {
applied_seq: self.applied_seq(),
primary_head_seq,
state: State::Recovering,
why: format!("{}{no_copy}", &said[at..]),
last_answer_unix_ms,
last_level_unix_ms: 0,
},
);
return r;
}
let holds = if no_copy.is_empty() {
"what it holds stops here".to_owned()
} else {
format!("it holds NO copy{no_copy}")
};
set_status(self.stores.root(), TwinStatus {
applied_seq: self.applied_seq(),
primary_head_seq,
state: State::Stalled,
why: format!("this twin could not take what its primary sent — it is NOT a current copy, and {holds}: {err:#}"),
last_answer_unix_ms,
last_level_unix_ms: 0,
});
}
r
}
fn step_inner(
&self,
said_why: &mut bool,
heard: &mut Option<(u64, i64)>,
) -> anyhow::Result<Progress> {
// A crash mid-install: the stores are half a snapshot; only a whole one fixes that.
let installing = snapshot::installing(self.stores.root());
if !installing {
self.reapply_head()?;
}
let (first, primary_head, their_id, their_generation) = match self
.source
.head_identity_generation()
{
Ok(h) => h,
Err(err) => {
// **An unanswered poll keeps what the last answer said** — the
// primary's head AND the time (2026-09-21, lane T4). This used
// to write `primary_head_seq: 0`, so the moment a primary died
// the twin that had followed it to 221 reported "a primary head
// it last saw at 0" (MEASURED under KVM by T3) — and a restore
// guard reading that could not tell a complete copy from an
// empty one. The primary saying nothing now is not the primary
// having had nothing then.
let last = status(self.stores.root());
set_status(
self.stores.root(),
TwinStatus {
applied_seq: self.applied_seq(),
primary_head_seq: last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
state: State::Unreachable,
why: err.to_string(),
last_answer_unix_ms: last
.as_ref()
.map(|s| s.last_answer_unix_ms)
.unwrap_or(0),
last_level_unix_ms: 0,
},
);
*said_why = true;
return Err(err);
}
};
let answered = now_ms();
let own = self.applied_seq();
*heard = Some((primary_head, answered));
// ── the identity check (2026-09-21, lane T4) ────────────────────────
//
// **Before any number is believed, the box that gave it is asked which
// log it is.** The sequence rule below catches a primary that is BEHIND
// this twin. It cannot catch one that is AHEAD of it for the wrong
// reason: a re-imaged main at the same address, left running for an
// afternoon, passes the twin's head, and from then on every number
// agrees — `first <= wanted`, `wanted <= primary_head` — and the twin
// splices one history onto another and applies a stranger's frames over
// the only surviving copy of its own. No arithmetic can see that. Only
// a name can, so [`LOG_ID`] gives the log one.
//
// Three cases, and the middle one is the join:
// * both named and different — DIVERGED, and it stays stopped,
// * this root unnamed — pin what the source says (the first poll
// after `become_twin`, which dropped this box's own id),
// * the source unnamed — a primary older than this change. Nothing is
// claimed and nothing is refused; the numbers are all there is, as
// they were before. Saying so beats inventing a verdict.
if let Some(theirs) = their_id.as_deref().map(str::trim).filter(|s| !s.is_empty()) {
match log_id(self.stores.root()) {
Some(mine) if mine != theirs => {
let why = format!(
"the box at the other end is log {theirs}; this twin holds log {mine}. Two different histories, \
whatever their sequence numbers say — this is a re-imaged or reinstalled primary at a familiar \
address, or an address that now points at another set. NOTHING is applied on top of it, and \
retrying cannot fix it: either point this twin at the box that holds {mine}, or — if {mine} is \
gone — restore a primary FROM this twin, which keeps the id, and it will follow again."
);
set_status(
self.stores.root(),
TwinStatus {
applied_seq: own,
primary_head_seq: primary_head,
state: State::Diverged,
why,
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
},
);
return Ok(Progress::Diverged { own, primary_head });
}
Some(_) => {}
None => {
pin_log_id(self.stores.root(), theirs)
.context("pinning the primary's log identity on this twin")?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, log_id = %theirs, "this twin now knows which log it follows");
}
}
}
// ── the generation fence (stinky row 18, 2026-10-10) ───────────────
//
// **A restored main keeps the id and the key, so the OLD main is the
// same name too.** If it comes back after its replacement was restored
// from this twin, and it took writes the twin never saw (or takes them
// now), its head passes this twin's and every check above and below
// agrees: the twin would splice the dead main's lost writes onto the
// restored history. A restore stamps a newer [`LOG_GENERATION`]; this
// twin pins the highest it has followed and refuses a lower one. A
// fence that cannot be read refuses the poll rather than reading as 0.
let pinned = pinned_generation(self.stores.root())?;
if their_generation < pinned {
let why = format!(
"the box at the other end is generation {their_generation} of this log; this twin already follows generation \
{pinned}, a later restore of it. This is the OLD main come back after its replacement was restored: the same \
log id and the same key, so nothing else can tell them apart. NOTHING is applied from it — its writes since \
the restore belong to no history; this poll handed it the proof, so it fences itself ({LOG_FENCED}). Point \
this twin back at the restored main, and retire the old box."
);
set_status(
self.stores.root(),
TwinStatus {
applied_seq: own,
primary_head_seq: primary_head,
state: State::Diverged,
why,
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
},
);
return Ok(Progress::Diverged { own, primary_head });
}
// ── the lineage check (2026-09-21, lane T4) ─────────────────────────
// **A log's head only ever goes forward.** Retention moves `first`; a
// restore continues the same numbers; nothing in gunnar rewinds a head.
// So a primary answering a head BELOW what this twin has already applied
// is not the log this twin has been following — a re-imaged box, a
// restore from somewhere else, or an address that now points at a
// different set.
//
// MEASURED on master 2026-09-21, before this: that case fell straight
// through. `wanted <= primary_head` is false, so the snapshot branch and
// the snapshot-needed refusal are both skipped, the fetch loop's
// condition is false, and the poll ended at the bottom of `step` setting
// `State::Following` with an EMPTY `why`. On the primary's side the lag
// is `head_seq.saturating_sub(applied)` = 0, so the heartbeat read `ok`.
// BOTH ENDS reported a healthy pairing, for ever, while not one entry
// would ever be applied again and the twin's contents belonged to
// another box. A restore leaning on that twin is the worst shape this
// repo knows — and re-imaging a main is a thing this estate does.
if !installing && primary_head < own {
let why = format!(
"the primary answers head {primary_head}, BELOW the {own} this twin has already applied. A log's head never goes \
backwards, so this is NOT the log this twin was following (a re-imaged or restored primary, or an address that now \
points at a different set). Nothing is applied on top of it; an operator decides whether to re-join from scratch."
);
set_status(
self.stores.root(),
TwinStatus {
applied_seq: own,
primary_head_seq: primary_head,
state: State::Diverged,
why,
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
},
);
return Ok(Progress::Diverged { own, primary_head });
}
// A newer generation that passed every check above is this twin's
// primary from now on; the pin only ever moves up.
if their_generation > pinned {
pin_generation(self.stores.root(), their_generation)
.context("pinning the primary's log generation on this twin")?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, generation = their_generation, "this twin now follows generation {their_generation} of its log and refuses any older one");
}
let wanted = own + 1;
if installing || (wanted <= primary_head && first > wanted) {
let root = self.stores.root();
// ★ twin-resync-space (2026-10-06): a twin makes ROOM for the resync instead of
// filling its volume part-way (see `snapshot`'s "Room on a twin's volume").
let twin = self.stores.is_twin();
let mut given_up = false;
if twin {
// A resync this volume could not hold even alone is not asked for again until
// the volume is larger: every snapshot holds main's writes for its whole stream.
if let Some(was) = snapshot::too_small_total(root) {
The file goes on; the console shows all of it.