From 664f1353db9c7ba6087a5a2754a780849ef2b579 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 9 Apr 2026 12:29:33 -0500 Subject: [PATCH] Decouple propagation_scores pruning from the compute worker Pruning used to only run at the end of a successful PropagationGridWorker pass, so a stretch of failed compute jobs (k8s OOM kills, SIGTERM) stopped prune from running and let the table accumulate ~5h of stale rows. A dedicated PropagationPruneWorker now runs every 15 minutes on its own Oban cron, and PropagationGridWorker also calls prune_old_scores at the start of each run as a second safety net. Bumped the delete timeout from 2m to 5m so the first catch-up pass has enough headroom. --- config/config.exs | 3 +- config/dev.exs | 1 + config/runtime.exs | 1 + lib/microwaveprop/propagation.ex | 11 +++- .../workers/propagation_grid_worker.ex | 5 ++ .../workers/propagation_prune_worker.ex | 24 +++++++++ .../workers/propagation_prune_worker_test.exs | 51 +++++++++++++++++++ 7 files changed, 93 insertions(+), 3 deletions(-) create mode 100644 lib/microwaveprop/workers/propagation_prune_worker.ex create mode 100644 test/microwaveprop/workers/propagation_prune_worker_test.exs diff --git a/config/config.exs b/config/config.exs index 74cf2dc2..fc8e291b 100644 --- a/config/config.exs +++ b/config/config.exs @@ -52,7 +52,8 @@ config :microwaveprop, Oban, crontab: [ {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, - {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker} + {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker} ]} ] diff --git a/config/dev.exs b/config/dev.exs index e73fd674..39b5fba1 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -88,6 +88,7 @@ config :microwaveprop, Oban, {Oban.Plugins.Cron, crontab: [ {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker} ]} ] diff --git a/config/runtime.exs b/config/runtime.exs index fde1c30f..4575cdef 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -152,6 +152,7 @@ if config_env() == :prod do {"0 8 * * *", Microwaveprop.Workers.SolarIndexWorker}, {"*/5 * * * *", Microwaveprop.Commercial.PollWorker}, {"5 * * * *", Microwaveprop.Workers.PropagationGridWorker}, + {"*/15 * * * *", Microwaveprop.Workers.PropagationPruneWorker}, {"*/30 * * * *", Microwaveprop.Workers.BackfillEnqueueWorker, args: %{"limit" => 500, "types" => ["hrrr", "weather", "terrain", "iemre"]}} ]} diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 8b7c862e..4a3a52f2 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -159,11 +159,18 @@ defmodule Microwaveprop.Propagation do result end - @doc "Remove scores with valid_times older than 2 hours." + @doc """ + Remove scores with valid_times older than 2 hours. Called on a cron by + `Microwaveprop.Workers.PropagationPruneWorker` and also at the start of + each `PropagationGridWorker.perform/1` as a safety net. The 5-minute + timeout gives enough headroom for a catch-up run (potentially millions of + rows across four indexes) after a stretch of failed compute jobs. + """ def prune_old_scores do cutoff = DateTime.add(DateTime.utc_now(), -2, :hour) - {deleted, _} = Repo.delete_all(from(gs in GridScore, where: gs.valid_time < ^cutoff), timeout: 120_000) + {deleted, _} = + Repo.delete_all(from(gs in GridScore, where: gs.valid_time < ^cutoff), timeout: 300_000) if deleted > 0 do Logger.info("PropagationScores: pruned #{deleted} old scores (before #{cutoff})") diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 31c8af6b..60a3bdf9 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -24,6 +24,11 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do def perform(%Oban.Job{}) do t_start = System.monotonic_time(:millisecond) + # Safety net: prune before compute so a row of failed runs doesn't let + # stale scores accumulate indefinitely. Pruning also runs on its own cron + # (`PropagationPruneWorker`) independent of this worker's success. + Propagation.prune_old_scores() + # HRRR takes ~45min to publish after the hour. Use 2 hours ago to ensure availability. two_hours_ago = DateTime.add(DateTime.utc_now(), -2, :hour) run_time = two_hours_ago |> HrrrClient.nearest_hrrr_hour() |> DateTime.truncate(:second) diff --git a/lib/microwaveprop/workers/propagation_prune_worker.ex b/lib/microwaveprop/workers/propagation_prune_worker.ex new file mode 100644 index 00000000..a9c563af --- /dev/null +++ b/lib/microwaveprop/workers/propagation_prune_worker.ex @@ -0,0 +1,24 @@ +defmodule Microwaveprop.Workers.PropagationPruneWorker do + @moduledoc """ + Standalone worker that prunes stale rows from `propagation_scores`. + + Pruning used to be tacked on to the end of `PropagationGridWorker.perform/1`, + which meant it only ran after a successful grid compute. When the compute + worker was being killed mid-run (OOM / SIGTERM), prune never ran and the + table grew unbounded. Running prune on its own cron keeps the table healthy + regardless of the compute worker's state. + """ + + use Oban.Worker, + queue: :propagation, + max_attempts: 3, + unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]] + + alias Microwaveprop.Propagation + + @impl Oban.Worker + def perform(%Oban.Job{}) do + Propagation.prune_old_scores() + :ok + end +end diff --git a/test/microwaveprop/workers/propagation_prune_worker_test.exs b/test/microwaveprop/workers/propagation_prune_worker_test.exs new file mode 100644 index 00000000..29607cf8 --- /dev/null +++ b/test/microwaveprop/workers/propagation_prune_worker_test.exs @@ -0,0 +1,51 @@ +defmodule Microwaveprop.Workers.PropagationPruneWorkerTest do + use Microwaveprop.DataCase, async: true + + alias Microwaveprop.Propagation.GridScore + alias Microwaveprop.Repo + alias Microwaveprop.Workers.PropagationPruneWorker + + @factors %{ + "humidity" => 85, + "time_of_day" => 60, + "td_depression" => 70, + "refractivity" => 55, + "sky" => 80, + "season" => 65, + "wind" => 50, + "rain" => 90, + "pressure" => 75 + } + + defp insert_score!(valid_time, lat \\ 32.875, lon \\ -97.0) do + Repo.insert!(%GridScore{lat: lat, lon: lon, valid_time: valid_time, band_mhz: 10_000, score: 72, factors: @factors}) + end + + describe "perform/1" do + test "deletes scores with valid_time older than 2 hours" do + now = DateTime.truncate(DateTime.utc_now(), :second) + old = DateTime.add(now, -3, :hour) + fresh = DateTime.add(now, -30, :minute) + future = DateTime.add(now, 6, :hour) + + old_score = insert_score!(old, 32.0, -97.0) + fresh_score = insert_score!(fresh, 32.1, -97.1) + future_score = insert_score!(future, 32.2, -97.2) + + assert :ok = PropagationPruneWorker.perform(%Oban.Job{}) + + refute Repo.get(GridScore, old_score.id) + assert Repo.get(GridScore, fresh_score.id) + assert Repo.get(GridScore, future_score.id) + end + + test "is a no-op when nothing is stale" do + now = DateTime.truncate(DateTime.utc_now(), :second) + fresh = DateTime.add(now, -30, :minute) + score = insert_score!(fresh) + + assert :ok = PropagationPruneWorker.perform(%Oban.Job{}) + assert Repo.get(GridScore, score.id) + end + end +end