From 4ae2297a35d04063bacb293c5daa157f992b6662 Mon Sep 17 00:00:00 2001 From: claude Date: Tue, 21 Jul 2026 21:05:52 +0200 Subject: [PATCH] =?UTF-8?q?feat(aura-registry):=20streaming=20trace=20writ?= =?UTF-8?q?e=20path=20=E2=80=94=20begin=5Frun,=20TraceStreamer,=20TapTrace?= =?UTF-8?q?Writer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 4 +- crates/aura-registry/src/trace_store.rs | 346 ++++++++++++++++++++++-- 2 files changed, 332 insertions(+), 18 deletions(-) diff --git a/crates/aura-registry/src/lib.rs b/crates/aura-registry/src/lib.rs index c2caafb..4fe79bf 100644 --- a/crates/aura-registry/src/lib.rs +++ b/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`. diff --git a/crates/aura-registry/src/trace_store.rs b/crates/aura-registry/src/trace_store.rs index 01e14df..43613fb 100644 --- a/crates/aura-registry/src/trace_store.rs +++ b/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)); - let body = serde_json::to_string(tap).map_err(|source| TraceStoreError::Parse { - file: path.display().to_string(), - source, - })?; - fs::write(&path, body)?; + write_tap_file(&run_dir, tap)?; } - let index = Index { - manifest: manifest.clone(), - taps: taps.iter().map(|t| t.tap.clone()).collect(), - }; - 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(()) + let order: Vec = taps.iter().map(|t| t.tap.clone()).collect(); + write_index(&run_dir, manifest, &order) + } + + /// Begin a streamed run write under `traces//` (directories created + /// now; `index.json` only at [`TraceStreamer::finish`] — the crash + /// discipline `read` already honors: no index → `NotFound`). + pub fn begin_run(&self, name: &str) -> Result { + let run_dir = self.dir.join(name); + fs::create_dir_all(&run_dir)?; + 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 `.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 { + 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, + col_w: BufWriter, + 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 = 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)> = 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); + } }