crates/gunnar-server/tests/twin_snapshot_slow.rs
//! **A primary that is slow between snapshot pieces, but alive, is waited
//! for** (wiring tracer B 2026-10-08). The primary sends nothing while it
//! walks a repository with no reachability index or takes its cut; past the
//! twin's per-piece bound the twin called it a stall, re-dialled and asked a
//! NEW cut — and on a repository that takes that long every time, it never
//! installed. A dead path is found by the channel's HTTP/2 keepalive (a PING
//! every 10 s, dropped after 20 s); the per-piece bound is only for a primary
//! that answers PINGs and still sends nothing for a long time.
//!
//! A real primary behind a relay that understands HTTP/2 framing: once the
//! snapshot's bytes begin to flow it HOLDS every frame from the primary for
//! [`HOLD`] — except PING frames, which pass at once, so the twin's keepalive
//! sees a live peer — and then lets everything through, in order. The twin
//! must install the snapshot over the one connection: no error, no re-dial.
//! Its own test binary: the relay's threads are this file's.
use std::collections::VecDeque;
use std::io::{ErrorKind, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, GrpcSource, Progress};
use gunnar_server::CoreControlPlane;
/// Longer than the bound the twin used to cut at (15 s), well inside the keepalive's view of a
/// live peer.
const HOLD: Duration = Duration::from_secs(20);
/// HTTP/2's PING frame type (RFC 9113 §6.7).
const PING: u8 = 0x6;
struct Relay {
at: SocketAddr,
passed: Arc<AtomicU64>,
budget: Arc<AtomicU64>,
held: Arc<AtomicBool>,
}
/// One frame off `r` (9-byte header + payload), or `None` at the end. A read timeout while
/// nothing of a frame has arrived is `Some(Err(TimedOut))`, so the caller can flush a hold.
fn frame(r: &mut TcpStream, partial: &mut Vec<u8>) -> Option<std::io::Result<Vec<u8>>> {
loop {
let want = if partial.len() < 9 {
9
} else {
9 + ((partial[0] as usize) << 16 | (partial[1] as usize) << 8 | partial[2] as usize)
};
if partial.len() >= 9 && partial.len() == want {
return Some(Ok(std::mem::take(partial)));
}
let mut buf = vec![0u8; want - partial.len()];
match r.read(&mut buf) {
Ok(0) => return None,
Ok(n) => partial.extend_from_slice(&buf[..n]),
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {
return Some(Err(e))
}
Err(_) => return None,
}
}
}
fn relay(to: SocketAddr) -> Relay {
let l = TcpListener::bind("127.0.0.1:0").expect("bind the relay");
let at = l.local_addr().expect("addr");
let passed = Arc::new(AtomicU64::new(0));
let budget = Arc::new(AtomicU64::new(u64::MAX));
let held = Arc::new(AtomicBool::new(false));
let (p, b, h) = (Arc::clone(&passed), Arc::clone(&budget), Arc::clone(&held));
std::thread::spawn(move || {
for twin_side in l.incoming().flatten() {
let Ok(primary_side) = TcpStream::connect(to) else {
continue;
};
let (mut tw_r, mut pr_w) = (
twin_side.try_clone().unwrap(),
primary_side.try_clone().unwrap(),
);
std::thread::spawn(move || {
let mut buf = [0u8; 4096];
while let Ok(n) = tw_r.read(&mut buf) {
if n == 0 || pr_w.write_all(&buf[..n]).is_err() {
break;
}
}
});
let (mut pr_r, mut tw_w, p, b, h) = (
primary_side,
twin_side,
Arc::clone(&p),
Arc::clone(&b),
Arc::clone(&h),
);
std::thread::spawn(move || {
pr_r.set_read_timeout(Some(Duration::from_millis(50)))
.unwrap();
// The connection preface's SETTINGS frame first: no preface on this side.
let mut partial = Vec::new();
let mut queue: VecDeque<Vec<u8>> = VecDeque::new();
let mut until: Option<Instant> = None;
loop {
let next = frame(&mut pr_r, &mut partial);
let holding = until.is_some_and(|u| Instant::now() < u);
if !holding && !queue.is_empty() {
while let Some(f) = queue.pop_front() {
if tw_w.write_all(&f).is_err() {
return;
}
}
}
let f = match next {
None => return,
Some(Err(_)) => continue,
Some(Ok(f)) => f,
};
// Once the snapshot flows, hold the primary's frames ONCE, for HOLD.
if until.is_none()
&& !h.load(Ordering::SeqCst)
&& p.load(Ordering::SeqCst) >= b.load(Ordering::SeqCst)
{
h.store(true, Ordering::SeqCst);
until = Some(Instant::now() + HOLD);
}
let holding = until.is_some_and(|u| Instant::now() < u);
if holding && f[3] != PING {
queue.push_back(f);
continue;
}
p.fetch_add(f.len() as u64, Ordering::SeqCst);
if tw_w.write_all(&f).is_err() {
return;
}
}
});
}
});
Relay {
at,
passed,
budget,
held,
}
}
#[test]
fn a_twin_waits_for_a_slow_primary_that_answers_its_pings() {
// ── the primary: a real control plane on loopback, its log retained past the start ──
let pd = tempfile::tempdir().expect("tempdir");
let primary = Arc::new(ServerStores::open(pd.path()).expect("primary stores"));
for i in 0..400 {
primary
.catalog()
.create_namespace(&format!("ns{i}"))
.expect("a namespace");
}
let head = primary.txlog().head_seq();
primary.txlog().retain_from(head + 1).expect("retain");
assert!(
primary.txlog().first_seq() > 1,
"a fresh twin needs a snapshot"
);
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(&primary)));
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 addr = {
let l = TcpListener::bind("127.0.0.1:0").expect("bind");
l.local_addr().expect("addr")
};
rt.spawn(async move {
let _ = gunnar_grpc::router(ctx, || 0).serve(addr).await;
});
let t0 = Instant::now();
while TcpStream::connect(addr).is_err() {
assert!(
t0.elapsed() < Duration::from_secs(10),
"the primary never listened"
);
std::thread::sleep(Duration::from_millis(20));
}
// ── the twin, through the relay ──
let r = relay(addr);
let td = tempfile::tempdir().expect("tempdir");
let twin_stores = Arc::new(ServerStores::open(td.path()).expect("twin stores"));
twin::become_twin(&twin_stores, &r.at.to_string()).expect("a fresh box becomes a twin");
let key = twin::twin_private_key(twin_stores.root()).expect("the twin's key");
let source = GrpcSource::connect(&r.at.to_string(), key, twin_stores.root()).expect("source");
// One head on this connection, to learn what a head costs on the wire; the step's own head
// is no more, so the hold begins INSIDE the snapshot.
use gunnar_server::twin::LogSource as _;
source
.head()
.expect("the primary answers through the relay");
let one_head = r.passed.load(Ordering::SeqCst);
r.budget.store(2 * one_head + 1024, Ordering::SeqCst);
let follower = Follower::new(Arc::clone(&twin_stores), Box::new(source));
let t = Instant::now();
let mut progress = follower
.step()
.expect("a slow primary that answers its pings is waited for, not cut");
while !matches!(
progress,
Progress::SnapshotInstalled { .. } | Progress::CaughtUp { .. }
) && t.elapsed() < Duration::from_secs(120)
{
progress = follower.step().expect("following over the one connection");
}
assert!(
r.held.load(Ordering::SeqCst),
"the relay held the primary's frames inside the snapshot"
);
assert!(
t.elapsed() >= HOLD,
"the hold was waited out: {:?}",
t.elapsed()
);
assert!(
matches!(progress, Progress::SnapshotInstalled { .. })
|| twin_stores.txlog().head_seq() >= head,
"{progress:?}"
);
assert!(
twin_stores
.catalog()
.namespace("ns399")
.expect("catalog")
.is_some(),
"the snapshot carried the primary's state"
);
}