feat(aura-registry): streaming trace write path — begin_run, TraceStreamer, TapTraceWriter

Adds the incremental write path the tap-subscriber cycle streams
record subscriptions through: TraceStore::begin_run opens a run
directory and returns a TraceStreamer; per tap, TapTraceWriter
appends (ts, cell) points into two temp column streams and finish()
assembles the canonical ColumnarTrace JSON by streaming
concatenation (io::copy), removing the temps. index.json is written
last by TraceStreamer::finish, preserving the store's crash
discipline (no index -> NotFound, treat-as-absent).

Compatibility is pinned as an equality property, not a frozen byte
assumption: a new test asserts the streamed output byte-identical to
TraceStore::write's for the same rows (including empty and bool
columns), so the two write paths cannot drift and every reader
(read, read_family, chart) stays untouched. TraceStore::write is
refactored onto shared write_tap_file/write_index helpers used by
both paths. Temp streams open read+write (OpenOptions), since
finish() seeks back and re-reads them.

Verification: cargo test -p aura-registry — 78 passed (3 new:
byte-equality incl. empty+bool, crash shape NotFound, caller-order
index).

refs #283

crates/aura-registry/src/lib.rs

@@ -37,7 +37,9 @@ pub use lineage::{
 };
 
 mod trace_store;
-pub use trace_store::{FamilyMember, NameKind, RunTraces, TraceStore, TraceStoreError, WriteKind};
+pub use trace_store::{
+    FamilyMember, NameKind, RunTraces, TraceStore, TraceStoreError, TraceStreamer, WriteKind,
+};
 
 /// An append-only run registry over a JSONL file: one serde_json line per
 /// `RunReport`.

crates/aura-registry/src/trace_store.rs

@@ -7,8 +7,11 @@
 
 use std::fmt;
 use std::fs;
+use std::fs::{File, OpenOptions};
+use std::io::{self, BufWriter, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 
+use aura_core::{Cell, ScalarKind, Timestamp};
 use aura_engine::{ColumnarTrace, RunManifest};
 use serde::{Deserialize, Serialize};
 
@@ -97,24 +100,19 @@ impl TraceStore {
         let run_dir = self.dir.join(name);
         fs::create_dir_all(&run_dir)?;
         for tap in taps {
-            let path = run_dir.join(format!("{}.json", tap.tap));
+            write_tap_file(&run_dir, tap)?;
-            let body = serde_json::to_string(tap).map_err(|source| TraceStoreError::Parse {
-                file: path.display().to_string(),
-                source,
-            })?;
-            fs::write(&path, body)?;
         }
-        let index = Index {
+        let order: Vec<String> = taps.iter().map(|t| t.tap.clone()).collect();
-            manifest: manifest.clone(),
+        write_index(&run_dir, manifest, &order)
-            taps: taps.iter().map(|t| t.tap.clone()).collect(),
+    }
-        };
+
-        let index_path = run_dir.join("index.json");
+    /// Begin a streamed run write under `traces/<name>/` (directories created
-        let body = serde_json::to_string(&index).map_err(|source| TraceStoreError::Parse {
+    /// now; `index.json` only at [`TraceStreamer::finish`] — the crash
-            file: index_path.display().to_string(),
+    /// discipline `read` already honors: no index → `NotFound`).
-            source,
+    pub fn begin_run(&self, name: &str) -> Result<TraceStreamer, TraceStoreError> {
-        })?;
+        let run_dir = self.dir.join(name);
-        fs::write(&index_path, body)?;
+        fs::create_dir_all(&run_dir)?;
-        Ok(())
+        Ok(TraceStreamer { run_dir })
     }
 
     /// Read a run's index + taps back, in index order. A missing run directory (no
@@ -239,6 +237,192 @@ impl TraceStore {
     }
 }
 
+/// One tap's canonical `<tap>.json` under `run_dir` — the single write shape
+/// both `TraceStore::write` and `TraceStreamer::write_full` share.
+fn write_tap_file(run_dir: &Path, tap: &ColumnarTrace) -> Result<(), TraceStoreError> {
+    let path = run_dir.join(format!("{}.json", tap.tap));
+    let body = serde_json::to_string(tap).map_err(|source| TraceStoreError::Parse {
+        file: path.display().to_string(),
+        source,
+    })?;
+    fs::write(&path, body)?;
+    Ok(())
+}
+
+/// The run's `index.json` (manifest + tap order) — shared by `write` and
+/// `TraceStreamer::finish`. Written last by both paths (crash discipline:
+/// no `index.json` → `NotFound`, treat-as-absent).
+fn write_index(
+    run_dir: &Path,
+    manifest: &RunManifest,
+    taps: &[String],
+) -> Result<(), TraceStoreError> {
+    let index = Index { manifest: manifest.clone(), taps: taps.to_vec() };
+    let index_path = run_dir.join("index.json");
+    let body = serde_json::to_string(&index).map_err(|source| TraceStoreError::Parse {
+        file: index_path.display().to_string(),
+        source,
+    })?;
+    fs::write(&index_path, body)?;
+    Ok(())
+}
+
+/// A begun streamed run write: hands out per-tap incremental writers and
+/// writes the closing `index.json`. Tap order in the index is the caller's
+/// (declared-tap order — the caller tracks it; this type keeps no order
+/// state so record-writer threads and post-run fold writes can interleave).
+pub struct TraceStreamer {
+    run_dir: PathBuf,
+}
+
+impl TraceStreamer {
+    /// An incremental writer for one tap; `Send`, moved into a writer thread.
+    pub fn tap_writer(&self, tap: &str, kind: ScalarKind) -> Result<TapTraceWriter, TraceStoreError> {
+        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,
+        })
+    }
+
+    /// Write one complete tap file at once (the fold path's one-row trace) —
+    /// the same bytes `TraceStore::write` produces for it.
+    pub fn write_full(&self, tap: &ColumnarTrace) -> Result<(), TraceStoreError> {
+        write_tap_file(&self.run_dir, tap)
+    }
+
+    /// Close the run: write `index.json` (manifest + `taps` in caller order).
+    pub fn finish(self, manifest: &RunManifest, taps: &[String]) -> Result<(), TraceStoreError> {
+        write_index(&self.run_dir, manifest, taps)
+    }
+}
+
+/// 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.
+fn tag(kind: ScalarKind) -> &'static str {
+    match kind {
+        ScalarKind::F64 => "F64",
+        ScalarKind::I64 => "I64",
+        ScalarKind::Bool => "Bool",
+        ScalarKind::Timestamp => "Timestamp",
+    }
+}
+
+/// Coerce one cell to the f64 a trace column stores — the same table as
+/// `ColumnarTrace::from_rows` (`scalar_to_f64`): f64 as-is, i64 as f64,
+/// bool 1.0/0.0, timestamp epoch as f64.
+fn cell_to_f64(kind: ScalarKind, cell: Cell) -> f64 {
+    match kind {
+        ScalarKind::F64 => cell.f64(),
+        ScalarKind::I64 => cell.i64() as f64,
+        ScalarKind::Bool => {
+            if cell.bool() {
+                1.0
+            } else {
+                0.0
+            }
+        }
+        ScalarKind::Timestamp => cell.ts().0 as f64,
+    }
+}
+
+/// Incremental single-column trace writer: appends each `(ts, cell)` to two
+/// temp streams, then `finish` assembles the canonical `ColumnarTrace` JSON
+/// by streaming concatenation (constant memory) and removes the temps. The
+/// output is byte-equal to `serde_json::to_string(&ColumnarTrace)` for the
+/// same rows — pinned by test against `TraceStore::write`, which is the
+/// anchor (not a frozen byte-shape assumption).
+pub struct TapTraceWriter {
+    tap: String,
+    kind: ScalarKind,
+    final_path: PathBuf,
+    ts_path: PathBuf,
+    col_path: PathBuf,
+    ts_w: BufWriter<File>,
+    col_w: BufWriter<File>,
+    any: bool,
+}
+
+impl TapTraceWriter {
+    /// Append one recorded point. Number formatting is `serde_json`'s own
+    /// (`to_writer` per value) — identical to the whole-struct serialize.
+    pub fn append(&mut self, ts: Timestamp, cell: Cell) -> Result<(), TraceStoreError> {
+        if self.any {
+            self.ts_w.write_all(b",")?;
+            self.col_w.write_all(b",")?;
+        }
+        serde_json::to_writer(&mut self.ts_w, &ts.0).map_err(|source| TraceStoreError::Parse {
+            file: self.ts_path.display().to_string(),
+            source,
+        })?;
+        serde_json::to_writer(&mut self.col_w, &cell_to_f64(self.kind, cell)).map_err(
+            |source| TraceStoreError::Parse {
+                file: self.col_path.display().to_string(),
+                source,
+            },
+        )?;
+        self.any = true;
+        Ok(())
+    }
+
+    /// Assemble `{"tap":…,"kinds":[…],"ts":[⟨ts-tmp⟩],"columns":[[⟨col-tmp⟩]]}`
+    /// and remove the temps.
+    pub fn finish(self) -> Result<(), TraceStoreError> {
+        let TapTraceWriter { tap, kind, final_path, ts_path, col_path, ts_w, col_w, any: _ } = self;
+        let mut ts_f = ts_w.into_inner().map_err(|e| TraceStoreError::Io(e.into_error()))?;
+        let mut col_f = col_w.into_inner().map_err(|e| TraceStoreError::Io(e.into_error()))?;
+        ts_f.flush()?;
+        col_f.flush()?;
+        ts_f.seek(SeekFrom::Start(0))?;
+        col_f.seek(SeekFrom::Start(0))?;
+
+        let mut out = BufWriter::new(File::create(&final_path)?);
+        // Field order (tap, kinds, ts, columns) = ColumnarTrace declaration
+        // order; string fields serialized via serde for correct escaping.
+        write!(
+            out,
+            "{{\"tap\":{},\"kinds\":[{}],\"ts\":[",
+            serde_json::to_string(&tap).map_err(|source| TraceStoreError::Parse {
+                file: final_path.display().to_string(),
+                source,
+            })?,
+            serde_json::to_string(tag(kind)).map_err(|source| TraceStoreError::Parse {
+                file: final_path.display().to_string(),
+                source,
+            })?,
+        )?;
+        io::copy(&mut ts_f, &mut out)?;
+        out.write_all(b"],\"columns\":[[")?;
+        io::copy(&mut col_f, &mut out)?;
+        out.write_all(b"]]}")?;
+        out.flush()?;
+        drop(out);
+        drop(ts_f);
+        drop(col_f);
+        fs::remove_file(&ts_path)?;
+        fs::remove_file(&col_path)?;
+        Ok(())
+    }
+}
+
 /// What can go wrong reading or writing the trace store.
 #[derive(Debug)]
 pub enum TraceStoreError {
@@ -447,4 +631,132 @@ mod tests {
         assert!(store.ensure_name_free("fresh", WriteKind::Run).is_ok());
         let _ = fs::remove_dir_all(&root);
     }
+
+    /// The load-bearing compatibility pin: for the same rows, the streamed
+    /// writer's file bytes equal `TraceStore::write`'s — the legacy path's
+    /// bytes, whatever they are, define the shape. Covers all four
+    /// `ScalarKind` arms (f64 fractional/negative, bool, i64, timestamp) —
+    /// every arm `tag`/`cell_to_f64` dispatch on, since those two helpers are
+    /// a local copy of `report.rs`'s private `kind_tag`/`scalar_to_f64` and
+    /// this is the only drift guard against them — plus the empty
+    /// (never-warm) tap.
+    #[test]
+    fn streamed_writer_is_byte_equal_to_write_for_the_same_rows() {
+        use aura_core::Cell;
+
+        type Case<'a> = (&'a str, ScalarKind, Vec<(Timestamp, Cell)>);
+        let cases: Vec<Case> = vec![
+            (
+                "f64_tap",
+                ScalarKind::F64,
+                vec![
+                    (Timestamp(2), Cell::from_f64(0.30000000000000004)),
+                    (Timestamp(3), Cell::from_f64(-1.5)),
+                    (Timestamp(5), Cell::from_f64(2.0)),
+                ],
+            ),
+            (
+                "bool_tap",
+                ScalarKind::Bool,
+                vec![(Timestamp(1), Cell::from_bool(true)), (Timestamp(2), Cell::from_bool(false))],
+            ),
+            (
+                "i64_tap",
+                ScalarKind::I64,
+                vec![(Timestamp(1), Cell::from_i64(-7)), (Timestamp(2), Cell::from_i64(42))],
+            ),
+            (
+                "timestamp_tap",
+                ScalarKind::Timestamp,
+                vec![
+                    (Timestamp(1), Cell::from_ts(Timestamp(100))),
+                    (Timestamp(2), Cell::from_ts(Timestamp(-3))),
+                ],
+            ),
+            ("empty_tap", ScalarKind::F64, vec![]),
+        ];
+
+        for (tap, kind, rows) in cases {
+            // legacy path
+            let legacy_root = temp_traces_root(&format!("byteeq-legacy-{tap}"));
+            let legacy = TraceStore::open(&legacy_root);
+            let scalar_rows: Vec<(Timestamp, Vec<Scalar>)> = rows
+                .iter()
+                .map(|&(t, c)| (t, vec![Scalar::from_cell(kind, c)]))
+                .collect();
+            let trace = ColumnarTrace::from_rows(tap, &[kind], &scalar_rows);
+            legacy.write("run", &sample_manifest(), &[trace]).expect("legacy write");
+            let legacy_bytes =
+                fs::read(legacy_root.join(format!("traces/run/{tap}.json"))).expect("legacy bytes");
+
+            // streamed path
+            let stream_root = temp_traces_root(&format!("byteeq-stream-{tap}"));
+            let store = TraceStore::open(&stream_root);
+            let streamer = store.begin_run("run").expect("begin_run");
+            let mut w = streamer.tap_writer(tap, kind).expect("tap_writer");
+            for (t, c) in rows {
+                w.append(t, c).expect("append");
+            }
+            w.finish().expect("finish");
+            let streamed_bytes =
+                fs::read(stream_root.join(format!("traces/run/{tap}.json"))).expect("streamed bytes");
+
+            assert_eq!(
+                streamed_bytes, legacy_bytes,
+                "streamed and legacy write bytes must be identical for {tap}"
+            );
+            // temps removed
+            assert!(!stream_root.join(format!("traces/run/{tap}.ts.tmp")).exists());
+            assert!(!stream_root.join(format!("traces/run/{tap}.col.tmp")).exists());
+
+            let _ = fs::remove_dir_all(&legacy_root);
+            let _ = fs::remove_dir_all(&stream_root);
+        }
+    }
+
+    /// A begun-but-unfinished streamed run (crash shape: tap file written or
+    /// half-written, no `index.json`) reads as `NotFound` — treat-as-absent.
+    #[test]
+    fn begun_but_unfinished_streamed_run_reads_not_found() {
+        use aura_core::Cell;
+        let root = temp_traces_root("stream-crash");
+        let store = TraceStore::open(&root);
+        let streamer = store.begin_run("crashed").expect("begin_run");
+        let mut w = streamer.tap_writer("t", ScalarKind::F64).expect("tap_writer");
+        w.append(Timestamp(1), Cell::from_f64(1.0)).expect("append");
+        // no w.finish(), no streamer.finish(): the crash shape.
+        drop(w);
+        match store.read("crashed") {
+            Err(TraceStoreError::NotFound(name)) => assert_eq!(name, "crashed"),
+            other => panic!("expected NotFound, got {other:?}"),
+        }
+        let _ = fs::remove_dir_all(&root);
+    }
+
+    /// `TraceStreamer::finish` + `write_full` produce a run `read` returns in
+    /// caller order — the fold path's one-row trace beside a streamed tap.
+    #[test]
+    fn streamer_finish_indexes_streamed_and_full_taps_in_caller_order() {
+        use aura_core::Cell;
+        let root = temp_traces_root("stream-index");
+        let store = TraceStore::open(&root);
+        let streamer = store.begin_run("mixed").expect("begin_run");
+        let mut w = streamer.tap_writer("series", ScalarKind::F64).expect("tap_writer");
+        w.append(Timestamp(1), Cell::from_f64(1.0)).expect("append");
+        w.finish().expect("finish series");
+        let fold_row = ColumnarTrace::from_rows(
+            "agg",
+            &[ScalarKind::I64],
+            &[(Timestamp(1), vec![Scalar::i64(1)])],
+        );
+        streamer.write_full(&fold_row).expect("write_full");
+        streamer
+            .finish(&sample_manifest(), &["series".to_string(), "agg".to_string()])
+            .expect("streamer finish");
+
+        let back = store.read("mixed").expect("read");
+        let names: Vec<&str> = back.taps.iter().map(|t| t.tap.as_str()).collect();
+        assert_eq!(names, vec!["series", "agg"], "index order = caller order");
+        let _ = fs::remove_dir_all(&root);
+    }
 }