crates/gunnar-server/tests/restore_after_snapshot.rs
//! **A main recharged from a twin whose log starts after its snapshot** (restore-after-snapshot,
//! 2026-10-07) — on the real wire.
//!
//! FOUND by the round 5.1 integration test: after a twin took a full resync, its log starts at
//! the snapshot's `at + 1` and everything up to `at` is in its BASE (617 MB there). gunnar's own
//! `restore_from` and `control twin replay` stream that base first (`Mirror.SnapshotFetch`), but
//! a base larger than one capped call of monetize's restore lane could never arrive: the box
//! staged snapshot pieces per stream and threw them away when the stream ended, and the twin
//! could only send its base from the start.
//!
//! Here a twin takes a snapshot of a primary (a 24 MiB pack in it), keeps it as its base and
//! follows past it; a fresh main is served on the real wire, and a LANE drives the recharge the
//! way monetize's gunnar guest does — per call: `Bootstrap.Status`, the base from where the box
//! holds it (`SnapshotFetch{from_item, from_offset, expect_id}`), then the log after it, until the
//! call's byte budget is spent. The base needs several calls; the recharged main must be the
//! primary, and nothing staged may be left on its volume.
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};
use gix_object::Kind as ObjKind;
use gunnar_core::{HashKind, NewRepo, Principal, PrincipalKind, RepoRef};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_proto::bootstrap_client::BootstrapClient;
use gunnar_proto::mirror_client::MirrorClient;
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, LocalSource, Progress};
use gunnar_server::CoreControlPlane;
use gunnar_store::conformance::pack::PackBuilder;
/// The big pack inside the snapshot: noise this large.
const BIG: usize = 24 << 20;
/// What one call of the lane carries: the base is cut inside at least three times.
const BUDGET: u64 = 7 << 20;
/// Where a restoring box stages a twin's snapshot across streams.
const SNAPSHOT_STAGING: &str = "txlog-restore-snapshot";
/// The `SnapshotAt` terminator's kind (gunnar_core::txlog::Kind::SnapshotAt).
const SNAPSHOT_AT: u32 = gunnar_core::txlog::Kind::SnapshotAt as u32;
fn free_addr() -> SocketAddr {
let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
l.local_addr().expect("addr")
}
/// A control plane for `stores`, served at `addr` until killed.
struct Served {
_rt: tokio::runtime::Runtime,
stop: Option<tokio::sync::oneshot::Sender<()>>,
addr: SocketAddr,
}
fn serve(stores: &Arc<ServerStores>) -> Served {
let addr = free_addr();
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(stores)));
let ctx = ServiceContext::new(
plane,
Arc::new(EventBus::new(16)),
ServerConfig::default().insecure_allow_all_for_local_development(),
);
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("rt");
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
rt.spawn(async move {
let _ = gunnar_grpc::router(ctx, || 0)
.serve_with_shutdown(addr, async {
let _ = rx.await;
})
.await;
});
wait("the box listens", Duration::from_secs(10), || {
std::net::TcpStream::connect(addr).ok()
});
Served {
_rt: rt,
stop: Some(tx),
addr,
}
}
impl Served {
fn kill(mut self) {
if let Some(tx) = self.stop.take() {
let _ = tx.send(());
}
let addr = self.addr;
wait("the box stops listening", Duration::from_secs(10), || {
std::net::TcpStream::connect(addr).is_err().then_some(())
});
}
}
fn wait<T>(what: &str, within: Duration, mut probe: impl FnMut() -> Option<T>) -> T {
let t0 = Instant::now();
loop {
if let Some(v) = probe() {
return v;
}
assert!(t0.elapsed() < within, "{what}: not within {within:?}");
std::thread::sleep(Duration::from_millis(25));
}
}
fn noise(seed: u64, n: usize) -> Vec<u8> {
let mut x = seed
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
(0..n)
.map(|_| {
x = x
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
(x >> 33) as u8
})
.collect()
}
fn one_blob_pack(hash: gix_hash::Kind, content: &[u8]) -> (Vec<u8>, Vec<u8>) {
let id = gix_object::compute_hash(hash, ObjKind::Blob, content).expect("hash");
let mut pb = PackBuilder::new(hash);
pb.object(ObjKind::Blob, content.to_vec());
let fixture = pb.build().expect("pack");
(fixture.bytes, id.as_bytes().to_vec())
}
/// Every pack this box holds resolved: a snapshot walks reachability, which refuses mid-index.
fn wait_until_indexed(stores: &ServerStores) {
// The progress rows see a pack only once the background drain has STARTED it: a pack still
// queued reads as `active: false, 0/0`, "finished" (twin/tests.rs found it on dwarf at load
// 25, 2026-10-16). MEASURED on oden at load 130 (GUN-FIELD-TEST, 2026-10-31): this wait
// returned and the second snapshot met "index in progress for this repository: pack 1 100%
// (1/1 objects, done) — reachability needs the whole graph". So the queue is emptied first
// through the engine's own `index_now` (it takes the absorb gate: it waits for a drain
// already running and absorbs what is pending); after it nothing is queued and the rows are
// the truth.
for row in stores.catalog().list(None).expect("the catalog") {
let repo = stores
.catalog()
.get(&RepoRef::new(&row.namespace, &row.name).expect("a catalogued name"))
.expect("the catalog")
.expect("listed, so catalogued");
stores
.repo_store(&repo)
.expect("its store")
.serve()
.index_now()
.expect("index the pending packs");
}
wait(
"the primary's packs are indexed",
Duration::from_secs(120),
|| {
stores
.engine()
.index_progress()
.into_iter()
.filter(|(root, _)| root.starts_with(stores.root()))
.all(|(_, p)| !p.active && p.objects_done >= p.objects_total)
.then_some(())
},
);
}
/// What the primary was told, so the comparison names things.
struct Primary {
stores: Arc<ServerStores>,
oids: Vec<Vec<u8>>,
}
fn push(primary: &ServerStores, name: &str, content: &[u8]) -> Vec<u8> {
let repo = primary
.catalog()
.get(&RepoRef::new("ns", "app").unwrap())
.unwrap()
.expect("ns/app");
let store = primary.repo_store(&repo).expect("store");
let (pack, oid) = one_blob_pack(store.hash_kind(), content);
store.serve().put_pack(&pack).expect("pack");
store
.serve()
.update_ref(name, None, Some(&oid))
.expect("ref");
oid
}
fn primary(dir: &Path) -> Primary {
let stores = Arc::new(ServerStores::open(dir).expect("primary"));
stores.catalog().create_namespace("ns").expect("namespace");
stores
.create_repo(
NewRepo::new(RepoRef::new("ns", "app").unwrap()).hash_kind(HashKind::Sha256),
false,
)
.expect("repo");
stores
.auth()
.add_principal(Principal::new("alice", "Alice", PrincipalKind::User).unwrap())
.expect("alice");
let oids = vec![
push(&stores, "refs/heads/main", b"before the snapshot\n"),
push(&stores, "refs/heads/big", &noise(7, BIG)),
];
Primary { stores, oids }
}
/// The twin learns which log it follows — what a twin over the wire is told on its first poll
/// (an in-process `LocalSource` carries no identity). A box restored from it is then restored
/// from a NAMED history, which a network restore requires.
fn pin_primarys_log_id(primary: &Primary, twin: &ServerStores) {
let id = twin::log_id(primary.stores.root()).expect("a primary mints its log's identity");
twin::pin_log_id(twin.root(), &id).expect("the twin pins its primary's log identity");
}
/// **The twin takes a snapshot of `primary` NOW** (its retention is past the twin): the base at
/// the primary's head — into a fresh box made a twin in `twin_dir`, or into the twin `again`.
fn snapshot_into(
primary: &Primary,
twin_dir: &Path,
again: Option<&Arc<ServerStores>>,
) -> Arc<ServerStores> {
let t = match again {
Some(t) => Arc::clone(t),
None => {
let t = Arc::new(ServerStores::open(twin_dir).expect("twin"));
twin::become_twin(&t, "primary.test:50051").expect("a fresh box becomes a twin");
drop(t);
let t = Arc::new(ServerStores::open(twin_dir).expect("the twin"));
pin_primarys_log_id(primary, &t);
t
}
};
wait_until_indexed(&primary.stores);
let at = primary.stores.txlog().head_seq();
let f = Follower::new(
Arc::clone(&t),
Box::new(LocalSource {
stores: Arc::clone(&primary.stores),
retained_from: at,
}),
);
assert_eq!(
f.step().expect("the snapshot"),
Progress::SnapshotInstalled { at }
);
assert_eq!(
twin::snapshot::base_at(t.root()).unwrap(),
Some(at),
"the twin keeps the snapshot as its base"
);
t
}
/// The primary goes on after the snapshot, and the twin follows it past its base.
fn go_on(primary: &mut Primary, twin: &Arc<ServerStores>, tag: &str) {
primary.oids.push(push(
&primary.stores,
&format!("refs/heads/after-{tag}"),
&noise(tag.len() as u64 + 3, 2 << 20),
));
primary
.stores
.auth()
.add_principal(Principal::new(tag, format!("Somebody {tag}"), PrincipalKind::User).unwrap())
.expect("principal");
let f = Follower::new(
Arc::clone(twin),
Box::new(LocalSource {
stores: Arc::clone(&primary.stores),
retained_from: 0,
}),
);
while !matches!(
f.step().expect("the twin follows"),
Progress::CaughtUp { .. }
) {}
assert_eq!(twin.txlog().head_seq(), primary.stores.txlog().head_seq());
}
/// How a call of the lane ends when its budget is spent.
#[derive(Clone, Copy, Debug)]
enum Cut {
/// The guest's own budget: it closes the request half at a piece boundary.
Close,
/// The host's deadline: the call is CANCELLED.
Abort,
}
#[derive(Debug)]
enum Call {
Done(u64),
Stopped(String),
}
/// **The lane**: one call = what the gunnar guest does in one call, over the real wire.
struct Lane {
rt: tokio::runtime::Runtime,
twin: SocketAddr,
budget: u64,
}
impl Lane {
fn new(twin: SocketAddr, budget: u64) -> Lane {
Lane {
rt: tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.unwrap(),
twin,
budget,
}
}
fn status(&self, main: SocketAddr) -> gunnar_proto::BootstrapStatusResponse {
self.rt.block_on(async {
let mut c = BootstrapClient::connect(format!("http://{main}"))
.await
.expect("connect");
c.status(gunnar_proto::BootstrapStatusRequest {})
.await
.expect("Bootstrap.Status")
.into_inner()
})
}
fn call(&self, main: SocketAddr, cut: Cut) -> Call {
let budget = self.budget;
let twin = self.twin;
let st = self.status(main);
self.rt.block_on(async move {
let mut mirror = MirrorClient::connect(format!("http://{twin}"))
.await
.expect("the twin");
let head = mirror
.log_head(gunnar_proto::LogHeadRequest::default())
.await
.expect("LogHead")
.into_inner();
let (tx, rx) = tokio::sync::mpsc::channel::<gunnar_proto::LogChunk>(4);
let restore = tokio::spawn(async move {
let mut c = BootstrapClient::connect(format!("http://{main}"))
.await
.expect("connect main");
c.restore(tokio_stream::wrappers::ReceiverStream::new(rx))
.await
.map(|r| r.into_inner())
});
let stop =
|restore: tokio::task::JoinHandle<
Result<gunnar_proto::RestoreResponse, tonic::Status>,
>,
tx: tokio::sync::mpsc::Sender<gunnar_proto::LogChunk>| async move {
match cut {
Cut::Close => {
drop(tx);
match restore.await.expect("the restore task") {
Ok(r) => {
Call::Stopped(format!("answered OK at head {}", r.head_seq))
}
Err(status) => Call::Stopped(status.message().to_owned()),
}
}
Cut::Abort => {
restore.abort();
drop(tx);
Call::Stopped("cancelled".into())
}
}
};
let mut sent = 0u64;
// The base first, while the box holds less than it.
let needs_base = head.first_seq > 1
&& st.log_head_seq + 1 < head.first_seq
&& !st.restore_snapshot_whole;
if needs_base {
let req = gunnar_proto::SnapshotFetchRequest {
from_item: st.restore_snapshot_items,
from_offset: st.restore_snapshot_bytes,
expect_id: st.restore_snapshot_id.clone(),
};
// The twin refuses a base that changed — at the open or with its first message —
// and the lane asks for the new one from its start.
let mut first = true;
let mut s = mirror
.snapshot_fetch(req)
.await
.expect("SnapshotFetch")
.into_inner();
loop {
let next = match s.message().await {
Ok(next) => next,
Err(status) if first && status.message().contains("snapshot-changed") => {
s = mirror
.snapshot_fetch(gunnar_proto::SnapshotFetchRequest::default())
.await
.expect("SnapshotFetch from the start")
.into_inner();
first = false;
continue;
}
Err(status) => panic!("the twin's snapshot stream: {status:?}"),
};
first = false;
let Some(mut chunk) = next else { break };
chunk.restore_head = head.head_seq;
chunk.log_id.clone_from(&head.log_id);
let terminator = chunk.kind == SNAPSHOT_AT && chunk.last;
sent += chunk.blob.len() as u64;
if tx.send(chunk).await.is_err() {
break;
}
if sent >= budget && !terminator {
return stop(restore, tx).await;
}
}
}
let from = if needs_base {
head.first_seq
} else {
head.first_seq.max(st.log_head_seq + 1)
};
let offset = if st.restoring && st.restore_partial_seq == from {
st.restore_partial_bytes
} else {
0
};
let (mut cursor, mut highest) = (from, 0u64);
while cursor <= head.head_seq {
let req = gunnar_proto::LogFetchRequest {
from: cursor,
to: head.head_seq,
max_bytes: 0,
from_offset: if cursor == from { offset } else { 0 },
};
let mut s = mirror.log_fetch(req).await.expect("LogFetch").into_inner();
while let Some(mut chunk) = s.message().await.expect("the twin's stream") {
chunk.restore_head = head.head_seq;
chunk.log_id.clone_from(&head.log_id);
let (seq, last) = (chunk.seq, chunk.last);
sent += chunk.blob.len() as u64;
highest = highest.max(seq);
if tx.send(chunk).await.is_err() {
break;
}
if sent >= budget && !(last && seq >= head.head_seq) {
return stop(restore, tx).await;
}
}
assert!(highest >= cursor, "the twin answered nothing at {cursor}");
cursor = highest + 1;
}
drop(tx);
match restore.await.expect("the restore task") {
Ok(r) => Call::Done(r.head_seq),
Err(status) => Call::Stopped(status.message().to_owned()),
}
})
}
}
/// The recharged main, restarted as the primary it was: the twin's log after the base frame for
/// frame, and the primary's objects, refs and principals.
fn assert_recharged(primary: &Primary, twin: &ServerStores, main_root: &Path, principals: &[&str]) {
assert!(
!main_root.join(SNAPSHOT_STAGING).exists(),
"nothing staged is left on the main's volume"
);
let main = ServerStores::open(main_root).expect("the recharged main, restarted");
let (t, m) = (twin.txlog(), main.txlog());
// Restarted, the main re-seals its secrets under a KEK of its own: one frame past the twin's head.
assert!(
m.head_seq() >= t.head_seq(),
"the main reached the twin's head: {} < {}",
m.head_seq(),
t.head_seq()
);
assert_eq!(
m.first_seq(),
t.first_seq(),
"and its log starts after the same base"
);
for seq in t.first_seq()..=t.head_seq() {
assert_eq!(m.read(seq).unwrap(), t.read(seq).unwrap(), "seq {seq}");
}
let at = RepoRef::new("ns", "app").unwrap();
let p = primary
.stores
.repo_store(&primary.stores.catalog().get(&at).unwrap().unwrap())
.unwrap();
let r = main
.repo_store(
&main
.catalog()
.get(&at)
.unwrap()
.expect("the repository came back"),
)
.unwrap();
for oid in &primary.oids {
assert!(
r.serve().has(oid).unwrap(),
"object {} missing on the recharged main",
hex::encode(oid)
);
assert_eq!(
p.serve().read(oid).unwrap(),
r.serve().read(oid).unwrap(),
"object {} differs",
hex::encode(oid)
);
}
let mut refs_p: Vec<_> = p
.serve()
.refs()
.unwrap()
.into_iter()
.map(|r| (r.name, r.oid))
.collect();
let mut refs_r: Vec<_> = r
.serve()
.refs()
.unwrap()
.into_iter()
.map(|r| (r.name, r.oid))
.collect();
refs_p.sort();
refs_r.sort();
assert_eq!(refs_p, refs_r, "refs");
for who in principals {
assert!(
main.auth().principal(who).unwrap().is_some(),
"{who} came back"
);
}
}
/// Calls until the box answers DONE; what the box held of the base after each call.
fn drive(
lane: &Lane,
main: SocketAddr,
cut: Cut,
calls: usize,
) -> (Option<u64>, Vec<(u64, u64)>, Vec<String>) {
let (mut held, mut said) = (Vec::new(), Vec::new());
for call in 0..calls {
let r = lane.call(main, cut);
let st = lane.status(main);
println!(
"call {call}: {r:?} — head {} restoring {} snapshot {}:{}+{} whole {}",
st.log_head_seq,
st.restoring,
st.restore_snapshot_id,
st.restore_snapshot_items,
st.restore_snapshot_bytes,
st.restore_snapshot_whole
);
if st.restoring && st.log_head_seq == 0 {
held.push((st.restore_snapshot_items, st.restore_snapshot_bytes));
}
match r {
Call::Done(head) => return (Some(head), held, said),
Call::Stopped(s) => said.push(s),
}
}
(None, held, said)
}
/// ★ **The base is cut three times and resumed inside it; then the log; the main is the primary.**
/// A 24 MiB pack in the base, 7 MiB per call, every call ending cleanly at its budget.
///
/// RED before the fix: the box threw every staged snapshot piece away when its stream ended, and
/// the twin could only send its base from the start — the box held `(0, 0)` call after call.
#[test]
fn a_base_larger_than_one_call_is_resumed_across_calls_and_the_main_is_the_primary() {
let (pd, td, md) = (
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
);
let mut primary = primary(pd.path());
let twin = snapshot_into(&primary, td.path(), None);
go_on(&mut primary, &twin, "bob");
let twin_served = serve(&twin);
let main = Arc::new(ServerStores::open(md.path()).expect("a fresh main"));
let main_served = serve(&main);
let lane = Lane::new(twin_served.addr, BUDGET);
let (done, held, said) = drive(&lane, main_served.addr, Cut::Close, 16);
assert_eq!(
done,
Some(twin.txlog().head_seq()),
"never recharged; the box held {held:?} of the base; it said {said:?}"
);
assert!(
held.len() >= 3,
"the base was cut inside three times or more: {held:?}"
);
assert!(
held.windows(2).all(|w| w[1] > w[0]),
"every call went further into the base: {held:?}"
);
main_served.kill();
drop(main);
assert_recharged(&primary, &twin, md.path(), &["alice", "bob"]);
twin_served.kill();
}
/// ★ **A main restarted in the middle of the base resumes from what it made durable.** The calls
/// are CANCELLED at their budget; after two the main goes down, and — as after a power cut — its
/// staged snapshot carries bytes past what its record counts. It comes back holding exactly what
/// it said before, and the lane finishes from there.
#[test]
fn a_main_restarted_inside_the_base_resumes_from_what_it_made_durable() {
let (pd, td, md) = (
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
);
let mut primary = primary(pd.path());
let twin = snapshot_into(&primary, td.path(), None);
go_on(&mut primary, &twin, "bob");
let twin_served = serve(&twin);
let lane = Lane::new(twin_served.addr, BUDGET);
let mut main = Arc::new(ServerStores::open(md.path()).expect("a fresh main"));
let mut main_served = serve(&main);
let mut held = (0u64, 0u64);
for _ in 0..2 {
lane.call(main_served.addr, Cut::Abort);
let st = wait(
"the cancelled stream's handler let go",
Duration::from_secs(20),
|| {
let st = lane.status(main_served.addr);
((st.restore_snapshot_items, st.restore_snapshot_bytes) > held).then_some(st)
},
);
held = (st.restore_snapshot_items, st.restore_snapshot_bytes);
}
let before = lane.status(main_served.addr);
main_served.kill();
drop(main);
let data = md.path().join(SNAPSHOT_STAGING).join("blobs");
assert!(
data.exists(),
"the main stages the base at {}",
data.display()
);
{
use std::io::Write as _;
std::fs::OpenOptions::new()
.append(true)
.open(&data)
.unwrap()
.write_all(&noise(99, 12_345))
.unwrap();
}
main = Arc::new(ServerStores::open(md.path()).expect("the main, back mid-restore"));
main_served = serve(&main);
let after = lane.status(main_served.addr);
assert!(after.restoring, "the main comes back still restoring");
assert_eq!(
(
after.restore_snapshot_id.as_str(),
after.restore_snapshot_items,
after.restore_snapshot_bytes
),
(
before.restore_snapshot_id.as_str(),
before.restore_snapshot_items,
before.restore_snapshot_bytes
),
"it holds what it made durable — no more, no less"
);
let (done, held, said) = drive(&lane, main_served.addr, Cut::Close, 16);
assert_eq!(
done,
Some(twin.txlog().head_seq()),
"never finished after the restart; held {held:?}; said {said:?}"
);
main_served.kill();
drop(main);
assert_recharged(&primary, &twin, md.path(), &["alice", "bob"]);
twin_served.kill();
}
/// **The twin took a NEW snapshot while the main held part of the old one.** The twin refuses to
/// splice the two (`[snapshot-changed]`), the lane asks from the start, the main drops what it
/// held of the old base and takes the new one, and the recharged main is the primary as it is now.
#[test]
fn a_base_that_changed_between_calls_is_taken_again_from_its_start_never_spliced() {
let (pd, td, md) = (
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
);
let mut primary = primary(pd.path());
let twin = snapshot_into(&primary, td.path(), None);
go_on(&mut primary, &twin, "bob");
let twin_served = serve(&twin);
let main = Arc::new(ServerStores::open(md.path()).expect("a fresh main"));
let main_served = serve(&main);
let lane = Lane::new(twin_served.addr, BUDGET);
lane.call(main_served.addr, Cut::Close);
let first = lane.status(main_served.addr);
assert!(
!first.restore_snapshot_id.is_empty()
&& (first.restore_snapshot_items, first.restore_snapshot_bytes) > (0, 0),
"part of the first base is staged: {first:?}"
);
// The twin falls behind a hole again and takes a new snapshot.
primary.oids.push(push(
&primary.stores,
"refs/heads/in-the-second-base",
b"only the second base has this\n",
));
primary
.stores
.auth()
.add_principal(Principal::new("carol", "Carol", PrincipalKind::User).unwrap())
.expect("carol");
drop(snapshot_into(&primary, td.path(), Some(&twin)));
go_on(&mut primary, &twin, "dave");
let (done, held, said) = drive(&lane, main_served.addr, Cut::Close, 16);
assert_eq!(
done,
Some(twin.txlog().head_seq()),
"never finished; held {held:?}; said {said:?}"
);
main_served.kill();
drop(main);
assert_recharged(
&primary,
&twin,
md.path(),
&["alice", "bob", "carol", "dave"],
);
twin_served.kill();
}