//! nornir Bencher harness for **funnel-core** — the funnel projection hot paths
//! the nornir funnel UI runs on every render, historized in the warehouse + gated
//! on regression by `nornir bench run funnel-core`.
//!
//! What's benched (the unguarded perf surface):
//! - `funnel-core.board_project` — `cards_from_funnel` + `board` over a big funnel
//! (the kanban board the viz paints).
//! - `funnel-core.smart_query` — `query_cards` filtering the whole card set.
//! - `funnel-core.plan_dag_whole` — `PlanDag::from_funnel(None)` over MANY plans
//! (a different shape from nornir's `plan_dag_assemble`, which is one long chain).
//!
//! The Plan-DAG single-chain assembly already has `nornir.plan_dag_assemble`; this
//! harness covers the board/query projection + a wide multi-plan Plan-DAG, which
//! were previously unbenched.
//!
//! Run: `nornir bench run funnel-core` (or `cargo run --release --example nornir-bench`).
use anyhow::{ensure, Result};
use nornir::bench::api::Bencher;
use nornir::bench::BenchResult;
use funnel_core::{
board, cards_from_funnel, query_cards, DagGrowth, Event, Funnel, ItemKind, IdeaId, NodeId,
PlanDag, PlanId, PlanStatus,
};
/// Headline scale: a funnel of `PLANS` decomposed ideas, each a plan of
/// `NODES_PER_PLAN` task nodes in a dependency chain, plus the mother idea cards.
/// Total cards ≈ PLANS * (1 + NODES_PER_PLAN).
const PLANS_HEAVY: u64 = 400;
const NODES_PER_PLAN: u64 = 12;
/// PREVIEW mode (`NORNIR_BENCH_PREVIEW=1`) — the fast verification loop.
///
/// Every arm still runs, through the SAME code paths, emitting the SAME metric keys;
/// only the scale (`PLANS`, `GROWTH_NODES`, iteration counts) shrinks, so a broken arm
/// still shows up as a failure in seconds instead of minutes. The emitted run is
/// stamped `<host>-preview` by `nornir_bench::api` — a preview number can therefore
/// never land in, or overwrite, the real host's README row.
fn preview() -> bool {
nornir::bench::api::preview_run()
}
/// Scale knob: the heavy value, or a small one under `NORNIR_BENCH_PREVIEW`.
fn scaled(heavy: u64, light: u64) -> u64 {
if preview() { light } else { heavy }
}
/// Number of plans in the synthetic funnel (heavy 400 / preview 20).
fn plans() -> u64 {
scaled(PLANS_HEAVY, 20)
}
/// Build the synthetic funnel once (shared by every bencher's `run`). Entirely
/// through the public event/apply seam, so it exercises the real fold path too.
fn big_funnel() -> Funnel {
let mut f = Funnel::default();
let now = chrono::Utc::now();
let mut node_seq = 0u64;
for p in 1..=plans() {
let idea = IdeaId::seq(p);
let plan = PlanId::seq(p);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: if p % 3 == 0 { "agent:autonom" } else { "human:rickard" }.into(),
text: format!("idea {p}: build feature #{p} with #tag{}", p % 7),
refs: vec![],
item_kind: if p % 5 == 0 { ItemKind::Error } else { ItemKind::Idea },
ts: now,
})
.expect("idea");
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: format!("decompose plan {p}"),
planner: "decompose".into(),
ts: now,
})
.expect("plan");
f.apply(&Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: now,
})
.expect("active");
let mut prev: Option<u64> = None;
for _ in 0..NODES_PER_PLAN {
node_seq += 1;
let nid = NodeId::seq(node_seq);
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: nid.clone(),
kind: if node_seq % 4 == 0 { "test:write" } else { "code:write" }.into(),
params: Default::default(),
targets: vec![format!("src/mod{node_seq}.rs")],
prompt_excerpt: Some(format!("task {node_seq}: implement thing #{node_seq}")),
ts: now,
})
.expect("node");
if let Some(pv) = prev {
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(pv),
to_node: nid.clone(),
ts: now,
})
.expect("edge");
}
prev = Some(node_seq);
}
}
f
}
/// Length of the single-plan chain the growth-timeline bench replays. One idea +
/// one plan + one activate + `GROWTH_NODES` NodeAdded + `GROWTH_NODES - 1` EdgeAdded.
const GROWTH_NODES_HEAVY: u64 = 300;
/// Chain length actually replayed (heavy 300 / preview 30).
fn growth_nodes() -> u64 {
scaled(GROWTH_NODES_HEAVY, 30)
}
/// The event log for one idea decomposed into a `GROWTH_NODES`-long dependency
/// chain, in write order — the realistic "watch this plan's DAG grow" animation the
/// live funnel pane replays. Returned as a `Vec<Event>` (not a folded `Funnel`)
/// because [`DagGrowth::from_events`] consumes the raw log.
fn growth_event_log() -> Vec<Event> {
let now = chrono::Utc::now();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
let mut evs: Vec<Event> = vec![
Event::IdeaSubmitted {
id: idea.clone(),
source: "human:rickard".into(),
text: "grow one big plan".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
},
Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "decompose into a long chain".into(),
planner: "decompose".into(),
ts: now,
},
Event::PlanStatusChanged { plan_id: plan.clone(), status: PlanStatus::Active, why: None, ts: now },
];
for i in 1..=growth_nodes() {
evs.push(Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: if i % 4 == 0 { "test:write" } else { "code:write" }.into(),
params: Default::default(),
targets: vec![format!("src/mod{i}.rs")],
prompt_excerpt: Some(format!("task {i}")),
ts: now,
});
if i > 1 {
evs.push(Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(i - 1),
to_node: NodeId::seq(i),
ts: now,
});
}
}
evs
}
fn metrics(pairs: &[(&str, serde_json::Value)]) -> serde_json::Map<String, serde_json::Value> {
pairs.iter().map(|(k, v)| (k.to_string(), v.clone())).collect()
}
/// Project the whole funnel into cards + group into the kanban board, repeatedly.
struct BoardProject;
impl Bencher for BoardProject {
fn id(&self) -> &'static str {
"funnel-core.board_project"
}
fn run(&self) -> Result<BenchResult> {
let f = big_funnel();
let iters = scaled(200, 5);
let t = std::time::Instant::now();
let mut cards_out = 0usize;
let mut total = 0usize;
for _ in 0..iters {
let cards = cards_from_funnel(&f);
cards_out = cards.len();
let b = board(&cards);
total = b.total;
}
let s = t.elapsed().as_secs_f64();
ensure!(cards_out > 0 && total == cards_out, "board projected nothing");
let per_sec = (cards_out as u64 * iters) as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("cards", cards_out.into()),
("cards_per_sec", per_sec.round().into()),
("iters", iters.into()),
("elapsed_ms", ((s * 1000.0).round()).into()),
]),
))
}
}
/// Filter the whole card set with a representative smart query, repeatedly.
struct SmartQuery;
impl Bencher for SmartQuery {
fn id(&self) -> &'static str {
"funnel-core.smart_query"
}
fn run(&self) -> Result<BenchResult> {
let f = big_funnel();
let cards = cards_from_funnel(&f);
let q = "type:feature is:open for:rickard sort:newest";
let iters = scaled(500, 10);
let t = std::time::Instant::now();
let mut hits = 0usize;
for _ in 0..iters {
hits = query_cards(&cards, q).len();
}
let s = t.elapsed().as_secs_f64();
ensure!(hits > 0, "query matched nothing");
let scanned_per_sec = (cards.len() as u64 * iters) as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("cards_scanned", cards.len().into()),
("hits", hits.into()),
("cards_scanned_per_sec", scanned_per_sec.round().into()),
("iters", iters.into()),
("elapsed_ms", ((s * 1000.0).round()).into()),
]),
))
}
}
/// Assemble the WHOLE-funnel Plan-DAG (many plans, mother+dep edges), repeatedly.
struct PlanDagWhole;
impl Bencher for PlanDagWhole {
fn id(&self) -> &'static str {
"funnel-core.plan_dag_whole"
}
fn run(&self) -> Result<BenchResult> {
let f = big_funnel();
let iters = scaled(100, 5);
let t = std::time::Instant::now();
let mut nodes = 0usize;
let mut edges = 0usize;
for _ in 0..iters {
let dag = PlanDag::from_funnel(&f, None);
nodes = dag.node_count();
edges = dag.edge_count();
}
let s = t.elapsed().as_secs_f64();
ensure!(nodes > 0 && edges > 0, "plan-dag assembled nothing");
let assemblies_per_sec = iters as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("nodes", nodes.into()),
("edges", edges.into()),
("assemblies_per_sec", assemblies_per_sec.round().into()),
("assemble_ms", (s * 1000.0 / iters as f64).into()),
("iters", iters.into()),
("elapsed_ms", ((s * 1000.0).round()).into()),
]),
))
}
}
/// Assemble the whole-funnel Plan-DAG once, then repeatedly extract its critical
/// path + per-node depth layering (the schedule-defining longest chain the viz
/// highlights + the layout layers it lays nodes out by).
struct PlanDagCritical;
impl Bencher for PlanDagCritical {
fn id(&self) -> &'static str {
"funnel-core.plan_dag_critical"
}
fn run(&self) -> Result<BenchResult> {
let f = big_funnel();
let dag = PlanDag::from_funnel(&f, None);
let iters = scaled(200, 5);
let t = std::time::Instant::now();
let mut cp_len = 0usize;
let mut depth_nodes = 0usize;
for _ in 0..iters {
cp_len = dag.critical_path_len();
depth_nodes = dag.depths().len();
}
let s = t.elapsed().as_secs_f64();
ensure!(cp_len > 0 && depth_nodes > 0, "critical path / depths empty");
let passes_per_sec = iters as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("nodes", dag.node_count().into()),
("edges", dag.edge_count().into()),
("critical_path_len", cp_len.into()),
("depth_nodes", depth_nodes.into()),
("passes_per_sec", passes_per_sec.round().into()),
("pass_ms", (s * 1000.0 / iters as f64).into()),
("iters", iters.into()),
("elapsed_ms", ((s * 1000.0).round()).into()),
]),
))
}
}
/// Fold the whole synthetic event log into a `Funnel` repeatedly — the write path
/// `Funnel::apply` runs on every event (validation + mutation), including the
/// cycle-guard reachability check each `EdgeAdded` now performs. Tracks that the
/// DAG-invariant enforcement doesn't regress the fold throughput.
struct FunnelBuild;
impl Bencher for FunnelBuild {
fn id(&self) -> &'static str {
"funnel-core.funnel_build"
}
fn run(&self) -> Result<BenchResult> {
// 26 events/plan (1 idea + 1 plan + 1 status + 12 nodes + 11 edges).
let events_per_iter = plans() * (3 + NODES_PER_PLAN + (NODES_PER_PLAN - 1));
let iters = scaled(20, 3);
let t = std::time::Instant::now();
let mut plans = 0usize;
let mut edges = 0usize;
for _ in 0..iters {
let f = big_funnel();
plans = f.plans.len();
edges = f.plans.values().map(|p| p.edges.len()).sum();
}
let s = t.elapsed().as_secs_f64();
ensure!(plans as u64 == self::plans() && edges > 0, "funnel folded nothing");
let events_per_sec = (events_per_iter * iters) as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("plans", plans.into()),
("edges", edges.into()),
("events_per_iter", events_per_iter.into()),
("events_per_sec", events_per_sec.round().into()),
("build_ms", (s * 1000.0 / iters as f64).into()),
("iters", iters.into()),
("elapsed_ms", ((s * 1000.0).round()).into()),
]),
))
}
}
/// Replay a single-plan event log into a birth-stamped growth timeline repeatedly —
/// the `DagGrowth::from_events` path the live funnel pane runs to animate the DAG
/// growing. A genuinely distinct (heavier) shape from the snapshot projections: one
/// `Funnel::apply` + `PlanDag::from_funnel` per event step, so it scales with the
/// log length, not one still frame.
struct PlanDagGrowth;
impl Bencher for PlanDagGrowth {
fn id(&self) -> &'static str {
"funnel-core.plan_dag_growth"
}
fn run(&self) -> Result<BenchResult> {
let evs = growth_event_log();
let iters = scaled(50, 3);
let t = std::time::Instant::now();
let mut node_births = 0usize;
let mut edge_births = 0usize;
let mut steps = 0usize;
for _ in 0..iters {
let g = DagGrowth::from_events(&evs, None);
node_births = g.nodes.len();
edge_births = g.edges.len();
steps = g.steps;
}
let s = t.elapsed().as_secs_f64();
ensure!(node_births > 0 && edge_births > 0, "growth timeline recorded no births");
let events_per_sec = (steps as u64 * iters) as f64 / s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("steps", steps.into()),
("node_births", node_births.into()),
("edge_births", edge_births.into()),
("events_per_sec", events_per_sec.round().into()),
("timeline_ms", (s * 1000.0 / iters as f64).into()),
("iters", iters.into()),
("elapsed_ms", ((s * 1000.0).round()).into()),
]),
))
}
}
/// The **live-pane** comparison: incremental `DagGrowth::extend` (one event per tick,
/// keeping a running accumulator) vs the naive way a pane would use the batch API —
/// calling `DagGrowth::from_events` on the whole log again on every tick. Streaming N
/// ticks is O(N) `from_funnel` frames; re-replaying the whole log each tick is
/// O(N^2). Same log as `plan_dag_growth`; reports both wall times + the speedup so a
/// regression in the incremental path (or a lost win) is visible.
struct PlanDagGrowthStream;
impl Bencher for PlanDagGrowthStream {
fn id(&self) -> &'static str {
"funnel-core.plan_dag_growth_stream"
}
fn run(&self) -> Result<BenchResult> {
let evs = growth_event_log();
let steps = evs.len();
// Incremental: open one stream, push each event once (the per-tick path).
let stream_iters = scaled(50, 3);
let t = std::time::Instant::now();
let mut node_births = 0usize;
let mut edge_births = 0usize;
for _ in 0..stream_iters {
let mut g = DagGrowth::new(None);
for ev in &evs {
g.extend(ev);
}
node_births = g.nodes.len();
edge_births = g.edges.len();
}
let stream_s = t.elapsed().as_secs_f64();
let stream_ms = stream_s * 1000.0 / stream_iters as f64;
// Naive live use of the batch API: re-replay evs[..=t] on every tick t. One
// full session (O(steps^2)); timed once as the baseline the stream beats.
let t = std::time::Instant::now();
let mut replay_node_births = 0usize;
for tick in 0..steps {
let g = DagGrowth::from_events(&evs[..=tick], None);
replay_node_births = g.nodes.len();
}
let replay_s = t.elapsed().as_secs_f64();
let replay_ms = replay_s * 1000.0;
ensure!(node_births > 0 && edge_births > 0, "streamed growth recorded no births");
ensure!(
replay_node_births == node_births,
"naive re-replay disagrees with the stream on the final node count"
);
let speedup = if stream_ms > 0.0 { replay_ms / stream_ms } else { 0.0 };
let events_per_sec = (steps as u64 * stream_iters) as f64 / stream_s;
Ok(BenchResult::measured(
self.id(),
metrics(&[
("steps", steps.into()),
("node_births", node_births.into()),
("edge_births", edge_births.into()),
("stream_ms", stream_ms.into()),
("replay_ms", replay_ms.into()),
("speedup_x", ((speedup * 100.0).round() / 100.0).into()),
("events_per_sec", events_per_sec.round().into()),
("iters", stream_iters.into()),
]),
))
}
}
nornir::register_bench!(BoardProject);
nornir::register_bench!(SmartQuery);
nornir::register_bench!(PlanDagWhole);
nornir::register_bench!(PlanDagCritical);
nornir::register_bench!(FunnelBuild);
nornir::register_bench!(PlanDagGrowth);
nornir::register_bench!(PlanDagGrowthStream);
fn main() -> Result<()> {
nornir::bench::api::run_main_json()
}