Aura/fieldtests/research-breakout-deepdive/t1b_residency.rs

// Task 1 (deep dive) — WHERE does the memory go? The docs (C12 realization
// cycle 0041; the streaming_seam.rs residency test) claim a M1FieldSource keeps
// O(one chunk) resident, "independent of window length". Yet the 7-year breakout
// run grew RSS to ~127 MiB. Is that the SOURCES (which the docs promise is
// bounded), or the recorder SINKS accumulating every bar into unbounded mpsc
// channels + the drain_trace Vec?
//
// This probe isolates the two by driving the SAME sources two ways over the SAME
// 7-year window:
//   (A) full breakout harness with all 4 recorder taps + drain_trace (what the
//       shipped example does), then
//   (B) the four M1FieldSource streams driven to exhaustion with NO harness, NO
//       sinks — peek/next only, the exact loop streaming_seam.rs calls bounded.
// VmHWM (peak RSS, /proc/self/status) is read after each. PUBLIC SURFACE ONLY:
// Source::{peek,next,resident_records} are the public trait methods the docs use.

use std::sync::Arc;

use chrono::TimeZone;
use data_server::{DataServer, DEFAULT_DATA_PATH};

use aura_engine::Source;
use aura_ingest::{M1Field, M1FieldSource};

#[path = "../../crates/aura-ingest/examples/shared/breakout_real.rs"]
mod breakout_real;
use breakout_real::*;

fn vmhwm_mib() -> u64 {
    let status = std::fs::read_to_string("/proc/self/status").unwrap_or_default();
    for line in status.lines() {
        if let Some(rest) = line.strip_prefix("VmHWM:") {
            let kib: u64 = rest.split_whitespace().next().and_then(|s| s.parse().ok()).unwrap_or(0);
            return kib / 1024;
        }
    }
    0
}

fn utc_span_ms(y0: i32, m0: u32, y1: i32, m1: u32) -> (i64, i64) {
    let from = chrono::Utc.with_ymd_and_hms(y0, m0, 1, 0, 0, 0).unwrap().timestamp_millis();
    let (ny, nm) = if m1 == 12 { (y1 + 1, 1) } else { (y1, m1 + 1) };
    let next = chrono::Utc.with_ymd_and_hms(ny, nm, 1, 0, 0, 0).unwrap().timestamp_millis();
    (from, next - 1)
}

fn main() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if !server.has_symbol(SYMBOL) {
        println!("skip: no local {SYMBOL} data");
        return;
    }
    let (from_ms, to_ms) = utc_span_ms(2018, 1, 2024, 12);

    println!("=== Task 1 deep dive: residency attribution (7-year GER40 window) ===");
    println!("VmHWM at start = {} MiB\n", vmhwm_mib());

    // (B) FIRST (so its peak is measured before the harness allocates): drive the
    // 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.
    {
        // 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().unwrap_or(0));
                Source::next(&mut s);
                total += 1;
            }
        }
        println!("(B) 4 raw sources, NO harness/sinks:");
        println!("    streamed {total} records, peak resident_records/source = {peak_resident}");
        println!("    VmHWM after = {} MiB", vmhwm_mib());
    }

    // (A) the full shipped breakout harness with 4 recorder taps + drain_trace.
    {
        let sources = open_ohlc_sources(&server, from_ms, to_ms).unwrap();
        let (mut h, taps) = build_harness();
        h.run(sources);
        let trace = drain_trace(&taps);
        println!("\n(A) full breakout harness + 4 recorder taps + drain_trace:");
        println!("    trace rows = {}", trace.len());
        println!("    VmHWM after = {} MiB", vmhwm_mib());
    }

    println!("\nVmHWM final = {} MiB", vmhwm_mib());
}