fix(grid-tasks): reclaim orphan running rows on hourly seed

Rust workers (prop-grid-rs) that die mid-claim (SIGKILL, OOM, node
drain) leave grid_tasks rows stuck in status='running' forever,
because claim_next uses FOR UPDATE SKIP LOCKED and nothing resets the
orphan. The /status page then shows a permanent spinner with stale
f00/f10/f18 badges — currently 21 rows claimed as far back as
2026-04-20.

GridTaskEnqueuer.reclaim_stale_running/1 flips rows whose claimed_at
is older than 15 minutes back to 'queued'. Rows that have already
burned through 5 claim/reclaim cycles become 'failed' with a
reclaim-orphan error so the next hourly seed can replace them.
Wired into PropagationGridWorker.seed_chain/0 so it runs every :05
cron tick before new rows are seeded.

Also rename the status panel "Retrying" column to "Failed" — it was
always showing terminal `failed` rows, never retrying ones.

lib/microwaveprop/propagation/grid_task_enqueuer.ex

@@ -10,12 +10,28 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
   unique index — re-seeding the same cycle is a no-op.
   """
 
+  import Ecto.Query
+
   alias Microwaveprop.Repo
 
   require Logger
 
   @max_forecast_hour 18
 
+  # A `running` row older than this is an orphan: the Rust worker that
+  # claimed it was SIGKILLed / OOMed / evicted before it could emit a
+  # terminal status. `claim_next` uses `FOR UPDATE SKIP LOCKED` so the
+  # row stays `running` indefinitely until something reclaims it.
+  # 15 min comfortably exceeds the ~3-4 min wall time of a healthy
+  # forecast-hour chain step; any row past that is stuck.
+  @stale_running_cutoff_seconds 15 * 60
+
+  # After this many claim/reclaim cycles, a row is permanently flipped
+  # to `failed` so the status-page spinner stops and the next hourly
+  # reseed can replace it. Rust's `claim_next` increments `attempt` on
+  # each pickup, so this is a total-attempts ceiling.
+  @max_reclaim_attempts 5
+
   @doc """
   Seed one kind='analysis' row (f00) plus 18 kind='forecast' rows
   (f01..f18) for `run_time`. This is the Phase 3 Stream A cutover
@@ -93,6 +109,49 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
     do_insert(rows, run_time)
   end
 
+  @doc """
+  Reclaim orphan `running` rows — Rust workers that died mid-claim
+  (SIGKILL, OOM, node drain) leave `status='running'` sticky, which
+  makes `claim_next` skip the row forever and the status panel shows
+  a permanent-looking spinner. Flip rows whose `claimed_at` is older
+  than `cutoff_seconds` back to `queued`. Rows that have already
+  burned through `@max_reclaim_attempts` become `failed` so the next
+  hourly seed replaces them instead of looping.
+
+  Called from `PropagationGridWorker.seed_chain/0` at :05 each hour.
+  """
+  @spec reclaim_stale_running(pos_integer()) :: %{requeued: non_neg_integer(), failed: non_neg_integer()}
+  def reclaim_stale_running(cutoff_seconds \\ @stale_running_cutoff_seconds) do
+    cutoff = DateTime.utc_now() |> DateTime.add(-cutoff_seconds, :second) |> DateTime.truncate(:second)
+    now = DateTime.truncate(DateTime.utc_now(), :second)
+
+    {failed_n, _} =
+      "grid_tasks"
+      |> where(
+        [t],
+        t.status == "running" and t.claimed_at < ^cutoff and t.attempt >= ^@max_reclaim_attempts
+      )
+      |> Repo.update_all(
+        set: [
+          status: "failed",
+          error: "reclaim-orphan: exceeded max_reclaim_attempts",
+          claimed_at: nil,
+          updated_at: now
+        ]
+      )
+
+    {requeued_n, _} =
+      "grid_tasks"
+      |> where([t], t.status == "running" and t.claimed_at < ^cutoff)
+      |> Repo.update_all(set: [status: "queued", claimed_at: nil, updated_at: now])
+
+    if failed_n > 0 or requeued_n > 0 do
+      Logger.info("GridTaskEnqueuer: reclaimed stale-running grid_tasks (requeued=#{requeued_n}, failed=#{failed_n})")
+    end
+
+    %{requeued: requeued_n, failed: failed_n}
+  end
+
   defp do_insert(rows, run_time) do
     {count, _} =
       Repo.insert_all("grid_tasks", rows,

lib/microwaveprop/workers/propagation_grid_worker.ex

@@ -39,6 +39,12 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
   end
 
   defp seed_chain do
+    # Reclaim any `running` grid_tasks rows orphaned by a Rust worker
+    # that died mid-claim (SIGKILL, OOM, node drain). Without this
+    # the status panel shows a permanent-looking spinner and the
+    # rows are never re-worked.
+    _ = GridTaskEnqueuer.reclaim_stale_running()
+
     # 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)

lib/microwaveprop_web/live/status_live.ex

@@ -551,7 +551,7 @@ defmodule MicrowavepropWeb.StatusLive do
               <th>Worker</th>
               <th>Running</th>
               <th>Queued</th>
-              <th>Retrying</th>
+              <th>Failed</th>
               <th>Done (1h)</th>
             </tr>
           </thead>

test/microwaveprop/propagation/grid_task_enqueuer_test.exs

@@ -50,4 +50,128 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuerTest do
 
     assert f0_count == 0
   end
+
+  describe "reclaim_stale_running/1" do
+    test "flips stale-running rows back to queued and clears claimed_at" do
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      stale = DateTime.add(now, -30 * 60, :second)
+
+      insert_task(%{
+        run_time: ~U[2026-04-19 20:00:00Z],
+        forecast_hour: 5,
+        valid_time: ~U[2026-04-20 01:00:00Z],
+        status: "running",
+        attempt: 1,
+        claimed_at: stale
+      })
+
+      assert %{requeued: 1, failed: 0} = GridTaskEnqueuer.reclaim_stale_running()
+
+      row =
+        Repo.one!(
+          from(t in "grid_tasks",
+            where: t.run_time == ^~U[2026-04-19 20:00:00Z] and t.forecast_hour == 5,
+            select: %{status: t.status, claimed_at: t.claimed_at, attempt: t.attempt}
+          )
+        )
+
+      assert row.status == "queued"
+      assert is_nil(row.claimed_at)
+      # attempt is NOT incremented — claim_next does that on re-claim.
+      assert row.attempt == 1
+    end
+
+    test "leaves recently-claimed running rows untouched" do
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      fresh = DateTime.add(now, -60, :second)
+
+      insert_task(%{
+        run_time: ~U[2026-04-19 21:00:00Z],
+        forecast_hour: 6,
+        valid_time: ~U[2026-04-20 03:00:00Z],
+        status: "running",
+        attempt: 1,
+        claimed_at: fresh
+      })
+
+      assert %{requeued: 0, failed: 0} = GridTaskEnqueuer.reclaim_stale_running()
+
+      row =
+        Repo.one!(
+          from(t in "grid_tasks",
+            where: t.run_time == ^~U[2026-04-19 21:00:00Z] and t.forecast_hour == 6,
+            select: %{status: t.status}
+          )
+        )
+
+      assert row.status == "running"
+    end
+
+    test "flips stale rows past max attempts to failed" do
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      stale = DateTime.add(now, -30 * 60, :second)
+
+      insert_task(%{
+        run_time: ~U[2026-04-19 22:00:00Z],
+        forecast_hour: 7,
+        valid_time: ~U[2026-04-20 05:00:00Z],
+        status: "running",
+        attempt: 5,
+        claimed_at: stale
+      })
+
+      assert %{requeued: 0, failed: 1} = GridTaskEnqueuer.reclaim_stale_running()
+
+      row =
+        Repo.one!(
+          from(t in "grid_tasks",
+            where: t.run_time == ^~U[2026-04-19 22:00:00Z] and t.forecast_hour == 7,
+            select: %{status: t.status, error: t.error, claimed_at: t.claimed_at}
+          )
+        )
+
+      assert row.status == "failed"
+      assert is_nil(row.claimed_at)
+      assert row.error =~ "reclaim"
+    end
+
+    test "does not touch queued or done rows even if claimed_at is ancient" do
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      stale = DateTime.add(now, -30 * 60, :second)
+
+      insert_task(%{
+        run_time: ~U[2026-04-19 23:00:00Z],
+        forecast_hour: 8,
+        valid_time: ~U[2026-04-20 07:00:00Z],
+        status: "done",
+        attempt: 1,
+        claimed_at: stale,
+        completed_at: stale
+      })
+
+      assert %{requeued: 0, failed: 0} = GridTaskEnqueuer.reclaim_stale_running()
+    end
+  end
+
+  defp insert_task(attrs) do
+    now = DateTime.truncate(DateTime.utc_now(), :microsecond)
+
+    row =
+      Map.merge(
+        %{
+          id: Ecto.UUID.bingenerate(),
+          status: "queued",
+          attempt: 0,
+          kind: "forecast",
+          claimed_at: nil,
+          completed_at: nil,
+          error: nil,
+          inserted_at: now,
+          updated_at: now
+        },
+        attrs
+      )
+
+    {1, _} = Repo.insert_all("grid_tasks", [row])
+  end
 end