feat(bench): surface framework + engine/ingest throughput surfaces

Tasks 3-4 of the bench-harness plan (refs #251). The surface framework
(1 discarded warmup + N measured reps, median per metric, cross-rep
fingerprint identity enforced — a surface disagreeing with itself is
nondeterminism, never averaged away) and the two library surfaces:

- engine_throughput: 1M-bar seeded walk through an SMA-cross bias graph
  (GraphBuilder -> compile -> bootstrap -> VecSource -> run), fingerprint =
  the SeriesReducer finalize row.
- ingest_throughput: synthetic 48-byte M1 archive drained via open_columns
  with a FRESH DataServer per rep, so the measured drain pays zip inflate +
  parse instead of the in-memory FileCache; fingerprint = count + bit-XOR
  of the close column.

Quick-mode determinism is pinned in-suite for both surfaces (C1 — the
harness's premise as currently-green tests); full-size numbers wait for the
release driver glue.

crates/aura-bench/src/main.rs

@@ -4,6 +4,7 @@
 //! bench instead of pinning a bogus baseline.
 
 mod baseline;
+mod surfaces;
 mod synth;
 
 fn main() {

crates/aura-bench/src/surfaces/engine.rs

@@ -0,0 +1,70 @@
+//! Engine throughput: bars/s through a bootstrapped harness — an SMA-cross
+//! bias graph fed by the seeded synthetic walk via `VecSource`. Fingerprint:
+//! the `SeriesReducer` finalize row `[last, max_drawdown, sign_flips]`.
+
+use super::{RepOutcome, Sizing};
+use crate::synth::synthetic_walk;
+use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
+use aura_engine::{GraphBuilder, Harness, Source, VecSource};
+use aura_std::{SeriesReducer, Sma, Sub};
+use std::collections::BTreeMap;
+use std::sync::mpsc;
+use std::time::Instant;
+
+pub const FULL_BARS: usize = 1_000_000;
+pub const QUICK_BARS: usize = 2_000;
+
+pub fn rep(sizing: Sizing) -> Result<RepOutcome, String> {
+    let n_bars = if sizing.quick { QUICK_BARS } else { FULL_BARS };
+    let (tx, rx) = mpsc::channel::<(Timestamp, Vec<Scalar>)>();
+    let mut g = GraphBuilder::new("bench_sma_cross");
+    let fast = g.add(Sma::builder().bind("length", Scalar::i64(12)));
+    let slow = g.add(Sma::builder().bind("length", Scalar::i64(48)));
+    let bias = g.add(Sub::builder());
+    let sink = g.add(SeriesReducer::builder(Firing::Any, tx));
+    let price = g.source_role("price", ScalarKind::F64);
+    g.feed(price, [fast.input("series"), slow.input("series")]);
+    g.connect(fast.output("value"), bias.input("lhs"));
+    g.connect(slow.output("value"), bias.input("rhs"));
+    g.connect(bias.output("value"), sink.input("col[0]"));
+    let flat = g
+        .build()
+        .map_err(|e| format!("bench graph must wire: {e:?}"))?
+        .compile_with_params(&[])
+        .map_err(|e| format!("bench graph must compile: {e:?}"))?;
+    let mut h = Harness::bootstrap(flat).map_err(|e| format!("bench graph must bootstrap: {e:?}"))?;
+    let prices = synthetic_walk(n_bars);
+    let src: Vec<Box<dyn Source>> = vec![Box::new(VecSource::new(prices))];
+    let t = Instant::now();
+    h.run(src);
+    let wall = t.elapsed().as_secs_f64();
+    let (_, row) = rx.try_iter().last().ok_or("SeriesReducer must emit its finalize row")?;
+    let fingerprint = format!(
+        "last={:.6} max_dd={:.6} flips={} n={}",
+        row[0].as_f64(),
+        row[1].as_f64(),
+        row[2].as_i64(),
+        n_bars
+    );
+    let mut metrics = BTreeMap::new();
+    metrics.insert("wall_s".to_string(), wall);
+    metrics.insert("bars_per_s".to_string(), n_bars as f64 / wall);
+    Ok(RepOutcome { metrics, fingerprint })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The determinism smoke for the engine surface: two quick reps in-process
+    /// must produce byte-identical fingerprints (C1 — this is the harness's
+    /// entire premise, pinned as a currently-green test).
+    #[test]
+    fn engine_surface_fingerprint_is_deterministic_in_quick_mode() {
+        let a = rep(Sizing { quick: true }).expect("quick rep runs");
+        let b = rep(Sizing { quick: true }).expect("quick rep runs");
+        assert_eq!(a.fingerprint, b.fingerprint);
+        assert!(a.fingerprint.starts_with("last="), "shape: {}", a.fingerprint);
+        assert!(a.metrics.contains_key("bars_per_s"));
+    }
+}

crates/aura-bench/src/surfaces/ingest.rs

@@ -0,0 +1,67 @@
+//! Ingest throughput: bars/s draining one field column out of a synthetic
+//! 48-byte M1 zip archive through the public ingest window API. A FRESH
+//! `DataServer` per repetition, so the measured drain pays zip inflate +
+//! record parse instead of the server's in-memory cache; the discarded warmup
+//! rep leaves the OS page cache warm and consistent.
+
+use super::{RepOutcome, Sizing};
+use crate::synth::{write_symbol_archive, ScratchDir};
+use aura_ingest::{open_columns, DataServer, M1Field};
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+pub const SYMBOL: &str = "BENCHI";
+pub const FULL_MONTHS: ((i32, u32), (i32, u32)) = ((2024, 1), (2025, 12));
+pub const QUICK_MONTHS: ((i32, u32), (i32, u32)) = ((2024, 1), (2024, 2));
+
+/// The archive is written once per surface run (setup, unmeasured) and shared
+/// by every repetition.
+pub struct IngestFixture {
+    pub dir: ScratchDir,
+}
+
+pub fn setup(sizing: Sizing) -> Result<IngestFixture, String> {
+    let dir = ScratchDir::new("ingest").map_err(|e| format!("scratch dir: {e}"))?;
+    let (from, to) = if sizing.quick { QUICK_MONTHS } else { FULL_MONTHS };
+    write_symbol_archive(&dir.0, SYMBOL, from, to);
+    Ok(IngestFixture { dir })
+}
+
+pub fn rep(fixture: &IngestFixture) -> Result<RepOutcome, String> {
+    let server = Arc::new(DataServer::new(&fixture.dir.0));
+    let mut srcs = open_columns(&server, SYMBOL, None, None, &[M1Field::Close])
+        .ok_or("bench archive must be present for the close column")?;
+    let mut src = srcs.pop().ok_or("one source for one field")?;
+    let t = Instant::now();
+    let mut n: u64 = 0;
+    let mut xor: u64 = 0;
+    while let Some((_, v)) = src.next() {
+        n += 1;
+        xor ^= v.as_f64().to_bits();
+    }
+    let wall = t.elapsed().as_secs_f64();
+    if n == 0 {
+        return Err("the synthetic archive must yield bars".to_string());
+    }
+    let mut metrics = BTreeMap::new();
+    metrics.insert("wall_s".to_string(), wall);
+    metrics.insert("bars_per_s".to_string(), n as f64 / wall);
+    Ok(RepOutcome { metrics, fingerprint: format!("n={n} xor={xor:016x}") })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Determinism smoke: two drains of the same archive fold to the same
+    /// fingerprint, and a fresh server per rep really re-reads the bytes.
+    #[test]
+    fn ingest_surface_fingerprint_is_deterministic_in_quick_mode() {
+        let fixture = setup(Sizing { quick: true }).expect("fixture writes");
+        let a = rep(&fixture).expect("first drain");
+        let b = rep(&fixture).expect("second drain");
+        assert_eq!(a.fingerprint, b.fingerprint);
+        assert!(a.fingerprint.starts_with("n="), "shape: {}", a.fingerprint);
+    }
+}

crates/aura-bench/src/surfaces/mod.rs

@@ -0,0 +1,141 @@
+//! The surface framework: a surface is a closure the harness calls once per
+//! repetition; `run_reps` applies the measurement discipline (1 discarded
+//! warmup + N measured reps, median per metric) and enforces cross-rep
+//! fingerprint identity — a surface disagreeing with itself is nondeterminism,
+//! reported as an infra error, never averaged away.
+//!
+//! The CLI glue that wires this framework's public surface into a running
+//! binary lands in Task 7; until then nothing outside `#[cfg(test)]` calls
+//! into it, so the module is exempted from the dead-code lint here rather
+//! than left to trip the workspace's `-D warnings` clippy gate.
+#![allow(dead_code)]
+
+pub mod engine;
+pub mod ingest;
+
+use crate::baseline::Measured;
+use std::collections::BTreeMap;
+
+/// Workload sizing: `quick` is the test/dev mode (tiny inputs, mechanics
+/// only); the committed baselines are always full-size.
+#[derive(Clone, Copy, Debug)]
+pub struct Sizing {
+    pub quick: bool,
+}
+
+/// One repetition's outcome.
+pub struct RepOutcome {
+    pub metrics: BTreeMap<String, f64>,
+    pub fingerprint: String,
+}
+
+pub fn median(mut xs: Vec<f64>) -> f64 {
+    xs.sort_by(|a, b| a.partial_cmp(b).expect("no NaN metrics"));
+    xs[xs.len() / 2]
+}
+
+/// 1 discarded warmup + `reps` measured runs; median per metric key;
+/// cross-rep fingerprint identity enforced.
+pub fn run_reps(
+    surface: &str,
+    reps: u32,
+    mut rep: impl FnMut() -> Result<RepOutcome, String>,
+) -> Result<Measured, String> {
+    let reps = reps.max(1);
+    let _warmup = rep()?; // page/alloc warmup, discarded
+    let mut runs = Vec::new();
+    for _ in 0..reps {
+        runs.push(rep()?);
+    }
+    let fingerprint = runs[0].fingerprint.clone();
+    for r in &runs[1..] {
+        if r.fingerprint != fingerprint {
+            return Err(format!(
+                "surface {surface} is nondeterministic across repetitions: {fingerprint:?} vs {:?}",
+                r.fingerprint
+            ));
+        }
+    }
+    let mut metrics = BTreeMap::new();
+    for key in runs[0].metrics.keys() {
+        let vals: Vec<f64> = runs.iter().filter_map(|r| r.metrics.get(key).copied()).collect();
+        metrics.insert(key.clone(), median(vals));
+    }
+    Ok(Measured { surface: surface.to_string(), metrics, fingerprint })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn median_picks_the_middle() {
+        assert_eq!(median(vec![3.0, 1.0, 2.0]), 2.0);
+        assert_eq!(median(vec![1.0, 2.0]), 2.0);
+    }
+
+    /// Task 3's actual deliverable: `run_reps` composed with the real engine
+    /// surface (a re-bootstrapped SMA-cross harness per call), not a mock
+    /// closure. Protects the wiring end to end — a warmup plus three fresh
+    /// bootstraps must agree on the determinism fingerprint and the returned
+    /// metrics map must carry both keys a `BaselineDoc` for this surface
+    /// needs (`wall_s`, `bars_per_s`), positive and finite.
+    #[test]
+    fn run_reps_drives_the_real_engine_surface_deterministically() {
+        let measured = run_reps("engine", 3, || engine::rep(Sizing { quick: true }))
+            .expect("the real engine surface must agree with itself across repeated bootstraps");
+        assert_eq!(measured.surface, "engine");
+        let bars_per_s = *measured.metrics.get("bars_per_s").expect("bars_per_s metric present");
+        assert!(bars_per_s.is_finite() && bars_per_s > 0.0, "bars_per_s: {bars_per_s}");
+        assert!(measured.metrics.contains_key("wall_s"));
+    }
+
+    /// Task 4's actual deliverable: `run_reps` composed with the real ingest
+    /// surface (a fresh `DataServer` re-opened and re-drained per call).
+    /// Protects that a warmup plus three fresh drains of the same synthetic
+    /// archive agree on the fingerprint, and that the metrics map carries
+    /// the schema the ingest `BaselineDoc` needs.
+    #[test]
+    fn run_reps_drives_the_real_ingest_surface_deterministically() {
+        let fixture = ingest::setup(Sizing { quick: true }).expect("ingest fixture must write");
+        let measured = run_reps("ingest", 3, || ingest::rep(&fixture))
+            .expect("the real ingest surface must agree with itself across repeated drains");
+        assert_eq!(measured.surface, "ingest");
+        let bars_per_s = *measured.metrics.get("bars_per_s").expect("bars_per_s metric present");
+        assert!(bars_per_s.is_finite() && bars_per_s > 0.0, "bars_per_s: {bars_per_s}");
+        assert!(measured.metrics.contains_key("wall_s"));
+    }
+
+    #[test]
+    fn run_reps_rejects_cross_rep_fingerprint_drift() {
+        let mut n = 0;
+        let r = run_reps("t", 2, move || {
+            n += 1;
+            Ok(RepOutcome { metrics: BTreeMap::new(), fingerprint: format!("fp{n}") })
+        });
+        assert!(r.is_err(), "differing fingerprints must be an error");
+    }
+
+    /// The measurement discipline itself: the first (warmup) call is never
+    /// mixed into the reported metric — its outlier value would otherwise
+    /// skew the median — and the median is computed independently per
+    /// metric key across only the measured reps.
+    #[test]
+    fn run_reps_discards_warmup_and_medians_only_measured_reps() {
+        let mut n = 0;
+        let measured = run_reps("t", 3, move || {
+            n += 1;
+            // Call 1 is the discarded warmup: an outlier that must never
+            // enter the median. Calls 2..=4 (the measured reps) report
+            // 1.0, 2.0, 3.0, whose median is 2.0.
+            let v = if n == 1 { 1_000_000.0 } else { (n - 1) as f64 };
+            let mut metrics = BTreeMap::new();
+            metrics.insert("m".to_string(), v);
+            Ok(RepOutcome { metrics, fingerprint: "fp".to_string() })
+        })
+        .expect("agreeing fingerprints must succeed");
+        assert_eq!(measured.surface, "t");
+        assert_eq!(measured.metrics.get("m"), Some(&2.0), "warmup outlier must not pollute the median");
+        assert_eq!(measured.fingerprint, "fp");
+    }
+}