crates/monetize-server/src/estate_order/twin.rs
//! **The twin sizes itself** (PRIVATE-GUNNAR.md §10): the buyer never buys
//! it. Its plan is an eighth of main's at the order (never under 4 GB); afterwards it GROWS one
//! plan step at a time when it stays behind — replication lag
//! (`TwinFill.lag_entries`) or load (`cpu_millicores_busy`) high for a
//! sustained window — and never:
//!
//! * on a spike: every sample of the window must be high (a flapping load
//! is not a trend);
//! * while it is furthest behind: a resize STOPS the twin, and stopping it
//! at the top of a rising lag makes the gap worse — it waits for the lag
//! to turn, unless the high window has lasted twice as long (then it will
//! not catch up by itself);
//! * within the cooldown after its last change;
//! * past main's plan (a twin bigger than main replicates nothing faster);
//! * over the estate's spend cap.
//!
//! [`decide`] is PURE: samples in, a verdict out, with HYSTERESIS (a twin
//! that went behind stays behind until its lag falls under `lag_low`, so a
//! lag that hovers at the threshold is one trend, not a flap) and a spacing
//! of at least an hour between two changes ([`MIN_SPACING_SECS`]), whatever
//! the configured cooldown says.
//!
//! # Where the samples come from (G12, 2026-10-10: the pull channel)
//!
//! gunnar's own numbers, REPORTED by the estate — monetize never dials an estate (owner rule
//! 07:39, ONE-MONETIZE-MODEL):
//!
//! ```text
//! twin ── Mirror.LogHead poll ──▶ main (WireGuard overlay, inside the estate)
//! main / twin ── Fleet.Report{twin_review, health words} ──▶ this monetize (the pull channel)
//! the newest `twin_review` (gunnar_repair::TwinReview) ──▶ [`ReportedTwinHealth`] ──▶ the window ──▶ [`decide`]
//! ```
//!
//! [`ReportedTwinHealth`] is the shop's reader (pure Rust: no door dialled, no process, no
//! kit): the lag is the review's `primary_head_seq − applied_seq`, the load the twin box's own
//! health words `cpu_millicores_busy` / `cpu_millicores_total` (none reported: no load
//! trigger). It reads at most once per `sample_every_secs` per estate ([`sample_when_due`])
//! and keeps a sample only when a review arrived since the last one; each sample is kept in
//! the estate's window (`<run>/twin-samples.json`) so a restart keeps the trend. The tests'
//! source is [`MockTwinHealth`]. A command (`[estate_orders] twin_health`, the kit's `cargo
//! xtask … estate twin-health`) is no longer run: a config that names one is refused at start
//! ([`super::TWIN_HEALTH_COMMAND_REMOVED`]).
use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Mutex;
use monetize_cloud::lay::PlanSpec;
use monetize_guardian::store::Store;
use crate::facts::TwinFacts;
/// One reading of the twin.
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TwinSample {
pub at_ms: u64,
/// `TwinFill.lag_entries`: how far the replica is behind main.
pub lag_entries: u64,
/// `cpu_millicores_busy`, and what the twin's plan has (`cores × 1000`).
pub cpu_millicores_busy: u64,
pub cpu_millicores: u64,
}
impl TwinSample {
fn load_pct(&self) -> u64 {
self.cpu_millicores_busy
.saturating_mul(100)
.checked_div(self.cpu_millicores)
.unwrap_or(0)
}
}
/// **The rule's numbers** (`[estate_orders.twin]`).
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct TwinRule {
/// Lag at or over this many entries is "behind".
pub lag_high: u64,
/// **Hysteresis**: a twin already behind stays behind while its lag is at
/// or over this (a lag that hovers at `lag_high` is one trend).
pub lag_low: u64,
/// Load at or over this percent of the twin's cores is "busy".
pub load_high_pct: u64,
/// Hysteresis for the load, as `lag_low` is for the lag.
pub load_low_pct: u64,
/// This many consecutive high samples make a trend.
pub sustain: usize,
/// No second change within this long of the last one — never under
/// [`MIN_SPACING_SECS`].
pub cooldown_secs: u64,
/// The estate is sampled at most this often ([`sample_when_due`]).
pub sample_every_secs: u64,
}
/// **At least an hour between two changes of the twin**, whatever
/// `cooldown_secs` says: a resize stops the twin, and two in an hour is a
/// twin that spends its time catching up from the last one.
pub const MIN_SPACING_SECS: u64 = 3_600;
impl Default for TwinRule {
fn default() -> Self {
TwinRule {
lag_high: 10_000,
lag_low: 5_000,
load_high_pct: 80,
load_low_pct: 60,
sustain: 6,
cooldown_secs: 6 * 3_600,
sample_every_secs: 600,
}
}
}
impl TwinRule {
/// The spacing actually kept between two changes.
pub fn spacing_secs(&self) -> u64 {
self.cooldown_secs.max(MIN_SPACING_SECS)
}
/// **How many samples, newest last, the twin has been behind for** — with
/// hysteresis: a sample at or over the high marks starts (or continues) a
/// run; one at or over the low marks only continues one already running.
pub fn high_run(&self, samples: &[TwinSample]) -> usize {
let mut run = 0;
for s in samples {
let enters = s.lag_entries >= self.lag_high || s.load_pct() >= self.load_high_pct;
let stays = s.lag_entries >= self.lag_low.min(self.lag_high)
|| s.load_pct() >= self.load_low_pct.min(self.load_high_pct);
run = if enters || (run > 0 && stays) {
run + 1
} else {
0
};
}
run
}
}
/// What [`decide`] said.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum TwinVerdict {
/// Stay as it is, and why.
Hold(String),
/// Grow to this plan, and why.
Grow { to: PlanSpec, why: String },
}
/// The estate's numbers [`decide`] weighs a growth against.
pub struct TwinContext<'a> {
pub twin: &'a PlanSpec,
pub main: &'a PlanSpec,
pub catalog: &'a [PlanSpec],
/// When the twin last changed plan (0: never).
pub last_change_ms: u64,
pub now_ms: u64,
/// The estate's provider cost a month now, and its spend cap (minor; 0 = none).
pub cost_minor: u64,
pub cap_minor: u64,
}
/// **Grow the twin one plan step, or hold** — the damped rule above.
pub fn decide(rule: &TwinRule, samples: &[TwinSample], cx: &TwinContext<'_>) -> TwinVerdict {
let n = rule.sustain.max(1);
if samples.len() < n {
return TwinVerdict::Hold(format!("{} sample(s), {n} make a trend", samples.len()));
}
let run = rule.high_run(samples);
if run < n {
return TwinVerdict::Hold(format!(
"not behind for long: {run} high sample(s) in a row, {n} make a trend"
));
}
if cx.last_change_ms > 0
&& cx.now_ms
< cx.last_change_ms
.saturating_add(rule.spacing_secs().saturating_mul(1_000))
{
return TwinVerdict::Hold(format!(
"the twin changed plan {} s ago; the cooldown is {} s",
cx.now_ms.saturating_sub(cx.last_change_ms) / 1_000,
rule.spacing_secs()
));
}
// ★ Not at the top of a rising lag: a resize stops the twin.
let window = &samples[samples.len() - run..];
let last = window[window.len() - 1];
let prev = window[window.len() - 2.min(window.len())];
let peak = window.iter().map(|s| s.lag_entries).max().unwrap_or(0);
let rising_at_peak =
window.len() >= 2 && last.lag_entries > prev.lag_entries && last.lag_entries >= peak;
if rising_at_peak && run < 2 * n {
return TwinVerdict::Hold(format!("the twin is furthest behind now ({} entries and rising): a resize would stop it at its worst", last.lag_entries));
}
// One step: the smallest plan bigger than the twin's, never past main's.
let bigger = |p: &PlanSpec| {
(p.cores, p.memory_mib) > (cx.twin.cores, cx.twin.memory_mib)
&& p.cores >= cx.twin.cores
&& p.memory_mib >= cx.twin.memory_mib
};
let within_main = |p: &PlanSpec| p.cores <= cx.main.cores && p.memory_mib <= cx.main.memory_mib;
let Some(to) = cx
.catalog
.iter()
.filter(|p| bigger(p) && within_main(p))
.min_by_key(|p| {
(
p.cores,
p.memory_mib,
p.price.minor_per_month,
p.name.clone(),
)
})
else {
return TwinVerdict::Hold(format!(
"the twin ({}) is as big as main ({}) allows",
cx.twin.name, cx.main.name
));
};
let after = cx
.cost_minor
.saturating_sub(cx.twin.price.minor_per_month)
.saturating_add(to.price.minor_per_month);
if cx.cap_minor > 0 && after > cx.cap_minor {
let m = |minor| {
monetize_pay::Money {
minor,
currency: to.price.currency,
}
.in_words()
};
return TwinVerdict::Hold(format!(
"growing the twin to {} would cost {} a month, over the estate's spend cap of {}",
to.name,
m(after),
m(cx.cap_minor)
));
}
let why = if last.lag_entries >= rule.lag_high {
format!(
"{run} samples behind (lag {} entries, the rule {})",
last.lag_entries, rule.lag_high
)
} else {
format!(
"{run} samples busy ({} % of its cores, the rule {} %)",
last.load_pct(),
rule.load_high_pct
)
};
TwinVerdict::Grow {
to: to.clone(),
why,
}
}
/// **What a source is asked with**: the estate, the time, where its run lives (its window
/// file), and the pull channel's store — what the estate's boxes last reported.
pub struct SampleAt<'a> {
pub company: &'a str,
pub apex: &'a str,
pub now_ms: u64,
/// `<estate-orders>/<company>`: `laid.json`, `twin-samples.json`.
pub run_dir: &'a Path,
/// The guardian's store (its boxes' newest reports); `None`: no pull channel here.
pub reports: Option<&'a Store>,
/// [`TwinRule::sample_every_secs`].
pub every_ms: u64,
/// **The estate's twin and the role it mirrors** (P3-FACTS: the product's facts' `twin`;
/// Gunnar's and Holger's `twin` mirroring `main`): whose reports carry the twin's review.
pub twin: &'a TwinFacts,
}
/// **Where the twin's samples come from.** The window, oldest first, with a
/// new sample taken when one is due.
pub trait TwinHealth: Send + Sync {
fn samples(&self, at: &SampleAt<'_>) -> Result<Vec<TwinSample>, String>;
}
/// No source: the twin keeps its plan, and the job says why.
pub struct NoTwinHealth;
impl TwinHealth for NoTwinHealth {
fn samples(&self, _at: &SampleAt<'_>) -> Result<Vec<TwinSample>, String> {
Err("no twin health source is linked here".into())
}
}
/// **The tests' source**: samples set per estate.
#[derive(Default)]
pub struct MockTwinHealth(pub Mutex<BTreeMap<String, Vec<TwinSample>>>);
impl MockTwinHealth {
pub fn set(&self, company: &str, samples: Vec<TwinSample>) {
self.0
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(company.to_string(), samples);
}
}
impl TwinHealth for MockTwinHealth {
fn samples(&self, at: &SampleAt<'_>) -> Result<Vec<TwinSample>, String> {
Ok(self
.0
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(at.company)
.cloned()
.unwrap_or_default())
}
}
/// The window file in an estate's run directory.
pub const WINDOW_FILE: &str = "twin-samples.json";
/// The window keeps at most this many samples.
pub const WINDOW_MAX: usize = 64;
/// **An estate's window**: the samples kept across restarts, oldest first.
pub fn window(run_dir: &Path) -> Vec<TwinSample> {
std::fs::read(run_dir.join(WINDOW_FILE))
.ok()
.and_then(|b| serde_json::from_slice(&b).ok())
.unwrap_or_default()
}
/// Add one sample to the window (the oldest dropped past [`WINDOW_MAX`]).
pub fn keep(run_dir: &Path, sample: TwinSample) -> Result<Vec<TwinSample>, String> {
let mut w = window(run_dir);
w.push(sample);
if w.len() > WINDOW_MAX {
w.drain(..w.len() - WINDOW_MAX);
}
let bytes = serde_json::to_vec(&w).map_err(|e| e.to_string())?;
super::write_atomic(&run_dir.join(WINDOW_FILE), &bytes)?;
Ok(w)
}
/// **One reading of the twin**: its lag and load, as the estate reported them.
#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize)]
pub struct TwinReading {
#[serde(default)]
pub lag_entries: u64,
#[serde(default)]
pub cpu_millicores_busy: Option<u64>,
#[serde(default)]
pub cpu_millicores_total: Option<u64>,
/// `ok` | `behind` | `silent` | `none` — a silent twin is not a reading.
#[serde(default)]
pub state: String,
}
impl TwinReading {
/// The last line of a text that is a reading (one JSON object per line: a bench's or a
/// test's file).
pub fn parse(out: &str) -> Result<TwinReading, String> {
out.lines()
.rev()
.map(str::trim)
.filter(|l| l.starts_with('{'))
.find_map(|l| serde_json::from_str::<TwinReading>(l).ok())
.ok_or_else(|| {
format!(
"no twin reading (the last line: {:?})",
out.lines()
.last()
.unwrap_or("")
.chars()
.take(200)
.collect::<String>()
)
})
}
/// As a sample at `at_ms`; `None` for a silent twin (nothing was read off
/// it) and for an estate whose primary names no twin (`none`).
pub fn sample(&self, at_ms: u64) -> Option<TwinSample> {
if self.state == "silent" || self.state == "none" {
return None;
}
Some(TwinSample {
at_ms,
lag_entries: self.lag_entries,
cpu_millicores_busy: self.cpu_millicores_busy.unwrap_or(0),
cpu_millicores: self.cpu_millicores_total.unwrap_or(0),
})
}
}
/// **A reader's window, read when due**: the estate's window as kept, with `read` asked for
/// one new reading only when the last sample is `every_ms` old (or there is none) — kept in
/// the window; a twin with no reading (silent, none) is an error by name and keeps nothing.
pub fn sample_when_due(
at: &SampleAt<'_>,
read: impl FnOnce() -> Result<TwinReading, String>,
) -> Result<Vec<TwinSample>, String> {
let w = window(at.run_dir);
let due = w
.last()
.is_none_or(|l| at.now_ms >= l.at_ms.saturating_add(at.every_ms));
if !due {
return Ok(w);
}
let r = read()?;
match r.sample(at.now_ms) {
Some(s) => keep(at.run_dir, s),
None => Err(format!(
"the twin is {} on main's heartbeat: no reading to keep",
if r.state == "none" {
"not named (none)"
} else {
"silent"
}
)),
}
}
/// **The shop's twin reader: the estate's own reports** (G12, the pull channel). The newest
/// `twin_review` an estate's main or twin box reported ([`TwinReading::of_reports`]), sampled
/// when due and only when it is newer than the last sample.
pub struct ReportedTwinHealth;
/// **The roles whose reports may carry the twin's review** (P3-FACTS): the role the twin
/// mirrors, then the twin — read off the product's facts' `twin`, never a fixed `main`/`twin`.
pub fn reviewing_roles(twin: &TwinFacts) -> [&str; 2] {
[twin.mirrors.as_str(), twin.role.as_str()]
}
impl TwinReading {
/// **The estate's own word on its twin**: `(heard_unix_ms, reading)` of the newest report
/// that carried a `twin_review`, or (F4, a sold estate's twin, every report) the twin box's
/// own words `lag_entries` / `twin_state` (gunnar-server `repair::twin_words`) when they are
/// newer — a review rides ONE report, the twin's words every one. Refused by name when
/// neither was reported.
pub fn of_reports(
store: &Store,
instance: &str,
of: &TwinFacts,
) -> Result<(u64, TwinReading), String> {
let records: Vec<(&str, monetize_guardian::BoxRecord)> = reviewing_roles(of)
.into_iter()
.filter_map(|r| store.box_record(instance, r).map(|rec| (r, rec)))
.collect();
let twin = records
.iter()
.find(|(r, _)| *r == of.role)
.map(|(_, rec)| rec);
let words = twin.map(|rec| &rec.report.health);
let word = |k: &str| {
words
.and_then(|w| w.get(k))
.and_then(|v| v.trim().parse::<u64>().ok())
};
let (cpu_busy, cpu_total) = (word(TWIN_WORD_CPU_BUSY), word(TWIN_WORD_CPU_TOTAL));
let state_of = |s: &str| match s {
"" => "none".to_string(),
"unreachable" => "silent".to_string(),
s => s.to_string(),
};
// The twin's own words (heard with its report): its lag every report.
let from_words = twin.zip(word(TWIN_WORD_LAG)).map(|(rec, lag)| {
let said = |k: &str| rec.report.health.get(k).map(String::as_str).unwrap_or("");
let state = if said("primary") == "silent" {
"silent".to_string()
} else {
state_of(said(TWIN_WORD_STATE))
};
(
rec.heard_unix_ms,
TwinReading {
lag_entries: lag,
cpu_millicores_busy: cpu_busy,
cpu_millicores_total: cpu_total,
state,
},
)
});
let from_reviews = records.iter().filter_map(|(_, rec)| {
let text = rec
.report
.twin_review
.as_deref()
.filter(|t| !t.trim().is_empty())?;
let review = serde_json::from_str::<serde_json::Value>(text).ok()?;
// A review that names no seq pair is no lag reading (a box that reviews its twin
// in other words): the twin's own words stand, never a lag of 0.
if review.get("primary_head_seq").is_none()
&& review.get("applied_seq").is_none()
&& review.get("state").is_none()
{
return None;
}
let seq = |k: &str| {
review
.get(k)
.and_then(serde_json::Value::as_u64)
.unwrap_or(0)
};
let state = state_of(
review
.get("state")
.and_then(serde_json::Value::as_str)
.unwrap_or(""),
);
Some((
rec.heard_unix_ms,
TwinReading {
lag_entries: seq("primary_head_seq").saturating_sub(seq("applied_seq")),
cpu_millicores_busy: cpu_busy,
cpu_millicores_total: cpu_total,
state,
},
))
});
// A review wins a tie (it is the deeper look).
from_words
.into_iter()
.chain(from_reviews)
.max_by_key(|(heard, _)| *heard)
.ok_or_else(|| format!("REFUSED [twin-health-no-report] {instance} has reported no twin review and its twin no lag on the pull channel (Fleet.Report twin_review / health {TWIN_WORD_LAG})"))
}
}
/// The twin box's own health words (gunnar-server `repair::twin_words`, every report).
pub const TWIN_WORD_STATE: &str = "twin_state";
pub const TWIN_WORD_LAG: &str = "lag_entries";
pub const TWIN_WORD_CPU_BUSY: &str = "cpu_millicores_busy";
pub const TWIN_WORD_CPU_TOTAL: &str = "cpu_millicores_total";
impl TwinHealth for ReportedTwinHealth {
fn samples(&self, at: &SampleAt<'_>) -> Result<Vec<TwinSample>, String> {
let store = at
.reports
.ok_or("no pull channel here: the estate's reports cannot be read")?;
sample_when_due(at, || {
let (heard, reading) = TwinReading::of_reports(store, at.company, at.twin)?;
match window(at.run_dir).last() {
Some(last) if heard <= last.at_ms => Err(format!(
"{} has not reported its twin since the last sample",
at.company
)),
_ => Ok(reading),
}
})
}
}
#[cfg(test)]
mod tests {
use super::*;
/// Gunnar's and Holger's twin: `twin`, mirroring `main`.
static GUNNARS_TWIN: std::sync::LazyLock<TwinFacts> = std::sync::LazyLock::new(|| TwinFacts {
role: "twin".into(),
mirrors: "main".into(),
});
use monetize_cloud::Price;
use monetize_pay::Currency;
fn plan(name: &str, cores: u32, gib: u64, minor: u64) -> PlanSpec {
PlanSpec {
name: name.into(),
cores,
memory_mib: gib * 1024,
price: Price {
minor_per_month: minor,
currency: Currency::Eur,
},
}
}
fn catalog() -> Vec<PlanSpec> {
vec![
plan("P1", 1, 2, 1_000),
plan("P2", 2, 4, 2_000),
plan("P4", 4, 8, 4_000),
plan("P16", 16, 64, 16_000),
]
}
fn s(at: u64, lag: u64, busy_pct: u64) -> TwinSample {
TwinSample {
at_ms: at * 60_000,
lag_entries: lag,
cpu_millicores_busy: busy_pct * 10,
cpu_millicores: 1_000,
}
}
fn cx<'a>(
cat: &'a [PlanSpec],
twin: usize,
main: usize,
last: u64,
cap: u64,
) -> TwinContext<'a> {
TwinContext {
twin: &cat[twin],
main: &cat[main],
catalog: cat,
last_change_ms: last,
now_ms: 100 * 3_600_000,
cost_minor: 30_000,
cap_minor: cap,
}
}
fn grows(v: &TwinVerdict) -> Option<&str> {
match v {
TwinVerdict::Grow { to, .. } => Some(to.name.as_str()),
TwinVerdict::Hold(_) => None,
}
}
/// Lag that stays high, and has turned (no longer rising at its peak): one step up.
#[test]
fn a_twin_that_stays_behind_grows_one_plan_step() {
let cat = catalog();
let samples: Vec<_> = [12_000, 15_000, 18_000, 20_000, 19_000, 17_000]
.iter()
.enumerate()
.map(|(i, l)| s(i as u64, *l, 20))
.collect();
assert_eq!(
grows(&decide(
&TwinRule::default(),
&samples,
&cx(&cat, 0, 3, 0, 0)
)),
Some("P2"),
"one step, not a leap to main's size"
);
}
/// Load, not lag: sustained busy cores grow it too.
#[test]
fn a_twin_whose_cores_stay_busy_grows() {
let cat = catalog();
let samples: Vec<_> = (0..6).map(|i| s(i, 100, 95)).collect();
assert_eq!(
grows(&decide(
&TwinRule::default(),
&samples,
&cx(&cat, 1, 3, 0, 0)
)),
Some("P4")
);
}
/// A flapping load is not a trend; nor is a short spike.
#[test]
fn a_flapping_or_short_spike_does_not_grow_it() {
let cat = catalog();
let flap: Vec<_> = (0..12)
.map(|i| s(i, if i % 2 == 0 { 50_000 } else { 10 }, 10))
.collect();
assert_eq!(
grows(&decide(&TwinRule::default(), &flap, &cx(&cat, 0, 3, 0, 0))),
None
);
let short: Vec<_> = (0..12)
.map(|i| s(i, if i >= 9 { 50_000 } else { 10 }, 10))
.collect();
assert_eq!(
grows(&decide(&TwinRule::default(), &short, &cx(&cat, 0, 3, 0, 0))),
None
);
assert_eq!(
grows(&decide(
&TwinRule::default(),
&short[..3],
&cx(&cat, 0, 3, 0, 0)
)),
None,
"too few samples"
);
}
/// Never at the top of a rising lag — unless it has stayed behind for
/// twice the window (then it will not catch up by itself).
#[test]
fn it_does_not_stop_the_twin_while_it_is_furthest_behind() {
let cat = catalog();
let rising: Vec<_> = (0..6).map(|i| s(i, 11_000 + i * 2_000, 20)).collect();
let v = decide(&TwinRule::default(), &rising, &cx(&cat, 0, 3, 0, 0));
assert!(
matches!(&v, TwinVerdict::Hold(w) if w.contains("furthest behind")),
"{v:?}"
);
let long: Vec<_> = (0..12).map(|i| s(i, 11_000 + i * 2_000, 20)).collect();
assert_eq!(
grows(&decide(&TwinRule::default(), &long, &cx(&cat, 0, 3, 0, 0))),
Some("P2"),
"behind for twice the window: grow anyway"
);
}
/// The cooldown holds a second change; the cap and main's size bound it.
#[test]
fn the_cooldown_the_spend_cap_and_mains_size_hold_it() {
let cat = catalog();
let behind: Vec<_> = (0..6).map(|i| s(i, 30_000 - i * 100, 20)).collect();
let now = 100 * 3_600_000;
let v = decide(
&TwinRule::default(),
&behind,
&cx(&cat, 0, 3, now - 3_600_000, 0),
);
assert!(
matches!(&v, TwinVerdict::Hold(w) if w.contains("cooldown")),
"{v:?}"
);
assert_eq!(
grows(&decide(
&TwinRule::default(),
&behind,
&cx(&cat, 0, 3, now - 7 * 3_600_000, 0)
)),
Some("P2"),
"past the cooldown"
);
let v = decide(&TwinRule::default(), &behind, &cx(&cat, 0, 3, 0, 30_500));
assert!(
matches!(&v, TwinVerdict::Hold(w) if w.contains("would cost 310.00 EUR a month, over the estate's spend cap of 305.00 EUR")),
"30 000 − 1 000 + 2 000 is over 30 500: {v:?}"
);
let v = decide(&TwinRule::default(), &behind, &cx(&cat, 1, 1, 0, 0));
assert!(
matches!(&v, TwinVerdict::Hold(w) if w.contains("as big as main")),
"{v:?}"
);
}
/// ★ **Hysteresis**: once behind, a twin stays behind while its lag is
/// over `lag_low` — a lag hovering just under `lag_high` after a breach is
/// ONE trend, not a flap — and a fall under `lag_low` ends it.
#[test]
fn a_lag_hovering_between_the_marks_is_one_trend_and_under_the_low_mark_ends_it() {
let rule = TwinRule::default();
let cat = catalog();
// 12 000 starts it; 9 000 (between 5 000 and 10 000) keeps it.
let hover: Vec<_> = [12_000, 9_000, 9_500, 9_200, 9_100, 9_000]
.iter()
.enumerate()
.map(|(i, l)| s(i as u64, *l, 10))
.collect();
assert_eq!(rule.high_run(&hover), 6);
assert_eq!(
grows(&decide(&rule, &hover, &cx(&cat, 0, 3, 0, 0))),
Some("P2")
);
// 9 000s with no breach first are not behind at all.
let never: Vec<_> = (0..6).map(|i| s(i, 9_000, 10)).collect();
assert_eq!(rule.high_run(&never), 0);
// Under the low mark: the run ends, and a new breach starts over.
let broken: Vec<_> = [12_000, 9_000, 4_000, 12_000, 9_000, 9_000]
.iter()
.enumerate()
.map(|(i, l)| s(i as u64, *l, 10))
.collect();
assert_eq!(rule.high_run(&broken), 3);
assert_eq!(grows(&decide(&rule, &broken, &cx(&cat, 0, 3, 0, 0))), None);
}
/// ★ **At least an hour between two changes**, whatever the config says:
/// a cooldown of one minute still waits the hour.
#[test]
fn two_changes_are_never_closer_than_an_hour() {
let rule = TwinRule {
cooldown_secs: 60,
..TwinRule::default()
};
assert_eq!(rule.spacing_secs(), MIN_SPACING_SECS);
let cat = catalog();
let behind: Vec<_> = (0..6).map(|i| s(i, 30_000 - i * 100, 20)).collect();
let now = 100 * 3_600_000;
let v = decide(&rule, &behind, &cx(&cat, 0, 3, now - 30 * 60_000, 0));
assert!(
matches!(&v, TwinVerdict::Hold(w) if w.contains("cooldown is 3600 s")),
"half an hour after the last: {v:?}"
);
assert_eq!(
grows(&decide(
&rule,
&behind,
&cx(&cat, 0, 3, now - 61 * 60_000, 0)
)),
Some("P2"),
"past the hour"
);
}
/// A reading is its sample; a silent twin and an estate with no twin are none; a text's
/// last JSON line is its reading.
#[test]
fn a_reading_is_its_sample_and_a_silent_twin_is_none() {
let out = "review: twin 10.6.0.3\n{\"lag_entries\": 12, \"cpu_millicores_busy\": 450, \"cpu_millicores_total\": 1000, \"state\": \"ok\"}\n";
let r = TwinReading::parse(out).unwrap();
assert_eq!(
r.sample(7),
Some(TwinSample {
at_ms: 7,
lag_entries: 12,
cpu_millicores_busy: 450,
cpu_millicores: 1_000
})
);
assert_eq!(
TwinReading {
state: "none".into(),
..r.clone()
}
.sample(7),
None,
"no twin named: no sample"
);
assert_eq!(
TwinReading {
state: "silent".into(),
..r
}
.sample(7),
None
);
assert!(TwinReading::parse("REFUSED no answer")
.unwrap_err()
.contains("no twin reading"));
}
fn report(
twin_review: Option<serde_json::Value>,
health: &[(&str, &str)],
) -> monetize_embed::fleet::Report {
monetize_embed::fleet::Report {
build: "test".into(),
health: health
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
twin_review: twin_review.map(|v| v.to_string()),
..Default::default()
}
}
/// A `gunnar_repair::TwinReview`, as a box reports it: the twin furthest behind by `lag`.
fn review(state: &str, lag: u64) -> serde_json::Value {
serde_json::json!({"twin_addr": "10.6.0.3:7443", "health": "behind", "head_seq": 9_000, "first_seq": 1, "primary_head_seq": 10_000 + lag, "applied_seq": 10_000, "state": state, "why": "", "recharge_possible": true, "verdict": "", "refusal": "", "reviewed_unix_ms": 1})
}
/// ★ **G12: the twin is read from the estate's OWN reports** (the pull channel: monetize
/// dials no door, runs no command): the newest `twin_review` a main or twin box reported
/// gives the lag (`primary_head_seq − applied_seq`), the twin box's health words the load.
#[test]
fn the_twin_is_read_from_the_estates_own_reports() {
let store = Store::in_memory().unwrap();
store.declare("acme", "gunnar").unwrap();
store
.record_report(
"acme",
"twin",
1_000,
&report(
None,
&[
("primary", "live"),
("cpu_millicores_busy", "1900"),
("cpu_millicores_total", "2000"),
],
),
)
.unwrap();
store
.record_report(
"acme",
"main",
1_500,
&report(Some(review("following", 25_000)), &[("serving", "yes")]),
)
.unwrap();
let (heard, r) = TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN).unwrap();
assert_eq!(heard, 1_500, "heard when the review's report arrived");
assert_eq!(
r.sample(9),
Some(TwinSample {
at_ms: 9,
lag_entries: 25_000,
cpu_millicores_busy: 1_900,
cpu_millicores: 2_000
})
);
// The twin's own review, newer, wins; an unreachable primary is a silent twin.
store
.record_report(
"acme",
"twin",
2_000,
&report(Some(review("unreachable", 0)), &[]),
)
.unwrap();
assert_eq!(
TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN)
.unwrap()
.1
.state,
"silent"
);
let dir =
std::env::temp_dir().join(format!("monetize-twin-reported-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let fresh = Store::in_memory().unwrap();
fresh
.record_report(
"acme",
"main",
1_500,
&report(Some(review("following", 7)), &[]),
)
.unwrap();
let at = |now_ms| SampleAt {
company: "acme",
apex: "gunnar.rs",
now_ms,
run_dir: &dir,
reports: Some(&fresh),
every_ms: 600_000,
twin: &GUNNARS_TWIN,
};
assert_eq!(
ReportedTwinHealth.samples(&at(2_000)).unwrap(),
vec![TwinSample {
at_ms: 2_000,
lag_entries: 7,
cpu_millicores_busy: 0,
cpu_millicores: 0
}],
"no load words: no load"
);
assert_eq!(
ReportedTwinHealth.samples(&at(3_000)).unwrap().len(),
1,
"not due: the window as kept"
);
let e = ReportedTwinHealth.samples(&at(700_000)).unwrap_err();
assert!(
e.contains("not reported its twin since"),
"the same review is no new sample: {e}"
);
fresh
.record_report(
"acme",
"main",
650_000,
&report(Some(review("following", 3)), &[]),
)
.unwrap();
assert_eq!(
ReportedTwinHealth
.samples(&at(700_000))
.unwrap()
.iter()
.map(|s| s.lag_entries)
.collect::<Vec<_>>(),
[7, 3]
);
let _ = std::fs::remove_dir_all(&dir);
}
/// ★ F4 x G12: **a sold estate's twin is read off its OWN words every report** — a review
/// rides one report and is gone with the next; the twin's `lag_entries` / `twin_state`
/// (gunnar-server `repair::twin_words`) come with each, and the newer word wins.
#[test]
fn a_twins_own_words_read_its_lag_on_every_report() {
let store = Store::in_memory().unwrap();
store.declare("acme", "gunnar").unwrap();
store
.record_report(
"acme",
"main",
1_000,
&report(Some(review("following", 50)), &[("serving", "yes")]),
)
.unwrap();
store
.record_report(
"acme",
"twin",
2_000,
&report(
None,
&[
("primary", "live"),
(TWIN_WORD_STATE, "following"),
(TWIN_WORD_LAG, "40"),
(TWIN_WORD_CPU_BUSY, "420"),
(TWIN_WORD_CPU_TOTAL, "2000"),
],
),
)
.unwrap();
let (heard, r) = TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN).unwrap();
assert_eq!(
(heard, r.sample(9)),
(
2_000,
Some(TwinSample {
at_ms: 9,
lag_entries: 40,
cpu_millicores_busy: 420,
cpu_millicores: 2_000
})
),
"the twin's newer word"
);
store
.record_report(
"acme",
"main",
3_000,
&report(Some(review("following", 70)), &[]),
)
.unwrap();
assert_eq!(
TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN)
.unwrap()
.1
.lag_entries,
70,
"a newer review"
);
store
.record_report(
"acme",
"twin",
4_000,
&report(
None,
&[
("primary", "silent"),
(TWIN_WORD_STATE, "unreachable"),
(TWIN_WORD_LAG, "0"),
],
),
)
.unwrap();
assert_eq!(
TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN)
.unwrap()
.1
.state,
"silent",
"a twin that hears no primary is no reading"
);
let only = Store::in_memory().unwrap();
only.record_report(
"acme",
"twin",
5_000,
&report(
None,
&[
("primary", "live"),
(TWIN_WORD_STATE, "following"),
(TWIN_WORD_LAG, "5"),
],
),
)
.unwrap();
assert_eq!(
TwinReading::of_reports(&only, "acme", &GUNNARS_TWIN).unwrap(),
(
5_000,
TwinReading {
lag_entries: 5,
cpu_millicores_busy: None,
cpu_millicores_total: None,
state: "following".into()
}
),
"no review ever: the words alone"
);
}
/// ★ **The twin is the product's, by its facts** (P3-FACTS): a product whose twin is
/// `mirror`, mirroring `vault`, is read from those two boxes' reports — never from a fixed
/// `main` and `twin` (before, its reports were never read, and it never grew).
#[test]
fn the_twin_is_read_from_the_roles_the_products_facts_name() {
let store = Store::in_memory().unwrap();
store.declare("acme", "pair").unwrap();
store
.record_report(
"acme",
"vault",
1_000,
&report(Some(review("following", 30)), &[("serving", "yes")]),
)
.unwrap();
store
.record_report(
"acme",
"mirror",
2_000,
&report(
None,
&[
("primary", "live"),
(TWIN_WORD_STATE, "behind"),
(TWIN_WORD_LAG, "900"),
],
),
)
.unwrap();
let mirror = TwinFacts {
role: "mirror".into(),
mirrors: "vault".into(),
};
assert_eq!(reviewing_roles(&mirror), ["vault", "mirror"]);
let (heard, r) = TwinReading::of_reports(&store, "acme", &mirror).unwrap();
assert_eq!(
(heard, r.lag_entries, r.state.as_str()),
(2_000, 900, "behind"),
"the mirror's own words"
);
assert!(
TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN)
.unwrap_err()
.contains("[twin-health-no-report]"),
"Gunnar's main and twin are not this product's boxes"
);
}
/// ★ **M2: a Holger twin's report reads its real lag** — the pattern's words on every
/// report (`twin_state` from Holger's one follow rule: `ok | behind | stalled | silent`,
/// `lag_entries`, the CPU words) and the review's seqs; and a twin still on the review
/// that carries no seqs (a Holger box before M2) is read off its words, never as lag 0.
#[test]
fn a_holger_twins_report_reads_its_real_lag() {
let store = Store::in_memory().unwrap();
store.declare("acme", "holger").unwrap();
fn words<'a>(state: &'a str, lag: &'a str) -> Vec<(&'a str, &'a str)> {
vec![
("serving", "no"),
("data", "ok"),
("serves_at", "10.6.0.3:50051"),
("primary", "live"),
("complete", "yes"),
(TWIN_WORD_STATE, state),
(TWIN_WORD_LAG, lag),
(TWIN_WORD_CPU_BUSY, "1900"),
(TWIN_WORD_CPU_TOTAL, "2000"),
]
}
let holger_review = |state: &str, head: u64, applied: u64| serde_json::json!({"role": "twin", "complete": true, "why": "", "last_whole_unix": 1, "follow": "mtls 10.6.0.2:7443", "primary": "10.6.0.2:50051", "verdict": "the twin is whole", "reviewed_unix_ms": 1, "state": state, "primary_head_seq": head, "applied_seq": applied});
store
.record_report(
"acme",
"twin",
1_000,
&report(
Some(holger_review("behind", 30_000, 5_000)),
&words("behind", "25000"),
),
)
.unwrap();
let (heard, r) = TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN).unwrap();
assert_eq!(
(heard, r.sample(9)),
(
1_000,
Some(TwinSample {
at_ms: 9,
lag_entries: 25_000,
cpu_millicores_busy: 1_900,
cpu_millicores: 2_000
})
),
"{r:?}"
);
store
.record_report(
"acme",
"twin",
2_000,
&report(None, &words("stalled", "26000")),
)
.unwrap();
assert_eq!(
TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN)
.unwrap()
.1
.lag_entries,
26_000,
"a stalled twin is behind by its lag"
);
// A Holger twin before M2's seqs: its review names no seq pair, so it is no lag reading.
let old = Store::in_memory().unwrap();
let seqless = serde_json::json!({"role": "twin", "complete": true, "why": "", "last_whole_unix": 1, "follow": "", "primary": "10.6.0.2:50051", "verdict": "the twin is whole", "reviewed_unix_ms": 1});
old.record_report(
"acme",
"twin",
3_000,
&report(Some(seqless), &words("behind", "12000")),
)
.unwrap();
let (_, r) = TwinReading::of_reports(&old, "acme", &GUNNARS_TWIN).unwrap();
assert_eq!(
(r.lag_entries, r.state.as_str()),
(12_000, "behind"),
"the twin's words, not a seq-less review's 0: {r:?}"
);
// A Holger main's review of its twins (the roster's furthest behind) reads the same way.
let main = Store::in_memory().unwrap();
let roster = serde_json::json!({"role": "main", "twins": [{"address": "10.6.0.3:50051", "state": "behind", "lag_entries": 700, "disk_total_bytes": 1, "seen_ago_ms": 5}], "verdict": "the twin 10.6.0.3:50051 is behind (700 entries behind)", "reviewed_unix_ms": 1, "state": "behind", "primary_head_seq": 1_700, "applied_seq": 1_000});
main.record_report(
"acme",
"main",
4_000,
&report(
Some(roster),
&[("serving", "yes"), ("twin", "paired"), ("epoch", "1")],
),
)
.unwrap();
assert_eq!(
TwinReading::of_reports(&main, "acme", &GUNNARS_TWIN)
.unwrap()
.1
.lag_entries,
700
);
}
/// ★ **M2: a lagging Holger twin is queued for a resize** — its reports, sampled when due,
/// make the window the damped rule grows one plan step on.
#[test]
fn a_lagging_holger_twin_grows_one_plan_step() {
let dir = std::env::temp_dir().join(format!("monetize-twin-holger-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let store = Store::in_memory().unwrap();
store.declare("acme", "holger").unwrap();
let every = 600_000;
let mut samples = Vec::new();
for (i, lag) in [12_000u64, 15_000, 18_000, 20_000, 19_000, 17_000]
.iter()
.enumerate()
{
let at = (i as u64 + 1) * every;
let lag_s = lag.to_string();
store
.record_report(
"acme",
"twin",
at - 1,
&report(
None,
&[
("primary", "live"),
(TWIN_WORD_STATE, "behind"),
(TWIN_WORD_LAG, lag_s.as_str()),
],
),
)
.unwrap();
let sa = SampleAt {
company: "acme",
apex: "holger.rs",
now_ms: at,
run_dir: &dir,
reports: Some(&store),
every_ms: every,
twin: &GUNNARS_TWIN,
};
samples = ReportedTwinHealth.samples(&sa).unwrap();
}
assert_eq!(
samples.iter().map(|s| s.lag_entries).collect::<Vec<_>>(),
[12_000, 15_000, 18_000, 20_000, 19_000, 17_000]
);
let cat = catalog();
assert_eq!(
grows(&decide(
&TwinRule::default(),
&samples,
&cx(&cat, 0, 3, 0, 0)
)),
Some("P2"),
"the ONE box resizes Holger's twin"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// ★ No reports, no review, no pull channel: no sample, said by name.
#[test]
fn an_estate_that_reports_no_review_keeps_no_sample() {
let store = Store::in_memory().unwrap();
assert!(TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN)
.unwrap_err()
.contains("[twin-health-no-report]"));
store
.record_report("acme", "main", 1_000, &report(None, &[("serving", "yes")]))
.unwrap();
assert!(TwinReading::of_reports(&store, "acme", &GUNNARS_TWIN)
.unwrap_err()
.contains("[twin-health-no-report]"));
let dir =
std::env::temp_dir().join(format!("monetize-twin-noreport-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let none = SampleAt {
company: "acme",
apex: "gunnar.rs",
now_ms: 5,
run_dir: &dir,
reports: None,
every_ms: 600_000,
twin: &GUNNARS_TWIN,
};
assert!(ReportedTwinHealth
.samples(&none)
.unwrap_err()
.contains("no pull channel"));
assert!(window(&dir).is_empty(), "nothing kept");
let _ = std::fs::remove_dir_all(&dir);
}
/// ★ A reader is asked only when a sample is due, and a silent twin keeps nothing.
#[test]
fn a_reader_is_asked_only_when_a_sample_is_due() {
let dir = std::env::temp_dir().join(format!("monetize-twin-due-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let at = |now_ms| SampleAt {
company: "acme",
apex: "gunnar.rs",
now_ms,
run_dir: &dir,
reports: None,
every_ms: 600_000,
twin: &GUNNARS_TWIN,
};
let ok = || {
Ok(TwinReading {
lag_entries: 5,
cpu_millicores_busy: Some(10),
cpu_millicores_total: Some(1_000),
state: "ok".into(),
})
};
assert_eq!(sample_when_due(&at(1_000), ok).unwrap().len(), 1);
assert_eq!(
sample_when_due(&at(2_000), || panic!("not due: not asked"))
.unwrap()
.len(),
1
);
let silent = || {
Ok(TwinReading {
state: "silent".into(),
..ok().unwrap()
})
};
assert!(sample_when_due(&at(700_000), silent)
.unwrap_err()
.contains("silent"));
assert_eq!(window(&dir).len(), 1, "nothing kept of a silent twin");
let _ = std::fs::remove_dir_all(&dir);
}
/// The window survives a restart and keeps the newest [`WINDOW_MAX`].
#[test]
fn the_window_is_kept_on_disk_and_bounded() {
let dir = std::env::temp_dir().join(format!("monetize-twin-window-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
for i in 0..(WINDOW_MAX as u64 + 5) {
keep(&dir, s(i, i, 0)).unwrap();
}
let w = window(&dir);
assert_eq!(w.len(), WINDOW_MAX);
assert_eq!(w[0].lag_entries, 5, "the oldest went first");
let _ = std::fs::remove_dir_all(&dir);
}
}
/// R5a: the tests' twin health source answers after a panic that held it.
#[cfg(test)]
mod poisoned_mock_tests {
use super::*;
#[test]
fn a_poisoned_mock_source_still_sets() {
let m = MockTwinHealth::default();
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _held = m.0.lock();
panic!("a panic while the source is held");
}));
m.set("acme", Vec::new());
assert!(m
.0
.lock()
.unwrap_or_else(|e| e.into_inner())
.contains_key("acme"));
}
}