crates/gunnar-server/src/twin/intake.rs
//! **The restore's intake: an entry larger than one call, staged** (restore-large-entry,
//! 2026-10-07).
//!
//! # The defect
//!
//! A main recharged from its twin through monetize's restore lane (the gunnar guest) is fed by
//! guest calls that the host cuts: each stream the guest opens is cancelled 30 s after it opens
//! (the doors' own timeout; the whole call is cut at 60 s). The lane resumed per ENTRY — "a replay
//! too long for one call resumes on the next tick from head + 1" — and `Bootstrap.Restore`
//! gathered each entry in memory and dropped it when the stream ended inside it. So ONE entry
//! that takes longer than a call to stream (a 200 MiB pack over a slow link, MEASURED by the r5
//! integration test, row R1-recharge-main-lost: 41 calls, every one "the stream ended inside an
//! entry; resend from the box's head + 1") could never be restored, and the main never came back.
//!
//! # What this does
//!
//! An entry that spans more than one piece (or more than [`IN_MEMORY_UP_TO`]) is written to
//! `<data-root>/`[`PARTIAL_DIR`] as it arrives — its bytes in `entry.blob`, and in `entry.rec`
//! what they are (seq, kind, blob length, meta) and how many of them are DURABLE. The record is
//! rewritten (atomically, fsynced) after the bytes it counts were `fdatasync`ed: every
//! [`COMMIT_EVERY`], when the stream ends however it ends (a clean end, a broken one, the sender's
//! cancel — tonic reads a client's RST_STREAM as a clean end — or a refusal), and when the intake
//! is dropped. The next stream finds it: `Bootstrap.Status.restore_partial_{seq,bytes}` says
//! where to resume, the lane asks the twin for `Mirror.LogFetch{from: seq, from_offset: bytes}`,
//! and a call cut at its deadline loses at most the piece that was in flight — never the entry.
//!
//! The entry is applied ONLY when it is whole: its length is the declared one and, when the
//! sender stamped one (`LogChunk.entry_crc`, the twin's own payload crc, checked before the last
//! piece left it), the crc32c over `meta ‖ blob` put together here is that crc. Then it is
//! appended at its number and applied ([`super::restore_apply`]), and only then is the staging
//! area removed — durably, so nothing leaks onto the volume.
//!
//! # Idempotent, crash-safe
//!
//! * An entry is applied once: `append_at` takes only `head + 1`, an entry at or below the head
//! is checked and dropped, and the staged entry is the one at `head + 1` or it is thrown away
//! when the intake opens.
//! * A crash between a write and its commit leaves bytes past the recorded count; the intake
//! truncates the file to the count when it opens. A file SHORTER than its record (a filesystem
//! that lied) is not trusted: the entry starts again at byte 0.
//! * A crash between the apply and the removal leaves a staging area for an entry at or below
//! the head; the next intake removes it.
//!
//! # Older peers
//!
//! * An older TWIN ignores `from_offset` and sends the entry from byte 0 with no `offset` on its
//! chunks; the bytes this box holds are skipped as they arrive. Skipping makes the stream no
//! shorter, so an entry that needs more than one call STILL cannot pass an older twin: when a
//! stream of such a sender ends without one byte past what was held, the refusal says so by
//! name ([`TWIN_TOO_OLD`]) instead of asking for the same resend for ever. The estate upgrades
//! its twin together with its main.
//! * An older MAIN ignores the new fields and gathers entries in memory as it always did.
use std::collections::HashSet;
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use anyhow::Context;
use gunnar_core::txlog::{Crc32c, Entry};
use gunnar_grpc::{ControlError, ControlResult};
use super::Applied;
use crate::stores::ServerStores;
/// `<data-root>/txlog-restore-partial/`: the entry a restoring box holds part of. Per box,
/// never a transaction (`gunnar_core::txlog::is_per_box_path`).
pub const PARTIAL_DIR: &str = "txlog-restore-partial";
/// The entry's blob bytes so far.
const DATA: &str = "entry.blob";
/// What they are and how many of them are durable.
const RECORD: &str = "entry.rec";
/// The first line of [`RECORD`].
const RECORD_MAGIC: &str = "gunnar-restore-partial v1";
/// Bytes written between two durable commits of the staged entry: what a power cut may cost a
/// resume, never more.
pub const COMMIT_EVERY: u64 = 8 << 20;
/// An entry this small that starts at its first byte is gathered in memory, as before: it takes
/// a few seconds on any link a box is restored over, and staging it would cost two fsyncs per
/// entry over a log of many small ones. Larger ones (and every resumed one) are staged.
pub const IN_MEMORY_UP_TO: u64 = 4 << 20;
/// The refusal word for a sender that streams every entry from byte 0 when a staged entry needs
/// more than one call.
pub const TWIN_TOO_OLD: &str = "restore-twin-too-old";
/// **What this box holds of an entry it is restoring**, as its record says.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Staged {
pub seq: u64,
pub kind: u16,
pub blob_len: u64,
pub meta: Vec<u8>,
/// Bytes of the blob that are on disk AND counted here — the resume point.
pub durable: u64,
}
impl Staged {
fn to_text(&self) -> String {
format!(
"{RECORD_MAGIC}\nseq {}\nkind {}\nblob_len {}\ndurable {}\nmeta {}\n",
self.seq,
self.kind,
self.blob_len,
self.durable,
hex::encode(&self.meta)
)
}
fn from_text(t: &str) -> Option<Staged> {
let mut lines = t.lines();
if lines.next()? != RECORD_MAGIC {
return None;
}
let mut field = |name: &str| {
lines
.next()?
.strip_prefix(name)?
.strip_prefix(' ')
.map(str::to_owned)
};
let seq = field("seq")?.parse().ok()?;
let kind = field("kind")?.parse().ok()?;
let blob_len = field("blob_len")?.parse().ok()?;
let durable: u64 = field("durable")?.parse().ok()?;
let meta = hex::decode(field("meta")?).ok()?;
(durable <= blob_len && seq > 0).then_some(Staged {
seq,
kind,
blob_len,
meta,
durable,
})
}
}
/// The staged entry's record, or `None` (nothing staged, or a record that does not parse).
pub fn staged(root: &Path) -> Option<Staged> {
std::fs::read_to_string(root.join(PARTIAL_DIR).join(RECORD))
.ok()
.as_deref()
.and_then(Staged::from_text)
}
/// **`(seq, bytes held durably)` of the entry a restoring box holds part of** — what
/// `Bootstrap.Status.restore_partial_{seq,bytes}` says. `None` when nothing is staged.
pub fn partial(root: &Path) -> Option<(u64, u64)> {
staged(root).map(|s| (s.seq, s.durable))
}
/// **Throw the staging area away, durably**: the files, the directory, then the data root's
/// entry for it — so a removal after an apply survives a power cut and no staged bytes are left
/// on the volume.
pub fn forget(root: &Path) -> anyhow::Result<()> {
let dir = root.join(PARTIAL_DIR);
match std::fs::remove_dir_all(&dir) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(e) => return Err(e).with_context(|| format!("removing {}", dir.display())),
}
let r = File::open(root).with_context(|| format!("opening {} to flush it", root.display()))?;
r.sync_all().with_context(|| {
format!(
"flushing the removal of {PARTIAL_DIR} from {}",
root.display()
)
})?;
Ok(())
}
/// The record, rewritten atomically and fsynced. The directory is made (and its entry in the
/// data root made durable) the first time.
fn write_record(root: &Path, rec: &Staged) -> anyhow::Result<()> {
let dir = root.join(PARTIAL_DIR);
if !dir.is_dir() {
std::fs::create_dir_all(&dir).with_context(|| format!("creating {}", dir.display()))?;
crate::durable::sync_dir(root)
.with_context(|| format!("flushing {PARTIAL_DIR} into {}", root.display()))?;
}
crate::durable::write_atomic_unframed(&dir, RECORD, rec.to_text().as_bytes())
}
/// **What the box still holds, made trustworthy before a stream uses it**: the staged entry is
/// the one at `next` (anything else was applied or belongs to another history, and goes), and its
/// file is cut back to the durable count (a crash may have left a tail nobody counted). A file
/// shorter than its count is not trusted at all.
fn sanitize(root: &Path, next: u64) -> anyhow::Result<Option<(Staged, File)>> {
if !root.join(PARTIAL_DIR).exists() {
return Ok(None);
}
let Some(rec) = staged(root) else {
forget(root)?;
return Ok(None);
};
if rec.seq != next {
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
staged = rec.seq, next, "a staged restore entry that is not the next one is dropped (applied already, or another history)"
);
forget(root)?;
return Ok(None);
}
let path = root.join(PARTIAL_DIR).join(DATA);
let file = match OpenOptions::new().read(true).write(true).open(&path) {
Ok(f) => f,
Err(_) => {
forget(root)?;
return Ok(None);
}
};
let len = file
.metadata()
.with_context(|| format!("reading {}", path.display()))?
.len();
if len < rec.durable {
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
seq = rec.seq, durable = rec.durable, on_disk = len,
"the staged restore entry is SHORTER than its record says was made durable; it is not trusted and starts again at byte 0"
);
drop(file);
forget(root)?;
return Ok(None);
}
let mut file = file;
if len > rec.durable {
file.set_len(rec.durable).with_context(|| {
format!(
"cutting {} back to its durable {} bytes",
path.display(),
rec.durable
)
})?;
file.sync_all()
.with_context(|| format!("flushing {}", path.display()))?;
}
// The next byte written is the first one not held.
file.seek(SeekFrom::Start(rec.durable))
.with_context(|| format!("seeking {}", path.display()))?;
Ok(Some((rec, file)))
}
/// The data roots that have an intake open: ONE stream at a time stages into a box.
static OPEN: Mutex<Option<HashSet<PathBuf>>> = Mutex::new(None);
/// This root's place in [`OPEN`], given back on drop.
struct Held(PathBuf);
impl Held {
fn take(root: &Path) -> Option<Held> {
let mut open = OPEN.lock().unwrap_or_else(|p| p.into_inner());
open.get_or_insert_with(HashSet::new)
.insert(root.to_path_buf())
.then(|| Held(root.to_path_buf()))
}
}
impl Drop for Held {
fn drop(&mut self) {
let mut open = OPEN.lock().unwrap_or_else(|p| p.into_inner());
if let Some(set) = open.as_mut() {
set.remove(&self.0);
}
}
}
/// Where the open entry's bytes go.
enum Body {
/// Applied already (`seq` below the next one): its pieces are checked and dropped.
Skip,
/// Small and starting at byte 0: gathered here.
Memory(Vec<u8>),
/// Staged on disk: `written` bytes in the file, `rec.durable` of them committed.
Spool {
file: File,
rec: Staged,
written: u64,
},
}
/// The entry this stream is receiving.
struct Open {
seq: u64,
kind: u16,
meta: Vec<u8>,
blob_len: u64,
/// The byte of the blob this stream's next chunk starts at.
pos: u64,
/// This stream's chunks name their offsets (a sender that knows resumed restores).
explicit: bool,
/// What the box held of it before this stream reached it.
held_before: u64,
body: Body,
}
/// **One `Bootstrap.Restore` stream's intake on the server** — see the module docs.
pub struct Intake<'a> {
stores: &'a ServerStores,
_held: Held,
next: u64,
/// What the box held when this stream began, until the stream reaches that entry.
kept: Option<(Staged, File)>,
open: Option<Open>,
applied: u64,
lost: u64,
/// **The twin's base this box holds part of** (restore-after-snapshot, 2026-10-07), staged
/// across streams ([`super::snapshot::RestoreSnapshot`]).
snap: Option<super::snapshot::RestoreSnapshot>,
/// This stream carried pieces of a resumable snapshot.
snap_touched: bool,
}
impl<'a> Intake<'a> {
/// **Open the box's intake for one stream**: [`super::restore_begin_by`] (the owner check, the
/// next sequence), then what is staged made trustworthy ([`sanitize`]). `Ok(None)`: another
/// stream's intake still holds this box.
pub fn open(stores: &'a ServerStores, owner: &str) -> anyhow::Result<Option<Intake<'a>>> {
let Some(held) = Held::take(stores.root()) else {
return Ok(None);
};
let next = super::restore_begin_by(stores, owner)?;
let kept = sanitize(stores.root(), next)?;
if let Some((rec, _)) = &kept {
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
seq = rec.seq, held = rec.durable, blob_len = rec.blob_len,
"a restore stream resumes an entry this box holds part of: {} of {} bytes of seq {}", rec.durable, rec.blob_len, rec.seq
);
}
let mut intake = Intake {
stores,
_held: held,
next,
kept,
open: None,
applied: 0,
lost: 0,
snap: None,
snap_touched: false,
};
// A twin's base staged by earlier streams: cut back to what was made durable — and, when
// it is whole (a stream ended, or the box went down, between its last piece and the
// install), installed now: the log then follows from its `at + 1`.
if super::is_restoring(stores.root()) {
intake.snap = super::snapshot::RestoreSnapshot::resume(stores.root())?;
}
if let Some(s) = &intake.snap {
let (items, bytes) = s.position();
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
id = s.id(), items, bytes, whole = s.whole().is_some(),
"a restore stream resumes the twin's snapshot this box holds part of: {items} item(s) and {bytes} byte(s) of the next"
);
if s.whole().is_some() {
intake
.install_snapshot()
.map_err(|e| anyhow::anyhow!("{e}"))?;
}
}
Ok(Some(intake))
}
/// **The twin's base is whole: install it** — the box then holds everything up to its `at`,
/// and its log follows from `at + 1`. The staging goes with the install.
fn install_snapshot(&mut self) -> ControlResult<()> {
let Some(s) = self.snap.take() else {
return Ok(());
};
let id = s.id().to_owned();
let staged = s
.finish()
.map_err(|err| ControlError::Internal(format!("the twin's snapshot {id}: {err:#}")))?;
let at = staged.at;
super::snapshot::install(self.stores, &staged).map_err(|err| {
ControlError::Internal(format!("installing the twin's snapshot {id}: {err:#}"))
})?;
drop(staged);
super::snapshot::forget_restore_snapshot_or_warn(self.stores.root());
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, id = %id, at, "the twin's snapshot, staged across streams, is installed: this box holds everything up to seq {at}; its log follows from {}", at + 1);
self.next = at + 1;
Ok(())
}
/// **One piece of a twin's base, resumable** (`snapshot_id` set): placed by its item and
/// offset, the bytes this box holds skipped, a gap refused naming where to resume, the item's
/// crc checked at its end, and the base installed at its terminator.
fn take_snapshot_piece(&mut self, chunk: &gunnar_proto::LogChunk) -> ControlResult<()> {
let root = self.stores.root();
let id = chunk.snapshot_id.trim();
let (item, offset) = (chunk.snapshot_item.unwrap_or(0), chunk.offset.unwrap_or(0));
self.snap_touched = true;
if self.snap.as_ref().is_none_or(|s| s.id() != id) {
if (item, offset) != (0, 0) {
let held = self
.snap
.as_ref()
.map(|s| format!("part of snapshot {}", s.id()))
.unwrap_or_else(|| "none of it".into());
return Err(ControlError::FailedPrecondition(format!(
"a piece at item {item} byte {offset} of the twin's snapshot {id} arrived, and this box holds {held}; resend snapshot {id} from its first item"
)));
}
if let Some(old) = self.snap.take() {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, old = old.id(), new = id, "the twin's snapshot is another one now; what this box staged of the old one is dropped and the new one is taken from its first item");
}
self.snap = Some(
super::snapshot::RestoreSnapshot::start(root, id).map_err(|err| {
ControlError::Internal(format!("staging the twin's snapshot {id}: {err:#}"))
})?,
);
}
let s = self.snap.as_mut().expect("a snapshot is staged");
let (done, held) = s.position();
if item < done {
return Ok(());
}
if item > done || offset > held {
return Err(ControlError::FailedPrecondition(format!(
"a piece at item {item} byte {offset} of the twin's snapshot {id} arrived, and this box holds {done} item(s) and {held} byte(s) of the next; resend from item {done} byte {held}"
)));
}
let skip = (held - offset).min(chunk.blob.len() as u64) as usize;
if skip == chunk.blob.len() && !chunk.last {
return Ok(());
}
let kind = u16::try_from(chunk.kind).map_err(|_| {
ControlError::InvalidArgument(format!("snapshot piece of kind {}", chunk.kind))
})?;
let piece = super::snapshot::Piece {
kind,
meta: &chunk.meta,
blob_len: chunk.blob_len,
blob: &chunk.blob[skip..],
last: chunk.last,
};
if let Err(err) = s.feed(piece, chunk.entry_crc) {
if let Some(bad) = err.downcast_ref::<super::snapshot::ItemCrcMismatch>() {
let said = bad.to_string();
self.snap = None;
super::snapshot::forget_restore_snapshot_or_warn(root);
return Err(ControlError::FailedPrecondition(format!(
"{said}: nothing of the twin's snapshot {id} is installed — this box drops what it staged of it and takes it again from its first item"
)));
}
return Err(ControlError::InvalidArgument(format!(
"the twin's snapshot {id}, item {item}: {err:#}"
)));
}
if s.whole().is_some() {
self.install_snapshot()?;
}
Ok(())
}
/// The next sequence this box takes.
pub fn next(&self) -> u64 {
self.next
}
/// One chunk of the stream.
pub fn take(&mut self, chunk: &gunnar_proto::LogChunk) -> ControlResult<()> {
if chunk.seq == 0 {
if let Some(o) = &self.open {
return Err(ControlError::InvalidArgument(format!(
"a snapshot piece arrived with seq {} unfinished",
o.seq
)));
}
// A twin's base, resumable across streams (restore-after-snapshot).
if !chunk.snapshot_id.trim().is_empty() {
return self.take_snapshot_piece(chunk);
}
let kind = u16::try_from(chunk.kind).map_err(|_| {
ControlError::InvalidArgument(format!("snapshot piece of kind {}", chunk.kind))
})?;
let piece = super::snapshot::Piece {
kind,
meta: &chunk.meta,
blob_len: chunk.blob_len,
blob: &chunk.blob,
last: chunk.last,
};
if let Some(at) = super::snapshot::restore_piece(self.stores, piece).map_err(|err| {
ControlError::Internal(format!("installing the snapshot: {err:#}"))
})? {
self.next = at + 1;
}
return Ok(());
}
let kind = u16::try_from(chunk.kind).map_err(|_| {
ControlError::InvalidArgument(format!(
"seq {}: kind {} is not a frame kind",
chunk.seq, chunk.kind
))
})?;
match &self.open {
Some(o) if o.seq == chunk.seq => {
if o.kind != kind || o.blob_len != chunk.blob_len {
return Err(ControlError::InvalidArgument(format!(
"seq {}: a later piece says kind {kind} of {} bytes, the first said kind {} of {}",
o.seq, chunk.blob_len, o.kind, o.blob_len
)));
}
if let Some(at) = chunk.offset {
if at != o.pos {
return Err(ControlError::InvalidArgument(format!(
"seq {}: a piece at byte {at} arrived where byte {} was due",
o.seq, o.pos
)));
}
}
}
Some(o) => {
return Err(ControlError::InvalidArgument(format!(
"seq {} arrived with seq {} unfinished",
chunk.seq, o.seq
)))
}
None => self.open_entry(chunk, kind)?,
}
let stores = self.stores;
let root = stores.root();
let o = self.open.as_mut().expect("an entry is open");
let len = chunk.blob.len() as u64;
if o.pos + len > o.blob_len {
return Err(ControlError::InvalidArgument(format!(
"seq {}: more than the {} blob bytes it declared arrived",
o.seq, o.blob_len
)));
}
match &mut o.body {
Body::Skip => {}
Body::Memory(v) => v.extend_from_slice(&chunk.blob),
Body::Spool { file, rec, written } => {
// Bytes below what the box holds (a sender that started lower) are skipped.
if o.pos + len > *written {
let from = (*written - o.pos) as usize;
file.write_all(&chunk.blob[from..])
.map_err(|e| internal(o.seq, "writing the staged entry", e))?;
*written = o.pos + len;
if *written - rec.durable >= COMMIT_EVERY {
commit(root, file, rec, *written)
.map_err(|e| internal(o.seq, "committing the staged entry", e))?;
}
}
}
}
o.pos += len;
if chunk.last {
self.finish_entry(chunk.entry_crc)?;
}
Ok(())
}
/// The first chunk of an entry in this stream.
fn open_entry(&mut self, chunk: &gunnar_proto::LogChunk, kind: u16) -> ControlResult<()> {
let seq = chunk.seq;
let start = chunk.offset.unwrap_or(0);
let explicit = chunk.offset.is_some();
let mut open = Open {
seq,
kind,
meta: chunk.meta.clone(),
blob_len: chunk.blob_len,
pos: start,
explicit,
held_before: 0,
body: Body::Skip,
};
if start > chunk.blob_len {
return Err(ControlError::InvalidArgument(format!(
"seq {seq}: a piece at byte {start} of a {}-byte blob",
chunk.blob_len
)));
}
if seq < self.next {
self.open = Some(open);
return Ok(());
}
if seq > self.next {
return Err(ControlError::FailedPrecondition(format!(
"seq {seq} arrived where {} was due; resend from {}",
self.next, self.next
)));
}
let stores = self.stores;
let root = stores.root();
// What the box already holds of THIS entry: the same number, kind, length and meta.
let held = match self.kept.take() {
Some((rec, file))
if rec.seq == seq
&& rec.kind == kind
&& rec.blob_len == chunk.blob_len
&& (chunk.meta.is_empty() || chunk.meta == rec.meta) =>
{
Some((rec, file))
}
Some((rec, _)) => {
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
seq, staged_kind = rec.kind, staged_len = rec.blob_len, kind, blob_len = chunk.blob_len,
"the staged restore entry is not the entry now arriving at its number; it is dropped and the entry starts at byte 0"
);
forget(root).map_err(|e| internal(seq, "dropping the staged entry", e))?;
None
}
None => None,
};
let held_bytes = held.as_ref().map(|(rec, _)| rec.durable).unwrap_or(0);
if start > held_bytes {
self.kept = held;
return Err(ControlError::FailedPrecondition(format!(
"seq {seq}: this stream starts it at byte {start}, and this box holds {held_bytes} byte(s) of it; resend seq {seq} from byte {held_bytes}"
)));
}
open.held_before = held_bytes;
open.body = match held {
Some((rec, file)) => {
if open.meta.is_empty() {
open.meta.clone_from(&rec.meta);
}
Body::Spool {
file,
rec,
written: held_bytes,
}
}
None if start == 0
&& (chunk.blob_len <= IN_MEMORY_UP_TO
|| (chunk.last && chunk.blob.len() as u64 == chunk.blob_len)) =>
{
Body::Memory(Vec::with_capacity(gunnar_proto::reserve_for(
chunk.blob_len,
)))
}
None => {
let rec = Staged {
seq,
kind,
blob_len: chunk.blob_len,
meta: open.meta.clone(),
durable: 0,
};
let file =
start_spool(root, &rec).map_err(|e| internal(seq, "staging the entry", e))?;
Body::Spool {
file,
rec,
written: 0,
}
}
};
self.open = Some(open);
Ok(())
}
/// The open entry's last piece arrived: check it, apply it, forget its staging.
fn finish_entry(&mut self, entry_crc: Option<u32>) -> ControlResult<()> {
let stores = self.stores;
let root = stores.root();
let o = self.open.take().expect("an entry is open");
if o.pos != o.blob_len {
return Err(ControlError::InvalidArgument(format!(
"seq {}: {} of {} blob bytes arrived",
o.seq, o.pos, o.blob_len
)));
}
let (blob, staged) = match o.body {
Body::Skip => return Ok(()),
Body::Memory(v) => (v, false),
Body::Spool {
mut file,
mut rec,
written,
} => {
if written != o.blob_len {
return Err(ControlError::InvalidArgument(format!(
"seq {}: {written} of {} blob bytes are staged",
o.seq, o.blob_len
)));
}
// Durable before the apply: an apply that fails leaves the whole entry staged,
// and the next stream finishes it with one empty last piece.
commit(root, &mut file, &mut rec, written)
.map_err(|e| internal(o.seq, "committing the staged entry", e))?;
let mut blob = Vec::with_capacity(gunnar_proto::reserve_for(o.blob_len));
file.seek(SeekFrom::Start(0))
.and_then(|_| file.read_to_end(&mut blob))
.map_err(|e| internal(o.seq, "reading the staged entry back", e))?;
if blob.len() as u64 != o.blob_len {
forget(root).map_err(|e| internal(o.seq, "dropping the staged entry", e))?;
return Err(ControlError::FailedPrecondition(format!(
"seq {}: {} of {} staged bytes read back; they are dropped and seq {} starts again from byte 0",
o.seq,
blob.len(),
o.blob_len,
o.seq
)));
}
(blob, true)
}
};
if let Some(want) = entry_crc {
let mut crc = Crc32c::new();
crc.update(&o.meta);
crc.update(&blob);
let got = crc.finish();
if got != want {
if staged {
forget(root).map_err(|e| internal(o.seq, "dropping the staged entry", e))?;
}
return Err(ControlError::FailedPrecondition(format!(
"seq {}: the {} bytes this box put together do not match the twin's crc ({got:08x}, the twin's {want:08x}); \
nothing was applied, the staged bytes are dropped and seq {} starts again from byte 0",
o.seq, o.blob_len, o.seq
)));
}
}
let e = Entry {
seq: o.seq,
kind: o.kind,
meta: o.meta,
blob,
};
match super::restore_apply(self.stores, &e) {
Ok(Applied::Lost(_)) => self.lost += 1,
Ok(_) => {}
Err(err) => {
return Err(ControlError::Internal(format!(
"restoring seq {}: {err:#}",
e.seq
)))
}
}
if staged {
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, seq = e.seq, blob_len = e.blob.len(), "a restore entry staged across streams is whole, checked and applied");
}
// Applied (durable in the log at its number): the staging goes, durably.
if staged || root.join(PARTIAL_DIR).exists() {
forget(root)
.map_err(|err| internal(e.seq, "removing the staged entry after its apply", err))?;
}
self.applied += 1;
self.next += 1;
Ok(())
}
/// The stream ended. `Err` when inside an entry: what this box holds of it, durably, and
/// where the next stream resumes — or, for a sender that cannot resume, why it never will.
pub fn end_of_stream(&mut self) -> ControlResult<()> {
let stores = self.stores;
let root = stores.root();
// Inside the twin's base: what arrived is made durable, and the box says where it stands.
if self.snap_touched {
if let Some(s) = self.snap.as_mut() {
s.commit().map_err(|err| {
ControlError::Internal(format!(
"committing the twin's snapshot {}: {err:#}",
s.id()
))
})?;
let (items, bytes) = s.position();
return Err(ControlError::FailedPrecondition(format!(
"the stream ended inside the twin's snapshot {}: this box holds {items} item(s) of it and {bytes} byte(s) of the next, \
durably; the next stream resumes there (Bootstrap.Status restore_snapshot_items/bytes, Mirror.SnapshotFetch \
from_item/from_offset) — nothing of it is sent twice",
s.id()
)));
}
}
let Some(mut o) = self.open.take() else {
return Ok(());
};
match &mut o.body {
Body::Skip => Err(ControlError::InvalidArgument(format!(
"the stream ended inside seq {}, which this box holds already; resend from the box's head + 1 ({})",
o.seq, self.next
))),
Body::Memory(_) => Err(ControlError::InvalidArgument(format!(
"the stream ended inside seq {} ({} of {} bytes, small enough to be gathered in memory, and dropped); \
resend from the box's head + 1 ({})",
o.seq, o.pos, o.blob_len, self.next
))),
Body::Spool { file, rec, written } => {
commit(root, file, rec, *written).map_err(|e| internal(o.seq, "committing the staged entry", e))?;
if !o.explicit && o.held_before > 0 && *written <= o.held_before {
return Err(ControlError::FailedPrecondition(format!(
"REFUSED [{TWIN_TOO_OLD}] the twin is too old for a resumed restore; upgrade the twin. Seq {seq} is \
{len} bytes and needs more than one call: this box holds {held} of them, durably, but this stream \
sent the entry from its first byte (no `offset` on its chunks — a twin, or a restore lane, built \
before resumed restores) and ended before reaching byte {held}, so skipping what is held gains \
nothing and every resend ends the same way. With the twin upgraded the next stream resumes seq \
{seq} at byte {held} (Mirror.LogFetch from_offset); nothing here is lost meanwhile",
seq = o.seq,
len = o.blob_len,
held = o.held_before,
)));
}
Err(ControlError::FailedPrecondition(format!(
"the stream ended inside seq {seq}: this box holds {w} of its {len} bytes, durably; the next stream \
resumes seq {seq} at byte {w} (Bootstrap.Status restore_partial_bytes, Mirror.LogFetch from_offset) — \
nothing of it is sent twice",
seq = o.seq,
w = written,
len = o.blob_len,
)))
}
}
}
/// `(applied, lost)` by this stream.
pub fn counts(&self) -> (u64, u64) {
(self.applied, self.lost)
}
}
impl Drop for Intake<'_> {
/// **A stream that never reached [`Intake::end_of_stream`]** — the handler returned early on a
/// broken stream or a refusal, or its future was dropped — still keeps every byte it wrote.
fn drop(&mut self) {
let root = self.stores.root().to_path_buf();
if let Some(Open {
seq,
body: Body::Spool { file, rec, written },
..
}) = self.open.as_mut()
{
if *written > rec.durable {
if let Err(err) = commit(&root, file, rec, *written) {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, seq = *seq, %err, "the staged restore entry could not be committed as its stream ended; the next stream resumes from its last commit");
}
}
}
}
}
impl gunnar_grpc::RestoreIntake for Intake<'_> {
fn take(&mut self, chunk: &gunnar_proto::LogChunk) -> ControlResult<()> {
Intake::take(self, chunk)
}
fn end_of_stream(&mut self) -> ControlResult<()> {
Intake::end_of_stream(self)
}
fn counts(&self) -> (u64, u64) {
Intake::counts(self)
}
}
/// Make `written` bytes durable and count them.
fn commit(root: &Path, file: &mut File, rec: &mut Staged, written: u64) -> anyhow::Result<()> {
file.sync_data().context("fdatasync of the staged entry")?;
if rec.durable != written {
rec.durable = written;
write_record(root, rec)?;
}
Ok(())
}
/// A fresh staging area for `rec` (anything there before is not this entry).
fn start_spool(root: &Path, rec: &Staged) -> anyhow::Result<File> {
forget(root)?;
write_record(root, rec)?;
let path = root.join(PARTIAL_DIR).join(DATA);
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(true)
.open(&path)
.with_context(|| format!("creating {}", path.display()))?;
// The data file's directory entry, durable before any count refers to its bytes.
crate::durable::sync_dir(&root.join(PARTIAL_DIR))
.with_context(|| format!("flushing {} into place", path.display()))?;
Ok(file)
}
fn internal(seq: u64, what: &str, e: impl Into<anyhow::Error>) -> ControlError {
ControlError::Internal(format!("seq {seq}: {what}: {:#}", e.into()))
}