diff --git a/lib/microwaveprop/workers/era5_submit_worker.ex b/lib/microwaveprop/workers/era5_submit_worker.ex index 3748d1cc..45619606 100644 --- a/lib/microwaveprop/workers/era5_submit_worker.ex +++ b/lib/microwaveprop/workers/era5_submit_worker.ex @@ -38,6 +38,16 @@ defmodule Microwaveprop.Workers.Era5SubmitWorker do @poll_delay_seconds 5 * 60 + # CDS enforces a server-side per-user cap of 150 in-flight jobs. Each + # successful submit adds 2 rows to era5_cds_jobs (single-level + + # pressure-level), so we stop submitting once the in-flight count is + # 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. + @cds_ceiling 150 + @cap_headroom 10 + @snooze_seconds 5 * 60 + @impl Oban.Worker def backoff(%Oban.Job{attempt: attempt}) do # The submit itself is cheap; retry fast so transient CDS/HTTP blips @@ -65,11 +75,24 @@ defmodule Microwaveprop.Workers.Era5SubmitWorker do {:ok, :already_submitted} + cds_queue_full?() -> + in_flight = Repo.aggregate(Era5CdsJob, :count, :id) + + Logger.info( + "Era5Submit: CDS in-flight cap reached (#{in_flight}/#{@cds_ceiling}) — snoozing #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}" + ) + + {:snooze, @snooze_seconds} + true -> submit_and_persist(year, month, tile_lat, tile_lon) end end + defp cds_queue_full? do + Repo.aggregate(Era5CdsJob, :count, :id) + 2 > @cds_ceiling - @cap_headroom + end + defp submit_and_persist(year, month, tile_lat, tile_lon) do days = Calendar.ISO.days_in_month(year, month) area = tile_area(tile_lat, tile_lon) diff --git a/test/microwaveprop/workers/era5_submit_worker_test.exs b/test/microwaveprop/workers/era5_submit_worker_test.exs index 4a91c762..493f530c 100644 --- a/test/microwaveprop/workers/era5_submit_worker_test.exs +++ b/test/microwaveprop/workers/era5_submit_worker_test.exs @@ -124,6 +124,88 @@ defmodule Microwaveprop.Workers.Era5SubmitWorkerTest do end end + describe "CDS in-flight cap" 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 + @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 + + rows = + for i <- 1..seed_count do + %{ + id: Ecto.UUID.generate(), + year: 2014, + month: 3, + tile_lat: 32, + tile_lon: -98 - i, + single_job_id: "seed-single-#{i}", + pressure_job_id: "seed-pressure-#{i}", + submitted_at: DateTime.truncate(DateTime.utc_now(), :second), + inserted_at: DateTime.truncate(DateTime.utc_now(), :second), + updated_at: DateTime.truncate(DateTime.utc_now(), :second), + poll_count: 0 + } + end + + Repo.insert_all(Era5CdsJob, rows) + + Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted when capped" end) + + Oban.Testing.with_testing_mode(:manual, fn -> + assert {:snooze, seconds} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args}) + assert is_integer(seconds) and seconds > 0 + end) + + # No new era5_cds_jobs row inserted for the requested tile. + refute Repo.one( + from j in Era5CdsJob, + where: j.year == 2014 and j.month == 3 and j.tile_lat == 32 and j.tile_lon == -98 + ) + end + + test "submits normally when in-flight count is well below the ceiling" do + # 10 in-flight — comfortably under the cap. + rows = + for i <- 1..10 do + %{ + id: Ecto.UUID.generate(), + year: 2013, + month: i, + tile_lat: 40, + tile_lon: -90, + single_job_id: "warm-single-#{i}", + pressure_job_id: "warm-pressure-#{i}", + submitted_at: DateTime.truncate(DateTime.utc_now(), :second), + inserted_at: DateTime.truncate(DateTime.utc_now(), :second), + updated_at: DateTime.truncate(DateTime.utc_now(), :second), + poll_count: 0 + } + end + + Repo.insert_all(Era5CdsJob, rows) + + Req.Test.stub(Era5Client, fn conn -> + dataset_id = + cond do + String.contains?(conn.request_path, "single-levels") -> "fresh-single" + String.contains?(conn.request_path, "pressure-levels") -> "fresh-pressure" + end + + Req.Test.json(conn, %{"jobID" => dataset_id}) + end) + + Oban.Testing.with_testing_mode(:manual, fn -> + assert {:ok, %Era5CdsJob{}} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args}) + end) + end + end + describe "unique constraint" do test "collapses duplicate (year, month, tile_lat, tile_lon) Oban enqueues" do Oban.Testing.with_testing_mode(:manual, fn ->