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.
This commit is contained in:
parent
b343bbe680
commit
993a136ad7
4 changed files with 104 additions and 2 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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!()
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue