diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs index 6bb6aa1..acbfd0b 100644 --- a/crates/aura-engine/src/harness.rs +++ b/crates/aura-engine/src/harness.rs @@ -70,6 +70,18 @@ pub trait Source { /// open-ended. Cursor-independent: a property of the data extent, not the /// read position. fn bounds(&self) -> Option<(Timestamp, Timestamp)>; + + /// Records this source holds resident right now — the streaming-ring probe + /// (#95). `None` (the default) means the source does not report a residency: + /// the eager [`VecSource`] materializes its whole stream and has no bounded + /// ring. A lazy chunk source overrides this to return `Some(len)` of the chunk + /// it currently borrows, so a `Box` consumer can budget memory + /// without downcasting. Mirrors [`bounds`](Source::bounds)' Option idiom: + /// `None` = "not reported," distinct from `Some(0)` = "reports, holds none" + /// (e.g. exhausted). Probes the aura source ring only, not whole-process RSS. + fn resident_records(&self) -> Option { + None + } } /// A `Source` over a fully materialized stream — the cycle-0011 eager shape, @@ -493,6 +505,16 @@ mod tests { points.iter().map(|&(t, v)| (Timestamp(t), Scalar::f64(v))).collect() } + #[test] + fn vec_source_reports_no_residency_by_default() { + // The eager VecSource has no streaming ring, so it takes the Source trait + // default: residency is "not reported" (None), distinct from Some(0). The + // probe is reachable through `&dyn Source` without downcasting (#95). + let v = VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0)])); + let probe: &dyn Source = &v; + assert_eq!(probe.resident_records(), None); + } + #[test] fn seeded_source_same_seed_same_stream() { // Same seed -> bit-identical stream (C1/C12 determinism). diff --git a/crates/aura-ingest/src/lib.rs b/crates/aura-ingest/src/lib.rs index a653217..407615c 100644 --- a/crates/aura-ingest/src/lib.rs +++ b/crates/aura-ingest/src/lib.rs @@ -182,7 +182,7 @@ fn decode(field: M1Field, bar: &M1Parsed) -> (Timestamp, Scalar) { /// 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 /// drains. The **source**'s resident footprint is O(one chunk), independent of -/// window length (see [`resident_records`](Self::resident_records)) — a per-source +/// window length (see [`resident_records`](aura_engine::Source::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 { @@ -237,18 +237,20 @@ impl M1FieldSource { } } - /// 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 source holds at most one `Arc` chunk and no accumulating - /// 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 { - self.chunk.as_ref().map_or(0, |c| c.len()) - } } impl aura_engine::Source for M1FieldSource { + /// Records resident in this source right now: the current chunk's length (or + /// 0 at exhaustion). The ring-residency probe (#95) — bounded by one chunk + /// length by construction (the source holds at most one `Arc` chunk and no + /// accumulating field), so it can never grow with window length; always `Some` + /// (this source reports). 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. + fn resident_records(&self) -> Option { + Some(self.chunk.as_ref().map_or(0, |c| c.len())) + } + fn peek(&self) -> Option { self.head.map(|(t, _)| t) } diff --git a/crates/aura-ingest/tests/streaming_seam.rs b/crates/aura-ingest/tests/streaming_seam.rs index 826be13..3891ae9 100644 --- a/crates/aura-ingest/tests/streaming_seam.rs +++ b/crates/aura-ingest/tests/streaming_seam.rs @@ -150,12 +150,19 @@ fn residency_is_bounded_by_one_chunk_independent_of_window_length() { let mut src = M1FieldSource::open(&server, SYMBOL, None, None, M1Field::Close) .expect("AAPL.US has archived data"); + // #95 adjacent friction resolved: residency is probeable through the trait — + // a `&dyn Source` reads it without downcasting to the concrete M1FieldSource. + assert!( + (&src as &dyn Source).resident_records().is_some(), + "M1FieldSource residency must be probeable via &dyn Source (no downcast)", + ); + // Drive the source the way `run` does (peek to pick, next to pop), sampling // resident records at every pull. let mut peak = 0usize; let mut total = 0usize; while src.peek().is_some() { - peak = peak.max(src.resident_records()); + peak = peak.max(src.resident_records().expect("M1FieldSource reports residency")); Source::next(&mut src); total += 1; } diff --git a/fieldtests/research-breakout-deepdive/t1b_residency.rs b/fieldtests/research-breakout-deepdive/t1b_residency.rs index b34b877..dcbf409 100644 --- a/fieldtests/research-breakout-deepdive/t1b_residency.rs +++ b/fieldtests/research-breakout-deepdive/t1b_residency.rs @@ -59,18 +59,18 @@ fn main() { // four field sources to exhaustion with no harness and no sinks. Per the docs // each keeps O(one chunk) resident — peak resident_records should stay tiny. { - // NOTE: resident_records() is an INHERENT method on the concrete - // M1FieldSource, NOT on the public Source trait (the streaming_seam test - // calls it on a concrete source). A consumer driving boxed `dyn Source`s - // therefore cannot read residency — recorded as a finding. We use the - // concrete type here, per field, exactly as the docs' residency test does. + // resident_records() is now on the public Source trait (#95 follow-up): a + // consumer driving boxed `dyn Source`s can read residency without a + // downcast. We use the concrete type here, per field, exactly as the docs' + // residency test does; the value is Option (None = a source that does not + // report, e.g. the eager VecSource), unwrapped to 0 for the max. let fields = [M1Field::Open, M1Field::High, M1Field::Low, M1Field::Close]; let mut peak_resident = 0usize; let mut total = 0u64; for f in fields { let mut s = M1FieldSource::open(&server, SYMBOL, Some(from_ms), Some(to_ms), f).unwrap(); while s.peek().is_some() { - peak_resident = peak_resident.max(s.resident_records()); + peak_resident = peak_resident.max(s.resident_records().unwrap_or(0)); Source::next(&mut s); total += 1; } diff --git a/fieldtests/research-breakout-deepdive/t1c_residency_scaling.rs b/fieldtests/research-breakout-deepdive/t1c_residency_scaling.rs index 828178b..c2571c5 100644 --- a/fieldtests/research-breakout-deepdive/t1c_residency_scaling.rs +++ b/fieldtests/research-breakout-deepdive/t1c_residency_scaling.rs @@ -44,7 +44,7 @@ fn drive(label: &str, from_ms: i64, to_ms: i64) { let mut total = 0u64; let mut peak_resident = 0usize; while s.peek().is_some() { - peak_resident = peak_resident.max(s.resident_records()); + peak_resident = peak_resident.max(s.resident_records().unwrap_or(0)); Source::next(&mut s); total += 1; }