crates/gunnar-server/tests/twin_crash_mid_group.rs
//! **A twin killed in the middle of a group commit loses nothing it acknowledged and applies
//! nothing twice** (perf twin-apply-fsync, 2026-10-06).
//!
//! The group commit makes a twin write a run of entries into its log with ONE fdatasync and take
//! their control-plane writes into ONE redb commit (`twin::take_group`). Its acknowledgement — the
//! head it answers the primary with — moves only after the log's fsync. Two windows are new and
//! both are killed here, in a CHILD process that dies by `abort()` (no destructor, no flush, the
//! page cache of a process that is gone):
//!
//! * `mid_group` — entries written into the log, not yet fsynced: the group's tail may be lost,
//! nothing the twin acknowledged may be;
//! * `log_ahead` — the group durable in the log, its redb commit not yet made: the twin comes back
//! with its log AHEAD of its control store and must re-take exactly the entries past its applied
//! marker, onto exactly the state the marker vouches for.
//!
//! After the kills a third child follows to the primary's head, and the parent judges the twin
//! against the primary: every document pointer equal, every document's history exactly as long as
//! the primary's (plus what the fresh box held before it became a twin) — a compare-and-swap
//! applied twice would add a row — every secret equal, and the marker at the head.
use std::io::Write as _;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
use gunnar_core::txlog::Entry;
use gunnar_core::{HashKind, NewRepo, Principal, PrincipalKind, RepoRef, TWIN_APPLIED};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, FrameSink, LocalSource, LogSource, Progress};
use gunnar_store::conformance::pack::PackBuilder;
const ACCOUNTS: usize = 40;
const INNER: &str = "the_twin_child_inner";
const ENV_ROOT: &str = "TWIN_CRASH_ROOT";
const ENV_CRASH: &str = "TWIN_CRASH_AT";
/// `mid_group` dies after this many entries reached the twin.
const MID_GROUP_AFTER: u64 = 500;
fn key_line() -> String {
let key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
.unwrap();
key.public_key().to_openssh().unwrap()
}
fn push(stores: &ServerStores, repo: &RepoRef, content: &[u8]) {
let entry = stores.catalog().get(repo).unwrap().unwrap();
let store = stores.repo_store(&entry).expect("store");
let hash = store.hash_kind();
let id = gix_object::compute_hash(hash, gix_object::Kind::Blob, content).expect("hash");
let mut pb = PackBuilder::new(hash);
pb.object(gix_object::Kind::Blob, content.to_vec());
store
.serve()
.put_pack(&pb.build().expect("pack").bytes)
.expect("pack");
let oid = id.as_bytes().to_vec();
store
.serve()
.update_ref("refs/heads/main", None, Some(&oid))
.expect("ref");
}
/// A source that dies after `after` entries reached the twin's sink.
struct DieAfter {
inner: LocalSource,
after: u64,
seen: AtomicU64,
}
struct Counting<'a> {
into: &'a mut dyn FrameSink,
after: u64,
seen: &'a AtomicU64,
}
impl FrameSink for Counting<'_> {
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
self.into.begin(seq, kind, meta, blob_len)
}
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
self.into.piece(bytes)
}
fn end(&mut self) -> anyhow::Result<()> {
self.into.end()?;
if self.seen.fetch_add(1, Ordering::SeqCst) + 1 == self.after {
println!("CRASH mid_group after {} entries", self.after);
let _ = std::io::stdout().flush();
std::process::abort();
}
Ok(())
}
}
impl LogSource for DieAfter {
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.inner.head()
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
self.inner.fetch(from, to, max_bytes)
}
fn fetch_into(
&self,
from: u64,
to: u64,
max_bytes: u64,
into: &mut dyn FrameSink,
) -> anyhow::Result<()> {
let mut c = Counting {
into,
after: self.after,
seen: &self.seen,
};
self.inner.fetch_into(from, to, max_bytes, &mut c)
}
}
/// The child: follow the primary, saying every acknowledged head (`ACK n`) as it moves, and die
/// where `TWIN_CRASH_AT` says.
#[test]
#[ignore = "run as a child by a_twin_killed_mid_group_loses_nothing_it_acknowledged_and_applies_nothing_twice"]
fn the_twin_child_inner() {
let root = std::path::PathBuf::from(std::env::var(ENV_ROOT).expect("run by the outer test"));
let crash = std::env::var(ENV_CRASH).unwrap_or_default();
let primary = Arc::new(ServerStores::open(root.join("primary")).unwrap());
let tw = Arc::new(ServerStores::open(root.join("twin")).unwrap());
assert!(tw.is_twin());
let start = tw.txlog().head_seq();
{
let tw = Arc::clone(&tw);
let crash = crash.clone();
std::thread::spawn(move || {
let mut said = u64::MAX;
loop {
let h = tw.txlog().head_seq();
if h != said {
println!("ACK {h}");
let _ = std::io::stdout().flush();
said = h;
}
if crash == "log_ahead"
&& h > start
&& tw.control_db().mark(TWIN_APPLIED).unwrap().unwrap_or(0) < h
{
println!("CRASH log_ahead at head {h}");
let _ = std::io::stdout().flush();
std::process::abort();
}
std::thread::sleep(Duration::from_micros(50));
}
});
}
let local = LocalSource {
stores: Arc::clone(&primary),
retained_from: 0,
};
let f = match crash.as_str() {
// Small fetches: groups are acknowledged one after another before the kill.
"mid_group" => Follower::new(
Arc::clone(&tw),
Box::new(DieAfter {
inner: local,
after: MID_GROUP_AFTER,
seen: AtomicU64::new(0),
}),
)
.with_fetch_bytes(32 << 10),
// One fetch: one large group, whose redb take is the window.
_ => Follower::new(Arc::clone(&tw), Box::new(local)),
};
loop {
match f.step().unwrap() {
Progress::CaughtUp { .. } => break,
Progress::Applied { .. } => {}
other => panic!("the twin did not follow: {other:?}"),
}
}
std::thread::sleep(Duration::from_millis(5));
println!("LEVEL {}", tw.txlog().head_seq());
}
fn run_child(root: &std::path::Path, crash: &str) -> (bool, String) {
let out = std::process::Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
INNER,
"--ignored",
"--nocapture",
"--test-threads=1",
])
.env(ENV_ROOT, root)
.env(ENV_CRASH, crash)
.output()
.expect("the child runs");
(
out.status.success(),
format!(
"{}\n{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
),
)
}
fn last_ack(text: &str) -> u64 {
text.lines()
.filter_map(|l| {
l.split_once("ACK ")
.and_then(|(_, v)| v.trim().parse::<u64>().ok())
})
.max()
.unwrap_or(0)
}
/// `(document → pointer, document → history length, secret → value, blob ids)` of a box.
type Shape = (
std::collections::BTreeMap<String, String>,
std::collections::BTreeMap<String, usize>,
std::collections::BTreeMap<String, Option<Vec<u8>>>,
std::collections::BTreeSet<String>,
);
fn shape(stores: &ServerStores) -> Shape {
let x = stores.control_db().export().unwrap();
let docs = x
.docs
.iter()
.map(|(n, id)| (n.clone(), hex::encode(id.as_bytes())))
.collect();
let hist = x
.docs
.keys()
.map(|n| {
(
n.clone(),
stores.raw_control_store().history(n).unwrap().len(),
)
})
.collect();
let blobs = x
.blobs
.keys()
.map(|id| hex::encode(id.as_bytes()))
.collect();
(docs, hist, x.secrets, blobs)
}
#[test]
fn a_twin_killed_mid_group_loses_nothing_it_acknowledged_and_applies_nothing_twice() {
let base = option_env!("CARGO_TARGET_TMPDIR")
.map(std::path::PathBuf::from)
.unwrap_or_else(std::env::temp_dir);
let dir = tempfile::tempdir_in(&base).unwrap();
let root = dir.path();
// ── the primary: sign-ups (runs of ControlBlob + ControlDoc) and a push each ──
let head = {
let primary = ServerStores::open(root.join("primary")).unwrap();
for i in 0..ACCOUNTS {
let id = format!("u{i:05}");
primary
.auth()
.add_principal(Principal::new(&id, &id, PrincipalKind::User).unwrap())
.unwrap();
primary.auth().register_key(&id, &key_line(), None).unwrap();
primary.catalog().create_namespace(&id).unwrap();
let r = RepoRef::new(&id, "r").unwrap();
primary
.create_repo(NewRepo::new(r.clone()).hash_kind(HashKind::Sha256), false)
.unwrap();
push(&primary, &r, format!("push of {i}\n").as_bytes());
}
primary.txlog().head_seq()
};
assert!(
head > MID_GROUP_AFTER + 200,
"the storm is longer than the first kill: {head}"
);
// ── the twin: a fresh box made a twin; what it held of its own before following ──
let before = {
let tw = ServerStores::open(root.join("twin")).unwrap();
twin::become_twin(&tw, "primary.test:50051").unwrap();
shape(&tw).1
};
// ── kill 1: mid-group, groups acknowledged before it ──
let (ok, text) = run_child(root, "mid_group");
assert!(
!ok && text.contains("CRASH mid_group"),
"the child died mid-group:\n{text}"
);
let acked = last_ack(&text);
let after_kill_1 = {
let tw = ServerStores::open(root.join("twin")).unwrap();
tw.txlog().head_seq()
};
assert!(
acked > 0,
"groups were acknowledged before the kill:\n{text}"
);
assert!(
after_kill_1 >= acked,
"the twin acknowledged {acked} and came back holding {after_kill_1}"
);
assert!(after_kill_1 <= MID_GROUP_AFTER, "the twin holds no more than reached it (head {after_kill_1} after {MID_GROUP_AFTER} entries in)");
// ── kill 2: the group durable in the log, its redb commit not made ──
let (ok, text) = run_child(root, "log_ahead");
assert!(
!ok && text.contains("CRASH log_ahead"),
"the child died between the log's fsync and redb's commit:\n{text}"
);
let acked = last_ack(&text);
let (after_kill_2, marker) = {
let tw = ServerStores::open(root.join("twin")).unwrap();
(
tw.txlog().head_seq(),
tw.control_db().mark(TWIN_APPLIED).unwrap().unwrap_or(0),
)
};
assert!(
after_kill_2 >= acked,
"the twin acknowledged {acked} and came back holding {after_kill_2}"
);
assert!(marker < after_kill_2, "the kill landed with the log ahead of the control store (marker {marker}, head {after_kill_2})");
// ── the twin follows to the head ──
let (ok, text) = run_child(root, "none");
assert!(ok, "the third child follows to the head:\n{text}");
assert!(
text.contains(&format!("LEVEL {head}")),
"level at {head}:\n{text}"
);
// ── judged against the primary ──
let primary = ServerStores::open(root.join("primary")).unwrap();
let tw = ServerStores::open(root.join("twin")).unwrap();
let (p_docs, p_hist, p_secrets, p_blobs) = shape(&primary);
let (t_docs, t_hist, t_secrets, t_blobs) = shape(&tw);
let missing: Vec<_> = p_blobs.difference(&t_blobs).collect();
assert!(
missing.is_empty(),
"every control blob of the primary is on the twin; missing {}",
missing.len()
);
for (name, ptr) in &p_docs {
assert_eq!(
t_docs.get(name),
Some(ptr),
"document {name}: the twin's pointer is the primary's"
);
let own = before.get(name).copied().unwrap_or(0);
assert_eq!(t_hist.get(name).copied().unwrap_or(0), own + p_hist[name], "document {name}: every version once — the twin held {own} of its own, the primary {} — nothing applied twice, nothing missing", p_hist[name]);
}
for (key, value) in &p_secrets {
assert_eq!(t_secrets.get(key), Some(value), "secret {key}");
}
assert_eq!(tw.txlog().head_seq(), head);
let marked = tw.control_db().mark(TWIN_APPLIED).unwrap().unwrap_or(0);
assert!(marked <= head && marked + twin::MARK_EVERY > head, "the applied marker {marked} is at most MARK_EVERY behind the head {head} (a run of data frames moves no redb page)");
eprintln!(
"killed mid-group (came back at {after_kill_1} of {MID_GROUP_AFTER} in) and between the log's fsync and redb's commit (log {after_kill_2}, marker {marker}); followed to {head}: {} documents ({} versions) and {} blobs equal, every version exactly once",
p_docs.len(),
p_hist.values().sum::<usize>(),
p_blobs.len()
);
}