From e6952a42c83faa1e43254fc2c9c0d33d352fd05e Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 14 Apr 2026 15:26:58 -0500 Subject: [PATCH] Run PropagationGridWorker hourly, retain only the current chain's window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 2 +- config/dev.exs | 2 +- config/runtime.exs | 14 +++++--- lib/microwaveprop/propagation/scores_file.ex | 27 ++++++++++++++ .../workers/propagation_grid_worker.ex | 35 +++++++++++++----- .../propagation/scores_file_test.exs | 36 +++++++++++++++++++ 6 files changed, 100 insertions(+), 16 deletions(-) diff --git a/config/config.exs b/config/config.exs index 7a4d0a90..c3912f2d 100644 --- a/config/config.exs +++ b/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}, diff --git a/config/dev.exs b/config/dev.exs index 97141680..4f164d76 100644 --- a/config/dev.exs +++ b/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}, diff --git a/config/runtime.exs b/config/runtime.exs index 4b78bd8a..7fe2e5c8 100644 --- a/config/runtime.exs +++ b/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 diff --git a/lib/microwaveprop/propagation/scores_file.ex b/lib/microwaveprop/propagation/scores_file.ex index c366e95b..7f3de4a3 100644 --- a/lib/microwaveprop/propagation/scores_file.ex +++ b/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 diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 6914ed50..61be3439 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/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")}" diff --git a/test/microwaveprop/propagation/scores_file_test.exs b/test/microwaveprop/propagation/scores_file_test.exs index ddc98b58..4b4ab42d 100644 --- a/test/microwaveprop/propagation/scores_file_test.exs +++ b/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]