From 09994b83ed4ccf60eba44f333e53671651483dad Mon Sep 17 00:00:00 2001 From: claude Date: Tue, 21 Jul 2026 19:55:30 +0200 Subject: [PATCH] =?UTF-8?q?feat(aura-std):=20tap=20subscriber=20sinks=20?= =?UTF-8?q?=E2=80=94=20TapFold,=20TapRecorder,=20TapLive?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The three C8 sinks realizing the per-tap subscription model (#283): TapFold (closed FoldKind vocabulary Count|Sum|Mean|Min|Max|First|Last, in-graph accumulation, exactly one finalize row), TapRecorder (bounded SyncSender<(Timestamp, Cell)>, lossless backpressure — a full channel blocks the sim, wall time never state), and TapLive (consumer-owned closure, inline on the sim thread). All three move (Timestamp, Cell) — 16 B, Copy — instead of (Timestamp, Vec): zero per-cycle heap on the tap path (#77 part 2). The shared kind-dispatched newest-cell read lives in tap_cell.rs. Known-good subset commit of iter tap-subscribers (tasks 1-3 of 7, each gated implementer -> spec-compliance -> quality DONE; the loop blocked on a task-4 plan defect, repaired separately): the trace-store streaming writer, the TapPlan wiring seam, and both entry-point rewires follow in this same cycle. Verified directly: cargo test -p aura-std — 109 passed, 0 failed. refs #283, refs #77 --- crates/aura-std/src/lib.rs | 7 + crates/aura-std/src/tap_cell.rs | 17 ++ crates/aura-std/src/tap_fold.rs | 263 ++++++++++++++++++++++++++++ crates/aura-std/src/tap_live.rs | 88 ++++++++++ crates/aura-std/src/tap_recorder.rs | 96 ++++++++++ 5 files changed, 471 insertions(+) create mode 100644 crates/aura-std/src/tap_cell.rs create mode 100644 crates/aura-std/src/tap_fold.rs create mode 100644 crates/aura-std/src/tap_live.rs create mode 100644 crates/aura-std/src/tap_recorder.rs diff --git a/crates/aura-std/src/lib.rs b/crates/aura-std/src/lib.rs index 4f55fe0..a176b61 100644 --- a/crates/aura-std/src/lib.rs +++ b/crates/aura-std/src/lib.rs @@ -43,6 +43,10 @@ mod sign; mod sma; mod sqrt; mod sub; +mod tap_cell; +mod tap_fold; +mod tap_live; +mod tap_recorder; mod when; pub use abs::Abs; pub use add::Add; @@ -70,4 +74,7 @@ pub use sign::Sign; pub use sma::Sma; pub use sqrt::Sqrt; pub use sub::Sub; +pub use tap_fold::{fold_binds_at, fold_output_kind, FoldKind, TapFold}; +pub use tap_live::TapLive; +pub use tap_recorder::TapRecorder; pub use when::When; diff --git a/crates/aura-std/src/tap_cell.rs b/crates/aura-std/src/tap_cell.rs new file mode 100644 index 0000000..3432954 --- /dev/null +++ b/crates/aura-std/src/tap_cell.rs @@ -0,0 +1,17 @@ +//! Shared single-column read for the tap sinks (`TapRecorder`, `TapFold`, +//! `TapLive`): the kind-dispatched newest-value read `Recorder::eval` does, +//! re-encoded as a bare `Cell` (the C7 kind-erased carrier) instead of a +//! kind-tagged `Scalar` — the #77-part-2 zero-heap payload. + +use aura_core::{Cell, Ctx, ScalarKind}; + +/// The newest value of input column 0, as a `Cell`. `None` until the column +/// is warm (the `?` on the window read), mirroring `Recorder`'s warm-up. +pub(crate) fn newest_cell(kind: ScalarKind, ctx: &Ctx<'_>) -> Option { + Some(match kind { + ScalarKind::F64 => Cell::from_f64(ctx.f64_in(0).get(0)?), + ScalarKind::I64 => Cell::from_i64(ctx.i64_in(0).get(0)?), + ScalarKind::Bool => Cell::from_bool(ctx.bool_in(0).get(0)?), + ScalarKind::Timestamp => Cell::from_ts(ctx.ts_in(0).get(0)?), + }) +} diff --git a/crates/aura-std/src/tap_fold.rs b/crates/aura-std/src/tap_fold.rs new file mode 100644 index 0000000..dd2baa1 --- /dev/null +++ b/crates/aura-std/src/tap_fold.rs @@ -0,0 +1,263 @@ +//! `TapFold` — a folding sink (C8) over one declared-tap column: accumulates +//! owned per-cycle state (no channel traffic, no heap per cycle) and emits +//! exactly one summary row on `finalize` (the #138 lifecycle; `SeriesReducer` +//! precedent). The fold is named from the closed [`FoldKind`] vocabulary +//! (C25: closed, typed; a new aggregate is a new named entry, never inline +//! logic in a data artifact). C7-pure: owned state + the channel handle. + +use aura_core::{Cell, Ctx, Node, Scalar, ScalarKind, Timestamp}; +use std::sync::mpsc::Sender; + +use crate::tap_cell::newest_cell; + +/// The closed fold vocabulary. Bind rules: `Sum`/`Mean`/`Min`/`Max` bind only +/// at f64 taps ([`fold_binds_at`]); `Count` at any kind; `First`/`Last` +/// kind-preserving at any kind. Output kinds: [`fold_output_kind`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FoldKind { + Count, + Sum, + Mean, + Min, + Max, + First, + Last, +} + +/// May `fold` bind at a tap of column kind `kind`? The arithmetic folds are +/// f64-only; `Count`/`First`/`Last` are kind-agnostic. +pub fn fold_binds_at(fold: FoldKind, kind: ScalarKind) -> bool { + match fold { + FoldKind::Sum | FoldKind::Mean | FoldKind::Min | FoldKind::Max => { + kind == ScalarKind::F64 + } + FoldKind::Count | FoldKind::First | FoldKind::Last => true, + } +} + +/// The output column kind of `fold` at a tap of column kind `kind`: +/// `Count` → i64; the arithmetic folds → f64; `First`/`Last` preserve `kind`. +pub fn fold_output_kind(fold: FoldKind, kind: ScalarKind) -> ScalarKind { + match fold { + FoldKind::Count => ScalarKind::I64, + FoldKind::Sum | FoldKind::Mean | FoldKind::Min | FoldKind::Max => ScalarKind::F64, + FoldKind::First | FoldKind::Last => kind, + } +} + +/// Owned accumulator for every fold kind at once (a handful of words; keeping +/// it total avoids a per-kind state enum). Sequential `sum += v` matches a +/// slice-driven mean's operation order, so streamed and slice `Mean` agree +/// bit-for-bit (the `SeriesFold` discipline, IEEE-754 non-associativity +/// respected). +#[derive(Clone, Copy, Debug)] +struct FoldState { + count: u64, + sum: f64, + min: f64, + max: f64, + first: Option<(Timestamp, Cell)>, + last: Option, +} + +impl FoldState { + fn new() -> Self { + Self { + count: 0, + sum: 0.0, + min: f64::INFINITY, + max: f64::NEG_INFINITY, + first: None, + last: None, + } + } + + fn fold(&mut self, ts: Timestamp, cell: Cell, kind: ScalarKind) { + self.count += 1; + if kind == ScalarKind::F64 { + let v = cell.f64(); + self.sum += v; + if v < self.min { + self.min = v; + } + if v > self.max { + self.max = v; + } + } + if self.first.is_none() { + self.first = Some((ts, cell)); + } + self.last = Some(cell); + } + + fn folded_any(&self) -> bool { + self.count > 0 + } + + /// The summary value; call only when `folded_any()`. + fn output(&self, fold: FoldKind, kind: ScalarKind) -> Scalar { + match fold { + FoldKind::Count => Scalar::i64(self.count as i64), + FoldKind::Sum => Scalar::f64(self.sum), + FoldKind::Mean => Scalar::f64(self.sum / self.count as f64), + FoldKind::Min => Scalar::f64(self.min), + FoldKind::Max => Scalar::f64(self.max), + FoldKind::First => { + Scalar::from_cell(kind, self.first.expect("folded_any checked").1) + } + FoldKind::Last => Scalar::from_cell(kind, self.last.expect("folded_any checked")), + } + } +} + +/// A single-column folding sink. Emits exactly one row on `finalize` (never +/// twice, never on a never-warm tap): `(row_ts, [output])`, where `row_ts` is +/// the last folded value's ts — except `First`, which carries the first warm +/// value's ts (the instant its reported value was determined). +pub struct TapFold { + kind: ScalarKind, + fold: FoldKind, + state: FoldState, + last_ts: Timestamp, + emitted: bool, + tx: Sender<(Timestamp, Vec)>, +} + +impl TapFold { + pub fn new(kind: ScalarKind, fold: FoldKind, tx: Sender<(Timestamp, Vec)>) -> Self { + debug_assert!(fold_binds_at(fold, kind), "caller validates fold/kind before binding"); + Self { kind, fold, state: FoldState::new(), last_ts: Timestamp(0), emitted: false, tx } + } +} + +impl Node for TapFold { + fn lookbacks(&self) -> Vec { + vec![1] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + let cell = newest_cell(self.kind, &ctx)?; // None until warm + self.state.fold(ctx.now(), cell, self.kind); + self.last_ts = ctx.now(); + None + } + + fn finalize(&mut self) { + if self.state.folded_any() && !self.emitted { + self.emitted = true; + let ts = match self.fold { + FoldKind::First => self.state.first.expect("folded_any checked").0, + _ => self.last_ts, + }; + let _ = self.tx.send((ts, vec![self.state.output(self.fold, self.kind)])); + } + } + + fn label(&self) -> String { + format!("TapFold({:?})", self.fold) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::AnyColumn; + use std::sync::mpsc; + + /// Drive a fold over an f64 series and return the emitted rows. + fn run_f64_fold(fold: FoldKind, values: &[f64]) -> Vec<(Timestamp, Vec)> { + let (tx, rx) = mpsc::channel(); + let mut sink = TapFold::new(ScalarKind::F64, fold, tx); + let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; + for (i, &v) in values.iter().enumerate() { + inputs[0].push(Scalar::f64(v)).unwrap(); + assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(i as i64 + 1))), None); + } + sink.finalize(); + rx.try_iter().collect() + } + + #[test] + fn mean_matches_slice_mean_bit_for_bit() { + let values = [0.1, 0.2, 0.30000000000000004, -1.5, 2.75]; + let rows = run_f64_fold(FoldKind::Mean, &values); + let slice_mean = values.iter().sum::() / values.len() as f64; + assert_eq!(rows, vec![(Timestamp(5), vec![Scalar::f64(slice_mean)])]); + } + + #[test] + fn count_sum_min_max_over_a_known_series() { + let values = [3.0, -1.0, 2.0]; + assert_eq!( + run_f64_fold(FoldKind::Count, &values), + vec![(Timestamp(3), vec![Scalar::i64(3)])] + ); + assert_eq!( + run_f64_fold(FoldKind::Sum, &values), + vec![(Timestamp(3), vec![Scalar::f64(4.0)])] + ); + assert_eq!( + run_f64_fold(FoldKind::Min, &values), + vec![(Timestamp(3), vec![Scalar::f64(-1.0)])] + ); + assert_eq!( + run_f64_fold(FoldKind::Max, &values), + vec![(Timestamp(3), vec![Scalar::f64(3.0)])] + ); + } + + #[test] + fn first_carries_the_first_warm_ts_and_last_the_last() { + let values = [7.0, 8.0, 9.0]; + assert_eq!( + run_f64_fold(FoldKind::First, &values), + vec![(Timestamp(1), vec![Scalar::f64(7.0)])], + "First reports the first warm value at ITS ts" + ); + assert_eq!( + run_f64_fold(FoldKind::Last, &values), + vec![(Timestamp(3), vec![Scalar::f64(9.0)])] + ); + } + + #[test] + fn never_warm_tap_emits_nothing_and_finalize_emits_once() { + let (tx, rx) = mpsc::channel(); + let mut sink = TapFold::new(ScalarKind::F64, FoldKind::Count, tx); + let inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; + // cold eval, then finalize: nothing emitted. + assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None); + sink.finalize(); + assert!(rx.try_recv().is_err(), "a never-warm fold emits no row"); + + // warm one value; a double finalize still emits exactly one row. + let (tx2, rx2) = mpsc::channel(); + let mut sink2 = TapFold::new(ScalarKind::I64, FoldKind::Last, tx2); + let mut inputs2 = vec![AnyColumn::with_capacity(ScalarKind::I64, 1)]; + inputs2[0].push(Scalar::i64(42)).unwrap(); + assert_eq!(sink2.eval(Ctx::new(&inputs2, Timestamp(9))), None); + sink2.finalize(); + sink2.finalize(); + let rows: Vec<_> = rx2.try_iter().collect(); + assert_eq!(rows, vec![(Timestamp(9), vec![Scalar::i64(42)])]); + } + + #[test] + fn bind_rules_and_output_kinds_are_the_specced_table() { + for fold in [FoldKind::Sum, FoldKind::Mean, FoldKind::Min, FoldKind::Max] { + assert!(fold_binds_at(fold, ScalarKind::F64)); + assert!(!fold_binds_at(fold, ScalarKind::I64)); + assert!(!fold_binds_at(fold, ScalarKind::Bool)); + assert!(!fold_binds_at(fold, ScalarKind::Timestamp)); + assert_eq!(fold_output_kind(fold, ScalarKind::F64), ScalarKind::F64); + } + for kind in [ScalarKind::F64, ScalarKind::I64, ScalarKind::Bool, ScalarKind::Timestamp] { + assert!(fold_binds_at(FoldKind::Count, kind)); + assert!(fold_binds_at(FoldKind::First, kind)); + assert!(fold_binds_at(FoldKind::Last, kind)); + assert_eq!(fold_output_kind(FoldKind::Count, kind), ScalarKind::I64); + assert_eq!(fold_output_kind(FoldKind::First, kind), kind); + assert_eq!(fold_output_kind(FoldKind::Last, kind), kind); + } + } +} diff --git a/crates/aura-std/src/tap_live.rs b/crates/aura-std/src/tap_live.rs new file mode 100644 index 0000000..3cba788 --- /dev/null +++ b/crates/aura-std/src/tap_live.rs @@ -0,0 +1,88 @@ +//! `TapLive` — the live-subscription sink (C8) for one declared tap: calls a +//! consumer-owned closure `(Timestamp, Cell)` inline per fired warm cycle, on +//! the sim thread, zero alloc. Loss policy is the closure's own (a UI +//! consumer typically `try_send`s into its own channel and drops when behind) +//! — per #283: consumer-owned logic and state, no registry ceremony. + +use aura_core::{Cell, Ctx, Node, ScalarKind, Timestamp}; + +use crate::tap_cell::newest_cell; + +/// A single-column live sink around a consumer closure. +pub struct TapLive { + kind: ScalarKind, + consumer: Box, +} + +impl TapLive { + pub fn new(kind: ScalarKind, consumer: impl FnMut(Timestamp, Cell) + Send + 'static) -> Self { + Self { kind, consumer: Box::new(consumer) } + } +} + +impl Node for TapLive { + fn lookbacks(&self) -> Vec { + vec![1] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + let cell = newest_cell(self.kind, &ctx)?; // None until warm + (self.consumer)(ctx.now(), cell); + None + } + + fn label(&self) -> String { + "TapLive".to_string() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::{AnyColumn, Scalar}; + use std::sync::mpsc; + + #[test] + fn calls_the_consumer_per_warm_cycle_in_order_and_not_before() { + let (tx, rx) = mpsc::channel(); + let mut sink = TapLive::new(ScalarKind::F64, move |ts, cell| { + let _ = tx.send((ts, cell.f64())); + }); + let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; + + // cold: consumer not called. + assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None); + assert!(rx.try_recv().is_err(), "no call before warm-up"); + + for (t, v) in [(2_i64, 1.5_f64), (3, 2.5)] { + inputs[0].push(Scalar::f64(v)).unwrap(); + assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None); + } + let got: Vec<(Timestamp, f64)> = rx.try_iter().collect(); + assert_eq!(got, vec![(Timestamp(2), 1.5), (Timestamp(3), 2.5)]); + } + + /// `newest_cell`'s four-kind dispatch (shared with `TapFold`) is + /// otherwise only exercised at F64/I64; the Bool and Timestamp arms + /// need a witness too, so a regression in either goes caught. + #[test] + fn calls_the_consumer_at_bool_and_timestamp_kinds_too() { + let (tx, rx) = mpsc::channel(); + let mut sink = TapLive::new(ScalarKind::Bool, move |ts, cell| { + let _ = tx.send((ts, cell.bool())); + }); + let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::Bool, 1)]; + inputs[0].push(Scalar::bool(true)).unwrap(); + assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None); + assert_eq!(rx.try_recv(), Ok((Timestamp(1), true))); + + let (tx2, rx2) = mpsc::channel(); + let mut sink2 = TapLive::new(ScalarKind::Timestamp, move |ts, cell| { + let _ = tx2.send((ts, cell.ts())); + }); + let mut inputs2 = vec![AnyColumn::with_capacity(ScalarKind::Timestamp, 1)]; + inputs2[0].push(Scalar::ts(Timestamp(99))).unwrap(); + assert_eq!(sink2.eval(Ctx::new(&inputs2, Timestamp(1))), None); + assert_eq!(rx2.try_recv(), Ok((Timestamp(1), Timestamp(99)))); + } +} diff --git a/crates/aura-std/src/tap_recorder.rs b/crates/aura-std/src/tap_recorder.rs new file mode 100644 index 0000000..04bf796 --- /dev/null +++ b/crates/aura-std/src/tap_recorder.rs @@ -0,0 +1,96 @@ +//! `TapRecorder` — the record-subscription sink (C8) for one declared tap: +//! sends `(ctx.now(), cell)` — 16 B, `Copy`, zero heap per cycle (#77 part 2) +//! — on a caller-chosen **bounded** `SyncSender`. Lossless by contract: a +//! full channel blocks the sim (wall time only, never state — the consumer +//! is one-way, C1 intact); a disconnected receiver returns `Err` immediately +//! and the run completes (the failure surfaces at the caller's join/finish). +//! C7-pure: no interior mutability, only the owned channel handle. + +use aura_core::{Cell, Ctx, Node, ScalarKind, Timestamp}; +use std::sync::mpsc::SyncSender; + +use crate::tap_cell::newest_cell; + +/// A single-column recording sink over a bounded channel. `None` until the +/// column is warm (records nothing), `None` always (pure consumer). +pub struct TapRecorder { + kind: ScalarKind, + tx: SyncSender<(Timestamp, Cell)>, +} + +impl TapRecorder { + pub fn new(kind: ScalarKind, tx: SyncSender<(Timestamp, Cell)>) -> Self { + Self { kind, tx } + } +} + +impl Node for TapRecorder { + fn lookbacks(&self) -> Vec { + vec![1] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + let cell = newest_cell(self.kind, &ctx)?; // None until warm + // Blocks when the writer is behind (lossless); Err on a dead + // receiver — ignored here, surfaced by the caller at join. + let _ = self.tx.send((ctx.now(), cell)); + None + } + + fn label(&self) -> String { + "TapRecorder".to_string() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::{AnyColumn, Scalar}; + use std::sync::mpsc; + + #[test] + fn records_ts_cell_pairs_after_warmup() { + let (tx, rx) = mpsc::sync_channel(8); + let mut sink = TapRecorder::new(ScalarKind::F64, tx); + let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; + + // cold: nothing recorded. + assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None); + assert!(rx.try_recv().is_err()); + + 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); + } + let got: Vec<(Timestamp, f64)> = rx.try_iter().map(|(t, c)| (t, c.f64())).collect(); + assert_eq!( + got, + vec![(Timestamp(2), 10.0), (Timestamp(3), 20.0), (Timestamp(4), 30.0)] + ); + } + + /// Backpressure losslessness: a capacity-1 channel with a concurrent + /// consumer never drops a value — the producer blocks instead. 100 sends + /// through a 1-slot channel arrive complete and in order. + #[test] + fn capacity_one_channel_is_lossless_under_backpressure() { + let (tx, rx) = mpsc::sync_channel::<(Timestamp, Cell)>(1); + let consumer = std::thread::spawn(move || { + let mut got = Vec::new(); + for (_, cell) in rx.iter() { + got.push(cell.i64()); + } + got + }); + { + let mut sink = TapRecorder::new(ScalarKind::I64, tx); + let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::I64, 1)]; + for i in 0..100_i64 { + inputs[0].push(Scalar::i64(i)).unwrap(); + assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(i + 1))), None); + } + } // sink (and tx) dropped → consumer's iter ends + let got = consumer.join().expect("consumer thread"); + assert_eq!(got, (0..100).collect::>(), "lossless, in order"); + } +}