gunnar-server 2.0.0

//! **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()))
}