Graham McIntire
33be6d008f
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.
104 lines
3.5 KiB
Elixir
104 lines
3.5 KiB
Elixir
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
|