diff --git a/.gitignore b/.gitignore index d65a2cdf..4e67ad03 100644 --- a/.gitignore +++ b/.gitignore @@ -35,6 +35,10 @@ microwaveprop-*.tar .env .env.* +# Kubernetes secret manifests (contain plaintext credentials) +/k8s/secret.yaml +/k8s/*-secret.yaml + # Trained ML model weights — tracked in git for Docker builds # GRIB2 test fixtures (large binary files, downloaded on-demand) diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index a9813bc8..fb5b5c73 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -374,8 +374,41 @@ defmodule Microwaveprop.Weather do end end + @doc """ + Cache-only read for the /weather LiveView mount hot path. Returns whatever + is in `GridCache` for the latest valid_time and fires a deduped background + fill task on a miss instead of blocking. The async task broadcasts + `weather:updated` when done, triggering every connected LiveView to refresh. + + Callers that genuinely need synchronous data (tests, scripts) should use + `load_weather_grid/1` instead. + """ @spec latest_weather_grid(map()) :: [map()] def latest_weather_grid(bounds) do + case latest_grid_valid_time() do + nil -> + [] + + latest_vt -> + case GridCache.fetch_bounds(latest_vt, bounds) do + {:ok, rows} -> + rows + + :miss -> + kickoff_async_grid_fill(latest_vt) + [] + end + end + end + + @doc """ + Synchronous cache-or-DB read. Blocks for several seconds on a cold cache + because `load_weather_grid_from_db/1` parses 176k rows with JSONB columns. + Used by tests and by `weather_point_detail/3` fallbacks. LiveView callers + should prefer `latest_weather_grid/1`. + """ + @spec load_weather_grid(map()) :: [map()] + def load_weather_grid(bounds) do case latest_grid_valid_time() do nil -> [] @@ -393,6 +426,40 @@ defmodule Microwaveprop.Weather do end end + defp filter_weather_bounds(rows, nil), do: rows + + defp filter_weather_bounds(rows, %{"south" => s, "north" => n, "west" => w, "east" => e}) do + Enum.filter(rows, fn %{lat: lat, lon: lon} -> + lat >= s and lat <= n and lon >= w and lon <= e + end) + end + + defp kickoff_async_grid_fill(valid_time) do + if GridCache.claim_fill(valid_time) do + Task.start(fn -> + try do + Logger.info("Weather.grid_cache async fill starting for #{valid_time}") + warm_grid_cache_and_broadcast(valid_time) + + Phoenix.PubSub.broadcast( + Microwaveprop.PubSub, + "weather:updated", + {:weather_updated, valid_time} + ) + + Logger.info("Weather.grid_cache async fill complete for #{valid_time}") + rescue + e -> + Logger.error("Weather.grid_cache async fill failed: #{inspect(e)}") + after + GridCache.release_fill(valid_time) + end + end) + end + + :ok + end + defp load_weather_grid_from_db(latest_vt) do from(h in HrrrProfile, where: h.valid_time == ^latest_vt and h.is_grid_point == true, @@ -418,14 +485,6 @@ defmodule Microwaveprop.Weather do |> Enum.map(&derive_and_clean/1) end - defp filter_weather_bounds(rows, nil), do: rows - - defp filter_weather_bounds(rows, %{"south" => s, "north" => n, "west" => w, "east" => e}) do - Enum.filter(rows, fn %{lat: lat, lon: lon} -> - lat >= s and lat <= n and lon >= w and lon <= e - end) - end - @doc """ Eagerly populate the `GridCache` with the full CONUS weather grid for `valid_time` and broadcast it to every node in the cluster. Called from diff --git a/lib/microwaveprop/weather/grid_cache.ex b/lib/microwaveprop/weather/grid_cache.ex index 16aaa6b5..8b71ae8f 100644 --- a/lib/microwaveprop/weather/grid_cache.ex +++ b/lib/microwaveprop/weather/grid_cache.ex @@ -19,6 +19,7 @@ defmodule Microwaveprop.Weather.GridCache do alias Phoenix.PubSub @table :weather_grid_cache + @lock_table :weather_grid_fill_locks @topic "weather:cache" @pubsub Microwaveprop.PubSub @@ -94,6 +95,25 @@ defmodule Microwaveprop.Weather.GridCache do :ok end + @doc """ + Atomically claim the right to fill the cache for `valid_time`. Returns + `true` if this caller won the claim and should run the fill; `false` if + another caller is already filling. Prevents N concurrent /weather mounts + after a pod restart from each firing the 15-second `load_weather_grid_from_db` + query and starving the Postgres connection pool. + """ + @spec claim_fill(DateTime.t()) :: boolean() + def claim_fill(valid_time) do + :ets.insert_new(@lock_table, {valid_time, :in_progress}) + end + + @doc "Release a fill lock claimed via `claim_fill/1`." + @spec release_fill(DateTime.t()) :: :ok + def release_fill(valid_time) do + :ets.delete(@lock_table, valid_time) + :ok + end + @spec sync() :: :ok def sync do GenServer.call(__MODULE__, :sync) @@ -102,6 +122,7 @@ defmodule Microwaveprop.Weather.GridCache do @impl true def init(_opts) do :ets.new(@table, [:set, :named_table, :public, read_concurrency: true]) + :ets.new(@lock_table, [:set, :named_table, :public]) PubSub.subscribe(@pubsub, @topic) {:ok, %{}} end diff --git a/test/microwaveprop/weather_grid_test.exs b/test/microwaveprop/weather_grid_test.exs index c2ae1e80..47bb545b 100644 --- a/test/microwaveprop/weather_grid_test.exs +++ b/test/microwaveprop/weather_grid_test.exs @@ -1,10 +1,19 @@ defmodule Microwaveprop.WeatherGridTest do - use Microwaveprop.DataCase, async: true + use Microwaveprop.DataCase, async: false alias Microwaveprop.Repo alias Microwaveprop.Weather + alias Microwaveprop.Weather.GridCache - describe "latest_weather_grid/1" do + setup do + # GridCache ETS state leaks between tests (it's a named table outside the + # test sandbox). Clear it so each test gets a fresh cache and the DB load + # path actually runs. + GridCache.clear() + :ok + end + + describe "load_weather_grid/1" do test "returns grid data for latest valid_time within bounds" do now = DateTime.truncate(DateTime.utc_now(), :second) @@ -35,7 +44,7 @@ defmodule Microwaveprop.WeatherGridTest do }) bounds = %{"south" => 32.0, "north" => 33.0, "west" => -98.0, "east" => -97.0} - result = Weather.latest_weather_grid(bounds) + result = Weather.load_weather_grid(bounds) assert length(result) == 2 point = Enum.find(result, &(&1.lat == 32.125)) @@ -49,7 +58,7 @@ defmodule Microwaveprop.WeatherGridTest do test "returns empty list when no data exists" do bounds = %{"south" => 32.0, "north" => 33.0, "west" => -98.0, "east" => -97.0} - assert Weather.latest_weather_grid(bounds) == [] + assert Weather.load_weather_grid(bounds) == [] end test "only returns points on 0.125 grid" do @@ -78,7 +87,7 @@ defmodule Microwaveprop.WeatherGridTest do }) bounds = %{"south" => 32.0, "north" => 33.0, "west" => -98.0, "east" => -97.0} - result = Weather.latest_weather_grid(bounds) + result = Weather.load_weather_grid(bounds) assert length(result) == 1 assert hd(result).lat == 32.125 @@ -111,7 +120,7 @@ defmodule Microwaveprop.WeatherGridTest do }) bounds = %{"south" => 32.0, "north" => 33.0, "west" => -98.0, "east" => -97.0} - result = Weather.latest_weather_grid(bounds) + result = Weather.load_weather_grid(bounds) assert length(result) == 1 assert hd(result).temperature == 28.0 @@ -150,7 +159,7 @@ defmodule Microwaveprop.WeatherGridTest do }) bounds = %{"south" => 32.0, "north" => 33.0, "west" => -98.0, "east" => -97.0} - [point] = Weather.latest_weather_grid(bounds) + [point] = Weather.load_weather_grid(bounds) # Derived fields should be present assert is_float(point.surface_rh)