Aura/crates/aura-std/src/recorder.rs

//! `Recorder` — a reusable recording sink (the glossary *sink* role, C8/C22):
//! a pure consumer that, each fired cycle, sends `(ctx.now(), row)` — the newest
//! value of each declared input column — to an out-of-graph `mpsc` destination it
//! holds. It produces nothing (`output: vec![]`), so it is a leaf in the DAG. The
//! `mpsc::Sender` keeps the engine's purity invariant (C7): the node carries no
//! `Rc`/`RefCell` interior mutability, only an owned channel handle. Supports all
//! four base scalar kinds so any column can be persisted; returns `None` (filters)
//! until every input column is warm.

use aura_core::{Cell, Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
use std::sync::mpsc::Sender;

/// A recording sink over `kinds.len()` input columns. Each fired cycle it reads
/// the newest value of every column and sends the row to `tx`; it returns `None`
/// (records, forwards nothing) and `None` during warm-up until all columns have a
/// value.
///
/// Multiple taps fire at their own node's cadence, so their recorded streams have
/// different lengths and different firing instants. To fuse several taps into one
/// trace, join on the recorded timestamp — never by positional index, which
/// misaligns the columns or panics. The engine's report layer ships a `join_on_ts`
/// helper (in `aura-engine`) for exactly this; it is named in prose rather than
/// linked because `aura-std` does not depend on `aura-engine`.
pub struct Recorder {
    kinds: Vec<ScalarKind>,
    tx: Sender<(Timestamp, Vec<Scalar>)>,
}

impl Recorder {
    /// A recorder over one input column per entry in `kinds`, each with the given
    /// `firing` policy, sending recorded `(timestamp, row)` pairs to `tx`. The
    /// `firing` policy is a property of the declared signature (carried by the
    /// `PrimitiveBuilder`'s `PortSpec`s), not of the built sink, so it is part of
    /// the construction contract but not stored on the instance.
    pub fn new(kinds: &[ScalarKind], _firing: Firing, tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
        Self { kinds: kinds.to_vec(), tx }
    }

    /// The param-generic recipe for a blueprint primitive. The channel + kinds + firing
    /// are non-param construction args (captured), not tunable params; the node
    /// declares none. The input ports (one per kind) are threaded into the schema
    /// statically; `tx`/`kinds` are cloned for the build closure.
    pub fn builder(
        kinds: Vec<ScalarKind>,
        firing: Firing,
        tx: Sender<(Timestamp, Vec<Scalar>)>,
    ) -> PrimitiveBuilder {
        let inputs = kinds
            .iter()
            .enumerate()
            .map(|(i, &kind)| PortSpec { kind, firing, name: format!("col[{i}]") })
            .collect();
        let build_kinds = kinds.clone();
        PrimitiveBuilder::new(
            "Recorder",
            NodeSchema {
                inputs,
                output: vec![],
                params: vec![],
                doc: "recording sink persisting its input series into the trace store",
            }, // sink: empty output (C8)
            move |_| Box::new(Recorder::new(&build_kinds, firing, tx.clone())),
        )
    }
}

impl Node for Recorder {
    fn lookbacks(&self) -> Vec<usize> {
        vec![1; self.kinds.len()]
    }

    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        let mut row = Vec::with_capacity(self.kinds.len());
        for (i, &kind) in self.kinds.iter().enumerate() {
            // newest of each column by kind; `?` returns None (warm-up) if cold.
            let scalar = match kind {
                ScalarKind::F64 => Scalar::f64(ctx.f64_in(i).get(0)?),
                ScalarKind::I64 => Scalar::i64(ctx.i64_in(i).get(0)?),
                ScalarKind::Bool => Scalar::bool(ctx.bool_in(i).get(0)?),
                ScalarKind::Timestamp => Scalar::ts(ctx.ts_in(i).get(0)?),
            };
            row.push(scalar);
        }
        let _ = self.tx.send((ctx.now(), row));
        None
    }

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

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

    #[test]
    fn recorder_captures_f64_stream_after_warmup() {
        let (tx, rx) = mpsc::channel();
        let mut rec = Recorder::new(&[ScalarKind::F64], Firing::Any, tx);

        // a sink declares no output (C8) — asserted on the param-generic builder
        let (tx_b, _rx_b) = mpsc::channel();
        assert!(
            Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_b)
                .schema()
                .output
                .is_empty(),
            "a sink declares no output (C8)"
        );
        // size the one f64 input column from the node's lookback, as bootstrap would.
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, rec.lookbacks()[0])];

        // cold: returns None and records nothing.
        assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(1))), None);
        assert!(rx.try_recv().is_err());

        // warm: returns None (pure consumer) but records (now, [F64(newest)]).
        for (t, v) in [(2_i64, 10.0_f64), (3, 20.0), (4, 30.0)] {
            inputs[0].push(Scalar::f64(v)).unwrap();
            assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(t))), None);
        }
        let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        assert_eq!(
            rows,
            vec![
                (Timestamp(2), vec![Scalar::f64(10.0)]),
                (Timestamp(3), vec![Scalar::f64(20.0)]),
                (Timestamp(4), vec![Scalar::f64(30.0)]),
            ]
        );
    }

    #[test]
    fn recorder_is_none_until_all_columns_warm() {
        let (tx, rx) = mpsc::channel();
        let mut rec = Recorder::new(&[ScalarKind::F64, ScalarKind::F64], Firing::Any, tx);
        let mut inputs = vec![
            AnyColumn::with_capacity(ScalarKind::F64, 1),
            AnyColumn::with_capacity(ScalarKind::F64, 1),
        ];

        // only column 0 present -> None, nothing recorded.
        inputs[0].push(Scalar::f64(1.0)).unwrap();
        assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(1))), None);
        assert!(rx.try_recv().is_err());

        // both present -> records the full row (still returns None).
        inputs[1].push(Scalar::f64(2.0)).unwrap();
        assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(2))), None);
        let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
        assert_eq!(rows, vec![(Timestamp(2), vec![Scalar::f64(1.0), Scalar::f64(2.0)])]);
    }

    #[test]
    fn input_slots_are_named_col_index() {
        let (tx, _rx) = std::sync::mpsc::channel();
        let r = Recorder::builder(vec![ScalarKind::F64, ScalarKind::F64], Firing::Any, tx);
        let names: Vec<String> = r.schema().inputs.iter().map(|p| p.name.clone()).collect();
        assert_eq!(names, ["col[0]", "col[1]"]);
    }
}