prop/test/microwaveprop/workers/era5_poll_worker_test.exs
Graham McIntire 993a136ad7
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.
2026-04-13 16:44:01 -05:00

142 lines
4.8 KiB
Elixir

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")
on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
:ok
end
defp insert_cds_job(overrides \\ %{}) do
attrs =
Map.merge(
%{
year: 2014,
month: 3,
tile_lat: 32,
tile_lon: -98,
single_job_id: "cds-single-123",
pressure_job_id: "cds-pressure-456",
submitted_at: ~U[2026-04-13 21:00:00Z]
},
overrides
)
{:ok, row} = %Era5CdsJob{} |> Era5CdsJob.changeset(attrs) |> Repo.insert()
row
end
describe "perform/1 — row lifecycle" do
test "returns :ok and does nothing when the era5_cds_jobs row no longer exists" do
# Row was already processed/deleted by a previous poll run; snooze is
# scheduled behind us in time; just no-op.
Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted" end)
missing_id = Ecto.UUID.generate()
assert :ok =
Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => missing_id}})
end
end
describe "perform/1 — still running" do
test "returns {:snooze, seconds} and bumps poll_count when either CDS job is still running" do
row = insert_cds_job()
Req.Test.stub(Era5Client, fn conn ->
assert conn.method == "GET"
Req.Test.json(conn, %{"status" => "running"})
end)
assert {:snooze, snooze_s} =
Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
assert snooze_s > 0
reloaded = Repo.get!(Era5CdsJob, row.id)
assert reloaded.poll_count == 1
assert reloaded.last_polled_at
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()
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, %{"status" => "failed", "message" => "bad input"})
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)
assert {:error, reason} =
Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
assert reason =~ "bad input"
refute Repo.get(Era5CdsJob, row.id)
# Both CDS jobs are deleted to free server-side quota.
assert_received {:delete, "/api/retrieve/v1/jobs/cds-single-123"}
assert_received {:delete, "/api/retrieve/v1/jobs/cds-pressure-456"}
end
end
end