crates/gunnar-cli/src/control/twin.rs
//! `gunnar control twin follow | restore | status` — the twin's verbs.
//! `.nornir/twin-transaction-log.md`.
//!
//! `follow`, `restore` and `catch-up` need the server half (the stores the log
//! is applied into) and are therefore only real on a build with `server`; a
//! `control`-only build says so by name. `status` is one `Mirror.LogHead`.
use std::path::Path;
use anyhow::Result;
use gunnar_proto::mirror_client::MirrorClient;
use super::{authed, connect, explain, private_key, runtime};
use crate::cli::ControlTwinCommand;
pub(crate) fn run(command: ControlTwinCommand) -> Result<()> {
match command {
ControlTwinCommand::Status { at, identity } => status(&at, identity.as_deref()),
ControlTwinCommand::Follow {
at,
data_root,
identity,
poll_secs,
once,
} => follow(&at, &data_root, identity.as_deref(), poll_secs, once),
ControlTwinCommand::Restore { from, into } => restore(&from, &into),
ControlTwinCommand::Join {
at,
primary,
identity,
} => join(&at, &primary, identity.as_deref()),
ControlTwinCommand::Certify {
at,
primary_at,
principal,
identity,
} => certify(&at, &primary_at, &principal, identity.as_deref()),
ControlTwinCommand::Replay {
twin,
into,
identity,
ignore_room,
stale_ok,
accept_lost,
max_silence,
primary_fenced,
} => replay(
&twin,
&into,
identity.as_deref(),
gunnar_repair::RechargeOpts {
accept_lost,
stale_ok,
max_silence_ms: max_silence.as_millis() as u64,
override_live_main: false,
ignore_room,
// A primary as the source has always been allowed here, deliberately.
allow_primary_source: true,
primary_fenced,
// Replay writes into a fresh `--into` dir, never over a live main.
in_place: false,
},
),
ControlTwinCommand::Retain {
at,
from,
allow_past_twin,
identity,
} => retain(&at, from, allow_past_twin, identity.as_deref()),
ControlTwinCommand::CatchUp {
at,
data_root,
identity,
stall_secs,
attempts,
dry_run,
} => catch_up(
&at,
&data_root,
identity.as_deref(),
stall_secs,
attempts,
dry_run,
),
}
}
fn retain(at: &str, from: u64, allow_past_twin: bool, identity: Option<&Path>) -> Result<()> {
let signer = private_key(identity)?;
let rt = runtime()?;
let mut client = MirrorClient::new(connect(&rt, at)?);
let head = rt
.block_on(client.log_retain(authed(
&signer,
gunnar_proto::LogRetainRequest {
from,
allow_past_twin,
},
)?))
.map_err(|status| anyhow::anyhow!("{}", explain(&status)))?
.into_inner();
println!("head_seq {}", head.head_seq);
println!("first_seq {}", head.first_seq);
println!("bytes {}", head.bytes);
Ok(())
}
fn status(at: &str, identity: Option<&Path>) -> Result<()> {
let signer = private_key(identity)?;
let rt = runtime()?;
let channel = connect(&rt, at)?;
let mut client = MirrorClient::new(channel);
let head = rt
.block_on(client.log_head(authed(&signer, gunnar_proto::LogHeadRequest::default())?))
.map_err(|status| anyhow::anyhow!("{}", explain(&status)))?
.into_inner();
println!("head_seq {}", head.head_seq);
println!("first_seq {}", head.first_seq);
println!("bytes {}", head.bytes);
if head.twin {
println!("twin yes");
println!("applied {}", head.applied_seq);
println!("primary {}", head.primary_head_seq);
println!("caught_up {}", head.caught_up);
println!("state {}", head.state);
// **The age, beside the numbers** (2026-09-21, lane T4). `caught_up` is
// a statement about a MOMENT, and until this line printed there was no
// way to ask which moment. A twin that stopped following ten days ago
// printed exactly what it printed ten days ago.
println!("as_of {}", as_of(&head));
if !head.why.is_empty() {
println!("why {}", head.why);
}
} else {
println!("twin no");
}
Ok(())
}
/// **The restore guard** lives in `gunnar-repair` (5.4 resilience, 2026-10-10): the ONE
/// implementation behind this verb, the admin console's button, the boot check and a pulled
/// monetize order. Its T3/T17 tests moved with it.
#[cfg(test)]
use gunnar_repair::{DEFAULT_MAX_SILENCE_MS as MAX_SILENCE_MS, FOLLOWER_NOT_RUNNING};
/// How current this twin is, in words — both numbers and both ages ([`gunnar_repair::as_of`]).
fn as_of(head: &gunnar_proto::LogHeadResponse) -> String {
gunnar_repair::as_of(&gunnar_repair::AsTwinHead::twin_head(head))
}
fn join(at: &str, primary: &str, identity: Option<&Path>) -> Result<()> {
// The fresh box takes this only from a key it was fed (h-bootstrap-open).
let signer = private_key(identity)?;
let rt = runtime()?;
let channel = connect(&rt, at)?;
let mut client = gunnar_proto::bootstrap_client::BootstrapClient::new(channel);
let answer = rt
.block_on(client.become_twin(authed(
&signer,
gunnar_proto::BecomeTwinRequest {
primary: primary.to_string(),
},
)?))
.map_err(|status| anyhow::anyhow!("{}", explain(&status)))?
.into_inner();
println!("{}", answer.twin_public_key);
eprintln!("{}", answer.detail);
if answer.restart_required {
eprintln!("RESTART REQUIRED: this box's git listener is still bound where a primary's was. It serves no git — every \
request is refused by name — but the socket only moves to loopback when the box comes back.");
}
Ok(())
}
/// **`twin certify`**: the courier for the twin's mutual-TLS credential.
///
/// Three calls, and the private key crosses none of them: the twin mints it and
/// answers a request (`Bootstrap.TwinCsr`), the primary's issuer signs and
/// registers it (`Admin.IssueTwinCertificate`), the twin checks it is for its own
/// key and pins the issuer (`Bootstrap.InstallTwinCertificate`). Idempotent end to
/// end, so a courier that dies half-way is simply run again.
fn certify(at: &str, primary_at: &str, principal: &str, identity: Option<&Path>) -> Result<()> {
let signer = private_key(identity)?;
let rt = runtime()?;
let mut twin = gunnar_proto::bootstrap_client::BootstrapClient::new(connect(&rt, at)?);
let csr_pem = rt
.block_on(twin.twin_csr(gunnar_proto::TwinCsrRequest {}))
.map_err(|status| {
anyhow::anyhow!(
"the twin at {at} would not give a signing request: {}",
explain(&status)
)
})?
.into_inner()
.csr_pem;
let mut primary = gunnar_proto::admin_client::AdminClient::new(connect(&rt, primary_at)?);
let issued = rt
.block_on(primary.issue_twin_certificate(authed(
&signer,
gunnar_proto::IssueTwinCertificateRequest {
principal: principal.to_string(),
csr_pem,
},
)?))
.map_err(|status| {
anyhow::anyhow!(
"the primary at {primary_at} would not certify {principal}: {}",
explain(&status)
)
})?
.into_inner();
let installed = rt
.block_on(twin.install_twin_certificate(authed(
&signer,
gunnar_proto::InstallTwinCertificateRequest {
cert_pem: issued.cert_pem,
ca_pem: issued.issuer_pem,
replication_port: issued.replication_port,
replication_host: issued.replication_host.clone(),
},
)?))
.map_err(|status| {
anyhow::anyhow!(
"the twin at {at} refused its certificate: {}",
explain(&status)
)
})?
.into_inner();
println!("paired {principal} for mutual TLS 1.3");
if issued.replication_host.is_empty() {
println!(
"replication port {} on the primary",
issued.replication_port
);
} else {
// The primary's listener is on its TUNNEL address: the twin now dials
// there, not the host it was joined to.
println!(
"replication {}:{} on the primary (its tunnel address)",
issued.replication_host, issued.replication_port
);
}
println!("days_left {}", installed.days_left);
eprintln!("{}", installed.detail);
Ok(())
}
/// **`twin replay`** — a thin caller of [`gunnar_repair::pump::recharge_from_twin`]: the same
/// refusals, the same pump (resumable base, poisoned body on a broken fetch), the same receipt as
/// the console's [Recharge from twin] and a pulled order. The fresh box takes the stream only
/// under a credential from a key it was FED (`--restore-pubkey`, or the installer's admin keys)
/// — the SAME key that reads the twin.
fn replay(
twin: &str,
into: &str,
identity: Option<&Path>,
opts: gunnar_repair::RechargeOpts,
) -> Result<()> {
let signer = private_key(identity)?;
let rt = runtime()?;
if opts.ignore_room {
eprintln!("--ignore-room: the box is not told how much is coming, so it cannot refuse for room; a volume that fills fails the apply mid-stream and the box stays restoring");
}
let source = gunnar_repair::proto::ProtoTwin {
addr: twin.to_owned(),
channel: connect(&rt, twin)?,
key: signer.clone(),
};
let target = gunnar_repair::proto::ProtoMain {
addr: into.to_owned(),
channel: connect(&rt, into)?,
key: signer,
};
let receipt = rt.block_on(replay_into(&source, &target, &opts))?;
println!("head_seq {}", receipt.head_seq);
println!("applied {}", receipt.applied);
println!("lost {}", receipt.lost);
println!("restart {}", receipt.needs_restart);
println!("{}", receipt.detail);
Ok(())
}
/// The verb, as `anyhow`: a refusal by name, the freshness line (said also when accepted: an
/// operator restoring the forge sees how current the copy is either way), and a failed pump as
/// an error that says the box was left restoring and the replay resumes.
async fn replay_into<T, M>(
twin: &T,
main: &M,
opts: &gunnar_repair::RechargeOpts,
) -> Result<gunnar_repair::Receipt>
where
T: gunnar_repair::pump::Twin,
M: gunnar_repair::pump::Main<T::Chunk>,
{
let receipt = gunnar_repair::recharge_from_twin(twin, main, opts, &|| false)
.await
.map_err(|refusal| anyhow::anyhow!("{}: {}", refusal.code.as_str(), refusal.message))?;
if !receipt.freshness.is_empty() {
eprintln!("{}", receipt.freshness);
}
eprintln!("{}", receipt.plan);
match receipt.phase {
gunnar_repair::Phase::Done => Ok(receipt),
gunnar_repair::Phase::Replaying => anyhow::bail!(
"{}; the fresh box was left restoring — run the replay again",
receipt.detail
),
gunnar_repair::Phase::Failed => anyhow::bail!(
"{}; the fresh box was left restoring — run the replay again",
receipt.why
),
}
}
/// A stream of the twin's chunks, as `Mirror.LogFetch` / `Mirror.SnapshotFetch` answer
/// (`catch-up` drains it).
#[cfg(feature = "server")]
type Chunks = std::pin::Pin<
Box<dyn tokio_stream::Stream<Item = Result<gunnar_proto::LogChunk, tonic::Status>> + Send>,
>;
#[cfg(feature = "server")]
mod with_server {
use std::path::Path;
use std::sync::atomic::AtomicBool;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use gunnar_proto::mirror_client::MirrorClient;
use gunnar_server::twin::{snapshot, Follower, LogSource, Progress};
use gunnar_server::ServerStores;
use crate::control::{authed, connect, explain, private_key, runtime, ControlChannel};
/// `Mirror.LogHead` / `Mirror.LogFetch` over the control plane, as the
/// follower reads them. One runtime, one channel, one signer.
struct Grpc {
rt: tokio::runtime::Runtime,
client: Mutex<MirrorClient<ControlChannel>>,
signer: ssh_key::PrivateKey,
/// The data root whose head and volume ride on each poll (the dual heartbeat).
stores: Arc<ServerStores>,
}
impl LogSource for Grpc {
fn head(&self) -> Result<(u64, u64)> {
let (first, head, _) = self.head_and_identity()?;
Ok((first, head))
}
/// **`gunnar control twin follow` gets the identity check too**
/// (2026-09-21, lane T4). This verb writes into a real data root exactly
/// as the server's own follower does, so it must be just as unable to
/// splice one history onto another: `Follower::step` refuses a source
/// whose log identity is not the one this root holds.
///
/// It does NOT yet verify the primary's SIGNATURE — that lives in
/// `GrpcSource::check_answer`, on the server's follower. This verb is a
/// nonce-less caller, so the answer comes back unsigned by design.
fn head_and_identity(&self) -> Result<(u64, u64, Option<String>)> {
let mut client = self.client.lock().expect("twin client").clone();
let head = self
.rt
.block_on(client.log_head(authed(
&self.signer,
gunnar_proto::LogHeadRequest {
report: Some(gunnar_server::twin::measure_report(&self.stores)),
nonce: String::new(),
generation_proof: None,
},
)?))
.map_err(|status| anyhow::anyhow!("{}", explain(&status)))?
.into_inner();
Ok((
head.first_seq,
head.head_seq,
Some(head.log_id).filter(|s| !s.trim().is_empty()),
))
}
fn fetch(
&self,
from: u64,
to: u64,
max_bytes: u64,
) -> Result<Vec<gunnar_core::txlog::Entry>> {
let mut whole = gunnar_server::twin::Gather::default();
self.fetch_into(from, to, max_bytes, &mut whole)?;
Ok(whole.out)
}
/// **Streamed into this data root's log as it arrives** (twin-streaming, 2026-10-06),
/// through the server follower's own reader of the `LogChunk` protocol: a 512 MiB pack
/// is never whole in this process.
fn fetch_into(
&self,
from: u64,
to: u64,
max_bytes: u64,
into: &mut dyn gunnar_server::twin::FrameSink,
) -> Result<()> {
let mut client = self.client.lock().expect("twin client").clone();
let request = authed(
&self.signer,
gunnar_proto::LogFetchRequest {
from,
to,
max_bytes,
from_offset: 0,
},
)?;
self.rt.block_on(async {
let mut stream = client
.log_fetch(request)
.await
.map_err(|status| anyhow::anyhow!("{}", explain(&status)))?
.into_inner();
gunnar_server::twin::feed_log_stream(
&mut stream,
None,
&|status| anyhow::anyhow!("{}", explain(&status)),
into,
)
.await
})
}
}
pub(super) fn follow(
at: &str,
data_root: &Path,
identity: Option<&Path>,
poll_secs: u64,
once: bool,
) -> Result<()> {
let signer = private_key(identity)?;
let rt = runtime()?;
let channel = connect(&rt, at)?;
let stores =
Arc::new(ServerStores::open(data_root).with_context(|| {
format!("opening the twin's data root {}", data_root.display())
})?);
let source = Grpc {
rt,
client: Mutex::new(MirrorClient::new(channel)),
signer,
stores: Arc::clone(&stores),
};
let follower = Follower::new(stores, Box::new(source));
if once {
match follower.step()? {
Progress::Applied { count, head } => println!("applied {count}; head {head}"),
Progress::CaughtUp { head } => println!("caught up at {head}"),
Progress::SnapshotInstalled { at } => println!("installed a snapshot at {at}; following resumes after it"),
Progress::SnapshotNeeded { wanted, first } => {
anyhow::bail!("snapshot needed: this twin needs {wanted}, the primary retains from {first}")
}
Progress::Diverged { own, primary_head } => anyhow::bail!(
"diverged: the box at --at answers head {primary_head}, BELOW the {own} this data root has \
already applied — a log's head never goes backwards, so --at is not the primary this copy \
was following"
),
}
return Ok(());
}
let stop = AtomicBool::new(false);
follower.run(Duration::from_secs(poll_secs.max(1)), &stop);
Ok(())
}
pub(super) fn restore(from: &Path, into: &Path) -> Result<()> {
let log_dir = from.join(gunnar_server::stores::TXLOG_DIR);
let log = Arc::new(
gunnar_core::txlog::Log::open(&log_dir)
.with_context(|| format!("opening the twin's log {}", log_dir.display()))?,
);
let stores = Arc::new(
ServerStores::open(into)
.with_context(|| format!("opening the fresh data root {}", into.display()))?,
);
// The identity of the log being restored, off the SOURCE data root: the
// box this builds IS that history, so its twin keeps following it. See
// `gunnar_server::twin::LOG_ID`.
let source_log_id = gunnar_server::twin::log_id(from).unwrap_or_default();
let head = gunnar_server::twin::restore(log, stores, &source_log_id)?;
println!("restored up to {head} into {}", into.display());
Ok(())
}
// ───────────────────────────────────────────────────────────────────────
// `gunnar control twin catch-up` — a twin behind a retention hole
// ───────────────────────────────────────────────────────────────────────
//
// **The gap this closes is a VERB, not a bench.** `Mirror.SnapshotFetch`
// (`crates/gunnar-proto/proto/gunnar/v1/mirror.proto:153`) and
// `Mirror.LogRetain` (:159) have shipped since 2026-09-15 (Lanes K and Q)
// and the mechanism is proven in `gunnar-server`
// (`twin::tests::a_twin_behind_a_retention_hole_catches_up_via_snapshot_and_then_follows`,
// `crates/gunnar-server/src/twin/tests.rs:1164`). What did not exist until
// now is anything an OPERATOR can type that drives them: the snapshot
// happened, if it happened at all, inside the twin's own background poll
// (`gunnar_server::twin::Follower::step`,
// `crates/gunnar-server/src/twin.rs:1073`), which
//
// * takes one SILENTLY whenever `first > applied + 1` — the operator
// learns that the primary's write path was stopped, if ever, from a log
// line after the fact;
// * has no deadline on a stream that stops arriving, so a primary that
// goes quiet mid-snapshot holds its own `Log::freeze()` until ITS guard
// fires (`SNAPSHOT_STALL_TIMEOUT`, 120 s,
// `crates/gunnar-grpc/src/service/mirror.rs:502`) while the operator
// watches a command that looks hung;
// * cannot be asked "is there a hole?" without also being told "and I
// have fixed it", so there is no way to establish the situation before
// paying for it.
//
// **Why the freeze is the dangerous part.** The producer holds
// `Log::freeze()` for the whole of `SnapshotFetch`
// (`crates/gunnar-grpc/src/service/mirror.rs:117-135`, whose own comment
// reads "while a snapshot streams EVERY framed write on the primary is
// stopped"), and `.nornir/twin-transaction-log.md` §Snapshot says it from
// the other side: the freeze stops new mutations and waits for the ones in
// flight, which is exactly what makes `at` a consistent sequence. Reads are
// unaffected; every push, every ref update and every control-plane write is
// not. On a forge that is serving, a snapshot is an outage of the write path
// for its whole duration — and an operator who is not told the number
// cannot decide to run it at 03:00 instead. So this verb reports it.
/// **Why the catch-up is allowed to run.** There are exactly two reasons;
/// everything else is a refusal by name out of [`hole_verdict`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) enum Verdict {
/// The primary's log starts ABOVE the sequence this twin needs next, so
/// the range is gone for good and only a snapshot crosses it
/// (`mirror.proto:143`: "`from` below `first_seq` is
/// FAILED_PRECONDITION naming `first_seq`: the twin needs a snapshot,
/// not a range").
Hole { wanted: u64, first: u64, head: u64 },
/// `txlog-snapshot-installing` is on this twin's disk: it died inside an
/// install and its stores are half a snapshot. Only a WHOLE one replaces
/// that — measured 2026-09-16 (`twin.rs:1112`): a twin that followed on
/// from a half install applied 29 entries onto stores the install had
/// already wiped, and said nothing.
HalfInstalled { wanted: u64, first: u64, head: u64 },
}
impl Verdict {
/// The one line that says what was established, printed BEFORE anything
/// freezes.
fn line(&self, applied: u64) -> String {
match self {
Verdict::Hole { wanted, first, head } => format!(
"hole the primary retains from {first} and has {head}; this twin holds \
{applied} and needs {wanted} — {} sequence(s) are GONE, so no LogFetch can ever \
answer for them again",
first - wanted
),
Verdict::HalfInstalled { wanted, first, head } => format!(
"half install this twin died inside a snapshot install, so its stores are half a \
snapshot and only a whole one replaces them (the primary retains {first}..={head}; \
this twin needs {wanted})"
),
}
}
}
/// **Is there a hole, and may a snapshot be taken to close it?**
///
/// `applied` is the twin's own head (`stores.txlog().head_seq()`), `head` is
/// the primary's answer to `Mirror.LogHead`, `installing` is
/// `snapshot::installing(<the twin's data root>)`.
///
/// Every refusal starts with a kebab-case NAME, so the operator reading a
/// terminal and a script matching on the first word are reading the same
/// fact. The one that carries the design is `no-hole`: a twin that is merely
/// BEHIND must use `gunnar control twin follow`, which reads the range the
/// primary still holds and freezes nothing. Stopping a forge's whole write
/// path to hand a twin sequences the log still has is not a slower way of
/// doing the same thing — it is a different and worse thing, so it is
/// refused rather than offered.
pub(super) fn hole_verdict(
applied: u64,
head: &gunnar_proto::LogHeadResponse,
installing: bool,
) -> Result<Verdict> {
let wanted = applied + 1;
// A twin answers `LogHead` with `twin = true` (`mirror.proto:196`). A
// snapshot OF a twin is a real thing — `control twin replay` takes one —
// but it is not what closes THIS twin's hole, and pointing `--at` at the
// wrong box is the easy mistake to make: both are gunnar control planes
// on :50051 and both answer this RPC.
if head.twin {
anyhow::bail!(
"not-a-primary: the box at --at answers LogHead with twin=true (it has applied {}, \
state {:?}). --at is the PRIMARY this twin follows — not another twin, and not this \
box",
head.applied_seq,
head.state
);
}
if installing {
return Ok(Verdict::HalfInstalled {
wanted,
first: head.first_seq,
head: head.head_seq,
});
}
if head.head_seq == 0 {
anyhow::bail!(
"empty-log: the primary's head_seq is 0, so its transaction log is empty — there is no \
state to snapshot and nothing for this twin to be behind"
);
}
if wanted > head.head_seq {
anyhow::bail!(
"already-level: this twin holds {applied} and the primary's head is {}, so it is level \
or ahead. There is no hole and nothing to fetch",
head.head_seq
);
}
if head.first_seq <= wanted {
anyhow::bail!(
"no-hole: the primary retains from {} and this twin needs {wanted} next, so every \
sequence it is missing is STILL ON THE PRIMARY. Fetch them with\n \
gunnar control twin follow --at <primary> --data-root <this root>\n\
which reads the range and freezes nothing. A snapshot would stop every framed write \
on the primary — every push, every ref update, every control-plane write — for the \
whole of the copy, to hand over a range the log still holds. This verb refuses rather \
than impose that",
head.first_seq
);
}
Ok(Verdict::Hole {
wanted,
first: head.first_seq,
head: head.head_seq,
})
}
/// **What one `Mirror.SnapshotFetch` stream delivered**, so the report can
/// say how much moved for the freeze it cost.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(super) struct Drained {
pieces: u64,
bytes: u64,
}
/// The first word of the refusal a stalled stream fails with. Named once, so
/// the retry loop can tell a stall (never retried) from a break, and so a
/// test asserts on the word rather than on a sentence.
pub(super) const STALL: &str = "stalled";
/// **Take the snapshot stream piece by piece, with a deadline on NO
/// PROGRESS.**
///
/// The mirror image of the primary's own `send_piece`
/// (`crates/gunnar-grpc/src/service/mirror.rs:506`), which cuts off a twin
/// that stops READING after 120 s. This is the half that cuts off a PRIMARY
/// that stops SENDING, and it exists for the same reason that one does:
/// `stream.next()` against a peer that has gone away — panicked, lost its
/// cable, black-holed its socket — waits until TCP gives up, which is
/// minutes to hours, and the primary's write path is stopped for every
/// second of it.
///
/// The deadline is on no progress and never on the transfer: a primary that
/// keeps sending pieces is never cut off, however slow, because a 44 GiB
/// snapshot legitimately takes a long time.
///
/// **Dropping `stream` is what releases the freeze**, so the stall arm drops
/// it BEFORE it writes the sentence that says so: the client's RST_STREAM
/// makes the producer's `tx.send` fail ("the twin hung up mid-snapshot",
/// `mirror.rs:512`) and the blocking thread returns, which drops the
/// `Log::freeze()` guard it was holding. Without that, a verb that gave up
/// and kept the connection open would have given up on nothing at all.
pub(super) async fn drain(
stream: super::Chunks,
stall: Duration,
sink: &mut dyn FnMut(&gunnar_proto::LogChunk) -> Result<()>,
) -> Result<Drained> {
use tokio_stream::StreamExt as _;
let mut stream = stream;
let mut out = Drained::default();
let opened = Instant::now();
loop {
// The timeout future borrows `stream`. It is bound to a `let` rather
// than matched in place so that the borrow has ended by the time the
// stall arm drops the stream.
let next = tokio::time::timeout(stall, stream.next()).await;
let next = match next {
Ok(next) => next,
Err(_) => {
let waited = opened.elapsed();
drop(stream);
anyhow::bail!(
"{STALL}: the primary sent no snapshot piece for {} s ({} piece(s), {} in \
before it went quiet). Mirror.SnapshotFetch holds Log::freeze() for the \
whole stream, so the primary had taken NO framed write for {:.1} s when \
this gave up — every push, ref update and control-plane write on it was \
waiting. The stream is dropped here, which is what releases that freeze; \
the primary's own guard (SNAPSHOT_STALL_TIMEOUT, 120 s) would otherwise \
have been the only thing that ever did. Nothing was installed on this twin \
and its log is untouched",
stall.as_secs_f64(),
out.pieces,
crate::control::human_bytes(out.bytes),
waited.as_secs_f64()
);
}
};
let Some(chunk) = next else {
return Ok(out);
};
let chunk = chunk.map_err(|status| {
anyhow::anyhow!(
"the primary's snapshot stream broke after {} piece(s), {} in: {}",
out.pieces,
crate::control::human_bytes(out.bytes),
explain(&status)
)
})?;
out.pieces += 1;
out.bytes += chunk.blob.len() as u64;
sink(&chunk)?;
}
}
/// **What the catch-up did**, as the verb prints it — a function of the
/// answer rather than a pile of `println!`s, so a test reads the same lines
/// an operator does. The shape `heartbeat_lines` uses in `control.rs:5872`.
#[derive(Debug)]
pub(super) struct Report {
/// What was established before anything froze.
verdict: Verdict,
/// The twin's own head when it started.
was: u64,
/// False for `--dry-run`: nothing was streamed and nothing froze.
streamed: bool,
/// The sequence the snapshot's state was taken at (its `SnapshotAt`).
at: u64,
pieces: u64,
bytes: u64,
/// How many times the stream had to be taken FROM THE BEGINNING.
attempts: u32,
/// **How long the primary's framed writes were stopped**, summed over
/// every attempt: from the moment each `SnapshotFetch` was opened until
/// its stream ended or was dropped.
froze: Duration,
/// **Where the twin's OWN log now starts.** The proof that a SNAPSHOT
/// closed the hole rather than a fetch that happened to work: only
/// `Log::restart_after`, which nothing but `snapshot::install` calls,
/// moves a twin's `first_seq` off 1 (`.nornir/twin-transaction-log.md`,
/// row R8).
first_after: u64,
/// The twin's head after the install and the follow that followed it.
applied: u64,
/// The primary's head at the last poll.
primary_head: u64,
/// The follower's state word. `following` is the one that means it worked.
state: String,
caught_up: bool,
}
impl Report {
pub(super) fn lines(&self) -> Vec<String> {
let mut out = vec![self.verdict.line(self.was)];
if !self.streamed {
out.push(
"dry run nothing was streamed and the primary was NEVER frozen; run it \
again without --dry-run to take the snapshot"
.to_owned(),
);
return out;
}
out.push(format!(
"snapshot taken at {} · {} piece(s) · {} · {} attempt(s)",
self.at,
self.pieces,
crate::control::human_bytes(self.bytes),
self.attempts
));
// Upper case, the way `loud()` (`control.rs:5834`) makes the line
// that is not "ok" the loud one in the block. This is the line that
// cost somebody else something.
out.push(format!(
"FROZEN {:.1} s — the primary took NO framed write for that long (every push, \
ref update and control-plane write on it waited); Mirror.SnapshotFetch holds \
Log::freeze() for the whole stream",
self.froze.as_secs_f64()
));
out.push(format!(
"resumed at this twin's own log now starts at {} — only a snapshot install moves \
that off 1",
self.first_after
));
out.push(format!("applied {}", self.applied));
out.push(format!("primary head {}", self.primary_head));
out.push(format!("state {}", self.state));
out.push(format!(
"following {}",
if self.caught_up {
"YES — the hole is closed and this twin is level with its primary".to_owned()
} else {
format!(
"NO — {} at {} of {}",
self.state, self.applied, self.primary_head
)
}
));
out
}
}
/// What the catch-up reads from the primary: the WHOLE `LogHead` answer (the
/// hole is established out of `first_seq`/`head_seq`, and `twin` catches a
/// misaimed `--at`), and `Mirror.SnapshotFetch`.
///
/// It is a `LogSource` as well, because the same connection that hands over
/// the snapshot is the one that must then prove the twin FOLLOWS again — a
/// snapshot installed and never followed on from is not a caught-up twin,
/// and B8's row is "snapshot catch-up → follows again".
pub(super) trait Primary: LogSource {
fn log_head(&self) -> Result<gunnar_proto::LogHeadResponse>;
/// Open `Mirror.SnapshotFetch` and feed every piece to `sink`, giving up
/// by name when none arrives for `stall`. **The primary's framed writes
/// are stopped from the moment the stream opens until it ends or is
/// dropped**, so the caller times exactly this call.
fn stream_snapshot(
&self,
stall: Duration,
sink: &mut dyn FnMut(&gunnar_proto::LogChunk) -> Result<()>,
) -> Result<Drained>;
}
impl Primary for Grpc {
fn log_head(&self) -> Result<gunnar_proto::LogHeadResponse> {
let mut client = self.client.lock().expect("twin client").clone();
// The dual heartbeat rides along, exactly as the follower's own poll
// sends it: a catch-up is a poll, and a primary whose console shows
// this twin should see it move.
let request = authed(
&self.signer,
gunnar_proto::LogHeadRequest {
report: Some(gunnar_server::twin::measure_report(&self.stores)),
nonce: String::new(),
generation_proof: None,
},
)?;
Ok(self
.rt
.block_on(client.log_head(request))
.map_err(|status| anyhow::anyhow!("{}", explain(&status)))?
.into_inner())
}
fn stream_snapshot(
&self,
stall: Duration,
sink: &mut dyn FnMut(&gunnar_proto::LogChunk) -> Result<()>,
) -> Result<Drained> {
let mut client = self.client.lock().expect("twin client").clone();
let request = authed(&self.signer, gunnar_proto::SnapshotFetchRequest::default())?;
self.rt.block_on(async {
let stream = client
.snapshot_fetch(request)
.await
.map_err(|status| anyhow::anyhow!("{}", explain(&status)))?
.into_inner();
drain(Box::pin(stream), stall, sink).await
})
}
}
/// The primary, shared between the snapshot drive and the follow-again that
/// proves it worked (`Follower::new` takes its source by value).
///
/// **`LogSource::snapshot` is deliberately left at its default `false`.**
/// This verb takes the ONE snapshot it was asked for, explicitly, and
/// reports the freeze it cost; a follow-again that could quietly take a
/// second one would be exactly the silent freeze the verb exists to remove.
/// So a hole that opens WHILE the catch-up runs (retention moving again
/// under it) is refused by name instead — see `hole-again` below.
struct Shared<P>(Arc<P>);
impl<P: Primary> LogSource for Shared<P> {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.0.head()
}
fn fetch(
&self,
from: u64,
to: u64,
max_bytes: u64,
) -> anyhow::Result<Vec<gunnar_core::txlog::Entry>> {
self.0.fetch(from, to, max_bytes)
}
}
/// One wire chunk as one staging piece. The same three checks
/// `gunnar_server::twin::GrpcSource::snapshot` makes (`twin.rs:1306-1310`):
/// every snapshot item carries `seq` 0 (`mirror.proto:151`), and a kind that
/// does not fit a `u16` is a primary this build cannot read rather than a
/// silent truncation.
fn feed(stager: &mut snapshot::Stager, chunk: &gunnar_proto::LogChunk) -> Result<()> {
anyhow::ensure!(
chunk.seq == 0,
"the primary's snapshot stream carried seq {} — every snapshot item has seq 0",
chunk.seq
);
let kind = u16::try_from(chunk.kind).map_err(|_| {
anyhow::anyhow!(
"the primary's snapshot carried a piece of kind {}",
chunk.kind
)
})?;
stager.feed(snapshot::Piece {
kind,
meta: &chunk.meta,
blob_len: chunk.blob_len,
blob: &chunk.blob,
last: chunk.last,
})
}
/// How many polls the twin is given to go from "snapshot installed" to
/// "level with the primary" before the verb reports what it reached. One
/// poll applies everything the primary had at that moment
/// (`Follower::step`'s inner loop runs to `primary_head`), so more than one
/// is only needed when the primary is still taking writes — and a bound is
/// what stops this verb becoming an unbounded `follow` on a busy forge.
const FOLLOW_POLLS: u32 = 8;
/// **The catch-up itself**, over any [`Primary`]: establish, drive, verify.
///
/// Written against the trait and not against the gRPC client so the tests
/// drive the REAL staging, the REAL `snapshot::install` and the REAL
/// `Follower` against a real primary's stores — the same shape
/// `replay_tests` uses for `control twin replay` below.
pub(super) fn run_catch_up<P: Primary + 'static>(
stores: Arc<ServerStores>,
primary: P,
stall: Duration,
attempts: u32,
dry_run: bool,
) -> Result<Report> {
// ── 1. establish: is there a hole at all? ───────────────────────────
let head = primary
.log_head()
.context("asking the primary for its log head")?;
let was = stores.txlog().head_seq();
let verdict = hole_verdict(was, &head, snapshot::installing(stores.root()))?;
if dry_run {
return Ok(Report {
verdict,
was,
streamed: false,
at: 0,
pieces: 0,
bytes: 0,
attempts: 0,
froze: Duration::ZERO,
first_after: stores.txlog().first_seq(),
applied: was,
primary_head: head.head_seq,
state: "unchanged".to_owned(),
caught_up: false,
});
}
// Progress on stderr, the finished report on stdout — `replay` above
// splits them the same way. The operator must see this BEFORE the wait,
// not in a summary afterwards.
eprintln!("{}", verdict.line(was));
eprintln!(
"opening Mirror.SnapshotFetch on the primary — ITS FRAMED WRITES STOP NOW and resume when \
the stream ends; no piece for {} s gives up by name and releases them",
stall.as_secs()
);
// ── 2. drive the snapshot, resuming a stream that breaks ────────────
let mut froze = Duration::ZERO;
let mut tries: u32 = 0;
let (drained, staged) = loop {
tries += 1;
// `Stager::begin` clears whatever an earlier attempt left, which is
// also the resume: there is no cursor to resume FROM.
let mut stager = snapshot::Stager::begin(stores.root())
.with_context(|| format!("staging a snapshot under {}", stores.root().display()))?;
let opened = Instant::now();
let got = {
let mut sink = |chunk: &gunnar_proto::LogChunk| feed(&mut stager, chunk);
primary.stream_snapshot(stall, &mut sink)
};
// The freeze lasted exactly as long as the stream was open, whether
// it ended, broke or was dropped.
froze += opened.elapsed();
match got.and_then(|drained| stager.finish().map(|staged| (drained, staged))) {
Ok(pair) => break pair,
Err(err) => {
// A stall is NEVER retried. Taking the stream again would
// freeze the primary from the start for a peer that has just
// proven it is not sending, which is the outage this verb is
// built to bound.
if format!("{err:#}").contains(STALL) {
return Err(err.context(format!(
"the catch-up gave up on attempt {tries} of {attempts} and did NOT take the \
stream again: a stall is not retried, because re-opening SnapshotFetch \
would freeze the primary from the start for a peer that has just stopped \
sending. The primary is frozen no longer"
)));
}
if tries >= attempts {
return Err(err.context(format!(
"the snapshot stream broke on all {attempts} attempt(s), {:.1} s of the \
primary's writes spent between them. Nothing was installed and this twin's \
log is untouched, so the catch-up is safe to run again",
froze.as_secs_f64()
)));
}
eprintln!("attempt {tries} of {attempts} broke: {err:#}");
eprintln!(
" SnapshotFetchRequest carries no cursor (mirror.proto:165), so the only resume \
there is is the WHOLE stream again — which freezes the primary from the start \
once more; {:.1} s of freeze spent so far",
froze.as_secs_f64()
);
}
}
};
// ── 3. install it whole, and follow again ───────────────────────────
let at = staged.at;
let (pieces, bytes) = (drained.pieces, drained.bytes);
snapshot::install(&stores, &staged)
.with_context(|| format!("installing the snapshot the primary took at {at}"))?;
drop(staged);
eprintln!(
"installed the snapshot taken at {at}; the primary's writes resumed after {:.1} s",
froze.as_secs_f64()
);
let primary = Arc::new(primary);
let follower = Follower::new(Arc::clone(&stores), Box::new(Shared(Arc::clone(&primary))));
let mut polls: u32 = 0;
let state = loop {
polls += 1;
match follower.step().context("following the primary again, after the snapshot")? {
Progress::CaughtUp { .. } => break "following",
Progress::Applied { count, head } => {
eprintln!("applied {count} entr(ies) the primary took while the snapshot streamed; head {head}");
if polls >= FOLLOW_POLLS {
break "behind";
}
}
// `Shared` cannot make one (its `snapshot` is the default
// `false`), so this arm means the install's own bookkeeping, not
// a second fetch.
Progress::SnapshotInstalled { at } => {
eprintln!("a snapshot at {at} was installed by the follow step");
if polls >= FOLLOW_POLLS {
break "behind";
}
}
Progress::Diverged { own, primary_head } => anyhow::bail!(
"diverged: after the snapshot the primary answers head {primary_head}, below the {own} this \
box holds — it is not the log this copy was following"
),
Progress::SnapshotNeeded { wanted, first } => anyhow::bail!(
"hole-again: the snapshot installed at {at}, but retention on the primary moved \
again while this ran — it now retains from {first} and this twin needs {wanted}. \
The install is durable and nothing is lost; run the catch-up again"
),
}
};
let applied = follower.applied_seq();
let (_retains_from, primary_head) = primary
.head()
.context("reading the primary's head after the catch-up")?;
Ok(Report {
verdict,
was,
streamed: true,
at,
pieces,
bytes,
attempts: tries,
froze,
first_after: stores.txlog().first_seq(),
applied,
primary_head,
state: state.to_owned(),
caught_up: applied >= primary_head,
})
}
/// `gunnar control twin catch-up --at <primary> --data-root <this twin>`.
pub(super) fn catch_up(
at: &str,
data_root: &Path,
identity: Option<&Path>,
stall_secs: u64,
attempts: u32,
dry_run: bool,
) -> Result<()> {
// Refused before the data root is even opened: a snapshot installs only
// over a follower-only log (`snapshot::install`,
// `crates/gunnar-server/src/twin/snapshot.rs:483`), and pointing this at
// a primary's own root would be asking to replace it with another box's
// state.
anyhow::ensure!(
gunnar_server::twin::is_twin_root(data_root),
"not-a-twin: {} has no twin.json, so it is not a twin's data root. --data-root is THIS \
box's root and --at is the primary; a snapshot installs only on a twin",
data_root.display()
);
let signer = private_key(identity)?;
let rt = runtime()?;
let channel = connect(&rt, at)?;
let stores =
Arc::new(ServerStores::open(data_root).with_context(|| {
format!("opening the twin's data root {}", data_root.display())
})?);
let source = Grpc {
rt,
client: Mutex::new(MirrorClient::new(channel)),
signer,
stores: Arc::clone(&stores),
};
let report = run_catch_up(
Arc::clone(&stores),
source,
// Zero would be a deadline that fires before the first piece can
// possibly arrive, which is a hang reported as a stall.
Duration::from_secs(stall_secs.max(1)),
attempts.max(1),
dry_run,
)?;
let caught_up = report.caught_up;
let (at_seq, applied, primary_head) = (report.at, report.applied, report.primary_head);
let streamed = report.streamed;
for line in report.lines() {
println!("{line}");
}
// The same shape `replication_flush` ends on (`control.rs:3367`): a verb
// whose whole point is a proof exits non-zero when it did not get one.
if streamed && !caught_up {
anyhow::bail!(
"the snapshot installed at {at_seq} but this twin is NOT level with its primary \
({applied} of {primary_head}); the hole is closed — finish with \
`gunnar control twin follow`, which freezes nothing"
);
}
Ok(())
}
#[cfg(test)]
mod catch_up_tests {
use std::sync::atomic::{AtomicU64, Ordering};
use gunnar_core::{HashKind, NewRepo, RepoRef};
use gunnar_server::twin::TwinConfig;
use super::*;
fn head_of(first_seq: u64, head_seq: u64) -> gunnar_proto::LogHeadResponse {
gunnar_proto::LogHeadResponse {
head_seq,
first_seq,
..Default::default()
}
}
/// **A twin that is merely BEHIND must never freeze its primary.**
///
/// The refusal is the feature. `Mirror.SnapshotFetch` stops every framed
/// write on the primary for the whole of the copy
/// (`gunnar-grpc/src/service/mirror.rs:117`), and a twin whose missing
/// range is still retained needs none of that — `twin follow` reads it
/// out of the log. So the verb refuses BY NAME and points at the verb
/// that costs nothing, rather than quietly doing the expensive thing
/// because it also works.
///
/// The boundary is asserted both ways round: `first == wanted` is the
/// oldest sequence the twin still needs and is NOT a hole; `first ==
/// wanted + 1` is one gone and is.
#[test]
fn a_twin_that_is_merely_behind_is_refused_by_name_and_sent_to_follow() {
let refusal = |applied: u64, first: u64, head: u64| {
format!(
"{:#}",
hole_verdict(applied, &head_of(first, head), false).expect_err("must refuse")
)
};
let say = refusal(40, 1, 100);
assert!(
say.starts_with("no-hole:"),
"the refusal must be named first: {say}"
);
assert!(
say.contains("gunnar control twin follow"),
"a refusal that does not name the verb that DOES work is a dead end: {say}"
);
assert!(say.contains("freezes nothing"), "{say}");
// The boundary. 40 applied ⇒ 41 wanted.
assert!(
refusal(40, 41, 100).starts_with("no-hole:"),
"first == wanted is not a hole"
);
assert_eq!(
hole_verdict(40, &head_of(42, 100), false).expect("42 > 41 is a hole"),
Verdict::Hole {
wanted: 41,
first: 42,
head: 100
}
);
// The other three refusals keep their own names, because "there is
// nothing to do" and "you are pointed at the wrong box" are not the
// same fact.
assert!(refusal(100, 42, 100).starts_with("already-level:"));
assert!(refusal(0, 1, 0).starts_with("empty-log:"));
let twin_answered = gunnar_proto::LogHeadResponse {
head_seq: 100,
first_seq: 90,
twin: true,
applied_seq: 100,
..Default::default()
};
assert!(format!(
"{:#}",
hole_verdict(40, &twin_answered, false).expect_err("a twin is not a primary")
)
.starts_with("not-a-primary:"));
// A half install is the one state that needs a snapshot with no hole
// in sight: the stores are half of one already.
assert!(matches!(
hole_verdict(40, &head_of(1, 100), true).expect("a half install must be allowed"),
Verdict::HalfInstalled { wanted: 41, .. }
));
}
/// A primary that stops sending mid-snapshot: one piece, then silence.
/// Its `Drop` is the observable that matters — dropping the stream is
/// what releases the primary's `Log::freeze()`, so a refusal that leaves
/// it open has given up on nothing at all.
struct WentQuiet(std::sync::Arc<std::sync::atomic::AtomicBool>);
impl Drop for WentQuiet {
fn drop(&mut self) {
self.0.store(true, Ordering::SeqCst);
}
}
impl tokio_stream::Stream for WentQuiet {
type Item = std::result::Result<gunnar_proto::LogChunk, tonic::Status>;
fn poll_next(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
std::task::Poll::Pending
}
}
fn a_piece() -> gunnar_proto::LogChunk {
gunnar_proto::LogChunk {
seq: 0,
kind: 1,
meta: Vec::new(),
blob_len: 4,
blob: b"abcd".to_vec(),
last: true,
restore_head: 0,
restore_bytes: 0,
log_id: String::new(),
offset: None,
entry_crc: None,
snapshot_item: None,
snapshot_id: String::new(),
}
}
/// **A primary that goes quiet mid-snapshot fails by name, and the
/// freeze it is holding is released rather than left running.**
///
/// Without the deadline `stream.next()` waits until TCP gives up —
/// minutes to hours — with every framed write on the primary stopped for
/// all of it, and a command that is indistinguishable from a snapshot
/// that is simply large. The only thing on this side that can end that
/// is dropping the stream, so the test asserts the drop and not only the
/// sentence.
#[test]
fn a_stalled_snapshot_stream_fails_by_name_and_drops_the_stream_that_holds_the_freeze() {
use tokio_stream::StreamExt as _;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("a runtime");
let dropped = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let stream: super::super::Chunks = Box::pin(
tokio_stream::iter(vec![Ok(a_piece())])
.chain(WentQuiet(std::sync::Arc::clone(&dropped))),
);
let mut took = 0u64;
let started = Instant::now();
let err = rt
.block_on(drain(stream, Duration::from_millis(80), &mut |_chunk| {
took += 1;
Ok(())
}))
.expect_err("a stream that goes quiet must not be waited on for ever");
assert!(
started.elapsed() < Duration::from_secs(5),
"the drain waited {:?} for a primary that is not sending",
started.elapsed()
);
assert_eq!(took, 1, "the piece that DID arrive was taken");
let say = format!("{err:#}");
assert!(
say.starts_with("stalled:"),
"the refusal must be named first: {say}"
);
assert!(
say.contains("Log::freeze()"),
"it must say what was being held: {say}"
);
assert!(
say.contains("dropped here"),
"it must say what releases it: {say}"
);
assert!(
dropped.load(Ordering::SeqCst),
"the stream holding the primary's freeze was still open after the refusal — the \
primary would stay frozen, silently, until its own 120 s guard fired"
);
}
/// A local primary as a [`Primary`]: real stores, a real snapshot out of
/// `snapshot::produce`, cut into real `LogChunk`s, and a `first_seq` an
/// operator's `twin retain` has moved above what the twin holds.
struct LocalPrimary {
stores: Arc<ServerStores>,
/// What the primary claims to retain from — `Mirror.LogRetain`'s
/// effect, without needing a second log.
retained_from: u64,
rt: tokio::runtime::Runtime,
/// Break the stream after this many pieces, once (the resume path).
break_after: AtomicU64,
/// Go quiet after this many pieces, for ever (the stall path).
quiet_after: AtomicU64,
/// How many times `SnapshotFetch` was opened — i.e. how many times
/// the primary was frozen.
opened: AtomicU64,
}
impl LocalPrimary {
fn new(stores: Arc<ServerStores>, retained_from: u64) -> Self {
Self {
stores,
retained_from,
rt: tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("a runtime"),
break_after: AtomicU64::new(0),
quiet_after: AtomicU64::new(0),
opened: AtomicU64::new(0),
}
}
}
impl LogSource for LocalPrimary {
fn head(&self) -> anyhow::Result<(u64, u64)> {
Ok((
self.retained_from.max(self.stores.txlog().first_seq()),
self.stores.txlog().head_seq(),
))
}
fn fetch(
&self,
from: u64,
to: u64,
max_bytes: u64,
) -> anyhow::Result<Vec<gunnar_core::txlog::Entry>> {
Ok(gunnar_server::twin::fetch_range(
self.stores.txlog(),
from,
to,
max_bytes,
)?)
}
}
impl Primary for LocalPrimary {
fn log_head(&self) -> Result<gunnar_proto::LogHeadResponse> {
let (first_seq, head_seq) = LogSource::head(self)?;
Ok(gunnar_proto::LogHeadResponse {
head_seq,
first_seq,
..Default::default()
})
}
fn stream_snapshot(
&self,
stall: Duration,
sink: &mut dyn FnMut(&gunnar_proto::LogChunk) -> Result<()>,
) -> Result<Drained> {
use tokio_stream::StreamExt as _;
self.opened.fetch_add(1, Ordering::SeqCst);
let mut chunks: Vec<std::result::Result<gunnar_proto::LogChunk, tonic::Status>> =
Vec::new();
{
let mut produce = |p: snapshot::Piece<'_>| {
chunks.push(Ok(gunnar_proto::LogChunk {
seq: 0,
kind: u32::from(p.kind),
meta: p.meta.to_vec(),
blob_len: p.blob_len,
blob: p.blob.to_vec(),
last: p.last,
restore_head: 0,
restore_bytes: 0,
log_id: String::new(),
offset: None,
entry_crc: None,
snapshot_item: None,
snapshot_id: String::new(),
}));
Ok(())
};
snapshot::produce(&self.stores, &mut produce)?;
}
let quiet = self.quiet_after.load(Ordering::SeqCst);
if quiet > 0 {
chunks.truncate(quiet as usize);
let stream: super::super::Chunks = Box::pin(tokio_stream::iter(chunks).chain(
WentQuiet(std::sync::Arc::new(std::sync::atomic::AtomicBool::new(
false,
))),
));
return self.rt.block_on(drain(stream, stall, sink));
}
// Once: the first attempt breaks part way, the next is whole.
let broken = self.break_after.swap(0, Ordering::SeqCst);
if broken > 0 {
chunks.truncate(broken as usize);
chunks.push(Err(tonic::Status::unavailable(
"the primary's disk went away",
)));
}
let stream: super::super::Chunks = Box::pin(tokio_stream::iter(chunks));
self.rt.block_on(drain(stream, stall, sink))
}
}
/// A primary with two repositories, and a twin that followed it once and
/// then went away while two more landed. Answers
/// `(primary, twin, the twin's head when it left, the primary's head now)`.
fn away_behind_a_hole(
pd: &std::path::Path,
td: &std::path::Path,
) -> (Arc<ServerStores>, Arc<ServerStores>, u64, u64) {
let primary = Arc::new(ServerStores::open(pd).expect("the primary's stores"));
primary
.catalog()
.create_namespace("ns")
.expect("a namespace");
primary
.create_repo(
NewRepo::new(RepoRef::new("ns", "app").unwrap()).hash_kind(HashKind::Sha256),
false,
)
.expect("a repository");
// The twin follows it once, the honest way: no hole yet.
TwinConfig {
primary: "primary.test:50051".into(),
joined_at_unix_ms: 0,
}
.save(td)
.expect("twin.json");
let twin = Arc::new(ServerStores::open(td).expect("the twin's stores"));
Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())))
.step()
.expect("the twin's first poll");
let seen = twin.txlog().head_seq();
assert_eq!(seen, primary.txlog().head_seq(), "the twin starts level");
// Then it is away, and work lands that it never sees.
primary
.catalog()
.create_namespace("later")
.expect("a namespace");
primary
.create_repo(
NewRepo::new(RepoRef::new("later", "inside-the-hole").unwrap())
.hash_kind(HashKind::Sha256),
false,
)
.expect("a repository inside the hole");
let now = primary.txlog().head_seq();
assert!(
now > seen + 1,
"the hole must be more than one sequence wide: {seen} → {now}"
);
(primary, twin, seen, now)
}
/// **B8, end to end, as the verb runs it**: a twin behind a retention
/// hole, a snapshot that breaks once and is taken again, an install, and
/// a twin that follows from there.
///
/// The proof that a SNAPSHOT closed the hole — rather than a fetch that
/// happened to work — is the twin's OWN `first_seq` moving off 1. Only
/// `Log::restart_after`, which nothing but the install calls, does that
/// (`.nornir/twin-transaction-log.md`, row R8).
#[test]
fn a_twin_behind_a_retention_hole_is_caught_up_by_the_verb_and_follows_again() {
let pd = tempfile::tempdir().expect("a temp dir");
let td = tempfile::tempdir().expect("a temp dir");
let (primary, twin, seen, hole_at) = away_behind_a_hole(pd.path(), td.path());
// ── the refusal, against the same real stores ───────────────────
// The primary still retains everything, so this is a twin that is
// merely behind: refused, and nothing on the twin moved.
let level = LocalPrimary::new(Arc::clone(&primary), 1);
let err = run_catch_up(Arc::clone(&twin), level, Duration::from_secs(5), 1, false)
.expect_err("a twin that is merely behind must be refused");
assert!(format!("{err:#}").starts_with("no-hole:"), "{err:#}");
assert_eq!(
twin.txlog().head_seq(),
seen,
"the refusal moved the twin's log"
);
assert_eq!(
twin.txlog().first_seq(),
1,
"the refusal installed a snapshot"
);
// ── the hole, and the catch-up ──────────────────────────────────
let source = LocalPrimary::new(Arc::clone(&primary), hole_at);
// The first attempt breaks two pieces in; the second is whole.
source.break_after.store(2, Ordering::SeqCst);
let report = run_catch_up(Arc::clone(&twin), source, Duration::from_secs(30), 3, false)
.expect("a hole is closed by a snapshot");
assert_eq!(
report.at, hole_at,
"the snapshot is taken at the primary's head"
);
assert_eq!(
report.attempts, 2,
"the broken stream was taken again from the beginning"
);
assert!(
report.pieces > 0 && report.caught_up,
"{} pieces, caught_up {}",
report.pieces,
report.caught_up
);
assert_eq!(report.state, "following");
assert_eq!(
(twin.txlog().first_seq(), twin.txlog().head_seq()),
(hole_at + 1, hole_at),
"only a snapshot install moves a twin's own first_seq off 1 — R8's proof"
);
assert!(
twin.catalog()
.get(&RepoRef::new("later", "inside-the-hole").unwrap())
.unwrap()
.is_some(),
"a repository created inside the hole is not on the twin"
);
assert!(
!snapshot::installing(td.path()),
"the install marker outlived the install"
);
// The freeze is in the report, as a number, loudly.
let lines = report.lines();
assert!(lines[0].starts_with("hole "), "{:?}", lines[0]);
let frozen = lines
.iter()
.find(|l| l.starts_with("FROZEN"))
.expect("the freeze must be reported");
assert!(frozen.contains("NO framed write"), "{frozen}");
assert_eq!(
report.first_after,
hole_at + 1,
"the report must carry R8's own proof"
);
assert!(
lines.iter().any(|l| l.starts_with("resumed at ")),
"{lines:?}"
);
assert!(
lines.iter().any(|l| l.starts_with("following YES")),
"{lines:?}"
);
// ── …and it follows again ───────────────────────────────────────
primary
.create_repo(
NewRepo::new(RepoRef::new("ns", "after-the-snapshot").unwrap())
.hash_kind(HashKind::Sha256),
false,
)
.expect("a repository after the snapshot");
let after = Follower::new(Arc::clone(&twin), Box::new(Arc::clone(primary.txlog())));
assert!(matches!(
after.step().expect("the poll after the snapshot"),
Progress::Applied { .. }
));
assert_eq!(
twin.txlog().head_seq(),
primary.txlog().head_seq(),
"the twin is level again"
);
assert!(twin
.catalog()
.get(&RepoRef::new("ns", "after-the-snapshot").unwrap())
.unwrap()
.is_some());
}
/// **A stall inside the verb fails by name, takes the stream ONCE, and
/// leaves the twin exactly as it found it.**
///
/// The retry is what makes this worth its own arm: a broken stream is
/// taken again (the arm above proves it), and a STALL must not be —
/// re-opening `SnapshotFetch` would freeze the primary from the start
/// again for a peer that has just proven it is not sending. `opened` is
/// the count of freezes this verb asked the primary for, and it must be
/// 1 even with `--attempts 3`.
#[test]
fn a_stalled_catch_up_fails_by_name_freezes_the_primary_once_and_installs_nothing() {
let pd = tempfile::tempdir().expect("a temp dir");
let td = tempfile::tempdir().expect("a temp dir");
let (primary, twin, seen, hole_at) = away_behind_a_hole(pd.path(), td.path());
let source = Arc::new(LocalPrimary::new(Arc::clone(&primary), hole_at));
source.quiet_after.store(1, Ordering::SeqCst);
let started = Instant::now();
let err = run_catch_up(
Arc::clone(&twin),
Shared(Arc::clone(&source)),
Duration::from_millis(120),
3,
false,
)
.expect_err("a primary that goes quiet must not hang the verb");
assert!(
started.elapsed() < Duration::from_secs(10),
"waited {:?}",
started.elapsed()
);
let say = format!("{err:#}");
assert!(
say.contains("stalled:"),
"the refusal must name the stall: {say}"
);
assert!(
say.contains("frozen no longer"),
"it must say the primary was let go: {say}"
);
assert!(
say.contains("not retried"),
"it must say why it stopped at one attempt: {say}"
);
assert_eq!(
source.opened.load(Ordering::SeqCst),
1,
"a stall was retried: the primary was frozen again for a peer that had stopped sending"
);
assert_eq!(
twin.txlog().head_seq(),
seen,
"the twin's log moved on a failed catch-up"
);
assert_eq!(
twin.txlog().first_seq(),
1,
"a snapshot was installed out of a stalled stream"
);
assert!(
!snapshot::installing(td.path()),
"the twin was left marked mid-install"
);
assert!(
!td.path().join(snapshot::STAGING).exists(),
"the staging directory of a failed attempt was left on the twin's disk"
);
}
/// [`Shared`] is the follow-again handle; the stall arm above reuses it
/// as the source so the fake's counters survive the call. It must
/// forward the head it wraps.
impl Primary for Shared<LocalPrimary> {
fn log_head(&self) -> Result<gunnar_proto::LogHeadResponse> {
self.0.log_head()
}
fn stream_snapshot(
&self,
stall: Duration,
sink: &mut dyn FnMut(&gunnar_proto::LogChunk) -> Result<()>,
) -> Result<Drained> {
self.0.stream_snapshot(stall, sink)
}
}
}
}
#[cfg(feature = "server")]
use with_server::{catch_up, follow, restore};
#[cfg(not(feature = "server"))]
fn follow(
_at: &str,
_data_root: &Path,
_identity: Option<&Path>,
_poll: u64,
_once: bool,
) -> Result<()> {
anyhow::bail!("this build has no server half, so it cannot hold a twin's stores; use a build with the `server` feature (the appliance medium — the same one a primary installs — made a twin by `gunnar control twin join`)")
}
#[cfg(not(feature = "server"))]
fn restore(_from: &Path, _into: &Path) -> Result<()> {
anyhow::bail!("this build has no server half, so it cannot rebuild a data root; use a build with the `server` feature (the appliance medium — the same one a primary installs — made a twin by `gunnar control twin join`)")
}
/// The catch-up STAGES a snapshot into a data root and installs it over that
/// box's stores, which is server work by construction. A `control`-only build
/// says so by name rather than opening a stream it could not do anything with —
/// and opening one would freeze the primary for nothing.
#[cfg(not(feature = "server"))]
fn catch_up(
_at: &str,
_data_root: &Path,
_identity: Option<&Path>,
_stall_secs: u64,
_attempts: u32,
_dry_run: bool,
) -> Result<()> {
anyhow::bail!("this build has no server half, so it cannot install a snapshot into a twin's stores; use a build with the `server` feature (the appliance medium — the same one a primary installs — made a twin by `gunnar control twin join`)")
}
/// The replay against a REAL `Bootstrap.Restore` (gunnar-grpc over `CoreControlPlane` on a fresh
/// data root) and a twin whose log is a real primary's.
#[cfg(all(test, feature = "server"))]
mod replay_tests {
use std::sync::Arc;
use gunnar_core::txlog::Log;
use gunnar_core::{HashKind, NewRepo, RepoRef};
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::is_restoring;
use super::*;
/// A twin over `log`, two entries a fetch; `break_later` breaks every fetch after the first
/// one chunk in. It answers as the primary it copies (`twin: false`): the verb has always
/// replayed from one, so the freshness guard is not what this test is about.
struct FakeTwin {
log: Arc<Log>,
break_later: bool,
}
struct Chunks(std::vec::IntoIter<Result<gunnar_proto::LogChunk, gunnar_repair::Fault>>);
impl gunnar_repair::pump::ChunkStream for Chunks {
type Chunk = gunnar_proto::LogChunk;
async fn next(&mut self) -> Option<Result<gunnar_proto::LogChunk, gunnar_repair::Fault>> {
self.0.next()
}
}
impl gunnar_repair::pump::Twin for FakeTwin {
type Chunk = gunnar_proto::LogChunk;
type Stream = Chunks;
fn addr(&self) -> String {
"twin:1".into()
}
async fn head(&self) -> Result<gunnar_repair::TwinHead, gunnar_repair::Fault> {
Ok(gunnar_repair::TwinHead {
first_seq: self.log.first_seq(),
head_seq: self.log.head_seq(),
log_id: "the-twins-log".into(),
..Default::default()
})
}
async fn snapshot(&self, _: u64, _: u64, _: &str) -> Result<Chunks, gunnar_repair::Fault> {
Err(gunnar_repair::Fault::new(
"not_found",
"this twin's log starts at 1",
))
}
async fn log(&self, from: u64, to: u64, _: u64) -> Result<Chunks, gunnar_repair::Fault> {
let to = to.min(from + 1);
let mut out: Vec<Result<gunnar_proto::LogChunk, gunnar_repair::Fault>> = self
.log
.range(from, to)
.map_err(|e| gunnar_repair::Fault::new("internal", e.to_string()))?
.into_iter()
.map(|e| {
Ok(gunnar_proto::LogChunk {
seq: e.seq,
kind: u32::from(e.kind),
blob_len: e.blob.len() as u64,
meta: e.meta,
blob: e.blob,
last: true,
restore_head: 0,
restore_bytes: 0,
log_id: String::new(),
offset: None,
entry_crc: None,
snapshot_item: None,
snapshot_id: String::new(),
})
})
.collect();
if self.break_later && from > self.log.first_seq() {
out.truncate(1);
out.push(Err(gunnar_repair::Fault::new(
"unavailable",
"the twin's disk went away",
)));
}
Ok(Chunks(out.into_iter()))
}
}
#[test]
fn a_broken_fetch_leaves_the_box_restoring_and_a_short_clean_stream_is_refused() {
// The primary the twin followed: enough entries to stop well short.
let pd = tempfile::tempdir().unwrap();
let primary = ServerStores::open(pd.path()).unwrap();
for ns in ["a", "b", "c"] {
primary.catalog().create_namespace(ns).unwrap();
primary
.create_repo(
NewRepo::new(RepoRef::new(ns, "app").unwrap()).hash_kind(HashKind::Sha256),
false,
)
.unwrap();
}
let log = Arc::clone(primary.txlog());
let (first, last) = (log.first_seq(), log.head_seq());
assert!(first == 1 && last >= 6, "{first}..={last}");
// The fresh box, serving Bootstrap.Restore.
let fd = tempfile::tempdir().unwrap();
let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
// The operator's key, fed to the box as `--restore-pubkey` would: the stream is taken
// only under a credential it signs (h-v1-6a), and only it resumes what it started.
let operator = ssh_key::PrivateKey::from_openssh(
std::str::from_utf8(&gunnar_core::mint_ed25519().unwrap()).unwrap(),
)
.unwrap();
let plane: Arc<dyn gunnar_grpc::ControlPlane> = Arc::new(
CoreControlPlane::new(Arc::clone(&fresh))
.with_restore_keys(vec![operator.public_key().to_openssh().unwrap()]),
);
let ctx = gunnar_grpc::ServiceContext::new(
plane,
Arc::new(gunnar_grpc::EventBus::new(16)),
gunnar_grpc::ServerConfig::default(),
);
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.unwrap();
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let at = listener.local_addr().unwrap().to_string();
rt.spawn(
tonic::transport::Server::builder()
.add_service(gunnar_proto::bootstrap_server::BootstrapServer::new(
gunnar_grpc::BootstrapService::new(ctx),
))
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)),
);
// 1. The server's half: a stream that ends CLEANLY after 2 entries but announced `last`.
let short: Vec<gunnar_proto::LogChunk> = log
.range(1, 2)
.unwrap()
.into_iter()
.map(|e| gunnar_proto::LogChunk {
seq: e.seq,
kind: u32::from(e.kind),
blob_len: e.blob.len() as u64,
meta: e.meta,
blob: e.blob,
last: true,
restore_head: last,
restore_bytes: 0,
log_id: "the-twins-log".into(),
offset: None,
entry_crc: None,
snapshot_item: None,
snapshot_id: String::new(),
})
.collect();
let mut client =
gunnar_proto::bootstrap_client::BootstrapClient::new(connect(&rt, &at).unwrap());
let refused = rt
.block_on(client.restore(authed(&operator, tokio_stream::iter(short)).unwrap()))
.expect_err("a short stream is not a restore");
assert_eq!(
refused.code(),
tonic::Code::FailedPrecondition,
"{refused:?}"
);
assert!(refused.message().contains("short stream"), "{refused:?}");
assert!(
is_restoring(fd.path()),
"refused short: the box stays restoring"
);
assert_eq!(fresh.txlog().head_seq(), 2);
// 2. The client's half: the twin's fetch breaks mid-way — the replay is an error, and the
// box did NOT finish (still restoring, below the head).
let opts = gunnar_repair::RechargeOpts {
allow_primary_source: true,
ignore_room: true,
..Default::default()
};
let main = gunnar_repair::proto::ProtoMain {
addr: at.clone(),
channel: connect(&rt, &at).unwrap(),
key: operator.clone(),
};
let err = rt
.block_on(replay_into(
&FakeTwin {
log: Arc::clone(&log),
break_later: true,
},
&main,
&opts,
))
.expect_err("a broken fetch is an error");
assert!(format!("{err:#}").contains("left restoring"), "{err:#}");
assert!(
is_restoring(fd.path()),
"a broken fetch never finishes the restore"
);
let reached = fresh.txlog().head_seq();
assert!(
reached < last,
"the box stopped short at {reached} of {last}"
);
// 3. Resumable: the replay run again finishes at the twin's head.
let answer = rt
.block_on(replay_into(
&FakeTwin {
log: Arc::clone(&log),
break_later: false,
},
&main,
&opts,
))
.expect("the resent replay");
assert_eq!(answer.head_seq, last);
assert!(answer.needs_restart);
assert!(
!is_restoring(fd.path()),
"the complete replay finished the restore"
);
}
}
/// The guard itself and its T3/T17 tests live in `gunnar-repair` (`plan::stale_source_tests`);
/// what stays here is what only this crate can see: the server's wording and the flag.
#[cfg(test)]
mod stale_source_tests {
use super::MAX_SILENCE_MS;
/// The follower-not-running refusal keys on the server's words; written in two crates, held
/// together here (the only crate that links both).
#[cfg(feature = "server")]
#[test]
fn the_follower_not_running_wording_is_the_servers() {
assert_eq!(
super::FOLLOWER_NOT_RUNNING,
gunnar_server::twin::FOLLOWER_NOT_RUNNING
);
}
/// `--primary-fenced` (R-13) is the caller's word that the old primary is fenced; never a default.
#[test]
fn primary_fenced_is_said_out_loud() {
use crate::cli::{Cli, Command, ControlCommand, ControlTwinCommand};
use clap::Parser as _;
let fenced = |extra: &[&str]| {
let mut argv = vec![
"gunnar", "control", "twin", "replay", "--twin", "t:1", "--into", "i:1",
];
argv.extend_from_slice(extra);
match Cli::try_parse_from(argv).expect("replay parses").command {
Command::Control(ControlCommand::Twin(ControlTwinCommand::Replay {
primary_fenced,
..
})) => primary_fenced,
other => panic!("expected twin replay, got {other:?}"),
}
};
assert!(!fenced(&[]));
assert!(fenced(&["--primary-fenced"]));
}
/// `--max-silence` reaches the verb, and its default IS [`MAX_SILENCE_MS`]: a default written
/// in two places that drifted would make the flag's help lie about what the guard does.
#[test]
fn the_max_silence_default_is_the_guards() {
use crate::cli::{Cli, Command, ControlCommand, ControlTwinCommand};
use clap::Parser as _;
let silence = |extra: &[&str]| {
let mut argv = vec![
"gunnar", "control", "twin", "replay", "--twin", "t:1", "--into", "i:1",
];
argv.extend_from_slice(extra);
match Cli::try_parse_from(argv).expect("replay parses").command {
Command::Control(ControlCommand::Twin(ControlTwinCommand::Replay {
max_silence,
..
})) => max_silence.as_millis() as u64,
other => panic!("expected twin replay, got {other:?}"),
}
};
assert_eq!(silence(&[]), MAX_SILENCE_MS);
assert_eq!(silence(&["--max-silence", "2h"]), 2 * 3_600_000);
assert_eq!(silence(&["--max-silence", "now"]), 0);
assert!(Cli::try_parse_from([
"gunnar",
"control",
"twin",
"replay",
"--twin",
"t:1",
"--into",
"i:1",
"--max-silence",
"a day"
])
.is_err());
}
}