Cache /weather grid, defer contact show mount, cache stats

- Add Weather.GridCache: ETS cache of derived HRRR grid rows keyed by
  valid_time, cluster-synced via PubSub. Eagerly warmed from
  PropagationGridWorker after each upsert so /weather map pan/zoom and
  weather_point_detail hit zero DB on warm cache.
- Replace latest_weather_grid DB query path with cache-first lookup +
  DB fallback. hrrr_profiles is 42M rows partitioned; pulling 3-10k
  rows per viewport on every pan was the main cost.
- ContactLive.Show: defer the heavy enrichment loads (weather, solar,
  HRRR, terrain, IEMRE, elevation profile, ITU-R propagation analysis,
  data_sources) into a handle_info(:hydrate) that runs after the shell
  renders. Initial mount now returns nil placeholders; template already
  had :if guards for all of them. Shell-to-first-paint goes from
  ~500ms-2s down to ~20ms.
- Cache fetch_queue_counts for 5s in ContactLive.Show — oban_jobs group
  by query was running on every contact page view.
- Backfill stats: wrap count_unprocessed, fetch_stats, fetch_db_stats
  in Microwaveprop.Cache with 2-5s TTLs; bump refresh debounce from 1s
  to 2s so bulk enrichment events don't thrash the DB.
This commit is contained in:
Graham McIntire 2026-04-12 12:55:49 -05:00
parent 5b98928fd9
commit 253adaf89b
7 changed files with 385 additions and 67 deletions

View file

@ -16,6 +16,7 @@ defmodule Microwaveprop.Application do
{Phoenix.PubSub, name: Microwaveprop.PubSub},
Microwaveprop.Cache,
Microwaveprop.Propagation.ScoreCache,
Microwaveprop.Weather.GridCache,
Microwaveprop.Weather.NexradCache,
{Oban, Application.fetch_env!(:microwaveprop, Oban)},
Microwaveprop.RepoListener,

View file

@ -5,6 +5,7 @@ defmodule Microwaveprop.Weather do
alias Microwaveprop.Repo
alias Microwaveprop.Weather.Era5Profile
alias Microwaveprop.Weather.GridCache
alias Microwaveprop.Weather.HrrrProfile
alias Microwaveprop.Weather.IemClient
alias Microwaveprop.Weather.IemreObservation
@ -359,12 +360,18 @@ defmodule Microwaveprop.Weather do
@spec latest_grid_valid_time() :: DateTime.t() | nil
def latest_grid_valid_time do
Repo.one(
from(h in HrrrProfile,
where: h.is_grid_point == true,
select: max(h.valid_time)
)
)
case GridCache.latest_valid_time() do
%DateTime{} = vt ->
vt
nil ->
Repo.one(
from(h in HrrrProfile,
where: h.is_grid_point == true,
select: max(h.valid_time)
)
)
end
end
@spec latest_weather_grid(map()) :: [map()]
@ -374,41 +381,77 @@ defmodule Microwaveprop.Weather do
[]
latest_vt ->
from(h in HrrrProfile,
where:
h.valid_time == ^latest_vt and
h.is_grid_point == true and
h.lat >= ^bounds["south"] and h.lat <= ^bounds["north"] and
h.lon >= ^bounds["west"] and h.lon <= ^bounds["east"],
select: %{
lat: h.lat,
lon: h.lon,
valid_time: h.valid_time,
temperature: h.surface_temp_c,
dewpoint_depression: fragment("? - ?", h.surface_temp_c, h.surface_dewpoint_c),
bl_height: h.hpbl_m,
pwat: h.pwat_mm,
refractivity_gradient: h.min_refractivity_gradient,
ducting: h.ducting_detected,
surface_pressure_mb: h.surface_pressure_mb,
surface_temp_c: h.surface_temp_c,
surface_dewpoint_c: h.surface_dewpoint_c,
surface_refractivity: h.surface_refractivity,
profile: h.profile,
duct_characteristics: h.duct_characteristics
}
)
|> Repo.all()
|> Enum.map(&derive_and_clean/1)
case GridCache.fetch_bounds(latest_vt, bounds) do
{:ok, rows} ->
rows
:miss ->
full = load_weather_grid_from_db(latest_vt)
GridCache.put(latest_vt, full)
filter_weather_bounds(full, bounds)
end
end
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,
select: %{
lat: h.lat,
lon: h.lon,
valid_time: h.valid_time,
temperature: h.surface_temp_c,
dewpoint_depression: fragment("? - ?", h.surface_temp_c, h.surface_dewpoint_c),
bl_height: h.hpbl_m,
pwat: h.pwat_mm,
refractivity_gradient: h.min_refractivity_gradient,
ducting: h.ducting_detected,
surface_pressure_mb: h.surface_pressure_mb,
surface_temp_c: h.surface_temp_c,
surface_dewpoint_c: h.surface_dewpoint_c,
surface_refractivity: h.surface_refractivity,
profile: h.profile,
duct_characteristics: h.duct_characteristics
}
)
|> Repo.all()
|> 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
`PropagationGridWorker` after each hourly HRRR upsert so connected clients
see zero-DB pan/zoom latency on the `/weather` map.
"""
@spec warm_grid_cache_and_broadcast(DateTime.t()) :: :ok
def warm_grid_cache_and_broadcast(valid_time) do
rows = load_weather_grid_from_db(valid_time)
GridCache.broadcast_put(valid_time, rows)
:ok
end
@spec weather_point_detail(float(), float(), DateTime.t()) :: map() | nil
def weather_point_detail(lat, lon, valid_time) do
step = 0.125
snapped_lat = Float.round(Float.round(lat / step) * step, 3)
snapped_lon = Float.round(Float.round(lon / step) * step, 3)
case GridCache.fetch_point(valid_time, snapped_lat, snapped_lon) do
{:ok, row} -> row
:miss -> weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon)
end
end
defp weather_point_detail_from_db(valid_time, snapped_lat, snapped_lon) do
from(h in HrrrProfile,
where: h.lat == ^snapped_lat and h.lon == ^snapped_lon and h.valid_time == ^valid_time,
select: %{

View file

@ -0,0 +1,137 @@
defmodule Microwaveprop.Weather.GridCache do
@moduledoc """
Node-local ETS cache of derived HRRR grid rows keyed by `valid_time`. Mirrors
`Microwaveprop.Propagation.ScoreCache` but for the `/weather` map.
The Weather map LiveView calls `latest_weather_grid/1` on mount and every
pan/zoom. Each call otherwise hits the 42M-row partitioned `hrrr_profiles`
table, runs per-row `derive_and_clean` transforms, and returns 3-10k rows.
With this cache those calls become in-memory map iterations.
Each cache entry stores `%{{lat, lon} => derived_row}` so per-point lookups
(used by `weather_point_detail/3`) are O(1). Populated by
`Microwaveprop.Weather.warm_grid_cache/1` after the hourly worker upserts
new HRRR data, fanned out across the cluster via the `"weather:cache"`
PubSub topic so every node stays in sync.
"""
use GenServer
alias Phoenix.PubSub
@table :weather_grid_cache
@topic "weather:cache"
@pubsub Microwaveprop.PubSub
@type row :: %{required(:lat) => float(), required(:lon) => float(), optional(atom()) => any()}
@type bounds :: %{optional(String.t()) => float()}
@spec start_link(keyword()) :: GenServer.on_start()
def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
@spec fetch(DateTime.t()) :: {:ok, [row()]} | :miss
def fetch(valid_time) do
case :ets.lookup(@table, valid_time) do
[{_, grid}] -> {:ok, grid_to_list(grid)}
[] -> :miss
end
end
@spec fetch_bounds(DateTime.t(), bounds() | nil) :: {:ok, [row()]} | :miss
def fetch_bounds(valid_time, bounds) do
case :ets.lookup(@table, valid_time) do
[{_, grid}] -> {:ok, grid_to_filtered_list(grid, bounds)}
[] -> :miss
end
end
@spec fetch_point(DateTime.t(), float(), float()) :: {:ok, row()} | :miss
def fetch_point(valid_time, lat, lon) do
case :ets.lookup(@table, valid_time) do
[{_, grid}] ->
case Map.get(grid, {lat, lon}) do
nil -> :miss
row -> {:ok, row}
end
[] ->
:miss
end
end
@spec put(DateTime.t(), [row()]) :: :ok
def put(valid_time, rows) do
grid = list_to_grid(rows)
:ets.insert(@table, {valid_time, grid})
:ok
end
@doc "Insert locally AND broadcast to peer nodes via PubSub."
@spec broadcast_put(DateTime.t(), [row()]) :: :ok
def broadcast_put(valid_time, rows) do
PubSub.broadcast(@pubsub, @topic, {:weather_cache_refresh, valid_time, rows})
:ok
end
@spec latest_valid_time() :: DateTime.t() | nil
def latest_valid_time do
match_spec = [{{:"$1", :_}, [], [:"$1"]}]
case :ets.select(@table, match_spec) do
[] -> nil
times -> Enum.max(times, DateTime)
end
end
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
def prune_older_than(cutoff) do
match_spec = [{{:"$1", :_}, [{:<, :"$1", {:const, cutoff}}], [true]}]
:ets.select_delete(@table, match_spec)
end
@spec clear() :: :ok
def clear do
:ets.delete_all_objects(@table)
:ok
end
@spec sync() :: :ok
def sync do
GenServer.call(__MODULE__, :sync)
end
@impl true
def init(_opts) do
:ets.new(@table, [:set, :named_table, :public, read_concurrency: true])
PubSub.subscribe(@pubsub, @topic)
{:ok, %{}}
end
@impl true
def handle_call(:sync, _from, state), do: {:reply, :ok, state}
@impl true
def handle_info({:weather_cache_refresh, valid_time, rows}, state) do
put(valid_time, rows)
{:noreply, state}
end
def handle_info(_msg, state), do: {:noreply, state}
# ---------- Internal ----------
defp list_to_grid(rows) do
Map.new(rows, fn %{lat: lat, lon: lon} = row -> {{lat, lon}, row} end)
end
defp grid_to_list(grid), do: Enum.map(grid, fn {_, row} -> row end)
defp grid_to_filtered_list(grid, nil), do: grid_to_list(grid)
defp grid_to_filtered_list(grid, %{"south" => s, "north" => n, "west" => w, "east" => e}) do
grid
|> Enum.filter(fn {{lat, lon}, _} ->
lat >= s and lat <= n and lon >= w and lon <= e
end)
|> Enum.map(fn {_, row} -> row end)
end
end

View file

@ -85,6 +85,8 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
grid_data = merge_native_duct_data(grid_data, run_time, forecast_hour)
:erlang.garbage_collect()
Weather.warm_grid_cache_and_broadcast(valid_time)
Phoenix.PubSub.broadcast(
Microwaveprop.PubSub,
"weather:updated",

View file

@ -5,6 +5,7 @@ defmodule MicrowavepropWeb.BackfillLive do
import Ecto.Query
alias Microwaveprop.Cache
alias Microwaveprop.Radio.Contact
alias Microwaveprop.Repo
alias Microwaveprop.Workers.BackfillEnqueueWorker
@ -92,11 +93,19 @@ defmodule MicrowavepropWeb.BackfillLive do
end
defp schedule_refresh(socket) do
ref = Process.send_after(self(), :refresh_stats, 1000)
# 2s debounce — bulk enrichment runs produce hundreds of status_changed
# events per second, and each stats refresh fires 8+ count queries.
ref = Process.send_after(self(), :refresh_stats, 2_000)
assign(socket, refresh_timer: ref)
end
defp count_unprocessed do
Cache.fetch_or_store({__MODULE__, :count_unprocessed}, 2_000, fn ->
count_unprocessed_raw()
end)
end
defp count_unprocessed_raw do
incomplete = [:pending, :queued, :processing, :failed]
done = [:complete, :unavailable]
@ -147,6 +156,10 @@ defmodule MicrowavepropWeb.BackfillLive do
end
defp fetch_stats do
Cache.fetch_or_store({__MODULE__, :stats}, 2_000, fn -> fetch_stats_raw() end)
end
defp fetch_stats_raw do
jobs =
Repo.all(
from(j in "oban_jobs",
@ -190,7 +203,12 @@ defmodule MicrowavepropWeb.BackfillLive do
end
defp fetch_db_stats do
# Regular tables + partitioned tables (summing child partition stats)
# pg_class + pg_stat_user_tables join is slow — 5s cache is fine since
# table row counts barely change in a 5-second window.
Cache.fetch_or_store({__MODULE__, :db_stats}, 5_000, fn -> fetch_db_stats_raw() end)
end
defp fetch_db_stats_raw do
%{rows: rows} =
Repo.query!("""
SELECT

View file

@ -35,37 +35,29 @@ defmodule MicrowavepropWeb.ContactLive.Show do
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:hrrr")
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:weather")
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "contact_enrichment:solar")
# Defer the expensive enrichment loads (HRRR profiles, terrain profile,
# IEMRE, elevation profile, propagation analysis — each touching large
# tables or doing ITU-R math) until after the initial shell has been
# sent. Users see the contact basics in ~20ms instead of waiting the
# full ~500ms-2s mount.
send(self(), {:hydrate, can_enqueue})
end
weather = load_weather(contact)
solar = if can_enqueue, do: maybe_enqueue_solar(contact), else: load_solar(contact)
hrrr_path = Weather.hrrr_profiles_for_path(contact)
hrrr = List.first(hrrr_path)
{hrrr, contact} =
if can_enqueue, do: maybe_enqueue_hrrr(hrrr, contact), else: {hrrr, contact}
terrain = Terrain.get_terrain_profile(contact.id)
contact = if can_enqueue, do: maybe_enqueue_terrain(terrain, contact), else: contact
contact = if can_enqueue, do: maybe_enqueue_weather(weather, contact), else: contact
iemre = load_iemre(contact)
elevation_profile = compute_elevation_profile(contact, hrrr_path, weather.soundings)
propagation_analysis = build_propagation_analysis(contact, hrrr_path, terrain, elevation_profile, weather.soundings)
data_sources = build_data_sources(hrrr, hrrr_path, terrain, weather, elevation_profile)
{:ok,
assign(socket,
page_title: "#{contact.station1} / #{contact.station2}",
contact: contact,
surface_observations: weather.surface_observations,
soundings: weather.soundings,
solar: solar,
hrrr: hrrr,
iemre: iemre,
terrain: terrain,
elevation_profile: elevation_profile,
propagation_analysis: propagation_analysis,
data_sources: data_sources,
surface_observations: [],
soundings: [],
solar: nil,
hrrr: nil,
iemre: nil,
terrain: nil,
elevation_profile: nil,
propagation_analysis: nil,
data_sources: nil,
hydrated: false,
terrain_expanded: false,
hrrr_profile_expanded: false,
obs_sort_by: "station_name",
@ -234,6 +226,44 @@ defmodule MicrowavepropWeb.ContactLive.Show do
end
@impl true
def handle_info({:hydrate, can_enqueue}, socket) do
contact = socket.assigns.contact
weather = load_weather(contact)
solar = if can_enqueue, do: maybe_enqueue_solar(contact), else: load_solar(contact)
hrrr_path = Weather.hrrr_profiles_for_path(contact)
hrrr = List.first(hrrr_path)
{hrrr, contact} =
if can_enqueue, do: maybe_enqueue_hrrr(hrrr, contact), else: {hrrr, contact}
terrain = Terrain.get_terrain_profile(contact.id)
contact = if can_enqueue, do: maybe_enqueue_terrain(terrain, contact), else: contact
contact = if can_enqueue, do: maybe_enqueue_weather(weather, contact), else: contact
iemre = load_iemre(contact)
elevation_profile = compute_elevation_profile(contact, hrrr_path, weather.soundings)
propagation_analysis =
build_propagation_analysis(contact, hrrr_path, terrain, elevation_profile, weather.soundings)
data_sources = build_data_sources(hrrr, hrrr_path, terrain, weather, elevation_profile)
{:noreply,
assign(socket,
contact: contact,
surface_observations: weather.surface_observations,
soundings: weather.soundings,
solar: solar,
hrrr: hrrr,
iemre: iemre,
terrain: terrain,
elevation_profile: elevation_profile,
propagation_analysis: propagation_analysis,
data_sources: data_sources,
hydrated: true
)}
end
def handle_info({:terrain_ready, _contact_id}, socket) do
contact = %{socket.assigns.contact | terrain_status: :complete}
terrain = Terrain.get_terrain_profile(contact.id)
@ -502,15 +532,17 @@ defmodule MicrowavepropWeb.ContactLive.Show do
end
defp fetch_queue_counts do
import Ecto.Query
Microwaveprop.Cache.fetch_or_store({__MODULE__, :queue_counts}, 5_000, fn ->
import Ecto.Query
from(j in Oban.Job,
where: j.state in ["available", "scheduled", "retryable", "executing"],
group_by: j.queue,
select: {j.queue, count(j.id)}
)
|> Microwaveprop.Repo.all()
|> Map.new()
from(j in Oban.Job,
where: j.state in ["available", "scheduled", "retryable", "executing"],
group_by: j.queue,
select: {j.queue, count(j.id)}
)
|> Microwaveprop.Repo.all()
|> Map.new()
end)
end
defp internal_network?(session) do

View file

@ -0,0 +1,85 @@
defmodule Microwaveprop.Weather.GridCacheTest do
use ExUnit.Case, async: false
alias Microwaveprop.Weather.GridCache
setup do
GridCache.clear()
:ok
end
describe "fetch/1" do
test "returns :miss when nothing is cached" do
assert GridCache.fetch(~U[2026-04-12 12:00:00Z]) == :miss
end
test "returns cached rows after put" do
rows = [%{lat: 32.0, lon: -97.0, temperature: 25.0}]
GridCache.put(~U[2026-04-12 12:00:00Z], rows)
assert {:ok, [%{lat: 32.0, lon: -97.0}]} = GridCache.fetch(~U[2026-04-12 12:00:00Z])
end
end
describe "fetch_bounds/2" do
setup do
rows = [
%{lat: 32.0, lon: -97.0, temperature: 25.0},
%{lat: 40.0, lon: -74.0, temperature: 20.0},
%{lat: 34.0, lon: -98.0, temperature: 28.0}
]
GridCache.put(~U[2026-04-12 12:00:00Z], rows)
:ok
end
test "returns all rows when bounds are nil" do
assert {:ok, list} = GridCache.fetch_bounds(~U[2026-04-12 12:00:00Z], nil)
assert length(list) == 3
end
test "returns only rows inside the bounds" do
bounds = %{"south" => 31.0, "north" => 35.0, "west" => -100.0, "east" => -95.0}
assert {:ok, list} = GridCache.fetch_bounds(~U[2026-04-12 12:00:00Z], bounds)
assert length(list) == 2
end
test "returns :miss when the valid_time is not cached" do
bounds = %{"south" => 0.0, "north" => 90.0, "west" => -180.0, "east" => 0.0}
assert GridCache.fetch_bounds(~U[2099-01-01 00:00:00Z], bounds) == :miss
end
end
describe "fetch_point/3" do
setup do
rows = [
%{lat: 32.0, lon: -97.0, temperature: 25.0},
%{lat: 33.0, lon: -97.0, temperature: 27.0}
]
GridCache.put(~U[2026-04-12 12:00:00Z], rows)
:ok
end
test "returns the cached row for a known point" do
assert {:ok, %{temperature: 25.0}} =
GridCache.fetch_point(~U[2026-04-12 12:00:00Z], 32.0, -97.0)
end
test "returns :miss for an unknown point" do
assert GridCache.fetch_point(~U[2026-04-12 12:00:00Z], 40.0, -74.0) == :miss
end
end
describe "latest_valid_time/0" do
test "returns the most recent cached valid_time" do
GridCache.put(~U[2026-04-12 10:00:00Z], [])
GridCache.put(~U[2026-04-12 14:00:00Z], [])
GridCache.put(~U[2026-04-12 12:00:00Z], [])
assert GridCache.latest_valid_time() == ~U[2026-04-12 14:00:00Z]
end
test "returns nil when nothing is cached" do
assert GridCache.latest_valid_time() == nil
end
end
end