Add IEM rate limiting, Req retry, and Oban retry for failed weather fetches
IEM was returning 503s under 10 concurrent workers. Three fixes: - Req transient retry with exponential backoff on IEM requests - WeatherFetchWorker Oban job retries failed ASOS/RAOB fetches 2-3h later with random jitter to avoid thundering herd - Import script concurrency reduced 10→5, weather queue capped at 3
This commit is contained in:
parent
6f16395f44
commit
5124cc09c1
6 changed files with 377 additions and 6 deletions
|
|
@ -44,7 +44,7 @@ config :microwaveprop, MicrowavepropWeb.Endpoint,
|
|||
|
||||
config :microwaveprop, Oban,
|
||||
repo: Microwaveprop.Repo,
|
||||
queues: [solar: 1],
|
||||
queues: [solar: 1, weather: 3],
|
||||
plugins: [
|
||||
{Oban.Plugins.Cron,
|
||||
crontab: [
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ config :microwaveprop, MicrowavepropWeb.Endpoint,
|
|||
|
||||
# Run Oban jobs inline during tests
|
||||
config :microwaveprop, Oban, testing: :inline
|
||||
config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}]
|
||||
config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}, retry: false]
|
||||
|
||||
# Route HTTP requests through Req.Test stubs
|
||||
config :microwaveprop, solar_req_options: [plug: {Req.Test, Microwaveprop.Weather.SolarClient}]
|
||||
|
|
|
|||
|
|
@ -74,7 +74,15 @@ defmodule Microwaveprop.Weather.IemClient do
|
|||
end
|
||||
|
||||
defp req_options do
|
||||
Application.get_env(:microwaveprop, :iem_req_options, []) ++ [retry: false]
|
||||
defaults = [retry: :transient, max_retries: 3, retry_delay: &retry_delay/1]
|
||||
overrides = Application.get_env(:microwaveprop, :iem_req_options, [])
|
||||
Keyword.merge(defaults, overrides)
|
||||
end
|
||||
|
||||
defp retry_delay(n) do
|
||||
base = Integer.pow(2, n) * 1_000
|
||||
jitter = :rand.uniform(500)
|
||||
base + jitter
|
||||
end
|
||||
|
||||
# --- Parsers ---
|
||||
|
|
|
|||
105
lib/microwaveprop/workers/weather_fetch_worker.ex
Normal file
105
lib/microwaveprop/workers/weather_fetch_worker.ex
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
defmodule Microwaveprop.Workers.WeatherFetchWorker do
|
||||
@moduledoc false
|
||||
use Oban.Worker, queue: :weather, max_attempts: 5
|
||||
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Weather
|
||||
alias Microwaveprop.Weather.IemClient
|
||||
alias Microwaveprop.Weather.SoundingParams
|
||||
alias Microwaveprop.Weather.Station
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: %{"fetch_type" => "asos"} = args}) do
|
||||
%{
|
||||
"station_id" => station_id,
|
||||
"station_code" => station_code,
|
||||
"start_dt" => start_dt_str,
|
||||
"end_dt" => end_dt_str
|
||||
} = args
|
||||
|
||||
{:ok, start_dt, _} = DateTime.from_iso8601(start_dt_str)
|
||||
{:ok, end_dt, _} = DateTime.from_iso8601(end_dt_str)
|
||||
|
||||
case Repo.get(Station, station_id) do
|
||||
nil ->
|
||||
:ok
|
||||
|
||||
station ->
|
||||
if Weather.has_surface_observations?(station.id, start_dt, end_dt) do
|
||||
:ok
|
||||
else
|
||||
case IemClient.fetch_asos(station_code, start_dt, end_dt) do
|
||||
{:ok, rows} ->
|
||||
Enum.each(rows, fn row ->
|
||||
if row.observed_at, do: Weather.upsert_surface_observation(station, row)
|
||||
end)
|
||||
|
||||
:ok
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def perform(%Oban.Job{args: %{"fetch_type" => "raob"} = args}) do
|
||||
%{
|
||||
"station_id" => station_id,
|
||||
"station_code" => station_code,
|
||||
"sounding_time" => sounding_time_str
|
||||
} = args
|
||||
|
||||
{:ok, sounding_time, _} = DateTime.from_iso8601(sounding_time_str)
|
||||
|
||||
case Repo.get(Station, station_id) do
|
||||
nil ->
|
||||
:ok
|
||||
|
||||
station ->
|
||||
if Weather.has_sounding?(station.id, sounding_time) do
|
||||
:ok
|
||||
else
|
||||
case IemClient.fetch_raob(station_code, sounding_time) do
|
||||
{:ok, [parsed | _]} ->
|
||||
params = SoundingParams.derive(parsed.profile)
|
||||
|
||||
sounding_attrs =
|
||||
if params do
|
||||
%{
|
||||
observed_at: parsed.observed_at,
|
||||
profile: parsed.profile,
|
||||
level_count: params.level_count,
|
||||
surface_pressure_mb: params.surface_pressure_mb,
|
||||
surface_temp_c: params.surface_temp_c,
|
||||
surface_dewpoint_c: params.surface_dewpoint_c,
|
||||
surface_refractivity: params.surface_refractivity,
|
||||
min_refractivity_gradient: params.min_refractivity_gradient,
|
||||
boundary_layer_depth_m: params.boundary_layer_depth_m,
|
||||
precipitable_water_mm: params.precipitable_water_mm,
|
||||
k_index: params.k_index,
|
||||
lifted_index: params.lifted_index,
|
||||
ducting_detected: params.ducting_detected,
|
||||
duct_characteristics: params.duct_characteristics
|
||||
}
|
||||
else
|
||||
%{
|
||||
observed_at: parsed.observed_at,
|
||||
profile: parsed.profile,
|
||||
level_count: length(parsed.profile)
|
||||
}
|
||||
end
|
||||
|
||||
Weather.upsert_sounding(station, sounding_attrs)
|
||||
:ok
|
||||
|
||||
{:ok, []} ->
|
||||
:ok
|
||||
|
||||
{:error, reason} ->
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -5,6 +5,7 @@ alias Microwaveprop.Weather
|
|||
alias Microwaveprop.Weather.IemClient
|
||||
alias Microwaveprop.Weather.SolarClient
|
||||
alias Microwaveprop.Weather.SoundingParams
|
||||
alias Microwaveprop.Workers.WeatherFetchWorker
|
||||
|
||||
# --- Helpers ---
|
||||
|
||||
|
|
@ -228,7 +229,21 @@ qsos
|
|||
end)
|
||||
|
||||
{:error, reason} ->
|
||||
IO.puts(" ASOS error #{s.station_code}: #{inspect(reason)}")
|
||||
jitter = :rand.uniform(3600)
|
||||
|
||||
IO.puts(
|
||||
" ASOS error #{s.station_code}: #{inspect(reason)}, enqueueing retry"
|
||||
)
|
||||
|
||||
%{
|
||||
"fetch_type" => "asos",
|
||||
"station_id" => station.id,
|
||||
"station_code" => s.station_code,
|
||||
"start_dt" => DateTime.to_iso8601(start_dt),
|
||||
"end_dt" => DateTime.to_iso8601(end_dt)
|
||||
}
|
||||
|> WeatherFetchWorker.new(schedule_in: 7200 + jitter)
|
||||
|> Oban.insert()
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -288,7 +303,20 @@ qsos
|
|||
:ok
|
||||
|
||||
{:error, reason} ->
|
||||
IO.puts(" RAOB error #{s.station_code}: #{inspect(reason)}")
|
||||
jitter = :rand.uniform(3600)
|
||||
|
||||
IO.puts(
|
||||
" RAOB error #{s.station_code}: #{inspect(reason)}, enqueueing retry"
|
||||
)
|
||||
|
||||
%{
|
||||
"fetch_type" => "raob",
|
||||
"station_id" => station.id,
|
||||
"station_code" => s.station_code,
|
||||
"sounding_time" => DateTime.to_iso8601(sounding_time)
|
||||
}
|
||||
|> WeatherFetchWorker.new(schedule_in: 7200 + jitter)
|
||||
|> Oban.insert()
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -297,7 +325,7 @@ qsos
|
|||
end)
|
||||
end
|
||||
end,
|
||||
max_concurrency: 10,
|
||||
max_concurrency: 5,
|
||||
timeout: 60_000,
|
||||
ordered: false
|
||||
)
|
||||
|
|
|
|||
230
test/microwaveprop/workers/weather_fetch_worker_test.exs
Normal file
230
test/microwaveprop/workers/weather_fetch_worker_test.exs
Normal file
|
|
@ -0,0 +1,230 @@
|
|||
defmodule Microwaveprop.Workers.WeatherFetchWorkerTest do
|
||||
use Microwaveprop.DataCase, async: true
|
||||
|
||||
alias Microwaveprop.Weather
|
||||
alias Microwaveprop.Weather.IemClient
|
||||
alias Microwaveprop.Weather.Sounding
|
||||
alias Microwaveprop.Weather.SurfaceObservation
|
||||
alias Microwaveprop.Workers.WeatherFetchWorker
|
||||
|
||||
defp create_asos_station(_) do
|
||||
{:ok, station} =
|
||||
Weather.find_or_create_station(%{
|
||||
station_code: "KORD",
|
||||
station_type: "asos",
|
||||
name: "Chicago O'Hare",
|
||||
lat: 41.98,
|
||||
lon: -87.9
|
||||
})
|
||||
|
||||
%{station: station}
|
||||
end
|
||||
|
||||
defp create_sounding_station(_) do
|
||||
{:ok, station} =
|
||||
Weather.find_or_create_station(%{
|
||||
station_code: "72451",
|
||||
station_type: "sounding",
|
||||
name: "Davenport",
|
||||
lat: 41.61,
|
||||
lon: -90.58
|
||||
})
|
||||
|
||||
%{station: station}
|
||||
end
|
||||
|
||||
defp asos_csv_body do
|
||||
"""
|
||||
#DEBUG,
|
||||
station,valid,tmpf,dwpf,relh,sknt,drct,mslp,alti,skyc1
|
||||
KORD,2026-03-28 18:53,45.0,30.0,55.0,10.0,270,1013.5,29.92,CLR
|
||||
KORD,2026-03-28 19:53,46.0,31.0,54.0,12.0,280,1013.2,29.91,FEW
|
||||
"""
|
||||
end
|
||||
|
||||
defp raob_json_body do
|
||||
%{
|
||||
"profiles" => [
|
||||
%{
|
||||
"valid" => "2026-03-28 12:00:00+00:00",
|
||||
"profile" => [
|
||||
%{"pres" => 1000.0, "hght" => 200.0, "tmpc" => 15.0, "dwpc" => 10.0, "drct" => 180, "sknt" => 5},
|
||||
%{"pres" => 925.0, "hght" => 800.0, "tmpc" => 10.0, "dwpc" => 5.0, "drct" => 200, "sknt" => 10},
|
||||
%{"pres" => 850.0, "hght" => 1500.0, "tmpc" => 5.0, "dwpc" => 0.0, "drct" => 220, "sknt" => 15},
|
||||
%{"pres" => 700.0, "hght" => 3000.0, "tmpc" => -5.0, "dwpc" => -10.0, "drct" => 240, "sknt" => 25},
|
||||
%{"pres" => 500.0, "hght" => 5500.0, "tmpc" => -20.0, "dwpc" => -30.0, "drct" => 260, "sknt" => 40}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
end
|
||||
|
||||
describe "ASOS perform/1" do
|
||||
setup :create_asos_station
|
||||
|
||||
test "fetches and upserts surface observations", %{station: station} do
|
||||
Req.Test.stub(IemClient, fn conn ->
|
||||
Req.Test.text(conn, asos_csv_body())
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"fetch_type" => "asos",
|
||||
"station_id" => station.id,
|
||||
"station_code" => "KORD",
|
||||
"start_dt" => "2026-03-28T16:53:00Z",
|
||||
"end_dt" => "2026-03-28T20:53:00Z"
|
||||
}
|
||||
|
||||
assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args})
|
||||
assert Repo.aggregate(SurfaceObservation, :count) == 2
|
||||
end
|
||||
|
||||
test "skips fetch when observations already exist", %{station: station} do
|
||||
start_dt = ~U[2026-03-28 16:53:00Z]
|
||||
end_dt = ~U[2026-03-28 20:53:00Z]
|
||||
|
||||
Weather.upsert_surface_observation(station, %{
|
||||
observed_at: ~U[2026-03-28 18:53:00Z],
|
||||
temp_f: 45.0,
|
||||
dewpoint_f: 30.0
|
||||
})
|
||||
|
||||
# Stub should NOT be called — if it is, it will raise
|
||||
Req.Test.stub(IemClient, fn conn ->
|
||||
Plug.Conn.send_resp(conn, 500, "Should not be called")
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"fetch_type" => "asos",
|
||||
"station_id" => station.id,
|
||||
"station_code" => "KORD",
|
||||
"start_dt" => DateTime.to_iso8601(start_dt),
|
||||
"end_dt" => DateTime.to_iso8601(end_dt)
|
||||
}
|
||||
|
||||
assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args})
|
||||
assert Repo.aggregate(SurfaceObservation, :count) == 1
|
||||
end
|
||||
|
||||
test "returns error on HTTP failure to trigger Oban retry", %{station: station} do
|
||||
Req.Test.stub(IemClient, fn conn ->
|
||||
Plug.Conn.send_resp(conn, 503, "Service Unavailable")
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"fetch_type" => "asos",
|
||||
"station_id" => station.id,
|
||||
"station_code" => "KORD",
|
||||
"start_dt" => "2026-03-28T16:53:00Z",
|
||||
"end_dt" => "2026-03-28T20:53:00Z"
|
||||
}
|
||||
|
||||
assert {:error, "IEM ASOS HTTP 503"} = WeatherFetchWorker.perform(%Oban.Job{args: args})
|
||||
end
|
||||
|
||||
test "returns :ok when station no longer exists" do
|
||||
Req.Test.stub(IemClient, fn conn ->
|
||||
Plug.Conn.send_resp(conn, 500, "Should not be called")
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"fetch_type" => "asos",
|
||||
"station_id" => Ecto.UUID.generate(),
|
||||
"station_code" => "GONE",
|
||||
"start_dt" => "2026-03-28T16:53:00Z",
|
||||
"end_dt" => "2026-03-28T20:53:00Z"
|
||||
}
|
||||
|
||||
assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args})
|
||||
end
|
||||
end
|
||||
|
||||
describe "RAOB perform/1" do
|
||||
setup :create_sounding_station
|
||||
|
||||
test "fetches and upserts sounding with derived params", %{station: station} do
|
||||
Req.Test.stub(IemClient, fn conn ->
|
||||
Req.Test.json(conn, raob_json_body())
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"fetch_type" => "raob",
|
||||
"station_id" => station.id,
|
||||
"station_code" => "72451",
|
||||
"sounding_time" => "2026-03-28T12:00:00Z"
|
||||
}
|
||||
|
||||
assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args})
|
||||
assert Repo.aggregate(Sounding, :count) == 1
|
||||
|
||||
sounding = Repo.one(Sounding)
|
||||
assert sounding.station_id == station.id
|
||||
assert sounding.level_count == 5
|
||||
assert sounding.surface_temp_c == 15.0
|
||||
end
|
||||
|
||||
test "skips when sounding already exists", %{station: station} do
|
||||
Weather.upsert_sounding(station, %{
|
||||
observed_at: ~U[2026-03-28 12:00:00Z],
|
||||
profile: [],
|
||||
level_count: 0
|
||||
})
|
||||
|
||||
Req.Test.stub(IemClient, fn conn ->
|
||||
Plug.Conn.send_resp(conn, 500, "Should not be called")
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"fetch_type" => "raob",
|
||||
"station_id" => station.id,
|
||||
"station_code" => "72451",
|
||||
"sounding_time" => "2026-03-28T12:00:00Z"
|
||||
}
|
||||
|
||||
assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args})
|
||||
assert Repo.aggregate(Sounding, :count) == 1
|
||||
end
|
||||
|
||||
test "handles empty profiles", %{station: station} do
|
||||
Req.Test.stub(IemClient, fn conn ->
|
||||
Req.Test.json(conn, %{"profiles" => []})
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"fetch_type" => "raob",
|
||||
"station_id" => station.id,
|
||||
"station_code" => "72451",
|
||||
"sounding_time" => "2026-03-28T12:00:00Z"
|
||||
}
|
||||
|
||||
assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args})
|
||||
assert Repo.aggregate(Sounding, :count) == 0
|
||||
end
|
||||
|
||||
test "returns error on HTTP failure", %{station: station} do
|
||||
Req.Test.stub(IemClient, fn conn ->
|
||||
Plug.Conn.send_resp(conn, 503, "Service Unavailable")
|
||||
end)
|
||||
|
||||
args = %{
|
||||
"fetch_type" => "raob",
|
||||
"station_id" => station.id,
|
||||
"station_code" => "72451",
|
||||
"sounding_time" => "2026-03-28T12:00:00Z"
|
||||
}
|
||||
|
||||
assert {:error, "IEM RAOB HTTP 503"} = WeatherFetchWorker.perform(%Oban.Job{args: args})
|
||||
end
|
||||
|
||||
test "returns :ok when station no longer exists" do
|
||||
args = %{
|
||||
"fetch_type" => "raob",
|
||||
"station_id" => Ecto.UUID.generate(),
|
||||
"station_code" => "GONE",
|
||||
"sounding_time" => "2026-03-28T12:00:00Z"
|
||||
}
|
||||
|
||||
assert :ok = WeatherFetchWorker.perform(%Oban.Job{args: args})
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue