From 14a22843219a234f58dc1a38304547544051eb2d Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 14 Apr 2026 08:26:37 -0500 Subject: [PATCH] Handle CDS "rejected" status + widen submit cap headroom MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 21 ++++++++ lib/microwaveprop/workers/era5_poll_worker.ex | 48 +++++++++++++------ .../workers/era5_submit_worker.ex | 10 +++- .../weather/era5_client_test.exs | 18 +++++++ .../workers/era5_poll_worker_test.exs | 45 +++++++++++++++++ .../workers/era5_submit_worker_test.exs | 12 +++-- 6 files changed, 134 insertions(+), 20 deletions(-) diff --git a/lib/microwaveprop/weather/era5_client.ex b/lib/microwaveprop/weather/era5_client.ex index 2d3bbfda..af55f42a 100644 --- a/lib/microwaveprop/weather/era5_client.ex +++ b/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 diff --git a/lib/microwaveprop/workers/era5_poll_worker.ex b/lib/microwaveprop/workers/era5_poll_worker.ex index df76e940..d23ef8d6 100644 --- a/lib/microwaveprop/workers/era5_poll_worker.ex +++ b/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 diff --git a/lib/microwaveprop/workers/era5_submit_worker.ex b/lib/microwaveprop/workers/era5_submit_worker.ex index 45619606..a79ac12d 100644 --- a/lib/microwaveprop/workers/era5_submit_worker.ex +++ b/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 diff --git a/test/microwaveprop/weather/era5_client_test.exs b/test/microwaveprop/weather/era5_client_test.exs index d0161dd2..4423626a 100644 --- a/test/microwaveprop/weather/era5_client_test.exs +++ b/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 diff --git a/test/microwaveprop/workers/era5_poll_worker_test.exs b/test/microwaveprop/workers/era5_poll_worker_test.exs index 610f77fb..a69f0ff5 100644 --- a/test/microwaveprop/workers/era5_poll_worker_test.exs +++ b/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() diff --git a/test/microwaveprop/workers/era5_submit_worker_test.exs b/test/microwaveprop/workers/era5_submit_worker_test.exs index 493f530c..ec997e61 100644 --- a/test/microwaveprop/workers/era5_submit_worker_test.exs +++ b/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