From ae240140c403deaa2b7ed29f04f7c4d68706d562 Mon Sep 17 00:00:00 2001 From: Brummel Date: Mon, 15 Jun 2026 10:00:44 +0200 Subject: [PATCH] spec: 0041 source ingestion seam (boss-signed) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-types the engine's ingestion input from a materialized Vec> into a `Source` producer trait the k-way merge drives by peek/next, and proves it against a real, lazily-streamed data-server window in the same cycle — closing the cycle-0011 deliberate eager-materialization gap (C12 Arc<[T]> cross-sim sharing). A VecSource adapter keeps every existing run call site byte-identical (behaviour-preserving); an M1FieldSource streams data-server chunks with O(one-chunk) residency, independent of window length. Foundational cycle of the World-II milestone; #66/#68/#69 inherit the seam. Signed under the boss auto-sign gate: precondition clean, grounding PASS on the final bytes, unanimous five-lens spec-skeptic panel (two editorial ambiguity rounds on the residency predicate repaired to a measured per-pull CHUNK_SIZE ceiling). refs #71 --- docs/specs/0041-source-ingestion-seam.md | 459 +++++++++++++++++++++++ 1 file changed, 459 insertions(+) create mode 100644 docs/specs/0041-source-ingestion-seam.md diff --git a/docs/specs/0041-source-ingestion-seam.md b/docs/specs/0041-source-ingestion-seam.md new file mode 100644 index 0000000..f8c3187 --- /dev/null +++ b/docs/specs/0041-source-ingestion-seam.md @@ -0,0 +1,459 @@ +# Source ingestion seam — Design Spec + +**Date:** 2026-06-15 +**Status:** Draft — awaiting user spec review +**Authors:** orchestrator + Claude + +## Goal + +Replace the engine's single materialized-`Vec` ingestion input with a **producer +seam** — a `Source` trait the k-way merge drives by `peek`/`next` — and prove it +against a real, lazily-streamed `data-server` window in the same cycle, so the +seam is designed against the real data source from the first line. + +`Harness::run` today takes `Vec>`: one fully +materialized stream per source. That is the **only** engine type encoding an +eager-dataflow assumption. At realistic research scale (20 years, 3 tick streams, +~7.9e9 ticks) the eager layout is ~190 GB resident (24 B/tick) — non-residable. +The merge loop itself never needs the whole stream: it only ever peeks the head +timestamp of each live source and pops one record (`harness.rs:237`–`257`). So +the eager `Vec` is gratuitous; a `peek`/`next` producer is the faithful seam, and +a `data-server`-backed producer streams `Arc<[T]>` chunks lazily — resident +footprint O(Σ node lookbacks + one chunk), flat across the horizon. + +This is the foundational cycle of the World-II milestone: the orchestration +families (#66 / #68 / #69) all bootstrap harnesses over data windows, so they +inherit this seam. Ledger cycle 0011 (`INDEX.md:530`) already names today's eager +materialization "a known, deliberate gap" whose zero-copy `Arc<[T]>` cross-sim +sharing is "the target for the orchestration cycle that introduces [the axes]" — +this milestone. This cycle closes that gap by making a streaming path exist; it +does not delete the eager convenience path (still valid for bounded loads). + +**Contracts touched:** C2 (bounded read-only windows), C3 (one merge at +ingestion, ms→epoch-ns normalized there), C4 (cycle granularity / tie-by-source- +index), C7 (SoA base columns), C12 (`Arc<[T]>` cross-sim sharing). + +## Architecture + +Three layers, one new boundary type: + +1. **`aura-engine`** gains a public `Source` trait and a `VecSource` adapter. + `Harness::run` is re-typed from `Vec>` to + `Vec>`. The merge loop's index arithmetic + (`streams[s][cursor[s]]`) becomes `sources[s].peek()` / `sources[s].next()`. + Every existing call site wraps its `Vec<(Timestamp, Scalar)>` in + `VecSource::new(..)` — behaviour-preserving, byte-identical run output, no + residency change. `VecSource` *is* the old behaviour, named. + +2. **`aura-ingest`** gains `M1FieldSource`: a streaming `Source` that holds a + `data-server` `SymbolChunkIter`, the current `Arc<[M1Parsed]>` chunk + (an `Arc` clone of the cache's chunk — zero-copy), a cursor into it, a field + selector, and a **pre-decoded head** record. It constructs each `Scalar` + per-pull from the borrowed chunk (no owned-`Vec` materialization of the + window) and refills via `next_chunk()` when the current chunk drains. + +3. **End-to-end**: a gated integration test bootstraps the signal-quality sample + harness, opens an `M1FieldSource` over a real `[from, to]` window, runs it + through the seam to a `RunReport`, and asserts determinism (C1). + +The `peek(&self)`/`next(&mut self)` split is what forces the pre-decoded head in +the streaming source: answering `peek` without `&mut` is impossible over an +underlying iterator whose refill (`next_chunk`) is `&mut`, so the head is decoded +ahead of demand at construction and after every `next`. This is the standard +peekable shape, determined by the trait signature (settled in the issue), not an +open design choice. + +`SourceSpec` (`harness.rs:49`) is unchanged: it stays the *wiring* metadata +(scalar `kind` + target slots), declared on the `FlatGraph`. A `Box` +is the *runtime* producer bound to source index `s` at `run` time. The two were +already distinct (the old `Vec` was the runtime half); this cycle only re-types +the runtime half from an owned `Vec` to a producer. + +### Scope boundary + +In scope: the trait, `VecSource`, the `run` re-type with every call site adapted, +`M1FieldSource` (the streaming source), and one streaming end-to-end test. + +Out of scope (deliberately, to keep the seam focused): +- Deleting the eager `load_m1_window` / `close_stream` path — it stays a valid + convenience for bounded loads; the CLI keeps using it (wrapped in `VecSource`). + The deliberate-eager gap is closed by a streaming path *existing*, not by + removing the eager one. +- Migrating the CLI's real-data commands to streaming — follow-on. +- Tick sources (`stream_tick_windowed`) — the same shape applies (`M1FieldSource` + generalizes), but the e2e and the SMA-cross sample consume M1 close; tick is + not built here. +- Seed-as-input (#66) — the seed lives at the recording edge, outside this seam. + +## Concrete code shapes + +### User-facing program — a streaming real-window backtest + +What a project author (or a World program) writes to backtest a strategy over a +real data window. This is the acceptance criterion's empirical evidence: the +audience's natural reach, shown. + +```rust +// Bootstrap the harness (unchanged — topology is FlatGraph as today). +let mut h = Harness::bootstrap(sma_cross_signal_quality_graph())?; + +// Open a streaming source over a real window. None = no archived file overlaps +// the window (data-server's file-level Option contract, propagated). +let close = M1FieldSource::open(&server, "AAPL.US", Some(from_ms), Some(to_ms), M1Field::Close) + .expect("window overlaps real data"); + +// Drive the seam. The source streams Arc<[M1Parsed]> chunks lazily: resident +// footprint is one chunk + the engine's per-node lookback windows, NOT the +// whole window. A 20-year window streams in the same memory as a one-day one. +h.run(vec![Box::new(close)]); + +let report = fold_recorded_into_report(/* sinks */); +``` + +The eager path still works unchanged for bounded loads (every existing call +site, now wrapped): + +```rust +let prices: Vec<(Timestamp, Scalar)> = + load_m1_window(&server, "AAPL.US", Some(from_ms), Some(to_ms))?.close_stream(); +h.run(vec![Box::new(VecSource::new(prices))]); // was: h.run(vec![prices]); +``` + +### Delivered code — the `Source` trait (aura-engine) + +```rust +/// The ingestion-boundary producer the k-way merge drives (C3). A source yields +/// timestamped scalars in ascending timestamp order (the C3 ingestion +/// precondition). Object-safe: the merge holds `Vec>`. +pub trait Source { + /// The timestamp of the head record, without consuming it. `None` = the + /// source is exhausted. Cheap and side-effect-free: it reads a pre-decoded + /// head, never advances the underlying stream. + fn peek(&self) -> Option; + + /// Pop the head `(timestamp, scalar)` and advance. `None` = exhausted. + /// After `next` returns `Some`, `peek` reflects the new head. + fn next(&mut self) -> Option<(Timestamp, Scalar)>; +} +``` + +### Delivered code — `VecSource` (aura-engine), behaviour-preserving + +```rust +/// A `Source` over a fully materialized stream — the cycle-0011 eager shape, +/// named. Every pre-seam call site wraps its `Vec<(Timestamp, Scalar)>` in this, +/// so run output is byte-identical (C1) and residency is unchanged. +pub struct VecSource { + stream: Vec<(Timestamp, Scalar)>, + cursor: usize, +} + +impl VecSource { + pub fn new(stream: Vec<(Timestamp, Scalar)>) -> Self { + Self { stream, cursor: 0 } + } +} + +impl Source for VecSource { + fn peek(&self) -> Option { + self.stream.get(self.cursor).map(|&(t, _)| t) + } + fn next(&mut self) -> Option<(Timestamp, Scalar)> { + let item = self.stream.get(self.cursor).copied()?; // (Timestamp, Scalar): Copy + self.cursor += 1; + Some(item) + } +} +``` + +### `Harness::run` — the merge loop re-type (before → after) + +```rust +// BEFORE (harness.rs:220) — eager Vec input, index arithmetic. +pub fn run(&mut self, streams: Vec>) { + assert_eq!(streams.len(), self.sources.len(), /* ... */); + let mut cursor: Vec = vec![0; streams.len()]; + loop { + let mut pick: Option = None; + for (s, stream) in streams.iter().enumerate() { + if cursor[s] < stream.len() { + match pick { + None => pick = Some(s), + Some(p) => if stream[cursor[s]].0 < streams[p][cursor[p]].0 { pick = Some(s); } + } + } + } + let s = match pick { Some(s) => s, None => break }; + let (ts, value) = streams[s][cursor[s]]; + cursor[s] += 1; + /* ... forward into target slots, eval in topo order ... unchanged ... */ + } +} + +// AFTER — producer seam, peek/next. The pick scan, tie-by-source-index, and the +// entire forward+eval body are preserved; only the head access changes. +pub fn run(&mut self, mut sources: Vec>) { + assert_eq!(sources.len(), self.sources.len(), /* ... */); + loop { + // pick the live source head with the smallest (timestamp, source index). + // strictly-`<` replace + source-order scan preserves C4 tie-breaking. + let mut pick: Option<(usize, Timestamp)> = None; + for (s, src) in sources.iter().enumerate() { + if let Some(ts) = src.peek() { + match pick { + None => pick = Some((s, ts)), + Some((_, best)) if ts < best => pick = Some((s, ts)), + _ => {} + } + } + } + let s = match pick { Some((s, _)) => s, None => break }; + let (ts, value) = sources[s].next().expect("peeked Some ⇒ next is Some"); + /* ... forward into target slots, eval in topo order ... byte-identical ... */ + } +} +``` + +Note the borrow shape: `self.sources` (the `SourceSpec` wiring) is read inside the +forward step while `sources` (the producers) is the separate `run` argument, so +the existing destructuring of `self` into disjoint field borrows is unaffected. + +### Delivered code — `M1FieldSource` (aura-ingest), the streaming source + +```rust +/// Which base column of an M1 bar a source streams (C7: a composite window is a +/// bundle of base columns; one `Source` per consumed field). +pub enum M1Field { Open, High, Low, Close, Spread, Volume } + +/// A streaming `Source` over a data-server M1 window. Holds one Arc chunk +/// (zero-copy clone of the cache's chunk) and a cursor; constructs each `Scalar` +/// per-pull; refills via `next_chunk()` when the chunk drains. Resident footprint +/// is O(one chunk), independent of window length. +pub struct M1FieldSource { + iter: SymbolChunkIter, // data-server's lazy chunk producer + chunk: Option>, // current chunk, borrowed (Arc refcount, not a data copy) + pos: usize, // cursor within `chunk` + field: M1Field, // which column this source projects + head: Option<(Timestamp, Scalar)>, // pre-decoded head ⇒ peek(&self) is cheap +} + +impl M1FieldSource { + /// Open over `[from_ms, to_ms]` (inclusive Unix-ms, data-server's contract). + /// `None` when no archived file overlaps the window (an unknown symbol or a + /// far-future window) — propagating `stream_m1_windowed`'s file-level Option. + /// A window that overlaps a file but holds zero matching bars yields a source + /// whose first `peek` is `None` (immediately exhausted), not `None` here. + pub fn open( + server: &Arc, symbol: &str, + from_ms: Option, to_ms: Option, field: M1Field, + ) -> Option { + let iter = server.stream_m1_windowed(symbol, from_ms, to_ms)?; + let mut s = Self { iter, chunk: None, pos: 0, field, head: None }; + s.advance(); // pre-decode the first head + Some(s) + } + + /// Project the bar at the cursor into (normalized ts, scalar) for `field`. + fn decode(&self, bar: &M1Parsed) -> (Timestamp, Scalar) { + let value = match self.field { + M1Field::Open => Scalar::F64(bar.open), + M1Field::High => Scalar::F64(bar.high), + M1Field::Low => Scalar::F64(bar.low), + M1Field::Close => Scalar::F64(bar.close), + M1Field::Spread => Scalar::F64(bar.spread), + M1Field::Volume => Scalar::I64(bar.volume), + }; + (unix_ms_to_epoch_ns(bar.time_ms), value) // ms → epoch-ns at the seam (C3) + } + + /// Decode the head at the cursor, refilling chunks as needed. Sets + /// `self.head = None` at exhaustion. + fn advance(&mut self) { + loop { + if self.chunk.is_none() { + self.chunk = self.iter.next_chunk(); // &mut refill; None = window exhausted + self.pos = 0; + if self.chunk.is_none() { self.head = None; return; } + } + let chunk = self.chunk.as_ref().unwrap(); + if self.pos < chunk.len() { + self.head = Some(self.decode(&chunk[self.pos])); + return; + } + self.chunk = None; // current chunk drained — loop to refill + } + } + + /// Records resident in this source right now: the current chunk's length (or + /// 0 at exhaustion). The residency probe — bounded by one chunk length by + /// construction (the struct holds at most one `Arc` chunk and no accumulating + /// field), so it can never grow with window length. Sampled by the residency + /// test to make the O(one chunk) claim a measured quantity, not prose. + pub fn resident_records(&self) -> usize { + self.chunk.as_ref().map_or(0, |c| c.len()) + } +} + +impl Source for M1FieldSource { + fn peek(&self) -> Option { + self.head.map(|(t, _)| t) + } + fn next(&mut self) -> Option<(Timestamp, Scalar)> { + let item = self.head?; + self.pos += 1; + self.advance(); // pre-decode the next head (may refill the chunk) + Some(item) + } +} +``` + +### N-field generalization (shape the trait permits; not all-tested here) + +Two `M1FieldSource`s over the same `(symbol, window)` get the **same** `Arc` +chunks from the data-server cache (the `SymbolGuard` keeps the symbol loaded; the +cache shares `Arc<[T]>`), so they stream different fields sharing the chunk's ts +axis zero-copy — C12's cross-source `Arc<[T]>` sharing: + +```rust +let close = M1FieldSource::open(&server, sym, f, t, M1Field::Close)?; +let volume = M1FieldSource::open(&server, sym, f, t, M1Field::Volume)?; +h.run(vec![Box::new(close), Box::new(volume)]); // co-scheduled by timestamp (C4) +``` + +The e2e headline uses the close source alone; a focused test asserts two sources +over one window both stream (the sharing property), but the full OHLCV-as-N- +sources harness is not built in this cycle. + +## Components + +| Component | Crate | Change | +|-----------|-------|--------| +| `Source` trait | aura-engine | **new** — `peek(&self)`, `next(&mut)`; object-safe; re-exported from the crate root | +| `VecSource` | aura-engine | **new** — behaviour-preserving adapter over `Vec<(Timestamp, Scalar)>` | +| `Harness::run` | aura-engine | **re-typed** — `Vec>`; merge loop drives `peek`/`next`; forward+eval body byte-identical | +| ~53 `run` call sites | aura-engine, aura-cli, aura-ingest | **adapted** — wrap streams in `VecSource::new(..)`; mostly tests | +| `M1Field`, `M1FieldSource` | aura-ingest | **new** — streaming M1 source, per-pull decode, lazy chunk refill, ms→ns normalization | +| streaming e2e test | aura-ingest | **new** — real-window backtest through the seam to a `RunReport` | +| `load_m1_window`, `close_stream`, `transpose_m1` | aura-ingest | **kept** — eager convenience for bounded loads; unchanged | + +## Data flow + +``` +data-server files ──stream_m1_windowed(sym, from_ms, to_ms)──▶ SymbolChunkIter + │ next_chunk() : Arc<[M1Parsed]> (lazy, N+2 prefetch) + ▼ +M1FieldSource { chunk: Arc<[M1Parsed]>, pos, field, head } ──decode(bar)──▶ (epoch-ns ts, Scalar) + │ peek() / next() + ▼ +Harness::run(Vec>) ──k-way merge by (peek ts, source idx)──▶ one chronological cycle stream + │ (forward into target slots → eval in topo order — unchanged) + ▼ + recorded sinks → RunReport +``` + +The only resident data at any instant: each source's current `Arc` chunk +(≤1 chunk) + one pre-decoded head, plus the engine's per-node lookback ring +buffers (already O(Σ lookbacks), unchanged). Nothing scales with window length. + +## Error handling + +- **Source open failure** is an `Option`, not a `Result`: `M1FieldSource::open` + returns `None` exactly when `stream_m1_windowed` does — no archived file + overlaps the window (unknown symbol / far-future window). This mirrors the + existing `load_m1_window` file-level Option contract (`lib.rs:101`–`109`). +- **Empty-but-overlapping window** (a file overlaps but holds zero matching + bars): `open` returns `Some`, and the source's first `peek` is `None` — + immediately exhausted. `run` over it drives zero cycles. This preserves the + existing distinction "data source present" vs "nothing to read from". +- **Mid-stream**: `next`/`peek` never error — exhaustion is `None`. `next_chunk` + yields `None` at window end; `advance` maps that to `head = None`. +- **`run`'s length assertion** (one source per `SourceSpec`) is unchanged. +- The ascending-timestamp ingestion precondition (C3) is unchanged: data-server + files are chronological, and `decode` preserves order within and across chunks. + +## Testing strategy + +1. **Behaviour-preserving (the load-bearing guarantee).** The full + `cargo test --workspace` suite stays green and byte-identical after the + re-type — every call site wrapped in `VecSource`. The existing C1-determinism + assertions (e.g. `real_bars.rs:118`–`120`, `blueprint.rs` sweep equality) are + the evidence; no test output changes. +2. **`VecSource` unit tests** (aura-engine): `peek` is non-consuming and idempotent; + `peek` then `next` agree; exhaustion yields `None`; an empty stream peeks `None`. +3. **`decode` projection unit tests** (aura-ingest, hermetic — pure `decode` on + hand-made `M1Parsed`, no data-server files): each `M1Field` projects to the + right `Scalar` kind/value with ts normalized ms→ns. `decode` is a pure + function of `(bar, field)`, so it is fully testable without an iterator or + files. (The `peek`/`next`/refill walk over a real iterator is exercised by the + gated e2e below, which is where a real `SymbolChunkIter` exists.) +4. **Streaming e2e + residency proof** (aura-ingest, gated — skips where the + local Pepperstone directory is absent, like `real_bars.rs`): bootstrap the + signal-quality sample harness, `M1FieldSource::open` over a real window + spanning **many** data-server chunks, `h.run` through the seam, fold to a + `RunReport`; assert `total_pips.is_finite()` and C1 (two runs bit-identical + JSON). **Residency predicate — the measurable O(one chunk) claim:** sampling + `resident_records()` at every pull across the whole run, its maximum is + **≤ one data-server chunk** — the `CHUNK_SIZE` bound (1024 records; + `data_server`'s documented chunk size, asserted against the constant, not a + magic literal). This per-pull bound is **independent of window length**: it + holds at every pull whether the window spans a handful of chunks or millions, + which is the O(one chunk)-not-O(window) content. The bound is correct on real + chunks regardless of boundary-chunk lengths (`filter_chunk` only ever yields a + chunk *≤* `CHUNK_SIZE`). A source that accumulated chunks would blow past the + one-chunk bound on a long window; a source that accumulated decoded state is + excluded structurally by §5. (Deliberately *not* an "equal peak across two + windows" assertion — boundary/file-final chunks are short, so per-window + maxima legitimately differ; the load-bearing invariant is the per-pull + one-chunk *ceiling*, which is what bounds residency.) +5. **Structural no-accumulation guarantee** (always-on, not gated): `M1FieldSource` + holds exactly `{ iter, one Option>, cursor, field, one head }` + — no `Vec>` or other field that grows with the stream — so + `resident_records()` is bounded by one chunk length *by construction*. This is + the always-on half of the residency claim (verifiable from the struct + definition + clippy); the gated item 4 is the dynamic-invariance half. +6. **Sharing property** (gated): two `M1FieldSource`s (close + volume) over one + window both stream to exhaustion in a single `run` — the N-sources-share-the- + window shape. + +## Acceptance criteria + +- [ ] `Source` (`peek` + `next`, object-safe) and `VecSource` exist in + aura-engine; `Harness::run` takes `Vec>`. +- [ ] Every `run` call site is adapted; `cargo test --workspace` is green and + run output is byte-identical to pre-seam (behaviour-preserving). +- [ ] `M1FieldSource` streams a data-server M1 window through the seam: per-pull + `Scalar` decode from a borrowed `Arc` chunk, lazy `next_chunk` refill, + ms→epoch-ns normalized at the seam (C3). +- [ ] A real short-window backtest runs end-to-end through the seam to a + `RunReport`, deterministically (C1). +- [ ] **Residency (measured predicate):** across a full run over a window + spanning many chunks, peak `M1FieldSource::resident_records()` is + **≤ one data-server chunk** (`CHUNK_SIZE` = 1024) — a per-pull ceiling + **independent of window length** — so the source holds O(one chunk), NOT + O(window length), measured on real multi-chunk data (Testing strategy §4). + The struct carries no accumulating field, so this bound holds by + construction (§5). The engine's per-node lookback ring buffers are + unchanged — already O(Σ lookbacks), pinned by the existing suite — so total + resident memory is flat across the horizon. +- [ ] `cargo clippy --workspace --all-targets -- -D warnings` clean; + `cargo doc --workspace --no-deps` clean. + +## Surfaced boundary frictions (recorded, per the issue's friction-first intent) + +- **Transpose granularity.** The streaming source does **per-pull** field decode + from the borrowed chunk, not `transpose_m1`'s per-window owned-SoA transpose. + The eager `M1Columns`/`transpose_m1` path is a different shape, kept for bounded + loads; the two now coexist (a follow-on may unify them). +- **`Arc` borrow lifetime.** Each source holds an `Arc<[M1Parsed]>` clone; across + disjoint sims (C1/C12) the data-server cache shares the same chunks zero-copy. + The `SymbolGuard` inside `SymbolChunkIter` keeps the symbol loaded for the + source's lifetime. +- **Data-server cache file retention.** The aura-side source is O(one chunk); the + data-server *cache* retains loaded files (with N+2 prefetch). Whether files + behind the cursor are evicted is the data-server's contract, **owned by that + lib, not this cycle** — the residency claim here is scoped to aura's source + footprint. Flagged for the milestone, not resolved here. +- **Error channel asymmetry.** `Option` at open (file-level), `None`-on- + exhaustion mid-stream — no `Result` anywhere, matching `stream_m1_windowed`. +- **`peek(&self)` vs `&mut` refill.** Resolved by the pre-decoded head; the head + decode happens in `advance`, called from `open` and `next`.