crates/gunnar-server/tests/txlog_retention.rs
//! **The default retention, on the real plane** (h-txlog-2x, 2026-09-28):
//! `gunnar_grpc::retention::pass` over a `CoreControlPlane` and a real
//! transaction log — trims behind the slowest reporting twin minus the window,
//! and trims nothing with no twin or a silent one.
use std::sync::Arc;
use gunnar_core::{NewRepo, RepoRef};
use gunnar_grpc::heartbeat::{Bounds, Registry};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_proto::TwinReport;
use gunnar_server::stores::ServerStores;
use gunnar_server::CoreControlPlane;
fn primary() -> (tempfile::TempDir, Arc<ServerStores>) {
let pd = tempfile::tempdir().unwrap();
let primary = Arc::new(ServerStores::open(pd.path()).unwrap());
for ns in ["a", "b", "c", "d", "e"] {
primary.catalog().create_namespace(ns).unwrap();
primary
.create_repo(NewRepo::new(RepoRef::new(ns, "app").unwrap()), false)
.unwrap();
}
assert!(
primary.txlog().head_seq() >= 10,
"a log long enough to trim"
);
(pd, primary)
}
fn ctx(stores: &Arc<ServerStores>, heartbeats: &Arc<Registry>) -> ServiceContext {
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(stores)));
ServiceContext::new(plane, Arc::new(EventBus::new(16)), ServerConfig::default())
.with_heartbeats(Arc::clone(heartbeats))
}
fn now() -> i64 {
estate_clock::now_unix_ms()
}
#[test]
fn a_primary_with_a_reporting_twin_trims_behind_it_and_keeps_the_window() {
let (_pd, stores) = primary();
let reg = Arc::new(Registry::new(Bounds::from_env()));
let c = ctx(&stores, ®);
let head = stores.txlog().head_seq();
assert_eq!(stores.txlog().first_seq(), 1);
let why =
gunnar_grpc::retention::pass(&c, 2).expect_err("no twin has reported: nothing is trimmed");
assert!(why.contains("no twin"), "{why}");
assert_eq!(stores.txlog().first_seq(), 1, "untouched");
// The log is kept in SEGMENTS: the first pass seals the one it writes and drops
// nothing that still holds a kept entry; the next pass, after more writes, drops
// the whole sealed segment below the point.
reg.record(
"gunnar-twin",
"10.13.8.3:40000",
&TwinReport {
applied_seq: head,
..TwinReport::default()
},
head,
now(),
);
gunnar_grpc::retention::pass(&c, 2).expect("a reporting twin: the pass retains behind it");
for ns in ["f", "g", "h", "i", "j"] {
stores.catalog().create_namespace(ns).unwrap();
stores
.create_repo(NewRepo::new(RepoRef::new(ns, "app").unwrap()), false)
.unwrap();
}
let head2 = stores.txlog().head_seq();
assert!(head2 > head + 2);
reg.record(
"gunnar-twin",
"10.13.8.3:40000",
&TwinReport {
applied_seq: head2,
..TwinReport::default()
},
head2,
now(),
);
let line = gunnar_grpc::retention::pass(&c, 2)
.expect("a reporting twin: the log is trimmed behind it");
let first = stores.txlog().first_seq();
assert!(
first > 1,
"the sealed segment below the twin is gone: {line}"
);
assert!(
first <= head2 + 1 - 2,
"the window of 2 applied entries is kept: first {first}, head {head2}: {line}"
);
assert_eq!(stores.txlog().head_seq(), head2, "the head is where it was");
assert!(
stores.txlog().range(first, head2).is_ok(),
"everything from the new start is readable"
);
assert!(line.contains("trimmed"), "{line}");
}
#[test]
fn a_silent_twin_stops_the_retention() {
let (_pd, stores) = primary();
let reg = Arc::new(Registry::new(Bounds::from_env()));
let c = ctx(&stores, ®);
let head = stores.txlog().head_seq();
// Heard once, an hour ago: silent now.
reg.record(
"gunnar-twin",
"10.13.8.3:40000",
&TwinReport {
applied_seq: head,
..TwinReport::default()
},
head,
now() - 3_600_000,
);
let why = gunnar_grpc::retention::pass(&c, 0).expect_err("a silent twin: nothing is trimmed");
assert!(why.contains("silent"), "{why}");
assert_eq!(stores.txlog().first_seq(), 1);
}
// ── cov-13 (2026-10-06): the log is bounded in BYTES ────────────────────────
use gunnar_core::txlog::{Kind, Meta};
use gunnar_grpc::retention::{pass_with, Policy};
const MIB: u64 = 1 << 20;
/// One push's pack frame of `len` bytes straight into the log (the retention reads the log,
/// not the stores).
fn push_frame(stores: &ServerStores, len: usize) -> u64 {
stores
.txlog()
.append(
Kind::Pack,
&Meta::Pack {
repo: "a/app".into(),
hash: "sha1".into(),
},
&vec![7u8; len],
)
.unwrap()
}
/// ★ **Many LARGE entries keep the log within its byte budget behind a twin that follows.**
/// MEASURED on the UpCloud mockup: 8 × 512 MiB pushes sat in the log (4.3 GB of /data) —
/// the 64-entry window is 64 PUSHES whatever their size. Twelve rounds of two 1 MiB pushes,
/// a following twin acknowledging each, a 3 MiB budget: after every pass the log holds at
/// most the budget, and never more than the budget plus one round between passes.
#[test]
fn many_large_entries_keep_the_log_within_its_byte_budget_behind_a_following_twin() {
let (_pd, stores) = primary();
let reg = Arc::new(Registry::new(Bounds::from_env()));
let c = ctx(&stores, ®);
let budget = 3 * MIB;
let policy = Policy {
window: 64,
budget_percent: 0,
budget_max_bytes: budget,
silent_grace_ms: 600_000,
};
let mut most = 0;
for _ in 0..12 {
push_frame(&stores, MIB as usize);
push_frame(&stores, MIB as usize);
most = most.max(stores.txlog().bytes());
let head = stores.txlog().head_seq();
reg.record(
"gunnar-twin",
"10.13.8.3:40000",
&TwinReport {
applied_seq: head,
..TwinReport::default()
},
head,
now(),
);
let _ = pass_with(&c, &policy, false);
assert!(
stores.txlog().bytes() <= budget,
"after a pass the log is within its {budget}-byte budget: {} bytes, first {} head {}",
stores.txlog().bytes(),
stores.txlog().first_seq(),
head
);
}
assert!(
most <= budget + 2 * MIB + 4096,
"between passes the log grew by one round at most: {most}"
);
assert!(
stores
.txlog()
.range(stores.txlog().first_seq(), stores.txlog().head_seq())
.is_ok(),
"what is kept is readable"
);
}
/// ★ **Only what the twin ACKNOWLEDGED is ever dropped**: a twin that follows but lags keeps
/// everything after its position, whatever the budget says — it catches up from the log.
#[test]
fn the_byte_budget_never_drops_what_a_following_twin_has_not_applied() {
let (_pd, stores) = primary();
let reg = Arc::new(Registry::new(Bounds::from_env()));
let c = ctx(&stores, ®);
let applied = stores.txlog().head_seq();
for _ in 0..6 {
push_frame(&stores, MIB as usize);
}
let head = stores.txlog().head_seq();
reg.record(
"gunnar-twin",
"10.13.8.3:40000",
&TwinReport {
applied_seq: applied,
..TwinReport::default()
},
head,
now(),
);
let _ = pass_with(
&c,
&Policy {
window: 64,
budget_percent: 0,
budget_max_bytes: MIB,
silent_grace_ms: 0,
},
false,
);
assert!(
stores.txlog().first_seq() <= applied + 1,
"the lagging twin's next entry is still in the log: first {}, applied {applied}",
stores.txlog().first_seq()
);
assert!(stores.txlog().range(applied + 1, head).is_ok());
}
/// ★ **A silent twin is waited for within its grace — and given up on past it, or at once
/// when the log passes TWICE its budget**: then the log is trimmed to the budget and the
/// twin is owed a full resync (on the primary's disk as well).
#[test]
fn a_silent_twin_is_waited_for_within_its_grace_and_given_up_on_past_it_or_at_twice_the_budget() {
let (pd, stores) = primary();
let reg = Arc::new(Registry::new(Bounds::from_env()));
let c = ctx(&stores, ®);
let applied = stores.txlog().head_seq();
// Heard a minute and a half ago: silent (bound 60 s), well inside a 10-minute grace.
reg.record(
"gunnar-twin",
"10.13.8.3:40000",
&TwinReport {
applied_seq: applied,
..TwinReport::default()
},
applied,
now() - 90_000,
);
let budget = 3 * MIB;
let policy = Policy {
window: 64,
budget_percent: 0,
budget_max_bytes: budget,
silent_grace_ms: 600_000,
};
for _ in 0..4 {
push_frame(&stores, MIB as usize);
}
let why = pass_with(&c, &policy, false)
.expect_err("over the budget, under twice it, within the grace: waited for");
assert!(why.contains("silent"), "{why}");
assert_eq!(stores.txlog().first_seq(), 1, "nothing trimmed");
assert!(
reg.owed()
.twins
.get("gunnar-twin")
.is_none_or(|o| o.resync_from == 0),
"{:?}",
reg.owed()
);
// Past twice the budget: given up on at once.
for _ in 0..4 {
push_frame(&stores, MIB as usize);
}
let line = pass_with(&c, &policy, false)
.expect("past twice the budget: the silent twin is not waited for");
assert!(
stores.txlog().bytes() <= budget,
"trimmed to the budget: {} bytes: {line}",
stores.txlog().bytes()
);
let first = stores.txlog().first_seq();
assert!(first > applied + 1, "{line}");
assert_eq!(
reg.owed().twins.get("gunnar-twin").map(|o| o.resync_from),
Some(first),
"owed a full resync: {:?}",
reg.owed()
);
let on_disk = std::fs::read_to_string(pd.path().join("twin-owed.json"))
.expect("the debt is on the primary's disk");
assert!(
on_disk.contains("gunnar-twin") && on_disk.contains("resync_from"),
"{on_disk}"
);
// Past the grace (and only over the budget): given up on as well.
let (_pd2, stores2) = primary();
let reg2 = Arc::new(Registry::new(Bounds::from_env()));
let c2 = ctx(&stores2, ®2);
let applied2 = stores2.txlog().head_seq();
reg2.record(
"gunnar-twin",
"10.13.8.3:40000",
&TwinReport {
applied_seq: applied2,
..TwinReport::default()
},
applied2,
now() - 3_600_000,
);
for _ in 0..4 {
push_frame(&stores2, MIB as usize);
}
pass_with(&c2, &policy, false).expect("silent an hour, the log over its budget: given up on");
assert!(stores2.txlog().bytes() <= budget);
assert!(
reg2.owed()
.twins
.get("gunnar-twin")
.is_some_and(|o| o.resync_from > 0),
"{:?}",
reg2.owed()
);
}