diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index d8b1eb2a..8b1bd346 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -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 diff --git a/lib/microwaveprop_web/live/map_live.ex b/lib/microwaveprop_web/live/map_live.ex index 9a9fc683..a3bbe388 100644 --- a/lib/microwaveprop_web/live/map_live.ex +++ b/lib/microwaveprop_web/live/map_live.ex @@ -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) diff --git a/test/microwaveprop/propagation_test.exs b/test/microwaveprop/propagation_test.exs index a5cdd22a..ab960df3 100644 --- a/test/microwaveprop/propagation_test.exs +++ b/test/microwaveprop/propagation_test.exs @@ -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 diff --git a/test/microwaveprop_web/live/map_live_test.exs b/test/microwaveprop_web/live/map_live_test.exs index c0de8c94..85f11733 100644 --- a/test/microwaveprop_web/live/map_live_test.exs +++ b/test/microwaveprop_web/live/map_live_test.exs @@ -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")