Graham McIntire
bd3b1148a4
Clicks on the map were hanging for 1-3s, triggering the Phoenix topbar progress bar visible to the user. Three hot-path improvements: 1. ScoresFile.read_point now uses a two-range :file.pread/2 — reads the 33-byte header + one cell byte (~60 bytes) instead of the full ~93 KB file. point_forecast walks 19 forecast hours per click, so the old full-read path was the dominant cost on cold cache. 2. ProfilesFile.read/1 and list_valid_times/0 cached via Microwaveprop.Cache (5s TTL). Decoded profile maps are ~10 MB each (92k cells); timeline scrub that re-clicks the same valid_time within seconds now hits ETS instead of re-gunzipping + decoding. Cache keys include base_dir so test setup that swaps dirs doesn't see stale entries. Writes + prune + retain_window invalidate. 3. Endpoint log_level now filters /metrics the same way it already filters /health — Prometheus scrapes every 5s and was producing visible log spam. Pod-local ETS is right for this workload (per-click read, tiny working set); Valkey / shared memstore would not help here since each pod needs its own fast-path lookup. File a follow-up if cross-pod score lookups ever show up in flame graphs.
487 lines
16 KiB
Elixir
487 lines
16 KiB
Elixir
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.<uniq>` → 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
|
||
@header_size 33
|
||
|
||
@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)
|
||
invalidate_list_cache(band_mhz)
|
||
: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 `<iso>.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)
|
||
|
||
deleted =
|
||
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)
|
||
|
||
if deleted > 0, do: invalidate_all_list_caches()
|
||
deleted
|
||
end
|
||
|
||
@doc """
|
||
Keep only the files whose `valid_time` falls inside the closed
|
||
window `[run_time, run_time + max_forecast_hour * 3600]`. Returns
|
||
the number of files deleted.
|
||
|
||
Called at the end of a `PropagationGridWorker` chain so any files
|
||
left over from the previous run_time (that the new chain didn't
|
||
overwrite by coincidence of valid_time) get cleaned up immediately
|
||
instead of waiting for the 2h prune cron.
|
||
"""
|
||
@spec retain_window(DateTime.t(), non_neg_integer()) :: non_neg_integer()
|
||
def retain_window(%DateTime{} = run_time, max_forecast_hour) when max_forecast_hour >= 0 do
|
||
lo = DateTime.to_unix(run_time)
|
||
hi = lo + max_forecast_hour * 3600
|
||
|
||
deleted =
|
||
base_dir()
|
||
|> list_score_files()
|
||
|> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
|
||
if valid_time_unix < lo or valid_time_unix > hi do
|
||
_ = File.rm(path)
|
||
acc + 1
|
||
else
|
||
acc
|
||
end
|
||
end)
|
||
|
||
if deleted > 0, do: invalidate_all_list_caches()
|
||
deleted
|
||
end
|
||
|
||
@doc """
|
||
List every `valid_time` that has a score file on disk for
|
||
`band_mhz`, ordered ascending. Empty list if the band directory
|
||
doesn't exist yet.
|
||
|
||
Cached for 5 seconds via `Microwaveprop.Cache`. The caller set is
|
||
tight (map click → `point_forecast` + `available_valid_times`;
|
||
hourly chain completion triggers a PubSub fan-out that'd flip this
|
||
TTL well within its useful lifetime) so sub-10s staleness is fine.
|
||
"""
|
||
@spec list_valid_times(non_neg_integer()) :: [DateTime.t()]
|
||
def list_valid_times(band_mhz) when is_integer(band_mhz) do
|
||
Microwaveprop.Cache.fetch_or_store({__MODULE__, :list_valid_times, base_dir(), band_mhz}, 5_000, fn ->
|
||
list_valid_times_raw(band_mhz)
|
||
end)
|
||
end
|
||
|
||
defp list_valid_times_raw(band_mhz) do
|
||
dir = Path.join(base_dir(), Integer.to_string(band_mhz))
|
||
|
||
case File.ls(dir) do
|
||
{:ok, files} ->
|
||
files
|
||
|> Enum.map(&parse_valid_time_dt/1)
|
||
|> Enum.reject(&is_nil/1)
|
||
|> Enum.sort(DateTime)
|
||
|
||
_ ->
|
||
[]
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Return the most recent `valid_time` across all bands on disk, or
|
||
`nil` if the scores tree is empty.
|
||
"""
|
||
@spec latest_valid_time() :: DateTime.t() | nil
|
||
def latest_valid_time do
|
||
case File.ls(base_dir()) do
|
||
{:ok, band_dirs} ->
|
||
band_dirs
|
||
|> Enum.flat_map(&band_times_or_empty/1)
|
||
|> max_datetime()
|
||
|
||
_ ->
|
||
nil
|
||
end
|
||
end
|
||
|
||
defp band_times_or_empty(band_dir) do
|
||
case Integer.parse(band_dir) do
|
||
{band_mhz, ""} -> list_valid_times(band_mhz)
|
||
_ -> []
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Read the file for `(band_mhz, valid_time)` and return every
|
||
scored cell as a list of `%{lat, lon, score}` maps. Optional
|
||
`bounds` map (`%{"north", "south", "east", "west"}` or
|
||
`%{north:, south:, east:, west:}`) filters the return to cells
|
||
inside that box. Cells with the no-data sentinel are excluded.
|
||
|
||
Returns `[]` when the file doesn't exist or is unreadable —
|
||
callers should treat this as "no scores available for this
|
||
band/time yet" rather than an error.
|
||
"""
|
||
@spec read_bounds(non_neg_integer(), DateTime.t(), map() | nil) ::
|
||
[%{lat: float(), lon: float(), score: non_neg_integer()}]
|
||
def read_bounds(band_mhz, %DateTime{} = valid_time, bounds \\ nil) do
|
||
case read(band_mhz, valid_time) do
|
||
{:ok, payload} -> extract_points(payload, bounds)
|
||
_ -> []
|
||
end
|
||
end
|
||
|
||
@doc """
|
||
Fetch the score for a single grid cell from the file for
|
||
`(band_mhz, valid_time)`. Returns `nil` if the file is missing or
|
||
if the cell is outside the grid / carries the no-data sentinel.
|
||
|
||
Reads only the 33-byte header plus the one cell byte via a single
|
||
multi-range `:file.pread/2` call — ~60 bytes off NFS instead of the
|
||
full ~93 KB file. Point-forecast walks 19 forecast hours per map
|
||
click, so the old full-read path dominated click latency whenever
|
||
the in-memory `ScoreCache` didn't already hold the cell.
|
||
"""
|
||
@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)
|
||
|
||
case :file.open(path, [:read, :binary, :raw]) do
|
||
{:ok, fd} ->
|
||
try do
|
||
read_point_from_fd(fd, lat, lon)
|
||
after
|
||
:file.close(fd)
|
||
end
|
||
|
||
{:error, _} ->
|
||
nil
|
||
end
|
||
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).
|
||
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>>} ->
|
||
read_cell(fd, lat, lon, lat_min, lon_min, step, n_rows, n_cols)
|
||
|
||
_ ->
|
||
nil
|
||
end
|
||
end
|
||
|
||
defp read_cell(fd, lat, lon, lat_min, lon_min, step, n_rows, n_cols) do
|
||
row = round((lat - lat_min) / step)
|
||
col = round((lon - lon_min) / step)
|
||
|
||
if row in 0..(n_rows - 1) and col in 0..(n_cols - 1) do
|
||
case :file.pread(fd, @header_size + row * n_cols + col, 1) do
|
||
{:ok, <<@no_data>>} -> nil
|
||
{:ok, <<score>>} -> score
|
||
_ -> nil
|
||
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)
|
||
<<score_byte(Map.get(lookup, {lat, lon}))::unsigned-8>>
|
||
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}
|
||
|
||
# ── Read helpers ────────────────────────────────────────────────
|
||
|
||
defp extract_points(%{scores: body} = payload, bounds) do
|
||
%{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols} = payload
|
||
{row_lo, row_hi, col_lo, col_hi} = bounds_to_index_range(payload, bounds)
|
||
|
||
for row <- row_lo..row_hi,
|
||
col <- col_lo..col_hi,
|
||
byte = :binary.at(body, row * n_cols + col),
|
||
byte != @no_data do
|
||
# guard against the ignored variables warning
|
||
_ = n_rows
|
||
lat = Float.round(lat_min + row * step, 3)
|
||
lon = Float.round(lon_min + col * step, 3)
|
||
%{lat: lat, lon: lon, score: byte}
|
||
end
|
||
end
|
||
|
||
defp bounds_to_index_range(payload, nil) do
|
||
{0, payload.n_rows - 1, 0, payload.n_cols - 1}
|
||
end
|
||
|
||
defp bounds_to_index_range(payload, bounds) do
|
||
%{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols} = payload
|
||
|
||
south = fetch_bound(bounds, :south, "south", lat_min)
|
||
north = fetch_bound(bounds, :north, "north", lat_min + (n_rows - 1) * step)
|
||
west = fetch_bound(bounds, :west, "west", lon_min)
|
||
east = fetch_bound(bounds, :east, "east", lon_min + (n_cols - 1) * step)
|
||
|
||
row_lo = clamp(trunc((south - lat_min) / step), 0, n_rows - 1)
|
||
row_hi = clamp(ceil((north - lat_min) / step), 0, n_rows - 1)
|
||
col_lo = clamp(trunc((west - lon_min) / step), 0, n_cols - 1)
|
||
col_hi = clamp(ceil((east - lon_min) / step), 0, n_cols - 1)
|
||
|
||
{min(row_lo, row_hi), max(row_lo, row_hi), min(col_lo, col_hi), max(col_lo, col_hi)}
|
||
end
|
||
|
||
defp fetch_bound(bounds, atom_key, string_key, default) do
|
||
case Map.get(bounds, atom_key) || Map.get(bounds, string_key) do
|
||
nil -> default
|
||
v when is_number(v) -> v
|
||
v when is_binary(v) -> String.to_float(v)
|
||
end
|
||
end
|
||
|
||
defp clamp(n, lo, _hi) when n < lo, do: lo
|
||
defp clamp(n, _lo, hi) when n > hi, do: hi
|
||
defp clamp(n, _lo, _hi), do: n
|
||
|
||
defp point_score(payload, lat, lon) do
|
||
%{lat_min: lat_min, lon_min: lon_min, step: step, n_rows: n_rows, n_cols: n_cols, scores: body} = payload
|
||
row = round((lat - lat_min) / step)
|
||
col = round((lon - lon_min) / step)
|
||
|
||
if row in 0..(n_rows - 1) and col in 0..(n_cols - 1) do
|
||
case :binary.at(body, row * n_cols + col) do
|
||
@no_data -> nil
|
||
score -> score
|
||
end
|
||
end
|
||
end
|
||
|
||
defp max_datetime([]), do: nil
|
||
defp max_datetime(times), do: Enum.max(times, DateTime)
|
||
|
||
defp invalidate_list_cache(band_mhz) do
|
||
Microwaveprop.Cache.invalidate({__MODULE__, :list_valid_times, base_dir(), band_mhz})
|
||
end
|
||
|
||
defp invalidate_all_list_caches do
|
||
# Cheaper than enumerating bands — clear every cached list entry.
|
||
# `Microwaveprop.Cache.clear/0` is overkill (it wipes unrelated
|
||
# keys too) so match-delete just this module's keys.
|
||
:ets.match_delete(:microwaveprop_cache, {{__MODULE__, :list_valid_times, :_, :_}, :_, :_})
|
||
:ok
|
||
end
|
||
|
||
defp parse_valid_time_dt(filename) do
|
||
with [_, iso] <- Regex.run(~r/^(.+)\.ntms$/, filename),
|
||
{:ok, dt, _} <- DateTime.from_iso8601(iso) do
|
||
dt
|
||
else
|
||
_ -> nil
|
||
end
|
||
end
|
||
|
||
# ── 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
|