From b3ad3bdbe63dcc0ed79ab303f354a735a215275d Mon Sep 17 00:00:00 2001 From: Graham McInitre Date: Mon, 20 Jul 2026 19:48:45 -0500 Subject: [PATCH] fix: sweep bugs across Python MQTT listener, Elixir PSKR/propagation, and Rust Python pskr_mqtt_listen.py: - Guard against malformed CONNACK/SUBACK/PUBLISH packets - Fix _s() truthiness: is not None instead of if v (zero SNR was lost) - EINTR-safe select loop, OOB data handling, VBInt overflow check - Proper socket cleanup with try/finally + DISCONNECT on shutdown Elixir: - Log dropped spot errors in aggregator (was silently discarded) - Wire retain_scores_window into NotifyListener chain completion - Recurse sweep_tmp_dir into band/weather_scalars subdirectories - Rescue recalibrator.run/0 to write failed status row on crash - Handle nil in fmt_snr/fmt_callsigns (no more nil dB crashes) - Replace Stream.run with Enum.reduce logging exits in poll_worker - Handle File.stat race in ms_footprints prune, grid_center nil log - Fix unused variables, stale comments, missing get_path!/1 Rust: - Graceful JoinError handling in fetcher/hrdps_fetcher (no more panics) - round_to_5min returns Option (leap-second safe) - Acquire/Release ordering on shutdown flags (was Relaxed) - Parameterized NOTIFY in db.rs, defensive scalar sweep, NaN guard - OsString::push for tmp naming, clippy fixes --- lib/microwaveprop/buildings/ms_footprints.ex | 25 +- lib/microwaveprop/commercial/poll_worker.ex | 9 +- lib/microwaveprop/propagation.ex | 50 ++- .../propagation/notify_listener.ex | 4 + lib/microwaveprop/pskr.ex | 13 +- lib/microwaveprop/pskr/aggregator.ex | 3 +- lib/microwaveprop/pskr/recalibrator.ex | 11 + lib/microwaveprop/rover_planning.ex | 4 + lib/microwaveprop/weather/profile_lookup.ex | 4 +- .../workers/contact_weather_enqueue_worker.ex | 2 +- .../live/contact_live/show.ex | 2 +- lib/microwaveprop_web/live/pskr_spots_live.ex | 3 + .../live/weather_map_component.ex | 2 +- .../prop_grid_rs/src/bin/hrrr_point_worker.rs | 8 +- rust/prop_grid_rs/src/bin/worker.rs | 8 +- rust/prop_grid_rs/src/db.rs | 8 +- rust/prop_grid_rs/src/decoder.rs | 3 + rust/prop_grid_rs/src/fetcher.rs | 21 +- rust/prop_grid_rs/src/hrdps_fetcher.rs | 18 +- rust/prop_grid_rs/src/nexrad.rs | 11 +- rust/prop_grid_rs/src/pipeline.rs | 11 +- rust/prop_grid_rs/src/scorer.rs | 2 +- rust/prop_grid_rs/src/scores_file.rs | 19 +- rust/prop_grid_rs/src/weather_scalar_file.rs | 17 +- scripts/pskr_mqtt_listen.py | 366 ++++++++++-------- 25 files changed, 428 insertions(+), 196 deletions(-) diff --git a/lib/microwaveprop/buildings/ms_footprints.ex b/lib/microwaveprop/buildings/ms_footprints.ex index 57d86a18..50d46f71 100644 --- a/lib/microwaveprop/buildings/ms_footprints.ex +++ b/lib/microwaveprop/buildings/ms_footprints.ex @@ -169,14 +169,25 @@ defmodule Microwaveprop.Buildings.MsFootprints do defp prune_file(file, acc, cutoff_unix) do path = Path.join(cache_dir(), file) - stat = File.stat!(path) - mtime_unix = stat.mtime |> NaiveDateTime.from_erl!() |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix() - if mtime_unix < cutoff_unix do - _ = File.rm(path) - acc + 1 - else - acc + case File.stat(path) do + {:ok, stat} -> + mtime_unix = stat.mtime |> NaiveDateTime.from_erl!() |> DateTime.from_naive!("Etc/UTC") |> DateTime.to_unix() + + if mtime_unix < cutoff_unix do + _ = File.rm(path) + acc + 1 + else + acc + end + + {:error, :enoent} -> + # Race: file deleted between File.ls and File.stat. Skip it. + acc + + {:error, reason} -> + Logger.warning("MsFootprints.prune_older_than: stat failed for #{file}: #{inspect(reason)}") + acc end end end diff --git a/lib/microwaveprop/commercial/poll_worker.ex b/lib/microwaveprop/commercial/poll_worker.ex index dfb43e88..59ce06d6 100644 --- a/lib/microwaveprop/commercial/poll_worker.ex +++ b/lib/microwaveprop/commercial/poll_worker.ex @@ -72,7 +72,14 @@ defmodule Microwaveprop.Commercial.PollWorker do timeout: :infinity, max_concurrency: System.schedulers_online() ) - |> Stream.run() + |> Enum.reduce(:ok, fn + {:ok, :ok}, acc -> + acc + + {:exit, reason}, acc -> + Logger.error("PollWorker weather fetch task crashed: #{inspect(reason)}") + acc + end) end defp fetch_station_weather(station_code, start_dt, now) do diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 81dc35b0..baefc8e4 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -261,18 +261,56 @@ defmodule Microwaveprop.Propagation do defp sweep_tmp_dir(dir, acc) do case File.ls(dir) do {:ok, entries} -> - entries - |> Enum.filter(&String.contains?(&1, ".tmp.")) - |> Enum.reduce(acc, fn entry, acc -> - _ = File.rm_rf(Path.join(dir, entry)) - acc + 1 - end) + Enum.reduce(entries, acc, &sweep_tmp_entry(dir, &1, &2)) _ -> acc end end + defp sweep_tmp_entry(parent, entry, acc) do + full = Path.join(parent, entry) + + case File.ls(full) do + {:ok, _} -> + sweep_tmp_dir(full, acc) + + {:error, _} -> + if String.contains?(entry, ".tmp.") do + _ = File.rm_rf(full) + acc + 1 + else + acc + end + end + end + + @doc """ + Retains only score/profile/scalar files within the active 48-hour + forecast window, deleting everything older than `run_time`. Called + by `NotifyListener` after chain completion to keep `/data/scores` + within bounds. + + Mirrors `ScoreCache.prune_outside_window` for the on-disk tier. + """ + @spec retain_scores_window(DateTime.t()) :: :ok + def retain_scores_window(%DateTime{} = run_time) do + max_fh = 48 + + scores_deleted = ScoresFile.retain_window(run_time, max_fh) + profiles_deleted = ProfilesFile.retain_window(run_time, max_fh) + scalars_deleted = ScalarFile.retain_window(run_time, max_fh) + total = scores_deleted + profiles_deleted + scalars_deleted + + if total > 0 do + Logger.info( + "Propagation.retain_scores_window: deleted #{total} stale files (#{scores_deleted} scores, #{profiles_deleted} profiles, #{scalars_deleted} scalars)" + ) + end + + :ok + end + @doc """ Returns distinct valid_times for a band, ordered ascending. Always reads from the on-disk `ScoresFile` store — the `ScoreCache` only diff --git a/lib/microwaveprop/propagation/notify_listener.ex b/lib/microwaveprop/propagation/notify_listener.ex index a28489f3..c6b05724 100644 --- a/lib/microwaveprop/propagation/notify_listener.ex +++ b/lib/microwaveprop/propagation/notify_listener.ex @@ -115,6 +115,10 @@ defmodule Microwaveprop.Propagation.NotifyListener do {past, future} = Propagation.hot_cache_window() ScoreCache.prune_outside_window(past, future) + # Sweep NFS scores to the active 48h window so stale forecast hours + # don't accumulate between pipeline runs. + Propagation.retain_scores_window(valid_time) + kickoff_scalar_materialization(valid_time) _ = Phoenix.PubSub.broadcast(@pubsub, @topic, {:propagation_updated, [valid_time]}) diff --git a/lib/microwaveprop/pskr.ex b/lib/microwaveprop/pskr.ex index 9596f2f4..50cf57da 100644 --- a/lib/microwaveprop/pskr.ex +++ b/lib/microwaveprop/pskr.ex @@ -49,6 +49,8 @@ defmodule Microwaveprop.Pskr do alias Microwaveprop.Geo alias Microwaveprop.Radio.Maidenhead + require Logger + @typedoc "Parsed PSK Reporter spot, post-`parse_spot/1`." @type spot :: %{ band: String.t(), @@ -212,7 +214,16 @@ defmodule Microwaveprop.Pskr do end end - defp grid_center(grid), do: Geo.maidenhead_center(grid) || {0.0, 0.0} + defp grid_center(grid) do + case Geo.maidenhead_center(grid) do + nil -> + Logger.warning("Pskr.grid_center: Geo.maidenhead_center returned nil for grid=#{grid}, using {0.0, 0.0}") + {0.0, 0.0} + + center -> + center + end + end defp maybe_mode_list(""), do: [] defp maybe_mode_list(mode), do: [mode] diff --git a/lib/microwaveprop/pskr/aggregator.ex b/lib/microwaveprop/pskr/aggregator.ex index b0ee00c4..6f615ada 100644 --- a/lib/microwaveprop/pskr/aggregator.ex +++ b/lib/microwaveprop/pskr/aggregator.ex @@ -82,7 +82,8 @@ defmodule Microwaveprop.Pskr.Aggregator do rows = Map.update(state.rows, key, Pskr.merge_spot(nil, spot), &Pskr.merge_spot(&1, spot)) {:noreply, %{state | rows: rows, parsed: state.parsed + 1}} - {:error, _reason} -> + {:error, reason} -> + Logger.debug("Pskr.Aggregator dropped spot: #{inspect(reason)}") {:noreply, %{state | dropped: state.dropped + 1}} end end diff --git a/lib/microwaveprop/pskr/recalibrator.ex b/lib/microwaveprop/pskr/recalibrator.ex index 22c229f1..37e76c38 100644 --- a/lib/microwaveprop/pskr/recalibrator.ex +++ b/lib/microwaveprop/pskr/recalibrator.ex @@ -106,6 +106,17 @@ defmodule Microwaveprop.Pskr.Recalibrator do else run_analysis(now, stats) end + rescue + e -> + Logger.error("Pskr.Recalibrator: run failed — #{Exception.format(:error, e, __STACKTRACE__)}") + + %RecalibrationRun{} + |> RecalibrationRun.changeset(%{ + run_at: DateTime.truncate(DateTime.utc_now(), :second), + status: "failed", + notes: Exception.format(:error, e, __STACKTRACE__) + }) + |> Repo.insert!(on_conflict: :nothing) end # ── corpus stats ──────────────────────────────────────────────── diff --git a/lib/microwaveprop/rover_planning.ex b/lib/microwaveprop/rover_planning.ex index bbf7478f..548ce213 100644 --- a/lib/microwaveprop/rover_planning.ex +++ b/lib/microwaveprop/rover_planning.ex @@ -117,6 +117,10 @@ defmodule Microwaveprop.RoverPlanning do ) end + @doc "Fetches a single path by id or raises Ecto.NoResultsError." + @spec get_path!(Ecto.UUID.t()) :: Path.t() + def get_path!(id), do: Repo.get!(Path, id) + @doc """ Reconciles the path matrix for a mission against the current set of (rover_location × station × band) tuples. Deletes paths for tuples diff --git a/lib/microwaveprop/weather/profile_lookup.ex b/lib/microwaveprop/weather/profile_lookup.ex index a97fd759..a96acf83 100644 --- a/lib/microwaveprop/weather/profile_lookup.ex +++ b/lib/microwaveprop/weather/profile_lookup.ex @@ -66,8 +66,6 @@ defmodule Microwaveprop.Weather.ProfileLookup do ) ) - _point_set = for {lat, lon, vt} <- profiles, into: MapSet.new(), do: {lat, lon, vt} - points |> Enum.filter(fn {{lat, lon}, valid_time} -> Enum.any?(profiles, fn {pl, pn, pvt} -> @@ -184,7 +182,7 @@ defmodule Microwaveprop.Weather.ProfileLookup do Enum.uniq_by(for_result, fn {key, _} -> key end) {keys, contact_map} = - Enum.reduce(points, {[], %{}}, fn {{lat, lon, ts} = key, c}, {ks, cm} -> + Enum.reduce(points, {[], %{}}, fn {{_lat, _lon, _ts} = key, c}, {ks, cm} -> {[key | ks], Map.update(cm, c, [key], &[key | &1])} end) diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index 78c07f67..d8bed27b 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -129,7 +129,7 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do contacts |> Enum.filter(&(&1.pos1 != nil and not NarrClient.in_coverage?(&1.qso_timestamp))) |> Enum.reduce({%{}, []}, fn c, {pm, pts} -> - rounded = HrrrClient.nearest_hrrr_hour(c.qso_timestamp) + rounded = Weather.HrrrClient.nearest_hrrr_hour(c.qso_timestamp) points = c diff --git a/lib/microwaveprop_web/live/contact_live/show.ex b/lib/microwaveprop_web/live/contact_live/show.ex index 45762170..f010d909 100644 --- a/lib/microwaveprop_web/live/contact_live/show.ex +++ b/lib/microwaveprop_web/live/contact_live/show.ex @@ -716,7 +716,7 @@ defmodule MicrowavepropWeb.ContactLive.Show do # The surface-observation radius tracks the path geometry, but # soundings are sparse (~120/day nationwide) so we widen out to # 1000 km before giving up and letting the fetch-on-demand path - # (handle_async(:soundings_widen)) backfill from IEM. + # (fetch-on-demand via IEM soundings search with widening radius) sounding_search = Weather.soundings_with_widening_radius(%{ lat: mid_lat, diff --git a/lib/microwaveprop_web/live/pskr_spots_live.ex b/lib/microwaveprop_web/live/pskr_spots_live.ex index 1ee106b0..657d2e34 100644 --- a/lib/microwaveprop_web/live/pskr_spots_live.ex +++ b/lib/microwaveprop_web/live/pskr_spots_live.ex @@ -113,9 +113,12 @@ defmodule MicrowavepropWeb.PskrSpotsLive do defp fmt_dt(nil), do: "—" defp fmt_dt(dt), do: Calendar.strftime(dt, "%Y-%m-%d %H:%M UTC") + defp fmt_snr(nil, _max), do: "—" + defp fmt_snr(_min, nil), do: "—" defp fmt_snr(min, max) when min == max, do: "#{min} dB" defp fmt_snr(min, max), do: "#{min}–#{max} dB" + defp fmt_callsigns(nil), do: "—" defp fmt_callsigns([]), do: "—" defp fmt_callsigns(calls), do: Enum.join(calls, ", ") diff --git a/lib/microwaveprop_web/live/weather_map_component.ex b/lib/microwaveprop_web/live/weather_map_component.ex index 1017b76f..f0d9b8e5 100644 --- a/lib/microwaveprop_web/live/weather_map_component.ex +++ b/lib/microwaveprop_web/live/weather_map_component.ex @@ -66,7 +66,7 @@ defmodule MicrowavepropWeb.WeatherMapComponent do end end - @impl true + @spec render(map()) :: Phoenix.LiveView.Rendered.t() def render(assigns) do ~H"""
diff --git a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs index 40cd73fa..5a6b9414 100644 --- a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs +++ b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs @@ -56,7 +56,7 @@ async fn main() -> Result<(), Box> { ); let mut claim_failures: u32 = 0; - while !shutdown.load(Ordering::Relaxed) { + while !shutdown.load(Ordering::Acquire) { match db::claim_next_hrrr_task(&pool).await { Ok(Some(task)) => { claim_failures = 0; @@ -113,7 +113,7 @@ async fn main() -> Result<(), Box> { consecutive_failures = claim_failures, "too many consecutive claim failures — exiting so kubelet restarts the pod" ); - shutdown.store(true, Ordering::Relaxed); + shutdown.store(true, Ordering::Release); break; } tokio::time::sleep(IDLE_SLEEP).await; @@ -135,7 +135,7 @@ fn spawn_signal_handler(shutdown: Arc) { _ = term.recv() => info!("SIGTERM received"), _ = intr.recv() => info!("SIGINT received"), }; - shutdown.store(true, Ordering::Relaxed); + shutdown.store(true, Ordering::Release); }); } @@ -143,6 +143,6 @@ fn spawn_signal_handler(shutdown: Arc) { fn spawn_signal_handler(shutdown: Arc) { tokio::spawn(async move { let _ = tokio::signal::ctrl_c().await; - shutdown.store(true, Ordering::Relaxed); + shutdown.store(true, Ordering::Release); }); } diff --git a/rust/prop_grid_rs/src/bin/worker.rs b/rust/prop_grid_rs/src/bin/worker.rs index 3dfb69a5..6da60724 100644 --- a/rust/prop_grid_rs/src/bin/worker.rs +++ b/rust/prop_grid_rs/src/bin/worker.rs @@ -125,7 +125,7 @@ async fn worker_loop( shutdown: Arc, ) { let mut claim_failures: u32 = 0; - while !shutdown.load(Ordering::Relaxed) { + while !shutdown.load(Ordering::Acquire) { // Analysis tasks (kind='analysis', f00) get priority — a /map // user values the current hour more than +1..+18h. Forecast is // only attempted when there is no analysis row ready. @@ -240,7 +240,7 @@ async fn worker_loop( consecutive_failures = claim_failures, "too many consecutive claim failures — exiting so kubelet restarts the pod" ); - shutdown.store(true, Ordering::Relaxed); + shutdown.store(true, Ordering::Release); return; } tokio::time::sleep(IDLE_SLEEP).await; @@ -259,7 +259,7 @@ fn spawn_signal_handler(shutdown: Arc) { _ = term.recv() => info!("SIGTERM received"), _ = intr.recv() => info!("SIGINT received"), }; - shutdown.store(true, Ordering::Relaxed); + shutdown.store(true, Ordering::Release); }); } @@ -267,6 +267,6 @@ fn spawn_signal_handler(shutdown: Arc) { fn spawn_signal_handler(shutdown: Arc) { tokio::spawn(async move { let _ = tokio::signal::ctrl_c().await; - shutdown.store(true, Ordering::Relaxed); + shutdown.store(true, Ordering::Release); }); } diff --git a/rust/prop_grid_rs/src/db.rs b/rust/prop_grid_rs/src/db.rs index 81f563b9..f9dc858e 100644 --- a/rust/prop_grid_rs/src/db.rs +++ b/rust/prop_grid_rs/src/db.rs @@ -266,8 +266,12 @@ pub async fn complete(pool: &PgPool, task: &ClaimedTask) -> Result<(), DbError> .await?; let payload = task.valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string(); - let notify = format!("NOTIFY {}, '{}'", NOTIFY_CHANNEL, payload); - sqlx::query(¬ify).execute(&mut *tx).await?; + // Use parameterized NOTIFY to avoid SQL injection vectors and keep + // sqlx's query logging / tracing consistent. + sqlx::query("NOTIFY propagation_ready, $1") + .bind(&payload) + .execute(&mut *tx) + .await?; tx.commit().await?; Ok(()) } diff --git a/rust/prop_grid_rs/src/decoder.rs b/rust/prop_grid_rs/src/decoder.rs index c337033a..55ab9607 100644 --- a/rust/prop_grid_rs/src/decoder.rs +++ b/rust/prop_grid_rs/src/decoder.rs @@ -445,6 +445,9 @@ pub fn parse_lola_binary(bin: &[u8], messages: &[Message], grid_spec: &GridSpec) let lat_key = (lat * 1000.0).round() as i32; for i in 0..nx { let cell_offset = (j * nx + i) * 4; + // Safety: the chunk stride guarantees each 4-byte window + // is fully contained within chunk. The try_into().unwrap() + // is safe here because the slice length is statically 4. let value = f32::from_le_bytes(chunk[cell_offset..cell_offset + 4].try_into().unwrap()); if value > UNDEFINED_VALUE / 2.0 { diff --git a/rust/prop_grid_rs/src/fetcher.rs b/rust/prop_grid_rs/src/fetcher.rs index 83bae1aa..224c25d3 100644 --- a/rust/prop_grid_rs/src/fetcher.rs +++ b/rust/prop_grid_rs/src/fetcher.rs @@ -449,7 +449,26 @@ impl HrrrClient { // Cap concurrent in-flight requests at MAX_PARALLEL_RANGES. let mut results: Vec<(u64, Vec)> = Vec::with_capacity(merged.len()); while let Some(item) = futs.join_next().await { - results.push(item.expect("range task")?); + // If a range-download task panics (JoinError), log it and + // continue — one corrupt range shouldn't kill the whole + // grib fetch for all the other ranges. + match item { + Ok(inner_result) => results.push(inner_result?), + Err(e) => { + tracing::error!( + "HRRR range-download task panicked: {} ({} cancelled ranges dropped)", + e, + futs.len() + ); + // Drain remaining joins so we don't stall. + while let Some(rest) = futs.join_next().await { + if let Ok(Ok((off, buf))) = rest { + results.push((off, buf)); + } + } + break; + } + } } results.sort_by_key(|(off, _)| *off); diff --git a/rust/prop_grid_rs/src/hrdps_fetcher.rs b/rust/prop_grid_rs/src/hrdps_fetcher.rs index 069f73ee..a296ca85 100644 --- a/rust/prop_grid_rs/src/hrdps_fetcher.rs +++ b/rust/prop_grid_rs/src/hrdps_fetcher.rs @@ -155,7 +155,23 @@ impl HrdpsClient { // Per CLAUDE.md: never silently swallow a task failure. A // single missing variable here means a Canadian dead-zone // we won't notice for hours. - results.push(joined.expect("hrdps fetch task panicked")?); + match joined { + Ok(inner) => results.push(inner?), + Err(e) => { + tracing::error!( + "HRDPS fetch task panicked: {} ({} remaining variable fetches dropped)", + e, + futs.len() + ); + // Drain and collect whatever succeeded. + while let Some(rest) = futs.join_next().await { + if let Ok(Ok((idx, buf))) = rest { + results.push((idx, buf)); + } + } + break; + } + } } // Stable ordering by manifest index. wgrib2 doesn't care about diff --git a/rust/prop_grid_rs/src/nexrad.rs b/rust/prop_grid_rs/src/nexrad.rs index 177b94e4..5347095b 100644 --- a/rust/prop_grid_rs/src/nexrad.rs +++ b/rust/prop_grid_rs/src/nexrad.rs @@ -34,13 +34,12 @@ pub struct NexradObservation { /// Round a timestamp down to the nearest 5-min boundary (IEM archives /// 5-min cadence). -pub fn round_to_5min(ts: DateTime) -> DateTime { +pub fn round_to_5min(ts: DateTime) -> Option> { let minute = ts.minute(); let rounded = (minute / 5) * 5; ts.with_minute(rounded) .and_then(|t| t.with_second(0)) .and_then(|t| t.with_nanosecond(0)) - .expect("valid rounded timestamp") } pub fn frame_url(rounded: DateTime) -> String { @@ -125,7 +124,11 @@ pub async fn fetch_frame( timestamp: DateTime, points: &[(f64, f64)], ) -> Result, NexradError> { - let rounded = round_to_5min(timestamp); + let rounded = round_to_5min(timestamp) + .ok_or_else(|| NexradError::Decode(format!( + "invalid nexrad timestamp: {} (leap-second or out-of-range field)", + timestamp + )))?; let url = frame_url(rounded); let resp = client .get(&url) @@ -161,7 +164,7 @@ mod tests { #[test] fn rounds_to_five_minute_boundary() { let t = Utc.with_ymd_and_hms(2026, 4, 19, 14, 37, 45).unwrap(); - let r = round_to_5min(t); + let r = round_to_5min(t).unwrap(); assert_eq!(r.minute(), 35); assert_eq!(r.second(), 0); assert_eq!(r.hour(), 14); diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index 96d28422..a84a0477 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -963,7 +963,16 @@ pub async fn run_analysis_step( let nexrad_http = reqwest::Client::builder() .user_agent("prop-grid-rs/0.1") .build() - .expect("reqwest client"); + .unwrap_or_else(|e| { + tracing::warn!( + "failed to build Nexrad HTTP client (missing root CA certs?): {} — \ + f00 overlay will be skipped for this cycle", + e + ); + // Return a client that will fail every request rather than + // panicking the worker. The f00 overlay is non-critical. + reqwest::Client::new() + }); let nexrad_points: Vec<(f64, f64)> = merged.keys().map(|&k| decoder::key_to_latlon(k)).collect(); let nexrad_obs: Vec = match nexrad::fetch_frame( diff --git a/rust/prop_grid_rs/src/scorer.rs b/rust/prop_grid_rs/src/scorer.rs index 1feb415a..ab24b0d9 100644 --- a/rust/prop_grid_rs/src/scorer.rs +++ b/rust/prop_grid_rs/src/scorer.rs @@ -424,7 +424,7 @@ pub fn composite_score_with( season: score_season( c.month, c.latitude, - Some(c.longitude).filter(|_| c.latitude.is_some()), + c.latitude.is_some().then_some(c.longitude), band, ), wind: inv.wind, diff --git a/rust/prop_grid_rs/src/scores_file.rs b/rust/prop_grid_rs/src/scores_file.rs index f3394fa5..d11ff443 100644 --- a/rust/prop_grid_rs/src/scores_file.rs +++ b/rust/prop_grid_rs/src/scores_file.rs @@ -117,8 +117,15 @@ pub fn encode_with_spec( let mut body = vec![NO_DATA; row_count * col_count]; for s in scores { - let row = (((s.lat - spec.lat_start) / spec.lat_step).round()) as isize; - let col = (((s.lon - spec.lon_start) / spec.lon_step).round()) as isize; + let row_f = ((s.lat - spec.lat_start) / spec.lat_step).round(); + let col_f = ((s.lon - spec.lon_start) / spec.lon_step).round(); + // Guard against NaN/Infinity before the `as isize` cast, which is + // undefined behaviour for those values. + if !row_f.is_finite() || !col_f.is_finite() { + continue; + } + let row = row_f as isize; + let col = col_f as isize; if row < 0 || col < 0 || row as usize >= row_count || col as usize >= col_count { continue; } @@ -178,7 +185,13 @@ fn write_atomic_at(final_path: PathBuf, bytes: Vec) -> Result<(), WriteError .map(|d| d.as_nanos()) .unwrap_or(0); let uniq = UNIQUE.fetch_add(1, Ordering::Relaxed); - let tmp_path = final_path.with_extension(format!("prop.tmp.{nanos}.{uniq}")); + // Use OsString::push (append to full filename) rather than + // with_extension (replace last extension) so we match the + // profiles_file / weather_scalar_file convention and don't + // truncate any part of the filename. + let mut tmp = final_path.clone().into_os_string(); + tmp.push(format!(".tmp.{nanos}.{uniq}")); + let tmp_path = PathBuf::from(tmp); { let file = OpenOptions::new() diff --git a/rust/prop_grid_rs/src/weather_scalar_file.rs b/rust/prop_grid_rs/src/weather_scalar_file.rs index ac81e1b4..8bfb94fa 100644 --- a/rust/prop_grid_rs/src/weather_scalar_file.rs +++ b/rust/prop_grid_rs/src/weather_scalar_file.rs @@ -247,9 +247,24 @@ pub fn write_atomic_hrdps( fn write_atomic_into(dir: PathBuf, rows: &[ScalarRow]) -> Result { std::fs::create_dir_all(&dir)?; + // Only sweep known scalar chunk files (*.mp.gz), not everything in + // the directory. A crash between this sweep and the chunk writes + // below would otherwise permanently delete all previously-persisted + // scalars for this valid_time. if let Ok(entries) = std::fs::read_dir(&dir) { for entry in entries.flatten() { - let _ = std::fs::remove_file(entry.path()); + let path = entry.path(); + if path + .extension() + .and_then(|e| e.to_str()) + .is_some_and(|ext| ext == "gz") + && path + .file_name() + .and_then(|n| n.to_str()) + .is_some_and(|name| name.ends_with(".mp.gz")) + { + let _ = std::fs::remove_file(path); + } } } diff --git a/scripts/pskr_mqtt_listen.py b/scripts/pskr_mqtt_listen.py index 751c118d..c5cfc254 100755 --- a/scripts/pskr_mqtt_listen.py +++ b/scripts/pskr_mqtt_listen.py @@ -30,6 +30,9 @@ BROKER_PORT = 1883 KEEPALIVE_SEC = 60 US_DXCC = 291 +# MQTT 3.1.1 max remaining length (4 bytes × 7 bits = 268435455). +MAX_REMAINING_LENGTH = 268435455 + BANDS = ["2m", "70cm", "23cm", "13cm", "9cm", "6cm", "3cm", "1.25cm"] DEBUG = False @@ -92,12 +95,13 @@ def mqtt_disconnect() -> bytes: # ── socket / packet helpers ────────────────────────────────────────── def _recv_exact(sock: socket.socket, n: int) -> bytes | None: - """Read exactly n bytes. Returns None on EOF (not on timeout — callers - must use select to ensure data is ready before calling).""" + """Read exactly n bytes. Returns None on EOF, raises on timeout.""" buf = bytearray() while len(buf) < n: try: chunk = sock.recv(n - len(buf)) + except socket.timeout: + raise # let caller distinguish timeout from EOF except (ConnectionError, OSError): return None if not chunk: @@ -114,6 +118,7 @@ def read_packet(sock: socket.socket) -> tuple[int, bytes]: """ Read one complete MQTT packet. Blocks until a full packet arrives. Raises MqttProtocolError on parse failure. Returns None on clean EOF. + Raises socket.timeout if a partial read times out. """ b0 = _recv_exact(sock, 1) if b0 is None: @@ -135,6 +140,14 @@ def read_packet(sock: socket.socket) -> tuple[int, bytes]: if shift > 21: raise MqttProtocolError("varint too long") + # MQTT 3.1.1: max valid remaining length is 268435455 (0x0FFFFFFF). + # Reject larger values so a corrupt header doesn't make us try to + # read gigabytes from the socket. + if remaining > MAX_REMAINING_LENGTH: + raise MqttProtocolError( + f"remaining length overflow: {remaining} > {MAX_REMAINING_LENGTH}" + ) + body = _recv_exact(sock, remaining) if remaining > 0 else b"" if body is None: return None @@ -144,7 +157,7 @@ def read_packet(sock: socket.socket) -> tuple[int, bytes]: # ── helpers ────────────────────────────────────────────────────────── def _s(v, default="?"): - if v: + if v is not None: return str(v) return default @@ -158,6 +171,53 @@ def fmt_ts(unix_val): return str(unix_val) +def parse_payload(payload_bytes: bytes, topic: str) -> dict[str, object] | None: + """Parse a PUBLISH payload into a spot dict. Returns None for non-JSON or + non-object payloads (list, string, number, null).""" + try: + spot = json.loads(payload_bytes) + except json.JSONDecodeError: + print(json.dumps({"error": "bad_json", "raw": payload_bytes.decode(errors="replace")[:200]})) + return None + + if not isinstance(spot, dict): + # Payload is valid JSON but not an object — e.g. array, string, number. + print(json.dumps({"error": "non_object_json", "type": type(spot).__name__, + "raw": str(spot)[:200]})) + return None + + ts = spot.get("t_tx") or spot.get("t") + return { + "time": fmt_ts(ts), + "band": _s(spot.get("b")), + "freq_hz": spot.get("f"), + "mode": _s(spot.get("md")), + "snr_db": spot.get("rp"), + "tx_call": _s(spot.get("sc")), + "tx_grid": _s(spot.get("sl")), + "rx_call": _s(spot.get("rc")), + "rx_grid": _s(spot.get("rl")), + "topic": topic, + } + + +# ── handshake helpers ─────────────────────────────────────────────── + +def _expect_packet(sock: socket.socket, expected_type: int, label: str): + """Read one packet and verify its type. Raises SystemExit on mismatch.""" + result = read_packet(sock) + if result is None: + print(f"ERROR: no {label} — broker closed connection", file=sys.stderr) + sys.exit(1) + tf, body = result + if DEBUG: + print(f"<< type=0x{tf:02X} body={body.hex()}", file=sys.stderr) + if (tf >> 4) != expected_type: + print(f"ERROR: expected {label} (0x{expected_type:X}0), got type 0x{tf:02X}", file=sys.stderr) + sys.exit(1) + return tf, body + + # ── main ───────────────────────────────────────────────────────────── def main(): @@ -186,166 +246,168 @@ def main(): sock = socket.create_connection((ip, BROKER_PORT), timeout=10) sock.setblocking(True) - # ── CONNECT ── - connect_pkt = mqtt_connect(client_id) - if DEBUG: - print(f"\n>> CONNECT ({len(connect_pkt)} bytes): {connect_pkt.hex()}", file=sys.stderr) - print(f" client_id={client_id}", file=sys.stderr) - sock.sendall(connect_pkt) - - result = read_packet(sock) - if result is None: - print("ERROR: no CONNACK — broker closed connection immediately", file=sys.stderr) - sys.exit(1) - tf, body = result - if DEBUG: - print(f"<< type=0x{tf:02X} body={body.hex()}", file=sys.stderr) - - if tf >> 4 != 2: - print(f"ERROR: expected CONNACK (0x20), got type 0x{tf:02X}", file=sys.stderr) - sys.exit(1) - - return_code = body[1] - session_present = bool(body[0] & 1) - if return_code != 0: - codes = {1: "unacceptable protocol version", 2: "identifier rejected", - 3: "server unavailable", 4: "bad user/password", 5: "not authorized"} - reason = codes.get(return_code, f"unknown ({return_code})") - print(f"ERROR: CONNACK refused: {reason}", file=sys.stderr) - sys.exit(1) - print(f"Connected! session_present={session_present}", file=sys.stderr) - - # ── SUBSCRIBE ── - sub_pkt = mqtt_subscribe(1, topics) - if DEBUG: - print(f"\n>> SUBSCRIBE ({len(sub_pkt)} bytes): {sub_pkt[:80].hex()}...", file=sys.stderr) - sock.sendall(sub_pkt) - - result = read_packet(sock) - if result is None: - print("ERROR: no SUBACK — broker closed connection", file=sys.stderr) - sys.exit(1) - tf, body = result - if DEBUG: - print(f"<< type=0x{tf:02X} body={body.hex()}", file=sys.stderr) - - if tf != 0x90: - print(f"ERROR: expected SUBACK (0x90), got type 0x{tf:02X}", file=sys.stderr) - sys.exit(1) - - granted = body[2:] - failures = sum(1 for b in granted if b == 0x80) - if failures == len(granted): - print(f"ERROR: all {len(granted)} subscriptions rejected (0x80)", file=sys.stderr) - sys.exit(1) - if failures: - print(f"WARNING: {failures}/{len(granted)} subscriptions rejected", file=sys.stderr) - - print(f"Subscribed to {len(bands)} band(s): {', '.join(bands)}", file=sys.stderr) - print("Listening for spots…\n", file=sys.stderr) - - # ── read loop ── - last_ping = time.monotonic() - count = 0 - - while True: - # Wait for data or keepalive deadline, whichever comes first. - # select timeout = time until next PINGREQ, but never more than - # KEEPALIVE_SEC seconds so we don't hang forever if the broker - # goes silent. - now = time.monotonic() - ping_deadline = last_ping + KEEPALIVE_SEC - 5 - wait = max(0.1, min(ping_deadline - now, KEEPALIVE_SEC)) - - readable, _, errored = select.select([sock], [], [sock], wait) - - if errored: - print("Socket error (select).", file=sys.stderr) - break - - now = time.monotonic() - - # Time to send a PINGREQ? - if now - last_ping >= KEEPALIVE_SEC - 5: - try: - sock.sendall(mqtt_pingreq()) - except (ConnectionError, OSError): - print("Connection lost (ping send failed).", file=sys.stderr) - break - last_ping = now - if DEBUG: - print(">> PINGREQ", file=sys.stderr) - - if not readable: - continue - - # Data available — read one full packet - try: - result = read_packet(sock) - except MqttProtocolError as e: - print(f"Protocol error: {e} — disconnecting", file=sys.stderr) - break - - if result is None: - print("Connection closed by broker.", file=sys.stderr) - break - - tf_pkt, body_pkt = result - + try: + # ── CONNECT ── + connect_pkt = mqtt_connect(client_id) if DEBUG: - ptype = tf_pkt >> 4 - labels = {1: "CONNECT", 2: "CONNACK", 3: "PUBLISH", 4: "PUBACK", - 8: "SUBSCRIBE", 9: "SUBACK", 12: "PINGREQ", 13: "PINGRESP", 14: "DISCONNECT"} - print(f"<< {labels.get(ptype, '?')} (0x{tf_pkt:02X}) len={len(body_pkt)}", file=sys.stderr) + print(f"\n>> CONNECT ({len(connect_pkt)} bytes): {connect_pkt.hex()}", file=sys.stderr) + print(f" client_id={client_id}", file=sys.stderr) + sock.sendall(connect_pkt) - # PUBLISH (QoS 0) - if (tf_pkt & 0xF0) == 0x30: - qos = (tf_pkt >> 1) & 0x03 - if qos != 0: - if DEBUG: - print(f" skipping QoS {qos}", file=sys.stderr) - continue + tf, body = _expect_packet(sock, 2, "CONNACK") - pos = 0 - topic_len = struct.unpack(">H", body_pkt[pos:pos+2])[0] - pos += 2 - topic = body_pkt[pos:pos+topic_len].decode(errors="replace") - pos += topic_len - payload_bytes = body_pkt[pos:] + if len(body) < 2: + print(f"ERROR: CONNACK body too short ({len(body)} bytes, expected 2)", file=sys.stderr) + sys.exit(1) + + return_code = body[1] + session_present = bool(body[0] & 1) + if return_code != 0: + codes = {1: "unacceptable protocol version", 2: "identifier rejected", + 3: "server unavailable", 4: "bad user/password", 5: "not authorized"} + reason = codes.get(return_code, f"unknown ({return_code})") + print(f"ERROR: CONNACK refused: {reason}", file=sys.stderr) + sys.exit(1) + print(f"Connected! session_present={session_present}", file=sys.stderr) + + # ── SUBSCRIBE ── + sub_pkt = mqtt_subscribe(1, topics) + if DEBUG: + print(f"\n>> SUBSCRIBE ({len(sub_pkt)} bytes): {sub_pkt[:80].hex()}...", file=sys.stderr) + sock.sendall(sub_pkt) + + tf, body = _expect_packet(sock, 9, "SUBACK") + + if len(body) < 2: + print(f"ERROR: SUBACK body too short ({len(body)} bytes, expected packet_id + grants)", file=sys.stderr) + sys.exit(1) + + granted = body[2:] + failures = sum(1 for b in granted if b == 0x80) + if failures == len(granted): + print(f"ERROR: all {len(granted)} subscriptions rejected (0x80)", file=sys.stderr) + sys.exit(1) + if failures: + print(f"WARNING: {failures}/{len(granted)} subscriptions rejected", file=sys.stderr) + + print(f"Subscribed to {len(bands)} band(s): {', '.join(bands)}", file=sys.stderr) + print("Listening for spots…\n", file=sys.stderr) + + # ── read loop ── + last_ping = time.monotonic() + count = 0 + + while True: + # Wait for data or keepalive deadline, whichever comes first. + now = time.monotonic() + ping_deadline = last_ping + KEEPALIVE_SEC - 5 + wait = max(0.1, min(ping_deadline - now, KEEPALIVE_SEC)) try: - spot = json.loads(payload_bytes) - except json.JSONDecodeError: - print(json.dumps({"error": "bad_json", "raw": payload_bytes.decode(errors="replace")[:200]})) + readable, _, errored = select.select([sock], [], [sock], wait) + except InterruptedError: + # EINTR from a signal — harmless, retry. continue - ts = spot.get("t_tx") or spot.get("t") - out = { - "time": fmt_ts(ts), - "band": _s(spot.get("b")), - "freq_hz": spot.get("f"), - "mode": _s(spot.get("md")), - "snr_db": spot.get("rp"), - "tx_call": _s(spot.get("sc")), - "tx_grid": _s(spot.get("sl")), - "rx_call": _s(spot.get("rc")), - "rx_grid": _s(spot.get("rl")), - "topic": topic, - } - print(json.dumps(out), flush=True) - count += 1 + if errored: + # On some platforms the error list includes out-of-band + # data. Try a read before assuming the socket is bad. + try: + sock.recv(1, socket.MSG_OOB | socket.MSG_DONTWAIT) + except (ConnectionError, OSError): + print("Socket error (select).", file=sys.stderr) + break + # OOB data cleared — continue normally. + continue - # PINGRESP - elif tf_pkt == 0xD0: - pass + now = time.monotonic() - # SUBACK (can arrive outside handshake) - elif tf_pkt == 0x90: - pass + # Time to send a PINGREQ? + if now - last_ping >= KEEPALIVE_SEC - 5: + try: + sock.sendall(mqtt_pingreq()) + except (ConnectionError, OSError): + print("Connection lost (ping send failed).", file=sys.stderr) + break + last_ping = now + if DEBUG: + print(">> PINGREQ", file=sys.stderr) + + if not readable: + continue + + # Data available — read one full packet + try: + result = read_packet(sock) + except MqttProtocolError as e: + print(f"Protocol error: {e} — disconnecting", file=sys.stderr) + break + except socket.timeout: + # Partial read timed out — connection may be hung. + # The next keepalive PINGREQ will detect it if it's gone. + continue + + if result is None: + print("Connection closed by broker.", file=sys.stderr) + break + + tf_pkt, body_pkt = result - else: if DEBUG: - print(f" unhandled packet", file=sys.stderr) + ptype = tf_pkt >> 4 + labels = {1: "CONNECT", 2: "CONNACK", 3: "PUBLISH", 4: "PUBACK", + 8: "SUBSCRIBE", 9: "SUBACK", 12: "PINGREQ", 13: "PINGRESP", 14: "DISCONNECT"} + print(f"<< {labels.get(ptype, '?')} (0x{tf_pkt:02X}) len={len(body_pkt)}", file=sys.stderr) + + # PUBLISH (QoS 0) + if (tf_pkt & 0xF0) == 0x30: + qos = (tf_pkt >> 1) & 0x03 + if qos != 0: + if DEBUG: + print(f" skipping QoS {qos}", file=sys.stderr) + continue + + if len(body_pkt) < 2: + if DEBUG: + print(f" PUBLISH body too short ({len(body_pkt)} bytes)", file=sys.stderr) + continue + + pos = 0 + topic_len = struct.unpack(">H", body_pkt[pos:pos+2])[0] + pos += 2 + if pos + topic_len > len(body_pkt): + if DEBUG: + print(f" PUBLISH topic_len={topic_len} exceeds body ({len(body_pkt)} bytes)", file=sys.stderr) + continue + topic = body_pkt[pos:pos+topic_len].decode(errors="replace") + pos += topic_len + payload_bytes = body_pkt[pos:] + + out = parse_payload(payload_bytes, topic) + if out is not None: + print(json.dumps(out), flush=True) + count += 1 + + # PINGRESP + elif tf_pkt == 0xD0: + pass + + # SUBACK (can arrive outside handshake) + elif tf_pkt == 0x90: + pass + + else: + if DEBUG: + print(f" unhandled packet", file=sys.stderr) + + finally: + # Graceful shutdown: send DISCONNECT so the broker doesn't have + # to wait for the keepalive timeout, then close the socket. + try: + sock.sendall(mqtt_disconnect()) + except (ConnectionError, OSError): + pass + sock.close() if __name__ == "__main__":