Reduce memory pressure: Stream large collections, GC between phases

- Stream profile storage and score upsert instead of materializing
  full 20k+ item lists (propagation_grid_worker, propagation.ex)
- GC between forecast hours and store/compute phases to reclaim
  ~400 MB of grid data between steps
- Single-pass field extraction in scorer.ex path_integrated_conditions
  instead of 6 separate Enum traversals
- Eliminate intermediate merged map in fetch_grid by combining
  merge + profile build into one pipe
- Fix UUID bug: bingenerate → generate in native grid worker
  (same issue previously fixed in nexrad_worker)

lib/microwaveprop/propagation.ex

@@ -106,26 +106,24 @@ defmodule Microwaveprop.Propagation do
   def upsert_scores(scores, opts \\ []) do
     now = DateTime.truncate(DateTime.utc_now(), :second)
 
-    entries =
-      Enum.map(scores, fn s ->
-        %{
-          id: Ecto.UUID.generate(),
-          lat: s.lat,
-          lon: s.lon,
-          valid_time: s.valid_time,
-          band_mhz: s.band_mhz,
-          score: s.score,
-          factors: s.factors,
-          inserted_at: now,
-          updated_at: now
-        }
-      end)
-
     result =
       Repo.transaction(
         fn ->
-          entries
+          scores
-          |> Enum.chunk_every(500)
+          |> Stream.map(fn s ->
+            %{
+              id: Ecto.UUID.generate(),
+              lat: s.lat,
+              lon: s.lon,
+              valid_time: s.valid_time,
+              band_mhz: s.band_mhz,
+              score: s.score,
+              factors: s.factors,
+              inserted_at: now,
+              updated_at: now
+            }
+          end)
+          |> Stream.chunk_every(500)
           |> Enum.reduce(0, fn chunk, acc ->
             {count, _} =
               Repo.insert_all(GridScore, chunk,

lib/microwaveprop/propagation/scorer.ex

@@ -388,8 +388,18 @@ defmodule Microwaveprop.Propagation.Scorer do
   def path_integrated_conditions(profiles, contact) do
     lon = Kernel.||(contact.pos1["lon"] || contact.pos1["lng"], -97.0)
 
-    temps = profiles |> Enum.map(& &1.surface_temp_c) |> Enum.reject(&is_nil/1)
+    # Single-pass extraction of all fields to avoid 6 separate Enum traversals
-    dewpoints = profiles |> Enum.map(& &1.surface_dewpoint_c) |> Enum.reject(&is_nil/1)
+    {temps, dewpoints, pressures, gradients, bl_depths, pwats} =
+      Enum.reduce(profiles, {[], [], [], [], [], []}, fn p, {t, d, pr, g, b, pw} ->
+        {
+          if(is_nil(p.surface_temp_c), do: t, else: [p.surface_temp_c | t]),
+          if(is_nil(p.surface_dewpoint_c), do: d, else: [p.surface_dewpoint_c | d]),
+          if(is_nil(p.surface_pressure_mb), do: pr, else: [p.surface_pressure_mb | pr]),
+          if(is_nil(p.min_refractivity_gradient), do: g, else: [p.min_refractivity_gradient | g]),
+          if(is_nil(p.hpbl_m), do: b, else: [p.hpbl_m | b]),
+          if(is_nil(p.pwat_mm), do: pw, else: [p.pwat_mm | pw])
+        }
+      end)
 
     if temps == [] or dewpoints == [] do
       nil
@@ -399,11 +409,6 @@ defmodule Microwaveprop.Propagation.Scorer do
       avg_temp_f = c_to_f(avg_temp_c)
       avg_dewpoint_f = c_to_f(avg_dewpoint_c)
 
-      pressures = profiles |> Enum.map(& &1.surface_pressure_mb) |> Enum.reject(&is_nil/1)
-      gradients = profiles |> Enum.map(& &1.min_refractivity_gradient) |> Enum.reject(&is_nil/1)
-      bl_depths = profiles |> Enum.map(& &1.hpbl_m) |> Enum.reject(&is_nil/1)
-      pwats = profiles |> Enum.map(& &1.pwat_mm) |> Enum.reject(&is_nil/1)
-
       %{
         abs_humidity: absolute_humidity(avg_temp_c, avg_dewpoint_c),
         temp_f: avg_temp_f,

lib/microwaveprop/weather/hrrr_client.ex

@@ -63,10 +63,12 @@ defmodule Microwaveprop.Weather.HrrrClient do
             %{}
         end
 
-      merged = merge_grid_data(sfc_grid, prs_grid)
+      # Merge and build profiles in one pass to avoid a full intermediate map copy.
-
+      # Use Map.merge to include any pressure-only points, then map directly to profiles.
       profiles =
-        Map.new(merged, fn {point, data} ->
+        sfc_grid
+        |> Map.merge(prs_grid, fn _point, sfc_data, prs_data -> Map.merge(sfc_data, prs_data) end)
+        |> Map.new(fn {point, data} ->
           profile = build_profile(data)
           {point, Map.put(profile, :run_time, hour_dt)}
         end)
@@ -350,7 +352,7 @@ defmodule Microwaveprop.Weather.HrrrClient do
   def download_grib_ranges_to_file(_url, [], _dest_path), do: :ok
 
   def download_grib_ranges_to_file(url, ranges, dest_path) do
-    merged = merge_ranges(ranges) |> Enum.sort_by(&elem(&1, 0))
+    merged = ranges |> merge_ranges() |> Enum.sort_by(&elem(&1, 0))
 
     file = File.open!(dest_path, [:write, :binary])
 

lib/microwaveprop/workers/hrrr_native_grid_worker.ex

@@ -129,7 +129,7 @@ defmodule Microwaveprop.Workers.HrrrNativeGridWorker do
 
             [
               Map.merge(profile, %{
-                id: Ecto.UUID.bingenerate(),
+                id: Ecto.UUID.generate(),
                 valid_time: valid_time,
                 run_time: valid_time,
                 lat: lat,

lib/microwaveprop/workers/propagation_grid_worker.ex

@@ -40,6 +40,8 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
       for fh <- 0..@max_forecast_hour do
         valid_time = DateTime.add(run_time, fh * 3600, :second)
         result = process_forecast_hour(points, run_time, fh, valid_time)
+        # Reclaim grid_data/profiles from this forecast hour before starting the next
+        :erlang.garbage_collect()
 
         case result do
           :ok -> valid_time
@@ -74,6 +76,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
          end) do
       {:ok, grid_data} ->
         store_hrrr_profiles(grid_data, valid_time)
+        :erlang.garbage_collect()
 
         Phoenix.PubSub.broadcast(
           Microwaveprop.PubSub,
@@ -111,8 +114,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
   defp format_duration(ms), do: "#{Float.round(ms / 1000, 1)}s"
 
   defp store_hrrr_profiles(grid_data, valid_time) do
-    stored =
+    count =
-      Enum.flat_map(grid_data, fn {{lat, lon}, profile} ->
+      grid_data
+      |> Stream.flat_map(fn {{lat, lon}, profile} ->
         temp = profile.surface_temp_c
 
         if temp && temp > -80 && temp < 60 do
@@ -150,14 +154,13 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
           []
         end
       end)
+      |> Stream.chunk_every(500)
+      |> Enum.reduce(0, fn chunk, acc ->
+        Weather.upsert_hrrr_profiles_batch(chunk)
+        acc + length(chunk)
+      end)
 
-    stored
+    Logger.info("PropagationGrid: stored #{count} HRRR profiles")
-    |> Enum.chunk_every(500)
-    |> Enum.each(fn chunk ->
-      Weather.upsert_hrrr_profiles_batch(chunk)
-    end)
-
-    Logger.info("PropagationGrid: stored #{length(stored)} HRRR profiles")
   end
 
   defp compute_scores(grid_data, valid_time) do
@@ -178,7 +181,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
       max_concurrency: System.schedulers_online() * 2,
       timeout: 30_000
     )
-    |> Enum.flat_map(fn
+    |> Stream.flat_map(fn
       {:ok, results} -> results
       {:exit, _reason} -> []
     end)