diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index e0124f86..a1894339 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -940,13 +940,63 @@ defmodule Microwaveprop.Weather do # authoritative, so don't fall back to a full ProfilesFile decode # in that case. if ScalarFile.exists?(valid_time) do - ScalarFile.read_bounds(valid_time, bounds) + read_via_scalar(valid_time, bounds) else read_and_derive_grid(valid_time, bounds) end end end + defp read_via_scalar(valid_time, bounds) do + fill_grid_cache_from_scalar(valid_time) + + case GridCache.fetch_bounds(valid_time, bounds) do + {:ok, rows} -> rows + :miss -> ScalarFile.read_bounds(valid_time, bounds) + end + end + + # Cap on cached valid_times. HRRR runs hourly with 18 forecast hours, so + # 24 covers analysis + 18 forecasts plus a few late stragglers from the + # previous run while a user is scrubbing. + @grid_cache_valid_time_cap 24 + + # Hydrate GridCache from the on-disk ScalarFile so concurrent viewport + # reads for the same valid_time don't each gunzip+msgpack-decode the + # same chunk files. Only the caller that wins `claim_fill` does the + # work; losers wait briefly for the ETS write to land. + defp fill_grid_cache_from_scalar(valid_time) do + lock_key = {:scalar_to_grid_cache, valid_time} + + if GridCache.claim_fill(lock_key) do + try do + rows = ScalarFile.read_bounds(valid_time, nil) + + if rows != [] do + GridCache.put(valid_time, rows) + _ = GridCache.prune_keep_latest(@grid_cache_valid_time_cap) + end + after + GridCache.release_fill(lock_key) + end + else + wait_for_grid_cache(valid_time) + end + end + + defp wait_for_grid_cache(valid_time) do + Enum.reduce_while(1..50, :miss, fn _, _ -> + case GridCache.fetch(valid_time) do + {:ok, _} -> + {:halt, :ok} + + :miss -> + Process.sleep(20) + {:cont, :miss} + end + end) + end + # Cold path: ScalarFile didn't have anything for this valid_time, so # decode the raw ProfilesFile and derive the requested viewport. Kicks # off a background materialization of the full scalar file so the diff --git a/lib/microwaveprop/weather/grid_cache.ex b/lib/microwaveprop/weather/grid_cache.ex index 62ad9df7..32215e79 100644 --- a/lib/microwaveprop/weather/grid_cache.ex +++ b/lib/microwaveprop/weather/grid_cache.ex @@ -119,6 +119,25 @@ defmodule Microwaveprop.Weather.GridCache do :ets.select_delete(@table, match_spec) end + @doc """ + Keep only the `keep` most-recent valid_times, dropping the rest. Returns + the number of entries removed. Used to bound memory when forecast hours + are cached on demand. + """ + @spec prune_keep_latest(non_neg_integer()) :: non_neg_integer() + def prune_keep_latest(keep) when is_integer(keep) and keep >= 0 do + case :ets.select(@table, [{{:"$1", :_}, [], [:"$1"]}]) do + [] -> + 0 + + times -> + sorted = Enum.sort(times, {:desc, DateTime}) + to_drop = Enum.drop(sorted, keep) + Enum.each(to_drop, &:ets.delete(@table, &1)) + length(to_drop) + end + end + @spec clear() :: :ok def clear do GenServer.call(__MODULE__, :clear) diff --git a/test/microwaveprop/weather/grid_cache_test.exs b/test/microwaveprop/weather/grid_cache_test.exs index eb7220e0..937eb9b8 100644 --- a/test/microwaveprop/weather/grid_cache_test.exs +++ b/test/microwaveprop/weather/grid_cache_test.exs @@ -195,4 +195,30 @@ defmodule Microwaveprop.Weather.GridCacheTest do assert GridCache.latest_valid_time() == nil end end + + describe "prune_keep_latest/1" do + test "drops oldest valid_times beyond the cap" do + GridCache.put(~U[2026-04-12 10:00:00Z], [%{lat: 32.0, lon: -97.0}]) + GridCache.put(~U[2026-04-12 11:00:00Z], [%{lat: 32.0, lon: -97.0}]) + GridCache.put(~U[2026-04-12 12:00:00Z], [%{lat: 32.0, lon: -97.0}]) + GridCache.put(~U[2026-04-12 13:00:00Z], [%{lat: 32.0, lon: -97.0}]) + + removed = GridCache.prune_keep_latest(2) + assert removed == 2 + + assert GridCache.fetch(~U[2026-04-12 10:00:00Z]) == :miss + assert GridCache.fetch(~U[2026-04-12 11:00:00Z]) == :miss + assert {:ok, _} = GridCache.fetch(~U[2026-04-12 12:00:00Z]) + assert {:ok, _} = GridCache.fetch(~U[2026-04-12 13:00:00Z]) + end + + test "is a no-op when the cap is not exceeded" do + GridCache.put(~U[2026-04-12 10:00:00Z], [%{lat: 32.0, lon: -97.0}]) + GridCache.put(~U[2026-04-12 11:00:00Z], [%{lat: 32.0, lon: -97.0}]) + + assert GridCache.prune_keep_latest(5) == 0 + assert {:ok, _} = GridCache.fetch(~U[2026-04-12 10:00:00Z]) + assert {:ok, _} = GridCache.fetch(~U[2026-04-12 11:00:00Z]) + end + end end diff --git a/test/microwaveprop/weather_grid_test.exs b/test/microwaveprop/weather_grid_test.exs index 144f1bf9..3cb3279c 100644 --- a/test/microwaveprop/weather_grid_test.exs +++ b/test/microwaveprop/weather_grid_test.exs @@ -647,6 +647,61 @@ defmodule Microwaveprop.WeatherGridTest do end end + describe "weather_grid_at/2 → GridCache from ScalarFile" do + alias Microwaveprop.Weather.ScalarFile + + setup do + # Tests below assert end state in ETS. + GridCache.clear() + :ok + end + + test "populates GridCache from ScalarFile on miss so the next read skips file IO" do + vt = ~U[2026-04-28 15:00:00Z] + + ProfilesFile.write!(vt, %{ + {32.125, -97.375} => %{ + surface_temp_c: 25.0, + surface_dewpoint_c: 18.0, + surface_pressure_mb: 1013.0, + hpbl_m: 500.0, + pwat_mm: 30.0, + profile: [] + }, + {32.250, -97.375} => %{ + surface_temp_c: 26.0, + surface_dewpoint_c: 19.0, + surface_pressure_mb: 1012.0, + hpbl_m: 600.0, + pwat_mm: 32.0, + profile: [] + } + }) + + # Materialize the ScalarFile so it's the source of truth. + :ok = Weather.materialize_scalar_file(vt) + assert ScalarFile.exists?(vt) + assert GridCache.fetch(vt) == :miss + + # First call: miss → reads ScalarFile → populates GridCache. + bounds = %{"south" => 32.0, "north" => 33.0, "west" => -98.0, "east" => -97.0} + rows = Weather.weather_grid_at(vt, bounds) + assert length(rows) == 2 + + # GridCache is now warm — confirms cache populate happened. + assert {:ok, cached_rows} = GridCache.fetch(vt) + assert length(cached_rows) == 2 + + # Wipe the on-disk ScalarFile. The next call should still return data + # because GridCache is the hot path. + _ = File.rm_rf(ScalarFile.dir_for(vt)) + refute ScalarFile.exists?(vt) + + rows_again = Weather.weather_grid_at(vt, bounds) + assert length(rows_again) == 2 + end + end + # Merges `attrs` (a single HRRR grid cell) into the ProfilesFile for # `attrs.valid_time`. Re-reads any existing file so multiple calls for # the same valid_time accumulate cells, mirroring how the Rust worker