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

//! 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. `passes` fresh drains per repetition lift the measured
/// wall out of the millisecond regime, where ordinary jitter would reach the
/// 10% NOTICE threshold and flap the report.
pub struct IngestFixture {
    pub dir: ScratchDir,
    pub passes: u32,
}

pub const FULL_PASSES: u32 = 20;

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, passes: if sizing.quick { 1 } else { FULL_PASSES } })
}

/// One measured drain: a FRESH `DataServer`, so the pass pays zip inflate +
/// parse instead of hitting the in-memory FileCache.
fn drain_once(dir: &std::path::Path) -> Result<(u64, u64), String> {
    let server = Arc::new(DataServer::new(dir));
    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 mut n: u64 = 0;
    let mut xor: u64 = 0;
    while let Some((_, v)) = src.next() {
        n += 1;
        xor ^= v.as_f64().to_bits();
    }
    if n == 0 {
        return Err("the synthetic archive must yield bars".to_string());
    }
    Ok((n, xor))
}

pub fn rep(fixture: &IngestFixture) -> Result<RepOutcome, String> {
    let t = Instant::now();
    let (n, xor) = drain_once(&fixture.dir.0)?;
    for _ in 1..fixture.passes {
        let again = drain_once(&fixture.dir.0)?;
        if again != (n, xor) {
            return Err(format!("ingest passes disagree: {:?} vs {:?}", (n, xor), again));
        }
    }
    let wall = t.elapsed().as_secs_f64();
    let total = n * u64::from(fixture.passes);
    let mut metrics = BTreeMap::new();
    metrics.insert("wall_s".to_string(), wall);
    metrics.insert("bars_per_s".to_string(), total 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);
    }
}