From 063e9e3ae41e08ea7d2bc29f65125f62f62eec2f Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 23 Apr 2026 12:57:39 -0500 Subject: [PATCH] fix(grid-tasks): reclaim orphan running rows on hourly seed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../propagation/grid_task_enqueuer.ex | 59 +++++++++ .../workers/propagation_grid_worker.ex | 6 + lib/microwaveprop_web/live/status_live.ex | 2 +- .../propagation/grid_task_enqueuer_test.exs | 124 ++++++++++++++++++ 4 files changed, 190 insertions(+), 1 deletion(-) diff --git a/lib/microwaveprop/propagation/grid_task_enqueuer.ex b/lib/microwaveprop/propagation/grid_task_enqueuer.ex index 1a52cc9e..238cd957 100644 --- a/lib/microwaveprop/propagation/grid_task_enqueuer.ex +++ b/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, diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 1ce9a964..5de8f6ba 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/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) diff --git a/lib/microwaveprop_web/live/status_live.ex b/lib/microwaveprop_web/live/status_live.ex index 92402f75..cec75103 100644 --- a/lib/microwaveprop_web/live/status_live.ex +++ b/lib/microwaveprop_web/live/status_live.ex @@ -551,7 +551,7 @@ defmodule MicrowavepropWeb.StatusLive do Worker Running Queued - Retrying + Failed Done (1h) diff --git a/test/microwaveprop/propagation/grid_task_enqueuer_test.exs b/test/microwaveprop/propagation/grid_task_enqueuer_test.exs index 1d037191..b5850626 100644 --- a/test/microwaveprop/propagation/grid_task_enqueuer_test.exs +++ b/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