From 4f82bd691e7a98df1f289300bea37245af31c70c Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 24 Apr 2026 18:48:47 -0500 Subject: [PATCH] perf(prop): cap ScoreCache LRU + drop eager warmers to fix hot-pod OOMs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ScoreCache held 437 entries × ~1.94 MiB each (~850 MiB per pod) of {band_mhz, valid_time} grids — data already on NFS as compact .prop files. NotifyListener and ScoreCacheReconciler both eagerly materialised every band into ETS on each Rust completion / 60s sweep, so 5 hot replicas wasted ~4.2 GiB of redundant cache and OOMed at 6 GiB. Bound the cache to 32 entries with eviction by oldest valid_time, drop the eager warm loops, and let LiveView callers lazy-fill via the existing ScoresFile read path. ScoreCacheReconciler had no remaining purpose and is deleted; runbook + prom_ex counters updated to match. Steady-state cache footprint ~60 MiB per pod instead of ~850 MiB. --- docs/runbook_propagation_pipeline.md | 92 +++---- lib/microwaveprop/application.ex | 1 - .../prom_ex/instrument_plugin.ex | 10 - lib/microwaveprop/propagation.ex | 8 +- .../propagation/notify_listener.ex | 58 +--- lib/microwaveprop/propagation/score_cache.ex | 63 ++++- .../propagation/score_cache_reconciler.ex | 142 ---------- .../propagation/notify_listener_test.exs | 82 ++---- .../score_cache_reconciler_test.exs | 249 ------------------ .../propagation/score_cache_test.exs | 56 ++++ 10 files changed, 189 insertions(+), 572 deletions(-) delete mode 100644 lib/microwaveprop/propagation/score_cache_reconciler.ex delete mode 100644 test/microwaveprop/propagation/score_cache_reconciler_test.exs diff --git a/docs/runbook_propagation_pipeline.md b/docs/runbook_propagation_pipeline.md index 0e19e0ec..d4e6acbc 100644 --- a/docs/runbook_propagation_pipeline.md +++ b/docs/runbook_propagation_pipeline.md @@ -6,9 +6,9 @@ how the system recovers. ``` prop-grid-rs (worker nodes) Postgres (10.0.15.30) prop pods (node1/2/3) - ├─ fetch GRIB2 (NOMADS) ├─ grid_tasks (SKIP LOCKED) ├─ ScoreCache (ETS) + ├─ fetch GRIB2 (NOMADS) ├─ grid_tasks (SKIP LOCKED) ├─ ScoreCache (ETS, capped LRU) ├─ decode ├─ LISTEN/NOTIFY channel ├─ NotifyListener - ├─ score 22 bands × 92k │ "propagation_ready" ├─ ScoreCacheReconciler (60s) + ├─ score 22 bands × 92k │ "propagation_ready" ├─ Lazy disk read on miss ├─ write /data/scores/.../.prop │ └─ PubSub → LiveView └─ NOTIFY propagation_ready ────┘ ``` @@ -19,45 +19,35 @@ prop-grid-rs (worker nodes) Postgres (10.0.15.30) prop pods (nod file via atomic rename. (Readers also accept legacy `.ntms` files with `NTMS` magic during a rolling deploy.) 2. Rust emits `NOTIFY propagation_ready ''`. -3. `NotifyListener` on each Elixir pod receives it and calls - `Propagation.warm_cache_and_broadcast/2` which reads the file once - and PubSub-broadcasts the payload to every peer. -4. Each pod's `ScoreCache` receives the message and writes into its - local ETS table. `/map` clients are notified via the - `"propagation:updated"` topic. +3. `NotifyListener` on each Elixir pod receives it, prunes + ScoreCache entries outside the active forecast window, and + broadcasts `propagation_updated` over PubSub. +4. LiveView clients see the broadcast and re-request scores. The + first request per `{band, valid_time}` lazy-fills `ScoreCache` + from the on-disk `.prop` file (~50–100 ms); subsequent requests + hit ETS until the LRU evicts. -Latency budget end-to-end: < 2s. +Latency budget end-to-end: < 2 s on the warm path; < 200 ms added +on the first lazy fill. ## Failure modes ### FM1 — NOTIFY dropped on reconnect -**Symptom.** `{band, valid_time}` files land on disk but ScoreCache -stays cold until the first lazy miss. +**Symptom.** `{band, valid_time}` files land on disk but +LiveView clients don't auto-refresh until the user interacts. **Why.** Postgres queues LISTEN notifications per-connection and discards any that are unread on disconnect. A DB restart, a network blip, or the Postgrex.Notifications connection hitting its idle timeout all drop whatever queued notifies were in flight. -**Recovery.** `ScoreCacheReconciler` sweeps `/data/scores` every 60s -per-pod and warms missing `{band, valid_time}` pairs from disk into -local ETS. Each pod reconciles independently — no cluster messaging -required. +**Recovery.** Lazy fill: the next `/map` request reads the on-disk +`.prop` file directly via `Propagation.scores_at_fresh/3`. Stale +ETS entries get evicted by NotifyListener's `prune_outside_window` +on the next NOTIFY that does land. -### FM2 — NotifyListener not running locally - -**Symptom.** One pod is cold while peers are warm. `/map` load time -on that pod is ~150ms slower than average until the reconciler runs. - -**Why.** The listener isn't currently part of the supervision tree -(see `lib/microwaveprop/application.ex`). When it is re-added, early -startup or Postgrex.Notifications boot failure can leave it -unsubscribed for several seconds while Rust writes land. - -**Recovery.** Same as FM1 — reconciler catches up within one tick. - -### FM3 — NFS stale read racing Rust write +### FM2 — NFS stale read racing Rust write **Symptom.** `ScoresFile.read/2` returns `{:error, reason}` when the reader opens the file exactly between Rust's `rename(2)` of the @@ -67,30 +57,22 @@ the reader opens the file exactly between Rust's `rename(2)` of the NFS client cache can briefly serve a stale directory listing or a no-longer-existing path. -**Recovery.** Reconciler calls `ScoresFile.read/2` directly and -pattern-matches `{:error, _} -> :error`, logging at debug and -skipping the key. The next tick re-reads the file after NFS -consistency settles. Lazy-miss path -(`Propagation.scores_at_fetch/3`) has historically surfaced this as -a LiveView error — the reconciler warms the cache before that miss -happens. +**Recovery.** LiveView callers retry on the next interaction; the +read settles after NFS consistency converges (sub-second). -### FM4 — Cluster partition splits ScoreCache broadcast +### FM3 — Cluster partition -**Symptom.** Pods on one side of a cluster partition have up-to-date -cache; pods on the other side don't, even though both can read NFS -and Postgres. +**Symptom.** Pods on one side of a cluster partition don't see +the other side's `propagation_updated` PubSub messages. -**Why.** `ScoreCache.broadcast_put/3` fans the payload out over -PubSub. A libcluster partition stops the message reaching partitioned -peers. +**Why.** Phoenix.PubSub fans messages over the BEAM cluster mesh. +A libcluster partition stops the message reaching partitioned peers. -**Recovery.** Each pod's reconciler reads from NFS directly, so it -converges regardless of PubSub reachability. Partition does not -cause divergent scores — only transient latency until the next -sweep. +**Recovery.** Each pod reads scores from the shared NFS mount on +demand, so partitions only delay LiveView refresh — they never cause +divergent answers. -### FM5 — Rust worker dead +### FM4 — Rust worker dead **Symptom.** `grid_tasks` rows with `status = 'queued'` accumulate; no new `.prop` files land; `/map` serves the last successful run's @@ -112,17 +94,17 @@ only as the cron entry point, not as a fallback compute path. ## Monitoring signals - `avg_over_time(beam_memory_total_bytes[1h])` on prop pods — steady - climb above 1 GiB means the cache isn't pruning (see - `ScoreCache.prune_older_than/1` in `NotifyListener`). -- Reconciler log line `"ScoreCacheReconciler: warmed N"` — if N is - consistently non-zero every minute, NOTIFY is dropping and - NotifyListener wants investigation. + climb above 1 GiB means the ScoreCache LRU cap isn't holding (see + `ScoreCache.max_entries/0`). +- `microwaveprop.propagation.scores_at.cache.count{hit=...}` — if the + miss rate is consistently above ~30 % the LRU cap may be too small + for the active session count. - `grid_tasks` queue depth in Prometheus — should drain to 0 between hourly cron ticks. ## Test coverage -- `test/microwaveprop/propagation/score_cache_reconciler_test.exs` - covers FM1 and FM3 (missing-file skip). -- Partition recovery (FM4) is untested in CI — requires multi-node +- `test/microwaveprop/propagation/score_cache_test.exs` covers the + LRU eviction path. +- Partition recovery (FM3) is untested in CI — requires multi-node cluster and libcluster orchestration. diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 901bcf36..d7b2b1cf 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -25,7 +25,6 @@ defmodule Microwaveprop.Application do child_spec: Task.Supervisor, name: Microwaveprop.TaskSupervisor, partitions: System.schedulers_online()}, Microwaveprop.Cache, Microwaveprop.Propagation.ScoreCache, - Microwaveprop.Propagation.ScoreCacheReconciler, Microwaveprop.Weather.GridCache, {Microwaveprop.Weather.IemRateLimiter, interval_ms: Application.get_env(:microwaveprop, :iem_rate_limiter_interval_ms, 700)}, diff --git a/lib/microwaveprop/prom_ex/instrument_plugin.ex b/lib/microwaveprop/prom_ex/instrument_plugin.ex index c4a346e9..8dc392ef 100644 --- a/lib/microwaveprop/prom_ex/instrument_plugin.ex +++ b/lib/microwaveprop/prom_ex/instrument_plugin.ex @@ -287,16 +287,6 @@ defmodule Microwaveprop.PromEx.InstrumentPlugin do event_name: [:microwaveprop, :weather, :grid_cache, :lookup], tags: [:hit], description: "GridCache hit/miss for the /weather map's grid fetch" - ), - counter("microwaveprop.propagation.notify_listener.warm.ok.count", - event_name: [:microwaveprop, :propagation, :notify_listener, :warm], - measurement: :ok, - description: "Bands successfully warmed per Rust propagation_ready notification" - ), - counter("microwaveprop.propagation.notify_listener.warm.err.count", - event_name: [:microwaveprop, :propagation, :notify_listener, :warm], - measurement: :err, - description: "Bands that failed to warm per Rust propagation_ready notification" ) ] ) diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 6c8ba3da..28fad1a2 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -256,10 +256,10 @@ defmodule Microwaveprop.Propagation do @doc """ The active forecast window for the `/map` UI: one hour in the past - through HRRR's 18-hour forecast horizon. Used by `ScoreCacheReconciler` - and `NotifyListener` to bound ETS growth — long-horizon GEFS `.prop` - files on disk must not balloon `propagation_score_cache` past the - memory the UI actually reads. + through HRRR's 18-hour forecast horizon. Used by `NotifyListener` to + bound ETS growth — long-horizon GEFS `.prop` files on disk must not + balloon `propagation_score_cache` past the memory the UI actually + reads. """ @spec hot_cache_window() :: {DateTime.t(), DateTime.t()} def hot_cache_window do diff --git a/lib/microwaveprop/propagation/notify_listener.ex b/lib/microwaveprop/propagation/notify_listener.ex index 1aab8975..87cab08d 100644 --- a/lib/microwaveprop/propagation/notify_listener.ex +++ b/lib/microwaveprop/propagation/notify_listener.ex @@ -15,7 +15,6 @@ defmodule Microwaveprop.Propagation.NotifyListener do use GenServer alias Microwaveprop.Propagation - alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Repo @@ -85,60 +84,29 @@ defmodule Microwaveprop.Propagation.NotifyListener do end @doc """ - Warm `ScoreCache` for every band at `valid_time`, prune old cache - entries, and broadcast `propagation_updated` to PubSub. Public so - tests can exercise the warm+telemetry path without standing up the + Prune ScoreCache entries outside the active forecast window, then + broadcast `propagation_updated` to PubSub so LiveView clients refresh. + + Note: this no longer eagerly materialises every band's 92k-cell grid + into ETS. Each entry weighed ~1.94 MiB and 23 bands × 19 forecast + hours kept hot pods around 850 MiB of redundant cache (data already + on NFS as `.prop` files), tripping the 6 GiB OOM. LiveView callers + now lazy-fill via `Propagation.scores_at/3` on first access; the + ScoreCache cap (`ScoreCache.max_entries/0`) keeps the working set + bounded thereafter. + + Public so tests can exercise the path without standing up the Postgrex.Notifications subscriber. """ @spec handle_propagation_ready(DateTime.t()) :: :ok def handle_propagation_ready(valid_time) do - {ok, err} = - Enum.reduce(BandConfig.all_bands(), {0, 0}, fn band, {ok, err} -> - case warm_band(band.freq_mhz, valid_time) do - :ok -> {ok + 1, err} - :error -> {ok, err + 1} - end - end) - - :telemetry.execute( - [:microwaveprop, :propagation, :notify_listener, :warm], - %{ok: ok, err: err}, - %{valid_time: valid_time} - ) - - # Prune anything outside the active map window. Old entries (>1h - # past) and long-horizon GEFS entries (>18h future) would otherwise - # accumulate and push the `propagation_score_cache` ETS table past - # 2 GiB — scheduler starvation from the GC sweeps tripped the - # /live probe and cycled the pods every ~25 min. {past, future} = Propagation.hot_cache_window() ScoreCache.prune_outside_window(past, future) _ = Phoenix.PubSub.broadcast(@pubsub, @topic, {:propagation_updated, [valid_time]}) - if err > 0 do - Logger.warning("NotifyListener: warmed #{ok} bands, #{err} failed for valid_time=#{valid_time}") - else - Logger.info("NotifyListener: warmed ScoreCache + broadcast for valid_time=#{valid_time}") - end + Logger.info("NotifyListener: broadcast propagation_updated for valid_time=#{valid_time}") :ok end - - defp warm_band(band_mhz, valid_time) do - case Propagation.warm_cache_and_broadcast(band_mhz, valid_time) do - :ok -> - :ok - - {:error, reason} -> - Logger.warning("NotifyListener: warm skipped band=#{band_mhz} valid_time=#{valid_time}: #{inspect(reason)}") - - :error - end - rescue - e -> - Logger.warning("NotifyListener: warm crashed band=#{band_mhz} valid_time=#{valid_time}: #{inspect(e)}") - - :error - end end diff --git a/lib/microwaveprop/propagation/score_cache.ex b/lib/microwaveprop/propagation/score_cache.ex index b0294223..41de2116 100644 --- a/lib/microwaveprop/propagation/score_cache.ex +++ b/lib/microwaveprop/propagation/score_cache.ex @@ -2,14 +2,20 @@ defmodule Microwaveprop.Propagation.ScoreCache do @moduledoc """ Node-local ETS cache of propagation scores keyed by `{band_mhz, valid_time}`. - Populated eagerly by `Microwaveprop.Workers.PropagationGridWorker` after each - hourly compute (via `broadcast_put/3`) and lazily by `Microwaveprop.Propagation.scores_at/3` - on a cache miss. `broadcast_put/3` fans the payload out across the cluster on - the `"propagation:cache"` PubSub topic so every BEAM node stays in sync with - a single compute. + Populated lazily by `Microwaveprop.Propagation.scores_at/3` on the first + request for a given `{band_mhz, valid_time}`; subsequent requests hit ETS + until the entry is evicted. The cache is bounded to `max_entries/0` total + entries via LRU-by-`valid_time` eviction; on overflow the oldest + `valid_time` is dropped first so the freshly-arrived hour stays warm. - Internally each cache entry stores a `%{{lat, lon} => score}` map so that - per-point lookups (used by `point_forecast/3` and `point_detail/4`) are O(1). + `broadcast_put/3` and the matching `"propagation:cache"` PubSub topic are + retained for the rare cluster-wide pre-warm path; the hot pipeline (Rust + Phase 2 cutover) does not call them — `NotifyListener` only emits the + lighter-weight `"propagation:updated"` PubSub so LiveView clients refresh + while the actual scores load on demand from `/data/scores/.../*.prop`. + + Each cache entry stores a `%{{lat, lon} => score}` map so per-point lookups + (used by `point_forecast/3` and `point_detail/4`) stay O(1) on hits. """ use GenServer @@ -19,6 +25,15 @@ defmodule Microwaveprop.Propagation.ScoreCache do @topic "propagation:cache" @pubsub Microwaveprop.PubSub + # Hard cap on cached `{band_mhz, valid_time}` entries. Each value is a + # 92k-cell `Map.new` weighing ~1.94 MiB; without a cap the eager warm + # path materialises 23 bands × 19 forecast hours = 437 entries per pod + # (~850 MiB) and OOMs the BEAM. The cap keeps the LiveView hot path + # warm (one band × the hour or two the user is actively scrubbing) + # and lets the rest fall back to direct `.prop` reads via + # `read_from_disk_and_cache/3`, which is sub-100 ms per file. + @max_entries 32 + @type score :: %{lat: float(), lon: float(), score: non_neg_integer()} @type bounds :: %{optional(String.t()) => float()} @@ -67,13 +82,47 @@ defmodule Microwaveprop.Propagation.ScoreCache do end end + @doc "Maximum number of `{band_mhz, valid_time}` entries kept in ETS." + @spec max_entries() :: pos_integer() + def max_entries, do: @max_entries + @spec put(non_neg_integer(), DateTime.t(), [score()]) :: :ok def put(band_mhz, valid_time, scores) do grid = list_to_grid(scores) :ets.insert(@table, {{band_mhz, valid_time}, grid}) + enforce_capacity() :ok end + # When the table is over the cap, evict whichever entry has the oldest + # `valid_time`. Picking the oldest matches the LiveView access pattern: + # the timeline scrubs forward, the hourly chain replaces the trailing + # edge, and the freshest hours are by far the most-requested. Loops + # until the table is back at or below the cap so that any historical + # over-shoot self-heals after this commit deploys. + defp enforce_capacity do + if :ets.info(@table, :size) > @max_entries do + case oldest_key() do + nil -> :ok + key -> :ets.delete(@table, key) + end + + enforce_capacity() + end + end + + defp oldest_key do + fun = fn + {{band, vt}, _}, nil -> + {band, vt} + + {{band, vt}, _}, {_, oldest} = acc -> + if DateTime.before?(vt, oldest), do: {band, vt}, else: acc + end + + :ets.foldl(fun, nil, @table) + end + @doc """ Insert locally AND broadcast to peer nodes via PubSub. diff --git a/lib/microwaveprop/propagation/score_cache_reconciler.ex b/lib/microwaveprop/propagation/score_cache_reconciler.ex deleted file mode 100644 index b54fe790..00000000 --- a/lib/microwaveprop/propagation/score_cache_reconciler.ex +++ /dev/null @@ -1,142 +0,0 @@ -defmodule Microwaveprop.Propagation.ScoreCacheReconciler do - @moduledoc """ - Periodic safety net that reconciles `ScoreCache` with the on-disk - `/data/scores` tree written by the Rust `prop-grid-rs` worker. - - The primary refresh path is `NOTIFY propagation_ready` → `NotifyListener` - → `Propagation.warm_cache_and_broadcast/2` → PubSub fan-out. That path - fails silently in three ways this reconciler is designed to cover: - - 1. **NOTIFY dropped on reconnect.** Postgres queues LISTEN notifies - per-connection and drops anything unread on disconnect. When - `Postgrex.Notifications` reconnects (DB restart, network blip) - a full run of score files may already be on disk. - 2. **No listener running locally.** Orphan pods, readiness misfires, - or early startup mean the LISTEN handler hasn't subscribed yet. - 3. **Rust writes completed pre-boot.** Pod starts, cache is empty, - and we don't hit the miss path until the first LiveView click. - - Every `interval_ms` the reconciler lists `/data/scores//*.prop` - per band, compares against `ScoreCache.valid_times/1`, and fills in - any {band, valid_time} pairs on disk but absent in the local ETS. - Warming is node-local (`ScoreCache.put/3`) — no PubSub fan-out, since - every pod runs its own reconciler and each one converges independently. - """ - use GenServer - - alias Microwaveprop.Propagation - alias Microwaveprop.Propagation.BandConfig - alias Microwaveprop.Propagation.ScoreCache - alias Microwaveprop.Propagation.ScoresFile - - require Logger - - @default_interval_ms 60_000 - @default_jitter_max_ms 20_000 - - @spec start_link(keyword()) :: GenServer.on_start() | :ignore - def start_link(opts) do - if Application.get_env(:microwaveprop, :start_score_cache_reconciler, true) do - GenServer.start_link(__MODULE__, opts, name: __MODULE__) - else - :ignore - end - end - - @doc """ - Runs one reconciliation pass synchronously. Returns the count of - {band, valid_time} pairs warmed into the cache. - """ - @spec sweep_once() :: non_neg_integer() - def sweep_once do - Enum.reduce(BandConfig.all_bands(), 0, fn %{freq_mhz: band_mhz}, acc -> - acc + warm_missing_for_band(band_mhz) - end) - end - - @doc """ - Returns the next sweep delay in milliseconds: the configured base - interval plus a uniformly random jitter in `[1, jitter_max_ms]`. - - Scheduling each pod's next tick with a randomized offset prevents the - cluster-wide "every replica sweeps at :00" thundering herd against the - shared NFS scores mount and Postgres. - """ - @spec next_sweep_interval() :: non_neg_integer() - def next_sweep_interval(base_ms \\ @default_interval_ms, jitter_max_ms \\ @default_jitter_max_ms) - - def next_sweep_interval(base_ms, jitter_max_ms) when jitter_max_ms > 0 do - base_ms + :rand.uniform(jitter_max_ms) - end - - def next_sweep_interval(base_ms, _jitter_max_ms), do: base_ms - - @impl true - def init(opts) do - interval = Keyword.get(opts, :interval_ms, @default_interval_ms) - jitter_max = Keyword.get(opts, :jitter_max_ms, @default_jitter_max_ms) - - _ = - if Keyword.get(opts, :run_on_start, true) do - _ = Process.send_after(self(), :sweep, 500) - end - - {:ok, %{interval_ms: interval, jitter_max_ms: jitter_max}} - end - - @impl true - def handle_info(:sweep, state) do - warmed = sweep_once() - - if warmed > 0 do - Logger.info("ScoreCacheReconciler: warmed #{warmed} {band, valid_time} pairs from disk") - end - - delay = next_sweep_interval(state.interval_ms, state.jitter_max_ms) - Process.send_after(self(), :sweep, delay) - {:noreply, state} - end - - def handle_info(_msg, state), do: {:noreply, state} - - defp warm_missing_for_band(band_mhz) do - {past_cutoff, future_cutoff} = Propagation.hot_cache_window() - - on_disk = - band_mhz - |> ScoresFile.list_valid_times() - |> Enum.filter(fn vt -> - DateTime.compare(vt, past_cutoff) != :lt and - DateTime.compare(vt, future_cutoff) != :gt - end) - |> MapSet.new() - - in_cache = MapSet.new(ScoreCache.valid_times(band_mhz)) - missing = MapSet.difference(on_disk, in_cache) - - Enum.reduce(missing, 0, fn valid_time, acc -> - case warm_one(band_mhz, valid_time) do - :ok -> acc + 1 - :error -> acc - end - end) - end - - # File may vanish between list and read if a prune runs concurrently, - # or be half-written during an atomic-rename race. `ScoresFile.read/2` - # reports both as `{:error, _}`; we skip and self-heal on the next - # sweep rather than caching an empty grid under that key. - defp warm_one(band_mhz, valid_time) do - case ScoresFile.read(band_mhz, valid_time) do - {:ok, payload} -> - scores = ScoresFile.extract_points(payload, nil) - ScoreCache.put(band_mhz, valid_time, scores) - :ok - - {:error, reason} -> - Logger.debug("ScoreCacheReconciler: skipped #{band_mhz}/#{valid_time}: #{inspect(reason)}") - - :error - end - end -end diff --git a/test/microwaveprop/propagation/notify_listener_test.exs b/test/microwaveprop/propagation/notify_listener_test.exs index 2d8c989b..89f3c81c 100644 --- a/test/microwaveprop/propagation/notify_listener_test.exs +++ b/test/microwaveprop/propagation/notify_listener_test.exs @@ -43,75 +43,27 @@ defmodule Microwaveprop.Propagation.NotifyListenerTest do ] end - describe "handle_propagation_ready/1 telemetry" do - test "emits propagation.notify_listener.warm with ok/err counts after warming every band" do + describe "handle_propagation_ready/1" do + test "does NOT eagerly warm ScoreCache (lazy fill via on-disk reads)" do valid_time = ~U[2026-04-21 12:00:00Z] - # Write a scores file for the first band only so the rest hit :enoent - # and contribute to the err tally. That gives us at least one ok and - # at least one err for an all-bands warm. - [first_band | _rest] = Enum.map(BandConfig.all_bands(), & &1.freq_mhz) - ScoresFile.write!(first_band, valid_time, sample_scores()) - - total_bands = length(BandConfig.all_bands()) - expected_err = total_bands - 1 - - test_pid = self() - handler_id = {__MODULE__, :notify_listener_warm} - - :telemetry.attach( - handler_id, - [:microwaveprop, :propagation, :notify_listener, :warm], - fn event, measurements, metadata, _config -> - send(test_pid, {:telemetry, event, measurements, metadata}) - end, - nil - ) - - try do - :ok = NotifyListener.handle_propagation_ready(valid_time) - - event = [:microwaveprop, :propagation, :notify_listener, :warm] - - assert_receive {:telemetry, ^event, %{ok: 1, err: ^expected_err}, %{valid_time: ^valid_time}} - after - :telemetry.detach(handler_id) - end - end - - test "emits telemetry even when every band warms successfully" do - valid_time = ~U[2026-04-21 13:00:00Z] - + # Write all bands so the old code path would have warmed every one. for band <- BandConfig.all_bands() do ScoresFile.write!(band.freq_mhz, valid_time, sample_scores()) end - total_bands = length(BandConfig.all_bands()) + assert :ets.info(:propagation_score_cache, :size) == 0 - test_pid = self() - handler_id = {__MODULE__, :notify_listener_warm_all_ok} + :ok = NotifyListener.handle_propagation_ready(valid_time) + ScoreCache.sync() - :telemetry.attach( - handler_id, - [:microwaveprop, :propagation, :notify_listener, :warm], - fn event, measurements, metadata, _config -> - send(test_pid, {:telemetry, event, measurements, metadata}) - end, - nil - ) - - try do - :ok = NotifyListener.handle_propagation_ready(valid_time) - - event = [:microwaveprop, :propagation, :notify_listener, :warm] - - assert_receive {:telemetry, ^event, %{ok: ^total_bands, err: 0}, %{valid_time: ^valid_time}} - after - :telemetry.detach(handler_id) - end + # Cache stays empty: NotifyListener no longer materialises the + # 92k-cell grid for every band into ETS. LiveView callers fall + # through to `read_from_disk_and_cache/3` on demand. + assert :ets.info(:propagation_score_cache, :size) == 0 end - test "broadcasts propagation:updated to PubSub after warming" do + test "broadcasts propagation:updated to PubSub" do valid_time = ~U[2026-04-21 14:00:00Z] ScoresFile.write!(10_000, valid_time, sample_scores()) @@ -121,6 +73,18 @@ defmodule Microwaveprop.Propagation.NotifyListenerTest do assert_receive {:propagation_updated, [^valid_time]}, 1_000 end + + test "prunes ScoreCache entries outside the active forecast window" do + stale = ~U[2026-04-12 00:00:00Z] + ScoreCache.put(10_000, stale, [%{lat: 1.0, lon: 1.0, score: 1}]) + + now_valid_time = DateTime.truncate(DateTime.utc_now(), :second) + ScoresFile.write!(10_000, now_valid_time, sample_scores()) + + :ok = NotifyListener.handle_propagation_ready(now_valid_time) + + assert ScoreCache.fetch(10_000, stale) == :miss + end end describe "lifecycle" do diff --git a/test/microwaveprop/propagation/score_cache_reconciler_test.exs b/test/microwaveprop/propagation/score_cache_reconciler_test.exs deleted file mode 100644 index 5f5c2d61..00000000 --- a/test/microwaveprop/propagation/score_cache_reconciler_test.exs +++ /dev/null @@ -1,249 +0,0 @@ -defmodule Microwaveprop.Propagation.ScoreCacheReconcilerTest do - # async: false — the test redirects the global :propagation_scores_dir - # Application env and mutates the shared ScoreCache ETS table. - use ExUnit.Case, async: false - - import ExUnit.CaptureLog - - alias Microwaveprop.Propagation.BandConfig - alias Microwaveprop.Propagation.Grid - alias Microwaveprop.Propagation.ScoreCache - alias Microwaveprop.Propagation.ScoreCacheReconciler - alias Microwaveprop.Propagation.ScoresFile - - setup do - dir = - Path.join( - System.tmp_dir!(), - "reconciler_test_#{System.unique_integer([:positive])}" - ) - - File.mkdir_p!(dir) - prev = Application.get_env(:microwaveprop, :propagation_scores_dir) - Application.put_env(:microwaveprop, :propagation_scores_dir, dir) - ScoreCache.clear() - - on_exit(fn -> - File.rm_rf!(dir) - ScoreCache.clear() - - if prev do - Application.put_env(:microwaveprop, :propagation_scores_dir, prev) - else - Application.delete_env(:microwaveprop, :propagation_scores_dir) - end - end) - - band_mhz = BandConfig.all_bands() |> hd() |> Map.fetch!(:freq_mhz) - {:ok, band_mhz: band_mhz} - end - - defp sample_scores do - %{lat_min: lat_min, lon_min: lon_min, lat_max: lat_max, lon_max: lon_max} = Grid.bounds() - - [ - %{lat: lat_min, lon: lon_min, score: 10}, - %{lat: lat_max, lon: lon_max, score: 90} - ] - end - - # In-window timestamps for reconciler tests — the reconciler now - # filters valid_times to `[now-1h, now+18h]`, so hardcoded fixtures - # in the past drift out of scope as time passes. `current_hour/1` - # returns an hour-aligned DateTime offset by N hours from now. - defp current_hour(offset_hours) do - DateTime.utc_now() - |> Map.put(:minute, 0) - |> Map.put(:second, 0) - |> Map.put(:microsecond, {0, 0}) - |> DateTime.add(offset_hours * 3600, :second) - end - - describe "next_sweep_interval/0" do - test "returns a value in [base + 1, base + jitter_max] with defaults" do - # Exercise the randomized default many times; each draw must land - # inside the advertised (60_000, 80_000] window. `:rand.uniform/1` - # returns >= 1 so the minimum is base + 1, not base. - for _ <- 1..500 do - delay = ScoreCacheReconciler.next_sweep_interval() - assert delay >= 60_001 - assert delay <= 80_000 - end - end - - test "honours explicit base and jitter arguments" do - for _ <- 1..200 do - delay = ScoreCacheReconciler.next_sweep_interval(1_000, 250) - assert delay >= 1_001 - assert delay <= 1_250 - end - end - - test "returns the base interval unchanged when jitter_max is 0" do - assert ScoreCacheReconciler.next_sweep_interval(5_000, 0) == 5_000 - end - end - - describe "sweep_once/0" do - test "warms ScoreCache for every {band, valid_time} on disk and missing from cache", - %{band_mhz: band_mhz} do - valid_time = current_hour(1) - ScoresFile.write!(band_mhz, valid_time, sample_scores()) - - assert :miss = ScoreCache.fetch(band_mhz, valid_time) - - warmed = ScoreCacheReconciler.sweep_once() - - assert warmed >= 1 - assert {:ok, _scores} = ScoreCache.fetch(band_mhz, valid_time) - end - - test "is a no-op when cache is already in sync with disk", - %{band_mhz: band_mhz} do - valid_time = current_hour(2) - ScoresFile.write!(band_mhz, valid_time, sample_scores()) - - # Prime the cache manually - ScoreCache.put(band_mhz, valid_time, sample_scores()) - - assert 0 = ScoreCacheReconciler.sweep_once() - end - - test "returns 0 when the scores directory is empty" do - assert 0 = ScoreCacheReconciler.sweep_once() - end - - test "skips valid_times outside the hot-cache window (now-1h..now+18h)", - %{band_mhz: band_mhz} do - # Three on-disk files: one way in the past, one in the window, - # one far in the future (simulating GEFS Day 2-7). Only the - # middle one should make it into ETS. - now = - DateTime.utc_now() - |> Map.put(:minute, 0) - |> Map.put(:second, 0) - |> Map.put(:microsecond, {0, 0}) - - stale_past = DateTime.add(now, -6 * 3600, :second) - in_window = now - far_future = DateTime.add(now, 72 * 3600, :second) - - ScoresFile.write!(band_mhz, stale_past, sample_scores()) - ScoresFile.write!(band_mhz, in_window, sample_scores()) - ScoresFile.write!(band_mhz, far_future, sample_scores()) - - _ = ScoreCacheReconciler.sweep_once() - - assert ScoreCache.fetch(band_mhz, stale_past) == :miss - assert {:ok, _} = ScoreCache.fetch(band_mhz, in_window) - assert ScoreCache.fetch(band_mhz, far_future) == :miss - end - end - - describe "GenServer lifecycle" do - # The gate in start_link/1 defaults :start_score_cache_reconciler to - # false in the test env. Flip it on for this block so start_supervised! - # gets a live pid rather than :ignore. - setup do - Application.put_env(:microwaveprop, :start_score_cache_reconciler, true) - - on_exit(fn -> - Application.put_env(:microwaveprop, :start_score_cache_reconciler, false) - end) - - :ok - end - - defp wait_for_cache(band_mhz, valid_time, timeout_ms) do - deadline = System.monotonic_time(:millisecond) + timeout_ms - poll_cache(band_mhz, valid_time, deadline) - end - - defp poll_cache(band_mhz, valid_time, deadline) do - case ScoreCache.fetch(band_mhz, valid_time) do - {:ok, _} = ok -> - ok - - :miss -> - if System.monotonic_time(:millisecond) >= deadline do - :timeout - else - Process.sleep(25) - poll_cache(band_mhz, valid_time, deadline) - end - end - end - - test "run_on_start: true warms cache from disk via handle_info(:sweep)", - %{band_mhz: band_mhz} do - valid_time = current_hour(3) - ScoresFile.write!(band_mhz, valid_time, sample_scores()) - - assert :miss = ScoreCache.fetch(band_mhz, valid_time) - - start_supervised!({ScoreCacheReconciler, run_on_start: true, interval_ms: 100}) - - # The boot timer is 500ms, so the first sweep lands somewhere - # around t+500ms. Give it generous headroom to avoid flakes on - # loaded CI workers. - assert {:ok, _scores} = wait_for_cache(band_mhz, valid_time, 2_000) - end - - test "run_on_start: false does not sweep within the wait window", - %{band_mhz: band_mhz} do - valid_time = current_hour(4) - ScoresFile.write!(band_mhz, valid_time, sample_scores()) - - start_supervised!({ScoreCacheReconciler, run_on_start: false, interval_ms: 100}) - - # Wait past the 500ms boot delay the other branch would have used. - Process.sleep(750) - - assert :miss = ScoreCache.fetch(band_mhz, valid_time) - end - - test "interval_ms rescheduling picks up files written after the first sweep", - %{band_mhz: band_mhz} do - first = current_hour(5) - second = current_hour(6) - - ScoresFile.write!(band_mhz, first, sample_scores()) - - # jitter_max_ms: 0 keeps the reschedule deterministic for the - # 2s wait_for_cache window; the jitter default is 20s. - start_supervised!({ScoreCacheReconciler, run_on_start: true, interval_ms: 150, jitter_max_ms: 0}) - - # First sweep warms `first` but knows nothing about `second`. - assert {:ok, _} = wait_for_cache(band_mhz, first, 2_000) - assert :miss = ScoreCache.fetch(band_mhz, second) - - # Drop the second file and wait for the next scheduled tick. - ScoresFile.write!(band_mhz, second, sample_scores()) - - assert {:ok, _} = wait_for_cache(band_mhz, second, 2_000) - end - - test "logs an info line when warmed > 0", %{band_mhz: band_mhz} do - valid_time = current_hour(7) - ScoresFile.write!(band_mhz, valid_time, sample_scores()) - - # config/test.exs pins the global logger to :warning, which drops the - # :info line before capture_log sees it. Raise the level for the - # duration of the test and restore it on exit. - prev_level = Logger.level() - Logger.configure(level: :info) - on_exit(fn -> Logger.configure(level: prev_level) end) - - log = - capture_log([level: :info], fn -> - start_supervised!({ScoreCacheReconciler, run_on_start: true, interval_ms: 100}) - - assert {:ok, _} = wait_for_cache(band_mhz, valid_time, 2_000) - # Give the logger a beat to flush before we snapshot the buffer. - Process.sleep(50) - end) - - assert log =~ "ScoreCacheReconciler: warmed" - end - end -end diff --git a/test/microwaveprop/propagation/score_cache_test.exs b/test/microwaveprop/propagation/score_cache_test.exs index ed635498..c66c684b 100644 --- a/test/microwaveprop/propagation/score_cache_test.exs +++ b/test/microwaveprop/propagation/score_cache_test.exs @@ -184,4 +184,60 @@ defmodule Microwaveprop.Propagation.ScoreCacheTest do ScoreCache.fetch(10_000, ~U[2026-04-12 12:00:00Z]) end end + + describe "max-entries eviction" do + test "caps entries at max_entries, evicting the oldest valid_time" do + cap = ScoreCache.max_entries() + base = ~U[2026-04-12 00:00:00Z] + score = [%{lat: 1.0, lon: 1.0, score: 1}] + + # Insert cap+1 entries, each at a distinct valid_time, oldest first. + for hour <- 0..cap do + ScoreCache.put(10_000, DateTime.add(base, hour * 3600, :second), score) + end + + # Size never exceeds the cap. + assert :ets.info(:propagation_score_cache, :size) == cap + + # The oldest valid_time was evicted. + assert ScoreCache.fetch(10_000, base) == :miss + + # The newest is retained. + newest = DateTime.add(base, cap * 3600, :second) + assert {:ok, [%{score: 1}]} = ScoreCache.fetch(10_000, newest) + end + + test "evicts across bands by valid_time when at capacity" do + cap = ScoreCache.max_entries() + base = ~U[2026-04-12 00:00:00Z] + score = [%{lat: 1.0, lon: 1.0, score: 1}] + + # Fill cap entries on band A at older valid_times. + for hour <- 0..(cap - 1) do + ScoreCache.put(10_000, DateTime.add(base, hour * 3600, :second), score) + end + + assert :ets.info(:propagation_score_cache, :size) == cap + + # One more on band B at a newer valid_time should evict the oldest + # band-A entry, not the new band-B one. + newer = DateTime.add(base, cap * 3600, :second) + ScoreCache.put(24_000, newer, score) + + assert :ets.info(:propagation_score_cache, :size) == cap + assert ScoreCache.fetch(10_000, base) == :miss + assert {:ok, _} = ScoreCache.fetch(24_000, newer) + end + + test "replacing an existing key does not change the cache size" do + base = ~U[2026-04-12 00:00:00Z] + ScoreCache.put(10_000, base, [%{lat: 1.0, lon: 1.0, score: 1}]) + size_before = :ets.info(:propagation_score_cache, :size) + + ScoreCache.put(10_000, base, [%{lat: 1.0, lon: 1.0, score: 99}]) + + assert :ets.info(:propagation_score_cache, :size) == size_before + assert {:ok, [%{score: 99}]} = ScoreCache.fetch(10_000, base) + end + end end