From 5124cc09c1ac3c6fdc89efbbd8f46c8a4b33f1da Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 29 Mar 2026 13:24:52 -0500 Subject: [PATCH] Add IEM rate limiting, Req retry, and Oban retry for failed weather fetches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- config/config.exs | 2 +- config/test.exs | 2 +- lib/microwaveprop/weather/iem_client.ex | 10 +- .../workers/weather_fetch_worker.ex | 105 ++++++++ priv/repo/import_weather.exs | 34 ++- .../workers/weather_fetch_worker_test.exs | 230 ++++++++++++++++++ 6 files changed, 377 insertions(+), 6 deletions(-) create mode 100644 lib/microwaveprop/workers/weather_fetch_worker.ex create mode 100644 test/microwaveprop/workers/weather_fetch_worker_test.exs diff --git a/config/config.exs b/config/config.exs index 8b2bf1b6..6ac80d9c 100644 --- a/config/config.exs +++ b/config/config.exs @@ -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: [ diff --git a/config/test.exs b/config/test.exs index 07737462..e49ddac0 100644 --- a/config/test.exs +++ b/config/test.exs @@ -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}] diff --git a/lib/microwaveprop/weather/iem_client.ex b/lib/microwaveprop/weather/iem_client.ex index 5cdfd572..17d37859 100644 --- a/lib/microwaveprop/weather/iem_client.ex +++ b/lib/microwaveprop/weather/iem_client.ex @@ -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 --- diff --git a/lib/microwaveprop/workers/weather_fetch_worker.ex b/lib/microwaveprop/workers/weather_fetch_worker.ex new file mode 100644 index 00000000..a2f197c7 --- /dev/null +++ b/lib/microwaveprop/workers/weather_fetch_worker.ex @@ -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 diff --git a/priv/repo/import_weather.exs b/priv/repo/import_weather.exs index 8e1291d4..a6f87cc2 100644 --- a/priv/repo/import_weather.exs +++ b/priv/repo/import_weather.exs @@ -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 ) diff --git a/test/microwaveprop/workers/weather_fetch_worker_test.exs b/test/microwaveprop/workers/weather_fetch_worker_test.exs new file mode 100644 index 00000000..db80bb6d --- /dev/null +++ b/test/microwaveprop/workers/weather_fetch_worker_test.exs @@ -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