src/bench_registry.rs
//! nornir's OWN linked bencher registrations (Task #39 — bench-move).
//!
//! The `register_bench!` calls MUST compile INTO the `nornir` library so that
//! `inventory` links them into every binary/test that uses the lib — crucially
//! the release completeness-gate binary. The bencher-discovery seam the coverage
//! gate sweeps, [`nornir_bench::api::registered_bench_ids`], reports exactly the
//! benchers linked into the CURRENT unit; when nornir's benchers lived only in
//! `examples/nornir-bench.rs` (an example is NOT linked into the lib) that seam
//! saw ZERO benchers at gate time and every hot-path fell back to a seeded
//! allowlist excuse — the gate had nothing real to check.
//!
//! With the registrations here, a gate run sees `nornir.dep_graph_build` /
//! `nornir.dep_graph_build_ext`, so the `nornir.dep_graph` hot-path is verified
//! for real (its allowlist entry is deleted). `examples/nornir-bench.rs` stays
//! the RUNNER: it links this lib, so `run_main_json()` iterates these same
//! `inventory` registrations — no bencher is defined in the example any more.
//!
//! ## Benches
//!
//! - `nornir.dep_graph_build` / `nornir.dep_graph_build_ext` — the B-bucket
//! dependency-Mímir workload, cloned from `tests/scenarios.rs::l3/l4`.
//! Generates a layered-DAG synthetic workspace of 50 real cargo crates, times
//! [`dep_graph::build`](crate::warehouse::dep_graph::build)
//! (which runs `cargo metadata` per repo), then times the warm
//! transitive-closure queries the Mímir answers (`dependents_transitive` /
//! `deps_transitive`) over every repo. The `_ext` variant additionally pins 300
//! locked "external" crates so the build also resolves external-dependency edges.
//!
//! The synthetic-workspace generator is the same one the scenario tests use,
//! `#[path]`-included exactly like `scenarios.rs` does. It is kept private here so
//! moving it into the lib adds NO public API surface — only the `inventory`
//! registrations become visible, via the discovery seam.
#![allow(dead_code)]
use anyhow::{Result, ensure};
use crate::bench::BenchResult;
use crate::bench::api::Bencher;
use crate::workspace::WorkspaceDescriptor;
#[path = "../tests/support/genworkspace.rs"]
mod genworkspace;
use genworkspace::{GenSpec, Shape, gen_workspace};
/// B-bucket: build a `WorkspaceGraph` over a synthetic layered DAG, then run
/// warm transitive-closure queries over every repo — the dependency-Mímir's
/// hot path.
///
/// This is a **latency / IO microbench, NOT a parallel-compute arm** (bencher
/// O1): the build phase is one `cargo metadata` subprocess per repo (I/O +
/// process bound) and the query phase is a warm in-memory closure battery. The
/// **authoritative metric is `queries_per_sec`** (~600K, run-to-run flat);
/// `build_ms` is host-contention wall-time that swings ~10× and must NOT be read
/// as a compute regression or the arm treated as multi-threaded. The
/// core-saturation LAW therefore exempts it (see
/// [`crate::release::gate::is_latency_microbench`]); the genuinely parallel
/// nornir headline arms are `knowledge_scan` / `warehouse_query`.
struct DepGraphBuild {
id: &'static str,
repos: usize,
ext_crates: usize,
}
impl Bencher for DepGraphBuild {
fn id(&self) -> &'static str {
self.id
}
fn run(&self) -> Result<BenchResult> {
// Build the synthetic workspace on disk (real cargo crates + descriptor).
let mut spec = GenSpec::new(Shape::LayeredDag, self.repos).seed(42);
if self.ext_crates > 0 {
spec = spec.ext_crates(self.ext_crates);
}
let ws = gen_workspace(&spec);
// ── Time the graph build (one `cargo metadata` per repo). ──
let desc = WorkspaceDescriptor::load(&ws.descriptor_path)?;
let t0 = std::time::Instant::now();
let g = crate::warehouse::dep_graph::build(&desc)?;
let build = t0.elapsed();
// ── Time the warm closure queries (the Mímir's hot answers). ──
// Two closures per repo, exactly the L3/L4 query battery.
let t1 = std::time::Instant::now();
let mut answered = 0u64;
for name in &ws.repo_names {
let _ = g.dependents_transitive(name);
let _ = g.deps_transitive(name);
answered += 2;
}
let queries = t1.elapsed();
let crates = ws.repo_names.len() as u64;
let queries_per_sec = if queries.as_secs_f64() > 0.0 {
answered as f64 / queries.as_secs_f64()
} else {
0.0
};
let mut metrics = serde_json::Map::new();
metrics.insert(
"build_ms".into(),
serde_json::json!(build.as_secs_f64() * 1000.0),
);
metrics.insert("crates".into(), serde_json::json!(crates));
metrics.insert("queries".into(), serde_json::json!(answered));
metrics.insert("queries_per_sec".into(), serde_json::json!(queries_per_sec));
metrics.insert("ext_crates".into(), serde_json::json!(self.ext_crates));
metrics.insert(
"elapsed_ms".into(),
serde_json::json!((build + queries).as_secs_f64() * 1000.0),
);
Ok(BenchResult {
name: self.id.into(),
metrics,
..Default::default()
})
}
}
// `nornir.dep_graph_build` — 50 repos, no externals (mirrors L3).
crate::register_bench!(DepGraphBuild {
id: "nornir.dep_graph_build",
repos: 50,
ext_crates: 0,
});
// `nornir.dep_graph_build_ext` — 50 repos + 300 locked externals (mirrors L4).
crate::register_bench!(DepGraphBuild {
id: "nornir.dep_graph_build_ext",
repos: 50,
ext_crates: 300,
});
/// The **vector-search** hot path (`BENCH_HOT_PATHS[nornir.vector_search]`): the
/// embedding similarity query the semantic-search index answers. Fills a real
/// [`crate::vector::VectorIndex`] with `rows` deterministic `dim`-D vectors
/// (over [`crate::vector::I8_SCAN_THRESHOLD`], so `search_auto` dispatches the
/// same int8/VNNI kernel `store::search` uses in production), then times a
/// battery of top-k nearest-neighbour queries — the exact cosine-similarity
/// scan the `vector_search` MCP tool / CLI runs per query.
#[cfg(feature = "vector")]
struct VectorSearch {
id: &'static str,
rows: usize,
dim: usize,
queries: usize,
k: usize,
}
#[cfg(feature = "vector")]
impl Bencher for VectorSearch {
fn id(&self) -> &'static str {
self.id
}
fn run(&self) -> Result<BenchResult> {
use crate::vector::VectorIndex;
// Deterministic pseudo-embedding: a smooth per-(row,dim) function so every
// vector is distinct and non-degenerate (no external model needed — we are
// benching the SEARCH kernel, not the embedder).
let mk = |seed: u64, dim: usize| -> Vec<f32> {
(0..dim)
.map(|d| {
let x = (seed.wrapping_mul(2_654_435_761).wrapping_add(d as u64)) as f32;
(x * 0.000_1).sin() + 0.001 * (d as f32 + 1.0)
})
.collect()
};
// ── Build the index (excluded from the query timing). ──
let mut idx = VectorIndex::new(self.dim)?;
let mut flat = Vec::with_capacity(self.rows * self.dim);
let mut ids = Vec::with_capacity(self.rows);
for r in 0..self.rows as u64 {
flat.extend(mk(r, self.dim));
ids.push(r * 7 + 3);
}
idx.add(&flat, &ids)?;
// ── Time the top-k query battery (the real similarity search). ──
let t0 = std::time::Instant::now();
let mut hits = 0u64;
for q in 0..self.queries as u64 {
let query = mk(q.wrapping_mul(31).wrapping_add(1_000), self.dim);
hits += idx.search_auto(&query, self.k).len() as u64;
}
let elapsed = t0.elapsed();
let queries_per_sec = if elapsed.as_secs_f64() > 0.0 {
self.queries as f64 / elapsed.as_secs_f64()
} else {
0.0
};
let mut metrics = serde_json::Map::new();
metrics.insert("rows".into(), serde_json::json!(self.rows));
metrics.insert("dim".into(), serde_json::json!(self.dim));
metrics.insert("queries".into(), serde_json::json!(self.queries));
metrics.insert("k".into(), serde_json::json!(self.k));
metrics.insert("hits".into(), serde_json::json!(hits));
metrics.insert("queries_per_sec".into(), serde_json::json!(queries_per_sec));
metrics.insert(
"elapsed_ms".into(),
serde_json::json!(elapsed.as_secs_f64() * 1000.0),
);
Ok(BenchResult {
name: self.id.into(),
metrics,
..Default::default()
})
}
}
// `nornir.vector_search` — 8192×128-D index, 256 top-10 similarity queries.
#[cfg(feature = "vector")]
crate::register_bench!(VectorSearch {
id: "nornir.vector_search",
rows: 8192,
dim: 128,
queries: 256,
k: 10,
});
/// The **warehouse-query** hot path (`BENCH_HOT_PATHS[nornir.warehouse_query]`):
/// a representative columnar read out of the local Iceberg warehouse. Writes one
/// coverage snapshot of `files` files (via the real `rows_for` + `append_report`
/// writer) into a temp-dir warehouse, then times
/// [`crate::warehouse::coverage::query_coverage`] — the Arrow `RecordBatch` scan
/// + decode + per-repo filter + sort that the Test-pane coverage panel and the
/// docs `nornir:gen:coverage` table read through. Fully offline (no network).
struct WarehouseQuery {
id: &'static str,
files: usize,
}
impl Bencher for WarehouseQuery {
fn id(&self) -> &'static str {
self.id
}
fn run(&self) -> Result<BenchResult> {
use crate::coverage::{Boundary, FileCoverage, FnCoverage, summarise};
use crate::warehouse::coverage::{append_report, query_coverage, rows_for};
use crate::warehouse::iceberg::IcebergWarehouse;
// A synthetic coverage report over `files` files (each with one fn), rolled
// up by the real `summarise`, so the row shapes match a live `nornir
// coverage` run exactly.
let file_cov: Vec<FileCoverage> = (0..self.files)
.map(|i| {
let file = format!("src/gen/m{i}.rs");
FileCoverage {
file: file.clone(),
krate: "nornir".into(),
lines: 20,
lines_covered: (i as u64 % 21),
regions: 12,
regions_covered: (i as u64 % 13),
functions: vec![FnCoverage {
name: format!("nornir::gen::m{i}::f{i}"),
file,
lines: 20,
lines_covered: (i as u64 % 21),
regions: 12,
regions_covered: (i as u64 % 13),
boundary: Boundary::Core,
line: 0,
end_line: 0,
}],
}
})
.collect();
let report = summarise("nornir", file_cov)?;
let (rows, fn_rows) = rows_for(&report, "bench-run", "nornir", 1_000);
let dir = tempfile::tempdir()?;
let wh = IcebergWarehouse::open(dir.path())?;
// ── Write one snapshot (excluded from the read timing). ──
wh.block_on(append_report(&wh, &rows, &fn_rows))?;
// ── Time the representative read query. ──
let t0 = std::time::Instant::now();
let back = wh.block_on(query_coverage(&wh, "nornir"))?;
let elapsed = t0.elapsed();
let read = back.len() as u64;
let rows_per_sec = if elapsed.as_secs_f64() > 0.0 {
read as f64 / elapsed.as_secs_f64()
} else {
0.0
};
let mut metrics = serde_json::Map::new();
metrics.insert("files".into(), serde_json::json!(self.files));
metrics.insert("rows_written".into(), serde_json::json!(rows.len()));
metrics.insert("rows_read".into(), serde_json::json!(read));
metrics.insert("rows_per_sec".into(), serde_json::json!(rows_per_sec));
metrics.insert(
"elapsed_ms".into(),
serde_json::json!(elapsed.as_secs_f64() * 1000.0),
);
Ok(BenchResult {
name: self.id.into(),
metrics,
..Default::default()
})
}
}
// `nornir.warehouse_query` — read back a 500-file coverage snapshot from Iceberg.
crate::register_bench!(WarehouseQuery {
id: "nornir.warehouse_query",
files: 500,
});
/// The **knowledge-graph query** hot path (R0.3 / R2.1): the call-graph questions
/// the MCP `callers_of` / `callees_of` / `call_path` tools and the viz code-graph
/// panel answer over a loaded [`crate::knowledge::query::KnowledgeView`]. Builds a
/// synthetic `nodes`-deep call chain (`f0 → f1 → … → fN`) with a shared fan-in
/// callee (`util::helper`, invoked by every node), warms the memoized inverted
/// index + adjacency ONCE, then times a battery of:
/// * `call_path(f0, fN)` — a full-depth BFS over the whole chain (now a dense
/// CSR integer BFS: nodes interned to `u32` ids once at index time, so the
/// traversal walks flat `Vec` arrays with zero per-node string hashing), and
/// * `callers_of("helper")` — an O(hits) posting-list lookup returning all N
/// fan-in edges.
/// Pure/offline (no warehouse, no network) — a clean scaling number for the
/// query-layer index, complementing `nornir.knowledge_scan` (the build side).
struct KnowledgeQuery {
id: &'static str,
nodes: usize,
queries: usize,
}
impl Bencher for KnowledgeQuery {
fn id(&self) -> &'static str {
self.id
}
fn run(&self) -> Result<BenchResult> {
use crate::knowledge::query::KnowledgeView;
use crate::knowledge::symbols::CallEdgeRow;
// A `nodes`-deep chain plus a shared fan-in callee, so both the deep
// `call_path` BFS and the wide `callers_of` posting lookup have real work.
let mut calls = Vec::with_capacity(self.nodes * 2);
for i in 0..self.nodes {
let caller = format!("app::m{i}::f{i}");
if i + 1 < self.nodes {
calls.push(CallEdgeRow {
crate_name: "app".into(),
caller_path: caller.clone(),
callee_ident: format!("app::m{}::f{}", i + 1, i + 1),
call_kind: "call".into(),
file: format!("src/m{i}.rs"),
line: 1,
});
}
calls.push(CallEdgeRow {
crate_name: "app".into(),
caller_path: caller,
callee_ident: "util::helper".into(),
call_kind: "call".into(),
file: format!("src/m{i}.rs"),
line: 2,
});
}
let view = KnowledgeView::new(vec![], calls);
let from = "f0".to_string();
let to = format!("f{}", self.nodes - 1);
// Warm the memoized index + adjacency OUTSIDE the timed region (its build
// is a one-time per-snapshot cost, not the per-query hot path).
let _ = view.call_path(&from, &to);
let _ = view.callers_of("helper");
// ── Time the query battery. ──
let t0 = std::time::Instant::now();
let mut path_hops = 0u64;
let mut fan_in = 0u64;
for _ in 0..self.queries {
if let Some(p) = view.call_path(&from, &to) {
path_hops += p.len() as u64;
}
fan_in += view.callers_of("helper").len() as u64;
}
let elapsed = t0.elapsed();
// Two queries per iteration (one deep BFS + one fan-in lookup).
let answered = self.queries as u64 * 2;
let queries_per_sec = if elapsed.as_secs_f64() > 0.0 {
answered as f64 / elapsed.as_secs_f64()
} else {
0.0
};
// ── R2.1 batched "Ragnar" stree leg ──────────────────────────────
// The battery above only touches the single-key HashMap path + the CSR
// BFS; it never routes a BATCH through the sorted-key Eytzinger S+tree
// (`callers_of_batch`) — Ragnar's pipelined SIMD + software-prefetch walk
// that overlaps cache-miss latency across the whole batch. Exercise it
// here so the heavy sweep captures ONE tree walk over the batch vs. N
// independent HashMap probes. New metrics only; nothing above changes.
//
// A mixed batch: every chain leaf `f{i}` (each a 1-hit or 0-hit lookup)
// plus the shared `helper` (an N-hit fan-in) — real routing work.
let batch_names: Vec<String> = (0..self.nodes)
.map(|i| format!("f{i}"))
.chain(std::iter::once("helper".to_string()))
.collect();
let name_refs: Vec<&str> = batch_names.iter().map(|s| s.as_str()).collect();
// Byte-identical guard (ONCE, outside timing): the batched stree path must
// return exactly what the per-name HashMap `callers_of` yields, positionally
// and in the same edge order — the stree only changes HOW a key is located.
let batched = view.callers_of_batch(&name_refs);
let per_name: Vec<Vec<u32>> = name_refs
.iter()
.map(|n| view.callers_of(n).iter().map(|e| e.line).collect())
.collect();
let batch_matches = batched.len() == per_name.len()
&& batched.iter().zip(&per_name).all(|(b, exp)| {
b.len() == exp.len() && b.iter().zip(exp).all(|(e, &l)| e.line == l)
});
// The BENCH RULE's byte-identical half, enforced in EVERY profile: benches run
// in release, where `debug_assert!` is compiled out, so a `debug_assert` here
// guarded nothing and `batch_speedup` below could have been earned by a batch
// path that returns different edges.
ensure!(
batch_matches,
"{}: `callers_of_batch` does not match per-name `callers_of` over {} names \
({} nodes) — batched returned {} lists ({} edges), per-name returned {} \
lists ({} edges); the R2.1 speedup would be measuring a wrong answer",
self.id,
name_refs.len(),
self.nodes,
batched.len(),
batched.iter().map(|b| b.len()).sum::<usize>(),
per_name.len(),
per_name.iter().map(|v| v.len()).sum::<usize>()
);
let batch_hits: u64 = per_name.iter().map(|v| v.len() as u64).sum();
// Time the batched path over the same `queries` budget.
let tb = std::time::Instant::now();
let mut batch_lookups = 0u64;
for _ in 0..self.queries {
for res in view.callers_of_batch(&name_refs) {
batch_lookups += res.len() as u64;
}
}
let batch_elapsed = tb.elapsed();
let names_routed = self.queries as u64 * name_refs.len() as u64;
let batch_names_per_sec = if batch_elapsed.as_secs_f64() > 0.0 {
names_routed as f64 / batch_elapsed.as_secs_f64()
} else {
0.0
};
// ── A/B baseline: the SAME batch resolved the OLD way ─────────────
// The Ragnar batch leg above was timed in isolation, so nothing here
// proved the stree actually BEATS the plain per-name HashMap loop —
// only the BENCH RULE's byte-identical half was guarded. Resolve the
// identical `name_refs` as N sequential single-key `callers_of` probes
// over the same `queries` budget: that is exactly what a caller pays
// WITHOUT the batch API, and what the pipelined SIMD + software-prefetch
// walk must out-run to earn its place. `batch_speedup = seq/batch` is
// the number the heavy sweep reads to keep-or-drop the R2.1 path.
let ts = std::time::Instant::now();
let mut seq_lookups = 0u64;
for _ in 0..self.queries {
for n in &name_refs {
seq_lookups += view.callers_of(n).len() as u64;
}
}
let seq_elapsed = ts.elapsed();
// Same total work both ways (byte-identical ⇒ identical hit counts): a
// second, independent cross-check of the equality guard above.
ensure!(
seq_lookups == batch_lookups,
"{}: the A/B legs of `batch_speedup` did not do the same work — the \
sequential per-name loop resolved {} edges, the batched stree path {}, \
over {} queries × {} names; a ratio of two different workloads is not a \
speedup",
self.id,
seq_lookups,
batch_lookups,
self.queries,
name_refs.len()
);
let batch_speedup = if batch_elapsed.as_secs_f64() > 0.0 {
seq_elapsed.as_secs_f64() / batch_elapsed.as_secs_f64()
} else {
0.0
};
let mut metrics = serde_json::Map::new();
metrics.insert("nodes".into(), serde_json::json!(self.nodes));
metrics.insert("queries".into(), serde_json::json!(answered));
metrics.insert("edges".into(), serde_json::json!(view.calls.len()));
metrics.insert("path_hops".into(), serde_json::json!(path_hops));
metrics.insert("fan_in_hits".into(), serde_json::json!(fan_in));
metrics.insert("queries_per_sec".into(), serde_json::json!(queries_per_sec));
// R2.1 batched stree leg.
metrics.insert("batch_names".into(), serde_json::json!(name_refs.len()));
metrics.insert("batch_hits".into(), serde_json::json!(batch_hits));
metrics.insert("batch_lookups".into(), serde_json::json!(batch_lookups));
metrics.insert("batch_matches".into(), serde_json::json!(batch_matches));
metrics.insert(
"batch_names_per_sec".into(),
serde_json::json!(batch_names_per_sec),
);
metrics.insert(
"batch_elapsed_ms".into(),
serde_json::json!(batch_elapsed.as_secs_f64() * 1000.0),
);
// A/B baseline (per-name HashMap loop) + the derived keep-or-drop ratio.
metrics.insert("seq_lookups".into(), serde_json::json!(seq_lookups));
metrics.insert(
"seq_elapsed_ms".into(),
serde_json::json!(seq_elapsed.as_secs_f64() * 1000.0),
);
metrics.insert("batch_speedup".into(), serde_json::json!(batch_speedup));
metrics.insert(
"elapsed_ms".into(),
serde_json::json!(elapsed.as_secs_f64() * 1000.0),
);
Ok(BenchResult {
name: self.id.into(),
metrics,
..Default::default()
})
}
}
// `nornir.knowledge_query` — 2000-node chain, 200 deep-BFS + fan-in query pairs.
crate::register_bench!(KnowledgeQuery {
id: "nornir.knowledge_query",
nodes: 2000,
queries: 200,
});
/// The **funnel Plan-DAG assembly** hot path: build a synthetic funnel whose one
/// idea decomposes into an `n`-node dependency chain, then time
/// [`funnel_core::plan_dag::PlanDag::from_funnel`] — the projection + edge
/// assembly the `nornir viz` Plan-DAG panel runs whenever the plan selection
/// changes. Pure/offline (no warehouse), so it's a clean scaling number for the
/// DAG-over-N-nodes path.
struct PlanDagAssemble {
id: &'static str,
nodes: usize,
}
impl Bencher for PlanDagAssemble {
fn id(&self) -> &'static str {
self.id
}
fn run(&self) -> Result<BenchResult> {
use funnel_core::event::{Event, ItemKind, PlanStatus};
use funnel_core::ids::{IdeaId, NodeId, PlanId};
use funnel_core::plan_dag::PlanDag;
use funnel_core::state::Funnel;
// Build the funnel: one idea → one Active plan → an n-node chain a→b→c→…
let mut f = Funnel::default();
let now = chrono::Utc::now();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "bench".into(),
text: "assemble".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})?;
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "chain".into(),
planner: "bench".into(),
ts: now,
})?;
f.apply(&Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: now,
})?;
for i in 1..=self.nodes as u64 {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some(format!("task {i}")),
ts: now,
})?;
if i > 1 {
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(i - 1),
to_node: NodeId::seq(i),
ts: now,
})?;
}
}
// ── Time the assembly (the hot path). ──
let iters = 200u64;
let t0 = std::time::Instant::now();
let mut last_nodes = 0usize;
for _ in 0..iters {
let dag = PlanDag::from_funnel(&f, None);
last_nodes = dag.node_count();
}
let elapsed = t0.elapsed();
let per = elapsed.as_secs_f64() * 1000.0 / iters as f64;
let mut metrics = serde_json::Map::new();
metrics.insert("nodes".into(), serde_json::json!(last_nodes));
metrics.insert("assemble_ms".into(), serde_json::json!(per));
metrics.insert(
"elapsed_ms".into(),
serde_json::json!(elapsed.as_secs_f64() * 1000.0),
);
metrics.insert(
"assemblies_per_sec".into(),
serde_json::json!(if elapsed.as_secs_f64() > 0.0 {
iters as f64 / elapsed.as_secs_f64()
} else {
0.0
}),
);
Ok(BenchResult {
name: self.id.into(),
metrics,
..Default::default()
})
}
}
// `nornir.plan_dag_assemble` — assemble the funnel Plan-DAG over a 200-node chain.
crate::register_bench!(PlanDagAssemble {
id: "nornir.plan_dag_assemble",
nodes: 200,
});
/// **Warehouse WRITE hot path** (previously un-benched): append N event-sourced
/// funnel events to the `funnel_events` Iceberg table — **each append is its own
/// snapshot** (the event-sourcing cost). `PlanDagAssemble` above covers the
/// in-memory DAG projection; nothing measured the *write* leg until now.
/// Authoritative metric: `events_per_sec`. Latency/IO microbench (one warehouse,
/// serial appends — not a parallel-compute arm).
///
/// Zero-copy / allocation candidate (see `.nornir/warehouse-write-bench.md`):
/// each `append_event` builds a fresh single-row Arrow `RecordBatch` (per-append
/// column builders) AND re-derives the `TableIdent` (`name.to_string()`) + runs
/// `ensure_table_schema` every call. **Batching many events into one append**
/// (one snapshot, builders sized once) is the win to prove here.
struct FunnelAppend {
id: &'static str,
events: usize,
/// `false` = one Iceberg snapshot per event (the fsync-bound baseline);
/// `true` = all N events in ONE snapshot via `append_events` (the win).
batch: bool,
}
impl Bencher for FunnelAppend {
fn id(&self) -> &'static str {
self.id
}
fn run(&self) -> Result<BenchResult> {
use crate::funnel::event::{Event, ItemKind};
use crate::funnel::ids::IdeaId;
use crate::warehouse::funnel::{append_event, append_events};
use crate::warehouse::iceberg::IcebergWarehouse;
let dir = tempfile::tempdir()?;
let wh = IcebergWarehouse::open(dir.path())?;
let now = chrono::Utc::now();
let evs: Vec<Event> = (0..self.events)
.map(|i| Event::IdeaSubmitted {
id: IdeaId::seq((i as u64) + 1),
source: "bench".into(),
text: format!("bench idea {i}"),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.collect();
let batch = self.batch;
let t0 = std::time::Instant::now();
wh.block_on(async {
if batch {
// One snapshot for all N events (fsync amortized).
append_events(&wh, &evs).await?;
} else {
for ev in &evs {
append_event(&wh, ev).await?;
}
}
anyhow::Ok(())
})?;
let elapsed = t0.elapsed();
let secs = elapsed.as_secs_f64();
let per_sec = if secs > 0.0 {
self.events as f64 / secs
} else {
0.0
};
let per_append_us = if self.events > 0 {
elapsed.as_micros() as f64 / self.events as f64
} else {
0.0
};
let mut metrics = serde_json::Map::new();
metrics.insert("events".into(), serde_json::json!(self.events));
metrics.insert("append_ms".into(), serde_json::json!(secs * 1000.0));
metrics.insert("events_per_sec".into(), serde_json::json!(per_sec));
metrics.insert("per_append_us".into(), serde_json::json!(per_append_us));
metrics.insert("elapsed_ms".into(), serde_json::json!(secs * 1000.0));
metrics.insert(
"mode".into(),
serde_json::json!(if self.batch { "batch" } else { "per_event" }),
);
metrics.insert(
"snapshots".into(),
serde_json::json!(if self.batch { 1 } else { self.events }),
);
Ok(BenchResult {
name: self.id.into(),
metrics,
..Default::default()
})
}
}
// `nornir.funnel_append` — 64 event-sourced appends, one snapshot each (the
// fsync-bound baseline). `nornir.funnel_append_batch` — the SAME 64 events in ONE
// snapshot via `append_events` (the fsync-amortizing win; this is the path
// `store_subtasks` now takes). Both light by design; the heavy batched-vs-serial
// sweep runs on the quiet bench box (Loki/Odin), never here.
crate::register_bench!(FunnelAppend {
id: "nornir.funnel_append",
events: 64,
batch: false,
});
crate::register_bench!(FunnelAppend {
id: "nornir.funnel_append_batch",
events: 64,
batch: true,
});
#[cfg(test)]
mod tests {
/// The whole point of moving the registrations into the lib (Task #39): the
/// discovery seam MUST be non-empty in a build that links this crate, and MUST
/// carry the `nornir.dep_graph` benchers — so the completeness gate verifies
/// that hot-path for real instead of excusing it via an allowlist entry.
/// Light smoke of the warehouse-WRITE bench (8 events, not the registered 64):
/// proves `FunnelAppend::run` opens a warehouse, appends event-sourced rows,
/// and reports a positive `events_per_sec`. The heavy sweep runs on the bench box.
#[test]
fn funnel_append_bench_runs_light() {
use crate::bench::api::Bencher;
// Per-event baseline (8 events, 8 snapshots).
let single = super::FunnelAppend {
id: "nornir.funnel_append",
events: 8,
batch: false,
}
.run()
.expect("funnel_append bench runs");
let eps = single
.metrics
.get("events_per_sec")
.and_then(|v| v.as_f64())
.unwrap_or(0.0);
let us = single
.metrics
.get("per_append_us")
.and_then(|v| v.as_f64())
.unwrap_or(0.0);
assert!(
eps > 0.0,
"events_per_sec must be positive: {:?}",
single.metrics
);
// Batched (8 events, 1 snapshot) — the amortized path.
let batched = super::FunnelAppend {
id: "nornir.funnel_append_batch",
events: 8,
batch: true,
}
.run()
.expect("funnel_append_batch bench runs");
let bus = batched
.metrics
.get("per_append_us")
.and_then(|v| v.as_f64())
.unwrap_or(0.0);
assert!(
batched
.metrics
.get("events_per_sec")
.and_then(|v| v.as_f64())
.unwrap_or(0.0)
> 0.0,
"batch events_per_sec must be positive: {:?}",
batched.metrics
);
assert_eq!(
batched.metrics.get("snapshots").and_then(|v| v.as_u64()),
Some(1),
"batch = 1 snapshot"
);
eprintln!(
"nornir.funnel_append light: per_event {us:.0} us/append (8 snap) vs batch {bus:.0} us/append (1 snap)"
);
}
/// Light smoke of the knowledge-graph query bench (200-node chain, 20 query
/// pairs — far below the registered 2000/200): proves it builds a view,
/// warms the index, answers a full-depth `call_path` and a fan-in
/// `callers_of`, and reports a positive `queries_per_sec`. The heavy scaling
/// sweep runs on the bench box (Loki/Odin).
#[test]
fn knowledge_query_bench_runs_light() {
use crate::bench::api::Bencher;
let out = super::KnowledgeQuery {
id: "nornir.knowledge_query",
nodes: 200,
queries: 20,
}
.run()
.expect("knowledge_query bench runs");
let qps = out
.metrics
.get("queries_per_sec")
.and_then(|v| v.as_f64())
.unwrap_or(0.0);
assert!(
qps > 0.0,
"queries_per_sec must be positive: {:?}",
out.metrics
);
// A 200-node chain yields a 200-hop path each iteration and 200 fan-in hits.
assert_eq!(
out.metrics.get("path_hops").and_then(|v| v.as_u64()),
Some(200 * 20)
);
assert_eq!(
out.metrics.get("fan_in_hits").and_then(|v| v.as_u64()),
Some(200 * 20)
);
// R2.1 batched stree leg: 200 chain leaves + `helper` = 201 routed names;
// the batched path must byte-match the per-name path and report throughput.
assert_eq!(
out.metrics.get("batch_names").and_then(|v| v.as_u64()),
Some(201)
);
assert_eq!(
out.metrics.get("batch_matches").and_then(|v| v.as_bool()),
Some(true)
);
// f1..f199 (1 hit each) + helper (200) + f0 (0) = 399 hits per pass.
assert_eq!(
out.metrics.get("batch_hits").and_then(|v| v.as_u64()),
Some(399)
);
assert_eq!(
out.metrics.get("batch_lookups").and_then(|v| v.as_u64()),
Some(399 * 20)
);
assert!(
out.metrics
.get("batch_names_per_sec")
.and_then(|v| v.as_f64())
.unwrap_or(0.0)
> 0.0,
"batch_names_per_sec must be positive: {:?}",
out.metrics
);
// A/B baseline leg: the sequential per-name loop resolves the SAME total
// edges as the batch (independent cross-check of byte-identity) and the
// keep-or-drop speedup ratio is present + positive.
assert_eq!(
out.metrics.get("seq_lookups").and_then(|v| v.as_u64()),
out.metrics.get("batch_lookups").and_then(|v| v.as_u64()),
"seq and batch must resolve identical edge counts: {:?}",
out.metrics
);
assert!(
out.metrics
.get("batch_speedup")
.and_then(|v| v.as_f64())
.unwrap_or(0.0)
> 0.0,
"batch_speedup must be present and positive: {:?}",
out.metrics
);
}
#[test]
fn dep_graph_benchers_are_linked_and_discoverable() {
let ids = crate::bench::api::registered_bench_ids();
assert!(
!ids.is_empty(),
"linked build must register at least one bencher: {ids:?}"
);
assert!(
ids.iter().any(|id| id == "nornir.dep_graph_build"),
"nornir.dep_graph_build must be linked + discoverable: {ids:?}"
);
assert!(
ids.iter().any(|id| id == "nornir.dep_graph_build_ext"),
"nornir.dep_graph_build_ext must be linked + discoverable: {ids:?}"
);
}
/// The bencher ids satisfy the `nornir.dep_graph` hot-path per the gate's
/// matching rule (`id == key` or `<key>_`/`<key>.` prefix), so the gate row
/// resolves covered from the LIVE registry — no allowlist excuse needed.
#[test]
fn dep_graph_hot_path_is_covered_by_registry() {
let registered = crate::bench::api::registered_bench_ids();
let rows = crate::autonom::bench_rows(&["nornir.dep_graph"], ®istered);
let dg = rows
.iter()
.find(|r| r.module == "nornir.dep_graph")
.unwrap();
assert!(
dg.has_bencher,
"dep_graph must be covered by the linked registry: {registered:?}"
);
}
/// The formerly-sentinelled hot-paths — `warehouse_query` and (under
/// `vector`) `vector_search` — are now benched
/// by LINKED registrations whose ids equal the hot-path key, so the
/// completeness gate resolves them covered from the live registry and their
/// `autonom-allow.toml` sentinels are deleted (a stale excuse would red the
/// gate). Mirrors `dep_graph_hot_path_is_covered_by_registry`.
#[test]
fn wired_hot_paths_are_covered_by_registry() {
let registered = crate::bench::api::registered_bench_ids();
let mut keys = vec!["nornir.warehouse_query"];
#[cfg(feature = "vector")]
keys.push("nornir.vector_search");
let rows = crate::autonom::bench_rows(&keys, ®istered);
for key in &keys {
let row = rows.iter().find(|r| &r.module == key).unwrap();
assert!(
row.has_bencher,
"{key} must be covered by the linked registry: {registered:?}"
);
}
}
}
// ╔══════════════════════════════════════════════════════════════════════════╗
// ║ APPENDED REGION — the three unbenched dependency-Mímir verbs (2026-07-23)║
// ╚══════════════════════════════════════════════════════════════════════════╝
//
// `deps_transitive` / `dependents_transitive` were the ONLY closure queries with
// an arm (inside `DepGraphBuild`'s warm query battery). The three verbs an agent
// actually calls to size a change — `affected_by_change`, `build_order`,
// `dep_path` — had **none**, so the blast-radius answers the MCP serves were
// unmeasured and, more to the point, unchecked: nothing asserted that the
// affected set is closed under dependents, that the build order is topological,
// or that a dep path is a real chain of edges.
//
// These arms bench the **served shapers** (`crate::mimir::*`, the exact functions
// the MCP tools call, JSON shaping included) rather than the raw graph methods,
// because the JSON hop annotation is part of what an agent waits for.
//
// Like `DepGraphBuild` above, the FIXTURE build (one `cargo metadata` per repo)
// is untimed setup; only the warm query battery is timed. Same latency-microbench
// character, same reason the core-saturation law does not apply to it.
/// Which dependency-Mímir verb an arm drives.
#[derive(Clone, Copy)]
enum DepVerb {
/// `affected_by_change` — changed repos ∪ everything transitively depending
/// on them, in build order. The re-run set.
AffectedByChange,
/// `build_order` — the whole workspace topologically sorted.
BuildOrder,
/// `dep_path` — shortest annotated route `from` → `to`.
DepPath,
}
impl DepVerb {
fn label(self) -> &'static str {
match self {
DepVerb::AffectedByChange => "affected_by_change",
DepVerb::BuildOrder => "build_order",
DepVerb::DepPath => "dep_path",
}
}
}
/// The blast-radius verbs over a synthetic layered DAG.
///
/// ## Floors — every one an `ensure!` that runs in release, and every one a
/// statement about the ANSWER, not about the clock
///
/// * **`affected_by_change`** — the returned set must contain the changed repo
/// and every one of its transitive dependents computed independently from the
/// generator's ground-truth edge list. A closure that stops one hop short still
/// returns a plausible list and a fine throughput number; this is what makes
/// that red.
/// * **`build_order`** — must be a permutation of the workspace (no repo lost, no
/// repo twice) AND topological: for every ground-truth edge `f → t`, `t` must
/// appear before `f`. A "build order" that is not one is worse than no answer.
/// * **`dep_path`** — the returned path must start at `from`, end at `to`, and
/// every consecutive pair must be a REAL edge. A shortest-path search that
/// returns a path through a non-edge is the failure mode a length check misses.
struct DepMimirVerbBench {
id: &'static str,
repos: usize,
verb: DepVerb,
iters: usize,
}
impl Bencher for DepMimirVerbBench {
fn id(&self) -> &'static str {
self.id
}
fn run(&self) -> Result<BenchResult> {
use std::collections::{BTreeSet, VecDeque};
let spec = GenSpec::new(Shape::LayeredDag, self.repos).seed(42);
let ws = gen_workspace(&spec);
let desc = WorkspaceDescriptor::load(&ws.descriptor_path)?;
let g = crate::warehouse::dep_graph::build(&desc)?;
// Ground truth straight from the generator — never from the graph under
// test, or the floor would be checking the answer against itself.
let edges: Vec<(String, String)> = ws.edges.clone();
let dependents_closure = |seed: &str| -> BTreeSet<String> {
let mut seen: BTreeSet<String> = BTreeSet::new();
let mut q: VecDeque<String> = VecDeque::new();
seen.insert(seed.to_string());
q.push_back(seed.to_string());
while let Some(cur) = q.pop_front() {
for (f, t) in &edges {
if t == &cur && seen.insert(f.clone()) {
q.push_back(f.clone());
}
}
}
seen
};
let mut per_iter = Vec::with_capacity(self.iters * self.repos);
let mut answered = 0u64;
let mut answer_size = 0usize;
let mut total = std::time::Duration::ZERO;
match self.verb {
DepVerb::AffectedByChange => {
for _ in 0..self.iters {
for repo in &ws.repo_names {
let changed = vec![repo.clone()];
let t = std::time::Instant::now();
let v = crate::mimir::affected_by_change(&g, &changed)?;
let el = t.elapsed();
total += el;
per_iter.push(el.as_secs_f64() * 1e6);
answered += 1;
let got: BTreeSet<String> = v["affected"]
.as_array()
.map(|a| {
a.iter()
.filter_map(|x| x.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default();
answer_size += got.len();
let want = dependents_closure(repo);
let missing: Vec<&String> =
want.iter().filter(|r| !got.contains(*r)).collect();
anyhow::ensure!(
missing.is_empty(),
"affected_by_change({repo}) lost {} transitive dependent(s) {:?} — \
the re-run set is NOT closed, so a real consumer would be skipped",
missing.len(),
missing
);
}
}
}
DepVerb::BuildOrder => {
let all: BTreeSet<String> = ws.repo_names.iter().cloned().collect();
for _ in 0..self.iters {
let t = std::time::Instant::now();
let v = crate::mimir::build_order(&g)?;
let el = t.elapsed();
total += el;
per_iter.push(el.as_secs_f64() * 1e6);
answered += 1;
let order: Vec<String> = v["build_order"]
.as_array()
.map(|a| {
a.iter()
.filter_map(|x| x.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default();
answer_size = order.len();
let as_set: BTreeSet<String> = order.iter().cloned().collect();
anyhow::ensure!(
as_set == all && order.len() == all.len(),
"build_order returned {} entries over {} distinct repos for a \
{}-repo workspace — an order that drops or duplicates a member is \
not an order",
order.len(),
as_set.len(),
all.len()
);
let pos: std::collections::HashMap<&str, usize> = order
.iter()
.enumerate()
.map(|(i, r)| (r.as_str(), i))
.collect();
for (f, t2) in &edges {
anyhow::ensure!(
pos[t2.as_str()] < pos[f.as_str()],
"build_order puts dependency {t2} AFTER its dependent {f} — \
building in this order cannot work"
);
}
}
}
DepVerb::DepPath => {
// Every ordered pair (a, c) that the ground truth says is
// reachable in ≥ 2 hops, so the battery exercises real BFS and
// not just a direct-edge lookup.
let mut pairs: Vec<(String, String)> = Vec::new();
for (a, b) in &edges {
for (b2, c) in &edges {
if b == b2 && a != c && !edges.iter().any(|(f, t)| f == a && t == c) {
pairs.push((a.clone(), c.clone()));
}
}
}
pairs.sort();
pairs.dedup();
anyhow::ensure!(
!pairs.is_empty(),
"the {}-repo layered DAG produced no 2-hop pair — the fixture would \
bench a direct-edge lookup and call it a path search",
self.repos
);
let edge_set: BTreeSet<(String, String)> = edges.iter().cloned().collect();
for _ in 0..self.iters {
for (from, to) in &pairs {
let t = std::time::Instant::now();
let v = crate::mimir::dep_path(&g, from, to)?;
let el = t.elapsed();
total += el;
per_iter.push(el.as_secs_f64() * 1e6);
answered += 1;
let path: Vec<String> = v["path"]
.as_array()
.map(|a| {
a.iter()
.filter_map(|x| x.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default();
answer_size += path.len();
anyhow::ensure!(
path.first().map(String::as_str) == Some(from.as_str())
&& path.last().map(String::as_str) == Some(to.as_str()),
"dep_path({from} → {to}) returned {path:?} — the ground truth says \
a route exists, so a missing or misanchored path is a lost answer"
);
for w in path.windows(2) {
anyhow::ensure!(
edge_set.contains(&(w[0].clone(), w[1].clone())),
"dep_path({from} → {to}) walks {} → {} which is NOT an edge — \
a path through a non-edge is a fabricated answer",
w[0],
w[1]
);
}
}
}
}
}
let mut metrics = serde_json::Map::new();
metrics.insert("verb".into(), serde_json::json!(self.verb.label()));
metrics.insert("crates".into(), serde_json::json!(ws.repo_names.len()));
metrics.insert("edges".into(), serde_json::json!(edges.len()));
metrics.insert("answer_entries".into(), serde_json::json!(answer_size));
let (p50, p99) = {
let mut xs = per_iter.clone();
xs.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let at = |q: f64| -> f64 {
if xs.is_empty() {
return 0.0;
}
let i = (((xs.len() - 1) as f64) * q).round() as usize;
xs[i.min(xs.len() - 1)]
};
(at(0.50), at(0.99))
};
metrics.insert(
"queries_per_sec".into(),
serde_json::json!(answered as f64 / total.as_secs_f64().max(1e-9)),
);
metrics.insert("queries".into(), serde_json::json!(answered));
metrics.insert("p50_us".into(), serde_json::json!(p50));
metrics.insert("p99_us".into(), serde_json::json!(p99));
metrics.insert(
"elapsed_ms".into(),
serde_json::json!(total.as_secs_f64() * 1000.0),
);
Ok(BenchResult {
name: self.id.into(),
metrics,
..Default::default()
})
}
}
// 30 repos, not 50: the fixture cost is one `cargo metadata` subprocess per repo
// and these are three more arms paying it. The query battery still covers EVERY
// repo (and, for `dep_path`, every 2-hop pair), so the answers are exercised
// exhaustively even though the graph is smaller than `dep_graph_build`'s.
crate::register_bench!(DepMimirVerbBench {
id: "nornir.mimir.deps.affected_by_change",
repos: 30,
verb: DepVerb::AffectedByChange,
iters: 4
});
crate::register_bench!(DepMimirVerbBench {
id: "nornir.mimir.deps.build_order",
repos: 30,
verb: DepVerb::BuildOrder,
iters: 64
});
crate::register_bench!(DepMimirVerbBench {
id: "nornir.mimir.deps.dep_path",
repos: 30,
verb: DepVerb::DepPath,
iters: 2
});
#[cfg(test)]
mod dep_verb_arm_tests {
use super::*;
use crate::bench::api::Bencher;
/// RED-WHEN-BROKEN: the three blast-radius arms run, and their answer floors
/// (closure completeness, topological order, real-edge paths) hold against a
/// small real workspace. Small `repos` so the `cargo metadata` fixture stays
/// cheap in `cargo test`; the assertions are about correctness, not speed.
#[test]
fn dep_mimir_verb_arms_hold_their_answer_floors() {
for verb in [
DepVerb::AffectedByChange,
DepVerb::BuildOrder,
DepVerb::DepPath,
] {
let out = DepMimirVerbBench {
id: "nornir.mimir.deps.test",
repos: 8,
verb,
iters: 1,
}
.run()
.unwrap_or_else(|e| panic!("{} arm failed its floor: {e}", verb.label()));
assert!(
out.metrics
.get("queries_per_sec")
.and_then(|v| v.as_f64())
.unwrap_or(0.0)
> 0.0,
"{} must report a throughput",
verb.label()
);
}
}
/// The three verbs are LINKED into the registry the completeness gate sweeps.
#[test]
fn dep_mimir_verb_arms_are_registered() {
let ids = crate::bench::api::registered_bench_ids();
for want in [
"nornir.mimir.deps.affected_by_change",
"nornir.mimir.deps.build_order",
"nornir.mimir.deps.dep_path",
] {
assert!(ids.iter().any(|id| id == want), "{want} must be registered");
}
}
}