From 993a136ad7be8151fe79ae64c884eed7747f2717 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 13 Apr 2026 16:43:51 -0500 Subject: [PATCH] Handle vanished CDS jobs by re-submitting tile-month When CDS returns 404 for a poll check, the job is gone (manually deleted, reaped, or never existed). Retrying will always 404, so treat it as terminal: drop the era5_cds_jobs row, delete the sibling CDS job, re-enqueue Era5SubmitWorker for the tile-month, and discard the Oban poll job. Fixes 30 retryable poll jobs left behind by the interrupted manual-cleanup script. --- lib/microwaveprop/weather/era5_client.ex | 9 +++- lib/microwaveprop/workers/era5_poll_worker.ex | 35 ++++++++++++++ .../weather/era5_client_test.exs | 16 +++++++ .../workers/era5_poll_worker_test.exs | 46 +++++++++++++++++++ 4 files changed, 104 insertions(+), 2 deletions(-) diff --git a/lib/microwaveprop/weather/era5_client.ex b/lib/microwaveprop/weather/era5_client.ex index dbb3a5b7..2d3bbfda 100644 --- a/lib/microwaveprop/weather/era5_client.ex +++ b/lib/microwaveprop/weather/era5_client.ex @@ -163,11 +163,15 @@ defmodule Microwaveprop.Weather.Era5Client do - `:running` — job is queued or in progress; caller should retry later. - `{:done, {:url, href} | {:body, binary}}` — ready; pass the source to `download_source_to_file/3`. + - `: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. - `{:failed, reason}` — CDS marked the job failed; don't retry. - `{:error, reason}` — transient HTTP/network error; caller should retry. """ @spec check_status(String.t()) :: :running + | :not_found | {:done, {:url, String.t()} | {:body, binary()}} | {:failed, String.t()} | {:error, String.t()} @@ -257,6 +261,8 @@ defmodule Microwaveprop.Weather.Era5Client do interpret_status_body(body, url, api_key) end + defp interpret_status_response({:ok, %{status: 404}}, _url, _api_key), do: :not_found + defp interpret_status_response({:ok, %{status: status, body: body}}, _url, _api_key) do {:error, "ERA5 poll HTTP #{status}: #{inspect(body)}"} end @@ -272,8 +278,7 @@ defmodule Microwaveprop.Weather.Era5Client do end end - defp interpret_status_body(%{"status" => status}, _url, _api_key) - when status in ["accepted", "running"] do + defp interpret_status_body(%{"status" => status}, _url, _api_key) when status in ["accepted", "running"] do :running end diff --git a/lib/microwaveprop/workers/era5_poll_worker.ex b/lib/microwaveprop/workers/era5_poll_worker.ex index c117f8a9..df76e940 100644 --- a/lib/microwaveprop/workers/era5_poll_worker.ex +++ b/lib/microwaveprop/workers/era5_poll_worker.ex @@ -24,6 +24,7 @@ defmodule Microwaveprop.Workers.Era5PollWorker do alias Microwaveprop.Weather.Era5BatchClient alias Microwaveprop.Weather.Era5CdsJob alias Microwaveprop.Weather.Era5Client + alias Microwaveprop.Workers.Era5SubmitWorker require Logger @@ -51,6 +52,12 @@ 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") + + {_, :not_found} -> + resubmit_vanished(row, "pressure-level") + {{:failed, reason}, _} -> fail_row(row, "single-level job failed: #{reason}") @@ -69,6 +76,34 @@ defmodule Microwaveprop.Workers.Era5PollWorker do 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 + Logger.warning( + "Era5Poll: CDS job vanished (#{leg}) 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"} + end + + defp enqueue_resubmit(row) do + %{ + "year" => row.year, + "month" => row.month, + "tile_lat" => row.tile_lat, + "tile_lon" => row.tile_lon + } + |> Era5SubmitWorker.new() + |> Oban.insert() + end + defp handle_both_done(row, single_source, pressure_source) do tmp_dir = System.tmp_dir!() diff --git a/test/microwaveprop/weather/era5_client_test.exs b/test/microwaveprop/weather/era5_client_test.exs index d20fff6e..d0161dd2 100644 --- a/test/microwaveprop/weather/era5_client_test.exs +++ b/test/microwaveprop/weather/era5_client_test.exs @@ -89,6 +89,22 @@ defmodule Microwaveprop.Weather.Era5ClientTest do assert {:failed, reason} = Era5Client.check_status("cds-job-123") assert reason =~ "input out of range" end + + test "returns :not_found when CDS returns 404 for the job" do + # 404 means the CDS job was deleted or never existed. This is terminal — + # retrying will always 404 — so callers should treat it as a distinct + # state from the generic {:error, _} retryable bucket. + Req.Test.stub(Era5Client, fn conn -> + conn + |> Plug.Conn.put_resp_content_type("application/json") + |> Plug.Conn.resp( + 404, + ~s({"detail":"job cds-job-123 deleted","status":404,"title":"job not found"}) + ) + end) + + assert :not_found = Era5Client.check_status("cds-job-123") + 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 2b73074b..610f77fb 100644 --- a/test/microwaveprop/workers/era5_poll_worker_test.exs +++ b/test/microwaveprop/workers/era5_poll_worker_test.exs @@ -1,9 +1,11 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do use Microwaveprop.DataCase, async: false + use Oban.Testing, repo: Microwaveprop.Repo alias Microwaveprop.Weather.Era5CdsJob alias Microwaveprop.Weather.Era5Client alias Microwaveprop.Workers.Era5PollWorker + alias Microwaveprop.Workers.Era5SubmitWorker setup do System.put_env("ERA5_CDS_API_KEY", "test-key") @@ -63,6 +65,50 @@ defmodule Microwaveprop.Workers.Era5PollWorkerTest do end end + describe "perform/1 — CDS job vanished (404)" do + test "deletes the row, deletes the other CDS job, re-enqueues Era5SubmitWorker, 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") -> + conn + |> Plug.Conn.put_resp_content_type("application/json") + |> Plug.Conn.resp(404, ~s({"detail":"job deleted","title":"job not found"})) + + 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) + + # Run with manual Oban testing so the re-enqueued Era5SubmitWorker is + # inserted into the queue but not executed inline — we only want to + # assert that the row was deleted and the re-submit was scheduled. + 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 =~ "deleted" + refute Repo.get(Era5CdsJob, row.id) + + # The sibling CDS job gets cleaned up too so it doesn't linger on CDS. + assert_received {:delete, "/api/retrieve/v1/jobs/cds-pressure-456"} + + # A fresh Era5SubmitWorker job is enqueued so the month gets re-fetched. + 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()