Aura/crates/aura-runner/src/tap_recorder.rs

//! `TapRecorder` — the `record` fold-registry entry's consumer (C8 sink) for
//! one declared tap: holds the streaming trace writer **in-graph** and
//! appends `(ctx.now(), cell)` per fired warm cycle — 16 B, `Copy`, zero heap
//! per cycle (#77 part 2); no writer thread, no channel (#283 design
//! revision). Lives in the assembly crate, not `aura-std`, because the
//! writer type is `aura-registry`'s (C28 layering). Lifecycle: `initialize`
//! opens the writer (deferred acquisition; an open error is stored and the
//! node degrades to inert — the run completes), `eval` appends (an I/O error
//! mid-run likewise stores + degrades), `finalize` finishes the writer and
//! reports exactly one terminal outcome over the wiring's channel.

use aura_core::{Cell, Ctx, Node, ScalarKind};
use aura_registry::{TapTraceWriter, TapWriterOpener, TraceStoreError};
use aura_std::newest_cell;
use std::sync::mpsc::Sender;

/// A single-column recording sink around an in-graph streaming writer.
pub struct TapRecorder {
    kind: ScalarKind,
    opener: Option<TapWriterOpener>,
    writer: Option<TapTraceWriter>,
    error: Option<TraceStoreError>,
    emitted: bool,
    outcome_tx: Sender<Result<(), TraceStoreError>>,
}

impl TapRecorder {
    pub fn new(
        kind: ScalarKind,
        opener: TapWriterOpener,
        outcome_tx: Sender<Result<(), TraceStoreError>>,
    ) -> Self {
        Self { kind, opener: Some(opener), writer: None, error: None, emitted: false, outcome_tx }
    }
}

impl Node for TapRecorder {
    fn lookbacks(&self) -> Vec<usize> {
        vec![1]
    }

    fn initialize(&mut self) {
        match self.opener.take().expect("initialize runs once per run").open() {
            Ok(w) => self.writer = Some(w),
            Err(e) => self.error = Some(e),
        }
    }

    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        let cell = newest_cell(self.kind, &ctx)?; // None until warm
        if let Some(w) = self.writer.as_mut()
            && let Err(e) = w.append(ctx.now(), cell)
        {
            // I/O failure mid-run: store it, go inert, keep consuming —
            // the run completes; the failure surfaces once at finalize.
            self.error = Some(e);
            self.writer = None;
        }
        None
    }

    fn finalize(&mut self) {
        if self.emitted {
            return;
        }
        self.emitted = true;
        let outcome = match (self.writer.take(), self.error.take()) {
            (Some(w), None) => w.finish(),
            (_, Some(e)) => Err(e),
            (None, None) => Ok(()), // finalize without initialize: benign
        };
        let _ = self.outcome_tx.send(outcome);
    }

    fn label(&self) -> String {
        "TapRecorder".to_string()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use aura_core::{AnyColumn, Scalar, Timestamp};
    use aura_registry::TraceStore;
    use std::path::PathBuf;
    use std::sync::mpsc;

    /// A fresh unique store root under the OS temp dir (aura-runner has no
    /// tempfile dev-dep; mirror of trace_store's own temp_traces_root).
    fn temp_store_root(name: &str) -> PathBuf {
        let dir = std::env::temp_dir()
            .join(format!("aura-runner-taprec-{name}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).expect("create temp store root");
        dir
    }

    #[test]
    fn initialize_opens_records_after_warmup_and_reports_one_ok_outcome() {
        let root = temp_store_root("happy");
        let store = TraceStore::open(&root);
        let streamer = store.begin_run("rec").expect("begin");
        let (tx, rx) = mpsc::channel();
        let mut sink =
            TapRecorder::new(ScalarKind::F64, streamer.register_tap("t", ScalarKind::F64), tx);
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];

        sink.initialize();
        // cold: nothing appended.
        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);

        for (t, v) in [(2_i64, 10.0_f64), (3, 20.0), (4, 30.0)] {
            inputs[0].push(Scalar::f64(v)).unwrap();
            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None);
        }
        sink.finalize();
        sink.finalize(); // second flush must not emit a second outcome

        let outcomes: Vec<Result<(), TraceStoreError>> = rx.try_iter().collect();
        assert_eq!(outcomes.len(), 1, "exactly one terminal outcome");
        assert!(outcomes[0].is_ok(), "happy path outcome: {outcomes:?}");

        let text = std::fs::read_to_string(root.join("traces").join("rec").join("t.json"))
            .expect("finished tap file");
        let v: serde_json::Value = serde_json::from_str(&text).expect("parse");
        assert_eq!(v["tap"], "t");
        assert_eq!(v["ts"], serde_json::json!([2, 3, 4]));
        assert_eq!(v["columns"], serde_json::json!([[10.0, 20.0, 30.0]]));
    }

    #[test]
    fn a_failed_open_degrades_to_inert_and_reports_one_err_outcome() {
        let root = temp_store_root("failopen");
        let store = TraceStore::open(&root);
        let streamer = store.begin_run("rec").expect("begin");
        let opener = streamer.register_tap("t", ScalarKind::F64);
        // Make the deferred open fail: remove the run dir before initialize.
        std::fs::remove_dir_all(root.join("traces").join("rec")).expect("remove run dir");

        let (tx, rx) = mpsc::channel();
        let mut sink = TapRecorder::new(ScalarKind::F64, opener, tx);
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];

        sink.initialize();
        // inert but alive: eval consumes without panicking.
        inputs[0].push(Scalar::f64(1.0)).unwrap();
        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(2))), None);
        sink.finalize();

        let outcomes: Vec<Result<(), TraceStoreError>> = rx.try_iter().collect();
        assert_eq!(outcomes.len(), 1, "exactly one terminal outcome");
        assert!(outcomes[0].is_err(), "the stored open error surfaces terminally");
    }

    /// Exercises the mid-run branch at `eval` (lines 51-57): a successfully
    /// opened writer whose *write* later faults must store the error, drop
    /// `self.writer` (go inert — no further append attempts), and still
    /// report exactly one terminal `Err` at `finalize`, same as a failed
    /// open. `/dev/full` always faults on write, so pointing the ts temp
    /// stream at it (after a normal open) is a deterministic mid-run I/O
    /// fault once the `BufWriter`'s internal buffer forces a real flush.
    #[test]
    #[cfg(unix)]
    fn a_mid_run_append_failure_degrades_to_inert_and_reports_one_err_outcome() {
        let root = temp_store_root("appendfail");
        let store = TraceStore::open(&root);
        let streamer = store.begin_run("rec").expect("begin");
        let opener = streamer.register_tap("t", ScalarKind::F64);
        // Retarget the ts temp stream at a permanently-full device: `open`
        // still succeeds (create/truncate are no-ops on a char device), only
        // a later flushed write faults.
        std::os::unix::fs::symlink(
            "/dev/full",
            root.join("traces").join("rec").join("t.ts.tmp"),
        )
        .expect("symlink t.ts.tmp to /dev/full");

        let (tx, rx) = mpsc::channel();
        let mut sink = TapRecorder::new(ScalarKind::F64, opener, tx);
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];

        sink.initialize();
        // Enough warm evals to force the BufWriter past its internal
        // capacity and trigger a real (faulting) flush; `eval` never emits a
        // row regardless of the underlying write outcome.
        for t in 0..20_000_i64 {
            inputs[0].push(Scalar::f64(1.0)).unwrap();
            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None);
        }
        sink.finalize();

        let outcomes: Vec<Result<(), TraceStoreError>> = rx.try_iter().collect();
        assert_eq!(outcomes.len(), 1, "exactly one terminal outcome");
        assert!(outcomes[0].is_err(), "the mid-run write fault surfaces terminally");
    }
}