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
This commit is contained in:
parent
f5bacf34d0
commit
1ec10bec1f
13 changed files with 1140 additions and 77 deletions
|
|
@ -167,19 +167,29 @@ if config_env() == :prod do
|
|||
terrain: 3,
|
||||
iemre: 3,
|
||||
era5: 2,
|
||||
# No global_limit: the Pro tracker leaked phantom slots across deploys
|
||||
# and stalled the whole backfill. The rate_limit protects CDS and the
|
||||
# per-pod local_limit scales naturally with replicas.
|
||||
# Split-worker ERA5 pipeline:
|
||||
#
|
||||
# Parallelism sizing: each job makes 2 CDS requests *serially* (single-
|
||||
# level then pressure-level), so each active worker holds 1 CDS slot at
|
||||
# a time. CDS-Beta's per-user ceiling is ~16 concurrent. local_limit 4
|
||||
# × 3 pods = 12 concurrent, leaving headroom. rate_limit 30/hour stays
|
||||
# above steady-state throughput so it doesn't re-become the bottleneck.
|
||||
era5_batch: [
|
||||
# `era5_submit` runs Era5SubmitWorker, which POSTs two CDS requests
|
||||
# and writes an era5_cds_jobs row. Each job is ~1s of real work, so
|
||||
# local_limit is the gate on how fast we burn CDS submission quota.
|
||||
# rate_limit protects CDS from submit bursts; the per-pod ceiling
|
||||
# (local_limit × replicas) determines peak concurrent in-flight.
|
||||
era5_submit: [
|
||||
local_limit: 4,
|
||||
rate_limit: [allowed: 30, period: {1, :hour}]
|
||||
],
|
||||
# `era5_poll` runs Era5PollWorker, which does a tiny GET per run and
|
||||
# either snoozes (releasing the slot) or streams the completed GRIB
|
||||
# and inserts. Needs high local_limit because in-flight CDS jobs
|
||||
# scale independently of concurrent poll work — a pod may have 50
|
||||
# tile-months in flight but be doing ~1 actual poll GET at a time.
|
||||
era5_poll: [
|
||||
local_limit: 20
|
||||
],
|
||||
# Legacy queue name retained briefly so any in-flight Era5MonthBatchWorker
|
||||
# jobs from the pre-split deploy drain cleanly (the worker now just
|
||||
# forwards to :era5_submit). Safe to delete after one rolling deploy.
|
||||
era5_batch: 1,
|
||||
rtma: 2,
|
||||
backfill_enqueue: 1,
|
||||
admin: 1,
|
||||
|
|
|
|||
|
|
@ -50,6 +50,7 @@ config :microwaveprop, Oban, testing: :inline
|
|||
config :microwaveprop, cache_contact_count: false
|
||||
config :microwaveprop, cache_contact_map: false
|
||||
config :microwaveprop, elevation_req_options: [plug: {Req.Test, Microwaveprop.Terrain.ElevationClient}, retry: false]
|
||||
config :microwaveprop, era5_req_options: [plug: {Req.Test, Microwaveprop.Weather.Era5Client}, retry: false]
|
||||
config :microwaveprop, hrrr_req_options: [plug: {Req.Test, Microwaveprop.Weather.HrrrClient}, retry: false]
|
||||
config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}, retry: false]
|
||||
|
||||
|
|
|
|||
|
|
@ -101,13 +101,8 @@ defmodule Microwaveprop.Weather.Era5BatchClient do
|
|||
|
||||
try do
|
||||
with :ok <- Task.await(single_task, :infinity),
|
||||
:ok <- Task.await(pressure_task, :infinity),
|
||||
{:ok, profiles} <- build_month_profiles(tile_lat, tile_lon, single_path, pressure_path) do
|
||||
inserted = bulk_insert_profiles(profiles)
|
||||
|
||||
Logger.info("ERA5Batch: stored #{inserted} profiles for #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}")
|
||||
|
||||
{:ok, inserted}
|
||||
:ok <- Task.await(pressure_task, :infinity) do
|
||||
decode_and_insert(year, month, tile_lat, tile_lon, single_path, pressure_path)
|
||||
end
|
||||
after
|
||||
Task.shutdown(single_task, :brutal_kill)
|
||||
|
|
@ -117,6 +112,31 @@ defmodule Microwaveprop.Weather.Era5BatchClient do
|
|||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Decodes a downloaded single-level + pressure-level GRIB2 pair from disk
|
||||
and bulk-inserts the resulting month-tile profiles into `era5_profiles`.
|
||||
|
||||
Shared between the legacy inline `do_fetch_month/4` path and the split
|
||||
`Era5PollWorker` path.
|
||||
"""
|
||||
@spec decode_and_insert(
|
||||
pos_integer(),
|
||||
pos_integer(),
|
||||
integer(),
|
||||
integer(),
|
||||
Path.t(),
|
||||
Path.t()
|
||||
) :: {:ok, non_neg_integer()} | {:error, term()}
|
||||
def decode_and_insert(year, month, tile_lat, tile_lon, single_path, pressure_path) do
|
||||
with {:ok, profiles} <- build_month_profiles(tile_lat, tile_lon, single_path, pressure_path) do
|
||||
inserted = bulk_insert_profiles(profiles)
|
||||
|
||||
Logger.info("ERA5Batch: stored #{inserted} profiles for #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}")
|
||||
|
||||
{:ok, inserted}
|
||||
end
|
||||
end
|
||||
|
||||
defp month_tile_cached?(year, month, tile_lat, tile_lon) do
|
||||
{first, last} = month_bounds(year, month)
|
||||
|
||||
|
|
|
|||
44
lib/microwaveprop/weather/era5_cds_job.ex
Normal file
44
lib/microwaveprop/weather/era5_cds_job.ex
Normal file
|
|
@ -0,0 +1,44 @@
|
|||
defmodule Microwaveprop.Weather.Era5CdsJob do
|
||||
@moduledoc """
|
||||
Tracks an in-flight ERA5 CDS request submitted by `Era5SubmitWorker` and
|
||||
awaiting download+decode by `Era5PollWorker`. Persisting the CDS job IDs
|
||||
decouples the Oban worker lifecycle from the 30+ minute CDS queue time,
|
||||
so a pod restart resumes polling instead of re-submitting.
|
||||
"""
|
||||
use Ecto.Schema
|
||||
|
||||
import Ecto.Changeset
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
|
||||
schema "era5_cds_jobs" do
|
||||
field :year, :integer
|
||||
field :month, :integer
|
||||
field :tile_lat, :integer
|
||||
field :tile_lon, :integer
|
||||
field :single_job_id, :string
|
||||
field :pressure_job_id, :string
|
||||
field :submitted_at, :utc_datetime
|
||||
field :last_polled_at, :utc_datetime
|
||||
field :poll_count, :integer, default: 0
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
@type t :: %__MODULE__{}
|
||||
|
||||
@required_fields ~w(year month tile_lat tile_lon single_job_id pressure_job_id submitted_at)a
|
||||
@optional_fields ~w(last_polled_at poll_count)a
|
||||
|
||||
@spec changeset(t() | Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()
|
||||
def changeset(job, attrs) do
|
||||
job
|
||||
|> cast(attrs, @required_fields ++ @optional_fields)
|
||||
|> validate_required(@required_fields)
|
||||
|> unique_constraint([:year, :month, :tile_lat, :tile_lon],
|
||||
name: :era5_cds_jobs_year_month_tile_lat_tile_lon_index,
|
||||
error_key: :tile_month
|
||||
)
|
||||
end
|
||||
end
|
||||
|
|
@ -135,13 +135,94 @@ defmodule Microwaveprop.Weather.Era5Client do
|
|||
@spec submit_and_download_to_file(String.t(), map(), Path.t()) :: :ok | {:error, term()}
|
||||
def submit_and_download_to_file(dataset, request, path) do
|
||||
with {:ok, source, api_key} <- submit_and_poll(dataset, request) do
|
||||
case source do
|
||||
{:body, body} -> File.write(path, body)
|
||||
{:url, href} -> download_href_to_file(href, api_key, path)
|
||||
download_source_to_file(source, api_key, path)
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Submits a CDS job and returns `{:ok, job_id}` without waiting for it to
|
||||
complete. Used by the split `Era5SubmitWorker` / `Era5PollWorker` flow so
|
||||
Oban worker slots aren't pinned to the 30+ minute CDS queue time.
|
||||
"""
|
||||
@spec submit_job(String.t(), map()) :: {:ok, String.t()} | {:error, String.t()}
|
||||
def submit_job(dataset, request) do
|
||||
api_key = api_key()
|
||||
|
||||
if is_nil(api_key) do
|
||||
{:error, "ERA5_CDS_API_KEY not configured"}
|
||||
else
|
||||
submit_request(dataset, request, api_key)
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Checks the status of a previously-submitted CDS job and, if complete,
|
||||
follows the results pointer to return the download source.
|
||||
|
||||
Returns:
|
||||
- `: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`.
|
||||
- `{: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
|
||||
| {:done, {:url, String.t()} | {:body, binary()}}
|
||||
| {:failed, String.t()}
|
||||
| {:error, String.t()}
|
||||
def check_status(job_id) do
|
||||
api_key = api_key()
|
||||
|
||||
if is_nil(api_key) do
|
||||
{:error, "ERA5_CDS_API_KEY not configured"}
|
||||
else
|
||||
do_check_status(job_id, api_key)
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Deletes a completed CDS job from the server so it doesn't count against
|
||||
our per-user job quota. Called by `Era5PollWorker` after a successful
|
||||
download. Treats 200/202/204/404 as success — 404 is fine because the
|
||||
job may already have been reaped by CDS's own cleanup.
|
||||
"""
|
||||
@spec delete_job(String.t()) :: :ok | {:error, String.t()}
|
||||
def delete_job(job_id) do
|
||||
api_key = api_key()
|
||||
|
||||
if is_nil(api_key) do
|
||||
{:error, "ERA5_CDS_API_KEY not configured"}
|
||||
else
|
||||
url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"
|
||||
|
||||
case Req.delete(
|
||||
url,
|
||||
[headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 30_000] ++ req_options()
|
||||
) do
|
||||
{:ok, %{status: status}} when status in [200, 202, 204, 404] -> :ok
|
||||
{:ok, %{status: status}} -> {:error, "ERA5 delete_job HTTP #{status}"}
|
||||
{:error, reason} -> {:error, "ERA5 delete_job failed: #{inspect(reason)}"}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Writes a CDS result source to `path`. `{:body, binary}` is written as a
|
||||
plain file; `{:url, href}` is streamed through Req's `into: File.stream!`
|
||||
so peak heap stays O(chunk) rather than O(full response).
|
||||
"""
|
||||
@spec download_source_to_file(
|
||||
{:body, binary()} | {:url, String.t()},
|
||||
String.t() | nil,
|
||||
Path.t()
|
||||
) :: :ok | {:error, String.t()}
|
||||
def download_source_to_file({:body, body}, _api_key, path), do: File.write(path, body)
|
||||
|
||||
def download_source_to_file({:url, href}, api_key, path) do
|
||||
download_href_to_file(href, api_key, path)
|
||||
end
|
||||
|
||||
defp do_submit_and_download(dataset, request) do
|
||||
with {:ok, source, api_key} <- submit_and_poll(dataset, request) do
|
||||
case source do
|
||||
|
|
@ -164,13 +245,57 @@ defmodule Microwaveprop.Weather.Era5Client do
|
|||
end
|
||||
end
|
||||
|
||||
defp do_check_status(job_id, api_key) do
|
||||
url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"
|
||||
|
||||
url
|
||||
|> Req.get([headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000] ++ req_options())
|
||||
|> interpret_status_response(url, api_key)
|
||||
end
|
||||
|
||||
defp interpret_status_response({:ok, %{status: 200, body: body}}, url, api_key) do
|
||||
interpret_status_body(body, url, api_key)
|
||||
end
|
||||
|
||||
defp interpret_status_response({:ok, %{status: status, body: body}}, _url, _api_key) do
|
||||
{:error, "ERA5 poll HTTP #{status}: #{inspect(body)}"}
|
||||
end
|
||||
|
||||
defp interpret_status_response({:error, reason}, _url, _api_key) do
|
||||
{:error, "ERA5 poll failed: #{inspect(reason)}"}
|
||||
end
|
||||
|
||||
defp interpret_status_body(%{"status" => "successful"}, url, api_key) do
|
||||
case fetch_results_source("#{url}/results", api_key) do
|
||||
{:ok, source} -> {:done, source}
|
||||
{:error, reason} -> {:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
defp interpret_status_body(%{"status" => status}, _url, _api_key)
|
||||
when status in ["accepted", "running"] do
|
||||
:running
|
||||
end
|
||||
|
||||
defp interpret_status_body(%{"status" => "failed"} = body, _url, _api_key) do
|
||||
reason = body["message"] || body["error"] || inspect(body)
|
||||
{:failed, reason}
|
||||
end
|
||||
|
||||
defp interpret_status_body(body, _url, _api_key) do
|
||||
{:error, "ERA5 poll unknown status body: #{inspect(body)}"}
|
||||
end
|
||||
|
||||
defp submit_request(dataset, request, api_key) do
|
||||
url = "#{@cds_url}/retrieve/v1/processes/#{dataset}/execution"
|
||||
|
||||
case Req.post(url,
|
||||
json: %{"inputs" => request},
|
||||
headers: [{"PRIVATE-TOKEN", api_key}],
|
||||
receive_timeout: 30_000
|
||||
case Req.post(
|
||||
url,
|
||||
[
|
||||
json: %{"inputs" => request},
|
||||
headers: [{"PRIVATE-TOKEN", api_key}],
|
||||
receive_timeout: 30_000
|
||||
] ++ req_options()
|
||||
) do
|
||||
{:ok, %{status: status, body: body}} when status in [200, 201, 202] ->
|
||||
job_id = body["jobID"] || body["request_id"] || body["jobId"]
|
||||
|
|
@ -201,7 +326,10 @@ defmodule Microwaveprop.Weather.Era5Client do
|
|||
defp poll_until_complete(job_id, api_key, attempt) do
|
||||
url = "#{@cds_url}/retrieve/v1/jobs/#{job_id}"
|
||||
|
||||
case Req.get(url, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000) do
|
||||
case Req.get(
|
||||
url,
|
||||
[headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 10_000] ++ req_options()
|
||||
) do
|
||||
{:ok, %{status: 200, body: %{"status" => "successful"}}} ->
|
||||
results_url = "#{url}/results"
|
||||
fetch_results_source(results_url, api_key)
|
||||
|
|
@ -226,7 +354,10 @@ defmodule Microwaveprop.Weather.Era5Client do
|
|||
end
|
||||
|
||||
defp fetch_results_source(url, api_key) do
|
||||
case Req.get(url, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000) do
|
||||
case Req.get(
|
||||
url,
|
||||
[headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000] ++ req_options()
|
||||
) do
|
||||
{:ok, %{status: 200, body: %{"asset" => %{"value" => %{"href" => href}}}}} ->
|
||||
# New CDS returns a JSON with download link
|
||||
{:ok, {:url, href}}
|
||||
|
|
@ -253,7 +384,10 @@ defmodule Microwaveprop.Weather.Era5Client do
|
|||
end
|
||||
|
||||
defp download_file(href, api_key) do
|
||||
case Req.get(href, headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000) do
|
||||
case Req.get(
|
||||
href,
|
||||
[headers: [{"PRIVATE-TOKEN", api_key}], receive_timeout: 120_000] ++ req_options()
|
||||
) do
|
||||
{:ok, %{status: 200, body: body}} -> {:ok, body}
|
||||
{:ok, %{status: status}} -> {:error, "ERA5 file download HTTP #{status}"}
|
||||
{:error, reason} -> {:error, "ERA5 file download failed: #{inspect(reason)}"}
|
||||
|
|
@ -266,10 +400,13 @@ defmodule Microwaveprop.Weather.Era5Client do
|
|||
# removed on HTTP error so a failed download never leaves a partial tmp file.
|
||||
defp download_href_to_file(href, api_key, path) do
|
||||
result =
|
||||
Req.get(href,
|
||||
headers: [{"PRIVATE-TOKEN", api_key}],
|
||||
receive_timeout: 300_000,
|
||||
into: File.stream!(path)
|
||||
Req.get(
|
||||
href,
|
||||
[
|
||||
headers: [{"PRIVATE-TOKEN", api_key}],
|
||||
receive_timeout: 300_000,
|
||||
into: File.stream!(path)
|
||||
] ++ req_options()
|
||||
)
|
||||
|
||||
case result do
|
||||
|
|
@ -286,6 +423,10 @@ defmodule Microwaveprop.Weather.Era5Client do
|
|||
end
|
||||
end
|
||||
|
||||
defp req_options do
|
||||
Application.get_env(:microwaveprop, :era5_req_options, [])
|
||||
end
|
||||
|
||||
defp build_profile(lat, lon, valid_time, single_grib, pressure_grib) do
|
||||
# Decode GRIB2 data using existing infrastructure
|
||||
with {:ok, single_points} <- extract_point(single_grib, lat, lon),
|
||||
|
|
|
|||
|
|
@ -1,29 +1,15 @@
|
|||
defmodule Microwaveprop.Workers.Era5MonthBatchWorker do
|
||||
@moduledoc """
|
||||
Fetches one month of ERA5 data for a 2° × 2° tile in a single CDS request
|
||||
and bulk-inserts all resulting hourly profiles into `era5_profiles`.
|
||||
Legacy entry point for the ERA5 month-tile backfill. **Forwards** to
|
||||
`Era5SubmitWorker` in the new split-worker pipeline — kept in place so
|
||||
already-enqueued callers (`Era5FetchWorker`, `mix era5.backfill`, any
|
||||
in-flight Oban rows) don't need to be changed on the same deploy.
|
||||
|
||||
This exists because fetching ERA5 per point-hour is tragically slow: each
|
||||
CDS request goes through a full submit → poll → assemble → download cycle
|
||||
whose overhead dwarfs the data transfer. Grouping by month+tile turns
|
||||
thousands of CDS jobs into a handful, and once a tile-month is cached in
|
||||
the DB every downstream query for a point inside that window is a local
|
||||
lookup.
|
||||
|
||||
Uniqueness on the full arg set means Oban collapses duplicate enqueues
|
||||
for the same tile-month — multiple `Era5FetchWorker` requests for the
|
||||
same region collapse into a single batch.
|
||||
|
||||
Runs on its own `:era5_batch` queue so its slow CDS poll cycles don't
|
||||
starve the fast `Era5FetchWorker` router that also lives in the ERA5
|
||||
namespace.
|
||||
On the next deploy cycle, callers will be updated to enqueue
|
||||
`Era5SubmitWorker` directly and this module can be removed.
|
||||
"""
|
||||
# The :era5_batch queue is rate-limited (see config/runtime.exs) to stay
|
||||
# under CDS-Beta's per-user concurrent ceiling. Retrying 5×, combined with
|
||||
# the generous backoff below, gives CDS up to a day to come back without
|
||||
# discarding the tile and silently dropping the historical contact.
|
||||
use Oban.Worker,
|
||||
queue: :era5_batch,
|
||||
queue: :era5_submit,
|
||||
max_attempts: 5,
|
||||
unique: [
|
||||
period: :infinity,
|
||||
|
|
@ -31,42 +17,26 @@ defmodule Microwaveprop.Workers.Era5MonthBatchWorker do
|
|||
keys: [:year, :month, :tile_lat, :tile_lon]
|
||||
]
|
||||
|
||||
alias Microwaveprop.Weather.Era5BatchClient
|
||||
alias Microwaveprop.Workers.Era5SubmitWorker
|
||||
|
||||
require Logger
|
||||
|
||||
@impl Oban.Worker
|
||||
def backoff(%Oban.Job{attempt: attempt}) do
|
||||
# A full month fetch can sit in the CDS queue for ages; back off generously.
|
||||
min(600 * Integer.pow(2, attempt - 1), _one_day = 86_400)
|
||||
# Forwarding is cheap; retry fast on transient failure.
|
||||
min(30 * Integer.pow(2, attempt - 1), 3_600)
|
||||
end
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: args}) do
|
||||
params = %{
|
||||
year: fetch_int!(args, "year"),
|
||||
month: fetch_int!(args, "month"),
|
||||
tile_lat: fetch_int!(args, "tile_lat"),
|
||||
tile_lon: fetch_int!(args, "tile_lon")
|
||||
}
|
||||
Logger.debug("Era5MonthBatchWorker forwarding to Era5SubmitWorker: #{inspect(args)}")
|
||||
|
||||
case Era5BatchClient.fetch_month_into_db(params) do
|
||||
{:ok, _count} ->
|
||||
:ok
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.warning(
|
||||
"ERA5Batch: #{params.year}-#{params.month} tile #{params.tile_lat},#{params.tile_lon} failed: #{inspect(reason)}"
|
||||
)
|
||||
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
|
||||
defp fetch_int!(args, key) do
|
||||
case Map.fetch!(args, key) do
|
||||
n when is_integer(n) -> n
|
||||
n when is_binary(n) -> String.to_integer(n)
|
||||
args
|
||||
|> Era5SubmitWorker.new()
|
||||
|> Oban.insert()
|
||||
|> case do
|
||||
{:ok, _job} -> :ok
|
||||
{:error, reason} -> {:error, reason}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
152
lib/microwaveprop/workers/era5_poll_worker.ex
Normal file
152
lib/microwaveprop/workers/era5_poll_worker.ex
Normal file
|
|
@ -0,0 +1,152 @@
|
|||
defmodule Microwaveprop.Workers.Era5PollWorker do
|
||||
@moduledoc """
|
||||
Second half of the split ERA5 backfill pipeline. Polls CDS for the two
|
||||
jobs submitted by `Era5SubmitWorker`; on completion, streams both GRIB2
|
||||
files to disk, decodes them, bulk-inserts the profiles, deletes the
|
||||
`era5_cds_jobs` row, and asks CDS to delete the completed jobs so they
|
||||
don't count against our per-user quota.
|
||||
|
||||
While either CDS job is still running the worker returns
|
||||
`{:snooze, N}` — Oban re-schedules the same job without counting an
|
||||
attempt and releases the worker slot immediately, so one pod can keep
|
||||
dozens of tile-months in flight without pinning workers to
|
||||
`Process.sleep`.
|
||||
|
||||
`max_attempts` is high because each attempt models a human-scale
|
||||
failure (e.g. CDS reports the job failed), not a poll retry — snoozes
|
||||
don't consume attempts.
|
||||
"""
|
||||
use Oban.Worker,
|
||||
queue: :era5_poll,
|
||||
max_attempts: 10
|
||||
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Weather.Era5BatchClient
|
||||
alias Microwaveprop.Weather.Era5CdsJob
|
||||
alias Microwaveprop.Weather.Era5Client
|
||||
|
||||
require Logger
|
||||
|
||||
# Re-check every 5 minutes. CDS month-tile jobs typically take 30-45 min,
|
||||
# so 5 minutes strikes a balance between responsiveness and noise.
|
||||
@snooze_seconds 5 * 60
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: %{"era5_cds_job_id" => id}}) do
|
||||
case Repo.get(Era5CdsJob, id) do
|
||||
nil ->
|
||||
Logger.debug("Era5Poll: row #{id} no longer exists — no-op")
|
||||
:ok
|
||||
|
||||
%Era5CdsJob{} = row ->
|
||||
handle_row(row)
|
||||
end
|
||||
end
|
||||
|
||||
defp handle_row(row) do
|
||||
single = Era5Client.check_status(row.single_job_id)
|
||||
pressure = Era5Client.check_status(row.pressure_job_id)
|
||||
|
||||
case {single, pressure} do
|
||||
{{:done, single_source}, {:done, pressure_source}} ->
|
||||
handle_both_done(row, single_source, pressure_source)
|
||||
|
||||
{{:failed, reason}, _} ->
|
||||
fail_row(row, "single-level job failed: #{reason}")
|
||||
|
||||
{_, {:failed, reason}} ->
|
||||
fail_row(row, "pressure-level job failed: #{reason}")
|
||||
|
||||
{{:error, reason}, _} ->
|
||||
{:error, "single-level check_status error: #{reason}"}
|
||||
|
||||
{_, {:error, reason}} ->
|
||||
{:error, "pressure-level check_status error: #{reason}"}
|
||||
|
||||
_both_running_or_mixed ->
|
||||
bump_poll_count(row)
|
||||
{:snooze, @snooze_seconds}
|
||||
end
|
||||
end
|
||||
|
||||
defp handle_both_done(row, single_source, pressure_source) do
|
||||
tmp_dir = System.tmp_dir!()
|
||||
|
||||
tag =
|
||||
"#{row.year}_#{pad(row.month)}_#{row.tile_lat}_#{row.tile_lon}_#{System.unique_integer([:positive])}"
|
||||
|
||||
single_path = Path.join(tmp_dir, "era5_single_#{tag}.grib2")
|
||||
pressure_path = Path.join(tmp_dir, "era5_pressure_#{tag}.grib2")
|
||||
|
||||
try do
|
||||
with :ok <-
|
||||
Era5Client.download_source_to_file(single_source, api_key(), single_path),
|
||||
:ok <-
|
||||
Era5Client.download_source_to_file(pressure_source, api_key(), pressure_path),
|
||||
{:ok, inserted} <-
|
||||
Era5BatchClient.decode_and_insert(
|
||||
row.year,
|
||||
row.month,
|
||||
row.tile_lat,
|
||||
row.tile_lon,
|
||||
single_path,
|
||||
pressure_path
|
||||
) do
|
||||
Repo.delete!(row)
|
||||
delete_cds_jobs(row)
|
||||
|
||||
Logger.info(
|
||||
"Era5Poll: stored #{inserted} profiles for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon}"
|
||||
)
|
||||
|
||||
{:ok, inserted}
|
||||
else
|
||||
{:error, reason} ->
|
||||
Logger.warning(
|
||||
"Era5Poll: download/decode failed for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon}: #{inspect(reason)}"
|
||||
)
|
||||
|
||||
{:error, reason}
|
||||
end
|
||||
after
|
||||
File.rm(single_path)
|
||||
File.rm(pressure_path)
|
||||
end
|
||||
end
|
||||
|
||||
defp fail_row(row, reason) do
|
||||
Logger.warning(
|
||||
"Era5Poll: CDS failure for #{row.year}-#{pad(row.month)} tile #{row.tile_lat},#{row.tile_lon}: #{reason}"
|
||||
)
|
||||
|
||||
delete_cds_jobs(row)
|
||||
Repo.delete!(row)
|
||||
{:error, reason}
|
||||
end
|
||||
|
||||
defp delete_cds_jobs(row) do
|
||||
# Best-effort — don't block progress on a CDS cleanup failure.
|
||||
for job_id <- [row.single_job_id, row.pressure_job_id] do
|
||||
case Era5Client.delete_job(job_id) do
|
||||
:ok ->
|
||||
:ok
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.warning("Era5Poll: failed to delete CDS job #{job_id}: #{reason}")
|
||||
:ok
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp bump_poll_count(row) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
row
|
||||
|> Era5CdsJob.changeset(%{poll_count: row.poll_count + 1, last_polled_at: now})
|
||||
|> Repo.update!()
|
||||
end
|
||||
|
||||
defp api_key, do: System.get_env("ERA5_CDS_API_KEY")
|
||||
|
||||
defp pad(n), do: String.pad_leading(Integer.to_string(n), 2, "0")
|
||||
end
|
||||
207
lib/microwaveprop/workers/era5_submit_worker.ex
Normal file
207
lib/microwaveprop/workers/era5_submit_worker.ex
Normal file
|
|
@ -0,0 +1,207 @@
|
|||
defmodule Microwaveprop.Workers.Era5SubmitWorker do
|
||||
@moduledoc """
|
||||
First half of the split ERA5 backfill pipeline.
|
||||
|
||||
Submits the two CDS jobs for a month-tile (single-level + pressure-level)
|
||||
in parallel, persists the returned CDS job IDs to `era5_cds_jobs`, and
|
||||
enqueues an `Era5PollWorker` job scheduled a few minutes out to check for
|
||||
completion.
|
||||
|
||||
The fast submit → enqueue-poll split means an Oban slot is held for ~1
|
||||
second (the time to POST and write a row), not the 30-45 minutes CDS
|
||||
takes to queue, assemble, and return a month-tile. A pod restart between
|
||||
submit and poll does not lose the CDS work: the row in `era5_cds_jobs`
|
||||
persists the job IDs and the poll worker picks up where we left off.
|
||||
|
||||
Idempotent: if the month-tile already has profiles (`era5_profiles`) or
|
||||
an in-flight `era5_cds_jobs` row, no submit is made.
|
||||
"""
|
||||
use Oban.Worker,
|
||||
queue: :era5_submit,
|
||||
max_attempts: 5,
|
||||
unique: [
|
||||
period: :infinity,
|
||||
states: [:available, :scheduled, :executing, :retryable],
|
||||
keys: [:year, :month, :tile_lat, :tile_lon]
|
||||
]
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Weather.Era5BatchClient
|
||||
alias Microwaveprop.Weather.Era5CdsJob
|
||||
alias Microwaveprop.Weather.Era5Client
|
||||
alias Microwaveprop.Weather.Era5Profile
|
||||
alias Microwaveprop.Workers.Era5PollWorker
|
||||
|
||||
require Logger
|
||||
|
||||
@poll_delay_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
|
||||
# don't stall the tile for an hour waiting on exponential backoff.
|
||||
min(30 * Integer.pow(2, attempt - 1), _one_hour = 3_600)
|
||||
end
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: args}) do
|
||||
year = fetch_int!(args, "year")
|
||||
month = fetch_int!(args, "month")
|
||||
tile_lat = fetch_int!(args, "tile_lat")
|
||||
tile_lon = fetch_int!(args, "tile_lon")
|
||||
|
||||
cond do
|
||||
month_tile_cached?(year, month, tile_lat, tile_lon) ->
|
||||
Logger.info("Era5Submit: #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} already cached — skip")
|
||||
|
||||
{:ok, :cached}
|
||||
|
||||
existing = find_in_flight(year, month, tile_lat, tile_lon) ->
|
||||
Logger.info(
|
||||
"Era5Submit: #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} already submitted (cds_job_id=#{existing.id}) — skip"
|
||||
)
|
||||
|
||||
{:ok, :already_submitted}
|
||||
|
||||
true ->
|
||||
submit_and_persist(year, month, tile_lat, tile_lon)
|
||||
end
|
||||
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)
|
||||
|
||||
# Run both submits in parallel — they're independent HTTP POSTs.
|
||||
single_task =
|
||||
Task.async(fn ->
|
||||
Era5Client.submit_job("reanalysis-era5-single-levels", single_request(year, month, days, area))
|
||||
end)
|
||||
|
||||
pressure_task =
|
||||
Task.async(fn ->
|
||||
Era5Client.submit_job("reanalysis-era5-pressure-levels", pressure_request(year, month, days, area))
|
||||
end)
|
||||
|
||||
try do
|
||||
with {:ok, single_id} <- Task.await(single_task, :infinity),
|
||||
{:ok, pressure_id} <- Task.await(pressure_task, :infinity),
|
||||
{:ok, row} <-
|
||||
insert_cds_job_row(year, month, tile_lat, tile_lon, single_id, pressure_id) do
|
||||
enqueue_poll(row)
|
||||
|
||||
Logger.info(
|
||||
"Era5Submit: submitted #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} (single=#{single_id}, pressure=#{pressure_id})"
|
||||
)
|
||||
|
||||
{:ok, row}
|
||||
end
|
||||
after
|
||||
Task.shutdown(single_task, :brutal_kill)
|
||||
Task.shutdown(pressure_task, :brutal_kill)
|
||||
end
|
||||
end
|
||||
|
||||
defp insert_cds_job_row(year, month, tile_lat, tile_lon, single_id, pressure_id) do
|
||||
%Era5CdsJob{}
|
||||
|> Era5CdsJob.changeset(%{
|
||||
year: year,
|
||||
month: month,
|
||||
tile_lat: tile_lat,
|
||||
tile_lon: tile_lon,
|
||||
single_job_id: single_id,
|
||||
pressure_job_id: pressure_id,
|
||||
submitted_at: DateTime.truncate(DateTime.utc_now(), :second)
|
||||
})
|
||||
|> Repo.insert()
|
||||
|> case do
|
||||
{:ok, row} -> {:ok, row}
|
||||
{:error, changeset} -> {:error, "era5_cds_jobs insert failed: #{inspect(changeset.errors)}"}
|
||||
end
|
||||
end
|
||||
|
||||
defp enqueue_poll(%Era5CdsJob{id: id}) do
|
||||
%{"era5_cds_job_id" => id}
|
||||
|> Era5PollWorker.new(schedule_in: @poll_delay_seconds)
|
||||
|> Oban.insert()
|
||||
end
|
||||
|
||||
defp find_in_flight(year, month, tile_lat, tile_lon) do
|
||||
Repo.one(
|
||||
from j in Era5CdsJob,
|
||||
where:
|
||||
j.year == ^year and j.month == ^month and j.tile_lat == ^tile_lat and
|
||||
j.tile_lon == ^tile_lon,
|
||||
limit: 1
|
||||
)
|
||||
end
|
||||
|
||||
defp month_tile_cached?(year, month, tile_lat, tile_lon) do
|
||||
{first_dt, last_dt} = month_bounds(year, month)
|
||||
tile_degrees = Era5BatchClient.tile_degrees()
|
||||
|
||||
Era5Profile
|
||||
|> where(
|
||||
[p],
|
||||
p.valid_time >= ^first_dt and p.valid_time < ^last_dt and
|
||||
p.lat >= ^(tile_lat * 1.0) and p.lat <= ^(tile_lat * 1.0 + tile_degrees) and
|
||||
p.lon >= ^(tile_lon * 1.0) and p.lon <= ^(tile_lon * 1.0 + tile_degrees)
|
||||
)
|
||||
|> Repo.exists?()
|
||||
end
|
||||
|
||||
defp month_bounds(year, month) do
|
||||
{:ok, first} = Date.new(year, month, 1)
|
||||
{:ok, first_dt} = DateTime.new(first, ~T[00:00:00], "Etc/UTC")
|
||||
days = Calendar.ISO.days_in_month(year, month)
|
||||
last_dt = DateTime.add(first_dt, days * 86_400, :second)
|
||||
{first_dt, last_dt}
|
||||
end
|
||||
|
||||
defp tile_area(tile_lat, tile_lon) do
|
||||
tile_degrees = Era5BatchClient.tile_degrees()
|
||||
|
||||
Enum.map(
|
||||
[tile_lat + tile_degrees, tile_lon, tile_lat, tile_lon + tile_degrees],
|
||||
&(&1 * 1.0)
|
||||
)
|
||||
end
|
||||
|
||||
defp single_request(year, month, days, area) do
|
||||
%{
|
||||
"product_type" => ["reanalysis"],
|
||||
"variable" => Era5Client.single_level_vars(),
|
||||
"year" => [Integer.to_string(year)],
|
||||
"month" => [pad(month)],
|
||||
"day" => Enum.map(1..days, &pad/1),
|
||||
"time" => Enum.map(0..23, &"#{pad(&1)}:00"),
|
||||
"area" => area,
|
||||
"data_format" => "grib"
|
||||
}
|
||||
end
|
||||
|
||||
defp pressure_request(year, month, days, area) do
|
||||
%{
|
||||
"product_type" => ["reanalysis"],
|
||||
"variable" => Era5Client.pressure_level_vars(),
|
||||
"pressure_level" => Era5Client.pressure_levels(),
|
||||
"year" => [Integer.to_string(year)],
|
||||
"month" => [pad(month)],
|
||||
"day" => Enum.map(1..days, &pad/1),
|
||||
"time" => Enum.map(0..23, &"#{pad(&1)}:00"),
|
||||
"area" => area,
|
||||
"data_format" => "grib"
|
||||
}
|
||||
end
|
||||
|
||||
defp pad(n), do: String.pad_leading(Integer.to_string(n), 2, "0")
|
||||
|
||||
defp fetch_int!(args, key) do
|
||||
case Map.fetch!(args, key) do
|
||||
v when is_integer(v) -> v
|
||||
v when is_binary(v) -> String.to_integer(v)
|
||||
end
|
||||
end
|
||||
end
|
||||
28
priv/repo/migrations/20260413210544_create_era5_cds_jobs.exs
Normal file
28
priv/repo/migrations/20260413210544_create_era5_cds_jobs.exs
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
defmodule Microwaveprop.Repo.Migrations.CreateEra5CdsJobs do
|
||||
use Ecto.Migration
|
||||
|
||||
# Tracks in-flight CDS requests for the split Era5SubmitWorker / Era5PollWorker
|
||||
# design. A row is created when Era5SubmitWorker successfully submits both
|
||||
# CDS jobs, and is deleted when Era5PollWorker downloads the results and
|
||||
# inserts profiles (or when either leg permanently fails). Persisting the
|
||||
# CDS job IDs lets a pod restart resume polling instead of losing the work.
|
||||
def change do
|
||||
create table(:era5_cds_jobs, primary_key: false) do
|
||||
add :id, :binary_id, primary_key: true
|
||||
add :year, :integer, null: false
|
||||
add :month, :integer, null: false
|
||||
add :tile_lat, :integer, null: false
|
||||
add :tile_lon, :integer, null: false
|
||||
add :single_job_id, :string, null: false
|
||||
add :pressure_job_id, :string, null: false
|
||||
add :submitted_at, :utc_datetime, null: false
|
||||
add :last_polled_at, :utc_datetime
|
||||
add :poll_count, :integer, default: 0, null: false
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
# Uniqueness so duplicate submits for the same tile-month collapse.
|
||||
create unique_index(:era5_cds_jobs, [:year, :month, :tile_lat, :tile_lon])
|
||||
end
|
||||
end
|
||||
82
test/microwaveprop/weather/era5_cds_job_test.exs
Normal file
82
test/microwaveprop/weather/era5_cds_job_test.exs
Normal file
|
|
@ -0,0 +1,82 @@
|
|||
defmodule Microwaveprop.Weather.Era5CdsJobTest do
|
||||
use Microwaveprop.DataCase, async: true
|
||||
|
||||
alias Microwaveprop.Weather.Era5CdsJob
|
||||
|
||||
describe "changeset/2" do
|
||||
@valid_attrs %{
|
||||
year: 2014,
|
||||
month: 3,
|
||||
tile_lat: 32,
|
||||
tile_lon: -98,
|
||||
single_job_id: "abc-single",
|
||||
pressure_job_id: "abc-pressure",
|
||||
submitted_at: ~U[2026-04-13 21:00:00Z]
|
||||
}
|
||||
|
||||
test "accepts a complete attrs map" do
|
||||
changeset = Era5CdsJob.changeset(%Era5CdsJob{}, @valid_attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "requires year, month, tile coordinates, and both CDS job IDs" do
|
||||
changeset = Era5CdsJob.changeset(%Era5CdsJob{}, %{})
|
||||
|
||||
refute changeset.valid?
|
||||
|
||||
required = ~w(year month tile_lat tile_lon single_job_id pressure_job_id submitted_at)a
|
||||
errors = Keyword.keys(changeset.errors)
|
||||
|
||||
for field <- required do
|
||||
assert field in errors, "expected #{inspect(field)} to be required"
|
||||
end
|
||||
end
|
||||
|
||||
test "last_polled_at and poll_count default to nil/0 and can be updated" do
|
||||
{:ok, row} =
|
||||
%Era5CdsJob{}
|
||||
|> Era5CdsJob.changeset(@valid_attrs)
|
||||
|> Repo.insert()
|
||||
|
||||
assert row.poll_count == 0
|
||||
assert is_nil(row.last_polled_at)
|
||||
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
{:ok, bumped} =
|
||||
row
|
||||
|> Era5CdsJob.changeset(%{last_polled_at: now, poll_count: 1})
|
||||
|> Repo.update()
|
||||
|
||||
assert bumped.poll_count == 1
|
||||
assert bumped.last_polled_at == now
|
||||
end
|
||||
end
|
||||
|
||||
describe "uniqueness" do
|
||||
test "collapses duplicate (year, month, tile_lat, tile_lon) inserts" do
|
||||
attrs = %{
|
||||
year: 2014,
|
||||
month: 3,
|
||||
tile_lat: 32,
|
||||
tile_lon: -98,
|
||||
single_job_id: "first-single",
|
||||
pressure_job_id: "first-pressure",
|
||||
submitted_at: ~U[2026-04-13 21:00:00Z]
|
||||
}
|
||||
|
||||
{:ok, _} =
|
||||
%Era5CdsJob{}
|
||||
|> Era5CdsJob.changeset(attrs)
|
||||
|> Repo.insert()
|
||||
|
||||
{:error, changeset} =
|
||||
%Era5CdsJob{}
|
||||
|> Era5CdsJob.changeset(%{attrs | single_job_id: "second-single", pressure_job_id: "second-pressure"})
|
||||
|> Repo.insert()
|
||||
|
||||
refute changeset.valid?
|
||||
assert {:tile_month, _} = List.keyfind(changeset.errors, :tile_month, 0) || {:tile_month, nil}
|
||||
end
|
||||
end
|
||||
end
|
||||
176
test/microwaveprop/weather/era5_client_test.exs
Normal file
176
test/microwaveprop/weather/era5_client_test.exs
Normal file
|
|
@ -0,0 +1,176 @@
|
|||
defmodule Microwaveprop.Weather.Era5ClientTest do
|
||||
# async: false — tests mutate the global ERA5_CDS_API_KEY env var to match
|
||||
# production's lookup path. Running concurrently with Era5BatchClientTest
|
||||
# (which *deletes* the env var to exercise the missing-key error path)
|
||||
# leaks state in both directions.
|
||||
use ExUnit.Case, async: false
|
||||
|
||||
alias Microwaveprop.Weather.Era5Client
|
||||
|
||||
setup do
|
||||
# A real key — stubs don't care about value, only presence.
|
||||
System.put_env("ERA5_CDS_API_KEY", "test-key")
|
||||
on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
|
||||
:ok
|
||||
end
|
||||
|
||||
describe "submit_job/2" do
|
||||
test "returns {:ok, job_id} when CDS accepts the request" do
|
||||
Req.Test.stub(Era5Client, fn conn ->
|
||||
assert conn.method == "POST"
|
||||
assert String.ends_with?(conn.request_path, "/reanalysis-era5-single-levels/execution")
|
||||
Req.Test.json(conn, %{"jobID" => "cds-job-123", "status" => "accepted"})
|
||||
end)
|
||||
|
||||
assert {:ok, "cds-job-123"} =
|
||||
Era5Client.submit_job("reanalysis-era5-single-levels", %{"foo" => "bar"})
|
||||
end
|
||||
|
||||
test "returns {:error, _} when CDS returns 400" do
|
||||
Req.Test.stub(Era5Client, fn conn ->
|
||||
conn
|
||||
|> Plug.Conn.put_resp_content_type("application/json")
|
||||
|> Plug.Conn.resp(400, ~s({"detail":"bad request"}))
|
||||
end)
|
||||
|
||||
assert {:error, reason} = Era5Client.submit_job("reanalysis-era5-single-levels", %{})
|
||||
assert reason =~ "HTTP 400"
|
||||
end
|
||||
|
||||
test "returns {:error, _} when API key is missing" do
|
||||
System.delete_env("ERA5_CDS_API_KEY")
|
||||
assert {:error, reason} = Era5Client.submit_job("reanalysis-era5-single-levels", %{})
|
||||
assert reason =~ "ERA5_CDS_API_KEY"
|
||||
end
|
||||
end
|
||||
|
||||
describe "check_status/1" do
|
||||
test "returns :running while CDS job is accepted" do
|
||||
Req.Test.stub(Era5Client, fn conn ->
|
||||
assert conn.method == "GET"
|
||||
assert String.ends_with?(conn.request_path, "/jobs/cds-job-123")
|
||||
Req.Test.json(conn, %{"status" => "accepted"})
|
||||
end)
|
||||
|
||||
assert :running = Era5Client.check_status("cds-job-123")
|
||||
end
|
||||
|
||||
test "returns :running while CDS job is running" do
|
||||
Req.Test.stub(Era5Client, fn conn ->
|
||||
Req.Test.json(conn, %{"status" => "running"})
|
||||
end)
|
||||
|
||||
assert :running = Era5Client.check_status("cds-job-123")
|
||||
end
|
||||
|
||||
test "returns {:done, {:url, href}} when CDS job completes with an asset href" do
|
||||
Req.Test.stub(Era5Client, fn conn ->
|
||||
case conn.request_path do
|
||||
"/api/retrieve/v1/jobs/cds-job-123" ->
|
||||
Req.Test.json(conn, %{"status" => "successful"})
|
||||
|
||||
"/api/retrieve/v1/jobs/cds-job-123/results" ->
|
||||
Req.Test.json(conn, %{
|
||||
"asset" => %{"value" => %{"href" => "https://example.com/file.grib"}}
|
||||
})
|
||||
end
|
||||
end)
|
||||
|
||||
assert {:done, {:url, "https://example.com/file.grib"}} =
|
||||
Era5Client.check_status("cds-job-123")
|
||||
end
|
||||
|
||||
test "returns {:failed, _} when CDS reports the job failed" do
|
||||
Req.Test.stub(Era5Client, fn conn ->
|
||||
assert conn.request_path == "/api/retrieve/v1/jobs/cds-job-123"
|
||||
Req.Test.json(conn, %{"status" => "failed", "message" => "input out of range"})
|
||||
end)
|
||||
|
||||
assert {:failed, reason} = Era5Client.check_status("cds-job-123")
|
||||
assert reason =~ "input out of range"
|
||||
end
|
||||
end
|
||||
|
||||
describe "delete_job/1" do
|
||||
test "sends DELETE to the CDS jobs endpoint and returns :ok on 204" do
|
||||
test_pid = self()
|
||||
|
||||
Req.Test.stub(Era5Client, fn conn ->
|
||||
send(test_pid, {:delete_request, conn.method, conn.request_path})
|
||||
Plug.Conn.resp(conn, 204, "")
|
||||
end)
|
||||
|
||||
assert :ok = Era5Client.delete_job("cds-job-123")
|
||||
assert_received {:delete_request, "DELETE", "/api/retrieve/v1/jobs/cds-job-123"}
|
||||
end
|
||||
|
||||
test "also treats 200/202/404 as success (CDS sometimes returns 200 or the job may already be gone)" do
|
||||
for status <- [200, 202, 404] do
|
||||
Req.Test.stub(Era5Client, fn conn -> Plug.Conn.resp(conn, status, "") end)
|
||||
assert :ok = Era5Client.delete_job("cds-job-#{status}")
|
||||
end
|
||||
end
|
||||
|
||||
test "returns {:error, _} for unexpected statuses" do
|
||||
Req.Test.stub(Era5Client, fn conn -> Plug.Conn.resp(conn, 500, "server error") end)
|
||||
assert {:error, reason} = Era5Client.delete_job("cds-job-500")
|
||||
assert reason =~ "HTTP 500"
|
||||
end
|
||||
end
|
||||
|
||||
describe "download_source_to_file/3" do
|
||||
test "writes a {:body, binary} source directly to the target path" do
|
||||
path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
|
||||
|
||||
try do
|
||||
assert :ok = Era5Client.download_source_to_file({:body, "grib-payload"}, nil, path)
|
||||
assert File.read!(path) == "grib-payload"
|
||||
after
|
||||
File.rm(path)
|
||||
end
|
||||
end
|
||||
|
||||
test "streams a {:url, href} source through Req into the target path" do
|
||||
Req.Test.stub(Era5Client, fn conn ->
|
||||
assert conn.method == "GET"
|
||||
|
||||
conn
|
||||
|> Plug.Conn.put_resp_content_type("application/octet-stream")
|
||||
|> Plug.Conn.resp(200, "streamed-grib-data")
|
||||
end)
|
||||
|
||||
path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
|
||||
|
||||
try do
|
||||
assert :ok =
|
||||
Era5Client.download_source_to_file(
|
||||
{:url, "https://example.com/file.grib"},
|
||||
"test-key",
|
||||
path
|
||||
)
|
||||
|
||||
assert File.read!(path) == "streamed-grib-data"
|
||||
after
|
||||
File.rm(path)
|
||||
end
|
||||
end
|
||||
|
||||
test "removes partial file and returns {:error, _} on HTTP error" do
|
||||
Req.Test.stub(Era5Client, fn conn ->
|
||||
Plug.Conn.resp(conn, 503, "")
|
||||
end)
|
||||
|
||||
path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
|
||||
|
||||
assert {:error, reason} =
|
||||
Era5Client.download_source_to_file(
|
||||
{:url, "https://example.com/file.grib"},
|
||||
"test-key",
|
||||
path
|
||||
)
|
||||
|
||||
assert reason =~ "HTTP 503"
|
||||
refute File.exists?(path)
|
||||
end
|
||||
end
|
||||
end
|
||||
96
test/microwaveprop/workers/era5_poll_worker_test.exs
Normal file
96
test/microwaveprop/workers/era5_poll_worker_test.exs
Normal file
|
|
@ -0,0 +1,96 @@
|
|||
defmodule Microwaveprop.Workers.Era5PollWorkerTest do
|
||||
use Microwaveprop.DataCase, async: false
|
||||
|
||||
alias Microwaveprop.Weather.Era5CdsJob
|
||||
alias Microwaveprop.Weather.Era5Client
|
||||
alias Microwaveprop.Workers.Era5PollWorker
|
||||
|
||||
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 — 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
|
||||
136
test/microwaveprop/workers/era5_submit_worker_test.exs
Normal file
136
test/microwaveprop/workers/era5_submit_worker_test.exs
Normal file
|
|
@ -0,0 +1,136 @@
|
|||
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
|
||||
Loading…
Add table
Reference in a new issue