diff --git a/lib/microwaveprop/workers/era5_poll_worker.ex b/lib/microwaveprop/workers/era5_poll_worker.ex index d23ef8d6..5edf6c17 100644 --- a/lib/microwaveprop/workers/era5_poll_worker.ex +++ b/lib/microwaveprop/workers/era5_poll_worker.ex @@ -32,6 +32,16 @@ defmodule Microwaveprop.Workers.Era5PollWorker do # so 5 minutes strikes a balance between responsiveness and noise. @snooze_seconds 5 * 60 + # A CDS job that's been in 'accepted' state for more than this is stuck + # in the server's queue and almost certainly won't run. (Observed in + # prod: pressure-level completes in ~90 min while single-level sits at + # 'accepted' for 16+ hours.) Once a row ages past this threshold we + # abandon the in-flight jobs, delete them from CDS, and re-submit the + # tile-month. 4 hours is ~8× the normal completion time — well above + # normal queue variance but tight enough that stuck jobs self-heal + # within a working session. + @stuck_after_seconds 4 * 3600 + @impl Oban.Worker def perform(%Oban.Job{args: %{"era5_cds_job_id" => id}}) do case Repo.get(Era5CdsJob, id) do @@ -53,10 +63,23 @@ defmodule Microwaveprop.Workers.Era5PollWorker do handle_both_done(row, single_source, pressure_source) _other -> - handle_non_both_done(row, single, pressure) + if stuck?(row) do + resubmit_vanished(row, "both", "stuck in CDS queue > 4h") + else + handle_non_both_done(row, single, pressure) + end end end + # True when the row is older than @stuck_after_seconds AND hasn't + # already landed both legs. CDS jobs that sit in 'accepted' state past + # this threshold are effectively dead and we should give up rather + # than polling forever. + defp stuck?(row) do + age_seconds = DateTime.diff(DateTime.utc_now(), row.submitted_at, :second) + age_seconds > @stuck_after_seconds + end + # The non-both-done dispatch is split out to keep handle_row/1 under # credo's cyclomatic complexity limit. `leg_outcome/1` collapses each # leg's status into a simple tag (:ok | :terminal | :failed | :error | diff --git a/test/microwaveprop/workers/era5_poll_worker_test.exs b/test/microwaveprop/workers/era5_poll_worker_test.exs index a69f0ff5..e243859e 100644 --- a/test/microwaveprop/workers/era5_poll_worker_test.exs +++ b/test/microwaveprop/workers/era5_poll_worker_test.exs @@ -14,6 +14,12 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do end defp insert_cds_job(overrides \\ %{}) do + # Default to "just now" so the stuck-after-4h guard doesn't misfire + # and re-submit in tests that want to exercise status-specific + # branches. Tests that care about the stuck path pass a stale + # submitted_at explicitly. + fresh_submit = DateTime.truncate(DateTime.utc_now(), :second) + attrs = Map.merge( %{ @@ -23,7 +29,7 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do tile_lon: -98, single_job_id: "cds-single-123", pressure_job_id: "cds-pressure-456", - submitted_at: ~U[2026-04-13 21:00:00Z] + submitted_at: fresh_submit }, overrides ) @@ -154,6 +160,67 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do end end + describe "perform/1 — stuck in CDS queue" do + # A CDS job that's been in 'accepted' state for many hours is stuck in + # the server's queue and unlikely to ever run. We track submitted_at + # on era5_cds_jobs, so once the row ages past @stuck_after, poll + # treats it as terminal and re-submits — same path as :not_found / + # :rejected. + test "re-submits when submitted_at > 4h ago and at least one leg is still running" do + stale_submit = + DateTime.utc_now() + |> DateTime.add(-5 * 3600, :second) + |> DateTime.truncate(:second) + + row = insert_cds_job(%{submitted_at: stale_submit}) + test_pid = self() + + Req.Test.stub(Era5Client, fn conn -> + cond do + conn.method == "GET" -> + Req.Test.json(conn, %{"status" => "accepted"}) + + conn.method == "DELETE" -> + send(test_pid, {:delete, conn.request_path}) + Plug.Conn.resp(conn, 204, "") + end + end) + + result = + Oban.Testing.with_testing_mode(:manual, fn -> + Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}}) + end) + + assert {:discard, reason} = result + assert reason =~ "stuck" + refute Repo.get(Era5CdsJob, row.id) + + assert_enqueued( + worker: Era5SubmitWorker, + args: %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98} + ) + end + + test "does NOT re-submit when submitted_at is recent even if running" do + # A 30-minute-old row with both legs running is normal — snooze. + fresh_submit = + DateTime.utc_now() + |> DateTime.add(-30 * 60, :second) + |> DateTime.truncate(:second) + + row = insert_cds_job(%{submitted_at: fresh_submit}) + + Req.Test.stub(Era5Client, fn conn -> + Req.Test.json(conn, %{"status" => "running"}) + end) + + assert {:snooze, _} = + Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}}) + + assert Repo.get(Era5CdsJob, row.id) + end + end + describe "perform/1 — failure" do test "returns {:error, _}, deletes the row, and deletes both CDS jobs when a leg fails" do row = insert_cds_job()