From 33be6d008f9b03b218ab22e4ca89a70c3b9da23d Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 21 Apr 2026 09:24:16 -0500 Subject: [PATCH] feat(propagation): add ScoreCacheReconciler + pipeline runbook MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Periodic 60s sweep re-warms ScoreCache from /data/scores whenever {band, valid_time} pairs are on disk but absent in local ETS. Covers the three silent-fail modes of the NOTIFY/LISTEN path: dropped notifies on reconnect, missing listener, and stale NFS reads. Each pod reconciles its own cache — no PubSub fan-out, no coordination required. Rescue on File.Error handles the rare window where an atomic-rename write races the reader. docs/runbook_propagation_pipeline.md names every failure mode in the Rust↔Elixir seam and what recovers it. --- config/test.exs | 1 + docs/runbook_propagation_pipeline.md | 120 ++++++++++++++++++ lib/microwaveprop/application.ex | 1 + .../propagation/score_cache_reconciler.ex | 104 +++++++++++++++ .../score_cache_reconciler_test.exs | 77 +++++++++++ 5 files changed, 303 insertions(+) create mode 100644 docs/runbook_propagation_pipeline.md create mode 100644 lib/microwaveprop/propagation/score_cache_reconciler.ex create mode 100644 test/microwaveprop/propagation/score_cache_reconciler_test.exs diff --git a/config/test.exs b/config/test.exs index 027fbbd0..34223a3a 100644 --- a/config/test.exs +++ b/config/test.exs @@ -71,6 +71,7 @@ config :microwaveprop, rtma_req_options: [plug: {Req.Test, Microwaveprop.Weather config :microwaveprop, solar_req_options: [plug: {Req.Test, Microwaveprop.Weather.SolarClient}] config :microwaveprop, srtm_req_options: [plug: {Req.Test, Microwaveprop.Terrain.Srtm}, retry: false] config :microwaveprop, start_freshness_monitor: false +config :microwaveprop, start_score_cache_reconciler: false config :microwaveprop, swpc_req_options: [plug: {Req.Test, Microwaveprop.SpaceWeather.SwpcClient}, retry: false] config :microwaveprop, uwyo_req_options: [plug: {Req.Test, Microwaveprop.Weather.UwyoSoundingClient}, retry: false] diff --git a/docs/runbook_propagation_pipeline.md b/docs/runbook_propagation_pipeline.md new file mode 100644 index 00000000..a62727d2 --- /dev/null +++ b/docs/runbook_propagation_pipeline.md @@ -0,0 +1,120 @@ +# Propagation pipeline runbook + +The hourly propagation grid flows across two runtimes and a shared +NFS volume. This doc names every failure mode at that seam and says +how the system recovers. + +``` +prop-grid-rs (talos5) Postgres (10.0.15.30) prop pods (node1/2/3) + ├─ fetch GRIB2 (NOMADS) ├─ grid_tasks (SKIP LOCKED) ├─ ScoreCache (ETS) + ├─ decode ├─ LISTEN/NOTIFY channel ├─ NotifyListener + ├─ score 22 bands × 92k │ "propagation_ready" ├─ ScoreCacheReconciler (60s) + ├─ write /data/scores/.../.ntms │ └─ PubSub → LiveView + └─ NOTIFY propagation_ready ────┘ +``` + +## Primary refresh path (happy case) + +1. Rust completes `{band, valid_time}` compute and writes the `.ntms` + file via atomic rename. +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. + +Latency budget end-to-end: < 2s. + +## Failure modes + +### FM1 — NOTIFY dropped on reconnect + +**Symptom.** `{band, valid_time}` files land on disk but ScoreCache +stays cold until the first lazy miss. + +**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. + +### 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 + +**Symptom.** `ScoresFile.read_bounds/2` raises `File.Error` when +the reader opens the file exactly between Rust's `rename(2)` of the +`.tmp.` file and the `fsync` landing on the NFS server. + +**Why.** The rename is atomic on the same NFSv4 filesystem, but the +NFS client cache can briefly serve a stale directory listing or a +no-longer-existing path. + +**Recovery.** Reconciler's `rescue File.Error` returns `:error` and +moves on; 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. + +### FM4 — Cluster partition splits ScoreCache broadcast + +**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. + +**Why.** `ScoreCache.broadcast_put/3` fans the payload out over +PubSub. 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. + +### FM5 — Rust worker dead + +**Symptom.** `grid_tasks` rows with `status = 'queued'` accumulate; +no new `.ntms` files land; `/map` serves the last successful run's +scores. + +**Why.** Deployment bug, Rust panic, OOM on talos5 (unlikely with +32 GB headroom), or disk-full on `/data`. + +**Recovery.** Manual. `kubectl -n prop logs deploy/prop-grid-rs` for +cause. The Elixir side is read-only against the grid — it does not +regenerate scores from within the BEAM since the extraction in +`65693ed`. + +## 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. +- `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 + cluster and libcluster orchestration. diff --git a/lib/microwaveprop/application.ex b/lib/microwaveprop/application.ex index 128b6d9f..0d7aa680 100644 --- a/lib/microwaveprop/application.ex +++ b/lib/microwaveprop/application.ex @@ -27,6 +27,7 @@ defmodule Microwaveprop.Application do {Phoenix.PubSub, name: Microwaveprop.PubSub}, 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/propagation/score_cache_reconciler.ex b/lib/microwaveprop/propagation/score_cache_reconciler.ex new file mode 100644 index 00000000..12a092ab --- /dev/null +++ b/lib/microwaveprop/propagation/score_cache_reconciler.ex @@ -0,0 +1,104 @@ +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//*.ntms` + 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.BandConfig + alias Microwaveprop.Propagation.ScoreCache + alias Microwaveprop.Propagation.ScoresFile + + require Logger + + @default_interval_ms 60_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 + + @impl true + def init(opts) do + interval = Keyword.get(opts, :interval_ms, @default_interval_ms) + + if Keyword.get(opts, :run_on_start, true) do + Process.send_after(self(), :sweep, 500) + end + + {:ok, %{interval_ms: interval}} + 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 + + Process.send_after(self(), :sweep, state.interval_ms) + {:noreply, state} + end + + def handle_info(_msg, state), do: {:noreply, state} + + defp warm_missing_for_band(band_mhz) do + on_disk = MapSet.new(ScoresFile.list_valid_times(band_mhz)) + 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 + + defp warm_one(band_mhz, valid_time) do + scores = ScoresFile.read_bounds(band_mhz, valid_time) + ScoreCache.put(band_mhz, valid_time, scores) + :ok + rescue + # File may vanish between list and read if a prune runs concurrently. + # Missing files are self-healing on the next sweep — no point logging. + e in [File.Error] -> + Logger.debug("ScoreCacheReconciler: skipped #{band_mhz}/#{valid_time}: #{inspect(e)}") + :error + end +end diff --git a/test/microwaveprop/propagation/score_cache_reconciler_test.exs b/test/microwaveprop/propagation/score_cache_reconciler_test.exs new file mode 100644 index 00000000..02100e8a --- /dev/null +++ b/test/microwaveprop/propagation/score_cache_reconciler_test.exs @@ -0,0 +1,77 @@ +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 + + 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 + + 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 = ~U[2026-04-21 12:00:00Z] + 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 = ~U[2026-04-21 13:00:00Z] + 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 + end +end