docs: narrow streaming residency claim from whole-process RSS to the source ring

The C12 realization and the 0041 spec claimed streaming residency is O(one chunk) at the process level ("a 20-year window streams in the same memory as a one-day one"). That holds only for the aura source ring (M1FieldSource::resident_records(), bounded by one chunk and correctly tested); whole-process RSS grows O(records-touched) because data-server's FileCache retains each window's parsed chunks (~56 B/record) read-only for the pass. Reproduced: 6 MiB @ 1mo -> 173 MiB @ ~10y, linear.

That retention is the replay-many optimization (load-once; close+volume share one parse via the field-agnostic FileKey), not a leak, and no always-on eviction can bound a single forward pass without regressing it. This narrows the docs/comments to the true per-source-ring property and names the FileCache per-window retention as the real, replay-amortized process cost. The streaming_seam assert is unchanged (it was always the ring bound; only its labelling confused ring vs process). The deferred opt-in non-retaining streaming read path is parked as Brummel/data-server#2.

closes #95
This commit is contained in:
2026-06-17 22:09:32 +02:00
parent dee25afcd5
commit 0d7c2c16f4
4 changed files with 32 additions and 13 deletions
+7 -2
View File
@@ -181,7 +181,10 @@ fn decode(field: M1Field, bar: &M1Parsed) -> (Timestamp, Scalar) {
/// A streaming [`Source`](aura_engine::Source) over a data-server M1 window. Holds /// A streaming [`Source`](aura_engine::Source) over a data-server M1 window. Holds
/// at most one `Arc` chunk (a zero-copy clone of the cache's chunk) and a cursor; /// at most one `Arc` chunk (a zero-copy clone of the cache's chunk) and a cursor;
/// constructs each `Scalar` per-pull; refills via `next_chunk()` when the chunk /// constructs each `Scalar` per-pull; refills via `next_chunk()` when the chunk
/// drains. Resident footprint is O(one chunk), independent of window length. /// drains. The **source**'s resident footprint is O(one chunk), independent of
/// window length (see [`resident_records`](Self::resident_records)) — a per-source
/// bound, not whole-process RSS: the data-server cache below retains the loaded
/// window's parsed chunks for the pass (the replay-many sharing model, #95).
pub struct M1FieldSource { pub struct M1FieldSource {
iter: SymbolChunkIter<M1Parsed>, iter: SymbolChunkIter<M1Parsed>,
chunk: Option<Arc<[M1Parsed]>>, chunk: Option<Arc<[M1Parsed]>>,
@@ -237,7 +240,9 @@ impl M1FieldSource {
/// Records resident in this source right now: the current chunk's length (or /// 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 /// 0 at exhaustion). The residency probe — bounded by one chunk length by
/// construction (the source holds at most one `Arc` chunk and no accumulating /// construction (the source holds at most one `Arc` chunk and no accumulating
/// field), so it can never grow with window length. /// field), so it can never grow with window length. This probes the **source
/// ring** only; the data-server `FileCache` below retains the loaded window's
/// parsed chunks for the pass and is not counted here (#95).
pub fn resident_records(&self) -> usize { pub fn resident_records(&self) -> usize {
self.chunk.as_ref().map_or(0, |c| c.len()) self.chunk.as_ref().map_or(0, |c| c.len())
} }
+9 -2
View File
@@ -130,6 +130,12 @@ fn streaming_close_source_backtests_end_to_end_deterministically() {
assert_eq!(r1.to_json(), r2.to_json()); assert_eq!(r1.to_json(), r2.to_json());
} }
/// Measures the **aura source ring** residency — `M1FieldSource::resident_records()`,
/// the single chunk the source holds — NOT whole-process RSS. The ring is bounded by
/// one chunk at every window length (the assert below). The data-server `FileCache`
/// beneath retains the loaded window's parsed chunks for the pass (the replay-many
/// sharing model), so process RSS grows O(records-touched); that gap is by design,
/// not a leak, and is tracked as #95.
#[test] #[test]
fn residency_is_bounded_by_one_chunk_independent_of_window_length() { fn residency_is_bounded_by_one_chunk_independent_of_window_length() {
let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
@@ -157,8 +163,9 @@ fn residency_is_bounded_by_one_chunk_independent_of_window_length() {
// multi-chunk window: more records than a single chunk holds, so the bound // multi-chunk window: more records than a single chunk holds, so the bound
// below is non-vacuous (a O(window) source would have peaked at `total`). // below is non-vacuous (a O(window) source would have peaked at `total`).
assert!(total > CHUNK_SIZE, "window must span multiple chunks (got {total} records)"); assert!(total > CHUNK_SIZE, "window must span multiple chunks (got {total} records)");
// O(one chunk), NOT O(window): the per-pull ceiling never exceeds one chunk, // O(one chunk) source ring, NOT O(window): the per-pull ceiling never exceeds
// regardless of how many chunks the window spans. // one chunk, regardless of how many chunks the window spans (whole-process RSS
// is a separate quantity — see this test's doc comment and #95).
assert!(peak <= CHUNK_SIZE, "resident records {peak} exceeded one chunk ({CHUNK_SIZE})"); assert!(peak <= CHUNK_SIZE, "resident records {peak} exceeded one chunk ({CHUNK_SIZE})");
assert!(peak > 0, "a non-empty window resided at least one record"); assert!(peak > 0, "a non-empty window resided at least one record");
} }
+8 -4
View File
@@ -561,10 +561,14 @@ only path. `Harness::run` is re-typed to a **producer seam** — a `Source` trai
(`peek`/`next`, object-safe) the k-way merge drives — and a streaming (`peek`/`next`, object-safe) the k-way merge drives — and a streaming
**`M1FieldSource`** (`aura-ingest`) pulls a data-server window lazily, borrowing **`M1FieldSource`** (`aura-ingest`) pulls a data-server window lazily, borrowing
**one** `Arc<[M1Parsed]>` chunk per pull (zero-copy *within* a source) and **one** `Arc<[M1Parsed]>` chunk per pull (zero-copy *within* a source) and
decoding each `Scalar` on demand: resident O(one chunk), not O(window length), decoding each `Scalar` on demand: the **source ring** is resident O(one chunk),
so a 20-year window streams in the same memory as a one-day one. The eager not O(window length) — the measured `resident_records()` predicate, a *per-source*
`load_m1_window`/`close_stream` path is kept for bounded loads (the gap closes by bound, not whole-process RSS. (Data-server's `FileCache` retains each window's
a streaming path *existing*, not by deleting the eager one). **Still open:** parsed chunks read-only for the pass — ~56 B/record — so process residency is
O(records-touched); that is the **replay-many** sharing C12 wants — one window
parsed once across a sweep family — not a leak. The single-pass cost is tracked as
#95.) The eager `load_m1_window`/`close_stream` path is kept for bounded loads (the
gap closes by a streaming path *existing*, not by deleting the eager one). **Still open:**
cross-*sim* `Arc<[T]>` sharing — one window shared zero-copy across many disjoint cross-*sim* `Arc<[T]>` sharing — one window shared zero-copy across many disjoint
sweep sims — has no consumer until the orchestration families (axes 24 above: sweep sims — has no consumer until the orchestration families (axes 24 above:
#66/#68/#69) are built, and remains the target for those cycles. #66/#68/#69) are built, and remains the target for those cycles.
+8 -5
View File
@@ -102,9 +102,10 @@ let mut h = Harness::bootstrap(sma_cross_signal_quality_graph())?;
let close = M1FieldSource::open(&server, "AAPL.US", Some(from_ms), Some(to_ms), M1Field::Close) let close = M1FieldSource::open(&server, "AAPL.US", Some(from_ms), Some(to_ms), M1Field::Close)
.expect("window overlaps real data"); .expect("window overlaps real data");
// Drive the seam. The source streams Arc<[M1Parsed]> chunks lazily: resident // Drive the seam. The source streams Arc<[M1Parsed]> chunks lazily: the source's
// footprint is one chunk + the engine's per-node lookback windows, NOT the // resident footprint is one chunk + the engine's per-node lookback windows, NOT
// whole window. A 20-year window streams in the same memory as a one-day one. // the whole window — a per-source bound (the data-server cache below still retains
// the loaded window for the pass; see "Data-server cache file retention" + #95).
h.run(vec![Box::new(close)]); h.run(vec![Box::new(close)]);
let report = fold_recorded_into_report(/* sinks */); let report = fold_recorded_into_report(/* sinks */);
@@ -433,8 +434,10 @@ buffers (already O(Σ lookbacks), unchanged). Nothing scales with window length.
O(window length), measured on real multi-chunk data (Testing strategy §4). O(window length), measured on real multi-chunk data (Testing strategy §4).
The struct carries no accumulating field, so this bound holds by The struct carries no accumulating field, so this bound holds by
construction (§5). The engine's per-node lookback ring buffers are construction (§5). The engine's per-node lookback ring buffers are
unchanged — already O(Σ lookbacks), pinned by the existing suite — so total unchanged — already O(Σ lookbacks), pinned by the existing suite. (This is the
resident memory is flat across the horizon. *source-ring* footprint; whole-process RSS is **not** flat across the horizon
the data-server cache retains the loaded window for the pass, see "Data-server
cache file retention" below + #95.)
- [ ] `cargo clippy --workspace --all-targets -- -D warnings` clean; - [ ] `cargo clippy --workspace --all-targets -- -D warnings` clean;
`cargo doc --workspace --no-deps` clean. `cargo doc --workspace --no-deps` clean.