From 690a3523e244e74a6d6280981fa031f393b538b7 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 14 Apr 2026 14:36:55 -0500 Subject: [PATCH] Dual-write propagation scores to binary files on /data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First step of the disk-backed scores migration from the DuckDB plan doc. Ended up shipping the raw-binary variant instead of Parquet because the data is disposable after ~2h — the ecosystem benefits of Parquet only pay off for long-lived datasets, and the binary path has zero new dependencies. Microwaveprop.Propagation.ScoresFile writes one file per (band_mhz, valid_time) tuple under the configured scores dir, default /data/scores in prod and priv/dev_scores in dev. Layout is a 33-byte header plus a dense n_rows × n_cols uint8 array (255 is the no-data sentinel). The whole CONUS grid serializes to ~93 KB per band, and writes use the temp-then-rename pattern so NFSv4 concurrent readers never see a partial file. Propagation.replace_scores/2 now materializes the score stream once and dual-writes: Postgres on the primary path, then one ScoresFile per band as a best-effort follow-up (any file error is logged but doesn't fail the DB write, so we can verify the file path in prod before cutting readers over). Propagation.prune_old_scores/0 also clears expired score files so the existing 15-minute prune cron covers both storage layers. Dev configuration points at priv/dev_scores/, added to .gitignore. Test configuration points at a per-run tmp directory. --- .gitignore | 3 + config/dev.exs | 5 + config/test.exs | 8 + lib/microwaveprop/propagation.ex | 88 +++++-- lib/microwaveprop/propagation/scores_file.ex | 231 ++++++++++++++++++ .../propagation/scores_file_test.exs | 135 ++++++++++ test/microwaveprop/propagation_test.exs | 22 ++ 7 files changed, 467 insertions(+), 25 deletions(-) create mode 100644 lib/microwaveprop/propagation/scores_file.ex create mode 100644 test/microwaveprop/propagation/scores_file_test.exs diff --git a/.gitignore b/.gitignore index 4e67ad03..904c6b63 100644 --- a/.gitignore +++ b/.gitignore @@ -44,6 +44,9 @@ microwaveprop-*.tar # GRIB2 test fixtures (large binary files, downloaded on-demand) /test/fixtures/grib2/*.grib2 +# Dev-only propagation score binaries written by PropagationGridWorker +/priv/dev_scores/ + # Dialyzer PLT files /priv/plts/ diff --git a/config/dev.exs b/config/dev.exs index 130703aa..78b5c909 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -98,6 +98,11 @@ config :microwaveprop, Oban, ]} ] +# Propagation score binary files land in a gitignored tree inside the +# project root in dev, so iex sessions / PropagationGridWorker runs +# drop files where the developer can poke at them. +config :microwaveprop, :propagation_scores_dir, Path.expand("priv/dev_scores", File.cwd!()) + # Enable dev routes for dashboard and mailbox config :microwaveprop, dev_routes: true diff --git a/config/test.exs b/config/test.exs index 3ad77019..331c7972 100644 --- a/config/test.exs +++ b/config/test.exs @@ -47,6 +47,14 @@ config :microwaveprop, MicrowavepropWeb.Endpoint, # Run Oban jobs inline during tests config :microwaveprop, Oban, testing: :inline + +# Score files land under a per-run tmp directory so replace_scores tests +# don't write into a shared location (and we can assert on the file +# contents from inside tests). +config :microwaveprop, + :propagation_scores_dir, + Path.join(System.tmp_dir!(), "microwaveprop_test_scores_#{System.unique_integer([:positive])}") + config :microwaveprop, cache_contact_count: false config :microwaveprop, cache_contact_map: false config :microwaveprop, elevation_req_options: [plug: {Req.Test, Microwaveprop.Terrain.ElevationClient}, retry: false] diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index da84467d..517d55ac 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -8,6 +8,7 @@ defmodule Microwaveprop.Propagation do alias Microwaveprop.Propagation.GridScore alias Microwaveprop.Propagation.ScoreCache alias Microwaveprop.Propagation.Scorer + alias Microwaveprop.Propagation.ScoresFile alias Microwaveprop.Repo alias Microwaveprop.Weather.SoundingParams @@ -167,33 +168,62 @@ defmodule Microwaveprop.Propagation do @spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()} def replace_scores(scores, %DateTime{} = valid_time) do now = DateTime.truncate(DateTime.utc_now(), :second) + # Materialize once — we need to traverse twice: once for the DB + # insert, once for the per-band ScoresFile writer. 475k maps × + # ~150 B each is ~70 MB, well within the pod memory budget. + scores_list = Enum.to_list(scores) - Repo.transaction( - fn -> - Repo.delete_all(from(gs in GridScore, where: gs.valid_time == ^valid_time), timeout: 300_000) + result = + Repo.transaction( + fn -> + Repo.delete_all(from(gs in GridScore, where: gs.valid_time == ^valid_time), timeout: 300_000) - scores - |> Stream.map(fn s -> - %{ - id: Ecto.UUID.generate(), - lat: s.lat, - lon: s.lon, - valid_time: s.valid_time, - band_mhz: s.band_mhz, - score: s.score, - factors: s.factors, - inserted_at: now, - updated_at: now - } - end) - |> Stream.chunk_every(1000) - |> Enum.reduce(0, fn chunk, acc -> - {count, _} = Repo.insert_all(GridScore, chunk) - acc + count - end) - end, - timeout: 600_000 - ) + scores_list + |> Stream.map(fn s -> + %{ + id: Ecto.UUID.generate(), + lat: s.lat, + lon: s.lon, + valid_time: s.valid_time, + band_mhz: s.band_mhz, + score: s.score, + factors: s.factors, + inserted_at: now, + updated_at: now + } + end) + |> Stream.chunk_every(1000) + |> Enum.reduce(0, fn chunk, acc -> + {count, _} = Repo.insert_all(GridScore, chunk) + acc + count + end) + end, + timeout: 600_000 + ) + + case result do + {:ok, _count} -> write_score_files(scores_list, valid_time) + _error -> :ok + end + + result + end + + # Dual-write the grid to disk-backed ScoresFile binaries so the map + # render path can eventually read directly from /data/scores without + # touching Postgres. Best-effort — any file error is logged and the + # DB write still counts as the source of truth until the cutover. + defp write_score_files(scores, valid_time) do + scores + |> Enum.group_by(& &1.band_mhz) + |> Enum.each(fn {band_mhz, band_scores} -> + try do + ScoresFile.write!(band_mhz, valid_time, band_scores) + rescue + e -> + Logger.warning("Propagation: ScoresFile write failed for band=#{band_mhz} vt=#{valid_time}: #{inspect(e)}") + end + end) end @doc """ @@ -274,6 +304,14 @@ defmodule Microwaveprop.Propagation do if deleted > 0 do Logger.info("PropagationScores: pruned #{deleted} old scores (before #{cutoff})") end + + file_deleted = ScoresFile.prune_older_than(cutoff) + + if file_deleted > 0 do + Logger.info("PropagationScores: pruned #{file_deleted} old score files (before #{cutoff})") + end + + :ok end @doc """ diff --git a/lib/microwaveprop/propagation/scores_file.ex b/lib/microwaveprop/propagation/scores_file.ex new file mode 100644 index 00000000..c57a63cd --- /dev/null +++ b/lib/microwaveprop/propagation/scores_file.ex @@ -0,0 +1,231 @@ +defmodule Microwaveprop.Propagation.ScoresFile do + @moduledoc """ + Binary on-disk storage for propagation score grids. One file per + `(band_mhz, valid_time)` tuple, ~93 KB each. Written in the + forecast-hour hot path as a companion to the Postgres upsert, and + eventually replacing it. + + ## Layout + + magic : 4 bytes "NTMS" + version : 1 byte 0x01 + band_mhz : 4 bytes uint32 little-endian + valid_time : 8 bytes int64 unix seconds little-endian + lat_min : 4 bytes float32 little-endian + lon_min : 4 bytes float32 little-endian + step_deg : 4 bytes float32 little-endian + n_rows : 2 bytes uint16 little-endian (lat dimension) + n_cols : 2 bytes uint16 little-endian (lon dimension) + scores : n_rows × n_cols bytes (uint8, 0-100 or 255 = no-data) + + A cell at `(lat, lon)` is at byte offset + `row * n_cols + col` in the scores body, where + `row = (lat - lat_min) / step` and `col = (lon - lon_min) / step` + (both rounded to the nearest integer). + + ## Atomic writes + + Files are written through `rename(2)`: `path.tmp.` → fsync → + rename to the final path. On the same NFSv4 filesystem the rename + is atomic, so a concurrent reader either sees the old file, the + new file, or the absence of either — never a partial file. + """ + + alias Microwaveprop.Propagation.Grid + + require Logger + + @magic "NTMS" + @version 1 + @no_data 255 + + @doc "Returns the root directory score files are written to." + @spec base_dir() :: String.t() + def base_dir do + Application.get_env(:microwaveprop, :propagation_scores_dir, "/data/scores") + end + + @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}.ntms"]) + end + + @doc """ + Write a score grid to disk for `(band_mhz, valid_time)`. `scores` + must be a list of `%{lat, lon, score}` maps. Missing cells are + stored as the no-data sentinel (255). + """ + @spec write!(non_neg_integer(), DateTime.t(), [map()]) :: :ok + def write!(band_mhz, %DateTime{} = valid_time, scores) when is_integer(band_mhz) and is_list(scores) do + path = path_for(band_mhz, valid_time) + File.mkdir_p!(Path.dirname(path)) + + binary = encode(band_mhz, valid_time, scores) + + tmp = path <> ".tmp." <> unique_suffix() + File.write!(tmp, binary, [:binary]) + File.rename!(tmp, path) + :ok + end + + @doc """ + Read a score grid from disk. Returns: + + * `{:ok, %{band_mhz, valid_time, lat_min, lon_min, step, n_rows, n_cols, scores}}` + where `scores` is the raw binary (row-major, uint8 per cell) + * `{:error, :enoent}` when the file doesn't exist + * `{:error, :invalid_format}` when the file isn't a score grid + """ + @spec read(non_neg_integer(), DateTime.t()) :: + {: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} -> {:error, :enoent} + {:error, other} -> {:error, other} + end + end + + @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. + """ + @spec prune_older_than(DateTime.t()) :: non_neg_integer() + def prune_older_than(%DateTime{} = cutoff) do + cutoff_unix = DateTime.to_unix(cutoff) + + base_dir() + |> list_score_files() + |> Enum.reduce(0, fn {path, valid_time_unix}, acc -> + if valid_time_unix < cutoff_unix do + _ = File.rm(path) + acc + 1 + else + acc + end + end) + end + + # ── Encoding ──────────────────────────────────────────────────── + + defp encode(band_mhz, valid_time, scores) do + %{lat_min: lat_min, lat_max: lat_max, lon_min: lon_min, lon_max: lon_max} = Grid.bounds() + step = Grid.step() + n_rows = round((lat_max - lat_min) / step) + 1 + n_cols = round((lon_max - lon_min) / step) + 1 + + lookup = + Map.new(scores, fn s -> + {{Float.round(s.lat, 3), Float.round(s.lon, 3)}, s.score} + end) + + body = build_body(lookup, lat_min, lon_min, step, n_rows, n_cols) + valid_time_unix = valid_time |> DateTime.truncate(:second) |> DateTime.to_unix() + + << + @magic::binary, + @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 + >> + end + + defp build_body(lookup, lat_min, lon_min, step, n_rows, n_cols) do + for row <- 0..(n_rows - 1), + col <- 0..(n_cols - 1), + into: <<>> do + lat = Float.round(lat_min + row * step, 3) + lon = Float.round(lon_min + col * step, 3) + <> + end + end + + defp score_byte(nil), do: @no_data + defp score_byte(s) when is_integer(s) and s >= 0 and s <= 100, do: s + defp score_byte(s) when is_integer(s) and s > 100, do: 100 + defp score_byte(s) when is_integer(s) and s < 0, do: 0 + defp score_byte(_), do: @no_data + + # ── Decoding ──────────────────────────────────────────────────── + + defp decode( + <<@magic, @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 + expected = n_rows * n_cols + + if byte_size(body) == expected do + case DateTime.from_unix(valid_time_unix) do + {:ok, valid_time} -> + {:ok, + %{ + band_mhz: band_mhz, + valid_time: valid_time, + lat_min: lat_min, + lon_min: lon_min, + step: step, + n_rows: n_rows, + n_cols: n_cols, + scores: body + }} + + _ -> + {:error, :invalid_format} + end + else + {:error, :invalid_format} + end + end + + defp decode(_), do: {:error, :invalid_format} + + # ── Prune helpers ─────────────────────────────────────────────── + + defp list_score_files(root) do + case File.ls(root) do + {:ok, entries} -> + for entry <- entries, + band_path = Path.join(root, entry), + File.dir?(band_path), + file <- files_in(band_path), + valid_time_unix = parse_valid_time(file), + valid_time_unix != nil do + {Path.join(band_path, file), valid_time_unix} + end + + _ -> + [] + end + end + + defp files_in(dir) do + case File.ls(dir) do + {:ok, entries} -> entries + _ -> [] + end + end + + defp parse_valid_time(filename) do + with [_, iso] <- Regex.run(~r/^(.+)\.ntms$/, filename), + {:ok, dt, _} <- DateTime.from_iso8601(iso) do + DateTime.to_unix(dt) + else + _ -> nil + end + end + + defp unique_suffix do + "#{System.system_time(:nanosecond)}.#{:erlang.unique_integer([:positive])}" + end +end diff --git a/test/microwaveprop/propagation/scores_file_test.exs b/test/microwaveprop/propagation/scores_file_test.exs new file mode 100644 index 00000000..555e7f8d --- /dev/null +++ b/test/microwaveprop/propagation/scores_file_test.exs @@ -0,0 +1,135 @@ +defmodule Microwaveprop.Propagation.ScoresFileTest do + use ExUnit.Case, async: true + + alias Microwaveprop.Propagation.Grid + alias Microwaveprop.Propagation.ScoresFile + + setup do + dir = + Path.join( + System.tmp_dir!(), + "scores_file_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) + + on_exit(fn -> + File.rm_rf!(dir) + + if prev do + Application.put_env(:microwaveprop, :propagation_scores_dir, prev) + else + Application.delete_env(:microwaveprop, :propagation_scores_dir) + end + end) + + %{dir: dir} + end + + 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.ntms"]) + 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] + + scores = [ + %{lat: 25.0, lon: -125.0, score: 10}, + %{lat: 25.0, lon: -124.875, score: 20}, + %{lat: 25.125, lon: -125.0, score: 30} + ] + + ScoresFile.write!(10_000, valid_time, scores) + + assert {:ok, payload} = ScoresFile.read(10_000, valid_time) + assert payload.band_mhz == 10_000 + assert DateTime.compare(payload.valid_time, valid_time) == :eq + assert payload.lat_min == Grid.bounds().lat_min + assert payload.lon_min == Grid.bounds().lon_min + assert payload.step == Grid.step() + + # The three scored cells we wrote come back at their grid positions. + assert score_at(payload, 25.0, -125.0) == 10 + assert score_at(payload, 25.0, -124.875) == 20 + assert score_at(payload, 25.125, -125.0) == 30 + end + + test "unscored cells read back as 255 (no-data sentinel)" do + valid_time = ~U[2026-04-14 18:00:00Z] + ScoresFile.write!(10_000, valid_time, [%{lat: 25.0, lon: -125.0, score: 42}]) + + {:ok, payload} = ScoresFile.read(10_000, valid_time) + assert score_at(payload, 25.0, -125.0) == 42 + # Any cell we didn't write should be the no-data sentinel. + assert score_at(payload, 26.5, -100.0) == 255 + end + end + + describe "read/2 errors" do + test "returns {:error, :enoent} when no file exists for the band/time" do + assert {:error, :enoent} = ScoresFile.read(10_000, ~U[2026-04-14 18:00:00Z]) + end + + test "returns {:error, :invalid_format} for a file that isn't a score grid" do + path = ScoresFile.path_for(10_000, ~U[2026-04-14 18:00:00Z]) + File.mkdir_p!(Path.dirname(path)) + File.write!(path, "not a score file") + + assert {:error, :invalid_format} = ScoresFile.read(10_000, ~U[2026-04-14 18:00:00Z]) + end + end + + describe "atomic write" do + test "leaves no .tmp files behind after a successful write", %{dir: dir} do + ScoresFile.write!(10_000, ~U[2026-04-14 18:00:00Z], [%{lat: 25.0, lon: -125.0, score: 1}]) + tmp_files = find_files_matching(dir, ~r/\.tmp\./) + assert tmp_files == [] + end + end + + describe "prune_older_than/1" do + test "deletes files whose valid_time is strictly before the cutoff", %{dir: _dir} do + old = ~U[2026-04-14 10:00:00Z] + fresh = ~U[2026-04-14 18:00:00Z] + + ScoresFile.write!(10_000, old, [%{lat: 25.0, lon: -125.0, score: 1}]) + ScoresFile.write!(10_000, fresh, [%{lat: 25.0, lon: -125.0, score: 2}]) + + cutoff = ~U[2026-04-14 15:00:00Z] + assert ScoresFile.prune_older_than(cutoff) == 1 + + assert {:error, :enoent} = ScoresFile.read(10_000, old) + assert {:ok, _} = ScoresFile.read(10_000, fresh) + end + + test "leaves foreign files in the scores dir alone", %{dir: dir} do + stray = Path.join(dir, "some_other_file.txt") + File.write!(stray, "do not touch") + + assert ScoresFile.prune_older_than(~U[2099-01-01 00:00:00Z]) == 0 + assert File.exists?(stray) + end + end + + # Look up the score for (lat, lon) in the binary returned by read/2. + defp score_at(payload, lat, lon) do + row = round((lat - payload.lat_min) / payload.step) + col = round((lon - payload.lon_min) / payload.step) + idx = row * payload.n_cols + col + :binary.at(payload.scores, idx) + end + + defp find_files_matching(dir, pattern) do + dir + |> Path.join("**/*") + |> Path.wildcard() + |> Enum.filter(fn path -> File.regular?(path) and Regex.match?(pattern, path) end) + end +end diff --git a/test/microwaveprop/propagation_test.exs b/test/microwaveprop/propagation_test.exs index 88233af0..60755ee7 100644 --- a/test/microwaveprop/propagation_test.exs +++ b/test/microwaveprop/propagation_test.exs @@ -4,6 +4,7 @@ defmodule Microwaveprop.PropagationTest do alias Microwaveprop.Propagation alias Microwaveprop.Propagation.GridScore alias Microwaveprop.Propagation.ScoreCache + alias Microwaveprop.Propagation.ScoresFile setup do ScoreCache.clear() @@ -247,6 +248,27 @@ defmodule Microwaveprop.PropagationTest do assert {:ok, 2} = Propagation.replace_scores(scores, valid_time) assert [nil, nil] = Repo.all(from gs in GridScore, select: gs.factors) end + + test "dual-writes one ScoresFile per band alongside the DB insert" do + valid_time = ~U[2026-07-15 13:00:00Z] + + scores = [ + %{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 75, factors: nil}, + %{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 24_000, score: 60, factors: nil} + ] + + assert {:ok, 2} = Propagation.replace_scores(scores, valid_time) + + # One binary file per band lands on disk at the configured path. + assert {:ok, grid_10g} = + ScoresFile.read(10_000, valid_time) + + assert {:ok, grid_24g} = + ScoresFile.read(24_000, valid_time) + + assert grid_10g.band_mhz == 10_000 + assert grid_24g.band_mhz == 24_000 + end end describe "point_detail/4 with forecast-hour rows" do