prop/test/microwaveprop/workers/era5_submit_worker_test.exs
Graham McIntire 1ec10bec1f
Split ERA5 backfill into submit/poll workers with persistent CDS state
The single Era5MonthBatchWorker pinned an Oban slot for the full 30-45
min a CDS month-tile takes to assemble, and lost the work entirely on a
rolling deploy because the in-flight Task died with the pod. This splits
the flow into two tiny workers and persists the CDS job IDs so deploys
survive.

New pipeline:

1. Era5SubmitWorker (:era5_submit queue) — POSTs both CDS requests in
   parallel, writes an `era5_cds_jobs` row with the returned job IDs,
   and enqueues an Era5PollWorker scheduled +5 min. ~1s of real work.
   Short-circuits when the month-tile is already cached or when a row
   already exists (in-flight from a previous attempt).

2. Era5PollWorker (:era5_poll queue) — reads the row, calls
   Era5Client.check_status/1 for both CDS job IDs, and:
     - returns {:snooze, 300} if either job is still running (Oban
       re-schedules without counting an attempt and releases the slot
       immediately — a pod can keep dozens of tile-months in flight
       without pinning workers)
     - streams both GRIB files to disk via Req into: File.stream!,
       decodes via Wgrib2.extract_grid_messages_from_file, bulk-inserts
       via Era5BatchClient.decode_and_insert/6, deletes the row, and
       DELETEs both completed jobs from CDS to free server-side quota
     - if either leg CDS-reports failed, deletes the row + both CDS
       jobs and returns {:error, reason}

Era5Client gains four testable building blocks:
  submit_job/2               (bare POST → {:ok, job_id})
  check_status/1             (GET → :running | {:done, src} | {:failed, reason})
  download_source_to_file/3  (streams {:url, href} or writes {:body, bin})
  delete_job/1               (DELETE /jobs/:id, treats 200/202/204/404 as :ok)

All Req calls now route through `era5_req_options` so tests can stub
CDS responses via Req.Test.stub(Era5Client, fn).

Era5MonthBatchWorker is retained as a thin forwarder to Era5SubmitWorker
so any jobs already in the :era5_batch queue on prod pods drain cleanly
on the next rolling deploy. Safe to delete in a follow-up.

Adds era5_cds_jobs table with a unique index on
(year, month, tile_lat, tile_lon) so duplicate submits collapse.

New queue config in runtime.exs:
  era5_submit: local_limit 4, rate_limit 30/hour (burst protection)
  era5_poll:   local_limit 20 (polls are cheap GETs)
  era5_batch:  kept at 1 for legacy job drain, delete next cycle
2026-04-13 16:26:26 -05:00

136 lines
4.6 KiB
Elixir

defmodule Microwaveprop.Workers.Era5SubmitWorkerTest do
use Microwaveprop.DataCase, async: false
use Oban.Testing, repo: Microwaveprop.Repo
import Ecto.Query
alias Microwaveprop.Weather.Era5CdsJob
alias Microwaveprop.Weather.Era5Client
alias Microwaveprop.Weather.Era5Profile
alias Microwaveprop.Workers.Era5PollWorker
alias Microwaveprop.Workers.Era5SubmitWorker
@default_args %{"year" => 2014, "month" => 3, "tile_lat" => 32, "tile_lon" => -98}
setup do
System.put_env("ERA5_CDS_API_KEY", "test-key")
on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
:ok
end
describe "perform/1 — short-circuit paths" do
test "returns {:ok, :cached} when the month-tile already has era5_profiles" do
# Seed a profile inside the (2014-03, tile 32,-98) window.
%Era5Profile{}
|> Era5Profile.changeset(%{
lat: 32.5,
lon: -97.25,
valid_time: ~U[2014-03-15 12:00:00Z],
profile: []
})
|> Repo.insert!()
Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted on cache hit" end)
Oban.Testing.with_testing_mode(:manual, fn ->
assert {:ok, :cached} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
assert [] = all_enqueued(worker: Era5PollWorker)
end)
end
test "returns {:ok, :already_submitted} when an era5_cds_jobs row already exists" do
{:ok, _} =
%Era5CdsJob{}
|> Era5CdsJob.changeset(%{
year: 2014,
month: 3,
tile_lat: 32,
tile_lon: -98,
single_job_id: "pre-existing-single",
pressure_job_id: "pre-existing-pressure",
submitted_at: DateTime.truncate(DateTime.utc_now(), :second)
})
|> Repo.insert()
Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted on duplicate submit" end)
Oban.Testing.with_testing_mode(:manual, fn ->
assert {:ok, :already_submitted} =
Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
end)
end
end
describe "perform/1 — happy path" do
test "submits both CDS jobs, persists the row, and enqueues a poll worker" do
Req.Test.stub(Era5Client, fn conn ->
assert conn.method == "POST"
dataset_id =
cond do
String.contains?(conn.request_path, "single-levels") -> "single-job-123"
String.contains?(conn.request_path, "pressure-levels") -> "pressure-job-456"
end
Req.Test.json(conn, %{"jobID" => dataset_id})
end)
Oban.Testing.with_testing_mode(:manual, fn ->
assert {:ok, %Era5CdsJob{} = row} =
Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
assert row.single_job_id == "single-job-123"
assert row.pressure_job_id == "pressure-job-456"
assert row.year == 2014
assert row.month == 3
assert row.tile_lat == 32
assert row.tile_lon == -98
# Row is persisted in the DB.
assert Repo.aggregate(
from(j in Era5CdsJob,
where: j.year == 2014 and j.month == 3 and j.tile_lat == 32 and j.tile_lon == -98
),
:count,
:id
) == 1
# A poll worker was enqueued for this row, scheduled ~5 min from now.
[poll_job] = all_enqueued(worker: Era5PollWorker)
assert poll_job.args["era5_cds_job_id"] == row.id
assert DateTime.after?(poll_job.scheduled_at, DateTime.utc_now())
end)
end
test "returns {:error, _} and persists no row if either CDS submit fails" do
Req.Test.stub(Era5Client, fn conn ->
if String.contains?(conn.request_path, "pressure-levels") do
Plug.Conn.resp(conn, 400, ~s({"detail":"bad pressure request"}))
else
Req.Test.json(conn, %{"jobID" => "single-ok"})
end
end)
Oban.Testing.with_testing_mode(:manual, fn ->
assert {:error, reason} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
assert is_binary(reason)
# No orphaned row.
assert Repo.aggregate(Era5CdsJob, :count, :id) == 0
# No poll worker enqueued.
assert [] = all_enqueued(worker: Era5PollWorker)
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 ->
{:ok, j1} = Oban.insert(Era5SubmitWorker.new(@default_args))
{:ok, j2} = Oban.insert(Era5SubmitWorker.new(@default_args))
assert j1.id == j2.id
end)
end
end
end