knut-bifrost 0.3.0

//! Cumulative run metrics and the poll-loop's failure backoff schedule.
//!
//! Two small, dependency-light pieces that harden the [`Listener`](crate::Listener)
//! poll loop (plan.md §7):
//!
//! - [`Metrics`] — counters accumulated across every round (`rounds_total`,
//!   `rounds_failed`, `tables_synced`, `rows_ingested`, `vertices_written`,
//!   `edges_written`, `deletes`, `assets_emitted`). The listener folds each
//!   [`RoundReport`](crate::RoundReport) into it and emits it as a structured
//!   `metrics` tracing event after every round; [`render_prometheus`] turns it
//!   into Prometheus text (a pure function — no HTTP server here).
//! - [`Backoff`] — exponential backoff for the sleep *between* rounds: a fully
//!   failed / no-progress round doubles the sleep up to a cap, and the first
//!   round that makes progress resets it to the base interval.

use std::time::Duration;

use crate::RoundReport;

/// Counters accumulated over the lifetime of a [`Listener`](crate::Listener)
/// run. All monotonic (never decrease); folded one round at a time by
/// [`Metrics::record_round`]. Cheap to clone and `Copy`-free only because it is
/// a plain struct of `u64`s.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Metrics {
    /// Rounds attempted (every `sync_once` pass, including failures/idles).
    pub rounds_total: u64,
    /// Rounds that made no progress: at least one table tried and *every*
    /// attempted table failed (a fully-failed round — see
    /// [`RoundReport::is_failure`]).
    pub rounds_failed: u64,
    /// Per-table successful syncs summed across rounds (one round can add many).
    pub tables_synced: u64,
    /// Source rows read and ingested across all tables/rounds.
    pub rows_ingested: u64,
    /// Vertices MERGE'd into the graph.
    pub vertices_written: u64,
    /// Edges MERGE'd into the graph.
    pub edges_written: u64,
    /// Nodes removed via `DETACH DELETE` (CDC deletes).
    pub deletes: u64,
    /// AIP-82 asset events emitted (one per ingested table, when an emitter is
    /// wired — counted from the synced tables of each round).
    pub assets_emitted: u64,
}

impl Metrics {
    /// Fold one round's [`RoundReport`] into the running totals. Returns whether
    /// the round counted as a failure (so the caller can drive backoff without
    /// re-deriving it).
    pub fn record_round(&mut self, report: &RoundReport) -> bool {
        self.rounds_total += 1;
        let failed = report.is_failure();
        if failed {
            self.rounds_failed += 1;
        }
        for (_table, rows, sink) in &report.synced {
            self.tables_synced += 1;
            self.rows_ingested += *rows as u64;
            self.vertices_written += sink.vertices_written as u64;
            self.edges_written += sink.edges_written as u64;
            self.deletes += sink.deleted as u64;
        }
        // Each synced table announces exactly one asset event (mirrors
        // `round_to_events`, which emits one event per synced table).
        self.assets_emitted += report.synced.len() as u64;
        failed
    }
}

/// Render [`Metrics`] as Prometheus text-exposition format. A pure function: a
/// `/metrics` HTTP endpoint can serve this later without pulling an HTTP server
/// dependency into the library. Every counter is a `# TYPE … counter` with the
/// `bifrost_` prefix.
pub fn render_prometheus(m: &Metrics) -> String {
    /// One `HELP`/`TYPE`/value triple for a counter.
    fn counter(out: &mut String, name: &str, help: &str, value: u64) {
        use std::fmt::Write;
        let _ = writeln!(out, "# HELP bifrost_{name} {help}");
        let _ = writeln!(out, "# TYPE bifrost_{name} counter");
        let _ = writeln!(out, "bifrost_{name} {value}");
    }

    let mut out = String::new();
    counter(
        &mut out,
        "rounds_total",
        "Sync rounds attempted.",
        m.rounds_total,
    );
    counter(
        &mut out,
        "rounds_failed",
        "Sync rounds that made no progress.",
        m.rounds_failed,
    );
    counter(
        &mut out,
        "tables_synced",
        "Per-table successful syncs.",
        m.tables_synced,
    );
    counter(
        &mut out,
        "rows_ingested",
        "Source rows ingested.",
        m.rows_ingested,
    );
    counter(
        &mut out,
        "vertices_written",
        "Vertices MERGE'd into the graph.",
        m.vertices_written,
    );
    counter(
        &mut out,
        "edges_written",
        "Edges MERGE'd into the graph.",
        m.edges_written,
    );
    counter(
        &mut out,
        "deletes",
        "Nodes DETACH DELETE'd (CDC).",
        m.deletes,
    );
    counter(
        &mut out,
        "assets_emitted",
        "AIP-82 asset events emitted.",
        m.assets_emitted,
    );
    out
}

/// Exponential backoff for the sleep *between* poll rounds.
///
/// The happy path sleeps the base `interval`. A failed (no-progress) round
/// doubles the next sleep — `interval → interval*2 → … → max` — and the first
/// round that makes progress resets the sleep back to `interval`. So a healthy
/// poller keeps its cadence exactly; a poller whose source/sink is down backs
/// off geometrically up to `max`, sparing the downstream.
#[derive(Debug, Clone)]
pub struct Backoff {
    /// Base (happy-path) interval and the floor the schedule resets to.
    interval: Duration,
    /// Ceiling the doubling never exceeds.
    max: Duration,
    /// The sleep the *next* round will use; `interval` while healthy.
    current: Duration,
}

impl Backoff {
    /// A schedule with base `interval` and ceiling `max`. `max` is clamped up to
    /// `interval` (a cap below the base interval would make no sense).
    pub fn new(interval: Duration, max: Duration) -> Self {
        let max = max.max(interval);
        Self {
            interval,
            max,
            current: interval,
        }
    }

    /// The base interval (the healthy cadence / the reset floor).
    pub fn interval(&self) -> Duration {
        self.interval
    }

    /// The cap the doubling never exceeds.
    pub fn max(&self) -> Duration {
        self.max
    }

    /// The sleep the next round will use given how this one went. On `failed`,
    /// returns the *current* sleep and then doubles it (capped at `max`) for the
    /// round after; on success, returns the base `interval` and resets the
    /// schedule. Returning-then-advancing means the first failed round still
    /// sleeps the base `interval`, and only *consecutive* failures grow it.
    pub fn next_sleep(&mut self, failed: bool) -> Duration {
        if !failed {
            self.current = self.interval;
            return self.interval;
        }
        let sleep = self.current;
        // Double for the next round, saturating at the cap (and guarding the
        // pathological `interval == 0` so we don't get stuck doubling zero).
        let doubled = self.current.saturating_mul(2);
        self.current = doubled.min(self.max).max(self.interval);
        sleep
    }
}

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

    fn synced_report() -> RoundReport {
        RoundReport {
            synced: vec![
                (
                    "a".into(),
                    10,
                    SinkReport {
                        vertices_written: 10,
                        edges_written: 4,
                        deleted: 1,
                    },
                ),
                (
                    "b".into(),
                    5,
                    SinkReport {
                        vertices_written: 5,
                        edges_written: 0,
                        deleted: 0,
                    },
                ),
            ],
            failed: vec![],
        }
    }

    fn failed_report() -> RoundReport {
        RoundReport {
            synced: vec![],
            failed: vec![("a".into(), "boom".into())],
        }
    }

    fn idle_report() -> RoundReport {
        RoundReport::default()
    }

    // ── metrics accumulation ────────────────────────────────────────────────

    #[test]
    fn record_round_accumulates_every_counter() {
        let mut m = Metrics::default();
        assert!(
            !m.record_round(&synced_report()),
            "a fully-synced round is not a failure"
        );

        assert_eq!(m.rounds_total, 1);
        assert_eq!(m.rounds_failed, 0);
        assert_eq!(m.tables_synced, 2);
        assert_eq!(m.rows_ingested, 15);
        assert_eq!(m.vertices_written, 15);
        assert_eq!(m.edges_written, 4);
        assert_eq!(m.deletes, 1);
        assert_eq!(m.assets_emitted, 2);

        // A second synced round adds on top (counters are cumulative).
        m.record_round(&synced_report());
        assert_eq!(m.rounds_total, 2);
        assert_eq!(m.tables_synced, 4);
        assert_eq!(m.rows_ingested, 30);
        assert_eq!(m.assets_emitted, 4);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics::record_round_accumulates_every_counter",
            m.rounds_total == 2
                && m.tables_synced == 4
                && m.rows_ingested == 30
                && m.assets_emitted == 4,
            &format!(
                "after 2 rounds: tables={}, rows={}, assets={}",
                m.tables_synced, m.rows_ingested, m.assets_emitted
            ),
        );
    }

    #[test]
    fn failed_round_increments_rounds_failed_only() {
        let mut m = Metrics::default();
        assert!(
            m.record_round(&failed_report()),
            "an all-failed round is a failure"
        );
        assert_eq!(m.rounds_total, 1);
        assert_eq!(m.rounds_failed, 1);
        assert_eq!(m.tables_synced, 0);
        assert_eq!(m.assets_emitted, 0);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics::failed_round_increments_rounds_failed_only",
            m.rounds_failed == 1 && m.tables_synced == 0,
            &format!(
                "rounds_failed={}, tables_synced={}",
                m.rounds_failed, m.tables_synced
            ),
        );
    }

    #[test]
    fn idle_round_counts_but_is_not_a_failure() {
        // An idle round (nothing synced, nothing failed) advances rounds_total
        // but must NOT count as a failure — otherwise idle pollers would back
        // off forever.
        let mut m = Metrics::default();
        let is_failure = m.record_round(&idle_report());
        assert!(!is_failure);
        assert_eq!(m.rounds_total, 1);
        assert_eq!(m.rounds_failed, 0);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics::idle_round_counts_but_is_not_a_failure",
            !is_failure && m.rounds_total == 1 && m.rounds_failed == 0,
            "idle round bumps rounds_total, not rounds_failed",
        );
    }

    #[test]
    fn partial_failure_is_progress_not_a_failure() {
        // One table synced, one failed → the round made progress, so it's not a
        // backoff-triggering failure.
        let mut m = Metrics::default();
        let report = RoundReport {
            synced: vec![(
                "a".into(),
                3,
                SinkReport {
                    vertices_written: 3,
                    ..Default::default()
                },
            )],
            failed: vec![("b".into(), "boom".into())],
        };
        let is_failure = m.record_round(&report);
        assert!(!is_failure);
        assert_eq!(m.rounds_failed, 0);
        assert_eq!(m.tables_synced, 1);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics::partial_failure_is_progress_not_a_failure",
            !is_failure && m.rounds_failed == 0 && m.tables_synced == 1,
            "1 synced + 1 failed = progress, not a backoff failure",
        );
    }

    // ── prometheus rendering ────────────────────────────────────────────────

    #[test]
    fn render_prometheus_emits_typed_counters() {
        let mut m = Metrics::default();
        m.record_round(&synced_report());
        let text = render_prometheus(&m);

        assert!(text.contains("# TYPE bifrost_rows_ingested counter"));
        assert!(text.contains("# HELP bifrost_rounds_total Sync rounds attempted."));
        assert!(text.contains("bifrost_rows_ingested 15"));
        assert!(text.contains("bifrost_vertices_written 15"));
        assert!(text.contains("bifrost_edges_written 4"));
        assert!(text.contains("bifrost_deletes 1"));
        assert!(text.contains("bifrost_assets_emitted 2"));
        assert!(text.contains("bifrost_rounds_total 1"));
        // Every metric line ends with a newline (valid exposition format).
        assert!(text.ends_with('\n'));
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics::render_prometheus_emits_typed_counters",
            text.contains("bifrost_rows_ingested 15")
                && text.contains("# TYPE bifrost_rows_ingested counter")
                && text.ends_with('\n'),
            "valid prometheus exposition with typed counters",
        );
    }

    // ── backoff schedule ────────────────────────────────────────────────────

    #[test]
    fn backoff_doubles_on_consecutive_failures_up_to_cap() {
        // base 1s, cap 10s → 1, 2, 4, 8, 10 (capped), 10, …
        let base = Duration::from_secs(1);
        let mut b = Backoff::new(base, Duration::from_secs(10));
        let seq: Vec<u64> = (0..6).map(|_| b.next_sleep(true).as_secs()).collect();
        assert_eq!(seq, vec![1, 2, 4, 8, 10, 10]);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics::backoff_doubles_on_consecutive_failures_up_to_cap",
            seq == vec![1, 2, 4, 8, 10, 10],
            &format!("backoff secs={seq:?}"),
        );
    }

    #[test]
    fn backoff_resets_to_base_on_success() {
        let base = Duration::from_secs(1);
        let mut b = Backoff::new(base, Duration::from_secs(16));
        // Climb a few failures.
        assert_eq!(b.next_sleep(true).as_secs(), 1);
        assert_eq!(b.next_sleep(true).as_secs(), 2);
        assert_eq!(b.next_sleep(true).as_secs(), 4);
        // A success drops straight back to the base interval.
        assert_eq!(b.next_sleep(false).as_secs(), 1);
        // And the next failure starts climbing from the base again.
        let after_reset_1 = b.next_sleep(true).as_secs();
        let after_reset_2 = b.next_sleep(true).as_secs();
        assert_eq!(after_reset_1, 1);
        assert_eq!(after_reset_2, 2);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics::backoff_resets_to_base_on_success",
            after_reset_1 == 1 && after_reset_2 == 2,
            "success resets backoff to base, then climbs again",
        );
    }

    #[test]
    fn backoff_healthy_path_keeps_cadence() {
        let base = Duration::from_secs(5);
        let mut b = Backoff::new(base, Duration::from_secs(50));
        // A run of successful rounds never deviates from the base interval.
        let mut steady = true;
        for _ in 0..10 {
            steady &= b.next_sleep(false) == base;
        }
        assert!(steady);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics::backoff_healthy_path_keeps_cadence",
            steady,
            "10 healthy rounds all sleep the base interval",
        );
    }

    #[test]
    fn backoff_clamps_cap_below_interval_up_to_interval() {
        // A nonsensical cap < interval is clamped up; the schedule never sleeps
        // less than the base interval.
        let base = Duration::from_secs(30);
        let mut b = Backoff::new(base, Duration::from_secs(5));
        assert_eq!(b.max(), base);
        assert_eq!(b.next_sleep(true), base);
        assert_eq!(b.next_sleep(true), base);
        crate::testmatrix::emit(
            crate::testmatrix::COMPONENT,
            "metrics::backoff_clamps_cap_below_interval_up_to_interval",
            b.max() == base,
            &format!("cap clamped up to base interval {}s", base.as_secs()),
        );
    }
}