crates/gunnar-grpc/src/heartbeat.rs
//! **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)]
);
}
}