Abandon ERA5 jobs stuck in CDS queue > 4h

Observed in prod: after the CDS cap guard shipped, jobs started
landing but one leg per tile-month would routinely stay in 'accepted'
state for 16+ hours while the other completed in ~90 minutes. CDS's
per-user queue fairness apparently serializes competing leg requests
and the single-level ones got stranded behind higher-priority work.

Add an age-based stuck detector in Era5PollWorker: if a row has been
submitted for more than 4 hours and neither leg is fully done, treat
both legs as terminal, clean them up from CDS, and re-enqueue the
submit. Re-uses the resubmit_vanished path. 4h is ~8× the normal
completion time so transient queue variance doesn't trip it, but
tight enough that stuck jobs self-heal in a working session.

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 |

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()