graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Drop dead Era5*/CDS code paths

Browse source 
After the NARR backfill landed, nothing reachable from the app calls
any of these modules. Net removal: ~2700 lines.

Deleted:
- Microwaveprop.Workers.{Era5Fetch,Era5Submit,Era5Poll,Era5MonthBatch}Worker
- Microwaveprop.Weather.{Era5BatchClient,Era5Client}
- Mix.Tasks.Era5Backfill (superseded by BackfillEnqueueWorker cron)
- Matching test files (~1100 LOC of test)
- era5/era5_submit/era5_poll/era5_batch queue blocks from runtime.exs
- era5_req_options Req.Test stub config from test.exs

Kept (actively used or intentionally preserved):
- Era5Profile schema (the era5_profiles table is the NARR target)
- Era5CdsJob schema + its test (inspection handle on the era5_cds_jobs
  table, which retains rows from the failed CDS runs)
- :era5 type key in ContactWeatherEnqueueWorker / BackfillEnqueueWorker
  (the symbol still means "historical backfill", just routed to NARR now)

Swapped the one remaining live Era5FetchWorker call site in
contact_live/show.ex's maybe_enqueue_era5 path to NarrFetchWorker
(with NarrClient.snap_to_analysis_hour for the 3-hourly slot).

Test suite: 1513 tests, 0 failures (-50 from the deleted era5 test
files). No compile warnings. Credo clean.
This commit is contained in:
Graham McIntire 2026-04-16 08:26:17 -05:00
parent d70444dd06
commit 1f2a729d40
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
17 changed files with 18 additions and 2693 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
config/runtime.exs
View file

@ -168,24 +168,13 @@ if config_env() == :prod do
hrrr: 1,
terrain: 3,
iemre: 3,
# Historical backfill for pre-2014 contacts (pre-HRRR archive).
# NARR is fetched anonymously from NCEI with no quota, no job
# queue. Replaced the ERA5/CDS pipeline which never produced a
# single row in prod. era5_cds_jobs table is kept for inspection
# but its workers/client are gone. See
# docs/plans/2026-04-15-merra2-historical-backfill.md.
narr: 6,
# ERA5 queues are DEPRECATED in favor of NARR via NarrFetchWorker
# (see docs/plans/2026-04-15-merra2-historical-backfill.md). The
# CDS pipeline never produced a single era5_profiles row in prod
# — every poll hit either CDS-rejected or long-running stuck
# states. NARR fetches anonymously from NCEI with no quota, no
# job queue, and the same era5_profiles table on the receiving
# side. These queues stay paused while the dead Era5* workers
# and the era5_cds_jobs table are slated for deletion in a
# follow-up PR.
era5: [local_limit: 2, paused: true],
era5_submit: [
local_limit: 4,
rate_limit: [allowed: 30, period: {1, :hour}],
paused: true
],
era5_poll: [local_limit: 20, paused: true],
era5_batch: [local_limit: 1, paused: true],
rtma: 2,
backfill_enqueue: 1,
admin: 1,

1
config/test.exs
View file

@ -58,7 +58,6 @@ config :microwaveprop,
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, giro_req_options: [plug: {Req.Test, Microwaveprop.Ionosphere.GiroClient}, 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]

376
lib/microwaveprop/weather/era5_batch_client.ex
View file

@ -1,376 +0,0 @@
defmodule Microwaveprop.Weather.Era5BatchClient do
@moduledoc """
Fetches ERA5 reanalysis data from Copernicus Climate Data Store (CDS) one
month-tile at a time and bulk-inserts the resulting profiles into
`era5_profiles`.
The per-point `Era5Client.fetch_profile/3` path is tragically slow because
every point-hour triggers its own asynchronous CDS job (submit → poll →
assemble → download). For backfill this balloons into thousands of
independent CDS jobs. This module instead groups requests by month and a
small lat/lon tile (2° × 2° by default) so one submit/poll/download cycle
produces many thousands of profile rows.
Idempotency: `fetch_month_into_db/1` is a no-op when the requested
(year, month, tile) already has profile rows in `era5_profiles`. Combined
with Oban uniqueness on `Era5MonthBatchWorker`, redundant fetches are
collapsed.
"""
import Ecto.Query
alias Microwaveprop.Repo
alias Microwaveprop.Weather.Era5Client
alias Microwaveprop.Weather.Era5Profile
alias Microwaveprop.Weather.Grib2.Wgrib2
alias Microwaveprop.Weather.SoundingParams
alias Microwaveprop.Weather.ThetaE
require Logger
# ERA5 hourly reanalysis runs on a 0.25° global grid. Tiles are aligned
# to an even-degree lattice (2° × 2°), producing 9 × 9 = 81 grid points
# per tile per hour. A month-tile expands to ~60k profiles.
@tile_degrees 2
@era5_step 0.25
@doc """
Returns the integer tile coordinate `{tile_lat, tile_lon}` for a point,
where `tile_lat` is the south edge and `tile_lon` is the west edge.
"""
@spec tile_for_point(float(), float()) :: {integer(), integer()}
def tile_for_point(lat, lon) do
{floor_to_tile(lat), floor_to_tile(lon)}
end
defp floor_to_tile(value) do
value
|> Kernel./(@tile_degrees)
|> Float.floor()
|> trunc()
|> Kernel.*(@tile_degrees)
end
@doc false
@spec tile_degrees() :: pos_integer()
def tile_degrees, do: @tile_degrees
@doc """
Fetches one month of ERA5 data covering the given tile and stores every
resulting hourly profile in `era5_profiles`. Returns
`{:ok, inserted_count}` or `{:error, reason}`.
"""
@spec fetch_month_into_db(%{
year: pos_integer(),
month: pos_integer(),
tile_lat: integer(),
tile_lon: integer()
}) :: {:ok, non_neg_integer()} | {:error, term()}
def fetch_month_into_db(%{year: year, month: month, tile_lat: tile_lat, tile_lon: tile_lon}) do
if month_tile_cached?(year, month, tile_lat, tile_lon) do
Logger.info("ERA5Batch: #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} already cached")
{:ok, 0}
else
do_fetch_month(year, month, tile_lat, tile_lon)
end
end
defp do_fetch_month(year, month, tile_lat, tile_lon) do
area = tile_area(tile_lat, tile_lon)
days = days_in_month(year, month)
Logger.info("ERA5Batch: fetching #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon} (#{days} days)")
tmp_dir = System.tmp_dir!()
job_tag = "#{year}_#{pad(month)}_#{tile_lat}_#{tile_lon}_#{System.unique_integer([:positive])}"
single_path = Path.join(tmp_dir, "era5_single_#{job_tag}.grib2")
pressure_path = Path.join(tmp_dir, "era5_pressure_#{job_tag}.grib2")
# Fire both CDS submits in parallel — they have no ordering dependency and
# each spends 30+ min blocking on CDS queue/download. Running them serially
# doubles wall time for zero benefit. Task.async links to the Oban worker,
# so if the worker exits (deploy, crash) both tasks die with it. We still
# shut down any unfinished task on early error so we don't waste CDS quota.
# Both downloads stream directly to disk (Req `into: File.stream!`), so
# peak heap stays O(chunk) rather than O(month-tile GRIB size ~50-200 MB).
single_task =
Task.async(fn -> fetch_single_level_month(year, month, days, area, single_path) end)
pressure_task =
Task.async(fn -> fetch_pressure_level_month(year, month, days, area, pressure_path) end)
try do
with :ok <- Task.await(single_task, :infinity),
: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)
Task.shutdown(pressure_task, :brutal_kill)
File.rm(single_path)
File.rm(pressure_path)
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)
Era5Profile
|> where(
[p],
p.valid_time >= ^first and p.valid_time < ^last 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")
last_dt = DateTime.add(first_dt, days_in_month(year, month) * 86_400, :second)
{first_dt, last_dt}
end
defp days_in_month(year, month), do: Calendar.ISO.days_in_month(year, month)
defp tile_area(tile_lat, tile_lon) do
# CDS area is [north, west, south, east]
Enum.map([tile_lat + @tile_degrees, tile_lon, tile_lat, tile_lon + @tile_degrees], &(&1 * 1.0))
end
defp pad(n), do: String.pad_leading(Integer.to_string(n), 2, "0")
# -- CDS requests ----------------------------------------------------------
defp fetch_single_level_month(year, month, days, area, dest_path) do
request = %{
"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"
}
Era5Client.submit_and_download_to_file("reanalysis-era5-single-levels", request, dest_path)
end
defp fetch_pressure_level_month(year, month, days, area, dest_path) do
request = %{
"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"
}
Era5Client.submit_and_download_to_file("reanalysis-era5-pressure-levels", request, dest_path)
end
# -- GRIB extraction -------------------------------------------------------
defp grid_spec_for_tile(tile_lat, tile_lon) do
count = round(@tile_degrees / @era5_step) + 1
%{
lon_start: tile_lon * 1.0,
lon_count: count,
lon_step: @era5_step,
lat_start: tile_lat * 1.0,
lat_count: count,
lat_step: @era5_step
}
end
defp build_month_profiles(tile_lat, tile_lon, single_path, pressure_path) do
grid_spec = grid_spec_for_tile(tile_lat, tile_lon)
with {:ok, single_msgs} <-
Wgrib2.extract_grid_messages_from_file(
single_path,
":(TMP|DPT|PRES|UGRD|VGRD|PWAT|HPBL):",
grid_spec
),
{:ok, pressure_msgs} <-
Wgrib2.extract_grid_messages_from_file(
pressure_path,
":(TMP|SPFH|HGT):",
grid_spec
) do
# Index both message sets by (datetime, lat, lon) → var:level map.
points_single = index_messages(single_msgs)
points_pressure = index_messages(pressure_msgs)
all_keys =
MapSet.union(
MapSet.new(Map.keys(points_single)),
MapSet.new(Map.keys(points_pressure))
)
profiles =
all_keys
|> Enum.map(fn {dt, lat, lon} ->
single_fields = Map.get(points_single, {dt, lat, lon}, %{})
pressure_fields = Map.get(points_pressure, {dt, lat, lon}, %{})
build_profile_attrs(lat, lon, dt, single_fields, pressure_fields)
end)
|> Enum.reject(&is_nil/1)
{:ok, profiles}
end
end
defp index_messages(messages) do
Enum.reduce(messages, %{}, fn msg, acc ->
key_name = "#{msg.var}:#{msg.level}"
Enum.reduce(msg.values, acc, fn {{lat, lon}, value}, inner ->
Map.update(
inner,
{msg.datetime, lat, lon},
%{key_name => value},
&Map.put(&1, key_name, value)
)
end)
end)
end
defp build_profile_attrs(_lat, _lon, nil, _single, _pressure), do: nil
defp build_profile_attrs(lat, lon, datetime, single_fields, pressure_fields) do
temp_c = kelvin_to_celsius(single_fields["TMP:2 m above ground"])
dewpoint_c = kelvin_to_celsius(single_fields["DPT:2 m above ground"])
pressure_mb = pa_to_mb(single_fields["PRES:surface"])
hpbl_m = single_fields["HPBL:surface"]
pwat_mm = single_fields["PWAT:entire atmosphere"]
profile =
Era5Client.pressure_levels()
|> Enum.map(fn level_str ->
level = String.to_integer(level_str)
pres_key = "#{level} mb"
%{
"pres" => level * 1.0,
"tmpc" => kelvin_to_celsius(pressure_fields["TMP:#{pres_key}"]),
"dwpc" =>
dewpoint_from_spfh(
pressure_fields["SPFH:#{pres_key}"],
level * 100.0
),
"hght" => geopotential_to_height(pressure_fields["HGT:#{pres_key}"])
}
end)
|> Enum.reject(&is_nil(&1["tmpc"]))
if profile == [] and is_nil(temp_c) do
nil
else
params = SoundingParams.derive(profile)
base = %{
valid_time: datetime,
lat: lat,
lon: lon,
profile: profile,
hpbl_m: hpbl_m,
pwat_mm: pwat_mm,
surface_temp_c: temp_c,
surface_dewpoint_c: dewpoint_c,
surface_pressure_mb: pressure_mb
}
if params do
Map.merge(base, %{
surface_refractivity: params.surface_refractivity,
min_refractivity_gradient: params.min_refractivity_gradient,
ducting_detected: params.ducting_detected || false,
duct_characteristics: params.duct_characteristics
})
else
base
end
end
end
defp kelvin_to_celsius(nil), do: nil
defp kelvin_to_celsius(k), do: k - 273.15
defp pa_to_mb(nil), do: nil
defp pa_to_mb(pa), do: pa / 100.0
defp geopotential_to_height(nil), do: nil
defp geopotential_to_height(gp), do: gp / 9.80665
# ERA5 pressure-level moisture is specific humidity (kg/kg). Convert to
# dewpoint in °C via the Magnus-Tetens inverse already used for HRRR native
# profiles. `level_pa` is the pressure level itself in Pa.
# ThetaE.dewpoint_from_spfh/2 returns Kelvin. Profile `dwpc` entries
# are Celsius (the downstream SoundingParams.compute_refractivity_profile
# passes p["dwpc"] straight to the Buck saturation-vapor-pressure
# equation which expects °C). Convert here.
defp dewpoint_from_spfh(nil, _level_pa), do: nil
defp dewpoint_from_spfh(spfh, _level_pa) when spfh <= 0, do: nil
defp dewpoint_from_spfh(spfh, level_pa), do: ThetaE.dewpoint_from_spfh(spfh, level_pa) - 273.15
# -- Bulk insert -----------------------------------------------------------
# Insert in chunks to keep each query under Postgres' parameter limit
# (65,535 params / ~15 fields ≈ 4,000 rows per statement).
@chunk_size 4_000
defp bulk_insert_profiles(profiles) do
now = DateTime.truncate(DateTime.utc_now(), :second)
profiles
|> Enum.chunk_every(@chunk_size)
|> Enum.reduce(0, fn chunk, acc ->
rows =
Enum.map(chunk, fn attrs ->
attrs
|> Map.put(:inserted_at, now)
|> Map.put(:updated_at, now)
end)
{count, _} =
Repo.insert_all(Era5Profile, rows,
on_conflict: :nothing,
conflict_target: [:lat, :lon, :valid_time]
)
acc + count
end)
end
end

541
lib/microwaveprop/weather/era5_client.ex
View file

@ -1,541 +0,0 @@
defmodule Microwaveprop.Weather.Era5Client do
@moduledoc """
Client for the Copernicus Climate Data Store (CDS) API.
Fetches ERA5 reanalysis data (pressure-level profiles + single-level surface fields)
for historical contact enrichment where HRRR is unavailable (pre-2014).
The CDS API is asynchronous: submit a request, poll for completion, download result.
Results are GRIB2 format, decoded with the existing GRIB2 infrastructure.
"""
alias Microwaveprop.Weather.Grib2.Extractor
alias Microwaveprop.Weather.SoundingParams
require Logger
@cds_url "https://cds.climate.copernicus.eu/api"
@poll_interval_ms 5_000
# CDS routinely takes 30+ min to return month-tile jobs under load. The prior
# 120-attempt (10 min) ceiling was burning through Era5MonthBatchWorker
# max_attempts before CDS even produced a result. 720 attempts × 5s = 1 hour
# bounds the worker but still gives CDS enough room.
@max_poll_attempts 720
@pressure_levels ~w(
1000 975 950 925 900 875 850 825 800 775 750 725 700
650 600 550 500 450 400 350 300 250 200 150 100
)
@single_level_vars ~w(
2m_temperature
2m_dewpoint_temperature
surface_pressure
10m_u_component_of_wind
10m_v_component_of_wind
total_column_water_vapour
boundary_layer_height
)
# `dewpoint_temperature` is only a single-level 2m variable in ERA5 — CDS
# rejects pressure-level requests that include it. Use specific_humidity on
# pressure levels and derive Td in the profile builder via ThetaE.
@pressure_level_vars ~w(temperature specific_humidity geopotential)
@doc false
@spec pressure_levels() :: [String.t()]
def pressure_levels, do: @pressure_levels
@doc false
@spec single_level_vars() :: [String.t()]
def single_level_vars, do: @single_level_vars
@doc false
@spec pressure_level_vars() :: [String.t()]
def pressure_level_vars, do: @pressure_level_vars
@doc false
@spec cds_url() :: String.t()
def cds_url, do: @cds_url
@doc false
@spec poll_interval_ms() :: pos_integer()
def poll_interval_ms, do: @poll_interval_ms
@doc false
@spec max_poll_attempts() :: pos_integer()
def max_poll_attempts, do: @max_poll_attempts
@doc """
Fetch ERA5 profile for a single point and time.
Returns {:ok, profile_attrs} or {:error, reason}.
The profile_attrs map has the same shape as HRRR profiles for interoperability.
"""
@spec fetch_profile(float(), float(), DateTime.t()) :: {:ok, map()} | {:error, term()}
def fetch_profile(lat, lon, timestamp) do
# Round to ERA5 grid (0.25°)
rlat = Float.round(lat * 4) / 4
rlon = Float.round(lon * 4) / 4
# ERA5 hourly — round to nearest hour
valid_time = DateTime.truncate(timestamp, :second)
valid_time = %{valid_time | minute: 0, second: 0}
# Build a small bounding box (0.25° around the point)
area = [rlat + 0.25, rlon - 0.25, rlat - 0.25, rlon + 0.25]
with {:ok, single_data} <- fetch_single_levels(valid_time, area),
{:ok, pressure_data} <- fetch_pressure_levels(valid_time, area) do
build_profile(rlat, rlon, valid_time, single_data, pressure_data)
end
end
@doc """
Fetch ERA5 single-level (surface) data for a time and area.
"""
@spec fetch_single_levels(DateTime.t(), [number()]) :: {:ok, binary()} | {:error, term()}
def fetch_single_levels(valid_time, area) do
request = %{
"product_type" => ["reanalysis"],
"variable" => @single_level_vars,
"year" => [Calendar.strftime(valid_time, "%Y")],
"month" => [Calendar.strftime(valid_time, "%m")],
"day" => [Calendar.strftime(valid_time, "%d")],
"time" => [Calendar.strftime(valid_time, "%H:00")],
"area" => area,
"data_format" => "grib"
}
do_submit_and_download("reanalysis-era5-single-levels", request)
end
@doc """
Fetch ERA5 pressure-level data for a time and area.
"""
@spec fetch_pressure_levels(DateTime.t(), [number()]) :: {:ok, binary()} | {:error, term()}
def fetch_pressure_levels(valid_time, area) do
request = %{
"product_type" => ["reanalysis"],
"variable" => @pressure_level_vars,
"pressure_level" => @pressure_levels,
"year" => [Calendar.strftime(valid_time, "%Y")],
"month" => [Calendar.strftime(valid_time, "%m")],
"day" => [Calendar.strftime(valid_time, "%d")],
"time" => [Calendar.strftime(valid_time, "%H:00")],
"area" => area,
"data_format" => "grib"
}
do_submit_and_download("reanalysis-era5-pressure-levels", request)
end
@doc false
@spec submit_and_download(String.t(), map()) :: {:ok, binary()} | {:error, term()}
def submit_and_download(dataset, request) do
do_submit_and_download(dataset, request)
end
@doc """
Like `submit_and_download/2` but streams the GRIB2 response directly to
`path` on disk rather than loading the whole body (50-200 MB for month
tiles) into the BEAM heap.
"""
@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
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`.
- `: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.
- `{:rejected, reason}` — CDS accepted the submit (gave us a job id) but
immediately rejected it, typically because our in-flight queue was
over the 150-job cap. Terminal; caller should clean up and re-submit
once cap space is available.
- `{: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()}}
| {:rejected, String.t()}
| {: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
{:body, body} -> {:ok, body}
{:url, href} -> download_file(href, api_key)
end
end
end
defp submit_and_poll(dataset, request) do
api_key = api_key()
if is_nil(api_key) do
{:error, "ERA5_CDS_API_KEY not configured"}
else
with {:ok, job_id} <- submit_request(dataset, request, api_key),
{:ok, source} <- poll_until_complete(job_id, api_key) do
{:ok, source, api_key}
end
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: 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
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
# CDS returns this when the submit was accepted into the user's session
# but immediately rejected server-side — usually because we were over the
# 150-job per-user cap. The worker should abandon the job and re-submit
# once there's room.
defp interpret_status_body(%{"status" => "rejected"} = body, _url, _api_key) do
reason = body["message"] || body["detail"] || "CDS job rejected (likely over cap)"
{:rejected, reason}
end
# `dismissed` is what a manually deleted job transitions to — treat it
# the same as rejected so we drop the row and re-submit.
defp interpret_status_body(%{"status" => "dismissed"} = body, _url, _api_key) do
reason = body["message"] || body["detail"] || "CDS job dismissed"
{:rejected, 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
] ++ req_options()
) do
{:ok, %{status: status, body: body}} when status in [200, 201, 202] ->
job_id = body["jobID"] || body["request_id"] || body["jobId"]
if job_id do
Logger.info("ERA5: submitted job #{job_id} for #{dataset}")
{:ok, job_id}
else
{:error, "ERA5: no job_id in response: #{inspect(body)}"}
end
{:ok, %{status: status, body: body}} ->
{:error, "ERA5 CDS HTTP #{status}: #{inspect(body)}"}
{:error, reason} ->
{:error, "ERA5 CDS request failed: #{inspect(reason)}"}
end
end
# Returns {:ok, {:url, href}} when CDS returns a JSON asset pointer (new CDS)
# or {:ok, {:body, binary}} for direct binary responses (legacy CDS).
defp poll_until_complete(job_id, api_key, attempt \\ 0)
defp poll_until_complete(_job_id, _api_key, attempt) when attempt >= @max_poll_attempts do
{:error, "ERA5: poll timeout after #{@max_poll_attempts} attempts"}
end
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] ++ req_options()
) do
{:ok, %{status: 200, body: %{"status" => "successful"}}} ->
results_url = "#{url}/results"
fetch_results_source(results_url, api_key)
{:ok, %{status: 200, body: %{"status" => status}}}
when status in ["accepted", "running"] ->
Process.sleep(@poll_interval_ms)
poll_until_complete(job_id, api_key, attempt + 1)
{:ok, %{status: 200, body: %{"status" => "failed"} = body}} ->
# CDS often omits "message" — fall back to dumping the whole body so the
# failure isn't summarised as `nil` in oban_jobs.errors.
reason = body["message"] || body["error"] || inspect(body)
{:error, "ERA5 job #{job_id} failed: #{reason}"}
{:ok, %{status: status, body: body}} ->
{:error, "ERA5 poll HTTP #{status}: #{inspect(body)}"}
{:error, reason} ->
{:error, "ERA5 poll failed: #{inspect(reason)}"}
end
end
defp fetch_results_source(url, api_key) 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}}
{:ok, %{status: 200, headers: headers, body: body}} ->
# Direct binary download
content_type =
%Req.Response{headers: headers}
|> Req.Response.get_header("content-type")
|> List.first("")
if String.contains?(content_type, "grib") or is_binary(body) do
{:ok, {:body, body}}
else
{:error, "ERA5: unexpected response format"}
end
{:ok, %{status: status, body: body}} ->
{:error, "ERA5 download HTTP #{status}: #{inspect(body)}"}
{:error, reason} ->
{:error, "ERA5 download failed: #{inspect(reason)}"}
end
end
defp download_file(href, api_key) 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)}"}
end
end
# Streams the GRIB2 response directly to `path`. Uses Req's `into:` with a
# `File.stream!` collectable, so chunks land on disk as they arrive and peak
# heap stays O(chunk size) instead of O(full response). Target path is
# 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_options()
)
case result do
{:ok, %{status: 200}} ->
:ok
{:ok, %{status: status}} ->
File.rm(path)
{:error, "ERA5 file download HTTP #{status}"}
{:error, reason} ->
File.rm(path)
{:error, "ERA5 file download failed: #{inspect(reason)}"}
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),
{:ok, pressure_points} <- extract_point(pressure_grib, lat, lon) do
# ERA5 temperatures are in Kelvin, convert to Celsius
temp_c = kelvin_to_celsius(single_points["TMP:2 m above ground"])
dewpoint_c = kelvin_to_celsius(single_points["DPT:2 m above ground"])
# ERA5 surface pressure is in Pa, convert to mb
pressure_mb = pa_to_mb(single_points["PRES:surface"])
hpbl_m = single_points["HPBL:surface"]
# ERA5 TCWV is in kg/m² which equals mm
pwat_mm = single_points["TCWV:entire atmosphere"]
# Build pressure-level profile
profile =
@pressure_levels
|> Enum.map(fn level_str ->
level = String.to_integer(level_str)
pres_key = "#{level} mb"
%{
"pres" => level * 1.0,
"tmpc" => kelvin_to_celsius(pressure_points["TMP:#{pres_key}"]),
"dwpc" => kelvin_to_celsius(pressure_points["DPT:#{pres_key}"]),
"hght" => geopotential_to_height(pressure_points["HGT:#{pres_key}"])
}
end)
|> Enum.reject(fn p -> is_nil(p["tmpc"]) end)
# Derive refractivity and ducting from profile
params = SoundingParams.derive(profile)
attrs = %{
valid_time: valid_time,
lat: lat,
lon: lon,
profile: profile,
hpbl_m: hpbl_m,
pwat_mm: pwat_mm,
surface_temp_c: temp_c,
surface_dewpoint_c: dewpoint_c,
surface_pressure_mb: pressure_mb
}
attrs =
if params do
Map.merge(attrs, %{
surface_refractivity: params.surface_refractivity,
min_refractivity_gradient: params.min_refractivity_gradient,
ducting_detected: params.ducting_detected || false,
duct_characteristics: params.duct_characteristics
})
else
attrs
end
{:ok, attrs}
end
rescue
e -> {:error, "ERA5 GRIB2 decode failed: #{Exception.message(e)}"}
end
defp extract_point(grib_data, lat, lon) do
case Extractor.extract_points(grib_data, lat, lon) do
{:ok, fields} when map_size(fields) > 0 -> {:ok, fields}
{:ok, _} -> {:error, "no data at #{lat},#{lon}"}
{:error, reason} -> {:error, reason}
end
end
defp kelvin_to_celsius(nil), do: nil
defp kelvin_to_celsius(k), do: k - 273.15
defp pa_to_mb(nil), do: nil
defp pa_to_mb(pa), do: pa / 100.0
defp geopotential_to_height(nil), do: nil
defp geopotential_to_height(gp), do: gp / 9.80665
defp api_key do
System.get_env("ERA5_CDS_API_KEY")
end
end

13
lib/microwaveprop/weather/narr_client.ex
View file

@ -147,9 +147,11 @@ defmodule Microwaveprop.Weather.NarrClient do
Pure-ish: shells out to `cdo -outputtab,name,lev,value -remapnn,lon=…_lat=…`
on the local file. No network.
Returns `{:ok, attrs}` where `attrs` has the same shape
`Era5BatchClient.build_profile_attrs/5` produces, or `{:error, reason}`
on missing file, cdo failure, or empty parser output.
Returns `{:ok, attrs}` — a map with the `era5_profiles` fields
(`:surface_temp_c`, `:surface_dewpoint_c`, `:surface_pressure_mb`,
`:hpbl_m`, `:pwat_mm`, `:profile` list plus the derived fields from
`SoundingParams.derive/1`) — or `{:error, reason}` on missing file,
cdo failure, or empty parser output.
"""
@spec extract_profile_from_file(Path.t(), float(), float()) ::
{:ok, map()} | {:error, term()}
@ -167,8 +169,9 @@ defmodule Microwaveprop.Weather.NarrClient do
GRIB1 with `cdo -merge`, then extracts the profile at `{lat, lon}`
using `extract_profile_from_file/3`.
Returns `{:ok, profile_attrs}` shaped like `Era5BatchClient.build_profile_attrs/5`
or `{:error, reason}`. Temp files are always cleaned up via `try/after`.
Returns `{:ok, profile_attrs}` shaped for direct `Repo.insert_all/3`
into the `era5_profiles` table, or `{:error, reason}`. Temp files are
always cleaned up via `try/after`.
"""
@spec fetch_profile_at(DateTime.t(), {float(), float()}) ::
{:ok, map()} | {:error, term()}

80
lib/microwaveprop/workers/era5_fetch_worker.ex
View file

@ -1,80 +0,0 @@
defmodule Microwaveprop.Workers.Era5FetchWorker do
@moduledoc """
Thin router that ensures an ERA5 profile is available for a single
`{lat, lon, valid_time}`. On a cache miss it enqueues the much more
efficient `Era5MonthBatchWorker` for the month-tile that contains the
requested point, rather than hitting the CDS API point-by-point.
Jobs are unique on `{lat, lon, valid_time}` so duplicate enrichment
requests collapse. The enqueued month-batch worker is itself unique on
`(year, month, tile_lat, tile_lon)` so the first enqueue wins and later
ones are no-ops.
"""
use Oban.Worker,
queue: :era5,
max_attempts: 5,
unique: [
period: :infinity,
states: [:available, :scheduled, :executing, :retryable]
]
alias Microwaveprop.Repo
alias Microwaveprop.Weather.Era5BatchClient
alias Microwaveprop.Weather.Era5Profile
alias Microwaveprop.Workers.Era5MonthBatchWorker
require Logger
@impl Oban.Worker
def backoff(%Oban.Job{attempt: attempt}) do
# Enqueueing the batch is cheap; retry fast so we stop blocking downstream.
min(30 * Integer.pow(2, attempt - 1), _one_hour = 3_600)
end
@impl Oban.Worker
def perform(%Oban.Job{args: %{"lat" => lat, "lon" => lon, "valid_time" => valid_time_str}}) do
{:ok, valid_time, _} = DateTime.from_iso8601(valid_time_str)
rlat = Float.round(lat * 4) / 4
rlon = Float.round(lon * 4) / 4
if has_era5_profile?(rlat, rlon, valid_time) do
:ok
else
{tile_lat, tile_lon} = Era5BatchClient.tile_for_point(rlat, rlon)
Logger.info(
"ERA5: enqueueing month batch for #{valid_time.year}-#{valid_time.month} tile #{tile_lat},#{tile_lon} (triggered by #{rlat},#{rlon} @ #{valid_time_str})"
)
%{
year: valid_time.year,
month: valid_time.month,
tile_lat: tile_lat,
tile_lon: tile_lon
}
|> Era5MonthBatchWorker.new()
|> Oban.insert()
:ok
end
end
defp has_era5_profile?(lat, lon, valid_time) do
import Ecto.Query
dlat = 0.13
dlon = 0.13
time_start = DateTime.add(valid_time, -1800, :second)
time_end = DateTime.add(valid_time, 1800, :second)
Era5Profile
|> where(
[p],
p.lat >= ^(lat - dlat) and p.lat <= ^(lat + dlat) and
p.lon >= ^(lon - dlon) and p.lon <= ^(lon + dlon) and
p.valid_time >= ^time_start and p.valid_time <= ^time_end
)
|> Repo.exists?()
end
end

42
lib/microwaveprop/workers/era5_month_batch_worker.ex
View file

@ -1,42 +0,0 @@
defmodule Microwaveprop.Workers.Era5MonthBatchWorker do
@moduledoc """
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.
On the next deploy cycle, callers will be updated to enqueue
`Era5SubmitWorker` directly and this module can be removed.
"""
use Oban.Worker,
queue: :era5_submit,
max_attempts: 5,
unique: [
period: :infinity,
states: [:available, :scheduled, :executing, :retryable],
keys: [:year, :month, :tile_lat, :tile_lon]
]
alias Microwaveprop.Workers.Era5SubmitWorker
require Logger
@impl Oban.Worker
def backoff(%Oban.Job{attempt: attempt}) do
# 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
Logger.debug("Era5MonthBatchWorker forwarding to Era5SubmitWorker: #{inspect(args)}")
args
|> Era5SubmitWorker.new()
|> Oban.insert()
|> case do
{:ok, _job} -> :ok
{:error, reason} -> {:error, reason}
end
end
end

243
lib/microwaveprop/workers/era5_poll_worker.ex
View file

@ -1,243 +0,0 @@
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
alias Microwaveprop.Workers.Era5SubmitWorker
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
# Snooze delay after a transient CDS HTTP error (5xx, network blip).
# Shorter than the normal poll interval so recovery is fast, but a
# snooze instead of an attempt so a sustained outage doesn't burn
# through max_attempts and orphan the row.
@error_snooze_seconds 60
# A CDS job that's been in 'accepted' state for more than this is stuck
# in the server's queue and almost certainly won't run. Observed in
# prod: pressure-level completes in ~90 min, but single-level
# routinely sits at 'accepted' for 12+ hours under CDS load. The
# previous 4h threshold was firing on normal slow runs, forcing a
# resubmit which then got rejected by the per-user cap — an infinite
# reject→resubmit spiral that burned CDS quota without producing any
# decoded data. Tightened from 18h to 8h while diagnosing the
# zero-ingestion problem: 18h was letting rows poll for ~13h without
# ever producing a profile, 8h recycles stuck rows within a work
# shift while still staying above the ~90 min normal pressure-level
# runtime.
@stuck_after_seconds 8 * 3600
@stuck_after_hours div(@stuck_after_seconds, 3600)
@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)
_other ->
if stuck?(row) do
resubmit_vanished(row, "both", "stuck in CDS queue > #{@stuck_after_hours}h")
else
handle_non_both_done(row, single, pressure)
end
end
end
# True when the row is older than @stuck_after_seconds AND hasn't
# already landed both legs. CDS jobs that sit in 'accepted' state past
# this threshold are effectively dead and we should give up rather
# than polling forever.
defp stuck?(row) do
age_seconds = DateTime.diff(DateTime.utc_now(), row.submitted_at, :second)
age_seconds > @stuck_after_seconds
end
# The non-both-done dispatch is split out to keep handle_row/1 under
# credo's cyclomatic complexity limit. `leg_outcome/1` collapses each
# leg's status into a simple tag (:ok | :terminal | :failed | :error |
# :running) plus a payload so the outer dispatch only has to match on
# the *pair* of tags, not the cross-product of individual statuses.
defp handle_non_both_done(row, single, pressure) do
case {leg_outcome(single), leg_outcome(pressure)} do
{{:terminal, reason}, _} ->
resubmit_vanished(row, "single-level", reason)
{_, {:terminal, reason}} ->
resubmit_vanished(row, "pressure-level", reason)
{{:failed, reason}, _} ->
fail_row(row, "single-level job failed: #{reason}")
{_, {:failed, reason}} ->
fail_row(row, "pressure-level job failed: #{reason}")
{{:error, reason}, _} ->
Logger.warning("Era5Poll: transient single-level check_status error (snoozing): #{reason}")
{:snooze, @error_snooze_seconds}
{_, {:error, reason}} ->
Logger.warning("Era5Poll: transient pressure-level check_status error (snoozing): #{reason}")
{:snooze, @error_snooze_seconds}
_both_running_or_one_done ->
bump_poll_count(row)
{:snooze, @snooze_seconds}
end
end
defp leg_outcome({:done, _source}), do: :ok
defp leg_outcome(:not_found), do: {:terminal, "deleted"}
defp leg_outcome({:rejected, reason}), do: {:terminal, "rejected: #{reason}"}
defp leg_outcome({:failed, reason}), do: {:failed, reason}
defp leg_outcome({:error, reason}), do: {:error, reason}
defp leg_outcome(:running), do: :running
# A CDS job is terminally unreachable — either 404 (deleted) or rejected
# by the server (over the per-user cap). 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. The submit worker's CDS cap guard will snooze the retry
# until there's room on the server. Returning `:discard` keeps Oban from
# wasting attempts on a retry that would hit the same terminal state.
defp resubmit_vanished(row, leg, reason) do
Logger.warning(
"Era5Poll: CDS job terminal (#{leg} #{reason}) 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 terminal (#{reason}); 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!()
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

241
lib/microwaveprop/workers/era5_submit_worker.ex
View file

@ -1,241 +0,0 @@
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
# CDS enforces a server-side per-user cap of 150 in-flight jobs. Each
# tile-month submit lives in ONE era5_cds_jobs row but carries TWO CDS
# job IDs (single-level + pressure-level), so the real in-flight count
# CDS sees is 2 × row_count. The cap guard multiplies accordingly.
# Snoozing (instead of failing or sleeping) releases the Oban slot
# immediately and lets the poll workers drain in-flight jobs before we
# push more in.
#
# Headroom is deliberately generous (30 job slots, effective ceiling
# 120 jobs = 60 rows) because the cap check races between concurrent
# workers: two workers that both observe row_count=59 can both submit,
# landing us at row_count=61 (122 jobs) after both finish. With 30
# slots of headroom the cluster would have to race 15 workers wide to
# clip CDS's hard 150 ceiling. Observed in prod: the old 1× math was
# allowing 74 rows (148 jobs) which triggered CDS reject storms.
@cds_ceiling 150
@cap_headroom 30
@snooze_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}
cds_queue_full?() ->
row_count = Repo.aggregate(Era5CdsJob, :count, :id)
in_flight_jobs = row_count * 2
Logger.info(
"Era5Submit: CDS in-flight cap reached (#{in_flight_jobs}/#{@cds_ceiling} jobs across #{row_count} rows) — snoozing #{year}-#{pad(month)} tile #{tile_lat},#{tile_lon}"
)
{:snooze, @snooze_seconds}
true ->
submit_and_persist(year, month, tile_lat, tile_lon)
end
end
defp cds_queue_full? do
in_flight_jobs_after_submit = Repo.aggregate(Era5CdsJob, :count, :id) * 2 + 2
in_flight_jobs_after_submit > @cds_ceiling - @cap_headroom
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

7
lib/microwaveprop_web/live/contact_live/show.ex
View file

@ -13,9 +13,10 @@ defmodule MicrowavepropWeb.ContactLive.Show do
alias Microwaveprop.Weather
alias Microwaveprop.Weather.HrrrClient
alias Microwaveprop.Weather.HrrrNativeProfile
alias Microwaveprop.Weather.NarrClient
alias Microwaveprop.Workers.ContactWeatherEnqueueWorker
alias Microwaveprop.Workers.Era5FetchWorker
alias Microwaveprop.Workers.HrrrFetchWorker
alias Microwaveprop.Workers.NarrFetchWorker
alias Microwaveprop.Workers.SolarIndexWorker
alias Microwaveprop.Workers.TerrainProfileWorker
@ -671,14 +672,14 @@ defmodule MicrowavepropWeb.ContactLive.Show do
lon = contact.pos1 && (contact.pos1["lon"] || contact.pos1["lng"])
if lat && lon do
valid_time = %{DateTime.truncate(contact.qso_timestamp, :second) | minute: 0, second: 0}
valid_time = NarrClient.snap_to_analysis_hour(contact.qso_timestamp)
%{
"lat" => lat,
"lon" => lon,
"valid_time" => DateTime.to_iso8601(valid_time)
}
|> Era5FetchWorker.new()
|> NarrFetchWorker.new()
|> Oban.insert()
end

82
lib/mix/tasks/era5_backfill.ex
View file

@ -1,82 +0,0 @@
defmodule Mix.Tasks.Era5Backfill do
@shortdoc "Enqueue ERA5 month-batch jobs for contacts without HRRR data"
@moduledoc """
Finds contacts where HRRR status is `:unavailable` (pre-2014 or missing
from the archive) and enqueues the month-tile batch worker for every
tile-month their path points need.
Usage:
mix era5_backfill # scan up to 1000 contacts
mix era5_backfill 5000 # scan up to 5000 contacts
Enqueues `Era5MonthBatchWorker` jobs deduplicated by
`(year, month, tile_lat, tile_lon)`. The per-point `Era5FetchWorker`
path still works for one-off enrichment, but this task goes straight
to the batch worker so backfills don't pay the per-point enqueue cost.
"""
use Mix.Task
import Ecto.Query
alias Microwaveprop.Radio
alias Microwaveprop.Radio.Contact
alias Microwaveprop.Repo
alias Microwaveprop.Weather.Era5BatchClient
alias Microwaveprop.Workers.Era5MonthBatchWorker
@impl Mix.Task
def run(args) do
Mix.Task.run("app.start")
Oban.pause_all_queues(Oban)
limit =
case args do
[n | _] -> String.to_integer(n)
_ -> 1000
end
contacts =
Contact
|> where([c], c.hrrr_status == :unavailable and not is_nil(c.pos1))
|> order_by([c], asc: c.qso_timestamp)
|> limit(^limit)
|> Repo.all()
Mix.shell().info("Found #{length(contacts)} contacts needing ERA5 data")
batches =
contacts
|> Enum.flat_map(fn contact ->
valid_time = %{DateTime.truncate(contact.qso_timestamp, :second) | minute: 0, second: 0}
contact
|> Radio.contact_path_points()
|> Enum.map(fn {lat, lon} ->
{tile_lat, tile_lon} = Era5BatchClient.tile_for_point(lat, lon)
%{
year: valid_time.year,
month: valid_time.month,
tile_lat: tile_lat,
tile_lon: tile_lon
}
end)
end)
|> Enum.uniq()
if batches == [] do
Mix.shell().info("No ERA5 batches to enqueue")
else
# Use Oban.insert/1 (not insert_all) so the worker's unique
# constraint collapses duplicate tile-month jobs against anything
# already queued.
Enum.each(batches, fn args ->
args
|> Era5MonthBatchWorker.new()
|> Oban.insert()
end)
Mix.shell().info("Enqueued #{length(batches)} ERA5 month-tile batches")
end
end
end

93
test/microwaveprop/weather/era5_batch_client_test.exs
View file

@ -1,93 +0,0 @@
defmodule Microwaveprop.Weather.Era5BatchClientTest do
use Microwaveprop.DataCase, async: true
alias Microwaveprop.Weather.Era5BatchClient
alias Microwaveprop.Weather.Era5Profile
describe "tile_for_point/2" do
test "floors positive coordinates to a 2° lattice" do
assert {32, 96} = Era5BatchClient.tile_for_point(32.75, 97.5)
assert {32, 96} = Era5BatchClient.tile_for_point(33.99, 97.99)
assert {34, 98} = Era5BatchClient.tile_for_point(34.0, 98.0)
end
test "floors negative longitudes toward -infinity" do
assert {32, -98} = Era5BatchClient.tile_for_point(32.75, -97.25)
assert {32, -98} = Era5BatchClient.tile_for_point(33.0, -97.01)
assert {32, -100} = Era5BatchClient.tile_for_point(33.0, -99.0)
end
test "floors negative latitudes toward -infinity" do
assert {-2, 0} = Era5BatchClient.tile_for_point(-1.25, 0.5)
assert {-34, -60} = Era5BatchClient.tile_for_point(-33.5, -59.5)
end
test "edges land on the tile to the right/north" do
# 30.0 exactly → tile 30 (south/west edge)
assert {30, 96} = Era5BatchClient.tile_for_point(30.0, 97.5)
end
end
describe "fetch_month_into_db/1 idempotency" do
# This describe block exercises the uncached path which would otherwise
# hit the real CDS API if the developer has ERA5_CDS_API_KEY set via
# direnv. Clear the env var for these tests so the expected "API key not
# configured" short-circuit fires and the tests stay hermetic.
setup do
previous = System.get_env("ERA5_CDS_API_KEY")
System.delete_env("ERA5_CDS_API_KEY")
on_exit(fn ->
if previous, do: System.put_env("ERA5_CDS_API_KEY", previous)
end)
:ok
end
test "returns {:ok, 0} when the month-tile already has any profile" do
# Seed one 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!()
# No CDS API key configured in tests → if the cache check fails, the
# function would return an ERA5_CDS_API_KEY error. {:ok, 0} means the
# short-circuit path was taken.
assert {:ok, 0} =
Era5BatchClient.fetch_month_into_db(%{
year: 2014,
month: 3,
tile_lat: 32,
tile_lon: -98
})
end
test "a profile outside the tile window does not count as cached" do
# This profile is in tile (30, -100), not (32, -98).
%Era5Profile{}
|> Era5Profile.changeset(%{
lat: 30.5,
lon: -99.25,
valid_time: ~U[2014-03-15 12:00:00Z],
profile: []
})
|> Repo.insert!()
# The (32, -98) tile is still uncached, so fetch_month_into_db tries
# to reach CDS and fails with missing API key, which bubbles up
# (rather than :ok, 0).
assert {:error, _} =
Era5BatchClient.fetch_month_into_db(%{
year: 2014,
month: 3,
tile_lat: 32,
tile_lon: -98
})
end
end
end

225
test/microwaveprop/weather/era5_client_test.exs
View file

@ -1,225 +0,0 @@
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
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
test "returns {:rejected, reason} when CDS marks the job as rejected" do
# CDS rejects submits when the user's in-flight queue is over the cap
# (150 jobs). The rejected job still gets a jobID and a GET /jobs/:id
# returns a body with status=rejected. This is terminal — re-polling
# will never transition to successful — so callers should treat it
# like {:failed, _} and re-submit the tile-month.
Req.Test.stub(Era5Client, fn conn ->
Req.Test.json(conn, %{
"jobID" => "cds-job-123",
"status" => "rejected",
"metadata" => %{"origin" => "api"}
})
end)
assert {:rejected, reason} = Era5Client.check_status("cds-job-123")
assert is_binary(reason)
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
describe "pressure_levels/0" do
test "covers upper troposphere and lower stratosphere for the skew-T plot" do
levels = Enum.map(Era5Client.pressure_levels(), &String.to_integer/1)
# Boundary-layer resolution near the surface stays intact.
assert 1000 in levels
assert 700 in levels
# Upper-air levels needed to fill out the skew-T above 700 mb.
for level <- [500, 300, 200, 100] do
assert level in levels, "expected #{level} mb in Era5Client.pressure_levels/0"
end
end
end
end

100
test/microwaveprop/workers/era5_fetch_worker_test.exs
View file

@ -1,100 +0,0 @@
defmodule Microwaveprop.Workers.Era5FetchWorkerTest do
use Microwaveprop.DataCase
use Oban.Testing, repo: Microwaveprop.Repo
alias Microwaveprop.Weather.Era5Profile
alias Microwaveprop.Workers.Era5FetchWorker
alias Microwaveprop.Workers.Era5MonthBatchWorker
describe "perform/1" do
test "enqueues a month-tile batch worker when no profile is cached" do
args = %{"lat" => 32.75, "lon" => -97.25, "valid_time" => "2014-01-15T12:00:00Z"}
Oban.Testing.with_testing_mode(:manual, fn ->
assert :ok = Era5FetchWorker.perform(%Oban.Job{args: args})
jobs = all_enqueued(worker: Era5MonthBatchWorker)
assert [job] = jobs
assert job.args["year"] == 2014
assert job.args["month"] == 1
# 32.75 → tile 32 (floor to even degree), -97.25 → tile -98
assert job.args["tile_lat"] == 32
assert job.args["tile_lon"] == -98
end)
end
test "is a no-op when the profile is already cached" do
valid_time = ~U[2014-01-15 12:00:00Z]
%Era5Profile{}
|> Era5Profile.changeset(%{
lat: 32.75,
lon: -97.25,
valid_time: valid_time,
profile: []
})
|> Repo.insert!()
args = %{"lat" => 32.75, "lon" => -97.25, "valid_time" => DateTime.to_iso8601(valid_time)}
Oban.Testing.with_testing_mode(:manual, fn ->
assert :ok = Era5FetchWorker.perform(%Oban.Job{args: args})
assert [] = all_enqueued(worker: Era5MonthBatchWorker)
end)
end
end
describe "unique constraint" do
test "deduplicates identical (lat, lon, valid_time) jobs across inserts" do
args = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
Oban.Testing.with_testing_mode(:manual, fn ->
{:ok, job1} = Oban.insert(Era5FetchWorker.new(args))
{:ok, job2} = Oban.insert(Era5FetchWorker.new(args))
# Second insert should hit the unique constraint and return the
# existing job (same id) instead of creating a new one.
assert job1.id == job2.id
count = Repo.aggregate(Oban.Job, :count, :id)
assert count == 1
end)
end
test "allows distinct hours on the same date" do
args_00 = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
args_06 = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T06:00:00Z"}
Oban.Testing.with_testing_mode(:manual, fn ->
{:ok, job1} = Oban.insert(Era5FetchWorker.new(args_00))
{:ok, job2} = Oban.insert(Era5FetchWorker.new(args_06))
assert job1.id != job2.id
assert Repo.aggregate(Oban.Job, :count, :id) == 2
end)
end
test "allows distinct grid points for the same valid_time" do
args_a = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
args_b = %{"lat" => 33.00, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
Oban.Testing.with_testing_mode(:manual, fn ->
{:ok, job1} = Oban.insert(Era5FetchWorker.new(args_a))
{:ok, job2} = Oban.insert(Era5FetchWorker.new(args_b))
assert job1.id != job2.id
assert Repo.aggregate(Oban.Job, :count, :id) == 2
end)
end
test "deduplicates across repeated Oban.insert calls" do
args = %{"lat" => 32.75, "lon" => -97.0, "valid_time" => "2014-01-01T00:00:00Z"}
Oban.Testing.with_testing_mode(:manual, fn ->
Enum.each(1..5, fn _ -> Oban.insert(Era5FetchWorker.new(args)) end)
assert Repo.aggregate(Oban.Job, :count, :id) == 1
end)
end
end
end

49
test/microwaveprop/workers/era5_month_batch_worker_test.exs
View file

@ -1,49 +0,0 @@
defmodule Microwaveprop.Workers.Era5MonthBatchWorkerTest do
use Microwaveprop.DataCase
alias Microwaveprop.Workers.Era5MonthBatchWorker
describe "worker config" do
test "max_attempts bumped to 5 so CDS slowness doesn't discard tiles after 3 tries" do
assert Era5MonthBatchWorker.__opts__()[:max_attempts] == 5
end
end
describe "unique constraint" do
test "collapses identical (year, month, tile_lat, tile_lon) jobs" do
args = %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}
Oban.Testing.with_testing_mode(:manual, fn ->
{:ok, job1} = Oban.insert(Era5MonthBatchWorker.new(args))
{:ok, job2} = Oban.insert(Era5MonthBatchWorker.new(args))
assert job1.id == job2.id
assert Repo.aggregate(Oban.Job, :count, :id) == 1
end)
end
test "different months for the same tile are distinct jobs" do
Oban.Testing.with_testing_mode(:manual, fn ->
{:ok, j1} =
Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}))
{:ok, j2} =
Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 4, tile_lat: 32, tile_lon: -98}))
assert j1.id != j2.id
end)
end
test "different tiles in the same month are distinct jobs" do
Oban.Testing.with_testing_mode(:manual, fn ->
{:ok, j1} =
Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}))
{:ok, j2} =
Oban.insert(Era5MonthBatchWorker.new(%{year: 2014, month: 3, tile_lat: 30, tile_lon: -98}))
assert j1.id != j2.id
end)
end
end
end

337
test/microwaveprop/workers/era5_poll_worker_test.exs
View file

@ -1,337 +0,0 @@
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
# Default to "just now" so the stuck-after guard doesn't misfire
# and re-submit in tests that want to exercise status-specific
# branches. Tests that care about the stuck path pass a stale
# submitted_at explicitly.
fresh_submit = DateTime.truncate(DateTime.utc_now(), :second)
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: fresh_submit
},
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 — CDS job rejected (over-cap)" do
test "deletes the row, cleans up the sibling CDS job, re-enqueues submit, 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") ->
Req.Test.json(conn, %{
"jobID" => "cds-single-123",
"status" => "rejected",
"metadata" => %{"origin" => "api"}
})
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)
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 =~ "rejected"
refute Repo.get(Era5CdsJob, row.id)
# Sibling CDS job is cleaned up so we don't burn queue slots waiting on
# a request we've already abandoned.
assert_received {:delete, "/api/retrieve/v1/jobs/cds-pressure-456"}
# A fresh Era5SubmitWorker job is enqueued so the tile-month gets
# re-submitted once the CDS cap has room.
assert_enqueued(
worker: Era5SubmitWorker,
args: %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}
)
end
end
describe "perform/1 — stuck in CDS queue" do
# A CDS job that's been in 'accepted' state for many hours is stuck in
# the server's queue and unlikely to ever run. We track submitted_at
# on era5_cds_jobs, so once the row ages past @stuck_after, poll
# treats it as terminal and re-submits — same path as :not_found /
# :rejected.
#
# Threshold is 8h (see @stuck_after_seconds in Era5PollWorker). We
# observed the previous 18h threshold held a row polling for ~13h
# without catching a stuck state while 0 profiles were being
# ingested. 8h is a diagnostic compromise: short enough to recycle
# stuck rows within a shift, still above the ~90 min pressure-level
# normal completion time so healthy runs aren't interrupted.
test "re-submits when submitted_at is older than 8h and at least one leg is still running" do
stale_submit =
DateTime.utc_now()
|> DateTime.add(-9 * 3600, :second)
|> DateTime.truncate(:second)
row = insert_cds_job(%{submitted_at: stale_submit})
test_pid = self()
Req.Test.stub(Era5Client, fn conn ->
cond do
conn.method == "GET" ->
Req.Test.json(conn, %{"status" => "accepted"})
conn.method == "DELETE" ->
send(test_pid, {:delete, conn.request_path})
Plug.Conn.resp(conn, 204, "")
end
end)
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 =~ "stuck"
refute Repo.get(Era5CdsJob, row.id)
assert_enqueued(
worker: Era5SubmitWorker,
args: %{year: 2014, month: 3, tile_lat: 32, tile_lon: -98}
)
end
test "does NOT re-submit a 6h-old row that's still running (under the 8h threshold)" do
# Observed in prod: CDS pressure-level completes in ~90 min, so
# anything under the 8h diagnostic threshold should keep snoozing,
# not trigger a re-submit storm. The single-level leg is allowed
# to keep running — only the combined age is stale, and 6h is
# still well within the envelope.
slowish_submit =
DateTime.utc_now()
|> DateTime.add(-6 * 3600, :second)
|> DateTime.truncate(:second)
row = insert_cds_job(%{submitted_at: slowish_submit})
Req.Test.stub(Era5Client, fn conn ->
Req.Test.json(conn, %{"status" => "running"})
end)
assert {:snooze, _} =
Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
assert Repo.get(Era5CdsJob, row.id)
end
test "does NOT re-submit when submitted_at is recent even if running" do
# A 30-minute-old row with both legs running is normal — snooze.
fresh_submit =
DateTime.utc_now()
|> DateTime.add(-30 * 60, :second)
|> DateTime.truncate(:second)
row = insert_cds_job(%{submitted_at: fresh_submit})
Req.Test.stub(Era5Client, fn conn ->
Req.Test.json(conn, %{"status" => "running"})
end)
assert {:snooze, _} =
Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
assert Repo.get(Era5CdsJob, row.id)
end
end
describe "perform/1 — transient CDS HTTP error" do
# Observed in prod 2026-04-15: a brief CDS network blip returned HTTP
# 5xx on check_status. The old {:error, _} branches consumed attempts
# and max_attempts=10 burned out in seconds, so 13 rows became
# orphaned (discarded with no cleanup, no active poll worker). The
# fix: treat HTTP errors as transient — snooze without counting an
# attempt. Terminal states (:failed, :rejected, :not_found) still
# drive the fail/resubmit paths.
test "snoozes and keeps the row when single-level check_status returns a transient HTTP error" do
row = insert_cds_job()
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(503, ~s({"error":"service unavailable"}))
conn.method == "GET" and String.contains?(conn.request_path, "cds-pressure-456") ->
Req.Test.json(conn, %{"status" => "running"})
end
end)
assert {:snooze, snooze_s} =
Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
assert snooze_s > 0
assert Repo.get(Era5CdsJob, row.id)
end
test "snoozes and keeps the row when pressure-level check_status returns a transient HTTP error" do
row = insert_cds_job()
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" => "running"})
conn.method == "GET" and String.contains?(conn.request_path, "cds-pressure-456") ->
conn
|> Plug.Conn.put_resp_content_type("application/json")
|> Plug.Conn.resp(502, ~s({"error":"bad gateway"}))
end
end)
assert {:snooze, snooze_s} =
Era5PollWorker.perform(%Oban.Job{args: %{"era5_cds_job_id" => row.id}})
assert snooze_s > 0
assert Repo.get(Era5CdsJob, row.id)
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

258
test/microwaveprop/workers/era5_submit_worker_test.exs
View file

@ -1,258 +0,0 @@
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 "CDS in-flight cap" do
# CDS rejects with "Number of queued requests is limited to 150" when
# a user has too many pending jobs. Each tile-month lives in ONE
# era5_cds_jobs row but carries TWO CDS job IDs (single-level +
# pressure-level), so the real in-flight count is 2 × row_count. The
# cap guard multiplies accordingly. The exact threshold (headroom) is
# tuned in the worker; the tests below exercise both the obvious
# "way above the cap" path and a row count that should only trip the
# snooze if the 2× math is in place.
@cds_ceiling 150
test "snoozes when era5_cds_jobs count is at or above the snooze threshold" do
# 140 in-flight is above any reasonable snooze threshold.
seed_count = @cds_ceiling - 10
rows =
for i <- 1..seed_count do
%{
id: Ecto.UUID.generate(),
year: 2014,
month: 3,
tile_lat: 32,
tile_lon: -98 - i,
single_job_id: "seed-single-#{i}",
pressure_job_id: "seed-pressure-#{i}",
submitted_at: DateTime.truncate(DateTime.utc_now(), :second),
inserted_at: DateTime.truncate(DateTime.utc_now(), :second),
updated_at: DateTime.truncate(DateTime.utc_now(), :second),
poll_count: 0
}
end
Repo.insert_all(Era5CdsJob, rows)
Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted when capped" end)
Oban.Testing.with_testing_mode(:manual, fn ->
assert {:snooze, seconds} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
assert is_integer(seconds) and seconds > 0
end)
# No new era5_cds_jobs row inserted for the requested tile.
refute Repo.one(
from j in Era5CdsJob,
where: j.year == 2014 and j.month == 3 and j.tile_lat == 32 and j.tile_lon == -98
)
end
test "snoozes when row count × 2 crosses the cap (each row = two CDS jobs)" do
# 65 rows = 130 in-flight CDS jobs — above the effective snooze
# threshold even though row count alone is nowhere near 150. This
# is the scenario that was escaping the old guard and driving the
# per-user-cap reject storms in prod.
rows =
for i <- 1..65 do
%{
id: Ecto.UUID.generate(),
year: 2014,
month: rem(i - 1, 12) + 1,
tile_lat: 32,
tile_lon: -98 - i,
single_job_id: "x2-single-#{i}",
pressure_job_id: "x2-pressure-#{i}",
submitted_at: DateTime.truncate(DateTime.utc_now(), :second),
inserted_at: DateTime.truncate(DateTime.utc_now(), :second),
updated_at: DateTime.truncate(DateTime.utc_now(), :second),
poll_count: 0
}
end
Repo.insert_all(Era5CdsJob, rows)
Req.Test.stub(Era5Client, fn _ -> raise "CDS should not be contacted when capped" end)
Oban.Testing.with_testing_mode(:manual, fn ->
assert {:snooze, seconds} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
assert is_integer(seconds) and seconds > 0
end)
refute Repo.one(
from j in Era5CdsJob,
where: j.year == 2014 and j.month == 3 and j.tile_lat == 32 and j.tile_lon == -98
)
end
test "submits normally when in-flight count is well below the ceiling" do
# 10 in-flight — comfortably under the cap.
rows =
for i <- 1..10 do
%{
id: Ecto.UUID.generate(),
year: 2013,
month: i,
tile_lat: 40,
tile_lon: -90,
single_job_id: "warm-single-#{i}",
pressure_job_id: "warm-pressure-#{i}",
submitted_at: DateTime.truncate(DateTime.utc_now(), :second),
inserted_at: DateTime.truncate(DateTime.utc_now(), :second),
updated_at: DateTime.truncate(DateTime.utc_now(), :second),
poll_count: 0
}
end
Repo.insert_all(Era5CdsJob, rows)
Req.Test.stub(Era5Client, fn conn ->
dataset_id =
cond do
String.contains?(conn.request_path, "single-levels") -> "fresh-single"
String.contains?(conn.request_path, "pressure-levels") -> "fresh-pressure"
end
Req.Test.json(conn, %{"jobID" => dataset_id})
end)
Oban.Testing.with_testing_mode(:manual, fn ->
assert {:ok, %Era5CdsJob{}} = Era5SubmitWorker.perform(%Oban.Job{args: @default_args})
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