From f1846c0a53576e183a78760c3acbae21c4e1f70f Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 19 Apr 2026 14:56:48 -0500 Subject: [PATCH] perf: reduce per-pod RSS and HRRR chain wall time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Telemetry showed the application-master process holding ~830 MiB of terms from warm_grid_cache_from_latest_profile — the data lives in the app master's heap and never GCs because the process is idle. Running it in a Task.start lets the terms die with the task. Mark GridCache, MrmsCache, NexradCache, and ScoreCache ETS tables :compressed. The scored-band-map and HRRR grid data are map-heavy; compression trims hundreds of MiB at a few percent CPU cost. Memoise HRRR .idx responses in Microwaveprop.Cache. Published idx files are immutable for a model run, but the hourly chain re-fetches the same URL dozens of times across forecast hours. Cuts ~10s per repeat out of hrrr_fetch_idx. Force a garbage collect at the end of HrrrFetchWorker.perform to reclaim the refc binary heap held from GRIB2 ranges before the Oban producer hands the process its next job. --- lib/microwaveprop/application.ex | 7 ++- lib/microwaveprop/propagation/score_cache.ex | 2 +- lib/microwaveprop/weather/grid_cache.ex | 2 +- lib/microwaveprop/weather/hrrr_client.ex | 28 ++++++----- lib/microwaveprop/weather/mrms_cache.ex | 2 +- lib/microwaveprop/weather/nexrad_cache.ex | 2 +- .../workers/hrrr_fetch_worker.ex | 21 ++++++--- .../weather/hrrr_client_test.exs | 46 +++++++++++++++++++ 8 files changed, 87 insertions(+), 23 deletions(-) diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 6421beda..7e7d7eb6 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -43,7 +43,12 @@ defmodule Microwaveprop.Application do # Warm GridCache from the latest persisted ProfilesFile so /weather # has data immediately after a pod restart. Runs after the GridCache # GenServer is up (it's in `children` above). - Microwaveprop.Weather.warm_grid_cache_from_latest_profile() + # + # Must run in a short-lived process, not inline in the application + # master: when executed here the loaded rows live on the app master's + # heap and never GC (it's idle after boot), pinning ~800 MiB for the + # lifetime of the pod. + Task.start(fn -> Microwaveprop.Weather.warm_grid_cache_from_latest_profile() end) # Load ML model in dev/test only (Nx/Axon not compiled for prod) if Application.get_env(:microwaveprop, :load_ml_model, false) do diff --git a/lib/microwaveprop/propagation/score_cache.ex b/lib/microwaveprop/propagation/score_cache.ex index a91c8a40..9e0edc4d 100644 --- a/lib/microwaveprop/propagation/score_cache.ex +++ b/lib/microwaveprop/propagation/score_cache.ex @@ -122,7 +122,7 @@ defmodule Microwaveprop.Propagation.ScoreCache do @impl true def init(_opts) do - :ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) + :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true]) PubSub.subscribe(@pubsub, @topic) {:ok, %{}} end diff --git a/lib/microwaveprop/weather/grid_cache.ex b/lib/microwaveprop/weather/grid_cache.ex index 8b71ae8f..87d1d5dd 100644 --- a/lib/microwaveprop/weather/grid_cache.ex +++ b/lib/microwaveprop/weather/grid_cache.ex @@ -121,7 +121,7 @@ defmodule Microwaveprop.Weather.GridCache do @impl true def init(_opts) do - :ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) + :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true]) :ets.new(@lock_table, [:set, :named_table, :public]) PubSub.subscribe(@pubsub, @topic) {:ok, %{}} diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index 62573ab7..75fbd02e 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -413,21 +413,27 @@ defmodule Microwaveprop.Weather.HrrrClient do end end + # A published HRRR idx file is immutable for the lifetime of its + # model run (60 min), but the hourly propagation chain fetches the + # same `.idx` URL dozens of times across forecast-hour steps and + # products. Memoising in `Microwaveprop.Cache` removes 100+ seconds + # per chain from the `hrrr_fetch_idx` span. + @idx_cache_ttl_ms to_timeout(hour: 1) + defp fetch_idx(url) do - Microwaveprop.Instrument.span([:hrrr, :fetch_idx], %{url: url}, fn -> - case Req.get(url, req_options()) do - {:ok, %{status: 200, body: body}} -> - {:ok, body} - - {:ok, %{status: status}} -> - {:error, "HRRR idx HTTP #{status}"} - - {:error, reason} -> - {:error, reason} - end + Microwaveprop.Cache.fetch_or_store({:hrrr_idx, url}, @idx_cache_ttl_ms, fn -> + Microwaveprop.Instrument.span([:hrrr, :fetch_idx], %{url: url}, fn -> do_fetch_idx(url) end) end) end + defp do_fetch_idx(url) do + case Req.get(url, req_options()) do + {:ok, %{status: 200, body: body}} -> {:ok, body} + {:ok, %{status: status}} -> {:error, "HRRR idx HTTP #{status}"} + {:error, reason} -> {:error, reason} + end + end + @doc """ Downloads the given byte ranges from a GRIB2 URL and concatenates them in offset order. Merges adjacent/overlapping ranges and runs diff --git a/lib/microwaveprop/weather/mrms_cache.ex b/lib/microwaveprop/weather/mrms_cache.ex index cbf8865e..a144c737 100644 --- a/lib/microwaveprop/weather/mrms_cache.ex +++ b/lib/microwaveprop/weather/mrms_cache.ex @@ -65,7 +65,7 @@ defmodule Microwaveprop.Weather.MrmsCache do @impl true def init(_opts) do - :ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) + :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true]) PubSub.subscribe(@pubsub, @topic) {:ok, %{}} end diff --git a/lib/microwaveprop/weather/nexrad_cache.ex b/lib/microwaveprop/weather/nexrad_cache.ex index 6f855246..5ebd440b 100644 --- a/lib/microwaveprop/weather/nexrad_cache.ex +++ b/lib/microwaveprop/weather/nexrad_cache.ex @@ -73,7 +73,7 @@ defmodule Microwaveprop.Weather.NexradCache do @impl true def init(_opts) do - :ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) + :ets.new(@table, [:set, :named_table, :public, :compressed, read_concurrency: true]) {:ok, %{}} end end diff --git a/lib/microwaveprop/workers/hrrr_fetch_worker.ex b/lib/microwaveprop/workers/hrrr_fetch_worker.ex index cc907912..83de03d7 100644 --- a/lib/microwaveprop/workers/hrrr_fetch_worker.ex +++ b/lib/microwaveprop/workers/hrrr_fetch_worker.ex @@ -27,13 +27,20 @@ defmodule Microwaveprop.Workers.HrrrFetchWorker do |> Enum.map(fn %{"lat" => lat, "lon" => lon} -> {lat, lon} end) |> Enum.reject(fn {lat, lon} -> Weather.has_hrrr_profile?(lat, lon, valid_time) end) - if point_tuples == [] do - Logger.info("HRRR batch: all #{length(points)} points already exist for #{valid_time_str}") - :ok - else - Logger.info("HRRR batch: fetching #{length(point_tuples)} points for #{valid_time_str}") - fetch_and_store_batch(point_tuples, valid_time, valid_time_str) - end + result = + if point_tuples == [] do + Logger.info("HRRR batch: all #{length(points)} points already exist for #{valid_time_str}") + :ok + else + Logger.info("HRRR batch: fetching #{length(point_tuples)} points for #{valid_time_str}") + fetch_and_store_batch(point_tuples, valid_time, valid_time_str) + end + + # Reclaim the refc binary heap held from GRIB2 ranges before the + # Oban producer hands this process its next job. Without this each + # worker steady-states at 100+ MiB of retained HRRR payload. + :erlang.garbage_collect(self()) + result end def perform(%Oban.Job{args: %{"lat" => raw_lat, "lon" => raw_lon, "valid_time" => valid_time_str}}) do diff --git a/test/microwaveprop/weather/hrrr_client_test.exs b/test/microwaveprop/weather/hrrr_client_test.exs index b3d51c8d..15057bec 100644 --- a/test/microwaveprop/weather/hrrr_client_test.exs +++ b/test/microwaveprop/weather/hrrr_client_test.exs @@ -171,6 +171,52 @@ defmodule Microwaveprop.Weather.HrrrClientTest do assert is_float(profile.surface_temp_c) assert is_struct(profile.run_time, DateTime) end + + test "caches idx responses across calls to the same HRRR run" do + # `.idx` files are static once a model run is published. The 114s + # `hrrr_fetch_grid` span in prod spends ~10s re-fetching the same idx + # URL across forecast hours / products. Cache eliminates that per-run. + grib_data = File.read!("test/fixtures/grib2/hrrr_tmp_2m.grib2") + grib_size = byte_size(grib_data) + + idx_text = """ + 1:0:d=2026032818:TMP:2 m above ground:anl: + 2:#{grib_size}:d=2026032818:DPT:2 m above ground:anl: + 3:#{grib_size * 2}:d=2026032818:PRES:surface:anl: + """ + + Microwaveprop.Cache.invalidate( + {:hrrr_idx, "https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t18z.wrfsfcf00.grib2.idx"} + ) + + {:ok, counter} = Agent.start_link(fn -> %{idx: 0, grib: 0} end) + + Req.Test.stub(HrrrClient, fn conn -> + if String.ends_with?(conn.request_path, ".idx") do + Agent.update(counter, &Map.update!(&1, :idx, fn n -> n + 1 end)) + Plug.Conn.send_resp(conn, 200, idx_text) + else + Agent.update(counter, &Map.update!(&1, :grib, fn n -> n + 1 end)) + [range] = Plug.Conn.get_req_header(conn, "range") + + if String.contains?(range, ",") do + Plug.Conn.send_resp(conn, 200, "full file") + else + Plug.Conn.send_resp(conn, 206, grib_data) + end + end + end) + + valid_time = ~U[2026-03-28 18:00:00Z] + + assert {:ok, _} = HrrrClient.fetch_profile(32.90, -97.04, valid_time) + assert {:ok, _} = HrrrClient.fetch_profile(32.90, -97.04, valid_time) + + %{idx: idx_count, grib: grib_count} = Agent.get(counter, & &1) + + assert idx_count == 1, "idx should be cached — expected 1 fetch, got #{idx_count}" + assert grib_count > 0, "grib fetches should still happen" + end end describe "build_profile/1" do