From 55dee38d224ea934a38fc02a0c2e962faa76e973 Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 25 Jun 2026 10:01:52 +0200 Subject: [PATCH] spec: 0070 streaming sink reductions (boss-signed) M3 fix for BLOCKER #138: an end-of-stream `finalize` lifecycle on sinks lets reductions fold online into O(trades)+O(1) state, so full-history stage1-r sweeps run in bounded memory per member instead of OOM-killing. Grounding-check PASS; signed autonomously under /boss. refs #138 --- docs/specs/0070-streaming-sink-reductions.md | 301 +++++++++++++++++++ 1 file changed, 301 insertions(+) create mode 100644 docs/specs/0070-streaming-sink-reductions.md diff --git a/docs/specs/0070-streaming-sink-reductions.md b/docs/specs/0070-streaming-sink-reductions.md new file mode 100644 index 0000000..1f6d64a --- /dev/null +++ b/docs/specs/0070-streaming-sink-reductions.md @@ -0,0 +1,301 @@ +# Streaming Sink Reductions (BLOCKER #138) — Design Spec + +**Date:** 2026-06-25 +**Status:** Draft — awaiting user spec review +**Authors:** orchestrator + Claude + +## Goal + +A full-history `stage1-r` parameter sweep must run in **bounded memory per +member** — O(trades)+O(1), independent of the cycle count — so that +multi-member full-history sweeps (and, downstream, walk-forward, #139) no +longer exhaust RAM+swap and SIGKILL-137. + +Today each swept member holds **~2 GiB** over ~5.5M one-minute cycles. The +dominant retainer is the unbounded `mpsc` channel each `Recorder` sink fills +**during** `Harness::run` (one row per fired cycle, nothing draining), plus a +second O(cycles) copy when `run_one` `.collect()`s it. The dense R-record +(14 columns × ~5.5M cycles ≈ 1.4 GB) dominates; the equity/exposure taps add +~0.5 GB; the `r_equity` tap is collected then **discarded** in the no-trace +sweep — pure waste. + +This is **M3** (the mechanism the user chose for #138): an end-of-stream +`finalize` lifecycle on sinks lets a sink **fold online into O(trades)+O(1) +state and flush a compact summary once**, instead of streaming every per-cycle +row out to a channel that buffers them all. + +Out of scope: the realistic-broker / cost layer (Stage 2); the +`--trace` persistence path (it writes full per-cycle streams to disk — +inherently O(cycles), kept as today, an explicit opt-in overlay, not the +BLOCKER); the SMA sample and `walkforward`/`mc` CLI paths (untouched here; +#139 builds on this). + +## Architecture + +Three layers, smallest-blast-radius first: + +1. **`aura-core` — the `finalize` lifecycle hook + a shared online fold.** + - `Node::finalize(&mut self) {}` — an **additive default-no-op** method on + the `Node` trait. Every existing node keeps compiling untouched; only a + folding sink overrides it. Object-safe (`&mut self`, no generics), like + `eval`. + - `SeriesFold` — a single-series online accumulator producing exactly the + three pip statistics `summarize` computes (`last`, `max_drawdown`, + `sign_flips`). One arithmetic implementation, in the crate **both** + `aura-engine` (`summarize`) and `aura-std` (the pip reducer sink) already + depend on — so there is **no dependency inversion** and the folded path is + **byte-identical to `summarize` by construction**, not merely by a pinned + test. + +2. **`aura-engine` — call `finalize` at end-of-stream; drive the shared fold.** + - `Harness::run` gains a **post-loop epilogue**: after the source loop + breaks (all sources exhausted), call `finalize()` on every node once, in + topological order. This runs **after** the deterministic event loop, so it + adds **no within-sim concurrency** (C1 intact). + - `summarize` is refactored to drive `SeriesFold` (one instance over the + equity series for `last`+`max_drawdown`, one over the exposure series for + `sign_flips`), keeping its public signature and its result byte-identical. + `summarize_r` is **unchanged**. + +3. **`aura-std` — two folding sink nodes (siblings of `Recorder`).** Both emit + the **same channel type** as `Recorder` (`(Timestamp, Vec)`) — no + metric types cross into `aura-std` — just compacted: + - `GatedRecorder` — a `Recorder` that emits a row only when a designated + **gate column** (a `bool`) is true, and **always** emits the genuine final + row once on `finalize` (deduped against the last gated emit). Fed the dense + R-record gated on the `closed` column, it yields the **O(trades)** closed + rows + the final row — exactly the rows `summarize_r` extracts — so + `summarize_r` consumes them **verbatim** (zero golden risk). + - `SeriesReducer` — a sink over one `f64` column that folds a `SeriesFold` + per `eval` and, on `finalize`, emits **one** row + `[last, max_drawdown, sign_flips]`. Used twice (equity, exposure); the + consumer reads `last`+`max_drawdown` from the equity instance and + `sign_flips` from the exposure instance. + +4. **`aura-cli` — wire the reducers; drain O(trades) instead of O(cycles).** + - `stage1_r_graph` gains a `reduce: bool`. When `reduce` (the no-trace sweep, + the BLOCKER path): wire `GatedRecorder` (R) + two `SeriesReducer`s + (equity, exposure); **omit** the `r_equity` tap entirely (it is unused + without `--trace`). When `!reduce` (today's `--trace` path and the single + `run_stage1_r`): wire the raw `Recorder`s exactly as today. + - The no-trace `run_one` builds `RunReport` from the folded rows: + `summarize_r(&gated_r_rows)` for the R block; the two single-summary rows + for the pip block. The `--trace` branch and `run_stage1_r` are **unchanged** + (they keep the raw recorders + `summarize`/`summarize_r` over full rows — + the byte-for-byte reference the equivalence test compares against). + +The single feedback-free invariant set is preserved: C1 (finalize is a +post-loop, single-threaded pass), C7 (the sinks hold only owned accumulator +state + an owned `Sender` — no `Rc`/`RefCell`, like `Recorder`), C8 (the sinks +declare empty output and emit nothing downstream). + +## Concrete code shapes + +### User-facing evidence (infra cycle — the invocation that must stop OOM-ing) + +```console +$ aura sweep --strategy stage1-r --real EURUSD \ + --fast 240,480,1920 --slow 1920,7680 --stop-length 1920 --stop-k 2.0 +# 6-member grid over ~5.5M one-minute cycles, no --trace. +# BEFORE: peak RSS ≈ members × ~2 GiB → swap exhaustion → SIGKILL (137). +# AFTER: peak RSS ≈ members × O(trades) → completes; metrics byte-identical. +``` + +The cycle delivers no new user surface; its empirical evidence is (a) the +must-fail memory fixture below and (b) the equivalence fixture proving the +metrics are unchanged. + +### The lifecycle hook (aura-core) + +```rust +// crates/aura-core/src/node.rs — additive, default no-op +trait Node { + fn eval(&mut self, ctx: Ctx) -> Option<&[Cell]>; // unchanged + // NEW: end-of-stream flush. default no-op → existing nodes untouched. + fn finalize(&mut self) {} +} +``` + +### The shared online fold (aura-core) — single source of truth + +```rust +// crates/aura-core/src/... — the exact three folds `summarize` performs today +struct SeriesFold { + last: f64, // total_pips = last value seen (0.0 if none) + peak: f64, // running max, seeded NEG_INFINITY (as summarize) + max_drawdown: f64, // running max(peak - v), >= 0.0 + prev_sign: Option, // sign0 of the previous value + sign_flips: u64, +} +impl SeriesFold { + fn fold(&mut self, v: f64) { + // operations and order kept identical to summarize's loops, so the + // result is bit-for-bit equal (IEEE-754 non-associativity respected). + self.last = v; + if v > self.peak { self.peak = v; } + let dd = self.peak - v; + if dd > self.max_drawdown { self.max_drawdown = dd; } + let s = sign0(v); + if let Some(p) = self.prev_sign { if s != p { self.sign_flips += 1; } } + self.prev_sign = Some(s); + } +} +``` + +```rust +// crates/aura-engine/src/report.rs — summarize now drives SeriesFold, signature unchanged +pub fn summarize(equity: &[(Timestamp, f64)], exposure: &[(Timestamp, f64)]) -> RunMetrics { + let mut eqf = SeriesFold::new(); + for &(_, v) in equity { eqf.fold(v); } // total_pips, max_drawdown + let mut exf = SeriesFold::new(); + for &(_, v) in exposure { exf.fold(v); } // sign_flips + RunMetrics { total_pips: eqf.last, max_drawdown: eqf.max_drawdown, + bias_sign_flips: exf.sign_flips, r: None } +} +// summarize_r: UNCHANGED. +``` + +### The folding sinks (aura-std) + +```rust +// crates/aura-std/src/gated_recorder.rs — Recorder that filters to O(trades) rows +struct GatedRecorder { + kinds: Vec, + gate_col: usize, // emit a row only when row[gate_col] is true + last: Option<(Timestamp, Vec)>, // the genuine final row + last_emitted: bool, // was `last` already emitted via the gate? + tx: Sender<(Timestamp, Vec)>, +} +impl Node for GatedRecorder { + fn eval(&mut self, ctx: Ctx) -> Option<&[Cell]> { + let row = read_declared_columns(ctx)?; // None during warm-up, as Recorder + let gated = row[self.gate_col].as_bool(); + if gated { let _ = self.tx.send((ctx.now(), row.clone())); } + self.last = Some((ctx.now(), row)); + self.last_emitted = gated; + None // sink: emits nothing downstream (C8) + } + fn finalize(&mut self) { + // ensure summarize_r's `record.last()` sees the true final row, without + // double-counting a final row that was already emitted as a closed trade. + if !self.last_emitted { + if let Some(row) = self.last.take() { let _ = self.tx.send(row); } + } + } +} +``` + +```rust +// crates/aura-std/src/series_reducer.rs — folds one f64 series, flushes once +struct SeriesReducer { fold: SeriesFold, warm: bool, tx: Sender<(Timestamp, Vec)> } +impl Node for SeriesReducer { + fn eval(&mut self, ctx: Ctx) -> Option<&[Cell]> { + let v = ctx.f64_in(0).get(0)?; // None during warm-up + self.fold.fold(v); + self.warm = true; + None + } + fn finalize(&mut self) { + // one summary row; ts is the final cycle's — carried for join symmetry, unused by the consumer. + let row = vec![Scalar::f64(self.fold.last), + Scalar::f64(self.fold.max_drawdown), + Scalar::i64(self.fold.sign_flips as i64)]; + let _ = self.tx.send((self.final_ts(), row)); + } +} +``` + +### The run loop epilogue (aura-engine) and the folded consumer (aura-cli) + +```rust +// crates/aura-engine/src/harness.rs ~371 — after `loop { ... }` breaks: +for &nidx in topo.iter() { nodes[nidx].node.finalize(); } +``` + +```rust +// crates/aura-cli/src/main.rs — no-trace run_one (the BLOCKER path) +h.run(data.run_sources()); // channels now fill only at finalize +let r_rows: Vec<_> = rx_r.try_iter().collect(); // O(trades): cheap +let eq = rx_eq.try_iter().next().unwrap_or(empty_summary()); // one folded row +let ex = rx_ex.try_iter().next().unwrap_or(empty_summary()); +let mut metrics = RunMetrics { + total_pips: eq.1[0].as_f64(), max_drawdown: eq.1[1].as_f64(), + bias_sign_flips: ex.1[2].as_i64() as u64, r: None, +}; +metrics.r = Some(summarize_r(&r_rows, 0.0)); // summarize_r unchanged +``` + +## Components + +| Component | Crate | Change | +|-----------|-------|--------| +| `Node::finalize` | aura-core | NEW default-no-op trait method | +| `SeriesFold` | aura-core | NEW online accumulator (the three pip folds) | +| `Harness::run` | aura-engine | post-loop `finalize()` epilogue (topo order) | +| `summarize` | aura-engine | refactor to drive `SeriesFold`; signature + result unchanged | +| `summarize_r` | aura-engine | UNCHANGED | +| `GatedRecorder` | aura-std | NEW filtering sink (gate column + final-row flush) | +| `SeriesReducer` | aura-std | NEW single-series folding sink | +| `Recorder` | aura-std | UNCHANGED (kept for `--trace` + SMA + tests) | +| `stage1_r_graph` | aura-cli | `reduce: bool`; wire reducers / omit `r_equity` when reducing | +| `run_one` (sweep) | aura-cli | no-trace branch builds `RunReport` from folded rows | +| `run_stage1_r`, `--trace` branch | aura-cli | UNCHANGED (reference path) | + +## Data flow + +`reduce` (no-trace sweep): +`price → graph → {GatedRecorder(R, gate=closed), SeriesReducer(equity), +SeriesReducer(exposure)}`. During `run`: per-cycle folds, nothing buffered +beyond O(trades) closed rows. At `finalize`: GatedRecorder flushes the final +row; each SeriesReducer flushes one summary row. `run_one` drains O(trades)+2 +rows → `summarize_r` + direct `RunMetrics` build. + +`!reduce` (`--trace`, single run): unchanged — raw `Recorder`s → +`.collect()` → `persist_traces_r` (trace) + `summarize`/`summarize_r`. + +## Error handling + +No new failure modes. A reducer sink whose declared column is missing/wrong +kind panics at the same "checked at wiring" boundary `Recorder`/`f64_field` use +today. `try_iter().next()` on an empty folded channel (a run with zero warm +cycles) falls back to the zero-summary, matching `summarize` on empty input +(`total_pips=0.0`, `max_drawdown=0.0`, `flips=0`). + +## Testing strategy + +- **End-to-end RED (the BLOCKER pin).** A counting-`#[global_allocator]` + integration test driving a real `Harness` + the folding sinks over a + synthetic stream with a FIXED small number of trades but a VARIABLE large + cycle count (e.g. 520 vs ~400k cycles): assert peak retained bytes through + the **recorder → channel → reduction** path does **not** scale with cycle + count (O(trades)+O(1)). Fails on today's `Recorder`+`.collect()` path; passes + with the folding sinks. (Reference scaffolding preserved at + `scratchpad/r_reduction_streaming.red-reference.rs` — but it must exercise + the channel path end-to-end, not `summarize_r` in isolation.) +- **Golden equivalence (byte-for-byte).** On the same synthetic + a real + fixture window, assert the folded `RunReport` (R block + pip block) equals + the raw-recorder `summarize`/`summarize_r` `RunReport` **field-for-field**. + This is the proof the folded path changed nothing observable. +- **`finalize` lifecycle unit tests** (aura-engine): `run` calls `finalize` + once per node after the stream ends, in topo order; a default node's + `finalize` is a no-op. +- **`GatedRecorder` unit tests** (aura-std): emits only gated rows; emits the + final row on `finalize`; does **not** double-emit when the final row was + gated-true. +- **`SeriesReducer` unit tests** (aura-std): folds match `SeriesFold`; one row + on `finalize`; nothing per-eval. +- **Existing goldens** (R metrics, pip metrics, sweep family, single run, + `--trace`): all must stay green and byte-identical. + +## Acceptance criteria + +1. A no-trace `stage1-r` sweep's **peak retained memory per member is + O(trades)+O(1)**, independent of cycle count (the RED test passes). +2. Every existing metric value (R and pip, single run and sweep) is + **byte-identical** (the equivalence test + all existing goldens pass). +3. `Node::finalize` is additive and default-no-op; no existing node is edited + to satisfy the compiler. +4. C1 (no within-sim concurrency), C7 (owned accumulator state, no interior + mutability), C8 (sinks emit nothing) are preserved. +5. `cargo test --workspace` and `cargo clippy --workspace --all-targets -- -D + warnings` are green.