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]