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"); - } -}