src/warehouse/step_feed.rs
//! **What a Runa step measured, filed** — the warehouse half of
//! `Telemetry.SubmitStepFeed`.
//!
//! A forge's Runa pipeline runs the tests (`Nextest` / `CargoTest`), the bench
//! example (`Bench`) and the coverage build (`Coverage`) of a checkout; the forge
//! reads what each step printed and files it here as a jera job, keyed by the
//! checkout's (repo, commit). These are the facts three Mímir lenses join with
//! the code room:
//!
//! | step | table(s) | lens |
//! |---|---|---|
//! | `Nextest` / `CargoTest` | `test_outcomes` (repo, crate, status, commit on the row) | `weak_tests` |
//! | `Bench` | `bench_runs` (+ `commit`) · `bench_results` · its own `test_outcomes` | `bench_gaps` |
//! | `Coverage` | `coverage` · `coverage_fn` (+ `line`, `end_line`, `commit`) | `change_risk` |
//! | `Trace` | `fn_trace_events` · `fn_trace_runs` (one filing per chunk of the run) | the 3D map's replay |
//! | `Profile` | `fn_profile` | the 3D map's heat layer |
//!
//! A trace's or profile's functions are named by symbol and source location; the
//! filer says which code room function each is ([`RoomMap`], the server's room)
//! and the rows carry its stable id.
//!
//! Every filing carries a `run_id` the PRODUCER minted, the same for a retried
//! filing: a run already in its table is accepted as 0 rows, never written twice.
use anyhow::{Context, Result, bail};
use super::coverage::{append_report, rows_for};
use super::iceberg::{IcebergWarehouse, StepTestOutcome};
use crate::bench::BenchRun;
use crate::coverage::CoverageReport;
/// What one step measured.
#[derive(Debug, Clone)]
pub enum Measured {
/// A test step's cases.
Tests(Vec<StepTestOutcome>),
/// The bench example's run.
Bench(Box<BenchRun>),
/// The coverage export, summarised.
Coverage(Box<CoverageReport>),
/// One self-standing chunk of a traced run.
Trace(Box<TraceChunk>),
/// A sampled profile.
Profile(Box<nornir_trace::wire::ProfileFiling>),
}
/// One chunk of a traced run as filed: its facts and its decoded events.
#[derive(Debug, Clone)]
pub struct TraceChunk {
pub filing: nornir_trace::wire::ChunkFiling,
pub events: Vec<nornir_trace::TraceEvent>,
/// Where it was measured (`runa <pipeline> <run> #<seq> <step>`).
pub source: String,
}
/// **Which code room function each filed function is**: its stable id
/// (`mimir_schema::moniker::stable`), `""` for none. The server's is its room.
pub trait RoomMap: Sync {
fn stable_ids(&self, fns: &[nornir_trace::wire::FiledFn]) -> Vec<String>;
}
/// Stable ids already worked out (in [`StepFeed::filed_fns`] order) — what a
/// server hands over after mapping on a blocking thread.
pub struct GivenIds(pub Vec<String>);
impl RoomMap for GivenIds {
fn stable_ids(&self, fns: &[nornir_trace::wire::FiledFn]) -> Vec<String> {
let mut v = self.0.clone();
v.resize(fns.len(), String::new());
v
}
}
/// No room in reach: nothing maps (the rows say so with a NULL stable id).
pub struct NoRoom;
impl RoomMap for NoRoom {
fn stable_ids(&self, fns: &[nornir_trace::wire::FiledFn]) -> Vec<String> {
vec![String::new(); fns.len()]
}
}
impl Measured {
/// The wire's `kind`.
pub fn kind(&self) -> &'static str {
match self {
Self::Tests(_) => "tests",
Self::Bench(_) => "bench",
Self::Coverage(_) => "coverage",
Self::Trace(_) => "trace",
Self::Profile(_) => "profile",
}
}
}
/// One filing: what was measured, of which (repo, commit), under which run.
#[derive(Debug, Clone)]
pub struct StepFeed {
pub repo: String,
pub commit: String,
pub run_id: String,
pub ts_micros: i64,
pub measured: Measured,
/// An `only: Affected` step: what the Mímir graph selected (`None` = every
/// test ran).
pub selection: Option<StepSelection>,
}
/// **What an `only: Affected` step ran** — `Test.Affected`'s answer, as the
/// forge passes it on (`SubmitStepFeedRequest.selection_json`).
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct StepSelection {
/// `affected` | `all`.
pub mode: String,
/// The commit the selection compared with (the previous green one).
pub baseline: String,
/// The one line that says why.
pub why: String,
/// Tests selected / tests the graph knows.
pub selected: u64,
pub tests_total: u64,
/// The runner names the step skipped.
pub skipped: Vec<String>,
/// The items that changed since the baseline, where they are now.
pub changed: Vec<super::coverage::ChangedSpan>,
}
impl StepSelection {
/// Parse the wire's JSON (`""` = none: every test ran).
pub fn parse(json: &str) -> Result<Option<Self>> {
if json.trim().is_empty() {
return Ok(None);
}
Ok(Some(serde_json::from_str(json).context("selection_json is no StepSelection")?))
}
/// Did the step run only what the graph selected?
pub fn is_affected(&self) -> bool {
self.mode == "affected"
}
/// The marker a skipped case is filed with.
pub fn marker(&self) -> String {
format!(
"selected by graph: {} of {} tests ran, {} skipped — nothing they reach changed since {}",
self.selected,
self.tests_total,
self.skipped.len(),
self.baseline.chars().take(12).collect::<String>()
)
}
}
impl StepFeed {
/// Decode the wire's payload by `kind` (`tests` rows are already typed; the
/// bench run and the coverage report travel as their own JSON).
#[allow(clippy::too_many_arguments)]
pub fn decode(
repo: String,
commit: String,
run_id: String,
ts_micros: i64,
kind: &str,
tests: Vec<StepTestOutcome>,
bench_json: &str,
coverage_json: &str,
selection_json: &str,
) -> Result<Self> {
let measured = match kind {
"tests" => Measured::Tests(tests),
"bench" => Measured::Bench(Box::new(serde_json::from_str(bench_json).context("bench_json is no BenchRun")?)),
"coverage" => Measured::Coverage(Box::new(serde_json::from_str(coverage_json).context("coverage_json is no CoverageReport")?)),
other => bail!("unknown step feed kind `{other}` (tests | bench | coverage)"),
};
Ok(Self { repo, commit, run_id, ts_micros, measured, selection: StepSelection::parse(selection_json)? })
}
/// The functions a trace chunk or a profile names, in the order a
/// [`RoomMap`] is asked them (empty for the other kinds).
pub fn filed_fns(&self) -> Vec<nornir_trace::wire::FiledFn> {
match &self.measured {
Measured::Trace(c) => c.filing.functions.clone(),
Measured::Profile(p) => p.functions.iter().map(|h| h.at.clone()).collect(),
_ => Vec::new(),
}
}
/// Decode a `trace` (a chunk: its JSON facts + its `NTR1` events) or a
/// `profile` filing.
#[allow(clippy::too_many_arguments)]
pub fn decode_run(
repo: String,
commit: String,
run_id: String,
ts_micros: i64,
kind: &str,
trace_json: &str,
trace_events: &[u8],
source: String,
) -> Result<Self> {
let measured = match kind {
"trace" => {
let filing: nornir_trace::wire::ChunkFiling = serde_json::from_str(trace_json).context("trace_json is no ChunkFiling")?;
let events = nornir_trace::wire::decode(trace_events, filing.base_seq).context("trace_events")?;
if let Some(e) = events.iter().find(|e| e.func as usize >= filing.functions.len()) {
bail!("trace event #{} names function {} of a chunk that has {}", e.seq_no, e.func, filing.functions.len());
}
Measured::Trace(Box::new(TraceChunk { filing, events, source }))
}
"profile" => Measured::Profile(Box::new(serde_json::from_str(trace_json).context("trace_json is no ProfileFiling")?)),
other => bail!("`{other}` is no run filing (trace | profile)"),
};
Ok(Self { repo, commit, run_id, ts_micros, measured, selection: None })
}
}
/// **File one step's measurement.** Returns the rows written (`0` = this run was
/// already filed, or it measured nothing).
pub async fn file(wh: &IcebergWarehouse, f: &StepFeed) -> Result<usize> {
file_mapped(wh, f, &NoRoom).await
}
/// [`file`], a trace's or profile's functions mapped to the room by `map`.
pub async fn file_mapped(wh: &IcebergWarehouse, f: &StepFeed, map: &dyn RoomMap) -> Result<usize> {
if f.repo.trim().is_empty() {
bail!("a step feed names no repo");
}
if f.run_id.trim().is_empty() {
bail!("a step feed carries no run id (a retried filing could not be told from a new one)");
}
match &f.measured {
Measured::Tests(cases) => {
// an `only: Affected` step: each test it skipped is filed as a
// `skipped` case under the same run, with the marker — no verdict
// (`weak_tests` counts no run of it), but the run says it ran 1 of N
let mut all = cases.clone();
if let Some(sel) = f.selection.as_ref().filter(|s| s.is_affected()) {
let ran: std::collections::BTreeSet<&str> = cases.iter().map(|c| c.name.as_str()).collect();
let marker = sel.marker();
all.extend(sel.skipped.iter().filter(|n| !ran.contains(n.as_str())).map(|n| StepTestOutcome {
name: n.clone(),
krate: String::new(),
status: SKIPPED.to_string(),
duration_ms: None,
message: Some(marker.clone()),
}));
}
wh.append_step_test_outcomes_async(&f.repo, &f.commit, &f.run_id, &all).await
}
Measured::Bench(run) => {
let run_id = uuid::Uuid::parse_str(&f.run_id).context("a bench filing's run id is a uuid (bench_runs keys by it)")?;
let filed = wh.append_bench_run_filed_async(&f.repo, run, run_id, &f.commit).await?;
Ok(if filed { 1 + run.results.len() + run.tests.len() } else { 0 })
}
Measured::Coverage(report) => {
if wh.run_filed_async(super::iceberg::TABLE_COVERAGE_FN, &f.run_id).await? {
return Ok(0);
}
let (rows, mut fn_rows) = rows_for(report, &f.run_id, &f.repo, f.ts_micros);
for r in &mut fn_rows {
r.commit = f.commit.clone();
}
// an `only: Affected` run counted the selected tests only: merged
// with the repo's newest earlier run for every unchanged function
if let Some(sel) = f.selection.as_ref().filter(|s| s.is_affected()) {
let previous = latest_run(super::coverage::query_coverage_fn(wh, &f.repo).await?, &f.run_id);
let m = super::coverage::merge_affected(&mut fn_rows, &previous, &sel.changed);
if !m.changed_uncovered.is_empty() {
// the rows carry the flag (`changed` and not executed); the log says it once
eprintln!(
"coverage {} @ {}: {} changed function(s) no selected test executed: {}",
f.repo,
f.commit.chars().take(12).collect::<String>(),
m.changed_uncovered.len(),
m.changed_uncovered.iter().take(20).cloned().collect::<Vec<_>>().join(", ")
);
}
}
append_report(wh, &rows, &fn_rows).await?;
Ok(fn_rows.len())
}
Measured::Trace(chunk) => file_trace_chunk(wh, f, chunk, map).await,
Measured::Profile(p) => {
if super::fn_trace::profile_filed(wh, &f.run_id).await? {
return Ok(0);
}
let at: Vec<nornir_trace::wire::FiledFn> = p.functions.iter().map(|h| h.at.clone()).collect();
let ids = map.stable_ids(&at);
let rows: Vec<super::fn_trace::ProfileRow> = p
.functions
.iter()
.zip(ids)
.map(|(h, stable_id)| super::fn_trace::ProfileRow {
stable_id,
symbol: h.at.symbol.clone(),
self_ns: h.self_ns,
total_ns: h.total_ns,
self_samples: h.self_samples,
total_samples: h.total_samples,
})
.collect();
super::fn_trace::append_profile(wh, &f.repo, &f.commit, &f.run_id, f.ts_micros, &p.target, p.samples, p.total_ns, &rows).await?;
Ok(rows.len())
}
}
}
/// One chunk: its functions mapped to the room, its events written with their
/// stable ids, its summary row with how much mapped.
async fn file_trace_chunk(wh: &IcebergWarehouse, f: &StepFeed, c: &TraceChunk, map: &dyn RoomMap) -> Result<usize> {
use super::fn_trace::{EventRow, RunRow, append_chunk, chunk_filed};
let fl = &c.filing;
if chunk_filed(wh, &f.run_id, fl.chunk).await? {
return Ok(0);
}
let ids = map.stable_ids(&fl.functions);
let rows: Vec<EventRow> = c
.events
.iter()
.map(|e| EventRow {
thread: e.thread,
seq_no: e.seq_no,
t_ns: e.t_ns,
stable_id: ids[e.func as usize].clone(),
kind: e.kind,
symbol: fl.functions[e.func as usize].symbol.clone(),
})
.collect();
let local = |i: usize| !fl.functions[i].repo.is_empty();
let run = RunRow {
run_id: f.run_id.clone(),
ts_micros: f.ts_micros,
repo: f.repo.clone(),
commit: f.commit.clone(),
target: fl.target.clone(),
source: c.source.clone(),
chunk: fl.chunk,
chunks: fl.chunks,
events: rows.len() as u64,
mapped_events: rows.iter().filter(|r| !r.stable_id.is_empty()).count() as u64,
functions: fl.functions.len() as u64,
mapped_functions: ids.iter().filter(|s| !s.is_empty()).count() as u64,
local_functions: (0..fl.functions.len()).filter(|&i| local(i)).count() as u64,
mapped_local_functions: (0..fl.functions.len()).filter(|&i| local(i) && !ids[i].is_empty()).count() as u64,
threads: fl.threads as u64,
first_ns: c.events.first().map(|e| e.t_ns).unwrap_or(0),
last_ns: c.events.last().map(|e| e.t_ns).unwrap_or(0),
total_events: fl.total_events,
duration_ns: fl.duration_ns,
truncated: fl.truncated,
binaries: serde_json::to_string(&fl.binaries).unwrap_or_default(),
};
append_chunk(wh, &f.repo, &f.commit, &run, &rows).await?;
Ok(rows.len())
}
/// The status a test an `only: Affected` step skipped is filed with.
pub const SKIPPED: &str = "skipped";
/// The rows of the newest coverage run in `rows` (by time, then run id) other
/// than `except` — the run an affected run merges with.
fn latest_run(rows: Vec<super::coverage::CoverageFnRow>, except: &str) -> Vec<super::coverage::CoverageFnRow> {
let newest = rows.iter().filter(|r| r.run_id != except).map(|r| (r.ts_micros, r.run_id.clone())).max();
match newest {
Some((_, id)) => rows.into_iter().filter(|r| r.run_id == id).collect(),
None => Vec::new(),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn case(name: &str, status: &str) -> StepTestOutcome {
StepTestOutcome { name: name.into(), krate: "shop".into(), status: status.into(), duration_ms: Some(1.0), message: None }
}
/// An `only: Affected` test step files what it ran as today, and each test
/// it skipped as a `skipped` case with the marker — which `weak_tests`
/// counts as no run.
#[test]
fn an_affected_test_step_files_its_skipped_tests_with_the_marker() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let sel = r#"{"mode":"affected","baseline":"c0ffee1234567890","why":"1 changed item","selected":1,"tests_total":3,"skipped":["cart::tests::adds_up","pay::tests::slow"],"changed":[]}"#;
let f = StepFeed::decode("shop".into(), "c2".into(), "run-1".into(), 1, "tests", vec![case("pay::tests::charges_once", "pass")], "", "", sel).unwrap();
assert_eq!(wh.block_on(file(&wh, &f)).unwrap(), 3, "1 ran + 2 skipped");
let (p, _) = wh.scan_preview_bounded("test_outcomes", 100, 1 << 20).unwrap();
let rows = crate::mimir_lens::rows::parse_test_outcomes(&crate::mimir_lens::rows::Table::new(p.columns, p.rows));
let skipped: Vec<&str> = rows.iter().filter(|r| r.status == SKIPPED).map(|r| r.test_name.as_str()).collect();
assert_eq!(skipped.len(), 2, "{rows:?}");
assert!(skipped.contains(&"cart::tests::adds_up"));
let marked = rows.iter().find(|r| r.status == SKIPPED).unwrap();
assert!(!marked.ran() && !marked.passed, "a skipped case is no run and no pass: {marked:?}");
// a step of every test files no marker
let all = StepFeed::decode("shop".into(), "c3".into(), "run-2".into(), 2, "tests", vec![case("a", "pass")], "", "", "").unwrap();
assert!(all.selection.is_none());
assert_eq!(wh.block_on(file(&wh, &all)).unwrap(), 1);
}
}