audit(0070): cycle close — lift finalize lifecycle to ledger C8, drop ephemera
Cycle 0070 (streaming sink reductions, #138) close. Durable rationale lifted to the ledger: C8 now records the end-of-stream `finalize` lifecycle (the second node phase beside `eval`), its `Harness::run` topo-order epilogue, and its C1/C7/C8 reconciliation — with the open `Recorder`->`Probe` / per-cycle-allocation question (#77) cross-referenced, since `finalize` is the "non-channel exit" that issue's accumulate-then-read option named as missing. The per-cycle spec and plan are git-rm'd (ephemeral, valid only for their cycle). Audit (architect) found no invariant breakage; noted debt, left as-is: the two-mode `stage1_r_graph` (reduce vs raw sink wiring) rests on the CLI-seam equivalence test rather than structural unification, and `SeriesReducer.last_ts` carries the last *warm* cycle's ts (latent only if `--trace` ever folds). refs #138
This commit is contained in:
@@ -275,6 +275,26 @@ out-of-graph side effect. In-graph routing stays engine-owned data (the edge tab
|
||||
of the graph is the node's own side effect — and that boundary is the
|
||||
determinism / graph-as-data boundary (C1/C7).
|
||||
|
||||
**Refinement (cycle 0070 — end-of-stream `finalize` lifecycle, 2026-06-25).** The
|
||||
node contract gains a second lifecycle phase beside `eval`: `finalize(&mut self)`,
|
||||
a **default-no-op** trait method the engine calls **once per node, in topological
|
||||
order, after the source loop drains** (the `Harness::run` epilogue). It lets a sink
|
||||
**fold online** — accumulate per `eval` into O(trades)/O(1) owned state and flush
|
||||
one compact summary at stream end — instead of pushing a record every fired cycle
|
||||
into an unbounded channel that buffers O(cycles) rows until the run ends (the
|
||||
retention BLOCKER #138 hit: a full-history Stage-1 sweep held ~2 GiB/member over
|
||||
~5.5M one-minute bars). C1/C7/C8 are preserved: `finalize` runs once *after* the
|
||||
deterministic event loop, adding no within-sim concurrency (C1); a folding sink
|
||||
holds only owned accumulator state, no interior mutability (C7); it still declares
|
||||
`output: vec![]` and its flush is the same out-of-graph side effect (C8) — one
|
||||
summary row, not a per-cycle stream. Realized by `GatedRecorder` (emits only gated
|
||||
rows + the genuine final row) and `SeriesReducer` (folds one column, emits one
|
||||
summary row), aura-std siblings of the per-cycle `Recorder`, which survives for the
|
||||
live / `--trace` / test-tap path. The recording sink's *own* per-cycle allocation
|
||||
(the `Recorder`→`Probe` rename + its accumulate-vs-stream choice, #77) stays open;
|
||||
`finalize` is the "non-channel exit from the graph" that issue's
|
||||
accumulate-then-read option named as missing.
|
||||
|
||||
**Refinement (Construction-layer milestone — render labels, 2026-06-05).** A node
|
||||
additionally exposes `label() -> String`, a **single-line, non-load-bearing**
|
||||
render symbol: a default trait method the run loop never calls (wiring is by
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,301 +0,0 @@
|
||||
# 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<Scalar>)`) — 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<f64>, // 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<ScalarKind>,
|
||||
gate_col: usize, // emit a row only when row[gate_col] is true
|
||||
last: Option<(Timestamp, Vec<Scalar>)>, // the genuine final row
|
||||
last_emitted: bool, // was `last` already emitted via the gate?
|
||||
tx: Sender<(Timestamp, Vec<Scalar>)>,
|
||||
}
|
||||
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<Scalar>)> }
|
||||
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.
|
||||
Reference in New Issue
Block a user