From 71c134d93b275681c494c0cfc5d42a8e3275d527 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 21 Apr 2026 09:42:29 -0500 Subject: [PATCH] chore(cleanup): drop dead scorer_diff + DB grid fallback; harden NotifyListener MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - AdminTaskWorker: remove scorer_diff no-op clause (propagation_scores table was dropped pre-cutover; queue is drained). - Weather: remove load_weather_grid_from_db/1 fallback — Rust writes ProfilesFile for every forecast hour, so the DB path was unreachable in practice and loaded stale pre-cutover data when it fired. - Weather.build_grid_cache_row/4 now prefers persisted fields from the ProfilesFile profile map (surface_refractivity, min_refractivity_gradient, ducting_detected, duct_characteristics) over re-deriving from the raw profile, matching the old DB-path semantics. - NotifyListener: wrap per-band warm in try/rescue so one corrupt or mid-rename .ntms file doesn't abort the remaining bands. Failures log a warning with band + valid_time; successes still emit one info line. - weather_grid_test: rewrite insert_hrrr_profile helper to write ProfilesFile directly and drop the "only returns points on 0.125 grid" test (filter was a property of the deleted DB fallback; Rust writes only grid points by construction). --- .../propagation/notify_listener.ex | 26 ++++- lib/microwaveprop/weather.ex | 59 +++------- .../workers/admin_task_worker.ex | 11 -- test/microwaveprop/weather_grid_test.exs | 107 ++++++++---------- 4 files changed, 88 insertions(+), 115 deletions(-) diff --git a/lib/microwaveprop/propagation/notify_listener.ex b/lib/microwaveprop/propagation/notify_listener.ex index 9be71687..b2a35241 100644 --- a/lib/microwaveprop/propagation/notify_listener.ex +++ b/lib/microwaveprop/propagation/notify_listener.ex @@ -85,9 +85,13 @@ defmodule Microwaveprop.Propagation.NotifyListener do end defp handle_propagation_ready(valid_time) do - Enum.each(BandConfig.all_bands(), fn band -> - Propagation.warm_cache_and_broadcast(band.freq_mhz, valid_time) - end) + {ok, err} = + Enum.reduce(BandConfig.all_bands(), {0, 0}, fn band, {ok, err} -> + case warm_band(band.freq_mhz, valid_time) do + :ok -> {ok + 1, err} + :error -> {ok, err + 1} + end + end) # Prune anything older than 2 hours so the cache doesn't grow # unbounded when Rust covers many forecast hours back-to-back. @@ -95,6 +99,20 @@ defmodule Microwaveprop.Propagation.NotifyListener do Phoenix.PubSub.broadcast(@pubsub, @topic, {:propagation_updated, [valid_time]}) - Logger.info("NotifyListener: warmed ScoreCache + broadcast for valid_time=#{valid_time}") + if err > 0 do + Logger.warning("NotifyListener: warmed #{ok} bands, #{err} failed for valid_time=#{valid_time}") + else + Logger.info("NotifyListener: warmed ScoreCache + broadcast for valid_time=#{valid_time}") + end + end + + defp warm_band(band_mhz, valid_time) do + Propagation.warm_cache_and_broadcast(band_mhz, valid_time) + :ok + rescue + e -> + Logger.warning("NotifyListener: warm failed for band=#{band_mhz} valid_time=#{valid_time}: #{inspect(e)}") + + :error end end diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 47d12b0e..2638d167 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -439,10 +439,10 @@ defmodule Microwaveprop.Weather do end @doc """ - Synchronous cache-or-DB read. Blocks for several seconds on a cold cache - because `load_weather_grid_from_db/1` parses 176k rows with JSONB columns. - Used by tests and by `weather_point_detail/3` fallbacks. LiveView callers - should prefer `latest_weather_grid/1`. + Synchronous cache-or-disk read. Blocks for ~1s on a cold cache while + `ProfilesFile.read/1` loads the latest grid from `/data/profiles`. + Used by tests and by `weather_point_detail/3` fallbacks. LiveView + callers should prefer `latest_weather_grid/1`. """ @spec load_weather_grid(map()) :: [map()] def load_weather_grid(bounds) do @@ -513,7 +513,7 @@ defmodule Microwaveprop.Weather do defp load_grid_rows_for(valid_time) do case ProfilesFile.read(valid_time) do {:ok, grid_data} -> build_grid_cache_rows(grid_data, valid_time) - {:error, _} -> load_weather_grid_from_db(valid_time) + {:error, _} -> [] end end @@ -551,31 +551,6 @@ defmodule Microwaveprop.Weather do :ok end - defp load_weather_grid_from_db(latest_vt) do - from(h in HrrrProfile, - where: h.valid_time == ^latest_vt and h.is_grid_point == true, - select: %{ - lat: h.lat, - lon: h.lon, - valid_time: h.valid_time, - temperature: h.surface_temp_c, - dewpoint_depression: fragment("? - ?", h.surface_temp_c, h.surface_dewpoint_c), - bl_height: h.hpbl_m, - pwat: h.pwat_mm, - refractivity_gradient: h.min_refractivity_gradient, - ducting: h.ducting_detected, - surface_pressure_mb: h.surface_pressure_mb, - surface_temp_c: h.surface_temp_c, - surface_dewpoint_c: h.surface_dewpoint_c, - surface_refractivity: h.surface_refractivity, - profile: h.profile, - duct_characteristics: h.duct_characteristics - } - ) - |> Repo.all(timeout: 60_000) - |> Enum.map(&derive_and_clean/1) - end - @doc """ Eagerly populate the `GridCache` with the full CONUS weather grid for `valid_time` and broadcast it to every node in the cluster. Used by the cold @@ -618,16 +593,13 @@ defmodule Microwaveprop.Weather do end @doc """ - Build derived weather grid cache rows directly from the in-memory HRRR - `grid_data` that `PropagationGridWorker` already has after fetching a - forecast hour. Avoids the 15s+ DB round trip that `load_weather_grid_from_db/1` - pays to SELECT + JSONB-decode 92k hrrr_profiles rows — the source of the - crash loop that stopped forecast hours from being written. + Build derived weather grid cache rows directly from an in-memory HRRR + `grid_data` map. Used by the cold-cache fill path after `ProfilesFile.read/1` + returns the grid written by the Rust worker. `grid_data` is `%{{lat, lon} => profile_map}` as produced by - `HrrrClient.fetch_grid/3` (optionally enriched via native duct merge). The - output matches the shape of `load_weather_grid_from_db/1` after - `derive_and_clean/1`. + `ProfilesFile.read/1`. Each row is pushed through `derive_and_clean/1` + to compute the derived fields consumed by the weather map LiveView. """ @spec build_grid_cache_rows(%{{float(), float()} => map()}, DateTime.t(), map() | nil) :: [map()] @@ -653,6 +625,9 @@ defmodule Microwaveprop.Weather do else sounding = derive_sounding(profile[:profile]) + # Prefer explicit values from the profile map (Rust-written ProfilesFile + # persists derived sounding params alongside the raw profile) and fall + # back to deriving from the profile list when absent. row = %{ lat: lat, lon: lon, @@ -661,14 +636,14 @@ defmodule Microwaveprop.Weather do dewpoint_depression: depression(temp_c, profile[:surface_dewpoint_c]), bl_height: profile[:hpbl_m], pwat: profile[:pwat_mm], - refractivity_gradient: sounding[:min_refractivity_gradient], - ducting: sounding[:ducting_detected], + refractivity_gradient: profile[:min_refractivity_gradient] || sounding[:min_refractivity_gradient], + ducting: profile[:ducting_detected] || sounding[:ducting_detected], surface_pressure_mb: profile[:surface_pressure_mb], surface_temp_c: temp_c, surface_dewpoint_c: profile[:surface_dewpoint_c], - surface_refractivity: sounding[:surface_refractivity], + surface_refractivity: profile[:surface_refractivity] || sounding[:surface_refractivity], profile: profile[:profile] || [], - duct_characteristics: sounding[:duct_characteristics] + duct_characteristics: profile[:duct_characteristics] || sounding[:duct_characteristics] } [derive_and_clean(row)] diff --git a/lib/microwaveprop/workers/admin_task_worker.ex b/lib/microwaveprop/workers/admin_task_worker.ex index 51a6cc97..40e1725e 100644 --- a/lib/microwaveprop/workers/admin_task_worker.ex +++ b/lib/microwaveprop/workers/admin_task_worker.ex @@ -202,17 +202,6 @@ defmodule Microwaveprop.Workers.AdminTaskWorker do :ok end - def perform(%Oban.Job{args: %{"task" => "scorer_diff"}}) do - # scorer_diff was a calibration tool that read the per-cell - # factors JSONB from the propagation_scores table. That table is - # gone — factors are no longer persisted across the CONUS grid — - # so the diff has nothing to compute against. Left as a no-op so - # any in-flight Oban rows from the pre-cutover era drain cleanly. - Logger.warning("AdminTask: scorer_diff is disabled — propagation_scores table + factors storage have been dropped") - - :ok - end - def perform(%Oban.Job{args: %{"task" => task}}) do Logger.error("AdminTask: unknown task #{task}") {:error, "unknown task: #{task}"} diff --git a/test/microwaveprop/weather_grid_test.exs b/test/microwaveprop/weather_grid_test.exs index 96d4fba5..10af499c 100644 --- a/test/microwaveprop/weather_grid_test.exs +++ b/test/microwaveprop/weather_grid_test.exs @@ -2,20 +2,40 @@ defmodule Microwaveprop.WeatherGridTest do use Microwaveprop.DataCase, async: false alias Microwaveprop.Propagation.ProfilesFile - alias Microwaveprop.Repo alias Microwaveprop.Weather alias Microwaveprop.Weather.GridCache setup do # GridCache ETS state leaks between tests (it's a named table outside the - # test sandbox). Clear it so each test gets a fresh cache and the DB load - # path actually runs. + # test sandbox). Clear it so each test gets a fresh cache and the disk + # load path actually runs. GridCache.clear() - # Microwaveprop.Cache now memoizes ProfilesFile.read/1 keyed by - # `(base_dir, valid_time)`. Tests within this file share the default - # base_dir and frequently reuse `DateTime.utc_now()` truncated to seconds, - # so a prior test's cached read leaks into the next one's assertion. + # Microwaveprop.Cache memoizes ProfilesFile.read/1 keyed by + # `(base_dir, valid_time)`; tests within this file share the scores_dir + # and frequently reuse `DateTime.utc_now()` truncated to seconds, so a + # prior test's cached read leaks into the next one's assertion. Microwaveprop.Cache.clear() + + dir = + Path.join( + System.tmp_dir!(), + "weather_grid_test_#{System.unique_integer([:positive])}" + ) + + File.mkdir_p!(Path.join(dir, "profiles")) + 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) + :ok end @@ -67,38 +87,6 @@ defmodule Microwaveprop.WeatherGridTest do assert Weather.load_weather_grid(bounds) == [] end - test "only returns points on 0.125 grid" do - now = DateTime.truncate(DateTime.utc_now(), :second) - - insert_hrrr_profile(%{ - lat: 32.125, - lon: -97.375, - valid_time: now, - surface_temp_c: 25.0, - surface_dewpoint_c: 18.0, - surface_pressure_mb: 1013.0, - hpbl_m: 500.0, - pwat_mm: 30.0 - }) - - insert_hrrr_profile(%{ - lat: 32.1337, - lon: -97.3821, - valid_time: now, - surface_temp_c: 25.0, - surface_dewpoint_c: 18.0, - surface_pressure_mb: 1013.0, - hpbl_m: 500.0, - pwat_mm: 30.0 - }) - - bounds = %{"south" => 32.0, "north" => 33.0, "west" => -98.0, "east" => -97.0} - result = Weather.load_weather_grid(bounds) - - assert length(result) == 1 - assert hd(result).lat == 32.125 - end - test "uses most recent valid_time only" do old = DateTime.utc_now() |> DateTime.add(-7200, :second) |> DateTime.truncate(:second) new = DateTime.truncate(DateTime.utc_now(), :second) @@ -545,27 +533,30 @@ defmodule Microwaveprop.WeatherGridTest do end end - defp insert_hrrr_profile(attrs) do - now = DateTime.truncate(DateTime.utc_now(), :second) + # Merges `attrs` (a single HRRR grid cell) into the ProfilesFile for + # `attrs.valid_time`. Re-reads any existing file so multiple calls for + # the same valid_time accumulate cells, mirroring how the Rust worker + # populates the grid in production. + defp insert_hrrr_profile(%{lat: lat, lon: lon, valid_time: valid_time} = attrs) do + existing = + case ProfilesFile.read(valid_time) do + {:ok, grid} -> grid + {:error, _} -> %{} + end - is_gp = - rem(round(attrs.lat * 1000), 125) == 0 and - rem(round(attrs.lon * 1000), 125) == 0 + profile_attrs = + attrs + |> Map.drop([:lat, :lon, :valid_time]) + |> Map.put_new(:profile, []) - defaults = %{ - id: Ecto.UUID.bingenerate(), - run_time: DateTime.add(attrs.valid_time, -3600, :second), - profile: [], - surface_refractivity: nil, - min_refractivity_gradient: nil, - ducting_detected: nil, - duct_characteristics: nil, - is_grid_point: is_gp, - inserted_at: now, - updated_at: now - } + grid = Map.put(existing, {lat, lon}, profile_attrs) - entry = Map.merge(defaults, attrs) - Repo.insert_all("hrrr_profiles", [entry]) + ProfilesFile.write!(valid_time, grid) + # write!/2 invalidates ProfilesFile's internal cache keyed by valid_time, + # but the read helper is also cached separately here; clear on every write + # so tests that rewrite the same valid_time see the new grid. + Microwaveprop.Cache.clear() + + :ok end end