gunnar-grpc 1.5.0

//! **The dual heartbeat between a primary and its twin** (2026-09-15).
//!
//! The twin already polls the primary (`Mirror.LogHead`, every two seconds);
//! before this module only the twin learned anything from it. Now each poll
//! carries a [`gunnar_proto::TwinReport`] — the twin's applied head and its
//! data volume's fill — and the primary keeps the last one per caller in a
//! [`Registry`]. `Admin.Health` reads it back as [`gunnar_proto::TwinHeartbeat`]
//! rows. The twin's side (when the primary last answered) is
//! [`primary_view`], fed by the follower.
//!
//! Every state is decided **at answer time** from the last record and the
//! clock, never by a timer: a twin that died cannot report its own death, so
//! `silent` has to be what the absence of polls reads as.
//!
//! # The rules
//!
//! * `silent` — no poll within [`Bounds::silent_after_ms`], or never since
//!   this process started.
//! * `behind` — the lag was above zero at each of the last
//!   [`Bounds::behind_polls`] + 1 polls AND either the twin's applied head did
//!   not move across them (stuck) or the lag grew at every one of them
//!   (falling behind). A healthy twin under a steady write rate shows a
//!   constant small lag with a moving head; that is `ok`.
//! * `ok` — otherwise.
//!
//! The history ring is allocated once per twin, at its first poll, and reused.

use std::collections::BTreeMap;
use std::sync::Mutex;

use gunnar_proto::{PrimaryHeartbeat, TwinHeartbeat, TwinReport};

/// `silent` after this long without a poll (the twin polls every 2 s).
pub const SILENT_AFTER_MS: u64 = 60_000;
/// `behind` after this many polls without the lag shrinking.
pub const BEHIND_POLLS: usize = 5;

/// The state words, written once.
pub const STATE_OK: &str = "ok";
pub const STATE_BEHIND: &str = "behind";
pub const STATE_SILENT: &str = "silent";

/// The two thresholds.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Bounds {
    pub silent_after_ms: u64,
    pub behind_polls: usize,
}

impl Default for Bounds {
    fn default() -> Self {
        Self { silent_after_ms: SILENT_AFTER_MS, behind_polls: BEHIND_POLLS }
    }
}

impl Bounds {
    /// `GUNNAR_HEARTBEAT_SILENT_MS` / `GUNNAR_HEARTBEAT_BEHIND_POLLS` override
    /// the defaults (a bench shortens the bound; unparsable values are ignored).
    pub fn from_env() -> Self {
        let mut b = Self::default();
        if let Some(v) = std::env::var("GUNNAR_HEARTBEAT_SILENT_MS").ok().and_then(|v| v.parse::<u64>().ok()) {
            b.silent_after_ms = v.max(1);
        }
        if let Some(v) = std::env::var("GUNNAR_HEARTBEAT_BEHIND_POLLS").ok().and_then(|v| v.parse::<usize>().ok()) {
            b.behind_polls = v.max(1);
        }
        b
    }
}

#[derive(Debug, Clone, Copy, Default)]
struct Sample {
    applied: u64,
    lag: u64,
}

#[derive(Debug)]
struct Row {
    address: String,
    last_seen_ms: i64,
    report: TwinReport,
    head_seq: u64,
    /// Ring of the last `behind_polls + 1` samples; `ring.len()` fixed at the first poll.
    ring: Box<[Sample]>,
    /// Next write index.
    at: usize,
    /// Samples written, saturating at `ring.len()`.
    filled: usize,
}

impl Row {
    fn push(&mut self, s: Sample) {
        self.ring[self.at] = s;
        self.at = (self.at + 1) % self.ring.len();
        self.filled = (self.filled + 1).min(self.ring.len());
    }

    /// Oldest → newest.
    fn samples(&self) -> impl Iterator<Item = &Sample> {
        let n = self.ring.len();
        let start = (self.at + n - self.filled) % n;
        (0..self.filled).map(move |i| &self.ring[(start + i) % n])
    }

    fn behind(&self) -> Option<String> {
        if self.filled < self.ring.len() {
            return None;
        }
        let mut it = self.samples();
        let first = *it.next()?;
        if first.lag == 0 {
            return None;
        }
        let (mut stuck, mut growing, mut prev, mut last) = (true, true, first, first);
        for s in it {
            if s.lag == 0 {
                return None;
            }
            stuck &= s.applied == first.applied;
            growing &= s.lag > prev.lag;
            prev = *s;
            last = *s;
        }
        let polls = self.ring.len() - 1;
        if stuck {
            Some(format!("applied head stuck at {} for {polls} polls with {} entries to go", last.applied, last.lag))
        } else if growing {
            Some(format!("lag grew at each of the last {polls} polls: {} → {} entries", first.lag, last.lag))
        } else {
            None
        }
    }
}

/// The primary's record of every twin that polled it.
#[derive(Debug)]
pub struct Registry {
    bounds: Bounds,
    rows: Mutex<BTreeMap<String, Row>>,
    /// ★ What each twin is OWED (cov-13): a full resync, a growth. Kept by the
    /// retention pass (and on disk through the control plane); read by the
    /// doors a twin follows through.
    owed: Mutex<crate::twin_owed::Book>,
}

impl Default for Registry {
    fn default() -> Self {
        Self::new(Bounds::from_env())
    }
}

impl Registry {
    pub fn new(bounds: Bounds) -> Self {
        Self { bounds, rows: Mutex::new(BTreeMap::new()), owed: Mutex::new(crate::twin_owed::Book::default()) }
    }

    /// The book of what each twin is owed, as it stands.
    pub fn owed(&self) -> crate::twin_owed::Book {
        self.owed.lock().expect("heartbeat registry").clone()
    }

    /// Replace the book (the retention pass, a load at start, a test).
    pub fn set_owed(&self, book: crate::twin_owed::Book) {
        *self.owed.lock().expect("heartbeat registry") = book;
    }

    /// **Why `principal` may not follow now**, or `None` — a twin owed a growth
    /// whose LAST report (this very poll's, `Mirror.LogHead` precedes every
    /// fetch) still holds less than main's grown set ([`crate::twin_owed::follow_refusal`]).
    pub fn follow_refusal(&self, principal: &str) -> Option<String> {
        let owed = self.owed.lock().expect("heartbeat registry").twins.get(principal).cloned()?;
        let reported = self.rows.lock().expect("heartbeat registry").get(principal).map(|r| r.report.disk_total_bytes).unwrap_or(0);
        crate::twin_owed::follow_refusal(&owed, reported)
    }

    pub fn bounds(&self) -> Bounds {
        self.bounds
    }

    /// One poll from `principal` at `address`, answered with `head_seq`.
    pub fn record(&self, principal: &str, address: &str, report: &TwinReport, head_seq: u64, now_ms: i64) {
        let mut rows = self.rows.lock().expect("heartbeat registry");
        let len = self.bounds.behind_polls + 1;
        let row = rows.entry(principal.to_owned()).or_insert_with(|| Row {
            address: String::new(),
            last_seen_ms: 0,
            report: TwinReport::default(),
            head_seq: 0,
            ring: vec![Sample::default(); len].into_boxed_slice(),
            at: 0,
            filled: 0,
        });
        if row.address != address {
            row.address.clear();
            row.address.push_str(address);
        }
        row.last_seen_ms = now_ms;
        row.report.clone_from(report);
        row.head_seq = head_seq;
        row.push(Sample { applied: report.applied_seq, lag: head_seq.saturating_sub(report.applied_seq) });
    }

    /// **The lowest applied position any twin has reported** since this server started, and
    /// which twin (2026-09-22, the retention guard). `None` = no twin has polled: nothing is
    /// known, and the caller says so rather than inventing a position.
    pub fn lowest_applied(&self) -> Option<(String, u64)> {
        let rows = self.rows.lock().expect("heartbeat registry");
        rows.iter().map(|(p, r)| (p.clone(), r.report.applied_seq)).min_by_key(|(_, a)| *a)
    }

    /// **The log twins at `head`?** Each of `members` (the enrolled twins) as it last
    /// reported: `Some(applied)` or `None` (not heard since this server started).
    pub fn applied_of(&self, members: &[String]) -> Vec<(String, Option<u64>)> {
        let rows = self.rows.lock().expect("heartbeat registry");
        members.iter().map(|m| (m.clone(), rows.get(m).map(|r| r.report.applied_seq))).collect()
    }

    /// Has `principal` polled (and so reported a position) since this server started?
    pub fn has_reported(&self, principal: &str) -> bool {
        self.rows.lock().expect("heartbeat registry").contains_key(principal)
    }

    /// One row per name in `members` and per principal that polled, sorted by principal.
    pub fn view(&self, members: &[String], now_ms: i64) -> Vec<TwinHeartbeat> {
        let rows = self.rows.lock().expect("heartbeat registry");
        let mut names: Vec<&str> = rows.keys().map(String::as_str).collect();
        for m in members {
            if !rows.contains_key(m) {
                names.push(m);
            }
        }
        names.sort_unstable();
        names.dedup();
        let silent_after_ms = self.bounds.silent_after_ms;
        names
            .into_iter()
            .map(|name| match rows.get(name) {
                None => TwinHeartbeat {
                    principal: name.to_owned(),
                    state: STATE_SILENT.to_owned(),
                    why: "a member of group twin that has not polled since this server started".to_owned(),
                    silent_after_ms,
                    ..Default::default()
                },
                Some(r) => {
                    let ago = now_ms.saturating_sub(r.last_seen_ms).max(0) as u64;
                    let lag = r.head_seq.saturating_sub(r.report.applied_seq);
                    let (state, why) = if ago > silent_after_ms {
                        (STATE_SILENT, format!("no poll for {} s (bound {} s)", ago / 1000, silent_after_ms / 1000))
                    } else if let Some(why) = r.behind() {
                        (STATE_BEHIND, why)
                    } else {
                        (STATE_OK, String::new())
                    };
                    TwinHeartbeat {
                        principal: name.to_owned(),
                        address: r.address.clone(),
                        last_seen_unix_ms: r.last_seen_ms,
                        seen_ago_ms: ago,
                        applied_seq: r.report.applied_seq,
                        head_seq: r.head_seq,
                        lag_entries: lag,
                        disk_total_bytes: r.report.disk_total_bytes,
                        disk_used_bytes: r.report.disk_used_bytes,
                        fill_permille: fill_permille(r.report.disk_used_bytes, r.report.disk_total_bytes),
                        state: state.to_owned(),
                        why,
                        twin_state: r.report.state.clone(),
                        silent_after_ms,
                        cpu_millicores_busy: r.report.cpu_millicores_busy,
                        cpu_millicores_total: r.report.cpu_millicores_total,
                    }
                }
            })
            .collect()
    }
}

/// `used * 1000 / total`, 0 when `total` is 0, capped at 1000.
pub fn fill_permille(used: u64, total: u64) -> u32 {
    if total == 0 {
        return 0;
    }
    ((used as u128 * 1000 / total as u128).min(1000)) as u32
}

/// The twin's view of its primary: `last_answer_ms` 0 = never answered.
///
/// **`own` is what the FOLLOWER recorded** — its state word and its `why` —
/// and it wins whenever it is not `following` (X2, 2026-09-23). This used to
/// take neither, and derived the whole line from `last_answer_ms` alone: so a
/// follower that could not start (`unreachable`, with the attempt count and
/// the real error), one whose every poll failed (`stalled`, with the error),
/// and one that had genuinely never been asked all printed the SAME sentence —
/// "the primary has not answered since this twin started". The one place on the
/// box that knew why was written and then thrown away here.
///
/// Answering is still what decides SILENT vs ok. The follower's word only
/// replaces the reason, and only when it has one: a twin that is following
/// normally reads exactly as it did before.
pub fn primary_view(
    address: &str,
    last_answer_ms: i64,
    primary_head_seq: u64,
    applied_seq: u64,
    now_ms: i64,
    silent_after_ms: u64,
    own: Option<(&str, &str)>,
) -> PrimaryHeartbeat {
    // What the follower itself recorded, when it is not the ordinary word.
    let recorded = own.filter(|(state, _)| !state.is_empty() && *state != "following");
    let (ago, state, why) = if last_answer_ms <= 0 {
        let why = match recorded {
            Some((state, why)) if !why.is_empty() => format!("the follower says {state}: {why}"),
            Some((state, _)) => format!("the follower says {state}"),
            None => "the primary has not answered since this twin started".to_owned(),
        };
        (0, STATE_SILENT, why)
    } else {
        let ago = now_ms.saturating_sub(last_answer_ms).max(0) as u64;
        if ago > silent_after_ms {
            let base = format!("no answer for {} s (bound {} s)", ago / 1000, silent_after_ms / 1000);
            let why = match recorded {
                Some((state, w)) if !w.is_empty() => format!("{base}; the follower says {state}: {w}"),
                Some((state, _)) => format!("{base}; the follower says {state}"),
                None => base,
            };
            (ago, STATE_SILENT, why)
        } else {
            // It IS answering. A follower that still reports a word of its own
            // (stalled mid-range, snapshot-needed, diverged) is not "ok", and
            // saying so is the whole point of carrying it.
            match recorded {
                Some((state, w)) if !w.is_empty() => (ago, state, format!("the follower says {state}: {w}")),
                Some((state, _)) => (ago, state, format!("the follower says {state}")),
                None => (ago, STATE_OK, String::new()),
            }
        }
    };
    PrimaryHeartbeat {
        address: address.to_owned(),
        last_answer_unix_ms: last_answer_ms.max(0),
        answer_ago_ms: ago,
        primary_head_seq,
        applied_seq,
        state: state.to_owned(),
        why,
        silent_after_ms,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn rep(applied: u64, used: u64, total: u64) -> TwinReport {
        TwinReport { applied_seq: applied, disk_total_bytes: total, disk_used_bytes: used, state: "following".into(), ..TwinReport::default() }
    }

    /// ★ **The twin's load reaches the primary's heartbeat** (2026-10-05): what
    /// the twin measured of its own cores on its poll is on its row, beside
    /// the lag — what a private estate's monetize sizes the twin by. A twin
    /// that did not say (an older one) is absent, never 0.
    #[test]
    fn the_twins_own_load_is_on_its_row_beside_the_lag() {
        let r = Registry::new(B);
        r.record("twin-a", "10.44.0.3:4000", &TwinReport { cpu_millicores_busy: Some(1_700), cpu_millicores_total: Some(2_000), ..rep(80, 1, 2) }, 120, 5_000);
        r.record("twin-b", "10.44.0.4:4000", &rep(120, 1, 2), 120, 5_000);
        let v = r.view(&[], 6_000);
        assert_eq!((v[0].lag_entries, v[0].cpu_millicores_busy, v[0].cpu_millicores_total), (40, Some(1_700), Some(2_000)));
        assert_eq!((v[1].cpu_millicores_busy, v[1].cpu_millicores_total), (None, None), "not said: absent, never 0");
    }

    const B: Bounds = Bounds { silent_after_ms: 60_000, behind_polls: 3 };

    #[test]
    fn a_member_never_heard_from_is_silent_and_a_polling_twin_is_ok_with_its_fill() {
        let r = Registry::new(B);
        let v = r.view(&["twin-a".into()], 1_000);
        assert_eq!(v.len(), 1);
        assert_eq!(v[0].state, STATE_SILENT);
        r.record("twin-a", "10.0.0.2:4000", &rep(10, 850, 1000), 12, 5_000);
        let v = r.view(&["twin-a".into()], 6_000);
        assert_eq!(v.len(), 1, "a member that polled is one row, not two");
        let t = &v[0];
        assert_eq!((t.state.as_str(), t.lag_entries, t.fill_permille, t.seen_ago_ms), (STATE_OK, 2, 850, 1_000));
        assert_eq!(t.address, "10.0.0.2:4000");
    }

    #[test]
    fn a_killed_twin_reads_silent_once_the_bound_passes_and_not_before() {
        let r = Registry::new(B);
        r.record("t", "a", &rep(5, 1, 2), 5, 100_000);
        assert_eq!(r.view(&[], 100_000 + 60_000)[0].state, STATE_OK, "at the bound exactly: still ok");
        let v = r.view(&[], 100_000 + 60_001);
        assert_eq!(v[0].state, STATE_SILENT);
        assert!(v[0].why.contains("no poll for 60 s"), "{}", v[0].why);
    }

    #[test]
    fn behind_is_a_stuck_head_or_a_growing_lag_never_a_steady_small_lag() {
        // Steady: head moves, lag constant 2 → ok.
        let r = Registry::new(B);
        for i in 0..10u64 {
            r.record("t", "a", &rep(i * 3, 0, 1), i * 3 + 2, i as i64 * 2000);
        }
        assert_eq!(r.view(&[], 18_000)[0].state, STATE_OK);
        // Stuck: applied fixed at 7, head moving or not.
        let r = Registry::new(B);
        for i in 0..4u64 {
            r.record("t", "a", &rep(7, 0, 1), 9, i as i64 * 2000);
        }
        let v = r.view(&[], 6_000);
        assert_eq!(v[0].state, STATE_BEHIND, "{}", v[0].why);
        assert!(v[0].why.contains("stuck at 7"));
        // Three polls are not enough history for behind_polls = 3.
        let r = Registry::new(B);
        for i in 0..3u64 {
            r.record("t", "a", &rep(7, 0, 1), 9, i as i64 * 2000);
        }
        assert_eq!(r.view(&[], 4_000)[0].state, STATE_OK);
        // Growing: applied moves but lag grows each poll.
        let r = Registry::new(B);
        for (i, (applied, head)) in [(1u64, 3u64), (2, 6), (3, 9), (4, 12)].into_iter().enumerate() {
            r.record("t", "a", &rep(applied, 0, 1), head, i as i64 * 2000);
        }
        let v = r.view(&[], 6_000);
        assert_eq!(v[0].state, STATE_BEHIND);
        assert!(v[0].why.contains("2 → 8"), "{}", v[0].why);
        // Caught up once → the ring forgets being behind only after lag 0 is in it.
        r.record("t", "a", &rep(12, 0, 1), 12, 8_000);
        assert_eq!(r.view(&[], 8_000)[0].state, STATE_OK);
    }

    #[test]
    fn the_twin_sees_its_primary_silent_after_the_bound_and_never_answered_as_silent() {
        assert_eq!(primary_view("p:1", 0, 0, 0, 5, 60_000, None).state, STATE_SILENT);
        assert_eq!(primary_view("p:1", 1_000, 9, 9, 61_000, 60_000, None).state, STATE_OK);
        let p = primary_view("p:1", 1_000, 9, 9, 61_001, 60_000, None);
        assert_eq!((p.state.as_str(), p.answer_ago_ms), (STATE_SILENT, 60_000 + 1));
    }

    #[test]
    fn fill_is_permille_and_survives_huge_volumes() {
        assert_eq!(fill_permille(0, 0), 0);
        assert_eq!(fill_permille(u64::MAX, u64::MAX), 1000);
        assert_eq!(fill_permille(799, 1000), 799);
    }
}

#[cfg(test)]
mod follower_word_tests {
    use super::*;

    /// ★ **A FOLLOWER THAT CANNOT START MUST NOT READ AS ONE NOBODY ASKED.**
    ///
    /// Before X2 these three printed one sentence — "the primary has not
    /// answered since this twin started" — so a box whose follower recorded
    /// `unreachable (attempt 412 …): <the real error>` was indistinguishable
    /// from a box that had genuinely never been polled. That ambiguity cost
    /// this estate three wrong conclusions in a night.
    ///
    /// To see this fail with the defect back, drop the `own` argument's effect
    /// and derive the reason from `last_answer_ms` alone.
    #[test]
    fn a_follower_that_never_started_says_so_instead_of_blaming_the_primary() {
        let bare = primary_view("p:1", 0, 0, 0, 5, 60_000, None);
        assert_eq!(bare.state, STATE_SILENT);
        assert_eq!(bare.why, "the primary has not answered since this twin started");

        let recorded = primary_view(
            "p:1",
            0,
            0,
            0,
            5,
            60_000,
            Some(("unreachable", "the follower is not running (attempt 412, starting again in 5 s): no twin.json")),
        );
        assert_eq!(recorded.state, STATE_SILENT, "it still has not answered");
        assert_ne!(recorded.why, bare.why, "the two must not print the same sentence");
        assert!(recorded.why.contains("unreachable"), "{}", recorded.why);
        assert!(recorded.why.contains("attempt 412"), "the real reason must survive: {}", recorded.why);
    }

    /// ★ **A STALLED POLL IS NOT `ok`**, even while the primary answers.
    #[test]
    fn a_stalled_follower_is_not_ok_even_when_the_primary_answers() {
        let ok = primary_view("p:1", 1_000, 9, 9, 2_000, 60_000, Some(("following", "")));
        assert_eq!((ok.state.as_str(), ok.why.as_str()), (STATE_OK, ""), "a following twin reads as it always did");

        let stalled = primary_view("p:1", 1_000, 9, 4, 2_000, 60_000, Some(("stalled", "the last poll failed: append: ENOSPC")));
        assert_eq!(stalled.state, "stalled", "the follower's own word wins: {}", stalled.why);
        assert!(stalled.why.contains("ENOSPC"), "{}", stalled.why);
    }

    /// ★ A twin that is simply following is untouched by all of this.
    #[test]
    fn a_healthy_twin_reads_exactly_as_before() {
        let with = primary_view("p:1", 1_000, 9, 9, 2_000, 60_000, Some(("following", "")));
        let without = primary_view("p:1", 1_000, 9, 9, 2_000, 60_000, None);
        assert_eq!((with.state, with.why), (without.state, without.why));
    }
}

/// **Have the LOG twins applied `head`?** — the flush's proof when the twin
/// FOLLOWS the transaction log (it pulls; no push replicator runs on the
/// primary). Pure: `Ok(verdict)` when every enrolled twin reported an applied
/// position at or past `head`; `Err(what is missing)` otherwise. No member =
/// no twin: `Err`, never a vacuous yes.
pub fn log_twins_at(applied: &[(String, Option<u64>)], head: u64) -> Result<String, String> {
    if applied.is_empty() {
        return Err("no twin is enrolled: nothing follows this log, so nothing is proven".to_owned());
    }
    let short: Vec<String> = applied
        .iter()
        .filter_map(|(p, a)| match a {
            Some(a) if *a >= head => None,
            Some(a) => Some(format!("{p} applied {a} of {head}")),
            None => Some(format!("{p} has not reported since this server started")),
        })
        .collect();
    if short.is_empty() {
        let names: Vec<&str> = applied.iter().map(|(p, _)| p.as_str()).collect();
        Ok(format!("every log twin ({}) has applied the primary's head {head}: every write committed before the flush is on the twin", names.join(", ")))
    } else {
        Err(short.join("; "))
    }
}

#[cfg(test)]
mod log_flush_tests {
    use super::*;

    #[test]
    fn a_log_twin_is_caught_up_only_at_or_past_the_head_and_silence_is_not_a_yes() {
        let at = |v: &[(&str, Option<u64>)]| v.iter().map(|(p, a)| (p.to_string(), *a)).collect::<Vec<_>>();
        assert!(log_twins_at(&at(&[("gunnar-twin", Some(402))]), 402).unwrap().contains("head 402"));
        assert!(log_twins_at(&at(&[("gunnar-twin", Some(403))]), 402).is_ok());
        assert!(log_twins_at(&at(&[("gunnar-twin", Some(401))]), 402).unwrap_err().contains("applied 401 of 402"));
        assert!(log_twins_at(&at(&[("gunnar-twin", None)]), 402).unwrap_err().contains("not reported"));
        assert!(log_twins_at(&at(&[("a", Some(402)), ("b", None)]), 402).is_err(), "every twin, not one");
        assert!(log_twins_at(&[], 402).unwrap_err().contains("no twin is enrolled"));
        let r = Registry::new(Bounds::from_env());
        r.record("gunnar-twin", "10.0.0.2:1", &TwinReport { applied_seq: 7, ..TwinReport::default() }, 9, 1);
        assert_eq!(r.applied_of(&["gunnar-twin".into(), "other".into()]), vec![("gunnar-twin".into(), Some(7)), ("other".into(), None)]);
    }
}