tests/data_submit_rpcs.rs
//! EPIC N-data — the **submit-RPC round-trip** matrix. The owner's law: a test
//! must inject REAL rows and assert REAL output, not merely "didn't panic".
//!
//! The live server's 🏆 Leaderboard / 🧪 Test / 🚀 Release panes were empty
//! because the only write path to the locked server warehouse was `Bench.Submit`.
//! This file proves the three data-fill RPCs that close that gap:
//!
//! * `Telemetry.SubmitBakeoff` → `agent_model_runs` (🏆 Leaderboard)
//! * `Telemetry.SubmitTestResults` → `test_results` (🧪 Test)
//! * `Telemetry.SubmitReleaseEvents` → `release_events` (🚀 Release) ← NEW
//!
//! For each: spin ONE ephemeral `nornir-server` over an EMPTY workspace
//! warehouse, SUBMIT real rows over the submit RPC, then READ them back over the
//! THIN viz RPCs the live panes use (`Viz.BakeoffResults`, `Viz.ReleaseEvents`,
//! `Warehouse.Scan`) and assert the EXACT values landed — including the
//! `release_events` `Option` fields (`depends_on` None vs empty, `elapsed_ms`)
//! round-tripping through the proto presence flags.
//!
//! cargo test --features viz,server --test data_submit_rpcs -- --nocapture
#![cfg(all(feature = "viz", feature = "server"))]
use std::io::Write as _;
use std::net::TcpListener;
use std::path::PathBuf;
use std::process::{Child, Command};
use std::time::{Duration, Instant};
use nornir::viz::remote;
use nornir::warehouse::agent_model_runs::AgentModelRunRow;
use nornir::warehouse::release_events::{ReleaseEventRow, phase, status};
use nornir::warehouse::test_results::TestResultRow;
const WS: &str = "ndata";
// ── ephemeral server (single empty-seeded workspace) ────────────────────────
fn free_port() -> u16 {
TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port()
}
struct Server {
child: Child,
root: PathBuf,
endpoint: String,
token: String,
}
impl Drop for Server {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
let _ = std::fs::remove_dir_all(&self.root);
}
}
/// Spawn a server monitoring ONE workspace whose warehouse starts EMPTY — the
/// submit RPCs are the only thing that ever writes to it, so a non-zero read-back
/// proves the RPC actually persisted.
fn spawn_server() -> Server {
let token = "ndata-submit-token-0123456789";
let port = free_port();
let endpoint = format!("http://127.0.0.1:{port}");
// Unique per spawn: the three tests run in parallel, so a shared root/HOME
// would collide (same registry + warehouse path).
static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
let uniq = N.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let root = std::env::temp_dir().join(format!("nornir-ndata-{}-{uniq}", std::process::id()));
std::fs::create_dir_all(&root).unwrap();
let server_home = root.join("home");
let registry_root = server_home.join(".nornir").join("workspaces");
std::fs::create_dir_all(®istry_root).unwrap();
// A path-member descriptor: the fetch loop skips path members (no clone), so
// it registers instantly. The warehouse is created lazily by the server.
let desc = root.join("ws.toml");
let mut f = std::fs::File::create(&desc).unwrap();
writeln!(f, "[workspace]\nname = \"{WS}\"\n").unwrap();
writeln!(
f,
"[repos.\"ndata-core\"]\npath = \"{}\"\n",
env!("CARGO_MANIFEST_DIR")
)
.unwrap();
f.flush().unwrap();
drop(f);
// Registry-only design: the server dropped the `NORNIR_MONITOR`/`--monitor` boot
// flag (commit 0c240f9) — a workspace is now served iff it's present in the redb
// registry at `<home>/.nornir/workspaces` at boot. Seed the monitored `ndata` row
// here, then DROP the handle so the server can take the registry's single-writer
// lock. The path-member descriptor makes the fetch loop skip cloning, so it
// serves instantly (the warehouse is still created lazily by the submit RPCs).
{
use nornir::registry::{Mode, Registry, Workspace};
let reg = Registry::open(®istry_root).expect("open registry");
reg.upsert(&Workspace::new(
WS.to_string(),
desc.to_string_lossy().into_owned(),
Mode::Monitored,
"3600s".to_string(),
None,
))
.expect("register ndata workspace");
} // reg dropped → releases registry.redb before the server boots
let child = Command::new(env!("CARGO_BIN_EXE_nornir-server"))
.env("NORNIR_SERVER_ADDR", format!("127.0.0.1:{port}"))
.env("NORNIR_SERVER_TOKEN", token)
.env("HOME", &server_home)
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.expect("spawn nornir-server");
let srv = Server {
child,
root,
endpoint: endpoint.clone(),
token: token.to_string(),
};
let deadline = Instant::now() + Duration::from_secs(45);
let mut last = String::from("no answer yet");
loop {
match remote::list_workspaces(&endpoint, token) {
Ok(ws) if ws.iter().any(|s| s == WS) => break,
Ok(ws) => last = format!("it lists {ws:?}"),
Err(e) => last = format!("Workspaces.List failed: {e:#}"),
}
assert!(
Instant::now() < deadline,
"server never served workspace `{WS}` — {last}"
);
std::thread::sleep(Duration::from_millis(300));
}
srv
}
// ── real input rows ─────────────────────────────────────────────────────────
fn bakeoff_rows() -> Vec<AgentModelRunRow> {
vec![
AgentModelRunRow {
run_id: "bo-run-1".into(),
ts_micros: 1_700_000_000_000,
agent: "claude".into(),
model: "opus".into(),
prompt_id: "p1".into(),
prompt: "sum two numbers".into(),
output: "42".into(),
latency_ms: 120.0,
tokens_in: 30,
tokens_out: 12,
tokens_per_s: 100.0,
score: 0.95,
ok: true,
error: None,
cost_usd: 0.003,
mcp_tool_calls: 2,
},
AgentModelRunRow {
run_id: "bo-run-1".into(),
ts_micros: 1_700_000_001_000,
agent: "kilo".into(),
model: "llama3".into(),
prompt_id: "p1".into(),
prompt: "sum two numbers".into(),
output: "".into(),
latency_ms: 800.0,
tokens_in: 30,
tokens_out: 0,
tokens_per_s: 0.0,
score: 0.0,
ok: false,
error: Some("model timed out".into()),
cost_usd: 0.0,
mcp_tool_calls: 0,
},
]
}
fn test_rows() -> Vec<TestResultRow> {
vec![
TestResultRow {
run_id: "tr-run-1".into(),
repo: "ndata-core".into(),
suite: "unit".into(),
test_name: "adds_ok".into(),
status: nornir::warehouse::test_results::status::PASS.into(),
duration_ms: 4.2,
ts_micros: 1_700_000_002_000,
message: "".into(),
aspect: "unit".into(),
metric: 0.0,
},
TestResultRow {
run_id: "tr-run-1".into(),
repo: "ndata-core".into(),
suite: "clippy".into(),
test_name: "no_warnings".into(),
status: "fail".into(),
duration_ms: 12.0,
ts_micros: 1_700_000_003_000,
message: "unused import".into(),
aspect: "clippy".into(),
metric: 1.0,
},
]
}
/// release_events with the two `Option` shapes exercised: a root op
/// (`depends_on = None`, `elapsed_ms = None`) and an end op that carries a real
/// (non-empty) deps list + an elapsed time. A THIRD row carries an EMPTY-but-
/// present deps list, so the None-vs-empty distinction is asserted explicitly.
fn release_rows() -> Vec<ReleaseEventRow> {
vec![
ReleaseEventRow {
run_id: "re-run-1".into(),
seq: 0,
ts_micros: 1_700_000_004_000,
component: "".into(),
repo: "".into(),
op: "run".into(),
phase: phase::START.into(),
status: status::RUNNING.into(),
detail: "order: a → b".into(),
depends_on: None, // root op, no upstream
elapsed_ms: None,
},
ReleaseEventRow {
run_id: "re-run-1".into(),
seq: 1,
ts_micros: 1_700_000_005_000,
component: "b".into(),
repo: "b".into(),
op: "test".into(),
phase: phase::END.into(),
status: status::OK.into(),
detail: "12 passed".into(),
depends_on: Some(vec!["a".into()]), // real upstream
elapsed_ms: Some(4321),
},
ReleaseEventRow {
run_id: "re-run-1".into(),
seq: 2,
ts_micros: 1_700_000_006_000,
component: "a".into(),
repo: "a".into(),
op: "gate".into(),
phase: phase::END.into(),
status: status::OK.into(),
detail: "all gates green".into(),
depends_on: Some(vec![]), // PRESENT but empty — distinct from None
elapsed_ms: Some(7),
},
]
}
// ════════════════════════════════════════════════════════════════════════════
// 1. Bakeoff.Submit → agent_model_runs → Viz.BakeoffResults
// ════════════════════════════════════════════════════════════════════════════
#[test]
fn submit_bakeoff_lands_and_reads_back() {
let srv = spawn_server();
let want = bakeoff_rows();
// empty before
let before = remote::fetch_bakeoff_results(&srv.endpoint, &srv.token, WS).unwrap();
assert!(
before.is_empty(),
"warehouse should start empty, had {}",
before.len()
);
// inject over the submit RPC
let accepted = remote::submit_bakeoff(&srv.endpoint, &srv.token, WS, &want).unwrap();
assert_eq!(accepted, want.len() as u32, "server accepted count");
// read back over the SAME thin RPC the live 🏆 Leaderboard uses
let got = remote::fetch_bakeoff_results(&srv.endpoint, &srv.token, WS).unwrap();
assert_eq!(got.len(), want.len(), "row count after submit");
// assert the REAL values landed (not just the count)
let claude = got
.iter()
.find(|r| r.agent == "claude")
.expect("claude row");
assert_eq!(claude.model, "opus");
assert!(
(claude.score - 0.95).abs() < 1e-9,
"score = {}",
claude.score
);
assert_eq!(claude.tokens_out, 12);
assert_eq!(claude.mcp_tool_calls, 2);
assert!(claude.ok);
assert!(claude.error.is_none(), "ok row carries no error");
let kilo = got.iter().find(|r| r.agent == "kilo").expect("kilo row");
assert!(!kilo.ok);
assert_eq!(
kilo.error.as_deref(),
Some("model timed out"),
"error string round-trips"
);
}
// ════════════════════════════════════════════════════════════════════════════
// 2. Test.Submit → test_results → Warehouse.Scan
// ════════════════════════════════════════════════════════════════════════════
#[test]
fn submit_test_results_lands_and_reads_back() {
let srv = spawn_server();
let want = test_rows();
let before = remote::scan_table(&srv.endpoint, &srv.token, "test_results", 100, WS)
.map(|p| p.rows.len())
.unwrap_or(0);
assert_eq!(before, 0, "test_results should start empty");
let accepted = remote::submit_test_results(&srv.endpoint, &srv.token, WS, &want).unwrap();
assert_eq!(accepted, want.len() as u32, "server accepted count");
// read back over Warehouse.Scan (the thin browser path the 🧪 pane shares)
let scan = remote::scan_table(&srv.endpoint, &srv.token, "test_results", 100, WS).unwrap();
assert_eq!(
scan.rows.len(),
want.len(),
"test_results rows after submit"
);
// locate the columns we assert on and check the real cell values
let col = |name: &str| {
scan.columns
.iter()
.position(|c| c == name)
.unwrap_or_else(|| panic!("column `{name}` in {:?}", scan.columns))
};
let c_test = col("test_name");
let c_status = col("status");
let c_msg = col("message");
let clippy = scan
.rows
.iter()
.find(|r| r[c_test] == "no_warnings")
.expect("clippy row present");
assert_eq!(clippy[c_status], "fail", "failing aspect status persisted");
assert_eq!(clippy[c_msg], "unused import", "failure message persisted");
let unit = scan
.rows
.iter()
.find(|r| r[c_test] == "adds_ok")
.expect("unit row present");
assert_eq!(
unit[c_status],
nornir::warehouse::test_results::status::PASS,
"passing aspect status persisted"
);
}
// ════════════════════════════════════════════════════════════════════════════
// 3. Release.Submit → release_events → Viz.ReleaseEvents (Option round-trip)
// ════════════════════════════════════════════════════════════════════════════
#[test]
fn submit_release_events_lands_and_reads_back() {
let srv = spawn_server();
let want = release_rows();
let before = remote::fetch_release_events(&srv.endpoint, &srv.token, WS).unwrap();
assert!(
before.is_empty(),
"release_events should start empty, had {}",
before.len()
);
let accepted = remote::submit_release_events(&srv.endpoint, &srv.token, WS, &want).unwrap();
assert_eq!(accepted, want.len() as u32, "server accepted count");
// read back over the thin RPC the live 🚀 Release tab uses; rows come sorted
// by (run_id, seq) — the durable ordering key.
let mut got = remote::fetch_release_events(&srv.endpoint, &srv.token, WS).unwrap();
got.sort_by_key(|r| r.seq);
assert_eq!(got.len(), want.len(), "release_events rows after submit");
// seq 0 — root op: depends_on None, elapsed_ms None survive the round-trip.
let root = &got[0];
assert_eq!(root.op, "run");
assert_eq!(root.phase, phase::START);
assert_eq!(root.detail, "order: a → b");
assert!(root.depends_on.is_none(), "root depends_on must stay None");
assert!(root.elapsed_ms.is_none(), "root elapsed_ms must stay None");
// seq 1 — real upstream + elapsed.
let b = &got[1];
assert_eq!(b.repo, "b");
assert_eq!(
b.depends_on.as_deref(),
Some(["a".to_string()].as_slice()),
"deps list"
);
assert_eq!(b.elapsed_ms, Some(4321), "elapsed_ms");
// seq 2 — PRESENT-but-EMPTY deps must NOT collapse to None (the proto
// presence flag is the whole point of this assertion).
let a = &got[2];
assert_eq!(a.repo, "a");
assert_eq!(
a.depends_on,
Some(vec![]),
"empty deps list must stay Some(empty), not None"
);
assert_eq!(a.elapsed_ms, Some(7));
}
// ════════════════════════════════════════════════════════════════════════════
// 4. BuildInfo.Submit → build_info + build_provenance → Warehouse.Scan
// (native landing for a dwarves build's build-info — no more test_results interim)
// ════════════════════════════════════════════════════════════════════════════
/// A captured `BuildProvenance` + its §f3 manifest/feed rows, for the submit
/// round-trip. Uses a real tempdir + artifact so the CAS sha256 is genuine.
fn build_info_fixture() -> (
tempfile::TempDir,
nornir_build_thing::BuildProvenance,
Vec<nornir_build_thing::build_info::BuildInfoRow>,
) {
use nornir_build_thing::build_info::{BuildInfoKind, BuildInfoRow, manifest_rows};
use nornir_build_thing::provenance::{BuildProvenance, HostInfo, Outcome, Toolchain};
use nornir_build_thing::variant::{BuildVariant, Profile};
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
std::fs::create_dir_all(root.join("target/release")).unwrap();
let art = root.join("target/release/dwarves-full");
std::fs::write(&art, b"ELF dwarves-full bytes").unwrap();
let v = BuildVariant {
name: "full".into(),
bin: Some("dwarves-full".into()),
features: vec!["full".into(), "cli".into()],
profile: Profile::Release,
..Default::default()
};
let rec = BuildProvenance::capture(
&v,
"dwarves",
root,
Toolchain {
rustc: "rustc 1.85.0".into(),
host: "x86_64".into(),
cargo: "cargo 1.85.0".into(),
// a build on the host: no container image to name
image: None,
image_sha256: None,
},
HostInfo {
cores: 12,
memory_bytes: 64 * 1024 * 1024 * 1024,
},
Outcome::Success,
&[art],
)
.unwrap();
let mut rows = manifest_rows(&rec, 1_700_000_010_000);
// A live-feed log row WITH an inline blob — the nullable column, present.
rows.push(
BuildInfoRow::new(&rec.id, 1_700_000_011_000, BuildInfoKind::Log)
.step("cargo-build")
.level("info")
.blob(b"compiling dwarves".to_vec()),
);
(dir, rec, rows)
}
#[test]
fn submit_build_info_and_provenance_land_natively() {
let srv = spawn_server();
let (_d, rec, rows) = build_info_fixture();
// Both tables start empty — a non-zero read-back proves the RPC persisted.
let bi_before = remote::scan_table(&srv.endpoint, &srv.token, "build_info", 200, WS)
.map(|p| p.rows.len())
.unwrap_or(0);
assert_eq!(bi_before, 0, "build_info should start empty");
// ── SubmitBuildInfo → build_info (the native §f3 landing) ──
let accepted = remote::submit_build_info(&srv.endpoint, &srv.token, WS, &rows).unwrap();
assert_eq!(accepted, rows.len() as u32, "server accepted count");
let scan = remote::scan_table(&srv.endpoint, &srv.token, "build_info", 200, WS).unwrap();
assert_eq!(scan.rows.len(), rows.len(), "build_info rows after submit");
let col = |name: &str| {
scan.columns
.iter()
.position(|c| c == name)
.unwrap_or_else(|| panic!("column `{name}` in {:?}", scan.columns))
};
let c_kind = col("kind");
let c_name = col("name");
let c_value = col("value");
let c_cid = col("content_id");
// provenance repo fact landed.
assert!(
scan.rows
.iter()
.any(|r| r[c_kind] == "provenance" && r[c_name] == "repo" && r[c_value] == "dwarves"),
"provenance repo=dwarves row"
);
// dep rows = the two enabled features.
let deps: std::collections::BTreeSet<&str> = scan
.rows
.iter()
.filter(|r| r[c_kind] == "dep")
.map(|r| r[c_value].as_str())
.collect();
assert_eq!(
deps,
["cli", "full"].into_iter().collect(),
"dep rows = features"
);
// RED-when-broken: the artifact CAS id (sha256) survives the round-trip.
let art_row = scan
.rows
.iter()
.find(|r| r[c_kind] == "artifact")
.expect("artifact row");
assert_eq!(
art_row[c_cid], rec.artifacts[0].sha256,
"artifact content_id = sha256"
);
// ── SubmitBuildProvenance → build_provenance ledger ──
let n =
remote::submit_build_provenance(&srv.endpoint, &srv.token, WS, std::slice::from_ref(&rec))
.unwrap();
assert_eq!(n, 1, "one provenance record accepted");
let prov_scan =
remote::scan_table(&srv.endpoint, &srv.token, "build_provenance", 100, WS).unwrap();
assert_eq!(
prov_scan.rows.len(),
1,
"build_provenance ledger row after submit"
);
let p_repo = prov_scan.columns.iter().position(|c| c == "repo").unwrap();
assert_eq!(prov_scan.rows[0][p_repo], "dwarves", "ledger repo landed");
}