feat(propagation): add ScoreCacheReconciler + pipeline runbook

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

@@ -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]
 

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 '<iso valid_time>'`.
+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.<uniq>` 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.

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)},

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/<band>/*.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

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