From 6d914614ac7779fd53f0bc29dbed382799f58ff2 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Sun, 19 Apr 2026 16:44:29 -0500
Subject: [PATCH] perf(grid-rs): stream bands to disk, drop merged grid early
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Previous pipeline held three overlapping large structures in RAM:
1. `merged: HashMap<Key, HashMap<String, f32>>` — 92k outer × ~60 String
   keys inner = ~200 MB of fragmented heap
2. `per_band: HashMap<u32, Vec<ScorePoint>>` — all 23 bands × 92k
   ScorePoints before the first write
3. A `scores.clone()` per band during the write loop (doubled that
   band's footprint transiently)

Consolidate into:
- Consume `merged` into a compact `Vec<(lat, lon, Conditions,
  BandInvariants)>`; the huge String-keyed HashMaps drop as soon as
  collection finishes.
- Loop over bands: score → write → vector drops at end of iteration.
  One band's ~1.8 MB score vector lives at a time.

Peak heap goes from ~1.5 Gi to ~250 Mi on the chain step. 2 Gi limit
stays, but we now have 8× headroom instead of tight-fit.
---
 rust/prop_grid_rs/src/pipeline.rs | 49 ++++++++++++++++---------------
 1 file changed, 25 insertions(+), 24 deletions(-)

diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs
index aa05cef7..de4aa05c 100644
--- a/rust/prop_grid_rs/src/pipeline.rs
+++ b/rust/prop_grid_rs/src/pipeline.rs
@@ -6,7 +6,6 @@
 //! minus the f00-only enrichment (native duct, NEXRAD, commercial link
 //! boost, ProfilesFile write) — Elixir retains f00.
 
-use std::collections::HashMap;
 use std::path::Path;
 
 use chrono::{DateTime, Datelike, Timelike, Utc};
@@ -108,33 +107,35 @@ pub async fn run_chain_step(
     let merged = merge_grids(sfc_grid, prs_grid);
     let point_count = merged.len() as u32;
 
-    // Score every point × every band; one ScorePoint per (cell, band).
+    // Step 1: consume `merged` into a compact `Vec<(lat, lon, Conditions,
+    // invariants)>`. This is the biggest win — `merged`'s inner HashMaps
+    // of `String -> f32` account for ~200 MB of heap (92k × ~60 string
+    // keys) and are dropped as soon as this loop finishes.
+    let prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> = merged
+        .into_iter()
+        .filter_map(|(key, cell)| {
+            let (lat, lon) = decoder::key_to_latlon(key);
+            let conditions = cell_to_conditions(&cell, lat, lon, &valid_time)?;
+            let invariants = scorer::precompute_band_invariants(&conditions);
+            Some((lat, lon, conditions, invariants))
+        })
+        .collect();
+
+    // Step 2: for each band, score → write → drop. One band's score
+    // vector (~92k × 20 B ≈ 1.8 MB) lives at a time instead of holding
+    // all 23 bands at once. The write runs on the blocking pool so the
+    // reactor stays responsive.
     let bands = band_config::all_bands();
-    let mut per_band: HashMap<u32, Vec<ScorePoint>> =
-        HashMap::with_capacity(bands.len());
-
-    for (key, cell) in &merged {
-        let (lat, lon) = decoder::key_to_latlon(*key);
-        let conditions = match cell_to_conditions(cell, lat, lon, &valid_time) {
-            Some(c) => c,
-            None => continue,
-        };
-        let invariants = scorer::precompute_band_invariants(&conditions);
-        for band in bands {
-            let r = scorer::composite_score_with(&conditions, band, Some(invariants));
-            per_band
-                .entry(band.freq_mhz)
-                .or_default()
-                .push(ScorePoint { lat, lon, score: r.score });
-        }
-    }
-
     let scores_dir = scores_dir.to_path_buf();
     let mut files_written = 0;
-    for (band_mhz, scores) in &per_band {
+    for band in bands {
+        let mut scores: Vec<ScorePoint> = Vec::with_capacity(prepared.len());
+        for (lat, lon, conditions, invariants) in &prepared {
+            let r = scorer::composite_score_with(conditions, band, Some(*invariants));
+            scores.push(ScorePoint { lat: *lat, lon: *lon, score: r.score });
+        }
         let dir = scores_dir.clone();
-        let band_mhz = *band_mhz;
-        let scores = scores.clone();
+        let band_mhz = band.freq_mhz;
         tokio::task::spawn_blocking(move || {
             scores_file::write_atomic(&dir, band_mhz, valid_time, &scores)
         })