From 6390093f9338305911e1726217bbc60cd394966e Mon Sep 17 00:00:00 2001
From: Brummel <mould@mycelium.de>
Date: Wed, 17 Jun 2026 09:59:58 +0200
Subject: [PATCH] refactor(aura-ingest): chunk-direct M1 transpose, drop the
 AoS intermediate
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

load_m1_window collected every bar into an intermediate Vec<M1Parsed> (a full
AoS materialization, un-presized so it grew by reallocation) before transpose_m1
copied it again into the SoA columns — ~2x peak load memory plus realloc churn.

Transpose each chunk straight into the columns instead, via a new shared
M1Columns::extend_from_bars (reserves per chunk for amortized growth). One fewer
full copy, halved peak. Behaviour-preserving: the resulting M1Columns is
byte-identical and transpose_m1's / load_m1_window's signatures are unchanged.
Pinned by a new parity test that needs no live DataServer.

closes #78
---
 crates/aura-ingest/src/lib.rs | 63 +++++++++++++++++++++++++++--------
 1 file changed, 49 insertions(+), 14 deletions(-)

diff --git a/crates/aura-ingest/src/lib.rs b/crates/aura-ingest/src/lib.rs
index 1c6ee3f..a9e7f6d 100644
--- a/crates/aura-ingest/src/lib.rs
+++ b/crates/aura-ingest/src/lib.rs
@@ -62,6 +62,31 @@ impl M1Columns {
         }
     }
 
+    /// Append one chunk of AoS bars, transposing into these SoA columns and
+    /// normalizing time ms->epoch-ns at the boundary (C3). Reserves per chunk so
+    /// repeated appends grow amortized; the push order is preserved (the
+    /// chronological merge order C3 relies on). Lets `load_m1_window` transpose
+    /// chunk-by-chunk without an intermediate AoS buffer.
+    fn extend_from_bars(&mut self, bars: &[M1Parsed]) {
+        let n = bars.len();
+        self.ts.reserve(n);
+        self.open.reserve(n);
+        self.high.reserve(n);
+        self.low.reserve(n);
+        self.close.reserve(n);
+        self.spread.reserve(n);
+        self.volume.reserve(n);
+        for b in bars {
+            self.ts.push(unix_ms_to_epoch_ns(b.time_ms));
+            self.open.push(b.open);
+            self.high.push(b.high);
+            self.low.push(b.low);
+            self.close.push(b.close);
+            self.spread.push(b.spread);
+            self.volume.push(b.volume);
+        }
+    }
+
     /// The close column as an engine source stream: each normalized timestamp
     /// zipped with its close as a [`Scalar::f64`]. Ascending in timestamp iff
     /// the bars were (the C3 ingestion precondition `Harness::run` relies on).
@@ -81,21 +106,14 @@ impl M1Columns {
 /// columns (C1) — it reads no clock and no external state.
 pub fn transpose_m1(bars: &[M1Parsed]) -> M1Columns {
     let mut c = M1Columns::with_capacity(bars.len());
-    for b in bars {
-        c.ts.push(unix_ms_to_epoch_ns(b.time_ms));
-        c.open.push(b.open);
-        c.high.push(b.high);
-        c.low.push(b.low);
-        c.close.push(b.close);
-        c.spread.push(b.spread);
-        c.volume.push(b.volume);
-    }
+    c.extend_from_bars(bars);
     c
 }
 
 /// Drain a data-server M1 window into transposed SoA columns. Reads the
-/// chronological chunk iterator to exhaustion into one AoS buffer, then
-/// transposes once at the boundary (C3 — one merge point, not per chunk).
+/// chronological chunk iterator to exhaustion, transposing each chunk directly
+/// into the SoA columns at the boundary (C3 — one merge point; no intermediate
+/// AoS buffer, so peak memory is the columns alone, not columns + an AoS copy).
 ///
 /// Each bound is an `Option<i64>` of inclusive Unix-ms (data-server's window
 /// contract): `None` means unbounded on that side (data-server skips the
@@ -122,12 +140,15 @@ pub fn load_m1_window(
     to_ms: Option<i64>,
 ) -> Option<M1Columns> {
     let mut it = server.stream_m1_windowed(symbol, from_ms, to_ms)?;
-    let mut bars: Vec<M1Parsed> = Vec::new();
+    // Transpose each chunk straight into the SoA columns — no intermediate AoS
+    // buffer (halves peak load memory, drops one full copy). The per-chunk push
+    // order is the chronological merge order (C3).
+    let mut cols = M1Columns::with_capacity(0);
     // M1Parsed is Copy; &Arc<[M1Parsed]> derefs to &[M1Parsed] in arg position.
     while let Some(chunk) = it.next_chunk() {
-        bars.extend_from_slice(&chunk);
+        cols.extend_from_bars(&chunk);
     }
-    Some(transpose_m1(&bars))
+    Some(cols)
 }
 
 /// Which base column of an M1 bar a source streams (C7: a composite window is a
@@ -301,6 +322,20 @@ mod tests {
         assert!(c.volume.is_empty());
     }
 
+    #[test]
+    fn chunked_accumulation_equals_single_transpose() {
+        // `load_m1_window` now transposes chunk-by-chunk into the SoA columns
+        // (no intermediate AoS buffer). That must be byte-identical to one
+        // transpose over the concatenation — the path it replaced. This pins the
+        // shared `extend_from_bars` logic without needing a live DataServer.
+        let all = [full_bar(1_000), full_bar(2_000), full_bar(3_000)];
+        let single = transpose_m1(&all);
+        let mut chunked = M1Columns::with_capacity(0);
+        chunked.extend_from_bars(&all[0..2]); // first chunk
+        chunked.extend_from_bars(&all[2..3]); // second chunk
+        assert_eq!(chunked, single);
+    }
+
     #[test]
     fn close_stream_zips_ts_with_close_in_order() {
         // distinct close per bar so order is observable; ts ascending.