gunnar-grpc 2.0.0

//! **What a primary owes each of its twins** (cov-13, 2026-10-06) — kept on the
//! primary beside its transaction log, read by the retention pass and by the
//! two doors a twin follows through (`Mirror.LogFetch`, `Mirror.SnapshotFetch`).
//!
//! Two debts, both born while a twin was SILENT (its VM died, or it was too busy to
//! poll), both settled by the twin itself once it answers:
//!
//! * **A full resync** ([`Owed::resync_from`]). The log is bounded in bytes
//!   (`crate::retention`): a twin silent past its grace while the log is over its
//!   budget is no longer waited for — the log is trimmed to the budget and the
//!   twin, when it polls again, finds the primary retaining from past its own
//!   head and takes a snapshot (`Mirror.SnapshotFetch`), exactly as any twin
//!   behind a retention hole does. The record says so on the primary and is
//!   cleared when the twin reports a position inside the log again.
//! * **A growth** ([`Owed::grow_to_bytes`]). An EMERGENCY grows main ALONE while
//!   its twin is silent (monetize's `grow_alone`): main's data set is then larger
//!   than the twin's volume. A twin that came back and followed would fill its
//!   disk part-way through main's data — the worst place to stop. So the doors
//!   REFUSE it, by name ([`crate::TWIN_TOO_SMALL`]), until its own report says
//!   its volume holds main's set; it waits in `recovering`, polling (so it reads
//!   heard, and monetize's twin lane grows it), and follows the moment it fits.
//!
//! Only a growth that happened WHILE the twin was silent creates the second
//! debt: a twin that was already smaller than main and following is not
//! stopped by this (that is the twin lane's fill and capacity rules, as before).
//!
//! On disk: `<data-root>/twin-owed.json` (per box, never framed), through
//! [`crate::ControlPlane::twin_owed_write`]. JSON by hand: this crate links
//! `serde_json` and not serde's derive.

use std::collections::BTreeMap;

/// A filesystem is a little smaller than its volume (xfs keeps its log and
/// metadata inside it), and main's striped set and the twin's one volume carry
/// different overheads: 2 % either way is "the same size". Main must have grown
/// by more than this for a debt to be born, and the twin must hold at least
/// this much less than the debt for the doors to refuse it.
pub const TOLERANCE_PERCENT: u64 = 2;

/// What one twin is owed. All zero = nothing (the entry is dropped).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Owed {
    /// Wall clock this twin was first seen silent by the retention pass; 0 = heard.
    pub silent_since_ms: i64,
    /// Main's data volume (bytes, statvfs total) when this twin went silent.
    pub main_total_at_silence: u64,
    /// The twin's own data volume as it last reported it; 0 = never reported.
    pub twin_total: u64,
    /// Non-zero: main's set grew to this many bytes while the twin was silent — the
    /// doors refuse it until it reports a volume at least this large.
    pub grow_to_bytes: u64,
    /// Non-zero: the log was trimmed past this twin while it was silent and retains
    /// from here — it takes a full resync (snapshot) when it answers.
    pub resync_from: u64,
    /// The sentence the last debt was recorded with.
    pub why: String,
}

impl Owed {
    pub fn is_empty(&self) -> bool {
        self.silent_since_ms == 0 && self.grow_to_bytes == 0 && self.resync_from == 0
    }
}

/// Every twin's debts, by principal.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Book {
    pub twins: BTreeMap<String, Owed>,
}

impl Book {
    pub fn to_json(&self) -> Vec<u8> {
        let twins: serde_json::Map<String, serde_json::Value> = self
            .twins
            .iter()
            .filter(|(_, o)| !o.is_empty())
            .map(|(p, o)| {
                (
                    p.clone(),
                    serde_json::json!({
                        "silent_since_ms": o.silent_since_ms,
                        "main_total_at_silence": o.main_total_at_silence,
                        "twin_total": o.twin_total,
                        "grow_to_bytes": o.grow_to_bytes,
                        "resync_from": o.resync_from,
                        "why": o.why,
                    }),
                )
            })
            .collect();
        serde_json::to_vec_pretty(&serde_json::json!({ "v": 1, "twins": twins })).unwrap_or_default()
    }

    /// A book from its file; anything unreadable is an EMPTY book, said by the caller.
    pub fn from_json(bytes: &[u8]) -> Result<Book, String> {
        if bytes.is_empty() {
            return Ok(Book::default());
        }
        let v: serde_json::Value = serde_json::from_slice(bytes).map_err(|e| format!("twin-owed.json is not JSON: {e}"))?;
        let mut book = Book::default();
        if let Some(m) = v.get("twins").and_then(|t| t.as_object()) {
            for (p, o) in m {
                let u = |k: &str| o.get(k).and_then(|x| x.as_u64()).unwrap_or(0);
                let owed = Owed {
                    silent_since_ms: o.get("silent_since_ms").and_then(|x| x.as_i64()).unwrap_or(0),
                    main_total_at_silence: u("main_total_at_silence"),
                    twin_total: u("twin_total"),
                    grow_to_bytes: u("grow_to_bytes"),
                    resync_from: u("resync_from"),
                    why: o.get("why").and_then(|x| x.as_str()).unwrap_or_default().to_owned(),
                };
                if !owed.is_empty() {
                    book.twins.insert(p.clone(), owed);
                }
            }
        }
        Ok(book)
    }
}

/// `a` is more than [`TOLERANCE_PERCENT`] above `b`.
pub fn clearly_above(a: u64, b: u64) -> bool {
    u128::from(a) * 100 > u128::from(b) * u128::from(100 + TOLERANCE_PERCENT)
}

/// Bytes in whole-GiB words, for the sentences.
pub fn gib(bytes: u64) -> String {
    format!("{:.1} GiB", bytes as f64 / (1u64 << 30) as f64)
}

/// **Why the doors refuse this twin**, or `None`: it owes no growth, or its last
/// report already holds main's set (less the tolerance). `reported_total` 0 =
/// it has not reported a volume since this server started — refused, because a
/// debt it has not shown it paid is still a debt.
pub fn follow_refusal(owed: &Owed, reported_total: u64) -> Option<String> {
    if owed.grow_to_bytes == 0 || (reported_total > 0 && !clearly_above(owed.grow_to_bytes, reported_total)) {
        return None;
    }
    let holds = if reported_total == 0 { "has not reported its volume since this primary started".to_owned() } else { format!("holds {}", gib(reported_total)) };
    Some(format!(
        "{} this twin's data volume {holds} and main's data set grew to {} while it was silent: it must be grown to at least that before it follows again, or it would fill part-way through main's data. It waits in `recovering` and keeps polling; nothing was sent. monetize's twin lane grows it when it reads heard (or grow its volume by hand); it follows the moment its report holds main's set",
        crate::TWIN_TOO_SMALL,
        gib(owed.grow_to_bytes)
    ))
}

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

    const GIB: u64 = 1 << 30;

    #[test]
    fn a_book_survives_its_file_and_drops_what_is_owed_nothing() {
        let mut b = Book::default();
        b.twins.insert("gunnar-twin".into(), Owed { silent_since_ms: 5, main_total_at_silence: 20 * GIB, twin_total: 20 * GIB, grow_to_bytes: 30 * GIB, resync_from: 77, why: "x".into() });
        b.twins.insert("quiet".into(), Owed::default());
        let back = Book::from_json(&b.to_json()).unwrap();
        assert_eq!(back.twins.len(), 1, "{back:?}");
        assert_eq!(back.twins["gunnar-twin"], b.twins["gunnar-twin"]);
        assert_eq!(Book::from_json(b"").unwrap(), Book::default());
        assert!(Book::from_json(b"{not json").is_err());
    }

    #[test]
    fn a_twin_owed_a_growth_is_refused_until_its_report_holds_mains_set() {
        let owed = Owed { grow_to_bytes: 30 * GIB, ..Owed::default() };
        let why = follow_refusal(&owed, 20 * GIB).expect("20 GiB behind a 30 GiB set is refused");
        assert!(why.starts_with(crate::TWIN_TOO_SMALL) && why.contains("30.0 GiB") && why.contains("20.0 GiB") && why.contains("recovering"), "{why}");
        assert!(follow_refusal(&owed, 0).is_some(), "an unreported volume has not shown the debt paid");
        assert_eq!(follow_refusal(&owed, 30 * GIB), None);
        assert_eq!(follow_refusal(&owed, 30 * GIB - GIB / 4), None, "a filesystem a little smaller than its volume is the same size");
        assert_eq!(follow_refusal(&Owed::default(), GIB), None, "no debt, no refusal");
    }
}