diff --git a/docs/runbook_propagation_pipeline.md b/docs/runbook_propagation_pipeline.md index a62727d2..789d0902 100644 --- a/docs/runbook_propagation_pipeline.md +++ b/docs/runbook_propagation_pipeline.md @@ -58,7 +58,7 @@ unsubscribed for several seconds while Rust writes land. ### FM3 — NFS stale read racing Rust write -**Symptom.** `ScoresFile.read_bounds/2` raises `File.Error` when +**Symptom.** `ScoresFile.read/2` returns `{:error, reason}` when the reader opens the file exactly between Rust's `rename(2)` of the `.tmp.` file and the `fsync` landing on the NFS server. @@ -66,11 +66,13 @@ the reader opens the file exactly between Rust's `rename(2)` of the NFS client cache can briefly serve a stale directory listing or a no-longer-existing path. -**Recovery.** Reconciler's `rescue File.Error` returns `:error` and -moves on; the next tick re-reads the file after NFS consistency -settles. Lazy-miss path (`Propagation.scores_at_fetch/3`) has -historically surfaced this as a LiveView error — the reconciler -warms the cache before that miss happens. +**Recovery.** Reconciler calls `ScoresFile.read/2` directly and +pattern-matches `{:error, _} -> :error`, logging at debug and +skipping the key. The next tick re-reads the file after NFS +consistency settles. Lazy-miss path +(`Propagation.scores_at_fetch/3`) has historically surfaced this as +a LiveView error — the reconciler warms the cache before that miss +happens. ### FM4 — Cluster partition splits ScoreCache broadcast @@ -99,7 +101,12 @@ scores. **Recovery.** Manual. `kubectl -n prop logs deploy/prop-grid-rs` for cause. The Elixir side is read-only against the grid — it does not regenerate scores from within the BEAM since the extraction in -`65693ed`. +`65693ed`. `lib/microwaveprop/workers/propagation_grid_worker.ex` +still exists as a seed worker: the hourly cron fires it with empty +args and it inserts 19 `grid_tasks` rows (f00 + f01..f18) for Rust +to drain. All fetch/decode/score helpers moved to +`rust/prop_grid_rs/src/pipeline.rs`; the Elixir module is retained +only as the cron entry point, not as a fallback compute path. ## Monitoring signals diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 03dcf2e3..f4fed9dd 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -330,12 +330,23 @@ defmodule Microwaveprop.Propagation do cluster. Called from `PropagationGridWorker` after each forecast hour so all pods have a warm cache by the time clients begin requesting the new hour. + + Returns `{:error, reason}` when the score file is missing or + corrupt — callers distinguish those from the successful empty-grid + case to avoid poisoning the cache with `[]` on a bad read. """ - @spec warm_cache_and_broadcast(non_neg_integer(), DateTime.t()) :: :ok + @spec warm_cache_and_broadcast(non_neg_integer(), DateTime.t()) :: + :ok | {:error, :enoent | :invalid_format} def warm_cache_and_broadcast(band_mhz, valid_time) do - scores = ScoresFile.read_bounds(band_mhz, valid_time) - ScoreCache.broadcast_put(band_mhz, valid_time, scores) - :ok + case ScoresFile.read(band_mhz, valid_time) do + {:ok, payload} -> + scores = ScoresFile.extract_points(payload, nil) + ScoreCache.broadcast_put(band_mhz, valid_time, scores) + :ok + + {:error, reason} -> + {:error, reason} + end end defp filter_bounds(scores, nil), do: scores diff --git a/lib/microwaveprop/propagation/notify_listener.ex b/lib/microwaveprop/propagation/notify_listener.ex index b2a35241..30c8f339 100644 --- a/lib/microwaveprop/propagation/notify_listener.ex +++ b/lib/microwaveprop/propagation/notify_listener.ex @@ -107,11 +107,18 @@ defmodule Microwaveprop.Propagation.NotifyListener do end defp warm_band(band_mhz, valid_time) do - Propagation.warm_cache_and_broadcast(band_mhz, valid_time) - :ok + case Propagation.warm_cache_and_broadcast(band_mhz, valid_time) do + :ok -> + :ok + + {:error, reason} -> + Logger.warning("NotifyListener: warm skipped band=#{band_mhz} valid_time=#{valid_time}: #{inspect(reason)}") + + :error + end rescue e -> - Logger.warning("NotifyListener: warm failed for band=#{band_mhz} valid_time=#{valid_time}: #{inspect(e)}") + Logger.warning("NotifyListener: warm crashed band=#{band_mhz} valid_time=#{valid_time}: #{inspect(e)}") :error end diff --git a/lib/microwaveprop/propagation/score_cache_reconciler.ex b/lib/microwaveprop/propagation/score_cache_reconciler.ex index 12a092ab..2bab079a 100644 --- a/lib/microwaveprop/propagation/score_cache_reconciler.ex +++ b/lib/microwaveprop/propagation/score_cache_reconciler.ex @@ -90,15 +90,21 @@ defmodule Microwaveprop.Propagation.ScoreCacheReconciler do end) end + # File may vanish between list and read if a prune runs concurrently, + # or be half-written during an atomic-rename race. `ScoresFile.read/2` + # reports both as `{:error, _}`; we skip and self-heal on the next + # sweep rather than caching an empty grid under that key. defp warm_one(band_mhz, valid_time) do - scores = ScoresFile.read_bounds(band_mhz, valid_time) - ScoreCache.put(band_mhz, valid_time, scores) - :ok - rescue - # File may vanish between list and read if a prune runs concurrently. - # Missing files are self-healing on the next sweep — no point logging. - e in [File.Error] -> - Logger.debug("ScoreCacheReconciler: skipped #{band_mhz}/#{valid_time}: #{inspect(e)}") - :error + case ScoresFile.read(band_mhz, valid_time) do + {:ok, payload} -> + scores = ScoresFile.extract_points(payload, nil) + ScoreCache.put(band_mhz, valid_time, scores) + :ok + + {:error, reason} -> + Logger.debug("ScoreCacheReconciler: skipped #{band_mhz}/#{valid_time}: #{inspect(reason)}") + + :error + end end end diff --git a/lib/microwaveprop/propagation/scores_file.ex b/lib/microwaveprop/propagation/scores_file.ex index 3e5f6fe2..f450b5a2 100644 --- a/lib/microwaveprop/propagation/scores_file.ex +++ b/lib/microwaveprop/propagation/scores_file.ex @@ -361,7 +361,15 @@ defmodule Microwaveprop.Propagation.ScoresFile do # ── Read helpers ──────────────────────────────────────────────── - defp extract_points(%{scores: body} = payload, bounds) do + @doc """ + Decode a `read/2` payload into `[%{lat, lon, score}]`. Exposed so + callers that need to distinguish "missing file" from "empty grid" + can invoke `read/2` directly and feed the payload here after their + own error-path handling. + """ + @spec extract_points(map(), map() | nil) :: + [%{lat: float(), lon: float(), score: non_neg_integer()}] + def 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) diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 2638d167..75a314c4 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -5,6 +5,7 @@ defmodule Microwaveprop.Weather do alias Microwaveprop.Propagation.ProfilesFile alias Microwaveprop.Radio + alias Microwaveprop.Radio.Contact alias Microwaveprop.Repo alias Microwaveprop.Weather.GefsProfile alias Microwaveprop.Weather.GridCache @@ -627,7 +628,9 @@ defmodule Microwaveprop.Weather do # 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. + # back to deriving from the profile list when absent. `prefer/3` uses + # `Map.fetch/2` so a legitimate `false` / `0` persisted value is not + # silently clobbered by a `||` fallback. row = %{ lat: lat, lon: lon, @@ -636,20 +639,31 @@ defmodule Microwaveprop.Weather do dewpoint_depression: depression(temp_c, profile[:surface_dewpoint_c]), bl_height: profile[:hpbl_m], pwat: profile[:pwat_mm], - refractivity_gradient: profile[:min_refractivity_gradient] || sounding[:min_refractivity_gradient], - ducting: profile[:ducting_detected] || sounding[:ducting_detected], + refractivity_gradient: prefer(profile, :min_refractivity_gradient, sounding[:min_refractivity_gradient]), + ducting: prefer(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: profile[:surface_refractivity] || sounding[:surface_refractivity], + surface_refractivity: prefer(profile, :surface_refractivity, sounding[:surface_refractivity]), profile: profile[:profile] || [], - duct_characteristics: profile[:duct_characteristics] || sounding[:duct_characteristics] + duct_characteristics: prefer(profile, :duct_characteristics, sounding[:duct_characteristics]) } [derive_and_clean(row)] end end + # Fetch `key` from `profile` verbatim when present (including `false`, + # `0`, or `[]`); only fall through to the derived value when the key is + # absent. `profile[key] || default` would discard legitimate `false` / + # `0` as if they weren't set. + defp prefer(profile, key, default) do + case Map.fetch(profile, key) do + {:ok, value} -> value + :error -> default + end + end + defp derive_sounding(profile) when is_list(profile) and length(profile) >= 3 do SoundingParams.derive(profile) || %{} end @@ -848,11 +862,19 @@ defmodule Microwaveprop.Weather do nearest HRRR hour. Lets the enrichment enqueuer skip :queued → :queued churn when the backing data is already present. - Returns `false` if `pos1` is nil — a contact with no position can't be - looked up. + Returns `false` if `pos1` or `qso_timestamp` is nil — a contact with no + position or no timestamp can't be looked up. """ - @spec hrrr_data_fully_present?(map()) :: boolean() + @spec hrrr_data_fully_present?( + Contact.t() + | %{ + required(:pos1) => term(), + required(:qso_timestamp) => DateTime.t() | nil, + optional(:pos2) => term() + } + ) :: boolean() def hrrr_data_fully_present?(%{pos1: nil}), do: false + def hrrr_data_fully_present?(%{qso_timestamp: nil}), do: false def hrrr_data_fully_present?(contact) do rounded = HrrrClient.nearest_hrrr_hour(contact.qso_timestamp) diff --git a/lib/microwaveprop/workers/admin_task_worker.ex b/lib/microwaveprop/workers/admin_task_worker.ex index e9cfcba2..408dff1d 100644 --- a/lib/microwaveprop/workers/admin_task_worker.ex +++ b/lib/microwaveprop/workers/admin_task_worker.ex @@ -198,9 +198,11 @@ defmodule Microwaveprop.Workers.AdminTaskWorker do end # Bulk-update via one `UPDATE ... FROM unnest(...)` statement per batch: - # one Postgres round-trip and one fsync instead of N, which was the - # dominant cost for runs of 10k profiles against the remote DB host. - @bulk_update_chunk 2000 + # one Postgres round-trip and one fsync instead of N. Chunk size is + # conservative to contain the blast radius of a single bad row + # (deleted UUID, bad duct payload) — the chunk re-runs row-by-row on + # failure so the remaining rows in the job still make progress. + @bulk_update_chunk 500 defp derive_native_row(profile) do p = %{ @@ -246,7 +248,33 @@ defmodule Microwaveprop.Workers.AdminTaskWorker do defp bulk_update_native_derivations(rows) do rows |> Enum.chunk_every(@bulk_update_chunk) - |> Enum.reduce(0, fn chunk, acc -> acc + execute_bulk_update(chunk) end) + |> Enum.reduce(0, fn chunk, acc -> acc + safe_execute_chunk(chunk) end) + end + + # Wraps the chunk UPDATE so one bad row doesn't kill the whole job. + # On Postgrex failure, fall back to per-row updates for just that + # chunk — rows that still succeed contribute to the count; rows that + # re-raise are logged and skipped. + defp safe_execute_chunk(chunk) do + execute_bulk_update(chunk) + rescue + e in [Postgrex.Error, DBConnection.EncodeError] -> + Logger.warning("AdminTask.native_derive: bulk chunk failed (#{inspect(e)}); falling back row-by-row") + + per_row_fallback(chunk) + end + + defp per_row_fallback(chunk) do + Enum.reduce(chunk, 0, fn row, acc -> + try do + execute_bulk_update([row]) + acc + 1 + rescue + e -> + Logger.warning("AdminTask.native_derive: skip id=#{row.id}: #{inspect(e)}") + acc + end + end) end defp execute_bulk_update(chunk) do diff --git a/mix.exs b/mix.exs index 3ecc79f5..66353b9c 100644 --- a/mix.exs +++ b/mix.exs @@ -14,7 +14,14 @@ defmodule Microwaveprop.MixProject do listeners: [Phoenix.CodeReloader], dialyzer: [ plt_add_apps: [:mix, :ex_unit], - plt_file: {:no_warn, "priv/plts/project.plt"} + plt_file: {:no_warn, "priv/plts/project.plt"}, + flags: [ + :error_handling, + :unknown, + :unmatched_returns, + :extra_return, + :missing_return + ] ] ] end diff --git a/test/microwaveprop/propagation/score_cache_reconciler_test.exs b/test/microwaveprop/propagation/score_cache_reconciler_test.exs index 02100e8a..5cc82fae 100644 --- a/test/microwaveprop/propagation/score_cache_reconciler_test.exs +++ b/test/microwaveprop/propagation/score_cache_reconciler_test.exs @@ -3,6 +3,8 @@ defmodule Microwaveprop.Propagation.ScoreCacheReconcilerTest do # Application env and mutates the shared ScoreCache ETS table. use ExUnit.Case, async: false + import ExUnit.CaptureLog + alias Microwaveprop.Propagation.BandConfig alias Microwaveprop.Propagation.Grid alias Microwaveprop.Propagation.ScoreCache @@ -74,4 +76,109 @@ defmodule Microwaveprop.Propagation.ScoreCacheReconcilerTest do assert 0 = ScoreCacheReconciler.sweep_once() end end + + describe "GenServer lifecycle" do + # The gate in start_link/1 defaults :start_score_cache_reconciler to + # false in the test env. Flip it on for this block so start_supervised! + # gets a live pid rather than :ignore. + setup do + Application.put_env(:microwaveprop, :start_score_cache_reconciler, true) + + on_exit(fn -> + Application.put_env(:microwaveprop, :start_score_cache_reconciler, false) + end) + + :ok + end + + defp wait_for_cache(band_mhz, valid_time, timeout_ms) do + deadline = System.monotonic_time(:millisecond) + timeout_ms + poll_cache(band_mhz, valid_time, deadline) + end + + defp poll_cache(band_mhz, valid_time, deadline) do + case ScoreCache.fetch(band_mhz, valid_time) do + {:ok, _} = ok -> + ok + + :miss -> + if System.monotonic_time(:millisecond) >= deadline do + :timeout + else + Process.sleep(25) + poll_cache(band_mhz, valid_time, deadline) + end + end + end + + test "run_on_start: true warms cache from disk via handle_info(:sweep)", + %{band_mhz: band_mhz} do + valid_time = ~U[2026-04-21 14:00:00Z] + ScoresFile.write!(band_mhz, valid_time, sample_scores()) + + assert :miss = ScoreCache.fetch(band_mhz, valid_time) + + start_supervised!({ScoreCacheReconciler, run_on_start: true, interval_ms: 100}) + + # The boot timer is 500ms, so the first sweep lands somewhere + # around t+500ms. Give it generous headroom to avoid flakes on + # loaded CI workers. + assert {:ok, _scores} = wait_for_cache(band_mhz, valid_time, 2_000) + end + + test "run_on_start: false does not sweep within the wait window", + %{band_mhz: band_mhz} do + valid_time = ~U[2026-04-21 15:00:00Z] + ScoresFile.write!(band_mhz, valid_time, sample_scores()) + + start_supervised!({ScoreCacheReconciler, run_on_start: false, interval_ms: 100}) + + # Wait past the 500ms boot delay the other branch would have used. + Process.sleep(750) + + assert :miss = ScoreCache.fetch(band_mhz, valid_time) + end + + test "interval_ms rescheduling picks up files written after the first sweep", + %{band_mhz: band_mhz} do + first = ~U[2026-04-21 16:00:00Z] + second = ~U[2026-04-21 17:00:00Z] + + ScoresFile.write!(band_mhz, first, sample_scores()) + + start_supervised!({ScoreCacheReconciler, run_on_start: true, interval_ms: 150}) + + # First sweep warms `first` but knows nothing about `second`. + assert {:ok, _} = wait_for_cache(band_mhz, first, 2_000) + assert :miss = ScoreCache.fetch(band_mhz, second) + + # Drop the second file and wait for the next scheduled tick. + ScoresFile.write!(band_mhz, second, sample_scores()) + + assert {:ok, _} = wait_for_cache(band_mhz, second, 2_000) + end + + test "logs an info line when warmed > 0", %{band_mhz: band_mhz} do + valid_time = ~U[2026-04-21 18:00:00Z] + ScoresFile.write!(band_mhz, valid_time, sample_scores()) + + # config/test.exs pins the global logger to :warning, which drops the + # :info line before capture_log sees it. Raise the level for the + # duration of the test and restore it on exit. + prev_level = Logger.level() + Logger.configure(level: :info) + on_exit(fn -> Logger.configure(level: prev_level) end) + + log = + capture_log([level: :info], fn -> + start_supervised!({ScoreCacheReconciler, run_on_start: true, interval_ms: 100}) + + assert {:ok, _} = wait_for_cache(band_mhz, valid_time, 2_000) + # Give the logger a beat to flush before we snapshot the buffer. + Process.sleep(50) + end) + + assert log =~ "ScoreCacheReconciler: warmed" + end + end end diff --git a/test/microwaveprop/weather_grid_test.exs b/test/microwaveprop/weather_grid_test.exs index 10af499c..8b4231bf 100644 --- a/test/microwaveprop/weather_grid_test.exs +++ b/test/microwaveprop/weather_grid_test.exs @@ -4,6 +4,7 @@ defmodule Microwaveprop.WeatherGridTest do alias Microwaveprop.Propagation.ProfilesFile alias Microwaveprop.Weather alias Microwaveprop.Weather.GridCache + alias Microwaveprop.Weather.SoundingParams setup do # GridCache ETS state leaks between tests (it's a named table outside the @@ -362,6 +363,46 @@ defmodule Microwaveprop.WeatherGridTest do assert length(Weather.build_grid_cache_rows(grid_data, now, nil)) == 2 end + test "explicit ducting_detected: false on the profile wins over derived true from sounding params" do + # Regression guard: the `||` fallback used to clobber legitimate `false` + # values written by the Rust ProfilesFile with whatever + # `SoundingParams.derive/1` produced. The profile below has a steep + # near-surface N drop (moist sfc → dry aloft) that derives ducts + # = [_], so `sounding[:ducting_detected]` would be `true`. With the + # explicit `false` on the profile map, the row must carry `false`. + now = DateTime.truncate(DateTime.utc_now(), :second) + + strong_inversion_profile = [ + %{"pres" => 1013.0, "tmpc" => 30.0, "dwpc" => 25.0, "hght" => 100.0}, + %{"pres" => 1000.0, "tmpc" => 35.0, "dwpc" => -5.0, "hght" => 200.0}, + %{"pres" => 925.0, "tmpc" => 20.0, "dwpc" => 0.0, "hght" => 800.0}, + %{"pres" => 850.0, "tmpc" => 15.0, "dwpc" => -5.0, "hght" => 1500.0}, + %{"pres" => 500.0, "tmpc" => -15.0, "dwpc" => -25.0, "hght" => 5600.0} + ] + + # Sanity check: this profile does derive a duct from SoundingParams. + derived = SoundingParams.derive(strong_inversion_profile) + assert derived.ducting_detected == true + + grid_data = %{ + {32.125, -97.375} => %{ + profile: strong_inversion_profile, + surface_temp_c: 30.0, + surface_dewpoint_c: 25.0, + surface_pressure_mb: 1013.0, + hpbl_m: 500.0, + pwat_mm: 30.0, + # Explicit false — persisted by the Rust worker. Must NOT be + # clobbered by the derived-from-sounding true above. + ducting_detected: false + } + } + + [row] = Weather.build_grid_cache_rows(grid_data, now) + + assert row.ducting == false + end + test "strips raw profile/duct_characteristics/surface_temp_c/surface_dewpoint_c fields" do now = DateTime.truncate(DateTime.utc_now(), :second) diff --git a/test/microwaveprop/weather_test.exs b/test/microwaveprop/weather_test.exs index b4c84cc7..0aa9565c 100644 --- a/test/microwaveprop/weather_test.exs +++ b/test/microwaveprop/weather_test.exs @@ -720,6 +720,16 @@ defmodule Microwaveprop.WeatherTest do qso_timestamp: @contact_time }) end + + test "returns false when qso_timestamp is nil" do + # A Contact with no timestamp can't be mapped to an HRRR hour; guard + # against HrrrClient.nearest_hrrr_hour(nil) crashing here. + refute Weather.hrrr_data_fully_present?(%{ + pos1: %{"lat" => 32.9, "lon" => -97.04}, + pos2: nil, + qso_timestamp: nil + }) + end end describe "round_to_hrrr_grid/2" do diff --git a/test/microwaveprop/workers/radar_frame_worker_test.exs b/test/microwaveprop/workers/radar_frame_worker_test.exs index e1719a98..4f31e12a 100644 --- a/test/microwaveprop/workers/radar_frame_worker_test.exs +++ b/test/microwaveprop/workers/radar_frame_worker_test.exs @@ -168,6 +168,25 @@ defmodule Microwaveprop.Workers.RadarFrameWorkerTest do end end + describe "NexradClient error contract" do + # Pins the exact "NEXRAD n0q HTTP " format emitted by + # NexradClient.fetch_decoded_frame/1 on a non-200 response. RadarFrameWorker + # pattern-matches on this literal prefix in `permanent_error?/1` to + # classify 4xx as permanent (stop retrying) and everything else as + # transient (let Oban retry). A silent format change would quietly turn + # permanent 404s into infinite retries — fail loudly here instead. + test "fetch_decoded_frame/1 returns error starting with \"NEXRAD n0q HTTP \" on 4xx" do + Req.Test.stub(NexradClient, fn conn -> + Plug.Conn.send_resp(conn, 404, "not found") + end) + + assert {:error, reason} = NexradClient.fetch_decoded_frame(~U[2024-09-15 18:30:00Z]) + assert is_binary(reason) + assert String.starts_with?(reason, "NEXRAD n0q HTTP ") + assert reason == "NEXRAD n0q HTTP 404" + end + end + describe "unique constraint" do test "rejects a second insert with the same frame_ts while one is pending" do frame_ts = "2024-09-15T18:30:00Z"