Run PropagationGridWorker hourly, retain only the current chain's window

Flip the cron from every-3-hours to every-hour now that a full
f00-f18 chain runs in ~45-60 min instead of ~170 min. With the
:propagation queue's concurrency-of-2, a slow chain won't block
the next one — they interleave, and the file store's last-writer
-wins semantics give the newer run_time's analysis data naturally.

Add ScoresFile.retain_window/2 that deletes any file whose
valid_time falls outside [run_time, run_time + 18h]. Call it at
fh=18 of the chain so leftover files from the previous chain
(specifically the old f00 whose valid_time the new chain doesn't
overwrite because it advanced by one hour) are discarded
immediately instead of waiting for the 2h prune cron.

The step transition path is extracted to handle_step_transition/2
to keep the nesting under credo's limit and make the final-step
logic easier to read.

config/config.exs

@@ -77,7 +77,7 @@ config :microwaveprop, Oban,
      crontab: [
        {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker},
        {"*/5 * * * *", Microwaveprop.Commercial.PollWorker},
-       {"5 */3 * * *", Microwaveprop.Workers.PropagationGridWorker},
+       {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker},
        {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker},
        # AsosAdjustmentWorker disabled — see runtime.exs for rationale.
        # {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker},

config/dev.exs

@@ -90,7 +90,7 @@ config :microwaveprop, Oban,
     {Oban.Plugins.Lifeline, rescue_after: to_timeout(minute: 45)},
     {Oban.Plugins.Cron,
      crontab: [
-       {"5 */3 * * *", Microwaveprop.Workers.PropagationGridWorker},
+       {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker},
        {"*/2 * * * *", Microwaveprop.Workers.MrmsFetchWorker},
        # AsosAdjustmentWorker disabled — see runtime.exs for rationale.
        # {"*/10 * * * *", Microwaveprop.Workers.AsosAdjustmentWorker},

config/runtime.exs

@@ -211,10 +211,14 @@ if config_env() == :prod do
        crontab: [
          {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker},
          {"*/5 * * * *", Microwaveprop.Commercial.PollWorker},
-         # Every 3 hours: one full run does f00–f18 (~5 min/hour × 19 ≈ 95 min).
-         # Hourly cron caused overlapping/failed runs because one full sweep
-         # exceeds a 60-minute window.
-         {"5 */3 * * *", Microwaveprop.Workers.PropagationGridWorker},
+         # Hourly at :05. With scores written as binary files (not
+         # Postgres) and HRRR grid profile persistence removed, a
+         # full f00-f18 chain runs in ~45-60 min, so hourly fits.
+         # If one chain slips past 60 min the :propagation queue's
+         # concurrency-of-2 lets the next chain interleave; files
+         # are keyed by (band, valid_time) so last-writer-wins gives
+         # newer analysis data naturally.
+         {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker},
          # MRMS refreshes the PrecipRate cache that AsosAdjustmentWorker
          # overlays onto the score grid. Both must run together — an
          # ASOS nudge without a fresh MRMS frame reverts to HRRR-only
@@ -235,7 +239,7 @@ if config_env() == :prod do
        ]}
     ]
 
-  config :microwaveprop, :email_from, {"NTMS Propagation", "graham@w5isp.com"}
+  config :microwaveprop, :email_from, {"NTMS Propagation", "prop@w5isp.com"}
   config :microwaveprop, hrrr_base_url: System.get_env("HRRR_BASE_URL", "http://skippy.w5isp.com:8080")
   config :microwaveprop, srtm_tiles_dir: "/data/srtm"
   # config :microwaveprop, start_freshness_monitor: true

lib/microwaveprop/propagation/scores_file.ex

@@ -110,6 +110,33 @@ defmodule Microwaveprop.Propagation.ScoresFile do
     end)
   end
 
+  @doc """
+  Keep only the files whose `valid_time` falls inside the closed
+  window `[run_time, run_time + max_forecast_hour * 3600]`. Returns
+  the number of files deleted.
+
+  Called at the end of a `PropagationGridWorker` chain so any files
+  left over from the previous run_time (that the new chain didn't
+  overwrite by coincidence of valid_time) get cleaned up immediately
+  instead of waiting for the 2h prune cron.
+  """
+  @spec retain_window(DateTime.t(), non_neg_integer()) :: non_neg_integer()
+  def retain_window(%DateTime{} = run_time, max_forecast_hour) when max_forecast_hour >= 0 do
+    lo = DateTime.to_unix(run_time)
+    hi = lo + max_forecast_hour * 3600
+
+    base_dir()
+    |> list_score_files()
+    |> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
+      if valid_time_unix < lo or valid_time_unix > hi do
+        _ = File.rm(path)
+        acc + 1
+      else
+        acc
+      end
+    end)
+  end
+
   @doc """
   List every `valid_time` that has a score file on disk for
   `band_mhz`, ordered ascending. Empty list if the band directory

lib/microwaveprop/workers/propagation_grid_worker.ex

@@ -27,6 +27,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
   alias Microwaveprop.Propagation.BandConfig
   alias Microwaveprop.Propagation.Grid
   alias Microwaveprop.Propagation.ScoreCache
+  alias Microwaveprop.Propagation.ScoresFile
   alias Microwaveprop.Weather
   alias Microwaveprop.Weather.GridCache
   alias Microwaveprop.Weather.HrrrClient
@@ -111,15 +112,9 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
         total_ms = System.monotonic_time(:millisecond) - t_start
         Logger.info("PropagationGrid: fh=#{fh} step finished in #{format_duration(total_ms)}")
 
-        case enqueue_next_step(run_time, fh) do
-          :final ->
-            Weather.prune_old_grid_profiles()
-            Propagation.prune_old_scores()
-            Logger.info("PropagationGrid: chain complete for run_time=#{run_time}")
-
-          {:ok, _job} ->
-            :ok
-        end
+        run_time
+        |> enqueue_next_step(fh)
+        |> handle_step_transition(run_time)
 
         :ok
 
@@ -128,6 +123,28 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
     end
   end
 
+  # `enqueue_next_step/2` returns `:final` at fh=18 (the whole chain
+  # is done) or `{:ok, job}` when the next hour has been scheduled.
+  # Split into its own function to keep `run_chain_step/2` under
+  # credo's nesting limit.
+  defp handle_step_transition(:final, run_time) do
+    Weather.prune_old_grid_profiles()
+    Propagation.prune_old_scores()
+    # Drop any files left over from the previous chain that the new
+    # chain didn't happen to overwrite (files are keyed by
+    # valid_time, so an "old f00" from a prior run escapes the
+    # natural last-writer-wins path).
+    dropped = ScoresFile.retain_window(run_time, @max_forecast_hour)
+
+    if dropped > 0 do
+      Logger.info("PropagationGrid: discarded #{dropped} leftover score files from prior chain")
+    end
+
+    Logger.info("PropagationGrid: chain complete for run_time=#{run_time}")
+  end
+
+  defp handle_step_transition({:ok, _job}, _run_time), do: :ok
+
   defp process_forecast_hour(points, run_time, forecast_hour, valid_time) do
     label = "f#{String.pad_leading(Integer.to_string(forecast_hour), 2, "0")}"
 

test/microwaveprop/propagation/scores_file_test.exs

@@ -185,6 +185,42 @@ defmodule Microwaveprop.Propagation.ScoresFileTest do
     end
   end
 
+  describe "retain_window/2" do
+    test "deletes files outside [run_time, run_time + hours] across every band" do
+      run_time = ~U[2026-04-14 16:00:00Z]
+
+      ScoresFile.write!(10_000, ~U[2026-04-14 15:00:00Z], [])
+      ScoresFile.write!(10_000, ~U[2026-04-14 16:00:00Z], [])
+      ScoresFile.write!(10_000, ~U[2026-04-14 17:00:00Z], [])
+      ScoresFile.write!(24_000, ~U[2026-04-14 14:00:00Z], [])
+      ScoresFile.write!(24_000, ~U[2026-04-14 16:00:00Z], [])
+      ScoresFile.write!(24_000, ~U[2026-04-15 10:00:00Z], [])
+      ScoresFile.write!(24_000, ~U[2026-04-15 11:00:00Z], [])
+
+      # Window: 16:00 -> 10:00 next day (18 hours forward).
+      assert ScoresFile.retain_window(run_time, 18) == 3
+
+      assert ScoresFile.list_valid_times(10_000) == [
+               ~U[2026-04-14 16:00:00Z],
+               ~U[2026-04-14 17:00:00Z]
+             ]
+
+      assert ScoresFile.list_valid_times(24_000) == [
+               ~U[2026-04-14 16:00:00Z],
+               ~U[2026-04-15 10:00:00Z]
+             ]
+    end
+
+    test "returns 0 and leaves foreign files alone", %{dir: dir} do
+      run_time = ~U[2026-04-14 16:00:00Z]
+      stray = Path.join(dir, "some_other_file.txt")
+      File.write!(stray, "keep me")
+
+      assert ScoresFile.retain_window(run_time, 18) == 0
+      assert File.exists?(stray)
+    end
+  end
+
   describe "prune_older_than/1" do
     test "deletes files whose valid_time is strictly before the cutoff", %{dir: _dir} do
       old = ~U[2026-04-14 10:00:00Z]