server/lib/src/follow_point.rs
//! **A follower's last word, and the one rule that judges it** — shared by the
//! twin's report to its primary ([`crate::grow::TwinRoster`]) and a paired
//! downstream's report to its upstream ([`crate::peers::PeerBoard`]).
//!
//! A follower says *"I have applied your sequence up to `applied_seq` of boot
//! `applied_boot`"*. The leader holds its own head ([`crate::grow::Hold`]: every
//! write admitted since this boot, from 1). From the two:
//!
//! * a position from ANOTHER boot is no position (applied 0, the lag is the
//! whole head) — the leader restarted and the old number counts nothing now;
//! * `lag = head − applied`;
//! * **stalled**: still reporting, the SAME position at every report while the
//! head was ahead of it, for longer than the stall bound — gunnar's heartbeat
//! rule for a stuck applied head;
//! * **silent**: no report for longer than the silence bound.
//!
//! The words are `ok | behind | stalled | silent`, silence first.
use serde::{Deserialize, Serialize};
/// What a leader keeps of one follower's report.
#[derive(Serialize, Deserialize, Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct FollowPoint {
/// The leader boot the position belongs to.
pub applied_boot: u64,
/// The leader's head at the start of the follower's last complete cycle.
pub applied_seq: u64,
/// When the leader last heard this follower (Unix ms).
pub last_seen_ms: i64,
/// Since when the follower has reported the SAME position (same boot) with
/// the leader's head ahead of it; `None` while it moves or is level.
#[serde(default)]
pub stuck_since_ms: Option<i64>,
}
/// How a leader reads a follower now.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Judged {
/// `ok | behind | stalled | silent`.
pub state: &'static str,
pub lag: u64,
/// The applied position in THIS boot (0 for another boot's).
pub applied: u64,
pub seen_ago_ms: u64,
pub silent: bool,
pub stalled: bool,
}
impl FollowPoint {
/// **The word heard again**: the new point, carrying the stuck-since mark
/// from `prev` while the follower repeats the same position behind `head`.
pub fn heard(
prev: Option<&FollowPoint>,
applied_boot: u64,
applied_seq: u64,
boot: u64,
head: u64,
now_ms: i64,
) -> FollowPoint {
let behind = applied_boot != boot || applied_seq < head;
let stuck_since_ms = match prev {
_ if !behind => None,
Some(p) if p.applied_boot == applied_boot && p.applied_seq == applied_seq => {
p.stuck_since_ms.or(Some(now_ms))
}
_ => Some(now_ms),
};
FollowPoint {
applied_boot,
applied_seq,
last_seen_ms: now_ms,
stuck_since_ms,
}
}
/// The applied position as THIS boot reads it.
pub fn applied_in(&self, boot: u64) -> u64 {
if self.applied_boot == boot {
self.applied_seq
} else {
0
}
}
/// **The verdict**, against the leader's `(boot, head)` now.
pub fn judge(
&self,
boot: u64,
head: u64,
now_ms: i64,
silent_after_ms: u64,
stalled_after_ms: u64,
) -> Judged {
let this_boot = self.applied_boot == boot;
let lag = if this_boot {
head.saturating_sub(self.applied_seq)
} else {
head
};
let seen_ago_ms = (now_ms - self.last_seen_ms).max(0) as u64;
let silent = seen_ago_ms > silent_after_ms;
let stalled = lag > 0
&& self
.stuck_since_ms
.is_some_and(|t| (now_ms - t).max(0) as u64 > stalled_after_ms);
let state = if silent {
"silent"
} else if stalled {
"stalled"
} else if lag > 0 {
"behind"
} else {
"ok"
};
Judged {
state,
lag,
applied: self.applied_in(boot),
seen_ago_ms,
silent,
stalled,
}
}
}
/// The worse of two state words (`ok < behind < stalled < silent`).
pub fn worse(a: &'static str, b: &'static str) -> &'static str {
fn rank(s: &str) -> u8 {
match s {
"ok" => 0,
"behind" => 1,
"stalled" => 2,
_ => 3,
}
}
if rank(b) > rank(a) {
b
} else {
a
}
}
// ── the twin's own words ────────────────────────────────────────────────────
/// **The twin's health words, every report** — the pattern's names, which the ONE
/// monetize's twin sizing reads off the pull channel (monetize-server
/// `estate_order::twin::TwinReading::of_reports`): where the twin stands against its
/// primary, and its load.
pub const TWIN_STATE: &str = "twin_state";
pub const TWIN_LAG_ENTRIES: &str = "lag_entries";
pub const TWIN_CPU_BUSY: &str = "cpu_millicores_busy";
pub const TWIN_CPU_TOTAL: &str = "cpu_millicores_total";
/// **What a twin reports of itself**: its state and lag as the one rule judges its last
/// point against the primary's head (`look`; none before its first answer, so no lag is
/// said), and its load (`cpu`: busy and total millicores; none when unmeasured).
pub fn twin_words(
h: &mut std::collections::BTreeMap<String, String>,
look: Option<&Judged>,
cpu: Option<(u64, u64)>,
) {
if let Some(j) = look {
h.insert(TWIN_STATE.to_owned(), j.state.to_owned());
h.insert(TWIN_LAG_ENTRIES.to_owned(), j.lag.to_string());
}
if let Some((busy, total)) = cpu {
h.insert(TWIN_CPU_BUSY.to_owned(), busy.to_string());
h.insert(TWIN_CPU_TOTAL.to_owned(), total.to_string());
}
}
/// **The review's seqs** (the same reader's `primary_head_seq` / `applied_seq` /
/// `state`): `state` and the leader's head and the follower's applied position.
pub fn review_point(
state: &str,
head: u64,
applied: u64,
) -> serde_json::Map<String, serde_json::Value> {
let mut m = serde_json::Map::new();
m.insert("state".into(), state.into());
m.insert("primary_head_seq".into(), head.into());
m.insert("applied_seq".into(), applied.into());
m
}
/// Pure: the load of a box with `cores` cores from `/proc/loadavg`'s text, as
/// `(busy, total)` millicores — the one-minute load times 1000, the cores times 1000.
pub fn cpu_of(loadavg: &str, cores: usize) -> Option<(u64, u64)> {
let load1: f64 = loadavg.split_whitespace().next()?.parse().ok()?;
if !load1.is_finite() || load1 < 0.0 || cores == 0 {
return None;
}
Some(((load1 * 1000.0).round() as u64, cores as u64 * 1000))
}
/// This box's own load ([`cpu_of`] over `/proc/loadavg`); `None` where it cannot be read.
pub fn own_cpu() -> Option<(u64, u64)> {
let cores = std::thread::available_parallelism().ok()?.get();
cpu_of(&std::fs::read_to_string("/proc/loadavg").ok()?, cores)
}
#[cfg(test)]
mod tests {
use super::*;
/// ★ The one rule: level is ok, moving is behind, repeating behind the head
/// past the bound is stalled, quiet past the bound is silent, and another
/// boot's position counts nothing.
#[test]
fn a_follow_point_is_judged_by_one_rule() {
let p = FollowPoint::heard(None, 7, 10, 7, 10, 1_000);
assert_eq!(p.stuck_since_ms, None);
assert_eq!(p.judge(7, 10, 1_000, 60_000, 5_000).state, "ok");
let p = FollowPoint::heard(Some(&p), 7, 10, 7, 15, 2_000);
assert_eq!(p.stuck_since_ms, Some(2_000));
let j = p.judge(7, 15, 2_000, 60_000, 5_000);
assert_eq!((j.state, j.lag, j.applied), ("behind", 5, 10));
let p = FollowPoint::heard(Some(&p), 7, 10, 7, 20, 9_000);
assert_eq!(
p.stuck_since_ms,
Some(2_000),
"the same position keeps its mark"
);
let j = p.judge(7, 20, 9_000, 60_000, 5_000);
assert!(j.stalled && j.state == "stalled" && j.lag == 10);
let moved = FollowPoint::heard(Some(&p), 7, 18, 7, 20, 10_000);
assert_eq!(
moved.stuck_since_ms,
Some(10_000),
"a move restarts the clock"
);
assert_eq!(moved.judge(7, 20, 10_000, 60_000, 5_000).state, "behind");
assert_eq!(moved.judge(7, 20, 80_000, 60_000, 5_000).state, "silent");
let other_boot = moved.judge(8, 3, 10_000, 60_000, 5_000);
assert_eq!((other_boot.applied, other_boot.lag), (0, 3));
assert_eq!(worse("behind", "stalled"), "stalled");
assert_eq!(worse("silent", "ok"), "silent");
}
/// ★ **A twin says where it stands in the pattern's words** (M2): state and lag from
/// the one rule, its load in millicores; no answer yet, no lag said.
#[test]
fn a_twin_says_its_lag_and_load_in_the_patterns_words() {
let p = FollowPoint::heard(None, 7, 60, 7, 100, 1_000);
let j = p.judge(7, 100, 1_000, 60_000, 5_000);
let mut h = std::collections::BTreeMap::new();
twin_words(&mut h, Some(&j), Some((420, 2_000)));
let w = |k: &str| h.get(k).cloned().unwrap_or_default();
assert_eq!(
(
w(TWIN_STATE),
w(TWIN_LAG_ENTRIES),
w(TWIN_CPU_BUSY),
w(TWIN_CPU_TOTAL)
),
("behind".into(), "40".into(), "420".into(), "2000".into()),
"{h:?}"
);
let mut none = std::collections::BTreeMap::new();
twin_words(&mut none, None, None);
assert!(
none.is_empty(),
"no answer and no load: nothing said, {none:?}"
);
let r = review_point("behind", 100, 60);
assert_eq!(
serde_json::Value::Object(r),
serde_json::json!({"state": "behind", "primary_head_seq": 100, "applied_seq": 60})
);
assert_eq!(cpu_of("0.42 0.30 0.10 1/100 999\n", 2), Some((420, 2_000)));
assert_eq!(cpu_of("1.5 1 1 1/1 1", 4), Some((1_500, 4_000)));
assert_eq!(cpu_of("nonsense", 4), None);
}
}