diff --git a/docs/runbook_propagation_pipeline.md b/docs/runbook_propagation_pipeline.md index 789d0902..b09cb67c 100644 --- a/docs/runbook_propagation_pipeline.md +++ b/docs/runbook_propagation_pipeline.md @@ -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 ''`. 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 diff --git a/lib/microwaveprop/propagation/notify_listener.ex b/lib/microwaveprop/propagation/notify_listener.ex index 9ba12cf7..dfcf1608 100644 --- a/lib/microwaveprop/propagation/notify_listener.ex +++ b/lib/microwaveprop/propagation/notify_listener.ex @@ -3,7 +3,7 @@ defmodule Microwaveprop.Propagation.NotifyListener do Listens for `NOTIFY propagation_ready ''` 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//.ntms` tree that Rust just + shared NFS `/data/scores//.prop` tree that Rust just wrote) and fans out the standard `"propagation:updated"` PubSub so LiveView clients refresh without polling. diff --git a/lib/microwaveprop/propagation/score_cache_reconciler.ex b/lib/microwaveprop/propagation/score_cache_reconciler.ex index 39798be9..04016291 100644 --- a/lib/microwaveprop/propagation/score_cache_reconciler.ex +++ b/lib/microwaveprop/propagation/score_cache_reconciler.ex @@ -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//*.ntms` + 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 diff --git a/lib/microwaveprop/propagation/scores_file.ex b/lib/microwaveprop/propagation/scores_file.ex index f450b5a2..1a78ff39 100644 --- a/lib/microwaveprop/propagation/scores_file.ex +++ b/lib/microwaveprop/propagation/scores_file.ex @@ -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 `.ntms`) are left - alone. + scores tree (anything that doesn't parse as `.prop` or the + legacy `.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>>} -> + <>} + 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, + <> - ) 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 diff --git a/lib/microwaveprop/workers/gefs_fetch_worker.ex b/lib/microwaveprop/workers/gefs_fetch_worker.ex index 78f20f9c..a998426d 100644 --- a/lib/microwaveprop/workers/gefs_fetch_worker.ex +++ b/lib/microwaveprop/workers/gefs_fetch_worker.ex @@ -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. diff --git a/lib/microwaveprop_web/live/map_live.ex b/lib/microwaveprop_web/live/map_live.ex index 9f7b6cfb..ed1eee5e 100644 --- a/lib/microwaveprop_web/live/map_live.ex +++ b/lib/microwaveprop_web/live/map_live.ex @@ -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 diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index 62c16879..4eb768ee 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -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 -//! (`/.ntms`) to the shared scores directory. +//! (`/.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 diff --git a/rust/prop_grid_rs/src/scores_file.rs b/rust/prop_grid_rs/src/scores_file.rs index 63918cb4..9560b12e 100644 --- a/rust/prop_grid_rs/src/scores_file.rs +++ b/rust/prop_grid_rs/src/scores_file.rs @@ -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 `//.ntms`. +/// Absolute path `//.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) -> 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) -> 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 `//.ntms`. Creates the band +/// Atomic write to `//.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 { 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(); diff --git a/test/microwaveprop/propagation/scores_file_test.exs b/test/microwaveprop/propagation/scores_file_test.exs index 4b4ab42d..564dbd64 100644 --- a/test/microwaveprop/propagation/scores_file_test.exs +++ b/test/microwaveprop/propagation/scores_file_test.exs @@ -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] diff --git a/test/microwaveprop_web/live/map_live_test.exs b/test/microwaveprop_web/live/map_live_test.exs index f078e746..4e4c06d7 100644 --- a/test/microwaveprop_web/live/map_live_test.exs +++ b/test/microwaveprop_web/live/map_live_test.exs @@ -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.