crates/gunnar-server/src/twin/snapshot.rs
//! **The snapshot** — what a twin installs when the primary no longer retains the range it
//! needs (`.nornir/twin-transaction-log.md` §"Snapshot").
//!
//! A snapshot is the MUST-NOT-LOSE classes (M1–M9 of the note's inventory) at one sequence
//! `at`, spelled as the same entries the log carries — `Kek`, `ControlBlob`, `ControlDoc`,
//! `Secret`, `File`, and per repository `RepoStore{created}`, one `Pack`, one `Refs`, `LfsPut` —
//! each with `seq` 0, and a `SnapshotAt{at}` terminator. So the one `apply` installs it, and the
//! one wire piece (`LogChunk`) carries it.
//!
//! **Consistent at `at`.** The primary takes its CUT under [`gunnar_core::txlog::Log::freeze`]:
//! no framed mutation starts, every one in flight reaches its store, and only then is the head
//! read and the state fixed (a read view of the control database, the framed files' handles and
//! lengths, every repository's refs). The log is written BEFORE the store, so without the wait
//! an entry at the head could still be on its way into the store and the twin, resuming at
//! `at + 1`, would never see it. Writes pause for the cut, not for the stream (R5.1-A).
//!
//! **Reused, not rewritten.** The objects of a repository are the engine's own export: the
//! reachability `select` (or `reachable`) over every ref tip, then `emit_pack` — the znippy
//! arm answers that as a byte-range copy out of its archive, the same path a clone takes. The
//! control database is [`gunnar_core::control::ControlDb::export_each`], row by row.
//!
//! **Streamed, never held** (2026-09-15, Lane Q). A 44 GiB data set does not fit in a 1 GiB
//! twin, so neither end holds the snapshot:
//!
//! * the producer ([`produce`]) writes every item through ONE preallocated [`PIECE_BYTES`]
//! buffer and hands each full slab to a sink as a [`Piece`] — `emit_pack` writes straight
//! into it, files and LFS objects are copied through it, control rows go one by one. Over the
//! wire the sink is a bounded channel into the gRPC stream (`Mirror.SnapshotFetch`), so a slow
//! twin holds the producer back instead of the producer piling pieces up. An item's length is
//! not known before a pack is emitted: every piece but an item's last says
//! [`BLOB_LEN_AT_LAST`], the last carries the item's total;
//! * the twin ([`Stager`]) appends each piece's bytes to ONE staging file on its data disk as
//! it arrives and keeps only the item table (kind, meta, offset, length). Nothing is touched
//! until the terminator has arrived and the file is synced;
//! * the install ([`install`]) reads each item back from the staging file — a small one into a
//! reused buffer, a large one as a read-only mapping (page cache the kernel can reclaim, the
//! push spool's rule in `storage::PackBytes`) — and deletes the staging file at the end.
//!
//! **Installed whole.** The twin wipes its repository stores and every framed file the snapshot
//! does not carry, force-sets every document to the snapshot's pointer, applies the rest with
//! [`super::apply_raw`], and restarts its log after `at`. A marker makes a crash mid-install ask
//! for the snapshot again rather than follow from a half-installed state; a crash while the
//! snapshot is still arriving leaves the stores untouched and the staging file is started over.
//!
//! **A twin keeps it** (ruling 2026-09-22, "the twin only stores"): a twin applies only the
//! control-plane items and keeps the staging file as [`BASE`] — its copy of everything at or
//! below `at` — and [`serve`] streams that base when the twin is asked for a snapshot. The box
//! recharged from it installs the base whole, and that is where the indexes are built.
//!
//! **Room on a twin's volume** (twin-resync-space, 2026-10-06). The staging file needs the
//! snapshot's size free on the twin's data disk, and it is staged BESIDE the copy it replaces —
//! the kept base and the log after it — so a resync can need about twice main's data. A twin is
//! grown to main's data set, not to twice it (cov-13: "main grew alone while the twin was
//! silent", then the twin is grown to main's set and resynced), and such a twin filled up
//! part-way, failed, and failed the same way on every later poll. So a twin's [`Stager`] makes
//! room ([`Stager::making_room`]): it stages beside the old copy while that fits (the old copy is
//! kept until the new one is whole — the common case, a twin with room), and the moment the
//! volume would run below [`staging_floor`] (or says `ENOSPC`) it GIVES THE STALE COPY UP: the
//! [`DROPPED`] marker first (durable), then its base and its log. It is a replica that was
//! already behind a hole main had trimmed; main holds the truth. The window is explicit — a
//! WARN `ALERT` line, the `twin.resync_without_copy` health alarm, the state word `resyncing`
//! with "holds NO copy" — and a twin in it serves nothing ([`not_whole`]: no snapshot, no log,
//! so no recharge from it). A crash anywhere in it restarts the resync on the next poll (the
//! marker makes [`installing`] true). A volume too small for main's data even alone says
//! [`TooSmall`] (`[twin-too-small]`), waits in `recovering`, and asks main for no further
//! snapshot until its volume is larger ([`TOO_SMALL`]) — every `SnapshotFetch` costs main a
//! frozen cut (R5.1-A) and a whole stream.
//!
//! Why not in place, store by store: a store-only twin's copy is ONE base file plus the log after
//! it, not stores. Any mix of the old and the new is consistent at no sequence (the control plane
//! at one point, the data at another), so an in-place swap has exactly the same no-copy window as
//! giving the old one up — with a layout change and a resumable snapshot protocol on top. And
//! growing every twin to twice main's set (a preflight) costs that volume for ever, to keep a
//! copy that is already behind a hole.
//!
//! Limits, named: the freeze is held for the CUT, not the stream (R5.1-A, 2026-10-07: it used to
//! pause the primary's writes for as long as the twin took to receive everything) — the cut reads
//! every repository's refs, so it grows with the number of repositories; a gc and a repository
//! store's removal wait for a stream in progress (at most `stores::SNAPSHOT_PIN_WAIT`); the control
//! database's read view keeps its file open while it streams; a repository's object-id list from
//! `select` is held in memory (ids, not objects); the staging file needs the snapshot's size free
//! on the twin's data disk (beside what it replaces while that fits; see "Room on a twin's volume").
use std::collections::{BTreeMap, BTreeSet};
use std::io::Write;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use anyhow::{anyhow, Context};
use gunnar_core::control::{BlobId, ExportRow};
use gunnar_core::txlog::{self, Crc32c, Kind, Meta, RefEdit};
use gunnar_store::git_storage_trait::Caps;
use crate::stores::ServerStores;
/// `<data-root>/txlog-snapshot-installing`: an install is in progress. Per box, never framed.
pub const INSTALLING: &str = "txlog-snapshot-installing";
/// `<data-root>/txlog-snapshot-staging/`: a snapshot arriving (or arrived and installing). Per
/// box, never framed; removed when the install ends or the next attempt starts.
pub const STAGING: &str = "txlog-snapshot-staging";
/// `<data-root>/txlog-snapshot-dropped`: **this twin gave its stale copy up** (its kept base and
/// its log) to make room for a full resync and holds NO copy until the resync installs
/// (twin-resync-space, 2026-10-06). Written BEFORE the first byte of the old copy goes; removed
/// by [`install`] once the new copy is whole. Per box, never framed.
pub const DROPPED: &str = "txlog-snapshot-dropped";
/// `<data-root>/txlog-snapshot-too-small`: the last resync ran out of room even with the stale
/// copy given up; it holds the volume's size then. The twin asks main for no further snapshot
/// until its volume is larger. Removed by [`install`]. Per box, never framed.
pub const TOO_SMALL: &str = "txlog-snapshot-too-small";
/// One piece's bytes at most: the producer's one buffer, and the wire's `LogChunk` slab (under
/// every client's default 4 MiB message ceiling with room for the meta).
pub const PIECE_BYTES: usize = 1 << 20;
/// `blob_len` on every piece of an item but its last: the length is known when the item ends.
pub const BLOB_LEN_AT_LAST: u64 = u64::MAX;
/// Items at or under this size are read back into the install's reused buffer; larger ones are
/// mapped.
const INLINE_BYTES: u64 = PIECE_BYTES as u64;
/// **One piece of a snapshot item**, borrowed: `meta` on the item's first piece only, `blob` a
/// slab of at most [`PIECE_BYTES`], `blob_len` the item's total on its `last` piece and
/// [`BLOB_LEN_AT_LAST`] before it.
#[derive(Debug, Clone, Copy)]
pub struct Piece<'a> {
pub kind: u16,
pub meta: &'a [u8],
pub blob_len: u64,
pub blob: &'a [u8],
pub last: bool,
}
/// Where the producer's pieces go.
pub type Sink<'s> = dyn FnMut(Piece<'_>) -> anyhow::Result<()> + 's;
fn not_found(e: &std::io::Error) -> bool {
e.kind() == std::io::ErrorKind::NotFound
}
/// **The producer's one buffer**, and the item being written through it. A full slab is handed
/// to the sink only when more bytes arrive, so an item's last piece is never an empty extra one.
struct Pieces<'s, 'k> {
sink: &'k mut Sink<'s>,
buf: Vec<u8>,
kind: u16,
meta: Vec<u8>,
first: bool,
total: u64,
/// The sink's error, kept whole while `Write` can only say "other".
failed: Option<anyhow::Error>,
/// Pieces handed over, for the log line.
pieces: u64,
}
impl<'s, 'k> Pieces<'s, 'k> {
fn new(sink: &'k mut Sink<'s>) -> Self {
Self {
sink,
buf: Vec::with_capacity(PIECE_BYTES),
kind: 0,
meta: Vec::new(),
first: true,
total: 0,
failed: None,
pieces: 0,
}
}
fn begin(&mut self, kind: Kind, meta: &Meta) {
debug_assert!(self.buf.is_empty() && self.first);
self.kind = kind as u16;
self.meta = meta.to_bytes();
self.total = 0;
}
fn hand_over(&mut self, last: bool) -> anyhow::Result<()> {
let piece = Piece {
kind: self.kind,
meta: if self.first { &self.meta } else { &[] },
blob_len: if last { self.total } else { BLOB_LEN_AT_LAST },
blob: &self.buf,
last,
};
let r = (self.sink)(piece);
self.buf.clear();
self.first = false;
self.pieces += 1;
r
}
/// The item ends: its last piece, with its total.
fn end(&mut self) -> anyhow::Result<()> {
if let Some(err) = self.failed.take() {
return Err(err);
}
self.hand_over(true)?;
self.first = true;
Ok(())
}
/// A whole small item.
fn item(&mut self, kind: Kind, meta: &Meta, blob: &[u8]) -> anyhow::Result<()> {
self.begin(kind, meta);
self.write_all(blob)?;
self.end()
}
/// A file's bytes as one item, copied through the buffer — at most `limit` of them (the
/// length the file had at a snapshot's cut; `u64::MAX` = to its end).
fn file(
&mut self,
kind: Kind,
meta: &Meta,
file: &std::fs::File,
limit: u64,
) -> anyhow::Result<()> {
self.file_at(kind, meta, file, 0, limit)
}
/// [`Self::file`] from byte `from` of the file: `limit` bytes of it (a frame's blob inside a
/// log segment).
fn file_at(
&mut self,
kind: Kind,
meta: &Meta,
file: &std::fs::File,
from: u64,
limit: u64,
) -> anyhow::Result<()> {
self.begin(kind, meta);
let mut at = 0u64;
loop {
if self.buf.len() == PIECE_BYTES {
self.hand_over(false)?;
}
let start = self.buf.len();
let room = ((PIECE_BYTES - start) as u64).min(limit - at) as usize;
if room == 0 {
break;
}
// SAFETY-free spare capacity use: grow the length over the spare room, read into it,
// cut back to what arrived. The buffer never reallocates (capacity PIECE_BYTES).
self.buf.resize(start + room, 0);
let n = match file.read_at(&mut self.buf[start..], from + at) {
Ok(n) => n,
Err(e) => {
self.buf.truncate(start);
return Err(e.into());
}
};
self.buf.truncate(start + n);
if n == 0 {
break;
}
at += n as u64;
self.total += n as u64;
}
self.end()
}
}
impl Write for Pieces<'_, '_> {
fn write(&mut self, mut data: &[u8]) -> std::io::Result<usize> {
let len = data.len();
while !data.is_empty() {
if self.buf.len() == PIECE_BYTES {
if let Err(err) = self.hand_over(false) {
let say = format!("{err:#}");
self.failed = Some(err);
return Err(std::io::Error::other(say));
}
}
let n = (PIECE_BYTES - self.buf.len()).min(data.len());
self.buf.extend_from_slice(&data[..n]);
self.total += n as u64;
data = &data[n..];
}
Ok(len)
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
/// **What a snapshot carries, fixed at one instant** (R5.1-A, 2026-10-07): the sequence `at`, the
/// KEK, a read view of the control database, every framed file as an open handle and its length,
/// and every repository's refs, HEAD and tips. Taken under the log's freeze — that is the whole
/// freeze now — and streamed from after it.
struct Cut {
at: u64,
kek: Vec<u8>,
control: gunnar_core::ControlView,
files: Vec<(&'static str, std::fs::File, u64)>,
repos: Vec<RepoCut>,
}
/// One repository at the cut.
struct RepoCut {
repo: gunnar_core::Repo,
key: String,
hash: String,
edits: Vec<RefEdit>,
tips: Vec<Vec<u8>>,
/// **Its pushes caught between their two frames** ([`pending_packs`]): each `Pack` frame at or
/// below the cut that no `Refs` frame of this repository follows yet, as an open handle on its
/// segment, the blob's offset and length. Streamed beside the reachable export.
pending: Vec<(std::fs::File, u64, u64)>,
}
/// Pure: **the `Pack` frames a cut at `at` must carry beside the reachable export** — per
/// repository, the ones after its last `Refs` frame. A push is two frames: its `Pack` (framed in
/// the ingest, under its own mutation guard) and, after the policy gates, its `Refs`. A cut that
/// falls between them finds the pushed objects in the store but reachable from no tip, and the
/// reachability export leaves them out; the `Refs` frame after the cut then names objects the
/// snapshot never had (R6 on the estate-box gate, 2026-10-09: the recharged main re-applied
/// seq 4104 for twenty minutes). `frames` is `(seq, kind, meta)` in sequence order.
///
/// **Several pushes into one repository** (R6 residual, 5.3.3): a `Refs` frame does not say which
/// pack it closes — A's pack, B's pack, A's refs | cut | B's refs — so it closes ONE push, not the
/// repository's every open pack. Per repository the open pushes are counted (a pack opens one, a
/// refs frame that sets a ref closes one; a refs frame that only deletes closes none), and every
/// pack since the count was last zero is carried: a carried pack whose objects are already
/// reachable costs bytes, a missing one costs the recharge.
pub fn pending_packs(frames: &[(u64, u16, Meta)], at: u64) -> BTreeMap<String, Vec<u64>> {
let mut open: BTreeMap<String, (u64, Vec<u64>)> = BTreeMap::new();
for (seq, kind, meta) in frames {
if *seq > at {
break;
}
match meta {
Meta::Pack { repo, .. } if *kind == Kind::Pack as u16 => {
let e = open.entry(repo.clone()).or_default();
e.0 += 1;
e.1.push(*seq);
}
Meta::Refs { repo, edits, .. }
if *kind == Kind::Refs as u16 && edits.iter().any(|e| e.new.is_some()) =>
{
if let Some(e) = open.get_mut(repo) {
e.0 = e.0.saturating_sub(1);
if e.0 == 0 {
open.remove(repo);
}
}
}
_ => {}
}
}
open.into_iter()
.map(|(repo, (_, seqs))| (repo, seqs))
.collect()
}
impl Cut {
/// Under the freeze: nothing framed is in flight, so the stores ARE the log's head.
fn take(stores: &ServerStores, at: u64) -> anyhow::Result<Self> {
// M6 — the key the secrets are sealed under: the file when there is one (a twin's is the
// primary's, applied), else the key this process holds (handed in by the environment).
let kek = match std::fs::read(gunnar_core::seal::kek_path(stores.control_dir())) {
Ok(bytes) => bytes,
Err(e) if not_found(&e) => stores.kek_bytes(),
Err(e) => return Err(e).context("reading the KEK for a snapshot"),
};
// M4, M5 — a read view: redb keeps this instant for as long as the stream needs it.
let control = stores
.control_db()
.view()
.context("a view of the control database for a snapshot")?;
// M7, M8 — each framed file as an OPEN handle and its length now: an append-only log is
// read up to that length, and a document replaced later (a rename) is still this inode.
let mut files = Vec::new();
for rel in txlog::FRAMED_FILES {
match std::fs::File::open(stores.root().join(rel)) {
Ok(f) => {
let len = f
.metadata()
.with_context(|| format!("measuring {rel} for a snapshot"))?
.len();
files.push((*rel, f, len));
}
Err(e) if not_found(&e) => {}
Err(e) => return Err(e).with_context(|| format!("reading {rel} for a snapshot")),
}
}
// M9, M2 — every repository and its refs; the objects are content-addressed and stay
// (the snapshot pin holds back a gc and a purge), so their bytes stream after the freeze.
// The pushes caught between their `Pack` and their `Refs` at this instant ([`pending_packs`]).
let mut frames: Vec<(u64, u16, Meta)> = Vec::new();
stores
.txlog()
.scan_meta(|seq, kind, meta| {
if kind == Kind::Pack as u16 || kind == Kind::Refs as u16 {
if let Ok(m) = Meta::from_bytes(meta) {
frames.push((seq, kind, m));
}
}
})
.context("scanning the log for pushes caught between their frames")?;
let mut pending_seqs = pending_packs(&frames, at);
let mut repos = Vec::new();
for repo in stores.catalog().list(None)? {
let key = repo.key();
let hash = repo.hash_kind.to_string();
let store = stores
.repo_store(&repo)
.with_context(|| format!("opening {key} for a snapshot"))?;
let git = store.unlogged();
let mut edits = Vec::new();
let mut tips: Vec<Vec<u8>> = Vec::new();
for r in git.refs()? {
if r.name == "HEAD" {
continue;
}
if let Some(oid) = &r.oid {
tips.push(oid.clone());
}
edits.push(RefEdit {
name: r.name,
old: None,
new: r.oid.as_deref().map(hex::encode),
peeled: r.peeled.as_deref().map(hex::encode),
symref: r.symref_target,
});
}
if let Some(target) = git.head()?.and_then(|h| h.symref_target) {
edits.push(RefEdit {
name: "HEAD".into(),
old: None,
new: None,
peeled: None,
symref: Some(target),
});
}
let mut pending = Vec::new();
for seq in pending_seqs.remove(&key).unwrap_or_default() {
let at = stores
.txlog()
.locate_blob(seq)
.with_context(|| format!("locating the pending pack of {key} at seq {seq}"))?;
let f = std::fs::File::open(&at.path).with_context(|| {
format!("opening the segment of {key}'s pending pack at seq {seq}")
})?;
pending.push((f, at.blob_offset, at.blob_len));
}
repos.push(RepoCut {
repo,
key,
hash,
edits,
tips,
pending,
});
}
Ok(Cut {
at,
kek,
control,
files,
repos,
})
}
}
/// **Produce a snapshot of `stores` into `sink`** — a primary for a twin behind a retention
/// hole, or a twin for a restore from it. Answers `at`.
///
/// **The log is frozen for the CUT only** (R5.1-A, 2026-10-07). It used to be frozen for the
/// whole stream — every framed write on the primary (every push, control write and audited RPC)
/// waited until the twin had the last piece of every repository's pack. Now the freeze holds while
/// [`Cut::take`] fixes the state at `at` — the head, a read view of the control database, the
/// framed files' handles and lengths, every repository's refs and tips — and the bytes stream from
/// the cut afterwards, writes going on beside them. A snapshot pin
/// ([`ServerStores::snapshot_pin`], taken BEFORE the freeze) holds back a gc and the removal of a
/// repository store until the stream is done, so what the cut names is still there to read. LFS
/// objects are listed after the freeze: an object stored later is also a later `LfsPut` (applied
/// again, idempotent by oid) and one removed later is a later `LfsDelete` — an object gone by the
/// time it is reached is skipped, and the replay ends in the same state.
pub fn produce(stores: &ServerStores, sink: &mut Sink<'_>) -> anyhow::Result<u64> {
let log = stores.txlog();
let _pin = stores.snapshot_pin();
let t0 = std::time::Instant::now();
let cut = {
let _frozen = log.freeze();
Cut::take(stores, log.head_seq())?
};
let at = cut.at;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, at, repos = cut.repos.len(), frozen_ms = t0.elapsed().as_millis() as u64, "a snapshot's cut is taken; writes go on while it streams");
let mut out = Pieces::new(sink);
out.item(Kind::Kek, &Meta::Kek {}, &cut.kek)?;
cut.control
.export_each(&mut |row| match row {
ExportRow::Blob(id, bytes) => out.item(
Kind::ControlBlob,
&Meta::ControlBlob {
id: hex::encode(id.as_bytes()),
},
bytes,
),
ExportRow::Doc(name, id) => out.item(
Kind::ControlDoc,
&Meta::ControlDoc {
name: name.to_string(),
expected: None,
new: hex::encode(id.as_bytes()),
},
&[],
),
ExportRow::Secret(key, value) => out.item(
Kind::Secret,
&Meta::Secret {
key: key.to_string(),
deleted: value.is_none(),
},
value.unwrap_or_default(),
),
})
.context("exporting the control database")??;
// The last-seen table at the cut, as ONE item (2026-10-21): a twin that joins later
// has every stamp, and keeps it when promoted.
let activity = cut
.control
.principal_activity()
.context("reading the last-seen table for a snapshot")?;
if !activity.is_empty() {
out.item(
Kind::Activity,
&Meta::Activity {},
&gunnar_core::activity::encode_rows(&activity)?,
)?;
}
for (rel, f, len) in &cut.files {
out.file(
Kind::File,
&Meta::File {
path: rel.to_string(),
deleted: false,
},
f,
*len,
)
.with_context(|| format!("streaming {rel} for a snapshot"))?;
}
// M1, M3 — per repository: its store, its objects (the engine's export over the cut's tips,
// written straight into the pieces), its refs and HEAD as they were at the cut, its LFS.
for r in &cut.repos {
let key = &r.key;
out.item(
Kind::RepoStore,
&Meta::RepoStore {
repo: key.clone(),
op: "created".into(),
hash: r.hash.clone(),
},
&[],
)?;
let store = stores
.repo_store(&r.repo)
.with_context(|| format!("opening {key} for a snapshot"))?;
let git = store.unlogged();
if !r.tips.is_empty() {
let want: Vec<&[u8]> = r.tips.iter().map(Vec::as_slice).collect();
let objects = match git.select(&want, &[])? {
Some(set) => set.objects,
None => git.reachable(&want, &[])?,
};
let objs: Vec<&[u8]> = objects.iter().map(Vec::as_slice).collect();
out.begin(
Kind::Pack,
&Meta::Pack {
repo: key.clone(),
hash: r.hash.clone(),
},
);
let emitted = git.emit_pack(&objs, &[], &Caps::modern(), &mut out);
if let Some(err) = out.failed.take() {
return Err(err).with_context(|| format!("streaming the objects of {key}"));
}
emitted.with_context(|| format!("exporting the objects of {key}"))?;
out.end()?;
}
// ★ Each push caught between its two frames: its pack, as it was framed, so the `Refs` frame
// after the cut finds every object it names (R6, 2026-10-09).
for (f, offset, len) in &r.pending {
out.file_at(
Kind::Pack,
&Meta::Pack {
repo: key.clone(),
hash: r.hash.clone(),
},
f,
*offset,
*len,
)
.with_context(|| format!("streaming a pending pack of {key}"))?;
}
if !r.edits.is_empty() {
out.item(
Kind::Refs,
&Meta::Refs {
repo: key.clone(),
atomic: false,
edits: r.edits.clone(),
},
&[],
)?;
}
let lfs_objects = store.git_dir().join("lfs").join("objects");
for path in files_under(&lfs_objects)? {
let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
continue;
};
if gunnar_lfs::Oid::parse(name).is_err() {
continue;
}
let f = match std::fs::File::open(&path) {
Ok(f) => f,
// Removed after the cut: its `LfsDelete` is a later frame (see above).
Err(e) if not_found(&e) => continue,
Err(e) => return Err(e).with_context(|| format!("opening LFS {name} of {key}")),
};
let size = f.metadata()?.len();
out.file(
Kind::LfsPut,
&Meta::Lfs {
repo: key.clone(),
oid: name.to_string(),
size,
},
&f,
size,
)
.with_context(|| format!("streaming LFS {name} of {key}"))?;
}
}
out.item(Kind::SnapshotAt, &Meta::SnapshotAt { at }, &[])?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, at, pieces = out.pieces, seconds = t0.elapsed().as_secs(), "produced a snapshot");
Ok(at)
}
fn files_under(dir: &Path) -> anyhow::Result<Vec<PathBuf>> {
let mut out = Vec::new();
let mut stack = vec![dir.to_path_buf()];
while let Some(d) = stack.pop() {
let rd = match std::fs::read_dir(&d) {
Ok(rd) => rd,
Err(e) if not_found(&e) => continue,
Err(e) => return Err(e).with_context(|| format!("listing {}", d.display())),
};
for entry in rd {
let entry = entry?;
let ft = entry.file_type()?;
if ft.is_dir() {
stack.push(entry.path());
} else if ft.is_file() {
out.push(entry.path());
}
}
}
out.sort();
Ok(out)
}
// ── the volume a snapshot is staged on ──────────────────────────────────────
/// **The volume a snapshot is staged on** (twin-resync-space, 2026-10-06): what the stager asks
/// of it — its size and what is still free — and how a piece is written to it. [`Disk`] is the
/// box's own filesystem (one `statvfs`, one `pwrite`); a test caps a directory in-process, so a
/// twin's volume "the size of main's data set" is a thing the suite can run anywhere.
pub trait Volume: Send + Sync {
/// `(total, free)` bytes of the volume holding `path` — free as an unprivileged write sees
/// it (`f_bavail`: the root reserve is not the box's to stage into); `None` when it cannot be
/// measured, which is never read as "full".
fn space(&self, path: &Path) -> Option<(u64, u64)>;
/// `bytes` at `off` of `file`, a file on this volume.
fn write_at(&self, file: &std::fs::File, bytes: &[u8], off: u64) -> std::io::Result<()> {
file.write_all_at(bytes, off)
}
}
/// **The box's own filesystem.**
#[derive(Debug, Clone, Copy, Default)]
pub struct Disk;
impl Volume for Disk {
fn space(&self, path: &Path) -> Option<(u64, u64)> {
let v = rustix::fs::statvfs(path).ok()?;
let frsize = if v.f_frsize == 0 {
v.f_bsize
} else {
v.f_frsize
};
let total = v.f_blocks.saturating_mul(frsize);
(total > 0).then(|| (total, v.f_bavail.saturating_mul(frsize)))
}
}
/// **What a twin keeps free for itself while a snapshot is staged** (twin-resync-space): 1 % of
/// its volume, at least 1 MiB and at most 256 MiB — room for the install's control-plane writes,
/// its own log lines and its status. A twin is grown to main's data set, and main keeps its own
/// 2 GiB floor and its log budget free inside that set, so main's snapshot plus this fits it.
pub fn staging_floor(total: u64) -> u64 {
(total / 100).clamp(1 << 20, 256 << 20)
}
fn is_no_space(e: &std::io::Error) -> bool {
matches!(e.raw_os_error(), Some(c) if c == rustix::io::Errno::NOSPC.raw_os_error() || c == rustix::io::Errno::DQUOT.raw_os_error())
}
/// What a twin gives up for room: its stale copy. Answers the bytes freed.
pub type GiveUp = dyn FnMut(&Short) -> anyhow::Result<u64> + Send;
struct Room {
give_up: Box<GiveUp>,
given_up: bool,
}
/// **The volume ran short mid-resync**: bytes staged so far, free of total, the floor kept.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Short {
pub staged: u64,
pub free: u64,
pub total: u64,
pub floor: u64,
/// The volume refused a write (`ENOSPC`) rather than the measure saying so first.
pub enospc: bool,
}
/// **A resync this twin's volume cannot hold, even with its stale copy given up** — said with
/// [`gunnar_grpc::TWIN_TOO_SMALL`], so the follower waits in `recovering` (and asks main for no
/// further snapshot until the volume is larger) instead of failing the same way every poll.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TooSmall {
pub staged: u64,
pub free: u64,
pub total: u64,
pub floor: u64,
}
impl std::fmt::Display for TooSmall {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{} this twin's data volume cannot hold its primary's data: a full resync had staged {} when {} of {} was left free (it keeps {} for itself), with its stale copy already given up. Grow this twin's volume — to at least main's data set — and it takes the resync again; until then it waits in `recovering` and asks main for no snapshot it cannot hold",
gunnar_grpc::TWIN_TOO_SMALL,
super::human(self.staged),
super::human(self.free),
super::human(self.total),
super::human(self.floor),
)
}
}
impl std::error::Error for TooSmall {}
/// **What the [`DROPPED`] marker says**: when, the head the twin held, its base's `at`, and the
/// bytes the give-up freed. Read tolerantly — the marker's PRESENCE is the fact; a field it does
/// not carry (a crash mid-write of an older build's) reads 0.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Dropped {
pub at_unix_ms: i64,
pub head: u64,
pub base_at: Option<u64>,
pub freed: u64,
}
impl Dropped {
fn to_text(&self) -> String {
format!(
"dropped_unix_ms {}\nhead {}\nbase_at {}\nfreed {}\n",
self.at_unix_ms,
self.head,
self.base_at
.map(|a| a.to_string())
.unwrap_or_else(|| "-".into()),
self.freed
)
}
fn parse(text: &str) -> Self {
let mut d = Dropped::default();
for line in text.lines() {
let mut w = line.split_whitespace();
match (w.next(), w.next()) {
(Some("dropped_unix_ms"), Some(v)) => d.at_unix_ms = v.parse().unwrap_or(0),
(Some("head"), Some(v)) => d.head = v.parse().unwrap_or(0),
(Some("base_at"), Some(v)) => d.base_at = v.parse().ok(),
(Some("freed"), Some(v)) => d.freed = v.parse().unwrap_or(0),
_ => {}
}
}
d
}
/// The sentence the status, the alarm and the refusals carry.
pub fn sentence(&self) -> String {
let base = self
.base_at
.map(|a| format!(", its kept base at {a}"))
.unwrap_or_default();
format!(
"this twin gave its stale copy up (it held up to seq {}{base}; {} freed) to make room for a full resync: it holds NO copy until the resync lands, and main is the only copy meanwhile",
self.head,
super::human(self.freed)
)
}
}
/// **Mark the give-up** — durable BEFORE the first byte of the stale copy goes.
pub fn mark_dropped(root: &Path, d: &Dropped) -> anyhow::Result<()> {
crate::durable::write_atomic_unframed(root, DROPPED, d.to_text().as_bytes())
.context("marking the stale copy given up")
}
/// The give-up in progress or done, when this box has marked one.
pub fn dropped(root: &Path) -> Option<Dropped> {
match std::fs::read_to_string(root.join(DROPPED)) {
Ok(text) => Some(Dropped::parse(&text)),
Err(e) if not_found(&e) => None,
// Unreadable but there: the copy is still to be treated as gone.
Err(_) => Some(Dropped::default()),
}
}
/// **Remember a resync the volume could not hold**: its size then.
pub fn note_too_small(root: &Path, t: &TooSmall) -> anyhow::Result<()> {
crate::durable::write_atomic_unframed(
root,
TOO_SMALL,
format!("total {}\nstaged {}\n", t.total, t.staged).as_bytes(),
)
.context("noting the resync the volume could not hold")
}
/// The volume size the last resync ran out of room on, when one did.
pub fn too_small_total(root: &Path) -> Option<u64> {
let text = std::fs::read_to_string(root.join(TOO_SMALL)).ok()?;
text.lines()
.find_map(|l| l.strip_prefix("total "))
.and_then(|v| v.trim().parse().ok())
}
/// Bytes of every file under `dir` (0 when it is not there).
pub fn bytes_under(dir: &Path) -> u64 {
files_under(dir)
.map(|fs| {
fs.iter()
.filter_map(|p| std::fs::metadata(p).ok())
.map(|m| m.len())
.sum()
})
.unwrap_or(0)
}
/// **Why this box holds no whole copy now**, or `None` (twin-resync-space): a twin that gave its
/// stale copy up for a resync, or one mid-install — its control plane may already be the new
/// snapshot's while its base is still the old one. A twin with either serves nothing: no
/// snapshot ([`serve`]) and no log, so no recharge reads from it.
pub fn not_whole(root: &Path) -> Option<String> {
if let Some(d) = dropped(root) {
return Some(format!("this twin holds no whole copy: {}", d.sentence()));
}
if root.join(INSTALLING).exists() {
return Some(format!(
"this twin holds no whole copy: it is inside a snapshot install ({INSTALLING}) — its control plane may already be the new snapshot's while its base is still the old one — and takes the snapshot again on its next poll"
));
}
None
}
// ── the twin's half: staged on disk as it arrives ───────────────────────────
/// One staged item: what it is, and where its blob lies in the staging file.
#[derive(Debug, Clone)]
struct StagedItem {
kind: u16,
meta: Vec<u8>,
off: u64,
len: u64,
}
/// **A snapshot arriving**: every piece's bytes appended to `<data-root>/txlog-snapshot-staging/
/// blobs` as it comes, the item table in memory. Dropped unfinished, the staging directory goes.
pub struct Stager {
dir: PathBuf,
file: Option<std::fs::File>,
off: u64,
items: Vec<StagedItem>,
/// Each staged item's crc32c over `meta ‖ blob` (restore-after-snapshot, 2026-10-07): kept
/// with a twin's base, sent with every item it serves, checked by the box it restores.
crcs: Vec<u32>,
open: Option<StagedItem>,
/// The open item's running crc.
open_crc: Crc32c,
at: Option<u64>,
/// Kept on drop: a restore's staging, resumed by the next stream ([`RestoreSnapshot`]).
keep: bool,
/// The volume the staging file is on.
volume: std::sync::Arc<dyn Volume>,
/// A twin's: what it may give up when the volume runs short ([`Stager::making_room`]).
room: Option<Room>,
}
impl Stager {
/// Start over under `root`, on the box's own filesystem: whatever an earlier attempt left is
/// removed first.
pub fn begin(root: &Path) -> anyhow::Result<Self> {
Self::begin_on(root, std::sync::Arc::new(Disk))
}
/// Start over under `root`, staging on `volume`.
pub fn begin_on(root: &Path, volume: std::sync::Arc<dyn Volume>) -> anyhow::Result<Self> {
let dir = root.join(STAGING);
match std::fs::remove_dir_all(&dir) {
Ok(()) => {}
Err(e) if not_found(&e) => {}
Err(e) => return Err(e).with_context(|| format!("clearing {}", dir.display())),
}
std::fs::create_dir_all(&dir).with_context(|| format!("creating {}", dir.display()))?;
let file = std::fs::OpenOptions::new()
.create_new(true)
.write(true)
.read(true)
.open(dir.join("blobs"))
.context("creating the staging file")?;
Ok(Self {
dir,
file: Some(file),
off: 0,
items: Vec::new(),
crcs: Vec::new(),
open: None,
open_crc: Crc32c::new(),
at: None,
volume,
room: None,
keep: false,
})
}
/// **A twin's stager makes room** (twin-resync-space, 2026-10-06): before each piece the
/// volume is measured, and when the piece would leave less than [`staging_floor`] free — or
/// the volume answers `ENOSPC` — `give_up` is called ONCE: it gives this twin's stale copy up
/// (the [`DROPPED`] marker first) and answers the bytes it freed. Short again after that, the
/// stream ends with [`TooSmall`]. `given_up`: the copy is already gone (a resync restarted
/// after the marker was written), so a shortage is [`TooSmall`] at once.
pub fn making_room(mut self, give_up: Box<GiveUp>, given_up: bool) -> Self {
self.room = Some(Room { give_up, given_up });
self
}
/// Room for `len` more bytes, giving the stale copy up when there is none. `enospc`: the
/// volume has just refused a write, whatever the measure says.
fn room_for(&mut self, len: u64, enospc: bool) -> anyhow::Result<()> {
let staged = self.off;
let Some(room) = self.room.as_mut() else {
return Ok(());
};
let (total, free) = match self.volume.space(&self.dir) {
Some(m) => m,
// Unmeasurable is never "full" — unless the volume itself said so.
None if enospc => (0, 0),
None => return Ok(()),
};
let floor = staging_floor(total);
if !enospc && free >= len.saturating_add(floor) {
return Ok(());
}
if room.given_up {
return Err(TooSmall {
staged,
free,
total,
floor,
}
.into());
}
room.given_up = true;
let freed = (room.give_up)(&Short {
staged,
free,
total,
floor,
enospc,
})?;
match self.volume.space(&self.dir) {
Some((total, free)) if free < len.saturating_add(staging_floor(total)) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, freed, free, total, "the stale copy is given up and the volume is still short for the resync");
Err(TooSmall {
staged,
free,
total,
floor: staging_floor(total),
}
.into())
}
_ => Ok(()),
}
}
/// One piece's bytes at `off`, with room made for them on a twin.
fn write_piece(&mut self, blob: &[u8], off: u64) -> anyhow::Result<()> {
let len = blob.len() as u64;
let mut refused = false;
loop {
self.room_for(len, refused)?;
match self
.volume
.write_at(self.file.as_ref().expect("staging file"), blob, off)
{
Ok(()) => return Ok(()),
// The volume ran out before the measure said so (a filesystem's own overhead):
// the same answer — give the stale copy up once, then it is too small.
Err(e) if is_no_space(&e) && self.room.is_some() => {
if refused {
let (total, free) = self.volume.space(&self.dir).unwrap_or((0, 0));
return Err(TooSmall {
staged: self.off,
free,
total,
floor: staging_floor(total),
}
.into());
}
refused = true;
}
Err(e) => return Err(e).context("writing a snapshot piece to the staging file"),
}
}
}
/// Take one piece.
pub fn feed(&mut self, p: Piece<'_>) -> anyhow::Result<()> {
anyhow::ensure!(
self.at.is_none(),
"a snapshot piece arrived after the SnapshotAt terminator"
);
let off = self.off;
match self.open.as_ref() {
Some(open) => anyhow::ensure!(
open.kind == p.kind,
"a snapshot piece of kind {} arrived inside an item of kind {}",
p.kind,
open.kind
),
None => {
self.open = Some(StagedItem {
kind: p.kind,
meta: p.meta.to_vec(),
off,
len: 0,
});
self.open_crc = Crc32c::new();
self.open_crc.update(p.meta);
}
}
if !p.blob.is_empty() {
self.write_piece(p.blob, off)?;
self.off += p.blob.len() as u64;
self.open.as_mut().expect("the open item").len += p.blob.len() as u64;
self.open_crc.update(p.blob);
}
if !p.last {
return Ok(());
}
let done = self.open.take().expect("the open item");
anyhow::ensure!(
done.len == p.blob_len,
"snapshot item of kind {}: {} of {} blob bytes arrived",
done.kind,
done.len,
p.blob_len
);
self.crcs.push(self.open_crc.finish());
if done.kind == Kind::SnapshotAt as u16 {
match Meta::from_bytes(&done.meta) {
Ok(Meta::SnapshotAt { at }) => self.at = Some(at),
other => anyhow::bail!("a SnapshotAt terminator whose meta is {other:?}"),
}
}
self.items.push(done);
Ok(())
}
/// Has the terminator arrived?
pub fn terminated(&self) -> Option<u64> {
self.at
}
/// Every piece arrived: synced, ready to install.
pub fn finish(mut self) -> anyhow::Result<Staged> {
anyhow::ensure!(
self.open.is_none(),
"the snapshot stream ended inside an item"
);
let at = self
.at
.ok_or_else(|| anyhow!("the snapshot stream carried no SnapshotAt terminator"))?;
let file = self.file.take().expect("staging file");
file.sync_all().context("syncing the staging file")?;
let staged = Staged {
at,
dir: std::mem::take(&mut self.dir),
file,
items: std::mem::take(&mut self.items),
crcs: std::mem::take(&mut self.crcs),
bytes: self.off,
};
Ok(staged)
}
}
impl Drop for Stager {
fn drop(&mut self) {
if !self.dir.as_os_str().is_empty() && !self.keep {
self.file.take();
let _ = std::fs::remove_dir_all(&self.dir);
}
}
}
/// **A snapshot that arrived whole**, on the twin's data disk. Dropped, the staging directory
/// goes.
pub struct Staged {
/// The log's head the state was taken at.
pub at: u64,
dir: PathBuf,
file: std::fs::File,
items: Vec<StagedItem>,
/// Each item's crc32c (meta ‖ blob), as it was staged.
crcs: Vec<u32>,
/// Blob bytes staged.
pub bytes: u64,
}
impl Staged {
/// How many items, the terminator included.
pub fn items(&self) -> usize {
self.items.len()
}
}
impl Drop for Staged {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.dir);
}
}
// ── the twin's kept base (ruling 2026-09-22, "the twin only stores") ────────
/// `<data-root>/txlog-snapshot-base/`: **the snapshot a twin installed, KEPT as bytes** —
/// `blobs` (the staging file, unchanged) and `items` (the item table). Per box, never framed.
///
/// A store-only twin does not apply a snapshot's data items (packs, refs, LFS, repository
/// stores, lines, files), and `restart_after(at)` moves its log past them — so this file IS
/// its copy of everything at or below `at`. A recharge from the twin streams it first
/// ([`serve`]) and the twin's log from `at + 1` after it; the new main installs it whole, and
/// that is where the indexes are built.
pub const BASE: &str = "txlog-snapshot-base";
const BASE_ITEMS: &str = "items";
const BASE_BLOBS: &str = "blobs";
/// Each item's crc32c (meta ‖ blob), `u32 LE` per item (restore-after-snapshot, 2026-10-07).
/// Absent beside a base kept before: the base then serves no crc and is named by its item table.
const BASE_CRCS: &str = "crcs";
/// The item table, length-prefixed: `kind u16 | meta_len u32 | meta | off u64 | len u64`, LE.
fn encode_items(items: &[StagedItem]) -> Vec<u8> {
let mut out = Vec::with_capacity(items.iter().map(|i| 22 + i.meta.len()).sum());
for i in items {
out.extend_from_slice(&i.kind.to_le_bytes());
out.extend_from_slice(&(i.meta.len() as u32).to_le_bytes());
out.extend_from_slice(&i.meta);
out.extend_from_slice(&i.off.to_le_bytes());
out.extend_from_slice(&i.len.to_le_bytes());
}
out
}
fn decode_items(mut b: &[u8]) -> anyhow::Result<Vec<StagedItem>> {
fn take<'a>(b: &mut &'a [u8], n: usize) -> anyhow::Result<&'a [u8]> {
anyhow::ensure!(b.len() >= n, "the kept snapshot's item table is truncated");
let (h, t) = b.split_at(n);
*b = t;
Ok(h)
}
let mut out = Vec::new();
while !b.is_empty() {
let kind = u16::from_le_bytes(take(&mut b, 2)?.try_into()?);
let mlen = u32::from_le_bytes(take(&mut b, 4)?.try_into()?) as usize;
let meta = take(&mut b, mlen)?.to_vec();
let off = u64::from_le_bytes(take(&mut b, 8)?.try_into()?);
let len = u64::from_le_bytes(take(&mut b, 8)?.try_into()?);
out.push(StagedItem {
kind,
meta,
off,
len,
});
}
Ok(out)
}
/// **Keep `snap` as this twin's base**: its item table written beside the staging file, and
/// the staging directory renamed over the previous base. Called by [`install`] on a twin BEFORE
/// the log restarts after `at`, so there is no instant at which the log has moved past data
/// that no file on this box holds (a crash before the rename leaves the install marker, and
/// the twin asks for the snapshot again).
fn keep_as_base(root: &Path, snap: &Staged) -> anyhow::Result<()> {
crate::durable::write_atomic_unframed(&snap.dir, BASE_ITEMS, &encode_items(&snap.items))
.context("writing the kept snapshot's item table")?;
if snap.crcs.len() == snap.items.len() {
let crcs: Vec<u8> = snap.crcs.iter().flat_map(|c| c.to_le_bytes()).collect();
crate::durable::write_atomic_unframed(&snap.dir, BASE_CRCS, &crcs)
.context("writing the kept snapshot's item crcs")?;
}
let base = root.join(BASE);
let old = root.join(format!("{BASE}.old"));
match std::fs::remove_dir_all(&old) {
Ok(()) => {}
Err(e) if not_found(&e) => {}
Err(e) => return Err(e).with_context(|| format!("clearing {}", old.display())),
}
match std::fs::rename(&base, &old) {
Ok(()) => {}
Err(e) if not_found(&e) => {}
Err(e) => return Err(e).with_context(|| format!("moving {} aside", base.display())),
}
std::fs::rename(&snap.dir, &base)
.with_context(|| format!("keeping the snapshot as {}", base.display()))?;
crate::durable::sync_dir(root)
.with_context(|| format!("flushing {} into place", base.display()))?;
let _ = std::fs::remove_dir_all(&old);
Ok(())
}
/// The `at` of the base this twin keeps, when it keeps one.
pub fn base_at(root: &Path) -> anyhow::Result<Option<u64>> {
let items = match std::fs::read(root.join(BASE).join(BASE_ITEMS)) {
Ok(b) => decode_items(&b)?,
Err(e) if not_found(&e) => return Ok(None),
Err(e) => return Err(e).context("reading the kept snapshot's item table"),
};
match items.last().map(|i| (i.kind, Meta::from_bytes(&i.meta))) {
Some((k, Ok(Meta::SnapshotAt { at }))) if k == Kind::SnapshotAt as u16 => Ok(Some(at)),
_ => anyhow::bail!("the kept snapshot does not end with its SnapshotAt terminator"),
}
}
/// **Stream the base this twin keeps**, piece by piece in the producer's shape (meta on an
/// item's first piece, [`BLOB_LEN_AT_LAST`] until its last), through ONE preallocated
/// [`PIECE_BYTES`] buffer. Answers the base's `at`.
pub fn stream_base(root: &Path, sink: &mut Sink<'_>) -> anyhow::Result<u64> {
stream_base_from(root, 0, 0, "", &mut |p, _| sink(p)).map(|(at, _)| at)
}
/// **Where a piece of a twin's base goes** (restore-after-snapshot, 2026-10-07): the item's index
/// in the base, the byte of its blob the piece starts at, the item's crc on its last piece (when
/// the base kept crcs), and which base ([`base_id`]). What lets a box being restored stage a base
/// across several capped calls and ask for the rest from where it holds it.
#[derive(Debug, Clone, Copy)]
pub struct PiecePos<'a> {
pub item: u64,
pub offset: u64,
pub crc: Option<u32>,
pub id: &'a str,
}
/// Where the pieces of a resumable snapshot go: each with its [`PiecePos`] — `None` for a
/// snapshot that cannot be resumed (a primary's own, a new cut every time).
pub type PosSink<'s> = dyn FnMut(Piece<'_>, Option<PiecePos<'_>>) -> anyhow::Result<()> + 's;
/// **A resumed snapshot asked of a twin whose base is another one now** (it took a new snapshot
/// since): never spliced — the asker starts again from the first item.
#[derive(Debug)]
pub struct SnapshotChanged {
pub asked: String,
pub now: String,
}
/// The refusal word for [`SnapshotChanged`].
pub const SNAPSHOT_CHANGED: &str = "snapshot-changed";
impl std::fmt::Display for SnapshotChanged {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"REFUSED [{SNAPSHOT_CHANGED}] the rest of snapshot {} was asked for, and this twin's base is snapshot {} now (it took a \
new one since): two snapshots are never spliced — ask for {} from its first item",
self.asked, self.now, self.now
)
}
}
impl std::error::Error for SnapshotChanged {}
fn read_crcs(dir: &Path, items: usize) -> Option<Vec<u32>> {
let b = std::fs::read(dir.join(BASE_CRCS)).ok()?;
(b.len() == items * 4).then(|| {
b.chunks_exact(4)
.map(|c| u32::from_le_bytes(c.try_into().expect("four bytes")))
.collect()
})
}
/// **The name of the base this twin keeps**: a digest of its item table, its item crcs and its
/// size — two bases differ in it whenever any item differs (a new snapshot, even at the same `at`).
pub fn base_id(root: &Path) -> anyhow::Result<Option<String>> {
let dir = root.join(BASE);
let items = match std::fs::read(dir.join(BASE_ITEMS)) {
Ok(b) => b,
Err(e) if not_found(&e) => return Ok(None),
Err(e) => return Err(e).context("reading the kept snapshot's item table"),
};
let mut h = blake3::Hasher::new();
h.update(b"gunnar-snapshot-base-v1\n");
h.update(&items);
h.update(&std::fs::read(dir.join(BASE_CRCS)).unwrap_or_default());
h.update(
&std::fs::metadata(dir.join(BASE_BLOBS))
.map(|m| m.len())
.unwrap_or(0)
.to_le_bytes(),
);
Ok(Some(h.finalize().to_hex()[..32].to_string()))
}
/// The kept base's size in blob bytes (0 = none).
pub fn base_bytes(root: &Path) -> u64 {
std::fs::metadata(root.join(BASE).join(BASE_BLOBS))
.map(|m| m.len())
.unwrap_or(0)
}
/// **[`stream_base`] from item `from_item`, byte `from_offset` of its blob, on** — each piece
/// with its [`PiecePos`] (restore-after-snapshot, 2026-10-07). The items before `from_item` and
/// the bytes below the offset are not read: a resumed 617 MB base costs the twin its rest per
/// call. `expect_id` (non-empty): the base the asker holds part of — another one now is
/// [`SnapshotChanged`]. An item resumed part-way carries its meta on its first piece again.
/// Answers the base's `at` and its id.
pub fn stream_base_from(
root: &Path,
from_item: u64,
from_offset: u64,
expect_id: &str,
sink: &mut PosSink<'_>,
) -> anyhow::Result<(u64, String)> {
let dir = root.join(BASE);
let items = decode_items(
&std::fs::read(dir.join(BASE_ITEMS)).context("reading the kept snapshot's item table")?,
)?;
let crcs = read_crcs(&dir, items.len());
let file = std::fs::File::open(dir.join(BASE_BLOBS)).context("opening the kept snapshot")?;
let at = base_at(root)?.ok_or_else(|| anyhow!("the kept snapshot vanished"))?;
let id = base_id(root)?.ok_or_else(|| anyhow!("the kept snapshot vanished"))?;
if !expect_id.trim().is_empty() && expect_id.trim() != id {
return Err(SnapshotChanged {
asked: expect_id.trim().to_owned(),
now: id,
}
.into());
}
anyhow::ensure!(
(from_item as usize) < items.len() || (from_item == 0 && items.is_empty()),
"item {from_item} of snapshot {id} was asked for, and it has {} item(s)",
items.len()
);
let mut buf = vec![0u8; PIECE_BYTES];
for (i, item) in items.iter().enumerate().skip(from_item as usize) {
let mut done = if i as u64 == from_item {
from_offset
} else {
0
};
anyhow::ensure!(
done <= item.len,
"byte {done} of item {i} of snapshot {id} was asked for, and it is {} byte(s)",
item.len
);
let mut first = true;
loop {
let n = (item.len - done).min(PIECE_BYTES as u64) as usize;
file.read_exact_at(&mut buf[..n], item.off + done)
.context("reading the kept snapshot")?;
let offset = done;
done += n as u64;
let last = done == item.len;
let pos = PiecePos {
item: i as u64,
offset,
crc: if last {
crcs.as_ref().map(|c| c[i])
} else {
None
},
id: &id,
};
sink(
Piece {
kind: item.kind,
meta: if first { &item.meta } else { &[] },
blob_len: if last { item.len } else { BLOB_LEN_AT_LAST },
blob: &buf[..n],
last,
},
Some(pos),
)?;
first = false;
if last {
break;
}
}
}
Ok((at, id))
}
/// **`Mirror.SnapshotFetch`, on whichever box is asked.** A primary produces one from its
/// stores ([`produce`]). A twin applies no data, so a snapshot of its stores would be a
/// snapshot of its control plane with every repository missing — it serves the base it KEPT
/// instead, which is exactly the state at its log's `first - 1`. A twin with no base has a log
/// that starts at 1 and needs no snapshot; asking it for one is refused by name rather than
/// answered with half a box.
pub fn serve(stores: &ServerStores, sink: &mut Sink<'_>) -> anyhow::Result<u64> {
serve_from(stores, 0, 0, "", &mut |p, _| sink(p))
}
/// [`serve`], resumable on a twin (restore-after-snapshot, 2026-10-07): its base from item
/// `from_item`, byte `from_offset` on, every piece with its [`PiecePos`] ([`stream_base_from`]).
/// A primary's snapshot is a new cut every time and is not resumed: asked from anywhere but its
/// start it refuses by name, and its pieces carry no position.
pub fn serve_from(
stores: &ServerStores,
from_item: u64,
from_offset: u64,
expect_id: &str,
sink: &mut PosSink<'_>,
) -> anyhow::Result<u64> {
if !stores.is_twin() {
anyhow::ensure!(
from_item == 0 && from_offset == 0 && expect_id.trim().is_empty(),
"this box is a primary: its snapshot is a new cut every time it is asked, so it cannot be resumed at item {from_item} \
byte {from_offset} — ask for it from its start (a resumed snapshot is a twin's kept base)"
);
return produce(stores, &mut |p| sink(p, None));
}
// ★ twin-resync-space: a twin that gave its stale copy up, or is mid-install, holds no state
// at any one sequence — serving its base then would be serving half a set.
if let Some(why) = not_whole(stores.root()) {
anyhow::bail!(
"{why}. Nothing is streamed: recharge from main, or wait for the resync to land"
);
}
match base_at(stores.root())? {
Some(_) => stream_base_from(stores.root(), from_item, from_offset, expect_id, sink)
.map(|(at, _)| at),
None => anyhow::bail!(
"this twin keeps no snapshot: its log starts at {} and holds everything from there — \
replay the log; a snapshot of a store-only twin's stores would carry no repository",
stores.txlog().first_seq()
),
}
}
/// An item's blob read back: small ones into the reused buffer, large ones mapped.
enum Blob<'b> {
Inline(&'b [u8]),
Mapped(memmap2::Mmap),
}
impl std::ops::Deref for Blob<'_> {
type Target = [u8];
fn deref(&self) -> &[u8] {
match self {
Blob::Inline(b) => b,
Blob::Mapped(m) => m,
}
}
}
fn blob_of<'b>(
file: &std::fs::File,
item: &StagedItem,
buf: &'b mut Vec<u8>,
) -> anyhow::Result<Blob<'b>> {
if item.len <= INLINE_BYTES {
buf.clear();
buf.resize(item.len as usize, 0);
file.read_exact_at(buf, item.off)
.context("reading a staged snapshot item")?;
return Ok(Blob::Inline(buf));
}
// SAFETY: the staging file is this box's alone, written only by the stager, which has
// finished; nothing writes it while the mapping lives.
let map = unsafe {
memmap2::MmapOptions::new()
.offset(item.off)
.len(item.len as usize)
.map(file)
}
.context("mapping a staged snapshot item")?;
Ok(Blob::Mapped(map))
}
/// **Install `snap` into `stores`** — a twin, or a fresh box being restored (both follower-only).
/// Afterwards the stores hold exactly the snapshot's state and the log's head is `snap.at`.
pub fn install(stores: &ServerStores, snap: &Staged) -> anyhow::Result<()> {
anyhow::ensure!(
stores.txlog().is_follower(),
"a snapshot installs only on a twin or on a box being restored (a follower-only log)"
);
let metas: Vec<Option<Meta>> = snap
.items
.iter()
.map(|i| Meta::from_bytes(&i.meta).ok())
.collect();
match (snap.items.last(), metas.last()) {
(Some(i), Some(Some(Meta::SnapshotAt { at })))
if i.kind == Kind::SnapshotAt as u16 && *at == snap.at => {}
_ => anyhow::bail!(
"the snapshot does not end with its SnapshotAt({}) terminator",
snap.at
),
}
let root = stores.root();
// A store-only twin (ruling 2026-09-22) installs the control plane and KEEPS the rest.
let twin = stores.is_twin();
let was_dropped = dropped(root).is_some();
crate::durable::write_atomic_unframed(root, INSTALLING, format!("{}\n", snap.at).as_bytes())?;
// Whole, not merged: every store this box holds goes, and every framed file the snapshot
// does not carry.
for repo in stores.catalog().list(None)? {
stores
.reclaim_repo(&repo.path())
.with_context(|| format!("clearing {} before the snapshot", repo.key()))?;
}
let carried: BTreeSet<&str> = metas
.iter()
.filter_map(|m| match m {
Some(Meta::File { path, .. }) => Some(path.as_str()),
_ => None,
})
.collect();
// A secret this box holds that the snapshot does not name (one a fresh box minted before it
// became a twin) is not the source's: tombstoned, as the source would have it.
let named: BTreeSet<&str> = metas
.iter()
.filter_map(|m| match m {
Some(Meta::Secret { key, .. }) => Some(key.as_str()),
_ => None,
})
.collect();
for (key, value) in stores.control_db().export()?.secrets {
if value.is_some() && !named.contains(key.as_str()) {
use gunnar_store::SecretStore as _;
stores
.raw_secrets()
.delete(&key)
.with_context(|| format!("removing {key}, which the snapshot does not carry"))?;
}
}
for rel in txlog::FRAMED_FILES {
if !carried.contains(*rel) {
match std::fs::remove_file(root.join(rel)) {
Ok(()) => {}
Err(e) if not_found(&e) => {}
Err(e) => {
return Err(e).with_context(|| {
format!("removing {rel}, which the snapshot does not carry")
})
}
}
}
}
let mut buf: Vec<u8> = Vec::with_capacity(INLINE_BYTES as usize);
for (item, meta) in snap.items.iter().zip(&metas) {
match (Kind::from_u16(item.kind), meta) {
(Some(Kind::SnapshotAt), _) => {}
// **An item this build cannot install is not a warning, it is the end of the
// install** (2026-09-16, lane AUDIT-TWIN). `apply_raw` answers `Applied::Unknown`
// for a kind a newer box invented, and this loop threw that answer away. On the
// FOLLOWING path an unknown frame costs nothing because the twin's log still holds
// it — but a SNAPSHOT item is never in any log: its `seq` is 0, `restart_after`
// moves the twin's log past it, and the bytes are only in a staging file that is
// deleted at the end of this function. So a twin taking a snapshot from a newer
// primary silently and PERMANENTLY dropped every item of a kind it did not know,
// then reported itself caught up. The install refuses instead: the marker stays,
// the twin asks for the snapshot again, and it says which kind stopped it.
(None, _) => anyhow::bail!(
"the snapshot carries an item of kind {}, which this build cannot install — the \
source is a NEWER gunnar. A snapshot item is in no log, so installing the rest \
would lose it silently; this box installs nothing and stays where it was.",
item.kind
),
// The snapshot's pointer, whatever this box's was: a force, not a replayed claim.
(Some(Kind::ControlDoc), Some(Meta::ControlDoc { name, new, .. })) => {
let new = BlobId::from_hex(new).ok_or_else(|| {
anyhow!("snapshot: document {name} points at {new:?}, not hex")
})?;
let raw = stores.raw_control_store();
let current = raw.read_doc(name)?;
if current != Some(new) {
anyhow::ensure!(
raw.commit_doc(name, current, new)?,
"snapshot: document {name} moved during the install"
);
}
}
// A store-only twin applies the control plane only; the data items stay in the
// staging file, which is KEPT below as this twin's base (ruling 2026-09-22).
(_, _) if twin && !super::twin_follows(item.kind, &item.meta) => {}
(kind, _) => {
let blob = blob_of(&snap.file, item, &mut buf)?;
super::apply_raw(stores, 0, item.kind, &item.meta, &blob)
.with_context(|| format!("installing a snapshot item ({kind:?})"))?;
}
}
}
// The identity core's tables are the snapshot's document now (RA5-10), or none.
if let Some(identity) = stores.identity() {
crate::identity::follow(root, &identity)
.map_err(|why| anyhow!("installing a snapshot: {why}"))?;
}
if twin {
keep_as_base(root, snap)?;
}
stores.txlog().restart_after(snap.at)?;
let _ = std::fs::remove_file(root.join(super::FRESH_DOCS));
// The copy is whole again: the give-up and the too-small note are over. Before the install
// marker, so a crash between leaves the marker that asks for the snapshot again.
for done in [DROPPED, TOO_SMALL] {
match std::fs::remove_file(root.join(done)) {
Ok(()) => {}
Err(e) if not_found(&e) => {}
Err(e) => return Err(e).with_context(|| format!("removing {done} after the install")),
}
}
std::fs::remove_file(root.join(INSTALLING))?;
if was_dropped {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, at = snap.at, items = snap.items.len(), bytes = snap.bytes, "installed a snapshot; following resumes after it — RESOLVED: the full resync landed and this twin holds a whole copy again");
} else {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, at = snap.at, items = snap.items.len(), bytes = snap.bytes, "installed a snapshot; following resumes after it");
}
Ok(())
}
/// Is an install in progress (a crash mid-install) under `root` — or has a twin given its stale
/// copy up for a resync that has not landed ([`DROPPED`])? Either way only a whole snapshot
/// makes the box a copy again.
pub fn installing(root: &Path) -> bool {
root.join(INSTALLING).exists() || root.join(DROPPED).exists()
}
/// **One staging slot per DATA ROOT, not one per process.**
///
/// A box has one data root, so in the appliance a single slot and this map
/// behave identically — this changes nothing an appliance does. It matters for
/// a process driving two roots, which is the test binary: with one slot,
/// [`restore_piece`] replaced a stager whose `dir` did not match, and replacing
/// it DROPS it, which deletes the other root's staging directory under a stream
/// that is still arriving. Latent rather than observed — the suite flake of
/// 2026-09-16 that prompted the look turned out to be an unindexed pack, not
/// this — but a per-root resource in a per-process slot is wrong on its own,
/// and it is wrong in the twin's most important path.
static RESTORE_STAGERS: Mutex<BTreeMap<PathBuf, Stager>> = Mutex::new(BTreeMap::new());
/// **Forget whatever a broken `Bootstrap.Restore` left staged** — called by
/// [`super::restore_begin`], so every restore stream stages from nothing.
///
/// [`restore_piece`] keeps the stager in a process-global slot that only the terminator or an
/// error inside [`Stager::feed`] empties. A stream that BREAKS empties neither: the half that
/// arrived stayed in the slot, and the resent stream's pieces were appended to it, so [`install`]
/// ran over the union of two snapshots. Mostly that is the same data twice; where the two differ
/// it is silent divergence, because `install` builds its `carried` and `named` sets from every
/// item it holds — a framed file or a secret the SOURCE deleted between the attempts was
/// "carried" by the stale half and kept, with its stale bytes, on a box whose whole contract is
/// that it holds exactly what the source holds. Installed whole, never merged.
///
/// Dropping the stager takes its staging directory with it ([`Stager`]'s `Drop`).
///
/// Scoped to `root` by [`RESTORE_STAGERS`]' key: a process driving two roots (the tests) must
/// not have one root's `restore_begin` — or one root's first piece — throw the other's stream
/// away mid-flight.
pub fn forget_restore_stager(root: &Path) {
let mut slots = RESTORE_STAGERS.lock().unwrap_or_else(|p| p.into_inner());
let _ = slots.remove(root);
}
/// **`Bootstrap.Restore`, a snapshot piece** (`seq` 0): staged on disk until the `SnapshotAt`
/// terminator, which installs the snapshot and answers `Some(at)` — the restore then continues
/// with the log from `at + 1`.
pub fn restore_piece(stores: &ServerStores, p: Piece<'_>) -> anyhow::Result<Option<u64>> {
let root = stores.root().to_path_buf();
let mut slots = RESTORE_STAGERS.lock().unwrap_or_else(|p| p.into_inner());
// This root's stager, begun if this is the stream's first piece. Another root's stager is
// left exactly where it is: it belongs to a stream that is still arriving.
if !slots.contains_key(&root) {
slots.insert(root.clone(), Stager::begin(&root)?);
}
let fed = slots.get_mut(&root).expect("stager").feed(p);
if let Err(err) = fed {
slots.remove(&root);
return Err(err);
}
if slots.get(&root).expect("stager").terminated().is_none() {
return Ok(None);
}
let staged = slots.remove(&root).expect("stager").finish()?;
install(stores, &staged)?;
// Whatever a resumable stream had staged of a twin's base is no part of this box now.
forget_restore_snapshot_or_warn(&root);
Ok(Some(staged.at))
}
// ── a twin's base, staged on the box being restored, across streams ─────────────────────────
/// `<data-root>/txlog-restore-snapshot/`: **the twin's base a box being restored holds part of**
/// (restore-after-snapshot, 2026-10-07) — `blobs` (its items' bytes, in order) and [`RESTORE_RECORD`]
/// (which base, the items staged whole, the open item and the bytes of it that are DURABLE). Per
/// box, never framed. Gone when the snapshot is installed, when the restore finishes, and when a
/// fresh box begins one.
pub const RESTORE_SNAPSHOT: &str = "txlog-restore-snapshot";
const RESTORE_RECORD: &str = "restore.rec";
const RESTORE_MAGIC: &[u8] = b"gunnar-restore-snapshot v1\n";
/// Bytes staged between two durable commits of a restore's snapshot.
pub const SNAPSHOT_COMMIT_EVERY: u64 = 8 << 20;
/// **What a box being restored holds of a twin's base**, as its record says: what
/// `Bootstrap.Status.restore_snapshot_*` answers, and where the next stream resumes it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HeldSnapshot {
/// The base ([`base_id`] on the twin).
pub id: String,
/// Items staged whole.
pub items: u64,
/// Bytes of the next item staged.
pub bytes: u64,
/// The terminator is staged: the whole base is here, waiting to be installed.
pub whole: bool,
}
/// The restore's record, decoded.
struct Record {
id: String,
off: u64,
at: Option<u64>,
items: Vec<StagedItem>,
crcs: Vec<u32>,
open: Option<(StagedItem, u32)>,
}
fn encode_record(id: &str, st: &Stager) -> Vec<u8> {
let mut out = RESTORE_MAGIC.to_vec();
out.extend_from_slice(&(id.len() as u32).to_le_bytes());
out.extend_from_slice(id.as_bytes());
out.extend_from_slice(&st.off.to_le_bytes());
out.push(u8::from(st.at.is_some()));
out.extend_from_slice(&st.at.unwrap_or(0).to_le_bytes());
let items = encode_items(&st.items);
out.extend_from_slice(&(items.len() as u64).to_le_bytes());
out.extend_from_slice(&items);
out.extend_from_slice(&(st.crcs.len() as u32).to_le_bytes());
for c in &st.crcs {
out.extend_from_slice(&c.to_le_bytes());
}
match &st.open {
Some(o) => {
out.push(1);
out.extend_from_slice(&encode_items(std::slice::from_ref(o)));
out.extend_from_slice(&st.open_crc.state().to_le_bytes());
}
None => out.push(0),
}
out
}
fn decode_record(b: &[u8]) -> anyhow::Result<Record> {
fn take<'a>(b: &mut &'a [u8], n: usize) -> anyhow::Result<&'a [u8]> {
anyhow::ensure!(b.len() >= n, "the restore's snapshot record is truncated");
let (h, t) = b.split_at(n);
*b = t;
Ok(h)
}
let mut b = b;
anyhow::ensure!(
take(&mut b, RESTORE_MAGIC.len())? == RESTORE_MAGIC,
"not a restore's snapshot record"
);
let n = u32::from_le_bytes(take(&mut b, 4)?.try_into()?) as usize;
let id = String::from_utf8(take(&mut b, n)?.to_vec())?;
let off = u64::from_le_bytes(take(&mut b, 8)?.try_into()?);
let has_at = take(&mut b, 1)?[0] == 1;
let at = u64::from_le_bytes(take(&mut b, 8)?.try_into()?);
let n = u64::from_le_bytes(take(&mut b, 8)?.try_into()?) as usize;
let items = decode_items(take(&mut b, n)?)?;
let n = u32::from_le_bytes(take(&mut b, 4)?.try_into()?) as usize;
let mut crcs = Vec::with_capacity(n);
for _ in 0..n {
crcs.push(u32::from_le_bytes(take(&mut b, 4)?.try_into()?));
}
let open = if take(&mut b, 1)?[0] == 1 {
anyhow::ensure!(b.len() >= 4, "the restore's open item is truncated");
let item = decode_items(&b[..b.len() - 4])?
.pop()
.ok_or_else(|| anyhow!("the restore's open item is missing"))?;
let state = u32::from_le_bytes(b[b.len() - 4..].try_into()?);
Some((item, state))
} else {
None
};
anyhow::ensure!(
crcs.len() == items.len(),
"the restore's snapshot record has {} crcs for {} items",
crcs.len(),
items.len()
);
let end = open
.as_ref()
.map(|(o, _)| o.off + o.len)
.or_else(|| items.last().map(|i| i.off + i.len))
.unwrap_or(0);
anyhow::ensure!(
end == off,
"the restore's snapshot record counts {off} bytes and its items end at {end}"
);
Ok(Record {
id,
off,
at: has_at.then_some(at),
items,
crcs,
open,
})
}
/// **What this box holds of a twin's base**, out of its record (the durable part, which is what
/// the next stream finds once the staging is cut back to it). `None`: nothing staged.
pub fn held_snapshot(root: &Path) -> Option<HeldSnapshot> {
let r = decode_record(&std::fs::read(root.join(RESTORE_SNAPSHOT).join(RESTORE_RECORD)).ok()?)
.ok()?;
Some(HeldSnapshot {
id: r.id,
items: r.items.len() as u64,
bytes: r.open.as_ref().map(|(o, _)| o.len).unwrap_or(0),
whole: r.at.is_some(),
})
}
/// **Throw a restore's staged snapshot away, durably** (the directory, then the data root's entry).
/// Nothing staged is `Ok`; a removal or a directory fsync that failed is the caller's error.
pub fn forget_restore_snapshot(root: &Path) -> anyhow::Result<()> {
let dir = root.join(RESTORE_SNAPSHOT);
match std::fs::remove_dir_all(&dir) {
Ok(()) => {}
Err(e) if not_found(&e) => return Ok(()),
Err(e) => return Err(e).with_context(|| format!("removing {}", dir.display())),
}
crate::durable::sync_dir(root).with_context(|| {
format!(
"flushing the removal of {RESTORE_SNAPSHOT} from {}",
root.display()
)
})
}
/// [`forget_restore_snapshot`] where what it throws away is already no part of the box (the
/// snapshot installed, the restore finished): a failure is said, by name, and is not the
/// caller's — the next restore start throws it away again.
pub fn forget_restore_snapshot_or_warn(root: &Path) {
if let Err(err) = forget_restore_snapshot(root) {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, err = %format!("{err:#}"), "a restore's staged snapshot could not be thrown away durably; the next restore start throws it away again");
}
}
/// **A twin's base staged on the box being restored, ACROSS streams** (restore-after-snapshot,
/// 2026-10-07). The [`Stager`] a twin's own resync uses, in [`RESTORE_SNAPSHOT`], kept on drop,
/// with its state written to [`RESTORE_RECORD`] after the bytes it counts were synced — every
/// [`SNAPSHOT_COMMIT_EVERY`], when a stream ends, and on drop. A crash between a write and its
/// commit leaves bytes nobody counted, and [`RestoreSnapshot::resume`] cuts them off.
pub struct RestoreSnapshot {
stager: Stager,
id: String,
/// Blob bytes durable (the record's count).
committed: u64,
/// Items in the last record written.
committed_items: usize,
}
/// A staged item that does not match the crc the twin vouched for.
#[derive(Debug)]
pub struct ItemCrcMismatch {
pub item: u64,
pub got: u32,
pub want: u32,
}
impl std::fmt::Display for ItemCrcMismatch {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"item {} of the snapshot was put together as crc {:08x} and the twin kept it as {:08x}",
self.item, self.got, self.want
)
}
}
impl std::error::Error for ItemCrcMismatch {}
impl RestoreSnapshot {
/// **What a previous stream staged, made trustworthy**: the staging file cut back to the
/// record's durable count. `None`: nothing (or nothing trustworthy — a record that does not
/// parse, or a file SHORTER than its count — which is then thrown away).
pub fn resume(root: &Path) -> anyhow::Result<Option<RestoreSnapshot>> {
let dir = root.join(RESTORE_SNAPSHOT);
let rec = match std::fs::read(dir.join(RESTORE_RECORD)) {
Ok(b) => b,
Err(e) if not_found(&e) => {
forget_restore_snapshot(root)?;
return Ok(None);
}
Err(e) => return Err(e).context("reading the restore's snapshot record"),
};
let r = match decode_record(&rec) {
Ok(r) => r,
Err(err) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "the restore's staged snapshot has a record that does not parse; it is dropped and the snapshot starts again");
forget_restore_snapshot(root)?;
return Ok(None);
}
};
let file = match std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(dir.join("blobs"))
{
Ok(f) => f,
Err(_) => {
forget_restore_snapshot(root)?;
return Ok(None);
}
};
let len = file
.metadata()
.context("reading the restore's staged snapshot")?
.len();
if len < r.off {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, id = %r.id, durable = r.off, on_disk = len, "the restore's staged snapshot is SHORTER than its record says was made durable; it is not trusted and starts again");
drop(file);
forget_restore_snapshot(root)?;
return Ok(None);
}
if len > r.off {
file.set_len(r.off)
.context("cutting the restore's staged snapshot back to its durable count")?;
file.sync_all()
.context("flushing the restore's staged snapshot")?;
}
let (open, open_crc) = match r.open {
Some((o, state)) => (Some(o), Crc32c::resume(state)),
None => (None, Crc32c::new()),
};
let items = r.items.len();
let stager = Stager {
dir,
file: Some(file),
off: r.off,
items: r.items,
crcs: r.crcs,
open,
open_crc,
at: r.at,
volume: std::sync::Arc::new(Disk),
room: None,
keep: true,
};
Ok(Some(RestoreSnapshot {
stager,
id: r.id,
committed: r.off,
committed_items: items,
}))
}
/// **Stage snapshot `id` from its first item**: whatever was staged before goes.
pub fn start(root: &Path, id: &str) -> anyhow::Result<RestoreSnapshot> {
forget_restore_snapshot(root)?;
let dir = root.join(RESTORE_SNAPSHOT);
std::fs::create_dir_all(&dir).with_context(|| format!("creating {}", dir.display()))?;
let file = std::fs::OpenOptions::new()
.create_new(true)
.write(true)
.read(true)
.open(dir.join("blobs"))
.context("creating the restore's snapshot staging file")?;
let stager = Stager {
dir,
file: Some(file),
off: 0,
items: Vec::new(),
crcs: Vec::new(),
open: None,
open_crc: Crc32c::new(),
at: None,
volume: std::sync::Arc::new(Disk),
room: None,
keep: true,
};
let mut s = RestoreSnapshot {
stager,
id: id.to_owned(),
committed: 0,
committed_items: 0,
};
s.write_record()?;
crate::durable::sync_dir(root)
.with_context(|| format!("flushing {} into {}", RESTORE_SNAPSHOT, root.display()))?;
Ok(s)
}
/// Which base.
pub fn id(&self) -> &str {
&self.id
}
/// `(items staged whole, bytes of the next one staged)` — in memory, the next byte wanted.
pub fn position(&self) -> (u64, u64) {
(
self.stager.items.len() as u64,
self.stager.open.as_ref().map(|o| o.len).unwrap_or(0),
)
}
/// The terminator is staged: the snapshot's `at`.
pub fn whole(&self) -> Option<u64> {
self.stager.at
}
/// One piece (already cut to the bytes not held). `want_crc`: the twin's crc of the item this
/// piece ends — checked before the item counts as staged ([`ItemCrcMismatch`]).
pub fn feed(&mut self, p: Piece<'_>, want_crc: Option<u32>) -> anyhow::Result<()> {
let before = self.stager.items.len();
self.stager.feed(p)?;
if self.stager.items.len() > before {
if let (Some(want), Some(&got)) = (want_crc, self.stager.crcs.last()) {
if got != want {
return Err(ItemCrcMismatch {
item: before as u64,
got,
want,
}
.into());
}
}
}
if self.stager.off - self.committed >= SNAPSHOT_COMMIT_EVERY || self.stager.at.is_some() {
self.commit()?;
}
Ok(())
}
/// **Make what is staged durable and count it**: the file synced, then the record.
pub fn commit(&mut self) -> anyhow::Result<()> {
if self.stager.off == self.committed
&& self.stager.items.len() == self.committed_items
&& self.stager.at.is_none()
{
return Ok(());
}
self.stager
.file
.as_ref()
.expect("staging file")
.sync_data()
.context("fdatasync of the restore's staged snapshot")?;
self.write_record()
}
fn write_record(&mut self) -> anyhow::Result<()> {
crate::durable::write_atomic_unframed(
&self.stager.dir,
RESTORE_RECORD,
&encode_record(&self.id, &self.stager),
)?;
self.committed = self.stager.off;
self.committed_items = self.stager.items.len();
Ok(())
}
/// The whole base, ready to install. Dropped after the install, it takes the staging with it.
pub fn finish(mut self) -> anyhow::Result<Staged> {
self.commit()?;
let mut stager = std::mem::replace(
&mut self.stager,
Stager {
dir: PathBuf::new(),
file: None,
off: 0,
items: Vec::new(),
crcs: Vec::new(),
open: None,
open_crc: Crc32c::new(),
at: None,
volume: std::sync::Arc::new(Disk),
room: None,
keep: true,
},
);
stager.keep = false;
stager.finish()
}
}
impl Drop for RestoreSnapshot {
/// A stream that never reached its end (dropped, refused, broken) keeps every byte it staged.
fn drop(&mut self) {
if self.stager.dir.as_os_str().is_empty() {
return;
}
if let Err(err) = self.commit() {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, id = %self.id, %err, "the restore's staged snapshot could not be committed as its stream ended; the next stream resumes from its last commit");
}
}
}