crates/gunnar-server/tests/twin_snapshot_stall.rs
//! **A twin whose primary goes silent in the middle of a snapshot is cut loose
//! within its bound, re-dials, and resumes** (r5.2; found by the r5.1
//! integration test: a twin whose primary died or restarted during
//! `Mirror.SnapshotFetch` hung for minutes on a half-open connection — the
//! follower's channel had no HTTP/2 keepalive and the snapshot stream no idle
//! bound, unlike `LogHead` and `LogFetch`).
//!
//! A real primary (a control plane on loopback, its log retained past the
//! start so a fresh twin needs a snapshot) behind a TCP relay that this test
//! FREEZES once the snapshot's bytes begin to flow: the relay keeps both
//! sockets open and passes nothing more — the half-open connection. The twin
//! follows through the relay: its step must fail as STALLED within
//! [`twin::SNAPSHOT_IDLE_WITHIN`] (its poll loop then hands back for a
//! re-dial), and a source dialled afresh to the primary installs the
//! snapshot. Its own test binary: the relay's threads are this file's.
use std::io::{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, Ended, Follower, GrpcSource, Progress};
use gunnar_server::CoreControlPlane;
fn free_addr() -> SocketAddr {
let l = TcpListener::bind("127.0.0.1:0").expect("bind");
l.local_addr().expect("addr")
}
/// **The relay**: every connection to it is relayed to `to`; once the
/// primary-to-twin direction has carried `budget` bytes in all, nothing more
/// is passed that way — both sockets stay open (a half-open connection, as a
/// primary that died under its tunnel looks to the twin).
struct Relay {
at: SocketAddr,
passed: Arc<AtomicU64>,
budget: Arc<AtomicU64>,
}
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 (p, b) = (Arc::clone(&passed), Arc::clone(&budget));
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) =
(primary_side, twin_side, Arc::clone(&p), Arc::clone(&b));
std::thread::spawn(move || {
let mut buf = [0u8; 512];
loop {
let room = b
.load(Ordering::SeqCst)
.saturating_sub(p.load(Ordering::SeqCst));
if room == 0 {
// Frozen: hold both sockets open, pass nothing.
std::thread::sleep(Duration::from_millis(50));
continue;
}
let want = buf.len().min(room.min(usize::MAX as u64) as usize);
match pr_r.read(&mut buf[..want]) {
Ok(0) | Err(_) => break,
Ok(n) => {
p.fetch_add(n as u64, Ordering::SeqCst);
if tw_w.write_all(&buf[..n]).is_err() {
break;
}
}
}
}
});
}
});
Relay { at, passed, budget }
}
#[test]
fn a_twin_whose_primary_goes_silent_mid_snapshot_is_cut_redials_and_resumes() {
// ── 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 = free_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.clone(), 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 relay freezes 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 stop = Arc::new(AtomicBool::new(false));
let s2 = Arc::clone(&stop);
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let t = Instant::now();
let ended = follower.run(Duration::from_millis(100), &s2);
let _ = tx.send((ended, t.elapsed()));
});
let bound = twin::SNAPSHOT_IDLE_WITHIN + Duration::from_secs(10);
let got = rx.recv_timeout(bound);
stop.store(true, Ordering::SeqCst);
let (ended, took) = got.unwrap_or_else(|_| {
panic!(
"the twin was still waiting on a silent snapshot stream {} s later",
bound.as_secs()
)
});
assert!(
r.passed.load(Ordering::SeqCst) > 2 * one_head,
"the relay froze inside the snapshot, not before it"
);
assert_eq!(
ended,
Ended::Redial,
"the silent connection is cut and re-dialled"
);
assert!(took < bound, "{took:?}");
// ── the primary is back: a source dialled afresh installs the snapshot ──
let again = GrpcSource::connect(&addr.to_string(), key, twin_stores.root()).expect("source");
let follower = Follower::new(Arc::clone(&twin_stores), Box::new(again));
let mut progress = follower.step().expect("the twin follows the primary again");
let t = Instant::now();
while !matches!(
progress,
Progress::SnapshotInstalled { .. } | Progress::CaughtUp { .. }
) && t.elapsed() < Duration::from_secs(60)
{
progress = follower.step().expect("following");
}
assert!(
twin_stores.txlog().head_seq() >= head
|| matches!(progress, Progress::SnapshotInstalled { .. }),
"{progress:?}"
);
assert!(
twin_stores
.catalog()
.namespace("ns399")
.expect("catalog")
.is_some(),
"the snapshot carried the primary's state"
);
}