From 92e281e4ec725fd0746287ac84b7432821df8297 Mon Sep 17 00:00:00 2001 From: claude Date: Tue, 21 Jul 2026 22:51:10 +0200 Subject: [PATCH] feat(aura-registry, aura-runner): deferred writer opener; TapRecorder moves in-graph to the assembly crate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/aura-registry/src/lib.rs | 3 +- crates/aura-registry/src/trace_store.rs | 106 ++++++++++--- crates/aura-runner/src/lib.rs | 2 + crates/aura-runner/src/tap_recorder.rs | 196 ++++++++++++++++++++++++ crates/aura-std/src/lib.rs | 3 +- crates/aura-std/src/tap_cell.rs | 11 +- crates/aura-std/src/tap_recorder.rs | 96 ------------ 7 files changed, 290 insertions(+), 127 deletions(-) create mode 100644 crates/aura-runner/src/tap_recorder.rs delete mode 100644 crates/aura-std/src/tap_recorder.rs diff --git a/crates/aura-registry/src/lib.rs b/crates/aura-registry/src/lib.rs index 4fe79bf..94c594e 100644 --- a/crates/aura-registry/src/lib.rs +++ b/crates/aura-registry/src/lib.rs @@ -38,7 +38,8 @@ pub use lineage::{ mod trace_store; pub use trace_store::{ - FamilyMember, NameKind, RunTraces, TraceStore, TraceStoreError, TraceStreamer, WriteKind, + FamilyMember, NameKind, RunTraces, TapTraceWriter, TapWriterOpener, TraceStore, + TraceStoreError, TraceStreamer, WriteKind, }; /// An append-only run registry over a JSONL file: one serde_json line per diff --git a/crates/aura-registry/src/trace_store.rs b/crates/aura-registry/src/trace_store.rs index 43613fb..54046cc 100644 --- a/crates/aura-registry/src/trace_store.rs +++ b/crates/aura-registry/src/trace_store.rs @@ -276,30 +276,19 @@ pub struct TraceStreamer { } impl TraceStreamer { - /// An incremental writer for one tap; `Send`, moved into a writer thread. + /// An incremental writer for one tap, opened now. `Send`. Expressed + /// through [`TraceStreamer::register_tap`] so there is exactly one open + /// path and the two cannot drift. pub fn tap_writer(&self, tap: &str, kind: ScalarKind) -> Result { - let final_path = self.run_dir.join(format!("{tap}.json")); - let ts_path = self.run_dir.join(format!("{tap}.ts.tmp")); - let col_path = self.run_dir.join(format!("{tap}.col.tmp")); - // Read+write handles, NOT `File::create`: `finish` seeks these same - // handles back and `io::copy`-reads them, and a `File::create` handle - // is O_WRONLY — reading it back fails with EBADF (os error 9). - let ts_w = BufWriter::new( - OpenOptions::new().read(true).write(true).create(true).truncate(true).open(&ts_path)?, - ); - let col_w = BufWriter::new( - OpenOptions::new().read(true).write(true).create(true).truncate(true).open(&col_path)?, - ); - Ok(TapTraceWriter { - tap: tap.to_string(), - kind, - final_path, - ts_path, - col_path, - ts_w, - col_w, - any: false, - }) + self.register_tap(tap, kind).open() + } + + /// The deferred twin of [`TraceStreamer::tap_writer`]: everything needed + /// to open one tap's incremental writer later — from inside a consumer + /// node's `initialize`, at run start — without holding the streamer. + /// `Send`; opens no handles until [`TapWriterOpener::open`]. + pub fn register_tap(&self, tap: &str, kind: ScalarKind) -> TapWriterOpener { + TapWriterOpener { run_dir: self.run_dir.clone(), tap: tap.to_string(), kind } } /// Write one complete tap file at once (the fold path's one-row trace) — @@ -314,6 +303,41 @@ impl TraceStreamer { } } +/// See [`TraceStreamer::register_tap`]. +pub struct TapWriterOpener { + run_dir: PathBuf, + tap: String, + kind: ScalarKind, +} + +impl TapWriterOpener { + /// Open the two temp column streams and return the incremental writer. + pub fn open(self) -> Result { + let final_path = self.run_dir.join(format!("{}.json", self.tap)); + let ts_path = self.run_dir.join(format!("{}.ts.tmp", self.tap)); + let col_path = self.run_dir.join(format!("{}.col.tmp", self.tap)); + // Read+write handles, NOT `File::create`: `finish` seeks these same + // handles back and `io::copy`-reads them, and a `File::create` handle + // is O_WRONLY — reading it back fails with EBADF (os error 9). + let ts_w = BufWriter::new( + OpenOptions::new().read(true).write(true).create(true).truncate(true).open(&ts_path)?, + ); + let col_w = BufWriter::new( + OpenOptions::new().read(true).write(true).create(true).truncate(true).open(&col_path)?, + ); + Ok(TapTraceWriter { + tap: self.tap, + kind: self.kind, + final_path, + ts_path, + col_path, + ts_w, + col_w, + any: false, + }) + } +} + /// The four-kind tag string (the on-disk `ColumnarTrace.kinds` form). A local /// copy of `report.rs`'s private `kind_tag`; the streamed-vs-write /// byte-equality test below is the drift guard. @@ -733,6 +757,42 @@ mod tests { let _ = fs::remove_dir_all(&root); } + /// `register_tap`/`TapWriterOpener::open` produce the exact same bytes as + /// `tap_writer` for the same rows — the deferred opener and the + /// open-now writer are one write path (`tap_writer` delegates to + /// `register_tap(..).open()`), so they cannot drift. + #[test] + fn register_tap_open_is_the_same_write_path_as_tap_writer() { + use aura_core::Cell; + let root = temp_traces_root("deferred-open"); + let store = TraceStore::open(&root); + let rows = + [(Timestamp(2), Cell::from_f64(0.25)), (Timestamp(3), Cell::from_f64(-1.5))]; + + // run "a": open-now. + let sa = store.begin_run("a").expect("begin a"); + let mut wa = sa.tap_writer("t", ScalarKind::F64).expect("open now"); + for &(ts, c) in &rows { + wa.append(ts, c).expect("append a"); + } + wa.finish().expect("finish a"); + + // run "b": deferred — the opener is Send and opens at use time. + let sb = store.begin_run("b").expect("begin b"); + let opener = sb.register_tap("t", ScalarKind::F64); + fn assert_send(_: &T) {} + assert_send(&opener); + let mut wb = opener.open().expect("deferred open"); + for &(ts, c) in &rows { + wb.append(ts, c).expect("append b"); + } + wb.finish().expect("finish b"); + + let a = fs::read(root.join("traces").join("a").join("t.json")).expect("read a"); + let b = fs::read(root.join("traces").join("b").join("t.json")).expect("read b"); + assert_eq!(a, b, "one open path, identical bytes"); + } + /// `TraceStreamer::finish` + `write_full` produce a run `read` returns in /// caller order — the fold path's one-row trace beside a streamed tap. #[test] diff --git a/crates/aura-runner/src/lib.rs b/crates/aura-runner/src/lib.rs index 0b46b83..8f251ca 100644 --- a/crates/aura-runner/src/lib.rs +++ b/crates/aura-runner/src/lib.rs @@ -18,10 +18,12 @@ pub mod member; pub mod project; pub mod reproduce; pub mod runner; +pub mod tap_recorder; pub mod translate; pub use project::Env; pub use runner::DefaultMemberRunner; +pub use tap_recorder::TapRecorder; /// A refusal a library function reports instead of exiting the process /// itself. The shell (`aura-cli`'s `dispatch_reproduce`) is the single place diff --git a/crates/aura-runner/src/tap_recorder.rs b/crates/aura-runner/src/tap_recorder.rs new file mode 100644 index 0000000..b03b8ea --- /dev/null +++ b/crates/aura-runner/src/tap_recorder.rs @@ -0,0 +1,196 @@ +//! `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, + writer: Option, + error: Option, + emitted: bool, + outcome_tx: Sender>, +} + +impl TapRecorder { + pub fn new( + kind: ScalarKind, + opener: TapWriterOpener, + outcome_tx: Sender>, + ) -> Self { + Self { kind, opener: Some(opener), writer: None, error: None, emitted: false, outcome_tx } + } +} + +impl Node for TapRecorder { + fn lookbacks(&self) -> Vec { + 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> = 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> = 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> = 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"); + } +} diff --git a/crates/aura-std/src/lib.rs b/crates/aura-std/src/lib.rs index a176b61..5c1d332 100644 --- a/crates/aura-std/src/lib.rs +++ b/crates/aura-std/src/lib.rs @@ -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; diff --git a/crates/aura-std/src/tap_cell.rs b/crates/aura-std/src/tap_cell.rs index 3432954..d2b8158 100644 --- a/crates/aura-std/src/tap_cell.rs +++ b/crates/aura-std/src/tap_cell.rs @@ -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 { +pub 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)?), diff --git a/crates/aura-std/src/tap_recorder.rs b/crates/aura-std/src/tap_recorder.rs deleted file mode 100644 index 04bf796..0000000 --- a/crates/aura-std/src/tap_recorder.rs +++ /dev/null @@ -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 { - 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"); - } -}