Make the map timeline read .ntms files authoritatively
Two related fixes so the main map reliably picks up new binary propagation score files as soon as PropagationGridWorker writes them. 1. Propagation.available_valid_times/1 previously preferred ScoreCache over ScoresFile, using the cache as an index of what was available. The cache is a lazy ETS of whatever hours have been fetched or broadcast, which is a strict subset of what's on disk. A new forecast hour landing on disk while the cache was warm with older entries was invisible to the timeline until the cache happened to catch up. Read directly from ScoresFile so the disk store is the source of truth. 2. Add Propagation.scores_at_fresh/3 that always reads the .ntms file and overwrites the cache entry, and use it from MapLive's propagation_updated handler. PropagationGridWorker publishes the cache_refresh on `propagation:cache` and the timeline ping on `propagation:updated` as separate PubSub broadcasts, so by the time MapLive runs through scores_at the ScoreCache GenServer may not have processed the refresh yet — fetch_bounds then returns the previous chain's bytes. scores_at_fresh takes disk as the source of truth for the refresh path and warms the cache as a side effect so subsequent readers see the new data.
This commit is contained in:
parent
1a394fa7ae
commit
2d9e5a2d1f
4 changed files with 189 additions and 44 deletions
|
|
@ -201,25 +201,19 @@ defmodule Microwaveprop.Propagation do
|
|||
end
|
||||
|
||||
@doc """
|
||||
Returns distinct valid_times for a band, ordered ascending.
|
||||
Filters out times more than 1 hour in the past, but always includes
|
||||
the most recent valid_time so there's always data to display.
|
||||
Returns distinct valid_times for a band, ordered ascending. Always
|
||||
reads from the on-disk `ScoresFile` store — the `ScoreCache` only
|
||||
holds whatever hours have been fetched or broadcast, which can be a
|
||||
partial view of what's actually on disk, so using it as the source
|
||||
of truth for the timeline makes new forecast hours invisible until
|
||||
the cache happens to catch up. Filters out times more than 1 hour in
|
||||
the past, but always includes the most recent valid_time so there's
|
||||
always data to display.
|
||||
"""
|
||||
@spec available_valid_times(non_neg_integer()) :: [DateTime.t()]
|
||||
def available_valid_times(band_mhz) do
|
||||
cutoff = DateTime.add(DateTime.utc_now(), -3600, :second)
|
||||
|
||||
case ScoreCache.valid_times(band_mhz) do
|
||||
[] -> available_valid_times_from_store(band_mhz, cutoff)
|
||||
cached -> filter_fresh(cached, cutoff)
|
||||
end
|
||||
end
|
||||
|
||||
defp filter_fresh(times, cutoff) do
|
||||
Enum.filter(times, fn t -> DateTime.compare(t, cutoff) != :lt end)
|
||||
end
|
||||
|
||||
defp available_valid_times_from_store(band_mhz, cutoff) do
|
||||
case ScoresFile.list_valid_times(band_mhz) do
|
||||
[] -> []
|
||||
times -> filter_or_latest(times, cutoff)
|
||||
|
|
@ -261,15 +255,34 @@ defmodule Microwaveprop.Propagation do
|
|||
Enum.map(scores, &Map.put(&1, :valid_time, time))
|
||||
|
||||
:miss ->
|
||||
full = ScoresFile.read_bounds(band_mhz, time)
|
||||
ScoreCache.put(band_mhz, time, full)
|
||||
|
||||
full
|
||||
|> filter_bounds(bounds)
|
||||
|> Enum.map(&Map.put(&1, :valid_time, time))
|
||||
read_from_disk_and_cache(band_mhz, time, bounds)
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Variant of `scores_at/3` that always reads from the `.ntms` file on
|
||||
disk and overwrites the cache entry, rather than returning whatever
|
||||
the cache happens to hold. Use from update paths (the map's
|
||||
`propagation_updated` handler) where the underlying file has just
|
||||
been rewritten but the cache may still contain the previous chain's
|
||||
scores because of the race between `propagation:cache` fan-out and
|
||||
`propagation:updated` delivery.
|
||||
"""
|
||||
@spec scores_at_fresh(non_neg_integer(), DateTime.t(), map() | nil) ::
|
||||
[%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}]
|
||||
def scores_at_fresh(band_mhz, %DateTime{} = valid_time, bounds \\ nil) do
|
||||
read_from_disk_and_cache(band_mhz, valid_time, bounds)
|
||||
end
|
||||
|
||||
defp read_from_disk_and_cache(band_mhz, time, bounds) do
|
||||
full = ScoresFile.read_bounds(band_mhz, time)
|
||||
ScoreCache.put(band_mhz, time, full)
|
||||
|
||||
full
|
||||
|> filter_bounds(bounds)
|
||||
|> Enum.map(&Map.put(&1, :valid_time, time))
|
||||
end
|
||||
|
||||
@doc """
|
||||
Load the full CONUS score set for `{band_mhz, valid_time}` from the
|
||||
on-disk binary file and broadcast it to every `ScoreCache` in the
|
||||
|
|
|
|||
|
|
@ -269,7 +269,16 @@ defmodule MicrowavepropWeb.MapLive do
|
|||
closest_to_now(valid_times)
|
||||
end
|
||||
|
||||
scores = Propagation.scores_at(band, selected_time, socket.assigns.bounds)
|
||||
# scores_at_fresh bypasses the cache and re-reads the .ntms 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
|
||||
# rides `propagation:cache`; the two topics deliver independently
|
||||
# so reading through the cache here can race against stale data.
|
||||
scores =
|
||||
if selected_time,
|
||||
do: Propagation.scores_at_fresh(band, selected_time, socket.assigns.bounds),
|
||||
else: []
|
||||
|
||||
send(self(), :preload_forecast)
|
||||
|
||||
|
|
|
|||
|
|
@ -343,29 +343,8 @@ defmodule Microwaveprop.PropagationTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "available_valid_times/1 with cache" do
|
||||
test "returns cached valid_times without touching the filesystem when warm" do
|
||||
t1 = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
t2 = DateTime.add(t1, 3600, :second)
|
||||
|
||||
ScoreCache.put(10_000, t1, [])
|
||||
ScoreCache.put(10_000, t2, [])
|
||||
|
||||
assert Propagation.available_valid_times(10_000) == [t1, t2]
|
||||
end
|
||||
|
||||
test "filters cached times older than the 1-hour cutoff" do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
stale = DateTime.add(now, -7200, :second)
|
||||
fresh = DateTime.add(now, 1800, :second)
|
||||
|
||||
ScoreCache.put(10_000, stale, [])
|
||||
ScoreCache.put(10_000, fresh, [])
|
||||
|
||||
assert Propagation.available_valid_times(10_000) == [fresh]
|
||||
end
|
||||
|
||||
test "falls back to the on-disk store on cold cache" do
|
||||
describe "available_valid_times/1" do
|
||||
test "reads from the on-disk .ntms store" do
|
||||
valid_time = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
Propagation.replace_scores(
|
||||
|
|
@ -376,6 +355,100 @@ defmodule Microwaveprop.PropagationTest do
|
|||
assert ScoreCache.valid_times(10_000) == []
|
||||
assert Propagation.available_valid_times(10_000) == [valid_time]
|
||||
end
|
||||
|
||||
test "filters valid_times older than the 1-hour cutoff" do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
stale = DateTime.add(now, -7200, :second)
|
||||
fresh = DateTime.add(now, 1800, :second)
|
||||
|
||||
Propagation.replace_scores(
|
||||
[%{lat: 25.0, lon: -125.0, valid_time: stale, band_mhz: 10_000, score: 50, factors: nil}],
|
||||
stale
|
||||
)
|
||||
|
||||
Propagation.replace_scores(
|
||||
[%{lat: 25.0, lon: -125.0, valid_time: fresh, band_mhz: 10_000, score: 80, factors: nil}],
|
||||
fresh
|
||||
)
|
||||
|
||||
assert Propagation.available_valid_times(10_000) == [fresh]
|
||||
end
|
||||
|
||||
test "sees a newly written forecast hour even when the cache is warm with earlier times" do
|
||||
# Reproduces the race where PropagationGridWorker writes a new .ntms
|
||||
# file but MapLive's propagation_updated handler runs before the
|
||||
# ScoreCache has absorbed the cache_refresh broadcast. The cached
|
||||
# view alone is not enough — the timeline must pick up the new file.
|
||||
t_earlier = ~U[2026-07-15 13:00:00Z]
|
||||
t_new = ~U[2026-07-15 14:00:00Z]
|
||||
|
||||
# Warm the cache with the earlier valid_time only.
|
||||
ScoreCache.put(10_000, t_earlier, [%{lat: 25.0, lon: -125.0, score: 60}])
|
||||
|
||||
# New .ntms file lands on disk for the next hour — but ScoreCache
|
||||
# has not yet received the cache_refresh for `t_new`.
|
||||
Propagation.replace_scores(
|
||||
[%{lat: 25.0, lon: -125.0, valid_time: t_new, band_mhz: 10_000, score: 85, factors: nil}],
|
||||
t_new
|
||||
)
|
||||
|
||||
assert t_new in Propagation.available_valid_times(10_000)
|
||||
end
|
||||
end
|
||||
|
||||
describe "scores_at_fresh/3" do
|
||||
test "bypasses a stale cache entry and returns the on-disk scores" do
|
||||
# Reproduces the race where MapLive handles a propagation_updated
|
||||
# message while the ScoreCache still holds the previous chain's
|
||||
# scores for the same (band, valid_time). The fresh path must
|
||||
# always return what's on disk and rewrite the cache accordingly.
|
||||
valid_time = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
stale_scores = [%{lat: 25.0, lon: -125.0, score: 10}]
|
||||
ScoreCache.put(10_000, valid_time, stale_scores)
|
||||
|
||||
Propagation.replace_scores(
|
||||
[%{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 95, factors: nil}],
|
||||
valid_time
|
||||
)
|
||||
|
||||
fresh = Propagation.scores_at_fresh(10_000, valid_time, nil)
|
||||
|
||||
assert [%{lat: 25.0, lon: -125.0, score: 95, valid_time: ^valid_time}] = fresh
|
||||
end
|
||||
|
||||
test "refreshes the cache so subsequent scores_at calls see the new data" do
|
||||
valid_time = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
ScoreCache.put(10_000, valid_time, [%{lat: 25.0, lon: -125.0, score: 10}])
|
||||
|
||||
Propagation.replace_scores(
|
||||
[%{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 95, factors: nil}],
|
||||
valid_time
|
||||
)
|
||||
|
||||
_ = Propagation.scores_at_fresh(10_000, valid_time, nil)
|
||||
|
||||
assert {:ok, [%{score: 95}]} = ScoreCache.fetch(10_000, valid_time)
|
||||
end
|
||||
|
||||
test "applies bounding-box filtering just like scores_at/3" do
|
||||
valid_time = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
Propagation.replace_scores(
|
||||
[
|
||||
%{lat: 25.0, lon: -125.0, valid_time: valid_time, band_mhz: 10_000, score: 80, factors: nil},
|
||||
%{lat: 40.0, lon: -80.0, valid_time: valid_time, band_mhz: 10_000, score: 70, factors: nil}
|
||||
],
|
||||
valid_time
|
||||
)
|
||||
|
||||
bounds = %{"south" => 20.0, "north" => 30.0, "west" => -130.0, "east" => -120.0}
|
||||
fresh = Propagation.scores_at_fresh(10_000, valid_time, bounds)
|
||||
|
||||
assert length(fresh) == 1
|
||||
assert hd(fresh).lat == 25.0
|
||||
end
|
||||
end
|
||||
|
||||
describe "warm_cache_and_broadcast/2" do
|
||||
|
|
|
|||
|
|
@ -3,6 +3,9 @@ defmodule MicrowavepropWeb.MapLiveTest do
|
|||
|
||||
import Phoenix.LiveViewTest
|
||||
|
||||
alias Microwaveprop.Propagation
|
||||
alias Microwaveprop.Propagation.ScoreCache
|
||||
|
||||
describe "mount" do
|
||||
test "renders full-page map", %{conn: conn} do
|
||||
{:ok, _lv, html} = live(conn, ~p"/map")
|
||||
|
|
@ -133,6 +136,53 @@ defmodule MicrowavepropWeb.MapLiveTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "propagation_updated handler" do
|
||||
setup do
|
||||
ScoreCache.clear()
|
||||
|
||||
on_exit(fn ->
|
||||
ScoreCache.clear()
|
||||
end)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
test "pushes freshly-written scores when the cache still holds stale ones", %{conn: conn} do
|
||||
# Use a valid_time close enough to now that available_valid_times
|
||||
# keeps it in the timeline.
|
||||
valid_time = DateTime.utc_now() |> DateTime.truncate(:second) |> Map.put(:minute, 0) |> Map.put(:second, 0)
|
||||
|
||||
# Seed the on-disk store and warm the cache with the same content
|
||||
# so the map has a baseline to render on mount.
|
||||
Propagation.replace_scores(
|
||||
[%{lat: 33.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 40, factors: nil}],
|
||||
valid_time
|
||||
)
|
||||
|
||||
Propagation.warm_cache_and_broadcast(10_000, valid_time)
|
||||
ScoreCache.sync()
|
||||
|
||||
{:ok, lv, _html} = live(conn, ~p"/map")
|
||||
|
||||
# Simulate a new forecast-hour compute: rewrite the .ntms 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.
|
||||
Propagation.replace_scores(
|
||||
[%{lat: 33.0, lon: -97.0, valid_time: valid_time, band_mhz: 10_000, score: 90, factors: nil}],
|
||||
valid_time
|
||||
)
|
||||
|
||||
# Deliver the propagation_updated PubSub message directly.
|
||||
send(lv.pid, {:propagation_updated, [valid_time]})
|
||||
|
||||
# The map must emit the new score (90), not the stale cached one (40).
|
||||
assert_push_event(lv, "update_scores", %{scores: scores})
|
||||
assert Enum.any?(scores, fn s -> s.score == 90 end)
|
||||
refute Enum.any?(scores, fn s -> s.score == 40 end)
|
||||
end
|
||||
end
|
||||
|
||||
describe "pipeline status chip" do
|
||||
test "renders the aggregated pipeline status chip on mount", %{conn: conn} do
|
||||
{:ok, _lv, html} = live(conn, ~p"/map")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue