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

@@ -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}
      ]}
   ]
 

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}
      ]}
   ]

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"]}}
        ]}

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})")

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)

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

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