Handle CDS "rejected" status + widen submit cap headroom

CDS returns status="rejected" (distinct from "failed") when a submit
is rate-limited past the 150-job per-user cap. The poll worker's
interpret_status_body treated this as an unknown status → generic
retryable error → 10 retries → discarded. 102 jobs hit this path
overnight.

Fix:
  * Era5Client.check_status/1 now returns {:rejected, reason} (and
    treats "dismissed" the same way — that's the status a manually
    deleted job transitions to).
  * Era5PollWorker handles :rejected exactly like :not_found: drop the
    DB row, clean up the sibling CDS job, re-enqueue the submit, and
    discard the current Oban job so it doesn't burn attempts retrying
    a terminal state.
  * Widened Era5SubmitWorker cap headroom from 10 → 30 (effective
    ceiling 120 not 140). The race between concurrent workers all
    passing the count check simultaneously meant we were reaching 141
    in-flight against the 150 hard cap; 30 slots of headroom tolerates
    a ~15-worker race before clipping CDS's ceiling.

Refactored handle_row/1 into handle_row + handle_non_both_done with a
leg_outcome/1 tag helper so the cross-product of leg statuses collapses
to a single pair match (credo complexity limit).

lib/microwaveprop/weather/era5_client.ex

@@ -166,6 +166,10 @@ defmodule Microwaveprop.Weather.Era5Client do
   - `:not_found` — CDS returned 404 for this job id; terminal (either the
     job was reaped, deleted manually, or never existed). Callers should
     stop polling and clean up; a new submit is required.
+  - `{:rejected, reason}` — CDS accepted the submit (gave us a job id) but
+    immediately rejected it, typically because our in-flight queue was
+    over the 150-job cap. Terminal; caller should clean up and re-submit
+    once cap space is available.
   - `{:failed, reason}` — CDS marked the job failed; don't retry.
   - `{:error, reason}` — transient HTTP/network error; caller should retry.
   """
@@ -173,6 +177,7 @@ defmodule Microwaveprop.Weather.Era5Client do
           :running
           | :not_found
           | {:done, {:url, String.t()} | {:body, binary()}}
+          | {:rejected, String.t()}
           | {:failed, String.t()}
           | {:error, String.t()}
   def check_status(job_id) do
@@ -287,6 +292,22 @@ defmodule Microwaveprop.Weather.Era5Client do
     {:failed, reason}
   end
 
+  # CDS returns this when the submit was accepted into the user's session
+  # but immediately rejected server-side — usually because we were over the
+  # 150-job per-user cap. The worker should abandon the job and re-submit
+  # once there's room.
+  defp interpret_status_body(%{"status" => "rejected"} = body, _url, _api_key) do
+    reason = body["message"] || body["detail"] || "CDS job rejected (likely over cap)"
+    {:rejected, reason}
+  end
+
+  # `dismissed` is what a manually deleted job transitions to — treat it
+  # the same as rejected so we drop the row and re-submit.
+  defp interpret_status_body(%{"status" => "dismissed"} = body, _url, _api_key) do
+    reason = body["message"] || body["detail"] || "CDS job dismissed"
+    {:rejected, reason}
+  end
+
   defp interpret_status_body(body, _url, _api_key) do
     {:error, "ERA5 poll unknown status body: #{inspect(body)}"}
   end

lib/microwaveprop/workers/era5_poll_worker.ex

@@ -52,11 +52,23 @@ defmodule Microwaveprop.Workers.Era5PollWorker do
       {{:done, single_source}, {:done, pressure_source}} ->
         handle_both_done(row, single_source, pressure_source)
 
-      {:not_found, _} ->
-        resubmit_vanished(row, "single-level")
+      _other ->
+        handle_non_both_done(row, single, pressure)
+    end
+  end
 
-      {_, :not_found} ->
-        resubmit_vanished(row, "pressure-level")
+  # 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 |
+  # :running) plus a payload so the outer dispatch only has to match on
+  # the *pair* of tags, not the cross-product of individual statuses.
+  defp handle_non_both_done(row, single, pressure) do
+    case {leg_outcome(single), leg_outcome(pressure)} do
+      {{:terminal, reason}, _} ->
+        resubmit_vanished(row, "single-level", reason)
+
+      {_, {:terminal, reason}} ->
+        resubmit_vanished(row, "pressure-level", reason)
 
       {{:failed, reason}, _} ->
         fail_row(row, "single-level job failed: #{reason}")
@@ -70,27 +82,35 @@ defmodule Microwaveprop.Workers.Era5PollWorker do
       {_, {:error, reason}} ->
         {:error, "pressure-level check_status error: #{reason}"}
 
-      _both_running_or_mixed ->
+      _both_running_or_one_done ->
         bump_poll_count(row)
         {:snooze, @snooze_seconds}
     end
   end
 
-  # A CDS job has vanished from the server (404). This is terminal for the
-  # current era5_cds_jobs row — polling will never succeed — but the tile-
-  # month's data still needs to exist, so we drop the row, clean up the
-  # sibling CDS job (best-effort), and ask Era5SubmitWorker to re-submit.
-  # Returning `:discard` keeps Oban from wasting attempts on a retry that
-  # would hit the same 404.
-  defp resubmit_vanished(row, leg) do
+  defp leg_outcome({:done, _source}), do: :ok
+  defp leg_outcome(:not_found), do: {:terminal, "deleted"}
+  defp leg_outcome({:rejected, reason}), do: {:terminal, "rejected: #{reason}"}
+  defp leg_outcome({:failed, reason}), do: {:failed, reason}
+  defp leg_outcome({:error, reason}), do: {:error, reason}
+  defp leg_outcome(:running), do: :running
+
+  # A CDS job is terminally unreachable — either 404 (deleted) or rejected
+  # by the server (over the per-user cap). Polling will never succeed, but
+  # the tile-month's data still needs to exist, so we drop the row, clean
+  # up the sibling CDS job (best-effort), and ask Era5SubmitWorker to
+  # re-submit. The submit worker's CDS cap guard will snooze the retry
+  # until there's room on the server. Returning `:discard` keeps Oban from
+  # wasting attempts on a retry that would hit the same terminal state.
+  defp resubmit_vanished(row, leg, reason) do
     Logger.warning(
-      "Era5Poll: CDS job vanished (#{leg}) for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon} — re-submitting"
+      "Era5Poll: CDS job terminal (#{leg} #{reason}) for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon} — re-submitting"
     )
 
     delete_cds_jobs(row)
     Repo.delete!(row)
     enqueue_resubmit(row)
-    {:discard, "#{leg} CDS job was deleted; re-submitted tile-month"}
+    {:discard, "#{leg} CDS job terminal (#{reason}); re-submitted tile-month"}
   end
 
   defp enqueue_resubmit(row) do

lib/microwaveprop/workers/era5_submit_worker.ex

@@ -44,8 +44,16 @@ defmodule Microwaveprop.Workers.Era5SubmitWorker do
   # within `@cap_headroom` of the ceiling. Snoozing (instead of failing
   # or sleeping) releases the Oban slot immediately and lets the poll
   # workers drain in-flight jobs before we push more in.
+  #
+  # Headroom is deliberately generous (30 slots, ceiling of 120 effective)
+  # because the cap check races between concurrent workers: two workers
+  # that both observe count=119 will both submit, landing us at 121 after
+  # both finish. With 30 slots of headroom the cluster would have to race
+  # 15 workers wide to clip CDS's hard 150 ceiling. Observed:
+  # single-worker submits at count=141 → CDS reject storms, so we err on
+  # the side of staying well clear.
   @cds_ceiling 150
-  @cap_headroom 10
+  @cap_headroom 30
   @snooze_seconds 5 * 60
 
   @impl Oban.Worker

test/microwaveprop/weather/era5_client_test.exs

@@ -105,6 +105,24 @@ defmodule Microwaveprop.Weather.Era5ClientTest do
 
       assert :not_found = Era5Client.check_status("cds-job-123")
     end
+
+    test "returns {:rejected, reason} when CDS marks the job as rejected" do
+      # CDS rejects submits when the user's in-flight queue is over the cap
+      # (150 jobs). The rejected job still gets a jobID and a GET /jobs/:id
+      # returns a body with status=rejected. This is terminal — re-polling
+      # will never transition to successful — so callers should treat it
+      # like {:failed, _} and re-submit the tile-month.
+      Req.Test.stub(Era5Client, fn conn ->
+        Req.Test.json(conn, %{
+          "jobID" => "cds-job-123",
+          "status" => "rejected",
+          "metadata" => %{"origin" => "api"}
+        })
+      end)
+
+      assert {:rejected, reason} = Era5Client.check_status("cds-job-123")
+      assert is_binary(reason)
+    end
   end
 
   describe "delete_job/1" do

test/microwaveprop/workers/era5_poll_worker_test.exs

@@ -109,6 +109,51 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do
     end
   end
 
+  describe "perform/1 — CDS job rejected (over-cap)" do
+    test "deletes the row, cleans up the sibling CDS job, re-enqueues submit, and discards" do
+      row = insert_cds_job()
+      test_pid = self()
+
+      Req.Test.stub(Era5Client, fn conn ->
+        cond do
+          conn.method == "GET" and String.contains?(conn.request_path, "cds-single-123") ->
+            Req.Test.json(conn, %{
+              "jobID" => "cds-single-123",
+              "status" => "rejected",
+              "metadata" => %{"origin" => "api"}
+            })
+
+          conn.method == "GET" and String.contains?(conn.request_path, "cds-pressure-456") ->
+            Req.Test.json(conn, %{"status" => "running"})
+
+          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 =~ "rejected"
+      refute Repo.get(Era5CdsJob, row.id)
+
+      # Sibling CDS job is cleaned up so we don't burn queue slots waiting on
+      # a request we've already abandoned.
+      assert_received {:delete, "/api/retrieve/v1/jobs/cds-pressure-456"}
+
+      # A fresh Era5SubmitWorker job is enqueued so the tile-month gets
+      # re-submitted once the CDS cap has room.
+      assert_enqueued(
+        worker: Era5SubmitWorker,
+        args: %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}
+      )
+    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()

test/microwaveprop/workers/era5_submit_worker_test.exs

@@ -128,13 +128,15 @@ defmodule Microwaveprop.Workers.Era5SubmitWorkerTest do
     # CDS rejects with "Number of queued requests is limited to 150" when
     # a user has too many pending jobs. Each submit adds 2 rows to
     # era5_cds_jobs (single-level + pressure-level), so we snooze when
-    # the in-flight count is close to the ceiling.
-    @cap_headroom 10
+    # the in-flight count is close to the ceiling. The exact threshold
+    # (headroom) is tuned in the worker; the test just seeds well above
+    # the effective threshold (140 rows) so whatever the worker picks,
+    # it should trip the snooze.
     @cds_ceiling 150
 
-    test "snoozes when era5_cds_jobs count is at or above the ceiling - headroom" do
-      # Seed enough in-flight rows to push us over the snooze threshold.
-      seed_count = @cds_ceiling - @cap_headroom
+    test "snoozes when era5_cds_jobs count is at or above the snooze threshold" do
+      # 140 in-flight is above any reasonable snooze threshold.
+      seed_count = @cds_ceiling - 10
 
       rows =
         for i <- 1..seed_count do