diff --git a/lib/microwaveprop/weather/iem_client.ex b/lib/microwaveprop/weather/iem_client.ex index a398dad1..44846de7 100644 --- a/lib/microwaveprop/weather/iem_client.ex +++ b/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 - {:ok, %{status: 200, body: body}} -> - {:ok, parse_network_json(body)} - - {:ok, %{status: status}} -> - {:error, "IEM network HTTP #{status}"} - - {:error, reason} -> - {:error, reason} - end + url + |> Req.get(req_options()) + |> handle_response(&parse_network_json/1, "IEM network") 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 - {:ok, %{status: 200, body: body}} -> - {:ok, parse_asos_csv(body)} - - {:ok, %{status: status}} -> - {:error, "IEM ASOS HTTP #{status}"} - - {:error, reason} -> - {:error, reason} - end + url + |> Req.get(req_options()) + |> handle_response(&parse_asos_csv/1, "IEM ASOS") 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 - {:ok, %{status: 200, body: body}} -> - rows = parse_asos_csv(body) - grouped = Enum.group_by(rows, & &1.station_code) - # Ensure every requested code has at least an empty list in the - # map so downstream iteration doesn't miss "no data" cases. - filled = Enum.reduce(codes, grouped, &Map.put_new(&2, &1, [])) - {:ok, filled} - - {:ok, %{status: status}} -> - {:error, "IEM ASOS HTTP #{status}"} - - {:error, reason} -> - {:error, reason} + parse = fn body -> + rows = parse_asos_csv(body) + grouped = Enum.group_by(rows, & &1.station_code) + # Ensure every requested code has at least an empty list in the + # map so downstream iteration doesn't miss "no data" cases. + Enum.reduce(codes, grouped, &Map.put_new(&2, &1, [])) 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 - {:ok, %{status: 200, body: body}} -> - {:ok, parse_raob_json(body)} - - {:ok, %{status: status}} -> - {:error, "IEM RAOB HTTP #{status}"} - - {:error, reason} -> - {:error, reason} - end + url + |> Req.get(req_options()) + |> handle_response(&parse_raob_json/1, "IEM RAOB") 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 - {:ok, %{status: 200, body: body}} -> - {:ok, parse_iemre_json(body)} - - {:ok, %{status: status}} -> - {:error, "IEM IEMRE HTTP #{status}"} - - {:error, reason} -> - {:error, reason} - end + url + |> Req.get(req_options()) + |> handle_response(&parse_iemre_json/1, "IEM IEMRE") 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, []) diff --git a/lib/microwaveprop/weather/iem_rate_limiter.ex b/lib/microwaveprop/weather/iem_rate_limiter.ex index 8c4cf09c..88ef1ec4 100644 --- a/lib/microwaveprop/weather/iem_rate_limiter.ex +++ b/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 - 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. + IEM rate-limits 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. + + **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) - {:ok, %{interval_ms: interval_ms, next_ms: System.system_time(:millisecond)}} + base = Keyword.get(opts, :interval_ms, @default_interval_ms) + 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 diff --git a/lib/mix/tasks/weather_rebatch_asos.ex b/lib/mix/tasks/weather_rebatch_asos.ex new file mode 100644 index 00000000..a42b8c1f --- /dev/null +++ b/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 diff --git a/test/microwaveprop/weather/iem_rate_limiter_test.exs b/test/microwaveprop/weather/iem_rate_limiter_test.exs index e18c6508..b17652fb 100644 --- a/test/microwaveprop/weather/iem_rate_limiter_test.exs +++ b/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 diff --git a/test/mix/tasks/weather_rebatch_asos_test.exs b/test/mix/tasks/weather_rebatch_asos_test.exs new file mode 100644 index 00000000..c7eb931e --- /dev/null +++ b/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