gunnar-grpc 1.5.0

//! **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));
    }
}