perf(weather): adaptive IemRateLimiter + rebatch mix task

- IemRateLimiter gains AIMD-style adaptive spacing. signal_429/0
  widens the current gap (*= 1.5, capped at max_interval_ms default
  10s); signal_success/0 narrows it back toward the configured base
  (*= 0.92, floored at interval_ms). Self-tunes to IEM's moving
  ceiling without needing the manual "safe for 4 pods" constant.

- IemClient now routes every response through a central handle_response
  helper that fires the widen/narrow feedback signals, eliminating the
  four near-identical case blocks.

- Mix.Tasks.Weather.RebatchAsos collapses any already-queued single-
  station "asos" jobs into the batched "asos_batch" shape the
  enqueuer now emits, so the pending backfill queue converts to the
  new per-request-efficient path instead of draining at the old rate.
  Idempotent; supports --dry-run.

2833 tests + credo green.

lib/microwaveprop/weather/iem_client.ex

@@ -46,16 +46,9 @@ defmodule Microwaveprop.Weather.IemClient do
     url = network_url(network)
     IemRateLimiter.acquire()
 
-    case Req.get(url, req_options()) do
+    url
-      {:ok, %{status: 200, body: body}} ->
+    |> Req.get(req_options())
-        {:ok, parse_network_json(body)}
+    |> handle_response(&parse_network_json/1, "IEM network")
-
-      {:ok, %{status: status}} ->
-        {:error, "IEM network HTTP #{status}"}
-
-      {:error, reason} ->
-        {:error, reason}
-    end
   end
 
   @spec fetch_asos(String.t(), DateTime.t(), DateTime.t()) :: {:ok, [map()]} | {:error, term()}
@@ -65,16 +58,9 @@ defmodule Microwaveprop.Weather.IemClient do
     Microwaveprop.Instrument.span([:iem, :fetch_asos], %{station: station_id}, fn ->
       IemRateLimiter.acquire()
 
-      case Req.get(url, req_options()) do
+      url
-        {:ok, %{status: 200, body: body}} ->
+      |> Req.get(req_options())
-          {:ok, parse_asos_csv(body)}
+      |> handle_response(&parse_asos_csv/1, "IEM ASOS")
-
-        {:ok, %{status: status}} ->
-          {:error, "IEM ASOS HTTP #{status}"}
-
-        {:error, reason} ->
-          {:error, reason}
-      end
     end)
   end
 
@@ -97,21 +83,17 @@ defmodule Microwaveprop.Weather.IemClient do
     Microwaveprop.Instrument.span([:iem, :fetch_asos_batch], %{count: length(codes)}, fn ->
       IemRateLimiter.acquire()
 
-      case Req.get(url, req_options()) do
+      parse = fn body ->
-        {:ok, %{status: 200, body: body}} ->
+        rows = parse_asos_csv(body)
-          rows = parse_asos_csv(body)
+        grouped = Enum.group_by(rows, & &1.station_code)
-          grouped = Enum.group_by(rows, & &1.station_code)
+        # Ensure every requested code has at least an empty list in the
-          # Ensure every requested code has at least an empty list in the
+        # map so downstream iteration doesn't miss "no data" cases.
-          # map so downstream iteration doesn't miss "no data" cases.
+        Enum.reduce(codes, grouped, &Map.put_new(&2, &1, []))
-          filled = Enum.reduce(codes, grouped, &Map.put_new(&2, &1, []))
-          {:ok, filled}
-
-        {:ok, %{status: status}} ->
-          {:error, "IEM ASOS HTTP #{status}"}
-
-        {:error, reason} ->
-          {:error, reason}
       end
+
+      url
+      |> Req.get(req_options())
+      |> handle_response(parse, "IEM ASOS")
     end)
   end
 
@@ -122,16 +104,9 @@ defmodule Microwaveprop.Weather.IemClient do
     Microwaveprop.Instrument.span([:iem, :fetch_raob], %{station: station_id}, fn ->
       IemRateLimiter.acquire()
 
-      case Req.get(url, req_options()) do
+      url
-        {:ok, %{status: 200, body: body}} ->
+      |> Req.get(req_options())
-          {:ok, parse_raob_json(body)}
+      |> handle_response(&parse_raob_json/1, "IEM RAOB")
-
-        {:ok, %{status: status}} ->
-          {:error, "IEM RAOB HTTP #{status}"}
-
-        {:error, reason} ->
-          {:error, reason}
-      end
     end)
   end
 
@@ -141,19 +116,31 @@ defmodule Microwaveprop.Weather.IemClient do
       url = iemre_url(lat, lon, date)
       IemRateLimiter.acquire()
 
-      case Req.get(url, req_options()) do
+      url
-        {:ok, %{status: 200, body: body}} ->
+      |> Req.get(req_options())
-          {:ok, parse_iemre_json(body)}
+      |> handle_response(&parse_iemre_json/1, "IEM IEMRE")
-
-        {:ok, %{status: status}} ->
-          {:error, "IEM IEMRE HTTP #{status}"}
-
-        {:error, reason} ->
-          {:error, reason}
-      end
     end)
   end
 
+  # Central response handler: fires adaptive-limiter feedback signals
+  # (widen on 429, narrow on 200) before returning the parsed body or
+  # an error tuple. `kind` is the upstream label used in error strings.
+  defp handle_response({:ok, %{status: 200, body: body}}, parser, _kind) do
+    _ = IemRateLimiter.signal_success()
+    {:ok, parser.(body)}
+  end
+
+  defp handle_response({:ok, %{status: 429}}, _parser, kind) do
+    _ = IemRateLimiter.signal_429()
+    {:error, "#{kind} HTTP 429"}
+  end
+
+  defp handle_response({:ok, %{status: status}}, _parser, kind) do
+    {:error, "#{kind} HTTP #{status}"}
+  end
+
+  defp handle_response({:error, reason}, _parser, _kind), do: {:error, reason}
+
   defp req_options do
     defaults = [retry: &retry?/2, max_retries: 5, retry_delay: &retry_delay/1]
     overrides = Application.get_env(:microwaveprop, :iem_req_options, [])

lib/microwaveprop/weather/iem_rate_limiter.ex

@@ -1,32 +1,37 @@
 defmodule Microwaveprop.Weather.IemRateLimiter do
   @moduledoc """
-  Per-pod serial gate for Iowa Environmental Mesonet HTTP calls.
+  Per-pod serial gate for Iowa Environmental Mesonet HTTP calls with
+  AIMD-style adaptive spacing.
 
-  IEM rate-limits bulk historical queries per source IP across all
+  IEM rate-limits per source IP across all in-flight requests, not
-  in-flight requests, not per-concurrency. Simply lowering the
+  per-concurrency. Simply lowering the `:weather` Oban queue to one
-  `:weather` Oban queue to one slot per pod still produced a steady
+  slot per pod still produced a steady stream of HTTP 429s because
-  stream of HTTP 429s because workers were issuing back-to-back
+  workers were issuing back-to-back requests inside each job (ASOS
-  requests inside each job (ASOS spanning multiple days, RAOB
+  spanning multiple days, RAOB batches). This limiter enforces a
-  batches). This limiter enforces a minimum gap between successive
+  minimum gap between successive calls — callers block in `acquire/0`
-  calls — callers block in `acquire/0` until their reserved slot
+  until their reserved slot arrives.
-  arrives.
+
+  **Adaptive gap.** The current gap lives between `interval_ms` (base,
+  floor) and `max_interval_ms` (ceiling). Each `signal_429/0` widens
+  the gap by `*= widen_factor`; each `signal_success/0` narrows it by
+  `*= narrow_factor`. Successes are the common case; 429s are the
+  feedback signal that IEM's moving ceiling has tightened. Over a
+  steady stream of calls the gap self-tunes to just under the IEM
+  429 threshold without requiring manual tuning.
+
+  Configured via:
+    * `:iem_rate_limiter_interval_ms` — base gap (default 1500)
+    * `:iem_rate_limiter_max_interval_ms` — ceiling (default 10_000)
 
-  Configured via `config :microwaveprop, :iem_rate_limiter_interval_ms`
-  (defaults to 1500ms ≈ 0.67 req/sec per pod; four pods give ~2.7
-  req/sec cluster-wide). Previously set at 700ms, which put four pods
-  at ~5.7 req/sec fleet-wide — above IEM's observed 429 threshold —
-  and the resulting retry storm (Req exponential backoff up to 32 s)
-  piled concurrent HTTP sleeps on top of live workers until Bandit's
-  acceptor and Oban's leader heartbeat both started tripping. The
-  limiter is per-pod, so cluster-synchronous bursts are still possible
-  when every pod reserves `now + interval` simultaneously; moving to
-  a global limiter is out of scope for this tuning pass.
   In tests, pass `interval_ms: 0` to disable.
   """
 
   use GenServer
 
   @default_interval_ms 1500
+  @default_max_interval_ms 10_000
+  @widen_factor 1.5
+  @narrow_factor 0.92
 
   @spec start_link(keyword()) :: GenServer.on_start()
   def start_link(opts \\ []) do
@@ -52,24 +57,86 @@ defmodule Microwaveprop.Weather.IemRateLimiter do
     end
   end
 
+  @doc """
+  Widen the gap because we just got a 429. Cheap cast.
+  """
+  @spec signal_429(GenServer.server()) :: :ok
+  def signal_429(server \\ __MODULE__) do
+    if registered?(server), do: GenServer.cast(server, :widen), else: :ok
+    :ok
+  end
+
+  @doc """
+  Narrow the gap toward base because we just got a 200 (or any non-429
+  success). Cheap cast.
+  """
+  @spec signal_success(GenServer.server()) :: :ok
+  def signal_success(server \\ __MODULE__) do
+    if registered?(server), do: GenServer.cast(server, :narrow), else: :ok
+    :ok
+  end
+
+  @doc """
+  Read the current gap. Primarily for introspection + tests.
+  """
+  @spec current_interval_ms(GenServer.server()) :: non_neg_integer()
+  def current_interval_ms(server \\ __MODULE__) do
+    if registered?(server), do: GenServer.call(server, :current_interval), else: 0
+  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)
+    base = Keyword.get(opts, :interval_ms, @default_interval_ms)
-    {:ok, %{interval_ms: interval_ms, next_ms: System.system_time(:millisecond)}}
+    max_interval = Keyword.get(opts, :max_interval_ms, @default_max_interval_ms)
+
+    state = %{
+      base_ms: base,
+      max_ms: max_interval,
+      current_ms: base,
+      next_ms: System.system_time(:millisecond)
+    }
+
+    {:ok, state}
   end
 
   @impl true
-  def handle_call(:reserve, _from, %{interval_ms: 0} = state) do
+  def handle_call(:reserve, _from, %{base_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
+  def handle_call(:reserve, _from, %{current_ms: gap, next_ms: next_ms} = state) do
     now = System.system_time(:millisecond)
     reserved = max(now, next_ms)
-    {:reply, reserved, %{state | next_ms: reserved + interval}}
+    {:reply, reserved, %{state | next_ms: reserved + gap}}
+  end
+
+  def handle_call(:current_interval, _from, state), do: {:reply, state.current_ms, state}
+
+  @impl true
+  def handle_cast(:widen, %{base_ms: 0} = state), do: {:noreply, state}
+
+  def handle_cast(:widen, state) do
+    new_gap =
+      (state.current_ms * @widen_factor)
+      |> round()
+      |> min(state.max_ms)
+      |> max(state.base_ms)
+
+    {:noreply, %{state | current_ms: new_gap}}
+  end
+
+  def handle_cast(:narrow, %{base_ms: 0} = state), do: {:noreply, state}
+
+  def handle_cast(:narrow, state) do
+    new_gap =
+      (state.current_ms * @narrow_factor)
+      |> round()
+      |> max(state.base_ms)
+
+    {:noreply, %{state | current_ms: new_gap}}
   end
 end

lib/mix/tasks/weather_rebatch_asos.ex

@@ -0,0 +1,112 @@
+defmodule Mix.Tasks.Weather.RebatchAsos do
+  @shortdoc "Collapse pending single-station ASOS jobs into batched jobs"
+
+  @moduledoc """
+  Converts every pending single-station `WeatherFetchWorker` ASOS job
+  into the batched (`asos_batch`) shape — one job per unique
+  `(start_dt, end_dt)` window covering every station that had a
+  pending job for that window.
+
+  Why: the worker still accepts the old `"asos"` fetch_type for
+  backward compat, but every such job pays the full `IemRateLimiter`
+  gap + IEM 429-retry tail for a single station's rows. Collapsing
+  them into batched jobs gets the queue through the backlog at the
+  new per-request efficiency.
+
+  Usage (idempotent — running it twice is a no-op):
+
+      mix weather.rebatch_asos               # rewrites everything
+      mix weather.rebatch_asos --dry-run     # report only
+
+  """
+
+  use Mix.Task
+
+  import Ecto.Query
+
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Workers.WeatherFetchWorker
+
+  require Logger
+
+  @impl Mix.Task
+  def run(args) do
+    {opts, _, _} = OptionParser.parse(args, strict: [dry_run: :boolean])
+    Mix.Task.run("app.start")
+    dry_run? = Keyword.get(opts, :dry_run, false)
+
+    jobs = load_pending_asos_jobs()
+
+    if jobs == [] do
+      Mix.shell().info("No pending `asos` jobs. Nothing to do.")
+    else
+      do_rebatch(jobs, dry_run?)
+    end
+  end
+
+  @doc false
+  def load_pending_asos_jobs do
+    query =
+      from(j in Oban.Job,
+        where:
+          j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and
+            j.state in ["available", "scheduled", "retryable"] and
+            fragment("?->>'fetch_type' = ?", j.args, "asos"),
+        select: %{id: j.id, args: j.args}
+      )
+
+    Repo.all(query)
+  end
+
+  defp do_rebatch(jobs, dry_run?) do
+    groups =
+      jobs
+      |> Enum.group_by(fn %{args: a} ->
+        {Map.get(a, "start_dt"), Map.get(a, "end_dt")}
+      end)
+      |> Enum.reject(fn {{s, e}, _} -> is_nil(s) or is_nil(e) end)
+
+    Mix.shell().info("Found #{length(jobs)} pending `asos` jobs across #{length(groups)} time windows.")
+
+    batch_args =
+      Enum.map(groups, fn {{start_dt, end_dt}, members} ->
+        # Deterministic sort (matches the enqueuer's ordering) so a
+        # re-run with identical membership hits Oban's unique dedup.
+        sorted = Enum.sort_by(members, fn %{args: a} -> a["station_code"] end)
+
+        %{
+          "fetch_type" => "asos_batch",
+          "station_ids" => Enum.map(sorted, fn %{args: a} -> a["station_id"] end),
+          "station_codes" => Enum.map(sorted, fn %{args: a} -> a["station_code"] end),
+          "start_dt" => start_dt,
+          "end_dt" => end_dt,
+          "_source_ids" => Enum.map(sorted, & &1.id)
+        }
+      end)
+
+    if dry_run? do
+      Enum.each(batch_args, fn a ->
+        Mix.shell().info("[dry-run] #{length(a["station_codes"])} stations for #{a["start_dt"]}..#{a["end_dt"]}")
+      end)
+    else
+      apply_rebatch(batch_args)
+    end
+  end
+
+  defp apply_rebatch(batch_args) do
+    {inserted, cancelled} =
+      Enum.reduce(batch_args, {0, 0}, fn a, {ins, can} ->
+        source_ids = a["_source_ids"]
+        job_args = Map.delete(a, "_source_ids")
+
+        Repo.transaction(fn ->
+          _ = job_args |> WeatherFetchWorker.new() |> Oban.insert!()
+          {:ok, _n} = Oban.cancel_all_jobs(from(j in Oban.Job, where: j.id in ^source_ids))
+        end)
+
+        {ins + 1, can + length(source_ids)}
+      end)
+
+    Mix.shell().info("Rebatched: inserted #{inserted} batch jobs, cancelled #{cancelled} single-station jobs.")
+  end
+end

test/microwaveprop/weather/iem_rate_limiter_test.exs

@@ -78,4 +78,43 @@ defmodule Microwaveprop.Weather.IemRateLimiterTest do
       assert elapsed_ms < 20
     end
   end
+
+  describe "adaptive gap" do
+    defp start_adaptive(base, max) do
+      name = :"iem_rate_limiter_test_#{System.unique_integer([:positive])}"
+      start_supervised!({IemRateLimiter, interval_ms: base, max_interval_ms: max, name: name})
+      name
+    end
+
+    test "signal_429 widens the current gap" do
+      name = start_adaptive(50, 500)
+      IemRateLimiter.signal_429(name)
+      IemRateLimiter.signal_429(name)
+      gap = IemRateLimiter.current_interval_ms(name)
+      assert gap > 50
+      assert gap <= 500
+    end
+
+    test "signal_success decays the gap back toward base" do
+      name = start_adaptive(50, 500)
+      # Inflate the gap
+      Enum.each(1..5, fn _ -> IemRateLimiter.signal_429(name) end)
+      high_gap = IemRateLimiter.current_interval_ms(name)
+      assert high_gap > 50
+
+      # Successes drive it back down
+      Enum.each(1..20, fn _ -> IemRateLimiter.signal_success(name) end)
+      low_gap = IemRateLimiter.current_interval_ms(name)
+
+      assert low_gap < high_gap
+      # But never below the configured base.
+      assert low_gap >= 50
+    end
+
+    test "max_interval_ms is a hard ceiling" do
+      name = start_adaptive(100, 400)
+      Enum.each(1..50, fn _ -> IemRateLimiter.signal_429(name) end)
+      assert IemRateLimiter.current_interval_ms(name) == 400
+    end
+  end
 end

test/mix/tasks/weather_rebatch_asos_test.exs

@@ -0,0 +1,89 @@
+defmodule Mix.Tasks.Weather.RebatchAsosTest do
+  use Microwaveprop.DataCase, async: false
+
+  alias Microwaveprop.Weather.IemClient
+  alias Microwaveprop.Workers.WeatherFetchWorker
+  alias Mix.Tasks.Weather.RebatchAsos
+
+  setup do
+    # Oban runs inline in tests, so the batch job the task inserts
+    # actually executes. Stub IEM so that execution is a quiet no-op
+    # (stations don't exist, so the batch handler just logs "missing"
+    # and returns :ok).
+    Req.Test.stub(IemClient, fn conn ->
+      Req.Test.text(conn, "#DEBUG\nstation,valid,tmpf\n")
+    end)
+
+    :ok
+  end
+
+  # Oban runs `testing: :inline` in this project, so Oban.insert! would
+  # execute the job immediately. We stash raw Oban.Job rows so we can
+  # assert on the queue state the rebatch task would see in prod.
+  defp insert_asos_job(station_id, station_code, start_dt, end_dt) do
+    changeset =
+      WeatherFetchWorker.new(%{
+        "fetch_type" => "asos",
+        "station_id" => station_id,
+        "station_code" => station_code,
+        "start_dt" => start_dt,
+        "end_dt" => end_dt
+      })
+
+    # Force state=available so the task picks it up; Oban.Job defaults
+    # to :available already but we set it explicitly for clarity.
+    changeset
+    |> Ecto.Changeset.put_change(:state, "available")
+    |> Repo.insert!()
+  end
+
+  defp pending_count_by_type do
+    import Ecto.Query
+
+    from(j in Oban.Job,
+      where: j.state in ["available", "scheduled", "retryable"],
+      select: {fragment("?->>'fetch_type'", j.args), count(j.id)},
+      group_by: fragment("?->>'fetch_type'", j.args)
+    )
+    |> Repo.all()
+    |> Map.new()
+  end
+
+  test "collapses N single-station jobs sharing a window into one batch" do
+    insert_asos_job(Ecto.UUID.generate(), "KDFW", "2026-03-28T16:00:00Z", "2026-03-28T20:00:00Z")
+    insert_asos_job(Ecto.UUID.generate(), "KFTW", "2026-03-28T16:00:00Z", "2026-03-28T20:00:00Z")
+    insert_asos_job(Ecto.UUID.generate(), "KAFW", "2026-03-28T16:00:00Z", "2026-03-28T20:00:00Z")
+
+    output = ExUnit.CaptureIO.capture_io(fn -> RebatchAsos.run([]) end)
+
+    assert output =~ "3 pending `asos` jobs across 1 time windows"
+    assert output =~ "inserted 1 batch jobs"
+    assert output =~ "cancelled 3 single-station jobs"
+  end
+
+  test "groups per (start_dt, end_dt) window" do
+    insert_asos_job(Ecto.UUID.generate(), "KDFW", "2026-03-28T16:00:00Z", "2026-03-28T20:00:00Z")
+    insert_asos_job(Ecto.UUID.generate(), "KFTW", "2026-03-28T16:00:00Z", "2026-03-28T20:00:00Z")
+    insert_asos_job(Ecto.UUID.generate(), "KAUS", "2026-03-29T10:00:00Z", "2026-03-29T14:00:00Z")
+
+    output = ExUnit.CaptureIO.capture_io(fn -> RebatchAsos.run([]) end)
+
+    assert output =~ "3 pending `asos` jobs across 2 time windows"
+    assert output =~ "inserted 2 batch jobs"
+  end
+
+  test "dry-run does not mutate the queue" do
+    insert_asos_job(Ecto.UUID.generate(), "KDFW", "2026-03-28T16:00:00Z", "2026-03-28T20:00:00Z")
+
+    ExUnit.CaptureIO.capture_io(fn -> RebatchAsos.run(["--dry-run"]) end)
+
+    counts = pending_count_by_type()
+    assert counts["asos"] == 1
+    assert counts["asos_batch"] in [nil, 0]
+  end
+
+  test "no pending asos jobs — short-circuits" do
+    output = ExUnit.CaptureIO.capture_io(fn -> RebatchAsos.run([]) end)
+    assert output =~ "Nothing to do"
+  end
+end