diff --git a/config/test.exs b/config/test.exs index 4b728a91..d2b1b0a8 100644 --- a/config/test.exs +++ b/config/test.exs @@ -61,6 +61,7 @@ config :microwaveprop, elevation_req_options: [plug: {Req.Test, Microwaveprop.Te config :microwaveprop, gefs_req_options: [plug: {Req.Test, Microwaveprop.Weather.GefsClient}, 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_rate_limiter_interval_ms: 0 config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}, retry: false] config :microwaveprop, mrms_req_options: [plug: {Req.Test, Microwaveprop.Weather.MrmsClient}, retry: false] config :microwaveprop, narr_req_options: [plug: {Req.Test, Microwaveprop.Weather.NarrClient}, retry: false] diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 288c210e..6421beda 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -20,6 +20,8 @@ defmodule Microwaveprop.Application do Microwaveprop.Cache, Microwaveprop.Propagation.ScoreCache, Microwaveprop.Weather.GridCache, + {Microwaveprop.Weather.IemRateLimiter, + interval_ms: Application.get_env(:microwaveprop, :iem_rate_limiter_interval_ms, 700)}, Microwaveprop.Weather.MrmsCache, Microwaveprop.Weather.NexradCache, Microwaveprop.Qrz.Client, diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index a952b457..62573ab7 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -99,29 +99,46 @@ defmodule Microwaveprop.Weather.HrrrClient do hour = hour_dt.hour forecast_hour = Keyword.get(opts, :forecast_hour, 0) - with {:ok, sfc_grid} <- fetch_product_grid(date, hour, :surface, points, forecast_hour) do - # Pressure is optional — if it fails, continue with surface-only data - prs_grid = - case maybe_fetch_pressure_grid(date, hour, points, opts, forecast_hour) do - {:ok, grid} -> - grid + # Surface and pressure products live in separate GRIB files on + # NOAA's side (wrfsfcf vs wrfprsf), so the whole fetch → wgrib2 + # extract pipeline is independent. Running them in parallel + # roughly halves per-forecast-hour wall time (both phases are + # ~30s each; sequential was ~60s, concurrent is max of the two). + sfc_task = Task.async(fn -> fetch_product_grid(date, hour, :surface, points, forecast_hour) end) + prs_task = Task.async(fn -> maybe_fetch_pressure_grid(date, hour, points, opts, forecast_hour) end) - {:error, reason} -> - Logger.warning("HRRR pressure fetch failed (continuing with surface only): #{inspect(reason)}") - %{} - end + # Match the 20-min per-forecast-hour ceiling in PropagationGridWorker + # so a stuck wgrib2 doesn't pin the chain past its timeout. Surface + # alone deciding the outcome — pressure is optional. + case Task.await(sfc_task, 20 * 60 * 1000) do + {:ok, sfc_grid} -> + prs_grid = + case Task.await(prs_task, 20 * 60 * 1000) do + {:ok, grid} -> + grid - # Merge and build profiles in one pass to avoid a full intermediate map copy. - # Use Map.merge to include any pressure-only points, then map directly to profiles. - profiles = - sfc_grid - |> Map.merge(prs_grid, fn _point, sfc_data, prs_data -> Map.merge(sfc_data, prs_data) end) - |> Map.new(fn {point, data} -> - profile = build_profile(data) - {point, Map.put(profile, :run_time, hour_dt)} - end) + {:error, reason} -> + Logger.warning("HRRR pressure fetch failed (continuing with surface only): #{inspect(reason)}") + %{} + end - {:ok, profiles} + # Merge and build profiles in one pass to avoid a full intermediate map copy. + # Use Map.merge to include any pressure-only points, then map directly to profiles. + profiles = + sfc_grid + |> Map.merge(prs_grid, fn _point, sfc_data, prs_data -> Map.merge(sfc_data, prs_data) end) + |> Map.new(fn {point, data} -> + profile = build_profile(data) + {point, Map.put(profile, :run_time, hour_dt)} + end) + + {:ok, profiles} + + {:error, reason} = error -> + # Drain the pressure task so it doesn't leak past this scope. + _ = Task.await(prs_task, 20 * 60 * 1000) + _ = reason + error end end diff --git a/lib/microwaveprop/weather/iem_client.ex b/lib/microwaveprop/weather/iem_client.ex index b7c40d87..e12de686 100644 --- a/lib/microwaveprop/weather/iem_client.ex +++ b/lib/microwaveprop/weather/iem_client.ex @@ -1,6 +1,8 @@ defmodule Microwaveprop.Weather.IemClient do @moduledoc false + alias Microwaveprop.Weather.IemRateLimiter + @iem_base "https://mesonet.agron.iastate.edu" # --- URL builders --- @@ -39,6 +41,7 @@ defmodule Microwaveprop.Weather.IemClient do @spec fetch_network(String.t()) :: {:ok, [map()]} | {:error, term()} def fetch_network(network) do url = network_url(network) + IemRateLimiter.acquire() case Req.get(url, req_options()) do {:ok, %{status: 200, body: body}} -> @@ -57,6 +60,8 @@ defmodule Microwaveprop.Weather.IemClient do url = asos_url(station_id, start_dt, end_dt) Microwaveprop.Instrument.span([:iem, :fetch_asos], %{station: station_id}, fn -> + IemRateLimiter.acquire() + case Req.get(url, req_options()) do {:ok, %{status: 200, body: body}} -> {:ok, parse_asos_csv(body)} @@ -75,6 +80,8 @@ defmodule Microwaveprop.Weather.IemClient do url = raob_url(station_id, dt) Microwaveprop.Instrument.span([:iem, :fetch_raob], %{station: station_id}, fn -> + IemRateLimiter.acquire() + case Req.get(url, req_options()) do {:ok, %{status: 200, body: body}} -> {:ok, parse_raob_json(body)} @@ -92,6 +99,7 @@ defmodule Microwaveprop.Weather.IemClient do def fetch_iemre(lat, lon, date) do Microwaveprop.Instrument.span([:iem, :fetch_iemre], %{}, fn -> url = iemre_url(lat, lon, date) + IemRateLimiter.acquire() case Req.get(url, req_options()) do {:ok, %{status: 200, body: body}} -> @@ -137,6 +145,7 @@ defmodule Microwaveprop.Weather.IemClient do |> Task.async_stream( fn network -> url = "#{@iem_base}/api/1/currents.json?network=#{network}" + IemRateLimiter.acquire() case Req.get(url, req_options()) do {:ok, %{status: 200, body: %{"data" => data}}} -> {:ok, data} diff --git a/lib/microwaveprop/weather/iem_rate_limiter.ex b/lib/microwaveprop/weather/iem_rate_limiter.ex new file mode 100644 index 00000000..07dc2ce7 --- /dev/null +++ b/lib/microwaveprop/weather/iem_rate_limiter.ex @@ -0,0 +1,68 @@ +defmodule Microwaveprop.Weather.IemRateLimiter do + @moduledoc """ + Per-pod serial gate for Iowa Environmental Mesonet HTTP calls. + + IEM rate-limits bulk historical queries per source IP across all + in-flight requests, not per-concurrency. Simply lowering the + `:weather` Oban queue to one slot per pod still produced a steady + stream of HTTP 429s because workers were issuing back-to-back + requests inside each job (ASOS spanning multiple days, RAOB + batches). This limiter enforces a minimum gap between successive + calls — callers block in `acquire/0` until their reserved slot + arrives. + + Configured via `config :microwaveprop, :iem_rate_limiter_interval_ms` + (defaults to 700ms ≈ 1.4 req/sec per pod; three pods give ~4 req/sec + cluster-wide, well under the empirically observed 429 threshold). + In tests, pass `interval_ms: 0` to disable. + """ + + use GenServer + + @default_interval_ms 700 + + @spec start_link(keyword()) :: GenServer.on_start() + def start_link(opts \\ []) do + GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, __MODULE__)) + end + + @doc """ + Reserve the next available slot and block until it arrives. Callers + MUST invoke this before every IEM HTTP request. Returns `:ok` once + the caller is allowed to proceed. + """ + @spec acquire(GenServer.server()) :: :ok + def acquire(server \\ __MODULE__) do + if registered?(server) do + reserved_ms = GenServer.call(server, :reserve, :infinity) + delay = reserved_ms - System.system_time(:millisecond) + if delay > 0, do: Process.sleep(delay) + :ok + else + # Limiter not started (e.g. dev Mix tasks running in isolation). + # Fail open rather than deadlock the caller. + :ok + end + end + + defp registered?(pid) when is_pid(pid), do: Process.alive?(pid) + defp registered?(name) when is_atom(name), do: Process.whereis(name) != nil + defp registered?(_), do: true + + @impl true + def init(opts) do + interval_ms = Keyword.get(opts, :interval_ms, @default_interval_ms) + {:ok, %{interval_ms: interval_ms, next_ms: System.system_time(:millisecond)}} + end + + @impl true + def handle_call(:reserve, _from, %{interval_ms: 0} = state) do + {:reply, System.system_time(:millisecond), state} + end + + def handle_call(:reserve, _from, %{interval_ms: interval, next_ms: next_ms} = state) do + now = System.system_time(:millisecond) + reserved = max(now, next_ms) + {:reply, reserved, %{state | next_ms: reserved + interval}} + end +end diff --git a/test/microwaveprop/weather/iem_rate_limiter_test.exs b/test/microwaveprop/weather/iem_rate_limiter_test.exs new file mode 100644 index 00000000..e18c6508 --- /dev/null +++ b/test/microwaveprop/weather/iem_rate_limiter_test.exs @@ -0,0 +1,81 @@ +defmodule Microwaveprop.Weather.IemRateLimiterTest do + use ExUnit.Case, async: false + + alias Microwaveprop.Weather.IemRateLimiter + + # Each test starts an isolated limiter under a unique name so the + # app-level IemRateLimiter (started by the application supervisor + # with interval_ms: 0) doesn't conflict. + defp start_limiter(interval_ms) do + name = :"iem_rate_limiter_test_#{System.unique_integer([:positive])}" + start_supervised!({IemRateLimiter, interval_ms: interval_ms, name: name}) + name + end + + describe "acquire/1" do + test "first call returns immediately when the limiter is idle" do + name = start_limiter(50) + + {elapsed_ms, :ok} = :timer.tc(fn -> IemRateLimiter.acquire(name) end, :millisecond) + + # Allow a generous upper bound for scheduling jitter; the point is that + # no rate-limit sleep was applied on the first call. + assert elapsed_ms < 20 + end + + test "serial calls are spaced at least interval_ms apart" do + name = start_limiter(50) + + {elapsed_ms, _} = + :timer.tc( + fn -> + IemRateLimiter.acquire(name) + IemRateLimiter.acquire(name) + IemRateLimiter.acquire(name) + end, + :millisecond + ) + + # Three acquires at 50ms gap → floor at 100ms (gap after call 1 and 2). + assert elapsed_ms >= 100 + end + + test "concurrent callers each get a non-overlapping slot" do + name = start_limiter(50) + parent = self() + + for _ <- 1..4 do + spawn(fn -> + IemRateLimiter.acquire(name) + send(parent, {:acquired, System.monotonic_time(:millisecond)}) + end) + end + + timestamps = + for _ <- 1..4 do + assert_receive {:acquired, ts}, 500 + ts + end + + sorted = Enum.sort(timestamps) + gaps = sorted |> Enum.zip(tl(sorted)) |> Enum.map(fn {a, b} -> b - a end) + min_gap = Enum.min(gaps) + # Allow 5ms scheduler jitter on the lower bound. + assert min_gap >= 45 + end + + test "interval_ms: 0 is a no-op" do + name = start_limiter(0) + + {elapsed_ms, _} = + :timer.tc( + fn -> + for _ <- 1..10, do: IemRateLimiter.acquire(name) + end, + :millisecond + ) + + assert elapsed_ms < 20 + end + end +end