crates/knut-bifrost/src/metrics.rs
//! 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()),
);
}
}