crates/gunnar-grpc/src/retention.rs
//! **The transaction log's default retention** (h-txlog-2x, 2026-09-28).
//!
//! Every push lands twice on a PRIMARY: once in its stores and once in the
//! transaction log the twin follows (`gunnar-core/src/txlog.rs`; measured 4.6×
//! on a 40 GiB set, `gunnar-growbench open_capacity.rs`). Until now only the
//! operator's `Mirror.LogRetain` trimmed the log, so on every estate nobody
//! trimmed it — and monetize grows the paid volume at 80 % used, so the log was
//! spending the customer's money on bytes the twin had already taken.
//!
//! The default: every [`EVERY`], the primary retains its log from **the slowest
//! reporting twin's applied position, minus a [`DEFAULT_WINDOW`] of entries**
//! (a twin that falls a little behind catches up from the log instead of a
//! snapshot). It NEVER trims:
//!
//! * with no twin enrolled — the log is then the only second copy of history a
//! restore could stream, and nothing says a twin is not about to join;
//! * while any enrolled twin has not reported since this server started, or is
//! `silent` — its position is unknown or stale, and the frames it needs next
//! might be the ones cut (the same rule `Mirror.LogRetain` refuses by);
//! * on a twin (its log IS the primary's numbers and its only copy: "the twin
//! only stores") — the pass is not started there.
//!
//! `GUNNAR_TXLOG_RETAIN=off` turns it off; `GUNNAR_TXLOG_RETAIN_WINDOW=<n>`
//! sets the window. The twin's own log is untouched by this.
//!
//! # ★ The log is bounded in BYTES too (cov-13, 2026-10-06)
//!
//! MEASURED on the UpCloud mockup: with 512 MiB pushes the log held 8 × 512 MiB
//! = 4.3 GB on /data — the 64-entry window is 64 PUSHES, whatever their size,
//! and nothing was trimmed at all while the twin was silent — so every pushed
//! byte took about twice its size on the set and a customer got roughly half
//! the disk before the floor. Now:
//!
//! * **a byte budget** ([`Policy::budget`]): the smaller of
//! [`DEFAULT_BUDGET_PERCENT`] of the data volume and [`DEFAULT_BUDGET_MAX_BYTES`]
//! (`GUNNAR_TXLOG_RETAIN_PERCENT`, `GUNNAR_TXLOG_RETAIN_BYTES`; 0 turns either
//! off). Retention keeps the smaller of the window and the budget behind the
//! slowest twin — only ever what that twin has ACKNOWLEDGED (applied, which a
//! twin reports only after the entry is durable in its own log), and only
//! what this primary's stores already hold (an entry is dropped only in a
//! whole segment sealed before the pass, never the one being written);
//! * **a silent twin is not waited for past its grace**
//! ([`DEFAULT_SILENT_GRACE`], `GUNNAR_TXLOG_SILENT_GRACE_SECS`): while the
//! log is over the budget and a twin has been silent that long — or at once,
//! when the log passes TWICE the budget — retention stops keeping the log for
//! it, trims to the budget, and records that the twin is owed a FULL RESYNC
//! ([`crate::twin_owed`]). It loses nothing by it: when it polls again it
//! finds the primary retaining from past its own head and takes a snapshot
//! (`Mirror.SnapshotFetch`) — the same door any twin behind a retention hole
//! has always taken. A silence shorter than the grace (a busy twin under a
//! storm went quiet for 190 s on the mockup) keeps the log, because a snapshot
//! holds the primary's writes for its whole stream;
//! * **checked every [`CHECK_EVERY`]**, not every ten minutes: a light pass
//! that trims only when a whole segment can go (or sealing the one being
//! written lets the next pass drop it), and the full window pass every
//! [`EVERY`] as before.
//!
//! The same pass keeps the other debt of a silent twin: main's set GREW while
//! it was silent (an emergency grows main alone) — the twin is then refused
//! `LogFetch`/`SnapshotFetch` until its volume holds main's set
//! ([`crate::TWIN_TOO_SMALL`]).
use std::time::{Duration, Instant};
use crate::heartbeat::STATE_SILENT;
use crate::twin_owed::{self, Book};
/// The log target: gunnar-core's `observe::TARGET_TWIN`, spelled here because
/// this crate does not link gunnar-core.
pub const TARGET: &str = "gunnar::twin";
/// How often the pass runs on a primary.
pub const EVERY: Duration = Duration::from_secs(600);
/// Entries kept behind the slowest twin by default.
pub const DEFAULT_WINDOW: u64 = 64;
/// `off` disables the default retention.
pub const RETAIN_ENV: &str = "GUNNAR_TXLOG_RETAIN";
/// The window, in entries.
pub const WINDOW_ENV: &str = "GUNNAR_TXLOG_RETAIN_WINDOW";
/// ★ How often the light, byte-bounded pass looks (cov-13).
pub const CHECK_EVERY: Duration = Duration::from_secs(30);
/// ★ The byte budget's share of the data volume, percent.
pub const DEFAULT_BUDGET_PERCENT: u64 = 5;
/// ★ The byte budget's ceiling, whatever the volume.
pub const DEFAULT_BUDGET_MAX_BYTES: u64 = 4 << 30;
/// ★ How long a silent twin is waited for while the log is over its budget.
pub const DEFAULT_SILENT_GRACE: Duration = Duration::from_secs(600);
/// `GUNNAR_TXLOG_RETAIN_PERCENT=<n>`: the budget's share of the data volume (0 = none).
pub const BUDGET_PERCENT_ENV: &str = "GUNNAR_TXLOG_RETAIN_PERCENT";
/// `GUNNAR_TXLOG_RETAIN_BYTES=<n>[KiB|MiB|GiB]`: the budget's ceiling (0 = none).
pub const BUDGET_BYTES_ENV: &str = "GUNNAR_TXLOG_RETAIN_BYTES";
/// `GUNNAR_TXLOG_SILENT_GRACE_SECS=<n>`: the silent twin's grace.
pub const SILENT_GRACE_ENV: &str = "GUNNAR_TXLOG_SILENT_GRACE_SECS";
/// One enrolled twin as the pass sees it: its principal, what it last reported
/// applied (`None`: not since this server started), gunnar's state word, and —
/// when silent — for how long.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TwinAt {
pub principal: String,
pub applied: Option<u64>,
pub state: String,
/// How long it has been silent, ms; 0 when heard.
pub silent_for_ms: u64,
}
/// ★ **What the retention keeps** (cov-13).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Policy {
/// Entries kept behind the slowest twin.
pub window: u64,
/// The budget's share of the data volume, percent; 0 = none.
pub budget_percent: u64,
/// The budget's ceiling, bytes; 0 = none.
pub budget_max_bytes: u64,
/// How long a silent twin is waited for while the log is over its budget.
pub silent_grace_ms: u64,
}
impl Policy {
/// The window alone, as before the byte bound: no budget, a silent twin waited for for ever.
pub fn window_only(window: u64) -> Self {
Self {
window,
budget_percent: 0,
budget_max_bytes: 0,
silent_grace_ms: u64::MAX,
}
}
/// **The byte budget** over a data volume of `data_total` bytes (`None`: unknown) —
/// the smaller of the share and the ceiling; `None` when neither is set.
pub fn budget(&self, data_total: Option<u64>) -> Option<u64> {
let share = data_total
.filter(|_| self.budget_percent > 0)
.map(|t| (u128::from(t) * u128::from(self.budget_percent) / 100) as u64);
let cap = (self.budget_max_bytes > 0).then_some(self.budget_max_bytes);
match (share, cap) {
(Some(a), Some(b)) => Some(a.min(b)),
(a, b) => a.or(b),
}
}
/// The policy from the environment (`None`: `GUNNAR_TXLOG_RETAIN=off`).
pub fn from_env(
retain: Option<&str>,
window: Option<&str>,
percent: Option<&str>,
bytes: Option<&str>,
grace_secs: Option<&str>,
) -> Option<Self> {
let window = window_from_env(retain, window)?;
Some(Self {
window,
budget_percent: percent
.and_then(|p| p.trim().parse().ok())
.unwrap_or(DEFAULT_BUDGET_PERCENT)
.min(100),
budget_max_bytes: bytes
.and_then(parse_bytes)
.unwrap_or(DEFAULT_BUDGET_MAX_BYTES),
silent_grace_ms: grace_secs
.and_then(|g| g.trim().parse::<u64>().ok())
.map(|g| g.saturating_mul(1000))
.unwrap_or(DEFAULT_SILENT_GRACE.as_millis() as u64),
})
}
}
/// `123`, `64MiB`, `4GiB`, `512KiB` (also `M`/`G`/`K`, `MB`/`GB`) → bytes.
pub fn parse_bytes(v: &str) -> Option<u64> {
let v = v.trim();
let split = v.find(|c: char| !c.is_ascii_digit()).unwrap_or(v.len());
let (n, unit) = v.split_at(split);
let n: u64 = n.parse().ok()?;
let shift = match unit.trim().to_ascii_lowercase().as_str() {
"" | "b" => 0,
"k" | "kb" | "kib" => 10,
"m" | "mb" | "mib" => 20,
"g" | "gb" | "gib" => 30,
_ => return None,
};
n.checked_mul(1u64 << shift)
}
/// ★ **One byte-bounded decision** (cov-13): where to retain from, and which silent twins
/// are given up on (owed a full resync).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Plan {
pub from: u64,
/// Silent twins no longer waited for: they take a snapshot when they poll again.
pub gave_up: Vec<String>,
/// One sentence: why this `from`.
pub why: String,
}
/// ★ **Where the log may be retained from, in bytes as well as entries** — pure.
///
/// `layout` is the log's segments (`(first, last, bytes)`, oldest first, the open one
/// last); `budget` the byte bound (`None`: the window alone, as [`point`]). `full` is the
/// ten-minute pass (any `from` past the start, sealing the open segment as before); the
/// light pass only acts when a whole segment can go, or when sealing the open one lets
/// the next pass drop it while the log is over its budget.
pub fn decide(
layout: &[(u64, u64, u64)],
head: u64,
twins: &[TwinAt],
policy: &Policy,
budget: Option<u64>,
full: bool,
) -> Result<Plan, String> {
if twins.is_empty() {
return Err(
"no twin is enrolled: the log is the only second copy of history, nothing is trimmed"
.to_owned(),
);
}
let first = layout.first().map(|s| s.0).unwrap_or(head + 1);
let bytes: u64 = layout.iter().map(|s| s.2).sum();
let over = budget.is_some_and(|b| bytes > b);
let hard = budget.is_some_and(|b| bytes > b.saturating_mul(2));
let mut gave_up = Vec::new();
let mut slowest: Option<(&str, u64)> = None;
for t in twins {
let silent = t.applied.is_none() || t.state == STATE_SILENT;
if silent {
// Given up on: over the budget, and silent past the grace — or the log at twice it.
if over && (t.silent_for_ms >= policy.silent_grace_ms || hard) {
gave_up.push(t.principal.clone());
continue;
}
return Err(match t.applied {
None => format!("twin '{}' has not reported since this server started: its position is unknown, nothing is trimmed", t.principal),
Some(applied) => format!(
"twin '{}' is silent (last applied {applied}): nothing is trimmed while a twin may need what is cut{}",
t.principal,
match budget {
Some(b) if over => format!(" (the log is {bytes} bytes, over its {b}-byte budget; it is given up on after {} s of silence, or at twice the budget)", policy.silent_grace_ms / 1000),
Some(b) => format!(" (the log is {bytes} bytes, within its {b}-byte budget)"),
None => String::new(),
}
),
});
}
let applied = t.applied.unwrap_or(0);
// A twin already behind the retention hole takes a snapshot whatever is kept.
if applied + 1 < first {
continue;
}
if slowest.is_none_or(|(_, a)| applied < a) {
slowest = Some((&t.principal, applied));
}
}
// Never past this box's own head, nor past what the slowest heard twin applied.
let acked = slowest.map(|(_, a)| a.min(head)).unwrap_or(head);
let from_window = (acked + 1).saturating_sub(policy.window).max(1);
let mut from = from_window;
let mut why = match slowest {
Some((p, a)) => format!(
"the slowest twin '{p}' applied {a}, less a window of {} entries",
policy.window
),
None => "no twin that is heard holds anything back".to_owned(),
};
if let Some(b) = budget {
// The newest whole segments that fit the budget (the open one is sealed by the trim).
let mut keep_from = head + 1;
let mut kept = 0u64;
for (sf, _, sb) in layout.iter().rev() {
if kept + sb > b {
break;
}
kept += sb;
keep_from = *sf;
}
let keep_from = keep_from.min(acked + 1);
if keep_from > from {
from = keep_from;
why = format!("{why}; the log is {bytes} bytes and its budget {b}: only what fits it is kept, and only what that twin acknowledged goes");
}
}
if !gave_up.is_empty() {
why = format!("{why}; NOT waited for any more (silent past the grace with the log over its budget — owed a full resync): {}", gave_up.join(", "));
}
if from <= first {
return Err(format!(
"nothing to trim: the log starts at {first} and {why} keeps it from {from}"
));
}
if !full {
let droppable = layout.iter().any(|(f, l, _)| l >= f && *l < from);
let seal_helps = over && layout.last().is_some_and(|(f, l, _)| l >= f && *f < from);
if !droppable && !seal_helps {
return Err(format!(
"nothing to trim yet: no whole segment lies below {from} ({why})"
));
}
}
Ok(Plan { from, gave_up, why })
}
/// **Where the log may be retained from**, pure — `Ok(from)` with
/// `from > first`, or `Err(why nothing is trimmed)`.
pub fn point(first: u64, head: u64, twins: &[TwinAt], window: u64) -> Result<u64, String> {
if twins.is_empty() {
return Err(
"no twin is enrolled: the log is the only second copy of history, nothing is trimmed"
.to_owned(),
);
}
let mut slowest: Option<(&str, u64)> = None;
for t in twins {
let Some(applied) = t.applied else {
return Err(format!("twin '{}' has not reported since this server started: its position is unknown, nothing is trimmed", t.principal));
};
if t.state == STATE_SILENT {
return Err(format!("twin '{}' is silent (last applied {applied}): nothing is trimmed while a twin may need what is cut", t.principal));
}
if slowest.is_none_or(|(_, a)| applied < a) {
slowest = Some((&t.principal, applied));
}
}
let (_, applied) = slowest.expect("at least one twin");
// Never past this box's own head (a twin reporting beyond it is a restart race),
// and `window` entries the slowest twin HAS applied stay (it can fall back that
// far and still catch up from the log instead of a snapshot).
let from = (applied.min(head) + 1).saturating_sub(window).max(1);
if from <= first {
return Err(format!("nothing to trim: the log starts at {first} and the slowest twin at {applied} minus the window of {window} keeps it from {from}"));
}
Ok(from)
}
/// The window and the switch, read from the environment (`None`: off).
pub fn window_from_env(retain: Option<&str>, window: Option<&str>) -> Option<u64> {
if retain
.map(str::trim)
.is_some_and(|v| v.eq_ignore_ascii_case("off"))
{
return None;
}
Some(
window
.and_then(|w| w.trim().parse().ok())
.unwrap_or(DEFAULT_WINDOW),
)
}
/// **One pass**: read the enrolled twins and the log's head, and retain from
/// [`point`] when there is one — the window alone ([`Policy::window_only`]). Answers the
/// log line it wrote.
pub fn pass(ctx: &crate::ServiceContext, window: u64) -> Result<String, String> {
pass_with(ctx, &Policy::window_only(window), true)
}
/// ★ **The silence and growth debts** (cov-13): a twin first seen silent is noted with
/// main's volume then; main's volume grown past it while the twin is still silent is a
/// growth it is owed ([`twin_owed`]); a twin heard again settles what it has paid.
/// Answers the lines to say.
pub fn note_debts(
book: &mut Book,
rows: &[gunnar_proto::TwinHeartbeat],
reported: &dyn Fn(&str) -> bool,
data_total: Option<u64>,
first_seq: u64,
now_ms: i64,
) -> Vec<String> {
let mut said = Vec::new();
for r in rows {
let o = book.twins.entry(r.principal.clone()).or_default();
let heard = r.state != STATE_SILENT && reported(&r.principal);
if !heard {
if o.silent_since_ms == 0 {
o.silent_since_ms = if r.last_seen_unix_ms > 0 {
r.last_seen_unix_ms
} else {
now_ms
};
o.main_total_at_silence = data_total.unwrap_or(0);
}
if o.main_total_at_silence == 0 {
o.main_total_at_silence = data_total.unwrap_or(0);
}
if r.disk_total_bytes > 0 {
o.twin_total = r.disk_total_bytes;
}
if let Some(total) = data_total {
let grew = o.main_total_at_silence > 0
&& twin_owed::clearly_above(total, o.main_total_at_silence);
let outgrown = o.twin_total == 0 || twin_owed::clearly_above(total, o.twin_total);
if grew && outgrown && total > o.grow_to_bytes {
o.grow_to_bytes = total;
o.why = format!(
"main's data set grew from {} to {} while twin '{}' was silent (its volume {}): it may not follow again until it holds main's set",
twin_owed::gib(o.main_total_at_silence),
twin_owed::gib(total),
r.principal,
if o.twin_total == 0 { "unknown".to_owned() } else { twin_owed::gib(o.twin_total) }
);
said.push(o.why.clone());
}
}
} else {
o.silent_since_ms = 0;
o.main_total_at_silence = 0;
if r.disk_total_bytes > 0 {
o.twin_total = r.disk_total_bytes;
}
if o.grow_to_bytes > 0
&& r.disk_total_bytes > 0
&& !twin_owed::clearly_above(o.grow_to_bytes, r.disk_total_bytes)
{
said.push(format!(
"twin '{}' holds main's grown set again ({} for {}): it follows",
r.principal,
twin_owed::gib(r.disk_total_bytes),
twin_owed::gib(o.grow_to_bytes)
));
o.grow_to_bytes = 0;
}
if o.resync_from > 0 && r.applied_seq + 1 >= first_seq {
said.push(format!("twin '{}' took its full resync: it reports {} applied and the log retains from {first_seq}", r.principal, r.applied_seq));
o.resync_from = 0;
}
if o.grow_to_bytes == 0 && o.resync_from == 0 {
o.why.clear();
}
}
}
book.twins.retain(|_, o| !o.is_empty());
said
}
/// ★ **One pass under `policy`** (cov-13): the debts of silent twins first, then the
/// window and the byte budget ([`decide`]); a twin given up on is recorded as owed a
/// full resync. `full` is the ten-minute pass (see [`decide`]). Answers the line it wrote.
pub fn pass_with(
ctx: &crate::ServiceContext,
policy: &Policy,
full: bool,
) -> Result<String, String> {
let head = ctx
.control
.log_head()
.map_err(|e| format!("the log's head: {e}"))?;
if head.twin {
return Err("this box is a twin: its log is the primary's numbers and its only copy, never trimmed here".to_owned());
}
let members: Vec<String> = ctx
.control
.list_group_members(crate::TWIN_GROUP)
.map(|ps| ps.into_iter().map(|p| p.id.to_string()).collect())
.unwrap_or_default();
let now = estate_clock::now_unix_ms();
let rows = ctx.heartbeats.view(&members, now);
let data = ctx.control.data_volume();
let mut book = ctx.heartbeats.owed();
let before = book.clone();
for line in note_debts(
&mut book,
&rows,
&|p| ctx.heartbeats.has_reported(p),
data.map(|d| d.0),
head.first_seq,
now,
) {
tracing::warn!(target: TARGET, "{line}");
}
let twins: Vec<TwinAt> = rows
.iter()
.map(|r| {
let applied = ctx
.heartbeats
.has_reported(&r.principal)
.then_some(r.applied_seq);
let silent = applied.is_none() || r.state == STATE_SILENT;
let since = book
.twins
.get(&r.principal)
.map(|o| o.silent_since_ms)
.unwrap_or(0);
TwinAt {
principal: r.principal.clone(),
// A member that never polled has no position (view draws it silent with seen 0).
applied,
state: r.state.clone(),
silent_for_ms: if silent && since > 0 {
now.saturating_sub(since).max(0) as u64
} else {
0
},
}
})
.collect();
let layout = ctx
.control
.log_layout()
.unwrap_or_else(|_| vec![(head.first_seq, head.head_seq, head.bytes)]);
let budget = policy.budget(data.map(|d| d.0));
let decided = decide(&layout, head.head_seq, &twins, policy, budget, full);
let result = match decided {
Err(why) => Err(why),
Ok(plan) => match ctx.control.log_retain(plan.from) {
Err(e) => Err(format!("retaining from {}: {e}", plan.from)),
Ok(after) => {
for p in &plan.gave_up {
let o = book.twins.entry(p.clone()).or_default();
let first_time = o.resync_from == 0;
o.resync_from = after.first_seq;
o.why = format!("the log was trimmed past twin '{p}' while it was silent (the log is bounded in bytes): it retains from {}, and the twin takes a full resync (snapshot) when it polls again", after.first_seq);
// Said loudly once; every later trim past the same silent twin is routine.
if first_time {
tracing::warn!(target: TARGET, twin = %p, first = after.first_seq, "{}", o.why);
} else {
tracing::info!(target: TARGET, twin = %p, first = after.first_seq, "{}", o.why);
}
}
// The log is kept in segments and only whole ones below `from` go: a pass over a
// young log may drop nothing and only seal the segment it writes.
let line = if after.first_seq > head.first_seq {
format!(
"the default retention trimmed the transaction log: it now starts at {} (was {}; retained from {}), head {}, {} bytes (was {}){}; {}",
after.first_seq,
head.first_seq,
plan.from,
after.head_seq,
after.bytes,
head.bytes,
budget.map(|b| format!(", budget {b}")).unwrap_or_default(),
plan.why
)
} else {
format!("the default retention retained from {}: no whole segment lay below it yet (the log still starts at {}), head {}; {}", plan.from, after.first_seq, after.head_seq, plan.why)
};
// INFO: with the byte bound the light pass trims as often as a busy set writes past
// the budget (every 30 s at most) — routine, not a warning. What IS a warning (a
// silent twin given up on, a growth owed) is said at WARN above and in `note_debts`.
tracing::info!(target: TARGET, first = after.first_seq, was = head.first_seq, head = after.head_seq, bytes = after.bytes, window = policy.window, "{line}");
Ok(line)
}
},
};
if book != before {
ctx.heartbeats.set_owed(book.clone());
if let Err(e) = ctx.control.twin_owed_write(&book.to_json()) {
tracing::debug!(target: TARGET, "twin-owed.json was not written (kept in memory): {e}");
}
}
result
}
/// **The pass, every [`EVERY`], for the life of the server** — on a primary,
/// unless `GUNNAR_TXLOG_RETAIN=off`. A pass that has nothing to do says why at
/// debug level only; a trim says so at warn.
pub async fn run(ctx: crate::ServiceContext) {
let env = |k: &str| std::env::var(k).ok();
let Some(policy) = Policy::from_env(
env(RETAIN_ENV).as_deref(),
env(WINDOW_ENV).as_deref(),
env(BUDGET_PERCENT_ENV).as_deref(),
env(BUDGET_BYTES_ENV).as_deref(),
env(SILENT_GRACE_ENV).as_deref(),
) else {
tracing::warn!(target: TARGET, "{RETAIN_ENV}=off: the transaction log is trimmed only by an operator's Mirror.LogRetain");
return;
};
// What the twins are owed survives a restart: read it before the first fetch is answered.
match ctx.control.twin_owed_read() {
Ok(bytes) => match Book::from_json(&bytes) {
Ok(book) => {
if !book.twins.is_empty() {
tracing::warn!(target: TARGET, owed = book.twins.len(), "twin-owed.json: {} twin(s) owed a resync or a growth", book.twins.len());
}
ctx.heartbeats.set_owed(book);
}
Err(e) => {
tracing::warn!(target: TARGET, "{e}: starting with an empty book of what the twins are owed")
}
},
Err(e) => tracing::debug!(target: TARGET, "twin-owed.json not read: {e}"),
}
tracing::info!(
target: TARGET,
window = policy.window,
budget_percent = policy.budget_percent,
budget_max_bytes = policy.budget_max_bytes,
silent_grace_s = policy.silent_grace_ms / 1000,
"the transaction log's retention: {} entries behind the slowest twin, at most {} % of the data volume and {} bytes, a silent twin waited for {} s while it is over",
policy.window,
policy.budget_percent,
policy.budget_max_bytes,
policy.silent_grace_ms / 1000
);
let mut every = tokio::time::interval(CHECK_EVERY);
every.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
// The first tick is immediate: skip it, so a restarting primary first hears its twins.
every.tick().await;
let mut last_full = Instant::now();
loop {
every.tick().await;
let full = last_full.elapsed() >= EVERY;
if full {
last_full = Instant::now();
}
let c = ctx.clone();
match tokio::task::spawn_blocking(move || pass_with(&c, &policy, full)).await {
Ok(Ok(_)) => {}
Ok(Err(why)) => tracing::debug!(target: TARGET, "default retention: {why}"),
Err(e) => tracing::warn!(target: TARGET, "default retention pass panicked: {e}"),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn at(p: &str, applied: Option<u64>, state: &str) -> TwinAt {
TwinAt {
principal: p.into(),
applied,
state: state.into(),
silent_for_ms: 0,
}
}
#[test]
fn the_log_is_retained_behind_the_slowest_reporting_twin_minus_the_window() {
// `window` entries the twin HAS applied are kept: 3 991..=4 990 is a thousand.
assert_eq!(
point(1, 5_000, &[at("gunnar-twin", Some(4_990), "ok")], 1_000),
Ok(3_991)
);
assert_eq!(
point(
1,
5_000,
&[at("a", Some(4_990), "ok"), at("b", Some(2_500), "behind")],
1_000
),
Ok(1_501),
"the SLOWEST twin decides"
);
assert_eq!(
point(2_000, 5_000, &[at("a", Some(4_990), "ok")], 0),
Ok(4_991),
"window 0: everything the twin has applied"
);
}
#[test]
fn nothing_is_trimmed_without_a_twin_or_with_a_silent_or_unheard_one() {
assert!(point(1, 5_000, &[], 1_000).unwrap_err().contains("no twin"));
let e = point(
1,
5_000,
&[at("a", Some(4_990), "ok"), at("b", None, "silent")],
1_000,
)
.unwrap_err();
assert!(e.contains("'b'") && e.contains("not reported"), "{e}");
let e = point(1, 5_000, &[at("a", Some(4_990), STATE_SILENT)], 1_000).unwrap_err();
assert!(e.contains("silent"), "{e}");
}
#[test]
fn nothing_is_trimmed_when_the_window_already_covers_the_log() {
let e = point(3_991, 5_000, &[at("a", Some(4_990), "ok")], 1_000).unwrap_err();
assert!(e.contains("nothing to trim"), "{e}");
let e = point(1, 500, &[at("a", Some(500), "ok")], 1_000).unwrap_err();
assert!(e.contains("nothing to trim"), "{e}");
// A twin reporting past this box's head (a restart race) never retains past the head.
assert_eq!(point(1, 5_000, &[at("a", Some(9_999), "ok")], 0), Ok(5_001));
}
/// **The default window is a few pushes, not a thousand** (h-txlog-2x, measured
/// on the mock 2026-09-29: one fill slice of 256 MiB costs the set ~1 GiB and a
/// handful of entries; 1000 entries kept the WHOLE log of a real estate, so the
/// default trimmed nothing). A following twin is seconds behind; one further
/// behind than the window takes a snapshot, which is correct, only slower.
#[test]
fn the_default_window_is_a_few_pushes() {
const {
assert!(DEFAULT_WINDOW <= 64);
assert!(DEFAULT_WINDOW >= 16);
}
}
// ── cov-13: bytes ───────────────────────────────────────────────────────
/// Ten closed segments of 100 bytes, ten entries each (1..=100), and an empty open one.
fn layout() -> Vec<(u64, u64, u64)> {
let mut l: Vec<(u64, u64, u64)> = (0..10).map(|i| (i * 10 + 1, i * 10 + 10, 100)).collect();
l.push((101, 100, 0));
l
}
fn silent(p: &str, applied: u64, for_ms: u64) -> TwinAt {
TwinAt {
principal: p.into(),
applied: Some(applied),
state: STATE_SILENT.into(),
silent_for_ms: for_ms,
}
}
const POLICY: Policy = Policy {
window: 64,
budget_percent: 0,
budget_max_bytes: 250,
silent_grace_ms: 600_000,
};
#[test]
fn the_budget_keeps_the_newest_whole_segments_that_fit_and_only_what_the_twin_applied_goes() {
// Caught up: the window alone keeps 37..; the budget keeps the newest two segments.
let p = decide(
&layout(),
100,
&[at("t", Some(100), "ok")],
&POLICY,
Some(250),
false,
)
.unwrap();
assert_eq!((p.from, p.gave_up.len()), (81, 0), "{p:?}");
// Lagging at 50: nothing it has not applied goes, whatever the budget.
let p = decide(
&layout(),
100,
&[at("t", Some(50), "behind")],
&POLICY,
Some(250),
false,
)
.unwrap();
assert_eq!(p.from, 51, "{p:?}");
// No budget: the window, as before.
assert_eq!(
decide(
&layout(),
100,
&[at("t", Some(100), "ok")],
&POLICY,
None,
true
)
.unwrap()
.from,
37
);
}
#[test]
fn a_silent_twin_is_given_up_on_past_its_grace_or_at_twice_the_budget_and_never_otherwise() {
// 1000 bytes over a 600 budget (under twice it), silent a minute: waited for.
let e = decide(
&layout(),
100,
&[silent("t", 40, 60_000)],
&POLICY,
Some(600),
false,
)
.unwrap_err();
assert!(
e.contains("silent") && e.contains("given up on after 600 s"),
"{e}"
);
// Silent past the grace: given up on, trimmed to the budget (the newest six segments).
let p = decide(
&layout(),
100,
&[silent("t", 40, 600_000)],
&POLICY,
Some(600),
false,
)
.unwrap();
assert_eq!(
(p.from, p.gave_up.clone()),
(41, vec!["t".to_string()]),
"{p:?}"
);
// Twice the budget: at once.
let p = decide(
&layout(),
100,
&[silent("t", 40, 1_000)],
&POLICY,
Some(250),
false,
)
.unwrap();
assert_eq!((p.from, p.gave_up.len()), (81, 1), "{p:?}");
// Within the budget: a silent twin keeps everything, however long.
assert!(decide(
&layout(),
100,
&[silent("t", 40, u64::MAX)],
&POLICY,
Some(5_000),
false
)
.is_err());
// A heard twin beside a given-up one still holds back what it has not applied.
let p = decide(
&layout(),
100,
&[silent("gone", 10, 900_000), at("alive", Some(70), "ok")],
&POLICY,
Some(250),
false,
)
.unwrap();
assert_eq!(
(p.from, p.gave_up.clone()),
(71, vec!["gone".to_string()]),
"{p:?}"
);
}
#[test]
fn the_light_pass_trims_only_whole_segments_or_seals_to_let_the_next_one() {
// Within the budget, the window would move `from` inside the first segment: nothing to do.
let one = vec![(1, 100, 1_000)];
assert!(decide(
&one,
100,
&[at("t", Some(50), "ok")],
&POLICY,
Some(5_000),
false
)
.is_err());
// The full pass still seals and retains (as the ten-minute pass always did).
assert!(decide(&one, 100, &[at("t", Some(100), "ok")], &POLICY, None, true).is_ok());
// Over the budget with one big open segment the twin has applied part of: seal it.
assert_eq!(
decide(
&one,
100,
&[at("t", Some(50), "ok")],
&POLICY,
Some(250),
false
)
.unwrap()
.from,
51
);
}
#[test]
fn the_budget_is_the_smaller_of_its_share_and_its_ceiling() {
let p = Policy {
window: 64,
budget_percent: 5,
budget_max_bytes: 4 << 30,
silent_grace_ms: 0,
};
assert_eq!(p.budget(Some(20 << 30)), Some(1 << 30), "5 % of 20 GiB");
assert_eq!(
p.budget(Some(300 << 30)),
Some(4 << 30),
"the ceiling on a big set"
);
assert_eq!(
p.budget(None),
Some(4 << 30),
"an unknown volume: the ceiling"
);
assert_eq!(Policy::window_only(64).budget(Some(1 << 40)), None);
let env = Policy::from_env(None, None, Some("10"), Some("2GiB"), Some("30")).unwrap();
assert_eq!(
(
env.budget_percent,
env.budget_max_bytes,
env.silent_grace_ms
),
(10, 2 << 30, 30_000)
);
let dflt = Policy::from_env(None, None, None, None, None).unwrap();
assert_eq!(
(
dflt.window,
dflt.budget_percent,
dflt.budget_max_bytes,
dflt.silent_grace_ms
),
(DEFAULT_WINDOW, 5, 4 << 30, 600_000)
);
assert_eq!(Policy::from_env(Some("off"), None, None, None, None), None);
assert_eq!(
(
parse_bytes("512KiB"),
parse_bytes("64M"),
parse_bytes("123"),
parse_bytes("lots")
),
(Some(512 << 10), Some(64 << 20), Some(123), None)
);
}
#[test]
fn a_twin_silent_through_mains_growth_is_owed_it_and_settles_when_its_volume_holds_it() {
const G: u64 = 1 << 30;
let row = |state: &str, seen: i64, total: u64, applied: u64| gunnar_proto::TwinHeartbeat {
principal: "t".into(),
state: state.into(),
last_seen_unix_ms: seen,
disk_total_bytes: total,
applied_seq: applied,
..Default::default()
};
let yes = |_: &str| true;
let mut book = Book::default();
// Heard, 20 GiB twin behind a 20 GiB set: nothing owed.
note_debts(
&mut book,
&[row("ok", 1_000, 20 * G, 5)],
&yes,
Some(20 * G),
1,
1_000,
);
assert!(book.twins.is_empty(), "{book:?}");
// Silent: noted with main's volume then.
note_debts(
&mut book,
&[row(STATE_SILENT, 1_000, 20 * G, 5)],
&yes,
Some(20 * G),
1,
90_000,
);
assert_eq!(book.twins["t"].main_total_at_silence, 20 * G);
assert_eq!(book.twins["t"].grow_to_bytes, 0, "nothing grew");
// Main grew alone while it was silent: the twin is owed main's set.
let said = note_debts(
&mut book,
&[row(STATE_SILENT, 1_000, 20 * G, 5)],
&yes,
Some(30 * G),
1,
400_000,
);
assert_eq!(book.twins["t"].grow_to_bytes, 30 * G, "{said:?}");
assert!(twin_owed::follow_refusal(&book.twins["t"], 20 * G).is_some());
// Back, still 20 GiB: still owed. Grown to 30: settled.
note_debts(
&mut book,
&[row("ok", 500_000, 20 * G, 5)],
&yes,
Some(30 * G),
1,
500_000,
);
assert_eq!(book.twins["t"].grow_to_bytes, 30 * G);
let said = note_debts(
&mut book,
&[row("ok", 600_000, 30 * G, 5)],
&yes,
Some(30 * G),
1,
600_000,
);
assert!(book.twins.is_empty(), "{book:?} {said:?}");
// A volume that only LOOKS smaller (fs overhead) is the same size.
let mut b2 = Book::default();
note_debts(
&mut b2,
&[row(STATE_SILENT, 1_000, 60 * G - G / 4, 5)],
&yes,
Some(60 * G),
1,
90_000,
);
note_debts(
&mut b2,
&[row(STATE_SILENT, 1_000, 60 * G - G / 4, 5)],
&yes,
Some(61 * G),
1,
400_000,
);
assert_eq!(
b2.twins["t"].grow_to_bytes, 0,
"main grew under 2 %: no debt"
);
}
#[test]
fn the_switch_and_the_window_come_from_the_environment() {
assert_eq!(window_from_env(None, None), Some(DEFAULT_WINDOW));
assert_eq!(window_from_env(Some("OFF"), None), None);
assert_eq!(window_from_env(Some("on"), Some("50")), Some(50));
assert_eq!(
window_from_env(None, Some("nonsense")),
Some(DEFAULT_WINDOW)
);
}
}