refactor(scores): rename file extension .ntms -> .prop with legacy reads

Score-grid files now land at `<band>/<iso>.prop` with a 4-byte "PROP"
magic header. Readers still accept the legacy `.ntms` extension and
"NTMS" magic so a rolling deploy doesn't invalidate any file already
on the shared NFS mount — legacy files age out through normal
retention.

Applied both sides of the seam: Elixir ScoresFile + regex parsers,
Rust scores_file writer + decoder. Updated the comments in all
modules that mention the extension (NotifyListener, Reconciler,
gefs_fetch_worker, map_live) and the runbook diagram. talos5 is a
DB host now, not a Rust worker — corrected the runbook accordingly.
This commit is contained in:
Graham McIntire 2026-04-21 15:54:12 -05:00
parent 989d310447
commit 410a1374fe
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
10 changed files with 184 additions and 33 deletions

View file

@ -5,18 +5,19 @@ 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)
prop-grid-rs (worker nodes) Postgres (talos5 — 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
├─ write /data/scores/.../.prop │ └─ 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.
1. Rust completes `{band, valid_time}` compute and writes the `.prop`
file via atomic rename. (Readers also accept legacy `.ntms` files
with `NTMS` magic during a rolling deploy.)
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
@ -92,11 +93,11 @@ 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
no new `.prop` 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`.
**Why.** Deployment bug, Rust panic, OOM on a worker node, 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

View file

@ -3,7 +3,7 @@ defmodule Microwaveprop.Propagation.NotifyListener do
Listens for `NOTIFY propagation_ready '<iso valid_time>'` from the Rust
`prop-grid-rs` worker. When a notification arrives, warms
`ScoreCache` for every band at that `valid_time` (reading from the
shared NFS `/data/scores/<band>/<iso>.ntms` tree that Rust just
shared NFS `/data/scores/<band>/<iso>.prop` tree that Rust just
wrote) and fans out the standard `"propagation:updated"` PubSub so
LiveView clients refresh without polling.

View file

@ -16,7 +16,7 @@ defmodule Microwaveprop.Propagation.ScoreCacheReconciler do
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`
Every `interval_ms` the reconciler lists `/data/scores/<band>/*.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

View file

@ -7,7 +7,7 @@ defmodule Microwaveprop.Propagation.ScoresFile do
## Layout
magic : 4 bytes "NTMS"
magic : 4 bytes "PROP"
version : 1 byte 0x01
band_mhz : 4 bytes uint32 little-endian
valid_time : 8 bytes int64 unix seconds little-endian
@ -35,7 +35,11 @@ defmodule Microwaveprop.Propagation.ScoresFile do
require Logger
@magic "NTMS"
@magic "PROP"
# Legacy magic from before the `.prop`/`PROP` rename. Still accepted
# by readers so score files written by an older pod remain usable
# across a rolling deploy. Writers always emit `@magic`.
@legacy_magic "NTMS"
@version 1
@no_data 255
@header_size 33
@ -49,6 +53,13 @@ defmodule Microwaveprop.Propagation.ScoresFile do
@doc "Returns the full path for a `(band_mhz, valid_time)` score file."
@spec path_for(non_neg_integer(), DateTime.t()) :: String.t()
def path_for(band_mhz, %DateTime{} = valid_time) when is_integer(band_mhz) do
iso = valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601()
Path.join([base_dir(), Integer.to_string(band_mhz), "#{iso}.prop"])
end
@doc false
@spec legacy_path_for(non_neg_integer(), DateTime.t()) :: String.t()
def legacy_path_for(band_mhz, %DateTime{} = valid_time) when is_integer(band_mhz) do
iso = valid_time |> DateTime.truncate(:second) |> DateTime.to_iso8601()
Path.join([base_dir(), Integer.to_string(band_mhz), "#{iso}.ntms"])
end
@ -84,6 +95,14 @@ defmodule Microwaveprop.Propagation.ScoresFile do
{:ok, map()} | {:error, :enoent | :invalid_format}
def read(band_mhz, %DateTime{} = valid_time) when is_integer(band_mhz) do
case File.read(path_for(band_mhz, valid_time)) do
{:ok, binary} -> decode(binary)
{:error, :enoent} -> read_legacy(band_mhz, valid_time)
{:error, other} -> {:error, other}
end
end
defp read_legacy(band_mhz, valid_time) do
case File.read(legacy_path_for(band_mhz, valid_time)) do
{:ok, binary} -> decode(binary)
{:error, :enoent} -> {:error, :enoent}
{:error, other} -> {:error, other}
@ -93,8 +112,8 @@ defmodule Microwaveprop.Propagation.ScoresFile do
@doc """
Delete every score file whose `valid_time` is strictly before
`cutoff`. Returns the number of files deleted. Foreign files in the
scores tree (anything that doesn't parse as `<iso>.ntms`) are left
alone.
scores tree (anything that doesn't parse as `<iso>.prop` or the
legacy `<iso>.ntms`) are left alone.
"""
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
def prune_older_than(%DateTime{} = cutoff) do
@ -236,8 +255,13 @@ defmodule Microwaveprop.Propagation.ScoresFile do
"""
@spec read_point(non_neg_integer(), DateTime.t(), float(), float()) :: non_neg_integer() | nil
def read_point(band_mhz, %DateTime{} = valid_time, lat, lon) do
path = path_for(band_mhz, valid_time)
Enum.find_value(
[path_for(band_mhz, valid_time), legacy_path_for(band_mhz, valid_time)],
&open_and_read_point(&1, lat, lon)
)
end
defp open_and_read_point(path, lat, lon) do
case :file.open(path, [:read, :binary, :raw]) do
{:ok, fd} ->
try do
@ -252,13 +276,14 @@ defmodule Microwaveprop.Propagation.ScoresFile do
end
defp read_point_from_fd(fd, lat, lon) do
# Peek the header to locate the cell — falls back to the full
# decode path only if the header is malformed (should never happen
# for files we wrote ourselves).
# Peek the header to locate the cell. Accepts both the current and
# legacy 4-byte magic — see `@legacy_magic`.
case :file.pread(fd, 0, @header_size) do
{:ok,
<<@magic, @version::unsigned-8, _band::unsigned-little-32, _vt::signed-little-64, lat_min::float-little-32,
lon_min::float-little-32, step::float-little-32, n_rows::unsigned-little-16, n_cols::unsigned-little-16>>} ->
<<magic::binary-size(4), @version::unsigned-8, _band::unsigned-little-32, _vt::signed-little-64,
lat_min::float-little-32, lon_min::float-little-32, step::float-little-32, n_rows::unsigned-little-16,
n_cols::unsigned-little-16>>}
when magic == @magic or magic == @legacy_magic ->
read_cell(fd, lat, lon, lat_min, lon_min, step, n_rows, n_cols)
_ ->
@ -328,10 +353,11 @@ defmodule Microwaveprop.Propagation.ScoresFile do
# ── Decoding ────────────────────────────────────────────────────
defp decode(
<<@magic, @version::unsigned-8, band_mhz::unsigned-little-32, valid_time_unix::signed-little-64,
<<magic::binary-size(4), @version::unsigned-8, band_mhz::unsigned-little-32, valid_time_unix::signed-little-64,
lat_min::float-little-32, lon_min::float-little-32, step::float-little-32, n_rows::unsigned-little-16,
n_cols::unsigned-little-16, body::binary>>
) do
)
when magic == @magic or magic == @legacy_magic do
expected = n_rows * n_cols
if byte_size(body) == expected do
@ -433,7 +459,7 @@ defmodule Microwaveprop.Propagation.ScoresFile do
end
defp parse_valid_time_dt(filename) do
with [_, iso] <- Regex.run(~r/^(.+)\.ntms$/, filename),
with [_, iso] <- Regex.run(~r/^(.+)\.(?:prop|ntms)$/, filename),
{:ok, dt, _} <- DateTime.from_iso8601(iso) do
dt
else
@ -468,7 +494,7 @@ defmodule Microwaveprop.Propagation.ScoresFile do
end
defp parse_valid_time(filename) do
with [_, iso] <- Regex.run(~r/^(.+)\.ntms$/, filename),
with [_, iso] <- Regex.run(~r/^(.+)\.(?:prop|ntms)$/, filename),
{:ok, dt, _} <- DateTime.from_iso8601(iso) do
DateTime.to_unix(dt)
else

View file

@ -142,7 +142,7 @@ defmodule Microwaveprop.Workers.GefsFetchWorker do
end
# Runs the propagation scorer across the fetched GEFS grid and
# persists scores to the `.ntms` file for this `valid_time` so the
# persists scores to the `.prop` file for this `valid_time` so the
# extended-horizon outlook picks them up via the usual
# `Propagation.scores_at/3` path. Also upserts raw profiles to
# `gefs_profiles` for future per-contact extended-horizon lookups.

View file

@ -408,7 +408,7 @@ defmodule MicrowavepropWeb.MapLive do
closest_to_now(valid_times)
end
# scores_at_fresh bypasses the cache and re-reads the .ntms file so
# scores_at_fresh bypasses the cache and re-reads the .prop file so
# the map reflects the newly written forecast hour even when the
# ScoreCache still holds the previous chain's bytes. The propagation
# update arrives via `propagation:updated` while the cache refresh

View file

@ -1,6 +1,6 @@
//! End-to-end f01..f18 chain step: fetch surface + pressure GRIBs, decode,
//! build per-cell `Conditions`, score every band, write score-grid files
//! (`<band>/<iso>.ntms`) to the shared scores directory.
//! (`<band>/<iso>.prop`) to the shared scores directory.
//!
//! This is the Rust equivalent of `PropagationGridWorker.process_forecast_hour/4`
//! minus the f00-only enrichment (native duct, NEXRAD, commercial link

View file

@ -2,7 +2,7 @@
//! `lib/microwaveprop/propagation/scores_file.ex`.
//!
//! Layout (all little-endian):
//! magic : 4 bytes "NTMS" (literal already on disk)
//! magic : 4 bytes "PROP" (decoder also accepts legacy "NTMS")
//! version : 1 byte 0x01
//! band_mhz : 4 bytes uint32
//! valid_time : 8 bytes int64 unix seconds
@ -27,14 +27,26 @@ use chrono::{DateTime, Utc};
use crate::grid;
pub const MAGIC: &[u8; 4] = b"NTMS";
pub const MAGIC: &[u8; 4] = b"PROP";
pub const LEGACY_MAGIC: &[u8; 4] = b"NTMS";
pub const VERSION: u8 = 1;
pub const NO_DATA: u8 = 255;
static UNIQUE: AtomicU64 = AtomicU64::new(1);
/// Absolute path `<base>/<band_mhz>/<iso>.ntms`.
/// Absolute path `<base>/<band_mhz>/<iso>.prop`. New writes use this;
/// readers additionally accept the legacy `.ntms` extension while
/// previously-written files drain through the normal retention window.
pub fn path_for(base_dir: &Path, band_mhz: u32, valid_time: DateTime<Utc>) -> PathBuf {
let iso = valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string();
base_dir
.join(band_mhz.to_string())
.join(format!("{iso}.prop"))
}
/// Legacy path used by scores written before the `.prop` rename. Kept so
/// readers can discover still-on-disk `.ntms` files during rollout.
pub fn legacy_path_for(base_dir: &Path, band_mhz: u32, valid_time: DateTime<Utc>) -> PathBuf {
let iso = valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string();
base_dir
.join(band_mhz.to_string())
@ -99,7 +111,7 @@ pub fn encode(
Ok(out)
}
/// Atomic write to `<base>/<band_mhz>/<iso>.ntms`. Creates the band
/// Atomic write to `<base>/<band_mhz>/<iso>.prop`. Creates the band
/// subdirectory on demand.
pub fn write_atomic(
base_dir: &Path,
@ -119,7 +131,7 @@ pub fn write_atomic(
.map(|d| d.as_nanos())
.unwrap_or(0);
let uniq = UNIQUE.fetch_add(1, Ordering::Relaxed);
let tmp_path = final_path.with_extension(format!("ntms.tmp.{nanos}.{uniq}"));
let tmp_path = final_path.with_extension(format!("prop.tmp.{nanos}.{uniq}"));
{
let file = OpenOptions::new()
@ -165,7 +177,7 @@ pub fn decode(bytes: &[u8]) -> Option<Decoded> {
if bytes.len() < 4 + 1 + 4 + 8 + 4 + 4 + 4 + 2 + 2 {
return None;
}
if &bytes[0..4] != MAGIC {
if &bytes[0..4] != MAGIC && &bytes[0..4] != LEGACY_MAGIC {
return None;
}
if bytes[4] != VERSION {
@ -266,12 +278,47 @@ mod tests {
fn path_for_layout() {
let ts = Utc.with_ymd_and_hms(2026, 4, 19, 15, 0, 0).unwrap();
let p = path_for(Path::new("/data/scores"), 10_000, ts);
assert_eq!(
p.to_str().unwrap(),
"/data/scores/10000/2026-04-19T15:00:00Z.prop"
);
}
#[test]
fn legacy_path_for_layout_matches_pre_rename() {
let ts = Utc.with_ymd_and_hms(2026, 4, 19, 15, 0, 0).unwrap();
let p = legacy_path_for(Path::new("/data/scores"), 10_000, ts);
assert_eq!(
p.to_str().unwrap(),
"/data/scores/10000/2026-04-19T15:00:00Z.ntms"
);
}
#[test]
fn decode_accepts_legacy_magic() {
// Flip the 4-byte magic on an otherwise-valid encoded blob and
// verify decode still accepts it. Readers must keep tolerating
// `NTMS`-prefixed files written before the rename.
let ts = Utc.with_ymd_and_hms(2026, 4, 19, 15, 0, 0).unwrap();
let scores = vec![ScorePoint {
lat: 32.0,
lon: -97.0,
score: 73,
}];
let mut bytes = encode(10_000, ts, &scores).unwrap();
bytes[0..4].copy_from_slice(LEGACY_MAGIC);
let decoded = decode(&bytes).expect("legacy magic must decode");
assert_eq!(decoded.band_mhz, 10_000);
}
#[test]
fn decode_rejects_unknown_magic() {
let ts = Utc.with_ymd_and_hms(2026, 4, 19, 15, 0, 0).unwrap();
let mut bytes = encode(10_000, ts, &[]).unwrap();
bytes[0..4].copy_from_slice(b"XXXX");
assert!(decode(&bytes).is_none());
}
#[test]
fn atomic_write_and_decode() {
let dir = tempfile::tempdir().unwrap();

View file

@ -35,10 +35,87 @@ defmodule Microwaveprop.Propagation.ScoresFileTest do
describe "path_for/2" do
test "nests files under base_dir/{band_mhz}/ with an ISO-8601 valid_time", %{dir: dir} do
path = ScoresFile.path_for(10_000, ~U[2026-04-14 18:00:00Z])
assert path == Path.join([dir, "10000", "2026-04-14T18:00:00Z.prop"])
end
test "writes with the new .prop extension", %{dir: dir} do
valid_time = ~U[2026-04-14 18:00:00Z]
ScoresFile.write!(10_000, valid_time, [%{lat: 25.0, lon: -125.0, score: 50}])
written = Path.join([dir, "10000", "2026-04-14T18:00:00Z.prop"])
assert File.exists?(written)
refute File.exists?(Path.join([dir, "10000", "2026-04-14T18:00:00Z.ntms"]))
end
end
describe "legacy_path_for/2" do
test "still points at the old .ntms extension for backwards reads", %{dir: dir} do
path = ScoresFile.legacy_path_for(10_000, ~U[2026-04-14 18:00:00Z])
assert path == Path.join([dir, "10000", "2026-04-14T18:00:00Z.ntms"])
end
end
describe "legacy .ntms file readability" do
test "read/2 falls through to the legacy .ntms path when .prop is missing", %{dir: dir} do
valid_time = ~U[2026-04-14 18:00:00Z]
# Write a valid file at the .prop path, then rename to .ntms to
# simulate a file written before the rename that survived through
# a rolling deploy. The decoder must accept the old magic bytes
# embedded in the header of already-on-disk files as well, but the
# simpler check — that the file is found by extension — is what
# this test targets.
ScoresFile.write!(10_000, valid_time, [%{lat: 25.0, lon: -125.0, score: 77}])
prop_path = Path.join([dir, "10000", "2026-04-14T18:00:00Z.prop"])
ntms_path = Path.join([dir, "10000", "2026-04-14T18:00:00Z.ntms"])
File.rename!(prop_path, ntms_path)
assert {:ok, payload} = ScoresFile.read(10_000, valid_time)
assert payload.band_mhz == 10_000
end
test "read_point/4 accepts a legacy .ntms file with NTMS magic", %{dir: dir} do
valid_time = ~U[2026-04-14 18:00:00Z]
ScoresFile.write!(10_000, valid_time, [%{lat: 25.0, lon: -125.0, score: 88}])
prop_path = Path.join([dir, "10000", "2026-04-14T18:00:00Z.prop"])
ntms_path = Path.join([dir, "10000", "2026-04-14T18:00:00Z.ntms"])
# Flip the 4-byte magic back to the legacy "NTMS" so both the
# extension and the in-file identifier reflect the pre-rename
# state. The reader must still serve the cell.
raw = File.read!(prop_path)
<<_magic::binary-size(4), rest::binary>> = raw
File.write!(ntms_path, <<"NTMS", rest::binary>>, [:binary])
File.rm!(prop_path)
assert ScoresFile.read_point(10_000, valid_time, 25.0, -125.0) == 88
end
test "list_valid_times/1 picks up .ntms and .prop files together", %{dir: _dir} do
t1 = ~U[2026-04-14 17:00:00Z]
t2 = ~U[2026-04-14 18:00:00Z]
ScoresFile.write!(10_000, t1, [%{lat: 25.0, lon: -125.0, score: 10}])
ScoresFile.write!(10_000, t2, [%{lat: 25.0, lon: -125.0, score: 20}])
# Rename the earlier one to .ntms to simulate a mixed-extension
# directory during a rolling deploy.
band_dir =
Path.join([
Application.fetch_env!(:microwaveprop, :propagation_scores_dir),
"10000"
])
File.rename!(
Path.join(band_dir, "2026-04-14T17:00:00Z.prop"),
Path.join(band_dir, "2026-04-14T17:00:00Z.ntms")
)
times = ScoresFile.list_valid_times(10_000)
assert length(times) == 2
assert t1 in times
assert t2 in times
end
end
describe "write!/3 + read/2 round trip" do
test "preserves the score for each scored grid point", %{dir: _dir} do
valid_time = ~U[2026-04-14 18:00:00Z]

View file

@ -460,7 +460,7 @@ defmodule MicrowavepropWeb.MapLiveTest do
# assertion below observes the propagation_updated delivery only.
assert_push_event(lv, "update_scores", %{scores: _initial_scores})
# Simulate a new forecast-hour compute: rewrite the .ntms file on
# Simulate a new forecast-hour compute: rewrite the .prop file on
# disk to the updated score, but do *not* touch the cache. This
# reproduces the race where the cache_refresh message has not yet
# been applied by the time MapLive handles propagation_updated.