chore(cleanup): drop dead scorer_diff + DB grid fallback; harden NotifyListener
- 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).
This commit is contained in:
parent
61da51c03f
commit
71c134d93b
4 changed files with 88 additions and 115 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)]
|
||||
|
|
|
|||
|
|
@ -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}"}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue