perf(weather): token-bucket IEM limiter + parallel HRRR surface/pressure

Two independent wins the latest prod telemetry pointed at:

1. Iowa Environmental Mesonet rate limiter. IEM throttles per source
   IP across all in-flight requests, so dropping the `:weather` Oban
   queue to 1/pod wasn't enough — workers were still issuing back-to-
   back requests inside each job and drawing a steady stream of HTTP
   429s (1,396 retryable on the queue). A GenServer-based token bucket
   serialises acquire() with a 700ms min gap per pod; three pods give
   ~4 req/sec cluster-wide, well under the observed 429 threshold.
   Wrapped around every IemClient fetch.

2. Parallel HRRR surface + pressure grid fetch. The two products live
   in separate wrfsfcf / wrfprsf GRIB files, so their fetch → wgrib2
   → decode pipelines are independent; running them sequentially was
   adding ~30s per forecast hour for no reason. Task.async the pair,
   merge at the end. Halves per-fh wall time and should unblock some
   of the missing-hour cases we saw on the 14:00Z chain.

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]

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,

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
 

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}

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

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