gunnar-grpc 3.0.0

//! **The data set's grow journal, read** — `/serialdat/grow-journal`, the
//! record `korp-start` keeps while it grows `/data` into enlarged member
//! volumes (`DATA-SET-GROWTH-DESIGN.md` §5; the writer is edda's
//! `korp-installer::grow`). gunnar reads it for `Admin.Capacity` and
//! `Admin.Health` and never writes it.
//!
//! One line per fsynced record, exactly:
//!
//! ```text
//! epoch=<u64> target_sectors=<u64> member=<u32|-> step=<G1..G6|PARTIAL> at=<unix-secs>
//! ```
//!
//! This parser is the reader's twin of `korp_installer::grow::Record::line`.
//! It is spelled here rather than taken from that crate because gunnar
//! resolves edda's crates from the registry, and the engine is on an
//! unmerged branch; the line above is the contract, and the test below
//! holds a line the writer's own test asserts byte for byte.
//!
//! Lives in `gunnar-grpc` because `Admin.Health` is answered here and
//! `gunnar-server` (which answers `Capacity`) depends on this crate — one
//! reader, two callers.

use std::path::Path;

/// Where the boot stack keeps the journal (the boot disk's own partition,
/// off the stripe).
pub const JOURNAL_PATH: &str = "/serialdat/grow-journal";

/// One record, as read.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Record {
    pub epoch: u64,
    pub target_sectors: u64,
    pub member: Option<u32>,
    pub step: String,
    pub at: u64,
}

impl Record {
    /// Parse one line; `None` for a line that is not a record (a torn tail).
    pub fn parse(line: &str) -> Option<Record> {
        let mut epoch = None;
        let mut target = None;
        let mut member: Option<Option<u32>> = None;
        let mut step = None;
        let mut at = None;
        for tok in line.split_whitespace() {
            let (k, v) = tok.split_once('=')?;
            match k {
                "epoch" => epoch = Some(v.parse().ok()?),
                "target_sectors" => target = Some(v.parse().ok()?),
                "member" => {
                    member = Some(if v == "-" {
                        None
                    } else {
                        Some(v.parse().ok()?)
                    })
                }
                "step" => step = Some(v.to_owned()),
                "at" => at = Some(v.parse().ok()?),
                _ => return None,
            }
        }
        Some(Record {
            epoch: epoch?,
            target_sectors: target?,
            member: member?,
            step: step?,
            at: at?,
        })
    }

    /// `G2:1`, `G4`.
    pub fn label(&self) -> String {
        match self.member {
            Some(m) => format!("{}:{m}", self.step),
            None => self.step.clone(),
        }
    }
}

/// What the journal says the set is doing.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Phase {
    /// No grow has ever run on this box.
    Idle,
    /// An epoch is open (no `G6`): the next boot resumes it.
    InProgress,
    /// The last epoch closed at its target.
    Done,
    /// The last boot refused: only some members had grown.
    Partial,
    /// The last epoch was a RUNTIME growth that ended with nothing touched
    /// (`ABORT`: a drain that timed out, an unmount that was refused, an
    /// operator who gave up while waiting). Closed, like `done`.
    Aborted,
    /// The last epoch's runtime growth hit a recoverable failure AFTER a
    /// member was touched (`FAIL`): the set serves at the size it has and
    /// the epoch stays open — the next boot, or `grow resume`, finishes it.
    Failed,
}

impl Phase {
    pub fn as_str(self) -> &'static str {
        match self {
            Phase::Idle => "idle",
            Phase::InProgress => "in_progress",
            Phase::Done => "done",
            Phase::Partial => "partial",
            Phase::Aborted => "aborted",
            Phase::Failed => "failed",
        }
    }
}

/// The journal, summarised.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Summary {
    pub path: String,
    pub records: usize,
    /// Lines that did not parse (a torn tail from a crash mid-append).
    pub torn: usize,
    pub phase: Phase,
    pub epoch: Option<u64>,
    pub target_sectors: Option<u64>,
    /// The last epoch's records, `G1`, `G2:0`, … in order.
    pub steps: Vec<String>,
    /// Members the last PARTIAL verdict found larger than the set.
    pub partial_members: Vec<u32>,
}

impl Summary {
    /// Summarise the journal's text.
    pub fn of(text: &str, path: &str) -> Summary {
        let mut records = Vec::new();
        let mut torn = 0;
        for line in text.lines().filter(|l| !l.trim().is_empty()) {
            match Record::parse(line) {
                Some(r) => records.push(r),
                None => torn += 1,
            }
        }
        let trailing_partial: Vec<&Record> = {
            let mut v: Vec<&Record> = records
                .iter()
                .rev()
                .take_while(|r| r.step == "PARTIAL")
                .collect();
            v.reverse();
            v
        };
        let epoch = records
            .iter()
            .filter(|r| r.step != "PARTIAL")
            .map(|r| r.epoch)
            .max();
        let mine: Vec<&Record> = records
            .iter()
            .filter(|r| Some(r.epoch) == epoch && r.step != "PARTIAL")
            .collect();
        let target_sectors = mine.last().map(|r| r.target_sectors);
        let steps: Vec<String> = mine.iter().map(|r| r.label()).collect();
        let phase = if !trailing_partial.is_empty() {
            Phase::Partial
        } else if epoch.is_none() {
            Phase::Idle
        } else if mine.iter().any(|r| r.step == "ABORT") {
            Phase::Aborted
        } else if mine.iter().any(|r| r.step == "G6") {
            Phase::Done
        } else if mine.iter().any(|r| r.step == "FAIL") {
            Phase::Failed
        } else {
            Phase::InProgress
        };
        Summary {
            path: path.to_owned(),
            records: records.len(),
            torn,
            phase,
            epoch,
            target_sectors,
            steps,
            partial_members: trailing_partial.iter().filter_map(|r| r.member).collect(),
        }
    }

    /// Read and summarise the journal at `path`; `None` when there is no
    /// journal file — a boot stack without the engine.
    pub fn read(path: &Path) -> Option<Summary> {
        let text = std::fs::read_to_string(path).ok()?;
        Some(Summary::of(&text, &path.display().to_string()))
    }

    /// The box's journal, at [`JOURNAL_PATH`].
    pub fn measure() -> Option<Summary> {
        Summary::read(Path::new(JOURNAL_PATH))
    }

    /// One sentence for a human.
    pub fn detail(&self) -> String {
        let gib = |s: u64| format!("{:.1} GiB", (s as f64) * 512.0 / (1u64 << 30) as f64);
        let torn = if self.torn > 0 {
            format!(" ({} torn line(s) skipped)", self.torn)
        } else {
            String::new()
        };
        match self.phase {
            Phase::Idle => format!("no grow has run on this box (journal empty{torn})"),
            Phase::Partial => format!(
                "PARTIAL RESIZE: {} member(s) [{}] are larger than the set and the rest are not — NOT grown; enlarge the remaining members to the same size and restart{torn}",
                self.partial_members.len(),
                self.partial_members.iter().map(|m| m.to_string()).collect::<Vec<_>>().join(", ")
            ),
            Phase::InProgress => format!(
                "epoch {} IN PROGRESS toward {} per member — recorded so far: {}; the next boot (or `gunnar control grow resume`) resumes it{torn}",
                self.epoch.unwrap_or(0),
                self.target_sectors.map(gib).unwrap_or_default(),
                self.steps.join(" ")
            ),
            Phase::Done => format!(
                "epoch {} done: the set grew to {} per member ({}){torn}",
                self.epoch.unwrap_or(0),
                self.target_sectors.map(gib).unwrap_or_default(),
                self.steps.join(" ")
            ),
            Phase::Aborted => format!(
                "epoch {} ABORTED at runtime with nothing touched ({}); the set is as it was{torn}",
                self.epoch.unwrap_or(0),
                self.steps.join(" ")
            ),
            Phase::Failed => format!(
                "epoch {} FAILED at runtime after a member was touched ({}); /data serves at the size it has and the next boot (or `gunnar control grow resume`) finishes the epoch{torn}",
                self.epoch.unwrap_or(0),
                self.steps.join(" ")
            ),
        }
    }

    /// As it goes on the wire.
    pub fn to_wire(&self) -> gunnar_proto::GrowState {
        gunnar_proto::GrowState {
            engine_present: true,
            phase: self.phase.as_str().to_owned(),
            epoch: self.epoch.unwrap_or(0),
            target_sectors: self.target_sectors.unwrap_or(0),
            steps: self.steps.clone(),
            partial_members: self.partial_members.clone(),
            detail: self.detail(),
            journal_path: self.path.clone(),
        }
    }
}

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

    /// The writer's own asserted line (`korp_installer::grow` tests) parses
    /// here field for field; a torn tail is counted, not read.
    #[test]
    fn the_reader_parses_the_writers_exact_line_and_skips_a_torn_tail() {
        let r = Record::parse("epoch=3 target_sectors=268433408 member=1 step=G2 at=1789394521")
            .unwrap();
        assert_eq!(
            r,
            Record {
                epoch: 3,
                target_sectors: 268_433_408,
                member: Some(1),
                step: "G2".into(),
                at: 1_789_394_521
            }
        );
        assert_eq!(r.label(), "G2:1");
        let g4 = Record::parse("epoch=3 target_sectors=268433408 member=- step=G4 at=1789394521")
            .unwrap();
        assert_eq!((g4.member, g4.label()), (None, "G4".into()));
        assert!(Record::parse("epoch=3 target_sec").is_none());
        assert!(Record::parse("epoch=3 target_sectors=1 member=- step=G1 at=1 x=y").is_none());
    }

    /// The four phases, from journal text: idle, an open epoch (resumes),
    /// a closed epoch, a partial verdict — and a partial after a done epoch
    /// is partial, because the last boot's verdict is the one that stands.
    #[test]
    fn the_summary_names_idle_in_progress_done_and_partial() {
        let idle = Summary::of("", "j");
        assert_eq!(
            (idle.phase, idle.epoch, idle.records),
            (Phase::Idle, None, 0)
        );
        assert!(idle.detail().contains("no grow has run"));

        let open = "epoch=1 target_sectors=100 member=- step=G1 at=1\nepoch=1 target_sectors=100 member=0 step=G2 at=2\nepoch=1 target_sectors=100 member=1 step=G2 at=3\nepoch=1 targ";
        let s = Summary::of(open, "j");
        assert_eq!(
            (s.phase, s.epoch, s.target_sectors, s.torn),
            (Phase::InProgress, Some(1), Some(100), 1)
        );
        assert_eq!(s.steps, vec!["G1", "G2:0", "G2:1"]);
        assert!(
            s.detail().contains("IN PROGRESS")
                && s.detail().contains("G2:1")
                && s.detail().contains("1 torn"),
            "{}",
            s.detail()
        );

        let done = format!(
            "{}\nepoch=1 target_sectors=100 member=- step=G6 at=9\n",
            open.trim_end_matches("epoch=1 targ")
        );
        let s = Summary::of(&done, "j");
        assert_eq!(s.phase, Phase::Done);
        assert!(s.steps.last().map(String::as_str) == Some("G6"));
        let wire = s.to_wire();
        assert!(
            wire.engine_present
                && wire.phase == "done"
                && wire.epoch == 1
                && wire.target_sectors == 100
                && wire.steps.len() == 4
        );

        let partial = format!("{done}epoch=1 target_sectors=200 member=0 step=PARTIAL at=10\nepoch=1 target_sectors=200 member=2 step=PARTIAL at=10\n");
        let s = Summary::of(&partial, "j");
        assert_eq!(
            (s.phase, s.partial_members.clone()),
            (Phase::Partial, vec![0, 2])
        );
        assert!(
            s.detail().contains("PARTIAL RESIZE: 2 member(s) [0, 2]"),
            "{}",
            s.detail()
        );
        assert_eq!(s.epoch, Some(1), "a partial verdict opens no epoch");

        // a missing file is "no engine", not an empty journal
        assert!(Summary::read(Path::new("/nonexistent/grow-journal")).is_none());
    }

    /// **The runtime records** (`korp_installer::grow::Step`'s runtime steps,
    /// written by gunnar-server's growth): a prelude alone is an open epoch
    /// (`in_progress`, resumed by the next boot or `grow resume`); `ABORT`
    /// closes it with nothing touched; `FAIL` after a G-step is `failed` and
    /// stays open; `G6` after a runtime prelude is `done` like any other.
    #[test]
    fn the_summary_names_the_runtime_phases_aborted_and_failed() {
        let prelude = "epoch=2 target_sectors=0 member=- step=DRAIN at=1\nepoch=2 target_sectors=0 member=- step=UNMOUNT at=2\nepoch=2 target_sectors=0 member=- step=STOP at=3\nepoch=2 target_sectors=0 member=- step=WAIT at=4\n";
        let s = Summary::of(prelude, "j");
        assert_eq!((s.phase, s.epoch), (Phase::InProgress, Some(2)));
        assert_eq!(s.steps, vec!["DRAIN", "UNMOUNT", "STOP", "WAIT"]);
        assert!(s.detail().contains("grow resume"), "{}", s.detail());

        let aborted = format!("{prelude}epoch=2 target_sectors=0 member=- step=ABORT at=5\n");
        let s = Summary::of(&aborted, "j");
        assert_eq!(s.phase, Phase::Aborted);
        assert_eq!(s.to_wire().phase, "aborted");
        assert!(
            s.detail().contains("ABORTED") && s.detail().contains("nothing touched"),
            "{}",
            s.detail()
        );

        let failed = format!("{prelude}epoch=2 target_sectors=0 member=- step=RESCAN at=5\nepoch=2 target_sectors=500 member=- step=G1 at=6\nepoch=2 target_sectors=500 member=0 step=G2 at=7\nepoch=2 target_sectors=500 member=- step=FAIL at=8\nepoch=2 target_sectors=500 member=- step=REOPEN at=9\n");
        let s = Summary::of(&failed, "j");
        assert_eq!((s.phase, s.target_sectors), (Phase::Failed, Some(500)));
        assert!(
            s.detail().contains("FAILED") && s.detail().contains("serves at the size it has"),
            "{}",
            s.detail()
        );

        let done = format!("{prelude}epoch=2 target_sectors=0 member=- step=RESCAN at=5\nepoch=2 target_sectors=500 member=- step=G1 at=6\nepoch=2 target_sectors=500 member=- step=G6 at=7\nepoch=2 target_sectors=500 member=- step=REOPEN at=8\n");
        let s = Summary::of(&done, "j");
        assert_eq!(
            s.phase,
            Phase::Done,
            "G6 closes a runtime epoch like any other"
        );
    }

    /// ★ cov-08 (the WireGuard upgrade drill d5, 2026-10-05): the OLD line
    /// journaled `FAIL` for a growth given up with NO member back larger (the
    /// spend ceiling refused the cloud's resize after main drained), and main
    /// carried "epoch 2 FAILED … the next boot finishes the epoch" into every
    /// capacity answer after its upgrade. The journal alone cannot tell that
    /// FAIL from T9's (members resized by the cloud, the engine refusing before
    /// any G-step: the next boot must finish it), so it reads `failed` — the
    /// writer says ABORT now when nothing was touched (`grow::tests`) — and the
    /// estate's next growth, once the ceiling pays, closes its OWN epoch: the
    /// reading is `done` again.
    #[test]
    fn an_old_lines_failed_epoch_reads_done_once_the_next_growth_closes_its_own() {
        let d5 = "epoch=1 target_sectors=31457280 member=- step=DRAIN at=1\nepoch=1 target_sectors=31457280 member=- step=G6 at=2\nepoch=1 target_sectors=31457280 member=- step=REOPEN at=3\n\
                  epoch=2 target_sectors=130023424 member=- step=DRAIN at=4\nepoch=2 target_sectors=130023424 member=- step=UNMOUNT at=5\nepoch=2 target_sectors=130023424 member=- step=STOP at=6\n\
                  epoch=2 target_sectors=130023424 member=- step=WAIT at=7\nepoch=2 target_sectors=130023424 member=- step=RESCAN at=8\nepoch=2 target_sectors=130023424 member=- step=FAIL at=9\n\
                  epoch=2 target_sectors=130023424 member=- step=REOPEN at=10\n";
        assert_eq!(
            (Summary::of(d5, "j").phase, Summary::of(d5, "j").epoch),
            (Phase::Failed, Some(2))
        );
        let recovered = format!(
            "{d5}epoch=3 target_sectors=165675008 member=- step=DRAIN at=11\nepoch=3 target_sectors=165675008 member=- step=UNMOUNT at=12\nepoch=3 target_sectors=165675008 member=- step=STOP at=13\n\
             epoch=3 target_sectors=165675008 member=- step=WAIT at=14\nepoch=3 target_sectors=165675008 member=- step=RESCAN at=15\nepoch=3 target_sectors=165675008 member=- step=G1 at=16\n\
             epoch=3 target_sectors=165675008 member=0 step=G2 at=17\nepoch=3 target_sectors=165675008 member=- step=G6 at=18\nepoch=3 target_sectors=165675008 member=- step=REOPEN at=19\n"
        );
        let s = Summary::of(&recovered, "j");
        assert_eq!((s.phase, s.epoch), (Phase::Done, Some(3)), "{}", s.detail());
        assert!(!s.detail().contains("FAILED"), "{}", s.detail());
    }
}