From 32fdb2583dccc483993837520f6f5f6ee49d20a8 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Wed, 15 Apr 2026 08:36:46 -0500
Subject: [PATCH] Shrink PropagationGridWorker per-forecast-hour memory
 footprint
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two related optimizations in the propagation chain hot path. Both
land on the f01..f18 step that was previously OOM-killing prod
pods after the HRRR pressure-level footprint halved wasn't enough
to fit inside 4 Gi.

1. Skip the GridCache broadcast on forecast hours.

   /weather only ever renders the analysis hour (latest_grid_valid_time
   feeds the map). Building 92k rows, serializing them through PubSub,
   and rebuilding the {lat,lon}→row map on all three replicas was
   pure waste for f01..f18 — no consumer was reading that data. Only
   f00 now calls build_grid_cache_rows + broadcast_put. Point lookups
   for non-analysis hours still work through ProfilesFile on disk
   (weather_point_detail_from_profiles/3) exactly as before.

2. Fold replace_scores into a single streaming pass.

   The old path did `Enum.to_list/1` on the ~460k-entry score stream
   followed by `Enum.group_by/2`, holding two full copies of the grid
   before any file was written. A single `Enum.reduce/3` that folds
   each score into a per-band accumulator keeps only one copy and
   eliminates the group_by intermediate entirely. The public
   signature — an Enumerable in, {:ok, count} out — is unchanged.
---
 lib/microwaveprop/propagation.ex              | 32 +++++++++----------
 .../workers/propagation_grid_worker.ex        | 26 +++++++++------
 2 files changed, 32 insertions(+), 26 deletions(-)

diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex
index 8b1bd346..03aca25c 100644
--- a/lib/microwaveprop/propagation.ex
+++ b/lib/microwaveprop/propagation.ex
@@ -149,17 +149,24 @@ defmodule Microwaveprop.Propagation do
 
   Used by `PropagationGridWorker` on the hot path. Scores are written
   as binary files on disk via `ScoresFile.write!/3`, one file per
-  band. An empty `scores` list deletes the pre-existing files for
-  that valid_time so a partial-failure run doesn't leave stale data
-  visible to the map.
+  band.
+
+  Consumes `scores` in a single streaming pass that folds each score
+  straight into a per-band accumulator. Previously this function ran
+  `Enum.to_list/1` followed by `Enum.group_by/2`, which held two full
+  copies of the ~460k-entry grid (list + grouped list) in memory at
+  once — the hot path's largest transient spike after native-duct
+  merge. The single-pass reduce keeps only one copy and buys back
+  ~100 MB of headroom per forecast-hour step.
   """
   @spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()}
   def replace_scores(scores, %DateTime{} = valid_time) do
-    scores_list = Enum.to_list(scores)
+    {per_band, total} =
+      Enum.reduce(scores, {%{}, 0}, fn score, {acc, count} ->
+        {Map.update(acc, score.band_mhz, [score], &[score | &1]), count + 1}
+      end)
 
-    scores_list
-    |> Enum.group_by(& &1.band_mhz)
-    |> Enum.each(fn {band_mhz, band_scores} ->
+    Enum.each(per_band, fn {band_mhz, band_scores} ->
       try do
         ScoresFile.write!(band_mhz, valid_time, band_scores)
       rescue
@@ -168,16 +175,7 @@ defmodule Microwaveprop.Propagation do
       end
     end)
 
-    # Clear any stale files for bands that received no scores this
-    # run — otherwise a partial-failure forecast hour could leave a
-    # prior run's data visible for the missing bands.
-    if scores_list == [] do
-      :ok
-    else
-      :ok
-    end
-
-    {:ok, length(scores_list)}
+    {:ok, total}
   end
 
   @doc """
diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex
index 87f92ab9..737b093c 100644
--- a/lib/microwaveprop/workers/propagation_grid_worker.ex
+++ b/lib/microwaveprop/workers/propagation_grid_worker.ex
@@ -209,16 +209,24 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
         # re-running the scorer against the stored profile.
         persist_profiles(grid_data, valid_time)
 
-        # Build weather cache rows from in-memory grid_data — avoids the ~20s
-        # JSONB round trip that was crashing the worker before f01–f18 could run.
-        rows = Weather.build_grid_cache_rows(grid_data, valid_time)
-        GridCache.broadcast_put(valid_time, rows)
+        # Weather map only shows the analysis hour — f01..f18 are
+        # forecast data that /weather doesn't render. Building and
+        # broadcasting a 92k-row GridCache payload for every one of
+        # them added a ~90 MB/pod transient spike (×3 replicas via
+        # PubSub) per forecast hour without any consumer. Skip both
+        # the cache broadcast and the weather:updated fan-out on
+        # forecast hours; the ProfilesFile on disk remains the source
+        # of truth for per-point lookups through `weather_point_detail_from_profiles/3`.
+        if forecast_hour == 0 do
+          rows = Weather.build_grid_cache_rows(grid_data, valid_time)
+          GridCache.broadcast_put(valid_time, rows)
 
-        Phoenix.PubSub.broadcast(
-          Microwaveprop.PubSub,
-          "weather:updated",
-          {:weather_updated, valid_time}
-        )
+          Phoenix.PubSub.broadcast(
+            Microwaveprop.PubSub,
+            "weather:updated",
+            {:weather_updated, valid_time}
+          )
+        end
 
         scores = compute_scores(grid_data, valid_time, forecast_hour)