crates/gunnar-server/tests/restore_large_entry.rs
//! **An entry larger than one call is restored — resumed INSIDE it** (restore-large-entry,
//! 2026-10-07).
//!
//! MEASURED by the r5 integration test (row R1-recharge-main-lost): a new main recharging from
//! its twin through monetize's restore lane restored up to a 200 MiB pack and then, every call,
//! "Bootstrap.Restore refused: the stream ended inside an entry; resend from the box's head + 1"
//! — 41 times, never progressing. Every stream the lane opens is cut a fixed time after it opens,
//! and the lane resumed per ENTRY: one entry that needs longer than a call to stream could never
//! be restored, and the main never came back.
//!
//! Here two real boxes — a twin that followed a primary, and a fresh main — are served on the
//! real wire, and a LANE drives the recharge the way monetize's gunnar guest does: per call, read
//! `Bootstrap.Status`, fetch the twin's log from where the box is (`Mirror.LogFetch`, from the
//! byte of the entry it holds part of when it says one), push it into `Bootstrap.Restore`, and
//! stop when the call's budget is spent — a byte budget here, so the test is deterministic: the
//! guest's is time. The big entry needs several calls; the lane must get through it.
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, Progress};
use gunnar_server::CoreControlPlane;
use gunnar_store::conformance::pack::PackBuilder;
/// The big entry: a pack holding one blob of noise this large (noise, so the pack is as large).
const BIG: usize = 24 << 20;
/// What one call of the lane carries: a third of the big entry and a bit — it is cut inside it
/// at least three times.
const BUDGET: u64 = 7 << 20;
/// Where a restoring box stages the entry it holds part of.
const PARTIAL_DIR: &str = "txlog-restore-partial";
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())
}
/// What the primary holds: small entries on both sides of ONE big pack.
struct Primary {
stores: Arc<ServerStores>,
oids: Vec<Vec<u8>>,
}
fn primary(dir: &Path) -> Primary {
let stores = Arc::new(ServerStores::open(dir).expect("primary stores"));
stores.catalog().create_namespace("ns").expect("namespace");
let repo = stores
.create_repo(
NewRepo::new(RepoRef::new("ns", "app").unwrap()).hash_kind(HashKind::Sha256),
false,
)
.expect("repo");
let store = stores.repo_store(&repo).expect("store");
let git = store.serve();
let hash = store.hash_kind();
let (p1, o1) = one_blob_pack(hash, b"before the big one\n");
let (p2, o2) = one_blob_pack(hash, &noise(7, BIG));
let (p3, o3) = one_blob_pack(hash, b"after the big one\n");
git.put_pack(&p1).expect("pack 1");
git.update_ref("refs/heads/main", None, Some(&o1))
.expect("main");
git.put_pack(&p2).expect("the big pack");
git.update_ref("refs/heads/big", None, Some(&o2))
.expect("big");
git.put_pack(&p3).expect("pack 3");
git.update_ref("refs/heads/after", None, Some(&o3))
.expect("after");
stores
.auth()
.add_principal(Principal::new("alice", "Alice", PrincipalKind::User).unwrap())
.expect("principal");
assert!(
p2.len() >= BIG,
"the big pack is as big as its noise: {}",
p2.len()
);
Primary {
stores,
oids: vec![o1, o2, o3],
}
}
/// A twin of `primary` — a fresh box made a twin, following it until level — in `dir`.
fn twin_of(primary: &Primary, dir: &Path) -> Arc<ServerStores> {
let twin = Arc::new(ServerStores::open(dir).expect("twin stores"));
twin::become_twin(&twin, "primary.test:50051").expect("a fresh box becomes a twin");
drop(twin);
let twin = Arc::new(ServerStores::open(dir).expect("the twin, reopened as one"));
// What a twin over the wire is told on its first poll (an in-process source carries no
// identity): a box restored from it is restored from a NAMED history, as a network
// restore requires.
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");
let f = Follower::new(
Arc::clone(&twin),
Box::new(Arc::clone(primary.stores.txlog())),
);
let head = primary.stores.txlog().head_seq();
for _ in 0..64 {
match f.step().expect("the twin follows") {
Progress::CaughtUp { .. } => break,
Progress::Applied { .. } => continue,
other => panic!("the twin cannot follow: {other:?}"),
}
}
assert_eq!(
twin.txlog().head_seq(),
head,
"the twin is level with the primary"
);
twin
}
/// The sequence of the big pack's frame in a log.
fn big_seq(log: &gunnar_core::txlog::Log) -> u64 {
(log.first_seq()..=log.head_seq())
.find(|s| log.read(*s).map(|e| e.blob.len() >= BIG).unwrap_or(false))
.expect("the big frame")
}
/// 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 (tonic reads it as a clean end, or an error).
Abort,
/// An old twin: no `from_offset` asked or honoured, no `offset` on the chunks; closed at
/// the budget.
OldTwinClose,
}
/// What one call did.
#[derive(Debug)]
enum Call {
/// The box answered the close: restored to this head.
Done(u64),
/// The budget ended the call; what the box said, if it said anything.
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 from = if st.restoring {
head.first_seq.max(st.log_head_seq + 1)
} else {
head.first_seq.max(1)
};
let old_twin = matches!(cut, Cut::OldTwinClose);
let offset = if st.restoring && st.restore_partial_seq == from && !old_twin {
st.restore_partial_bytes
} else {
0
};
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 (mut cursor, mut sent, mut highest) = (from, 0u64, 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);
if old_twin {
chunk.offset = None;
chunk.entry_crc = None;
}
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 match cut {
Cut::Close | Cut::OldTwinClose => {
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())
}
};
}
}
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()),
}
})
}
}
/// Every frame of `restored` is `twin`'s, byte for byte.
fn assert_frames_equal(twin: &ServerStores, restored: &ServerStores) {
let (t, r) = (twin.txlog(), restored.txlog());
assert_eq!(
r.head_seq(),
t.head_seq(),
"the restored main is at the twin's head"
);
for seq in t.first_seq()..=t.head_seq() {
assert_eq!(
r.read(seq).unwrap(),
t.read(seq).unwrap(),
"seq {seq} differs between the twin's log and the restored main's"
);
}
}
/// The restored main, restarted as the primary it was, holds the primary's objects and refs.
fn assert_same_objects(primary: &Primary, restored: &ServerStores) {
let at = RepoRef::new("ns", "app").unwrap();
let p = primary
.stores
.repo_store(&primary.stores.catalog().get(&at).unwrap().unwrap())
.unwrap();
let r = restored
.repo_store(
&restored
.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 restored 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");
}
/// ★ **The defect, and its cure.** A 24 MiB entry, a lane that carries 7 MiB per call, calls that
/// END cleanly at their budget (the guest's own deadline): the entry is cut inside at least three
/// times, each call resumes it at the byte the box holds durably, and the restore finishes. The
/// staging goes the moment the entry is applied — nothing is left on the volume — and the
/// restored main is the twin's log frame for frame and the primary's objects and refs.
///
/// RED before the fix: the box dropped the half entry every call and the lane asked for it again
/// from byte 0 — 12 calls, never past the big entry.
#[test]
fn an_entry_larger_than_one_call_is_cut_three_times_and_resumed_inside_it() {
let (pd, td, md) = (
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
);
let primary = primary(pd.path());
let twin = twin_of(&primary, td.path());
let big = big_seq(twin.txlog());
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 mut inside = Vec::new();
let mut done = None;
for call in 0..12 {
let said = lane.call(main_served.addr, Cut::Close);
let st = lane.status(main_served.addr);
println!(
"call {call}: {said:?} — head {} restoring {} partial {}:{}",
st.log_head_seq, st.restoring, st.restore_partial_seq, st.restore_partial_bytes
);
if st.log_head_seq >= big {
assert!(
!md.path().join(PARTIAL_DIR).exists(),
"call {call}: the big entry is applied and its staging is still on the volume"
);
} else if st.log_head_seq + 1 == big {
inside.push(st.restore_partial_bytes);
}
if let Call::Done(head) = said {
done = Some(head);
break;
}
}
assert_eq!(
done,
Some(twin.txlog().head_seq()),
"the restore never finished; the box held, call after call, {inside:?} bytes of seq {big}"
);
assert!(
inside.len() >= 3,
"the big entry was cut inside {} time(s), not three: {inside:?}",
inside.len()
);
assert!(
inside.windows(2).all(|w| w[1] > w[0]),
"every call went further INTO the entry: {inside:?}"
);
assert!(
!md.path().join(PARTIAL_DIR).exists(),
"nothing staged is left on the volume"
);
assert_frames_equal(&twin, &main);
main_served.kill();
drop(main);
let restored = ServerStores::open(md.path()).expect("the restored main, restarted");
assert_same_objects(&primary, &restored);
twin_served.kill();
}
/// ★ **A main restarted in the middle of the entry resumes from what it made durable.** The
/// calls are CANCELLED at their budget (the host's deadline, not the guest's close); after two
/// of them the main goes down — and, as a crash would, its staged file carries bytes past what
/// its record counts. It comes back, says the same durable count it said before, and the lane
/// finishes the entry from there; the result is the twin's log and the primary's objects.
#[test]
fn a_main_restarted_inside_the_entry_resumes_from_what_it_made_durable() {
let (pd, td, md) = (
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
);
let primary = primary(pd.path());
let twin = twin_of(&primary, td.path());
let big = big_seq(twin.txlog());
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 = Vec::new();
for _ in 0..2 {
lane.call(main_served.addr, Cut::Abort);
// The cancelled stream's handler commits as it ends; give it the moment it needs.
let st = wait(
"the cancelled stream's handler let go",
Duration::from_secs(20),
|| {
let st = lane.status(main_served.addr);
(st.restore_partial_seq == 0
|| st.restore_partial_bytes > held.last().copied().unwrap_or(0))
.then_some(st)
},
);
held.push(st.restore_partial_bytes);
}
let before = lane.status(main_served.addr);
// The main goes down mid-entry...
main_served.kill();
drop(main);
// ...and, as after a power cut, its staged file holds a tail nobody counted.
let data = md.path().join(PARTIAL_DIR).join("entry.blob");
if data.exists() {
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_partial_seq, after.restore_partial_bytes),
(before.restore_partial_seq, before.restore_partial_bytes),
"it holds what it made durable — no more, no less"
);
let mut done = None;
for _ in 0..12 {
if let Call::Done(head) = lane.call(main_served.addr, Cut::Close) {
done = Some(head);
break;
}
}
assert_eq!(
done,
Some(twin.txlog().head_seq()),
"the restore never finished after the restart (held {held:?} of seq {big} before it)"
);
assert!(
!md.path().join(PARTIAL_DIR).exists(),
"nothing staged is left on the volume"
);
assert_frames_equal(&twin, &main);
main_served.kill();
drop(main);
let restored = ServerStores::open(md.path()).expect("the restored main, restarted");
assert_same_objects(&primary, &restored);
twin_served.kill();
}
/// **An older twin cannot carry an entry that needs more than one call, and the box says so by
/// name** instead of asking for the same resend for ever. The old twin ignores `from_offset` and
/// sends the entry from byte 0 with no `offset` on its chunks; the box skips what it holds, which
/// makes the stream no shorter: once a stream of such a sender ends without one byte past what was
/// held, the refusal is `restore-twin-too-old` — "upgrade the twin" — and nothing held is lost.
#[test]
fn an_old_twin_that_cannot_resume_inside_an_entry_is_named_not_looped_on() {
let (pd, td, md) = (
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
);
let primary = primary(pd.path());
let twin = twin_of(&primary, td.path());
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 first = lane.call(main_served.addr, Cut::OldTwinClose);
let mut held = lane.status(main_served.addr).restore_partial_bytes;
assert!(
held > 0,
"the first call staged part of the big entry: {first:?}"
);
// Each resend from byte 0 ends where the last did (the budget is the same): at the latest
// the second one gains nothing, and is named.
let mut named = None;
for _ in 0..3 {
let Call::Stopped(said) = lane.call(main_served.addr, Cut::OldTwinClose) else {
panic!("an old twin cannot finish the big entry")
};
let now = lane.status(main_served.addr).restore_partial_bytes;
assert!(now >= held, "nothing held is lost: {now} < {held}");
held = now;
if said.contains("restore-twin-too-old") {
named = Some(said);
break;
}
}
let said = named.expect("the box never named the old twin");
assert!(
said.contains("upgrade the twin"),
"the box says what to do: {said}"
);
// The twin upgraded: the same box finishes from what it held.
let mut done = None;
for _ in 0..12 {
if let Call::Done(head) = lane.call(main_served.addr, Cut::Close) {
done = Some(head);
break;
}
}
assert_eq!(done, Some(twin.txlog().head_seq()));
assert_frames_equal(&twin, &main);
main_served.kill();
twin_served.kill();
}
/// **Bytes put together across calls are applied only if they ARE the entry.** One staged byte
/// flips on the main's disk between two calls: the last piece's crc (the twin's own payload crc,
/// checked before it left) does not match what the box assembled, nothing is applied, the
/// staging is dropped, and the entry starts again from byte 0 — and then finishes, the twin's
/// log frame for frame.
#[test]
fn a_staged_entry_whose_bytes_rotted_between_calls_is_refused_by_its_crc_and_taken_again() {
let (pd, td, md) = (
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
);
let primary = primary(pd.path());
let twin = twin_of(&primary, td.path());
let big = big_seq(twin.txlog());
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 st = lane.status(main_served.addr);
assert_eq!(
st.restore_partial_seq, big,
"the first call staged part of the big entry"
);
{
use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
let mut f = std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(md.path().join(PARTIAL_DIR).join("entry.blob"))
.unwrap();
f.seek(SeekFrom::Start(1000)).unwrap();
let mut b = [0u8; 1];
f.read_exact(&mut b).unwrap();
f.seek(SeekFrom::Start(1000)).unwrap();
f.write_all(&[b[0] ^ 0x5A]).unwrap();
}
let mut refused = None;
let mut done = None;
for _ in 0..16 {
match lane.call(main_served.addr, Cut::Close) {
Call::Done(head) => {
done = Some(head);
break;
}
Call::Stopped(said) if said.contains("crc") => refused = Some(said),
Call::Stopped(_) => {}
}
}
let refused = refused.expect("the rotted entry was refused by its crc");
assert!(refused.contains("starts again from byte 0"), "{refused}");
assert_eq!(
done,
Some(twin.txlog().head_seq()),
"taken again from byte 0, the entry finished"
);
assert_frames_equal(&twin, &main);
main_served.kill();
twin_served.kill();
}