Brummel/Aura
1
0
Fork 0
Code Issues 33 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(aura-registry, aura-runner): deferred writer opener; TapRecorder moves in-graph to the assembly crate

Browse Source 
Two #283 groundwork moves for the fold-registry model:

- TraceStreamer::register_tap returns a TapWriterOpener — everything
  needed to open one tap's incremental writer later, from inside a
  consumer node's initialize, without holding the streamer. tap_writer
  is re-expressed as register_tap(..).open(), so there is exactly one
  open path; pinned by a byte-equality test between the two.

- TapRecorder leaves aura-std (which must not know aura-registry, C28
  layering) and is reworked in aura-runner as the record entry's
  consumer: it holds the TapTraceWriter in-graph — initialize opens
  (deferred; an open error is stored and the node degrades to inert),
  eval appends per fired warm cycle (zero heap, no thread, no
  channel), finalize finishes the writer and reports exactly one
  terminal outcome. The SyncSender-based aura-std variant and its
  planned writer-thread protocol are retired; newest_cell becomes pub
  for the assembly crate.

Verification: cargo clippy --workspace --all-targets -D warnings
clean; aura-core/engine/registry/std/runner suites green (new tests:
opener equivalence, happy-path capture + single Ok outcome,
failed-open inert + single Err outcome).

refs #283 refs #77
This commit is contained in:
Claude
2026-07-21 22:51:10 +02:00
parent b8ba324a41
commit 92e281e4ec
7 changed files with 290 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-std/src/lib.rs
+1 -2
View File
@@ -46,7 +46,6 @@ mod sub;
mod tap_cell;
mod tap_fold;
mod tap_live;
mod tap_recorder;
mod when;
pub use abs::Abs;
pub use add::Add;
@@ -74,7 +73,7 @@ pub use sign::Sign;
pub use sma::Sma;
pub use sqrt::Sqrt;
pub use sub::Sub;
pub use tap_cell::newest_cell;
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;
crates/aura-std/src/tap_cell.rs
+6 -5
View File
@@ -1,13 +1,14 @@
//! 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.
//! Shared single-column read for the tap sinks (`TapFold`, `TapLive`, and
//! the assembly crate's `aura_runner::TapRecorder`): 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<Cell> {
pub fn newest_cell(kind: ScalarKind, ctx: &Ctx<'_>) -> Option<Cell> {
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)?),
crates/aura-std/src/tap_recorder.rs
-96
View File
@@ -1,96 +0,0 @@
//! `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<usize> {
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::<Vec<i64>>(), "lossless, in order");
}
}