fix propagation forecast persistence

lib/microwaveprop/propagation.ex

@@ -288,8 +288,9 @@ defmodule Microwaveprop.Propagation do
   end
 
   @doc """
-  Retains only score/profile/scalar files within the active 48-hour
+  Retains score files through GEFS's 168-hour horizon and profile/scalar
-  forecast window, deleting everything older than `run_time`. Called
+  files through HRRR's 48-hour horizon, deleting files older than
+  `run_time`. Called
   by `NotifyListener` after chain completion to keep `/data/scores`
   within bounds.
 
@@ -297,11 +298,9 @@ defmodule Microwaveprop.Propagation do
   """
   @spec retain_scores_window(DateTime.t()) :: :ok
   def retain_scores_window(%DateTime{} = run_time) do
-    max_fh = 48
+    scores_deleted = ScoresFile.retain_window(run_time, 168)
-
+    profiles_deleted = ProfilesFile.retain_window(run_time, 48)
-    scores_deleted = ScoresFile.retain_window(run_time, max_fh)
+    scalars_deleted = ScalarFile.retain_window(run_time, 48)
-    profiles_deleted = ProfilesFile.retain_window(run_time, max_fh)
-    scalars_deleted = ScalarFile.retain_window(run_time, max_fh)
     total = scores_deleted + profiles_deleted + scalars_deleted
 
     if total > 0 do
@@ -442,46 +441,34 @@ defmodule Microwaveprop.Propagation do
   @spec warm_cache_and_broadcast(non_neg_integer(), DateTime.t()) ::
           :ok | {:error, :enoent | :invalid_format}
   def warm_cache_and_broadcast(band_mhz, valid_time) do
-    hrrr =
+    hrrr = read_score_points(&ScoresFile.read/2, band_mhz, valid_time)
-      case ScoresFile.read(band_mhz, valid_time) do
+    hrdps = read_score_points(&ScoresFile.read_hrdps/2, band_mhz, valid_time)
-        {:ok, payload} -> ScoresFile.extract_points(payload, nil)
-        _ -> nil
-      end
-
-    hrdps =
-      case ScoresFile.read_hrdps(band_mhz, valid_time) do
-        {:ok, payload} -> ScoresFile.extract_points(payload, nil)
-        _ -> nil
-      end
 
     # Merge priority: HRRR > HRDPS > GEFS. GEFS provides extended-horizon
     # coverage beyond HRRR's 48h window. Within the f24-f48 overlap, cells
     # already present in HRRR/HRDPS are skipped so the coarser GEFS scores
     # don't override the higher-resolution ones.
     merged = (hrrr || []) ++ (hrdps || [])
-    merged_keys = MapSet.new(merged, fn %{lat: lat, lon: lon} -> {lat, lon} end)
+    gefs = read_score_points(&ScoresFile.read_gefs/2, band_mhz, valid_time) || []
+    scores = ScoresFile.merge_preferred(merged, gefs)
 
-    gefs =
+    case {hrrr, hrdps, scores} do
-      case ScoresFile.read_gefs(band_mhz, valid_time) do
-        {:ok, payload} ->
-          payload
-          |> ScoresFile.extract_points(nil)
-          |> Enum.reject(fn %{lat: lat, lon: lon} -> MapSet.member?(merged_keys, {lat, lon}) end)
-
-        _ ->
-          []
-      end
-
-    case {hrrr, hrdps, gefs} do
       {nil, nil, []} ->
         {:error, :enoent}
 
-      {h, c, g} ->
+      _ ->
-        ScoreCache.broadcast_put(band_mhz, valid_time, (h || []) ++ (c || []) ++ g)
+        ScoreCache.broadcast_put(band_mhz, valid_time, scores)
         :ok
     end
   end
 
+  defp read_score_points(reader, band_mhz, valid_time) do
+    case reader.(band_mhz, valid_time) do
+      {:ok, payload} -> ScoresFile.extract_points(payload, nil)
+      _ -> nil
+    end
+  end
+
   defp filter_bounds(scores, nil), do: scores
 
   defp filter_bounds(scores, %{"south" => s, "north" => n, "west" => w, "east" => e}) do

lib/microwaveprop/propagation/notify_listener.ex

@@ -120,7 +120,7 @@ defmodule Microwaveprop.Propagation.NotifyListener do
     {past, future} = Propagation.hot_cache_window()
     ScoreCache.prune_outside_window(past, future)
 
-    # Sweep NFS scores to the active 48h window anchored on run_time so
+    # Sweep NFS files to their active forecast windows anchored on run_time so
     # stale forecast hours from the previous cycle don't accumulate.
     # Calling with valid_time would slide the window forward for each
     # completed forecast-hour task and delete the current run's earlier

lib/microwaveprop/propagation/path_compute.ex

@@ -378,25 +378,33 @@ defmodule Microwaveprop.Propagation.PathCompute do
   end
 
   defp collect_profile_fields(profiles) do
-    Enum.reduce(profiles, {[], [], [], [], [], [], [], [], [], []}, fn p, {ts, ds, ps, gs, bs, ws, wus, wvs, scs, pms} ->
+    fields = [
-      {
+      :surface_temp_c,
-        if(p.surface_temp_c == nil, do: ts, else: [p.surface_temp_c | ts]),
+      :surface_dewpoint_c,
-        if(p.surface_dewpoint_c == nil, do: ds, else: [p.surface_dewpoint_c | ds]),
+      :surface_pressure_mb,
-        if(p.surface_pressure_mb == nil, do: ps, else: [p.surface_pressure_mb | ps]),
+      :min_refractivity_gradient,
-        if(p.min_refractivity_gradient == nil,
+      :hpbl_m,
-          do: gs,
+      :pwat_mm,
-          else: [p.min_refractivity_gradient | gs]
+      :wind_u,
-        ),
+      :wind_v,
-        if(p.hpbl_m == nil, do: bs, else: [p.hpbl_m | bs]),
+      :cloud_cover_pct,
-        if(p.pwat_mm == nil, do: ws, else: [p.pwat_mm | ws]),
+      :precip_mm
-        if(Map.get(p, :wind_u) == nil, do: wus, else: [Map.get(p, :wind_u) | wus]),
+    ]
-        if(Map.get(p, :wind_v) == nil, do: wvs, else: [Map.get(p, :wind_v) | wvs]),
+
-        if(Map.get(p, :cloud_cover_pct) == nil, do: scs, else: [Map.get(p, :cloud_cover_pct) | scs]),
+    profiles
-        if(Map.get(p, :precip_mm) == nil, do: pms, else: [Map.get(p, :precip_mm) | pms])
+    |> Enum.reduce(Map.new(fields, &{&1, []}), fn profile, acc ->
-      }
+      Enum.reduce(fields, acc, fn field, field_acc ->
+        Map.update!(field_acc, field, &prepend_if_present(&1, Map.get(profile, field)))
+      end)
+    end)
+    |> then(fn values ->
+      List.to_tuple(Enum.map(fields, &Map.fetch!(values, &1)))
     end)
   end
 
+  defp prepend_if_present(values, nil), do: values
+  defp prepend_if_present(values, value), do: [value | values]
+
   defp build_conditions(
          avg_temp_c,
          avg_dewpoint_c,
@@ -406,28 +414,9 @@ defmodule Microwaveprop.Propagation.PathCompute do
          {pressures, gradients, bl_depths, pwats, wind_us, wind_vs, sky_covers, precip_mms},
          native_duct
        ) do
-    wind_speed =
+    wind_speed = max_wind_speed(wind_us, wind_vs)
-      if wind_us != [] and wind_vs != [] do
+    sky_cover = max_or_default(sky_covers, nil)
-        wind_us
+    rain_rate = precip_mms |> Enum.map(&Scorer.precip_to_rate_mmhr/1) |> max_or_default(0.0)
-        |> Enum.zip(wind_vs)
-        |> Enum.map(fn {u, v} -> Scorer.wind_speed_kts(u, v) end)
-        |> Enum.filter(& &1)
-        |> case do
-          [] -> nil
-          speeds -> Enum.max(speeds)
-        end
-      end
-
-    sky_cover = if sky_covers == [], do: nil, else: Enum.max(sky_covers)
-
-    rain_rate =
-      if precip_mms == [] do
-        0.0
-      else
-        precip_mms
-        |> Enum.map(&Scorer.precip_to_rate_mmhr/1)
-        |> Enum.max()
-      end
 
     %{
       abs_humidity: Scorer.absolute_humidity(avg_temp_c, avg_dewpoint_c),
@@ -453,6 +442,17 @@ defmodule Microwaveprop.Propagation.PathCompute do
     }
   end
 
+  defp max_wind_speed(wind_us, wind_vs) do
+    wind_us
+    |> Enum.zip(wind_vs)
+    |> Enum.map(fn {u, v} -> Scorer.wind_speed_kts(u, v) end)
+    |> Enum.reject(&is_nil/1)
+    |> max_or_default(nil)
+  end
+
+  defp max_or_default([], default), do: default
+  defp max_or_default(values, _default), do: Enum.max(values)
+
   @doc false
   @spec compute_loss_budget(float(), float(), map(), map() | nil, map() | nil) :: map()
   def compute_loss_budget(dist_km, freq_ghz, band_config, terrain_result, conditions) do

lib/microwaveprop/propagation/scores_file.ex

@@ -304,18 +304,31 @@ defmodule Microwaveprop.Propagation.ScoresFile do
     # GEFS fills cells that neither HRRR nor HRDPS cover. Within the
     # f24-f48 overlap window HRRR wins; beyond 48h GEFS is the only
     # source for extended-horizon outlooks.
-    merged_keys = MapSet.new(merged, fn %{lat: lat, lon: lon} -> {lat, lon} end)
-
     gefs_cells =
       case read_gefs(band_mhz, valid_time) do
         {:ok, payload} -> extract_points(payload, bounds)
         _ -> []
       end
 
-    extras =
+    merge_preferred(merged, gefs_cells)
-      Enum.reject(gefs_cells, fn %{lat: lat, lon: lon} -> MapSet.member?(merged_keys, {lat, lon}) end)
+  end
 
-    merged ++ extras
+  @doc """
+  Merges score cells with `preferred` taking precedence over `fallback`.
+
+  Coordinates are compared by canonical propagation-grid index rather than
+  raw floating-point equality, so equivalent cells from different decoders
+  cannot leak through because of insignificant coordinate drift.
+  """
+  @spec merge_preferred([map()], [map()]) :: [map()]
+  def merge_preferred(preferred, fallback) do
+    preferred_keys = MapSet.new(preferred, &canonical_cell/1)
+    preferred ++ Enum.reject(fallback, &MapSet.member?(preferred_keys, canonical_cell(&1)))
+  end
+
+  defp canonical_cell(%{lat: lat, lon: lon}) do
+    bounds = Grid.bounds()
+    {round((lat - bounds.lat_min) / Grid.step()), round((lon - bounds.lon_min) / Grid.step())}
   end
 
   @doc """

lib/microwaveprop/workers/gefs_fetch_worker.ex

@@ -39,7 +39,6 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do
         {:ok, profiles} ->
           valid_time = run_time |> DateTime.add(fh * 3600, :second) |> DateTime.truncate(:second)
           score_and_persist(profiles, run_time, fh, valid_time)
-          :ok
 
         {:error, :wgrib2_not_available} ->
           Logger.warning("GefsFetch: wgrib2 missing on host — cancelling job")
@@ -188,27 +187,30 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do
     # Write via ScoresFile.write_gefs! so extended-horizon scores land in
     # `.gefs.prop` files — separate from HRRR's `.prop` files. Within the
     # f24-f48 overlap window HRRR wins; beyond 48h GEFS is the only source.
-    total =
+    with {:ok, total} <- persist_scores(scores, valid_time) do
-      scores
+      Logger.info("GefsFetch: wrote #{total} GEFS scores for fh=#{fh} @ #{valid_time}")
-      |> Enum.group_by(& &1.band_mhz)
+      broadcast_updated(valid_time)
-      |> Enum.reduce(0, fn {band_mhz, band_scores}, acc ->
-        try do
-          ScoresFile.write_gefs!(band_mhz, valid_time, band_scores)
-          acc + length(band_scores)
-        rescue
-          e ->
-            Logger.error("GefsFetch: ScoresFile.write_gefs! failed band=#{band_mhz} fh=#{fh}: #{inspect(e)}")
 
-            acc
+      attrs = Enum.map(profiles, &build_profile_attrs(run_time, fh, &1))
-        end
+      {count, _} = Weather.upsert_gefs_profiles_batch(attrs)
-      end)
+      Logger.info("GefsFetch: upserted #{count}/#{length(attrs)} profile rows for fh=#{fh}")
+      :ok
+    end
+  end
 
-    Logger.info("GefsFetch: wrote #{total} GEFS scores for fh=#{fh} @ #{valid_time}")
+  @doc false
-    broadcast_updated(valid_time)
+  @spec persist_scores([map()], DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()}
-
+  def persist_scores(scores, valid_time) do
-    attrs = Enum.map(profiles, &build_profile_attrs(run_time, fh, &1))
+    scores
-    {count, _} = Weather.upsert_gefs_profiles_batch(attrs)
+    |> Enum.group_by(& &1.band_mhz)
-    Logger.info("GefsFetch: upserted #{count}/#{length(attrs)} profile rows for fh=#{fh}")
+    |> Enum.reduce_while({:ok, 0}, fn {band_mhz, band_scores}, {:ok, total} ->
+      try do
+        ScoresFile.write_gefs!(band_mhz, valid_time, band_scores)
+        {:cont, {:ok, total + length(band_scores)}}
+      rescue
+        error -> {:halt, {:error, {:score_write_failed, band_mhz, error}}}
+      end
+    end)
   end
 
   defp profiles_to_grid(profiles) do

test/microwaveprop/propagation/notify_listener_test.exs

@@ -88,6 +88,16 @@ defmodule Microwaveprop.Propagation.NotifyListenerTest do
       assert ScoreCache.fetch(10_000, stale) == :miss
     end
 
+    test "retains GEFS score files through 168 hours" do
+      run_time = ~U[2026-04-21 00:00:00Z]
+      extended_time = DateTime.add(run_time, 168, :hour)
+      ScoresFile.write_gefs!(10_000, extended_time, sample_scores())
+
+      {:ok, _task_pid} = NotifyListener.handle_propagation_ready(run_time, run_time)
+
+      assert File.exists?(ScoresFile.path_for_gefs(10_000, extended_time))
+    end
+
     test "kicks off async ScalarFile materialization for the new valid_time" do
       vt = ~U[2026-04-21 16:00:00Z]
       ScoresFile.write!(10_000, vt, sample_scores())

test/microwaveprop/propagation/scores_file_test.exs

@@ -224,6 +224,20 @@ defmodule Microwaveprop.Propagation.ScoresFileTest do
   end
 
   describe "read_bounds/3" do
+    test "canonical grid cells from GEFS do not duplicate preferred cells" do
+      preferred = [%{lat: 32.875, lon: -97.0, score: 80}]
+
+      fallback = [
+        %{lat: 32.87500001, lon: -97.00000001, score: 20},
+        %{lat: 33.0, lon: -97.0, score: 30}
+      ]
+
+      assert ScoresFile.merge_preferred(preferred, fallback) == [
+               %{lat: 32.875, lon: -97.0, score: 80},
+               %{lat: 33.0, lon: -97.0, score: 30}
+             ]
+    end
+
     test "returns every scored cell (no-data excluded) when bounds is nil" do
       ScoresFile.write!(10_000, ~U[2026-04-14 18:00:00Z], [
         %{lat: 25.0, lon: -125.0, score: 10},
@@ -335,6 +349,19 @@ defmodule Microwaveprop.Propagation.ScoresFileTest do
   end
 
   describe "retain_window/2" do
+    test "recognizes GEFS files through the extended horizon" do
+      run_time = ~U[2026-04-14 00:00:00Z]
+      inside = DateTime.add(run_time, 168, :hour)
+      outside = DateTime.add(run_time, 174, :hour)
+
+      ScoresFile.write_gefs!(10_000, inside, [])
+      ScoresFile.write_gefs!(10_000, outside, [])
+
+      assert ScoresFile.retain_window(run_time, 168) == 1
+      assert File.exists?(ScoresFile.path_for_gefs(10_000, inside))
+      refute File.exists?(ScoresFile.path_for_gefs(10_000, outside))
+    end
+
     test "deletes files outside [run_time, run_time + hours] across every band" do
       run_time = ~U[2026-04-14 16:00:00Z]
 

test/microwaveprop/workers/gefs_fetch_worker_test.exs

@@ -3,6 +3,7 @@ defmodule Microwaveprop.Workers.GefsFetchWorkerTest do
   use ExUnitProperties
 
   alias Microwaveprop.Propagation
+  alias Microwaveprop.Propagation.ScoresFile
   alias Microwaveprop.Weather
   alias Microwaveprop.Weather.GefsClient
   alias Microwaveprop.Weather.GefsProfile
@@ -84,6 +85,32 @@ defmodule Microwaveprop.Workers.GefsFetchWorkerTest do
     end
   end
 
+  describe "persist_scores/2" do
+    test "returns an error when any score file cannot be written" do
+      blocker = Path.join(System.tmp_dir!(), "gefs_write_blocker_#{System.unique_integer([:positive])}")
+      File.write!(blocker, "not a directory")
+      previous = Application.get_env(:microwaveprop, :propagation_scores_dir)
+      Application.put_env(:microwaveprop, :propagation_scores_dir, Path.join(blocker, "scores"))
+
+      on_exit(fn ->
+        File.rm!(blocker)
+
+        if previous do
+          Application.put_env(:microwaveprop, :propagation_scores_dir, previous)
+        else
+          Application.delete_env(:microwaveprop, :propagation_scores_dir)
+        end
+      end)
+
+      scores = [%{band_mhz: 10_000, lat: 32.875, lon: -97.0, score: 50}]
+
+      assert {:error, {:score_write_failed, 10_000, _}} =
+               GefsFetchWorker.persist_scores(scores, ~U[2026-04-19 12:00:00Z])
+
+      refute File.exists?(ScoresFile.path_for_gefs(10_000, ~U[2026-04-19 12:00:00Z]))
+    end
+  end
+
   describe "most_recent_available_run/1" do
     test "returns the run cycle 5 or more hours earlier" do
       # 15:00 UTC → 10:00 UTC → snaps to 06Z