src/warehouse/clone_events.rs
//! Iceberg writer/reader for the **clone/populate-outcome stream** (`clone_events`,
//! PLAN #6).
//!
//! When the server's poll loop (or a `Workspaces.Fetch` RPC, or the fat-CLI
//! `nornir workspace fetch`) clones+fetches a monitored workspace's git members,
//! per-member failures used to be *log-only* — an `eprintln!("…fetch error…")`
//! that a thin viz/CLI client could never see. It just saw missing data and no
//! reason. This module makes those outcomes **first-class, persisted, and readable
//! remotely**: one row per member attempt (clone/fetch + republish), `ok` or
//! `error`, carrying the error chain and the elapsed time.
//!
//! Write path: [`record_fetch_report`] turns a [`crate::monitor::FetchReport`] into
//! rows and appends them through the warehouse handle the *server already holds*
//! (redb is single-writer — the monitor itself can't open a second handle, so the
//! call sites that own a handle do the write). Best-effort + non-fatal: a logging
//! row failure must never abort a fetch/republish.
//!
//! Read path: [`query_clone_events`] scans the table, scopes by workspace, and
//! returns rows newest-first (Iceberg gives no scan order, so reads sort by
//! `ts_micros`). The `Viz.CloneEvents` RPC + `nornir workspace events` both read
//! through it.
//!
//! Schema: [`super::iceberg_schema::clone_events`].
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow::array::{Array, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use chrono::{DateTime, TimeZone, Utc};
use iceberg::arrow::schema_to_arrow_schema;
use super::iceberg::{IcebergWarehouse, TABLE_CLONE_EVENTS, append_batch};
// Column indices match `iceberg_schema::clone_events` field order.
const COL_TS_MICROS: usize = 0;
const COL_WORKSPACE: usize = 1;
const COL_MEMBER: usize = 2;
const COL_REMOTE: usize = 3;
const COL_OP: usize = 4;
const COL_STATUS: usize = 5;
const COL_DETAIL: usize = 6;
const COL_ELAPSED_MS: usize = 7;
/// Canonical op tags.
pub mod op {
/// A member clone/fetch attempt (today both go through `clone_or_fetch`).
pub const CLONE_FETCH: &str = "clone-fetch";
/// The post-fetch warehouse republish for the whole workspace.
pub const REPUBLISH: &str = "republish";
/// One member's turn in the Mímir v2 consumer (commits behind its offset
/// consumed into the code room). `ok` → `seq N @ sha, K commit(s) …`;
/// `error` → why the member could not be consumed.
pub const CONSUME: &str = "consume";
}
/// Canonical status tags.
pub mod status {
pub const OK: &str = "ok";
pub const ERROR: &str = "error";
}
/// One row of the `clone_events` table — a single member's clone/fetch (or the
/// workspace republish) outcome.
///
/// `serde` derives let the server ship the exact rows to a thin viz client over
/// the `Viz.CloneEvents` RPC (serialize → JSON → deserialize), so the remote 🧬
/// nornir pane + `nornir --server workspace events` render the identical populate
/// status the embedded path reads from the warehouse.
///
/// This is a CORE (ungated) type — `warehouse` compiles in the default CLI build,
/// which does NOT pull the optional `nornir-wire` leaf — so its wire-crossing twin
/// is a field-identical **mirror** in `nornir_wire::view::rows::CloneEventRow`,
/// with the boundary [`From`] conversion below (gated to the wire feature sets).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct CloneEventRow {
/// Event time, microseconds since the Unix epoch (UTC).
pub ts_micros: i64,
/// The workspace this fetch belonged to.
pub workspace: String,
/// The member that was fetched (or `*` for a workspace-level `republish`).
pub member: String,
/// The git remote that was cloned/fetched (empty for a republish row).
pub remote: String,
/// `clone-fetch` | `republish` (see [`op`]).
pub op: String,
/// `ok` | `error` (see [`status`]).
pub status: String,
/// On error: the error chain (`{e:#}`). On ok: the resolved SHA (clone-fetch)
/// or the snapshot id (republish). Stored as "" never null so SQL readers see
/// no surprise nulls.
pub detail: String,
/// Wall-clock duration of the op, milliseconds.
pub elapsed_ms: i64,
}
/// Boundary conversion into the `nornir-wire` mirror (`CloneEventRow` → the
/// wire-stable [`nornir_wire::view::rows::CloneEventRow`]). Gated to the feature
/// sets that pull the wire leaf.
#[cfg(any(feature = "server", feature = "viz", feature = "mcp"))]
impl From<CloneEventRow> for nornir_wire::view::rows::CloneEventRow {
fn from(r: CloneEventRow) -> Self {
Self {
ts_micros: r.ts_micros,
workspace: r.workspace,
member: r.member,
remote: r.remote,
op: r.op,
status: r.status,
detail: r.detail,
elapsed_ms: r.elapsed_ms,
}
}
}
/// The reverse boundary conversion (wire mirror → core `CloneEventRow`), so a
/// client that deserialized the wire mirror can hand the core type back to code
/// that still wants it (the viz pane + `workspace_populate_verdict`, until the
/// pane moves to the wire type in Step 4).
#[cfg(any(feature = "server", feature = "viz", feature = "mcp"))]
impl From<nornir_wire::view::rows::CloneEventRow> for CloneEventRow {
fn from(r: nornir_wire::view::rows::CloneEventRow) -> Self {
Self {
ts_micros: r.ts_micros,
workspace: r.workspace,
member: r.member,
remote: r.remote,
op: r.op,
status: r.status,
detail: r.detail,
elapsed_ms: r.elapsed_ms,
}
}
}
/// Append a batch of `clone_events` rows (one Iceberg snapshot). Mirrors
/// [`super::release_events::append_release_events`].
pub async fn append_clone_events(wh: &IcebergWarehouse, rows: &[CloneEventRow]) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
// Blank/torn metadata JSON self-heals (drop+recreate the 0-row table) so the
// populate-outcome append can't EOF-fail.
let table = wh
.load_table_for_append(
TABLE_CLONE_EVENTS,
super::iceberg_schema::clone_events()?,
&[],
)
.await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(
TimestampMicrosecondArray::from(rows.iter().map(|r| r.ts_micros).collect::<Vec<_>>())
.with_timezone("+00:00"),
),
Arc::new(StringArray::from(
rows.iter().map(|r| r.workspace.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.member.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.remote.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.op.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.status.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.detail.clone()).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
rows.iter().map(|r| r.elapsed_ms).collect::<Vec<_>>(),
)),
];
let batch = RecordBatch::try_new(arrow_schema, cols)?;
append_batch(wh.catalog(), table, batch).await?;
Ok(())
}
/// Build `clone_events` rows from a [`crate::monitor::FetchReport`] — one per
/// fetched member outcome (ok or error). Pure (no I/O), so the writer + tests
/// share it. Stamped now (UTC micros).
pub fn rows_from_report(report: &crate::monitor::FetchReport) -> Vec<CloneEventRow> {
let ts = Utc::now().timestamp_micros();
report
.outcomes
.iter()
.map(|o| CloneEventRow {
ts_micros: ts,
workspace: report.workspace.clone(),
member: o.member.clone(),
remote: o.remote.clone(),
op: o.op.clone(),
status: o.status.clone(),
detail: o.detail.clone(),
elapsed_ms: o.elapsed_ms,
})
.collect()
}
/// Persist a fetch report's per-member outcomes to `clone_events`. Best-effort +
/// **non-fatal**: a logging-row failure is logged (stderr) and swallowed so a
/// warehouse hiccup never aborts an otherwise-successful fetch/republish. Called
/// from every path that owns a warehouse handle (server sweep, `Workspaces.Fetch`,
/// the fat-CLI fetch).
pub fn record_fetch_report(wh: &IcebergWarehouse, report: &crate::monitor::FetchReport) {
let rows = rows_from_report(report);
if rows.is_empty() {
return;
}
if let Err(e) = wh.block_on(append_clone_events(wh, &rows)) {
eprintln!(
" ⚠ clone_events: dropped {} populate-outcome row(s) for `{}` (non-fatal): {e:#}",
rows.len(),
report.workspace
);
}
}
/// Keep only the rows worth persisting from a **periodic poll sweep**: errors
/// (so "why didn't this member populate" survives) and members that actually
/// changed this tick. The steady-state "ok, nothing changed" rows are dropped —
/// otherwise the 60s poll loop appends one row per member *every tick forever*,
/// growing the table unboundedly (the cause of the multi-MB `Viz.CloneEvents`
/// payload). User-initiated fetches still log every outcome via
/// [`record_fetch_report`]; only the unattended sweep is thinned.
pub fn rows_from_sweep(report: &crate::monitor::FetchReport) -> Vec<CloneEventRow> {
rows_from_report(report)
.into_iter()
.filter(|r| r.status == status::ERROR || report.changed.iter().any(|c| c == &r.member))
.collect()
}
/// Like [`record_fetch_report`] but only persists the sweep-worthy rows
/// ([`rows_from_sweep`]). Used by the server poll loop so an idle workspace
/// stops minting a no-op row per member every tick. Non-fatal.
pub fn record_sweep_report(wh: &IcebergWarehouse, report: &crate::monitor::FetchReport) {
let rows = rows_from_sweep(report);
if rows.is_empty() {
return;
}
if let Err(e) = wh.block_on(append_clone_events(wh, &rows)) {
eprintln!(
" ⚠ clone_events: dropped {} sweep-outcome row(s) for `{}` (non-fatal): {e:#}",
rows.len(),
report.workspace
);
}
}
/// Record a single workspace-level `republish` outcome (ok → snapshot id, or the
/// error chain). Non-fatal. `member` is `*` (a republish covers the whole
/// workspace, not one member).
pub fn record_republish(
wh: &IcebergWarehouse,
workspace: &str,
status_: &str,
detail: &str,
elapsed_ms: i64,
) {
let row = CloneEventRow {
ts_micros: Utc::now().timestamp_micros(),
workspace: workspace.to_string(),
member: "*".into(),
remote: String::new(),
op: op::REPUBLISH.into(),
status: status_.to_string(),
detail: detail.to_string(),
elapsed_ms,
};
if let Err(e) = wh.block_on(append_clone_events(wh, std::slice::from_ref(&row))) {
eprintln!(" ⚠ clone_events: dropped republish row for `{workspace}` (non-fatal): {e:#}");
}
}
/// Record the per-member outcomes of one consumer run (op [`op::CONSUME`]):
/// `(member, ok, detail, elapsed_ms)`. They are MEMBER rows, so the populate
/// roll-up ([`workspace_populate_verdict`]) goes red on a member the consumer
/// could not consume — a workspace whose clones are fine but whose code room is
/// not is not green. Non-fatal.
pub fn record_consume(wh: &IcebergWarehouse, workspace: &str, outcomes: &[(String, bool, String, i64)]) {
let now = Utc::now().timestamp_micros();
let rows: Vec<CloneEventRow> = outcomes
.iter()
.map(|(member, ok, detail, elapsed_ms)| CloneEventRow {
ts_micros: now,
workspace: workspace.to_string(),
member: member.clone(),
remote: String::new(),
op: op::CONSUME.into(),
status: if *ok { status::OK } else { status::ERROR }.into(),
detail: detail.clone(),
elapsed_ms: *elapsed_ms,
})
.collect();
if rows.is_empty() {
return;
}
if let Err(e) = wh.block_on(append_clone_events(wh, &rows)) {
eprintln!(" ⚠ clone_events: dropped {} consume row(s) for `{workspace}` (non-fatal): {e:#}", rows.len());
}
}
/// Which clone events to read.
#[derive(Debug, Clone)]
pub enum CloneSelector {
/// Every event whose `workspace` matches.
Workspace(String),
/// Everything in the table.
All,
}
/// Read clone events, scoped by `sel`, returned **newest-first** (`ts_micros`
/// descending) so the recent populate status is at the top — Iceberg gives no scan
/// order, so the reader sorts.
pub async fn query_clone_events(
wh: &IcebergWarehouse,
sel: &CloneSelector,
) -> Result<Vec<CloneEventRow>> {
// Blank/torn Iceberg metadata JSON → 0 rows, never a JSON-EOF crash (load+read).
let batches: Vec<RecordBatch> =
super::iceberg::load_and_read_all(wh, TABLE_CLONE_EVENTS).await?;
let mut out: Vec<CloneEventRow> = Vec::new();
for b in &batches {
let ts = col_ts(b, COL_TS_MICROS)?;
let workspace = col_str(b, COL_WORKSPACE)?;
let member = col_str(b, COL_MEMBER)?;
let remote = col_str(b, COL_REMOTE)?;
let op_ = col_str(b, COL_OP)?;
let st = col_str(b, COL_STATUS)?;
let detail = col_str(b, COL_DETAIL)?;
let elapsed = col_i64(b, COL_ELAPSED_MS)?;
for i in 0..b.num_rows() {
let row = CloneEventRow {
ts_micros: ts.value(i),
workspace: workspace.value(i).to_string(),
member: member.value(i).to_string(),
remote: remote.value(i).to_string(),
op: op_.value(i).to_string(),
status: st.value(i).to_string(),
detail: detail.value(i).to_string(),
elapsed_ms: elapsed.value(i),
};
let keep = match sel {
CloneSelector::Workspace(w) => &row.workspace == w,
CloneSelector::All => true,
};
if keep {
out.push(row);
}
}
}
out.sort_by(|a, b| b.ts_micros.cmp(&a.ts_micros));
Ok(out)
}
/// The rolled-up **populate verdict** for ONE workspace (PO1): the latest
/// clone/fetch outcome per member collapsed into a single workspace-level state.
/// `green` ⟺ every member's most-recent op is `ok`; `red` ⟺ at least one member's
/// most-recent op is `error`; `stale` ⟺ no clone_events recorded at all (never
/// populated). The first failing member + its error detail are surfaced so the
/// roster shows *which* member broke and *why* — the same "see what the user sees"
/// data the per-member pane renders, collapsed for the multi-workspace roster.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct PopulateVerdict {
/// `"green"` (all latest ok) · `"red"` (≥1 latest error) · `"stale"` (no events).
pub state: String,
/// How many distinct members have a latest outcome.
pub members: usize,
/// How many members' latest outcome is `error`.
pub failed: usize,
/// The first failing member (alphabetical), or empty when none failed.
pub failing_member: String,
/// That member's latest error detail (the git error chain), or empty.
pub last_error: String,
/// RFC3339 of the newest event across the workspace, or empty when stale.
pub last_synced: String,
}
/// Verdict states.
pub mod verdict {
pub const GREEN: &str = "green";
pub const RED: &str = "red";
pub const STALE: &str = "stale";
}
/// Roll a workspace's `clone_events` rows (any order) up into a single
/// [`PopulateVerdict`] (PO1). Pure (no I/O) so the CLI roster, the viz roster, and
/// the server roll-up RPC share ONE verdict computation. For each member we keep
/// only its NEWEST event (`republish` rows — member `*` — are ignored: they're a
/// workspace-level op, not a member populate). An empty input is `stale`
/// (RAGNARÖK: never-populated ⇒ stale, not silently green).
pub fn workspace_populate_verdict(rows: &[CloneEventRow]) -> PopulateVerdict {
use std::collections::BTreeMap;
// member → its newest row (by ts_micros).
let mut latest: BTreeMap<&str, &CloneEventRow> = BTreeMap::new();
let mut newest_ts: i64 = i64::MIN;
for r in rows {
if r.member == "*" {
// workspace-level republish — still track the freshness timestamp.
newest_ts = newest_ts.max(r.ts_micros);
continue;
}
newest_ts = newest_ts.max(r.ts_micros);
latest
.entry(r.member.as_str())
.and_modify(|cur| {
if r.ts_micros > cur.ts_micros {
*cur = r;
}
})
.or_insert(r);
}
// RAGNARÖK (LAW#2): a workspace with ZERO populated members is STALE, never
// silently GREEN — even when a workspace-level `*` republish row advanced
// `newest_ts`. The old guard required BOTH `latest.is_empty() && newest_ts ==
// i64::MIN`, so a republish-only workspace (member `*` rows are skipped for
// `latest` above but they DO bump `newest_ts`) slipped past the stale branch:
// `latest.is_empty()` was true, `newest_ts != MIN`, so with no failing member
// the tail reported `green / members:0`. Gate on the member roster alone — no
// member with a latest outcome ⇒ nothing was populated ⇒ stale.
if latest.is_empty() {
return PopulateVerdict {
state: verdict::STALE.into(),
members: 0,
failed: 0,
failing_member: String::new(),
last_error: String::new(),
last_synced: String::new(),
};
}
let failed: Vec<&&CloneEventRow> = latest
.values()
.filter(|r| r.status == status::ERROR)
.collect();
// First failing member (BTreeMap iterates members alphabetically).
let (failing_member, last_error) = latest
.values()
.find(|r| r.status == status::ERROR)
.map(|r| (r.member.clone(), r.detail.clone()))
.unwrap_or_default();
PopulateVerdict {
state: if failed.is_empty() {
verdict::GREEN
} else {
verdict::RED
}
.into(),
members: latest.len(),
failed: failed.len(),
failing_member,
last_error,
last_synced: if newest_ts == i64::MIN {
String::new()
} else {
ts_to_rfc3339(newest_ts)
},
}
}
/// Render a human-readable populate-status view: newest events first, each line
/// `mark op member — detail (elapsed)`. Backs `nornir workspace events`.
pub fn render_events(rows: &[CloneEventRow]) -> String {
if rows.is_empty() {
return "(no clone/populate events recorded)\n".to_string();
}
let mut out = String::new();
for r in rows {
let mark = match r.status.as_str() {
status::OK => "✓",
status::ERROR => "✗",
_ => "·",
};
let when = ts_to_rfc3339(r.ts_micros);
let detail = if r.detail.is_empty() {
String::new()
} else {
format!(" — {}", r.detail)
};
out.push_str(&format!(
"{mark} {when} {op:<12} {member}{detail} ({elapsed}ms)\n",
op = r.op,
member = r.member,
elapsed = r.elapsed_ms,
));
}
out
}
// ─── column helpers ─────────────────────────────────────────────────────
fn col_str<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a StringArray> {
b.column(idx)
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("clone_events col {idx} is not StringArray"))
}
fn col_i64<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a Int64Array> {
b.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| anyhow!("clone_events col {idx} is not Int64Array"))
}
fn col_ts<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a TimestampMicrosecondArray> {
b.column(idx)
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
.ok_or_else(|| anyhow!("clone_events col {idx} is not TimestampMicrosecondArray"))
}
/// Format `ts_micros` as RFC3339 (UTC). Convenience for renderers/tests.
pub fn ts_to_rfc3339(ts_micros: i64) -> String {
let dt: DateTime<Utc> = Utc
.timestamp_micros(ts_micros)
.single()
.unwrap_or_else(Utc::now);
dt.to_rfc3339()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::monitor::{FetchReport, MemberOutcome};
fn outcome(member: &str, status_: &str, detail: &str) -> MemberOutcome {
MemberOutcome {
member: member.into(),
remote: format!("git@example.com:{member}.git"),
op: op::CLONE_FETCH.into(),
status: status_.into(),
detail: detail.into(),
elapsed_ms: 7,
}
}
/// LAW (inject-assert): a periodic sweep over an idle workspace must NOT mint a
/// row for steady-state "ok, unchanged" members — only errors and members that
/// actually changed. This is what stops the unbounded `clone_events` growth that
/// overflowed the `Viz.CloneEvents` gRPC message-size limit.
#[test]
fn sweep_thinning_keeps_only_errors_and_changed() {
let report = FetchReport {
workspace: "nordisk".into(),
fetched: 3,
changed: vec!["facett".into()],
errors: vec![("korp".into(), "fatal: could not read".into())],
outcomes: vec![
outcome("nornir", status::OK, "abc123"), // ok + unchanged → DROP
outcome("facett", status::OK, "def456"), // ok + changed → KEEP
outcome("korp", status::ERROR, "fatal: could not read"), // error → KEEP
],
};
let kept = rows_from_sweep(&report);
let members: Vec<&str> = kept.iter().map(|r| r.member.as_str()).collect();
assert_eq!(
members,
vec!["facett", "korp"],
"only changed + error rows survive a sweep"
);
// The no-op heartbeat row for the unchanged-ok member is gone.
assert!(
!members.contains(&"nornir"),
"steady-state ok/unchanged row must be dropped"
);
// A user-initiated fetch (record_fetch_report path) still logs everything.
assert_eq!(
rows_from_report(&report).len(),
3,
"full report keeps all outcomes"
);
}
/// LAW 1 (inject-assert): feed a FetchReport with one OK member and one FAILED
/// member → assert exactly two rows persist with the right
/// (member, op, status, detail) and read back newest-first, scoped by
/// workspace.
#[test]
fn fetch_report_failure_persists_and_reads_back() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let report = FetchReport {
workspace: "nordisk".into(),
fetched: 2,
changed: vec!["facett".into()],
errors: vec![(
"korp".into(),
"clone-fetch https://github.com/nordisk/korp: Couldn't obtain Username".into(),
)],
outcomes: vec![
MemberOutcome {
member: "facett".into(),
remote: "git@github.com:nordisk/facett.git".into(),
op: op::CLONE_FETCH.into(),
status: status::OK.into(),
detail: "abc123".into(),
elapsed_ms: 42,
},
MemberOutcome {
member: "korp".into(),
remote: "https://github.com/nordisk/korp".into(),
op: op::CLONE_FETCH.into(),
status: status::ERROR.into(),
detail: "clone-fetch https://github.com/nordisk/korp: Couldn't obtain Username"
.into(),
elapsed_ms: 7,
},
],
};
// Write through the (non-fatal) recorder — the same call the server uses.
record_fetch_report(&wh, &report);
// Read back, scoped to the workspace, newest-first.
let rows = wh
.block_on(query_clone_events(
&wh,
&CloneSelector::Workspace("nordisk".into()),
))
.unwrap();
assert_eq!(rows.len(), 2, "one row per fetched member");
// The FAILED member is first-class: find it and assert the error detail
// round-tripped exactly.
let failed: Vec<_> = rows.iter().filter(|r| r.status == status::ERROR).collect();
assert_eq!(failed.len(), 1);
assert_eq!(failed[0].member, "korp");
assert_eq!(failed[0].op, "clone-fetch");
assert_eq!(failed[0].remote, "https://github.com/nordisk/korp");
assert!(
failed[0].detail.contains("Couldn't obtain Username"),
"the error chain is readable, got: {}",
failed[0].detail
);
assert_eq!(failed[0].elapsed_ms, 7);
// The OK member round-trips with its SHA in `detail`.
let ok: Vec<_> = rows.iter().filter(|r| r.status == status::OK).collect();
assert_eq!(ok.len(), 1);
assert_eq!(ok[0].member, "facett");
assert_eq!(ok[0].detail, "abc123");
// A different workspace's scope sees nothing.
let other = wh
.block_on(query_clone_events(
&wh,
&CloneSelector::Workspace("holger".into()),
))
.unwrap();
assert!(other.is_empty(), "workspace scope isolates events");
// The human render names the failed member + its error.
let txt = render_events(&rows);
assert!(txt.contains("korp"));
assert!(txt.contains("Couldn't obtain Username"));
assert!(txt.contains('✗'), "failure marker present");
// JSON round-trips for the RPC contract.
let json = serde_json::to_string(&rows).unwrap();
let back: Vec<CloneEventRow> = serde_json::from_str(&json).unwrap();
assert_eq!(back, rows);
}
/// PF-cloneevents regression: an UNBOUNDED clone-event ledger (a long-lived
/// nordisk that never sweep-thinned) overflows the OLD 4 MiB tonic encode
/// cap — which is why the server now needs BOTH the 2 000-row truncation
/// (`VIZ_CLONE_EVENTS_CAP`) AND the 64 MiB Viz-service cap (the belt for the
/// other large, *un-truncated* viz payloads — Architecture, timeline). The
/// 2 000-row bounded payload must fit 4 MiB (truncation works); the unbounded
/// ledger must exceed 4 MiB (motivating the cap) yet still fit 64 MiB.
/// Multi-value across the boundaries: empty / one / at-truncation / unbounded.
#[test]
fn clone_events_payload_sizes_motivate_both_truncation_and_64mib_cap() {
const OLD_CAP: usize = 4 << 20; // tonic default — what the bug overflowed
const NEW_CAP: usize = 64 << 20; // the Viz-service cap we set on the server
const ROW_CAP: usize = 2_000; // VIZ_CLONE_EVENTS_CAP in nornir-server
let mk = |n: usize| -> Vec<CloneEventRow> {
(0..n)
.map(|i| CloneEventRow {
ts_micros: 1_700_000_000_000_000 + i as i64,
workspace: "nordisk".into(),
member: format!("member-{i:05}"),
remote: format!("ssh://git@git.gunnar.rs:2222/nordisk/member-{i:05}.git"),
op: "clone-fetch".into(),
status: if i % 7 == 0 { "error" } else { "ok" }.into(),
// Realistic worst-case detail: a full git error chain / resolved SHA.
detail: format!(
"fatal: could not read from remote repository for member-{i:05}; \
resolved sha 0123456789abcdef0123456789abcdef01234567 after retry {}",
i % 5
),
elapsed_ms: (i % 9000) as i64,
})
.collect()
};
// Boundary: empty + one row are trivially small.
assert!(
serde_json::to_string(&mk(0)).unwrap().len() < 16,
"empty payload tiny"
);
assert!(
serde_json::to_string(&mk(1)).unwrap().len() < OLD_CAP,
"one row never overflows"
);
// At-truncation: the bounded 2 000-row payload fits the OLD 4 MiB cap —
// proving `VIZ_CLONE_EVENTS_CAP` is a working bound for CloneEvents.
let bounded = serde_json::to_string(&mk(ROW_CAP)).unwrap();
assert!(
bounded.len() < OLD_CAP,
"the 2 000-row truncation keeps CloneEvents under 4 MiB ({} bytes)",
bounded.len(),
);
// Unbounded: a never-thinned ledger (what nordisk accumulated) blows past
// 4 MiB — the motivation for the 64 MiB cap, since other viz payloads
// (Architecture graph, timeline) are NOT row-truncated.
let unbounded = serde_json::to_string(&mk(20_000)).unwrap();
assert!(
unbounded.len() > OLD_CAP,
"an unbounded ledger ({} bytes) overflows the old 4 MiB cap — the bug",
unbounded.len(),
);
assert!(
unbounded.len() < NEW_CAP,
"and it fits the 64 MiB Viz-service cap ({} bytes < {NEW_CAP})",
unbounded.len(),
);
}
/// PO1 (inject-assert): roll a workspace's clone_events up into ONE verdict.
/// A workspace with one failing member rolls up RED + names the failing member
/// and surfaces its error; the LATEST op per member wins (a later ok heals an
/// earlier error); an all-ok workspace is GREEN; an empty input is STALE.
#[test]
fn populate_verdict_rolls_up_per_member_latest() {
let row = |ts: i64, member: &str, st: &str, detail: &str| CloneEventRow {
ts_micros: ts,
workspace: "nordisk".into(),
member: member.into(),
remote: format!("git@x:{member}.git"),
op: op::CLONE_FETCH.into(),
status: st.into(),
detail: detail.into(),
elapsed_ms: 1,
};
// facett ok (latest), korp error (latest) → RED, korp named + its error.
let rows = vec![
row(1, "facett", status::ERROR, "transient blip"), // older facett error
row(5, "facett", status::OK, "abc123"), // newer facett ok → heals
row(3, "korp", status::ERROR, "Couldn't obtain Username"),
];
let v = workspace_populate_verdict(&rows);
assert_eq!(
v.state,
verdict::RED,
"one member's latest is an error → red"
);
assert_eq!(v.members, 2);
assert_eq!(v.failed, 1, "only korp's latest is red (facett healed)");
assert_eq!(v.failing_member, "korp");
assert!(
v.last_error.contains("Couldn't obtain Username"),
"error surfaced: {}",
v.last_error
);
assert!(v.last_synced.contains('T'), "newest ts rendered RFC3339");
// All-ok workspace → green, no failing member.
let green = workspace_populate_verdict(&[
row(1, "a", status::OK, "sha-a"),
row(2, "b", status::OK, "sha-b"),
]);
assert_eq!(green.state, verdict::GREEN);
assert_eq!(green.failed, 0);
assert!(green.failing_member.is_empty());
// Never-populated workspace (no events) → stale, not silently green.
let stale = workspace_populate_verdict(&[]);
assert_eq!(stale.state, verdict::STALE);
assert_eq!(stale.members, 0);
assert!(stale.last_synced.is_empty());
}
/// ★4 REGRESSION (RAGNARÖK / LAW#2): a workspace carrying ONLY workspace-level
/// `*` republish rows — zero populated members — must read STALE, never a
/// silent GREEN. The republish rows bump `newest_ts` (freshness) without
/// contributing a member to `latest`; the old `&& newest_ts == i64::MIN`
/// conjunct let that green over 0 members. RED-on-empty first, then the
/// inverse GREEN once one real member event lands.
#[test]
fn republish_only_workspace_is_stale_never_green() {
let republish = |ts: i64| CloneEventRow {
ts_micros: ts,
workspace: "nordisk".into(),
member: "*".into(),
remote: String::new(),
op: op::REPUBLISH.into(),
status: status::OK.into(),
detail: "snapshot-42".into(),
elapsed_ms: 7,
};
let member = |ts: i64, m: &str, st: &str| CloneEventRow {
ts_micros: ts,
workspace: "nordisk".into(),
member: m.into(),
remote: format!("git@x:{m}.git"),
op: op::CLONE_FETCH.into(),
status: st.into(),
detail: "sha".into(),
elapsed_ms: 1,
};
// RED: republish rows only (advance newest_ts) but 0 members → NOT green.
let only_republish = workspace_populate_verdict(&[republish(10), republish(20)]);
assert_ne!(
only_republish.state,
verdict::GREEN,
"0 populated members must never read green (was the false-green bug)"
);
assert_eq!(only_republish.state, verdict::STALE);
assert_eq!(only_republish.members, 0);
assert_eq!(only_republish.failed, 0);
// GREEN (inverse): the SAME republish rows plus one real member ok event
// now has a populated member → green.
let with_member =
workspace_populate_verdict(&[republish(10), member(15, "korp", status::OK)]);
assert_eq!(with_member.state, verdict::GREEN);
assert_eq!(with_member.members, 1);
}
/// A republish outcome is recordable as its own workspace-level row.
#[test]
fn republish_outcome_records() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
record_republish(&wh, "nordisk", status::OK, "snapshot-99", 1234);
let rows = wh
.block_on(query_clone_events(&wh, &CloneSelector::All))
.unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].op, "republish");
assert_eq!(rows[0].member, "*");
assert_eq!(rows[0].detail, "snapshot-99");
assert_eq!(rows[0].elapsed_ms, 1234);
}
}