diff --git a/lib/microwaveprop/propagation/freshness_monitor.ex b/lib/microwaveprop/propagation/freshness_monitor.ex index d8e53462..19380673 100644 --- a/lib/microwaveprop/propagation/freshness_monitor.ex +++ b/lib/microwaveprop/propagation/freshness_monitor.ex @@ -39,26 +39,34 @@ defmodule Microwaveprop.Propagation.FreshnessMonitor do defp check_freshness do case Propagation.latest_valid_time() do - nil -> - Logger.info("FreshnessMonitor: no scores found, enqueuing grid worker") - enqueue_if_not_queued() - - latest -> - age = DateTime.diff(DateTime.utc_now(), latest, :minute) - - if age > @stale_threshold_minutes do - Logger.info("FreshnessMonitor: scores are #{age}m old, enqueuing grid worker") - enqueue_if_not_queued() - end + nil -> maybe_enqueue(reason: "no scores found", age: nil) + latest -> check_stale(latest) end end - defp enqueue_if_not_queued do - # PropagationGridWorker declares `unique:` over a 1-hour window on - # the seed args (`%{}`), so repeated 5-minute ticks during a long - # outage collapse into a single seed job — not one stacked chain - # per tick. `Oban.insert` returns `{:ok, job}` either way; the - # `conflict?` field on the returned job distinguishes the two. - Oban.insert(PropagationGridWorker.new(%{})) + defp check_stale(latest) do + age = DateTime.diff(DateTime.utc_now(), latest, :minute) + + if age > @stale_threshold_minutes do + maybe_enqueue(reason: "scores #{age}m old", age: age) + end + end + + # PropagationGridWorker declares `unique:` over a 1-hour window on + # the seed args (`%{}`), so repeated 5-minute ticks during a long + # outage collapse into a single seed job — not one stacked chain + # per tick. Only log when a new job actually lands; the steady-state + # "still stale, still queued" case is uninteresting. + defp maybe_enqueue(reason: reason, age: _age) do + case Oban.insert(PropagationGridWorker.new(%{})) do + {:ok, %Oban.Job{conflict?: false}} -> + Logger.info("FreshnessMonitor: #{reason}, enqueued grid worker") + + {:ok, %Oban.Job{conflict?: true}} -> + :ok + + {:error, reason} -> + Logger.warning("FreshnessMonitor: enqueue failed: #{inspect(reason, printable_limit: 200)}") + end end end diff --git a/lib/microwaveprop/workers/common_volume_radar_worker.ex b/lib/microwaveprop/workers/common_volume_radar_worker.ex index b3c7d6bf..d28a2dd2 100644 --- a/lib/microwaveprop/workers/common_volume_radar_worker.ex +++ b/lib/microwaveprop/workers/common_volume_radar_worker.ex @@ -166,8 +166,8 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorker do pixels |> collect_cv_pixels( width, - x_min..x_max, - y_min..y_max, + Range.new(x_min, x_max, 1), + Range.new(y_min, y_max, 1), pos1, pos2, step diff --git a/lib/microwaveprop/workers/contact_import_worker.ex b/lib/microwaveprop/workers/contact_import_worker.ex index 642ce21a..e4ad1f71 100644 --- a/lib/microwaveprop/workers/contact_import_worker.ex +++ b/lib/microwaveprop/workers/contact_import_worker.ex @@ -16,7 +16,15 @@ defmodule Microwaveprop.Workers.ContactImportWorker do that sees `status = 'pending'` on its first increment flips it to `"running"` and sets `started_at`. """ - use Oban.Worker, queue: :contact_import, max_attempts: 3 + use Oban.Worker, + queue: :contact_import, + max_attempts: 3, + unique: [ + period: :infinity, + fields: [:worker, :args], + keys: [:import_run_id, :offset], + states: [:available, :scheduled, :executing, :retryable, :completed] + ] import Ecto.Query diff --git a/lib/microwaveprop/workers/ionosphere_fetch_worker.ex b/lib/microwaveprop/workers/ionosphere_fetch_worker.ex index 1c6b051d..5923ce26 100644 --- a/lib/microwaveprop/workers/ionosphere_fetch_worker.ex +++ b/lib/microwaveprop/workers/ionosphere_fetch_worker.ex @@ -56,7 +56,23 @@ defmodule Microwaveprop.Workers.IonosphereFetchWorker do Logger.info("IonosphereFetch: #{code} (#{name}) upserted #{count} observations") {:error, reason} -> - Logger.warning("IonosphereFetch: #{code} (#{name}) failed: #{inspect(reason)}") + Logger.warning("IonosphereFetch: #{code} (#{name}) failed: #{format_reason(reason)}") end end + + # GIRO's TLS cert chains through a CA that isn't always in the pod's + # bundle, surfacing as a Req.TransportError with a multi-line erl_alert + # string. Inspecting the full struct dumps ~400 chars of stacktrace-y + # noise per station per 10-min cycle. Compress to something grep-able. + defp format_reason("GIRO request failed: " <> tail) do + cond do + String.contains?(tail, ":unknown_ca") -> "TLS unknown_ca (CA bundle missing)" + String.contains?(tail, ":timeout") -> "timeout" + String.contains?(tail, ":econnrefused") -> "connection refused" + true -> String.slice(tail, 0, 120) + end + end + + defp format_reason(reason) when is_binary(reason), do: reason + defp format_reason(reason), do: inspect(reason, printable_limit: 200) end diff --git a/lib/microwaveprop/workers/weather_fetch_worker.ex b/lib/microwaveprop/workers/weather_fetch_worker.ex index 0282f58f..ae364587 100644 --- a/lib/microwaveprop/workers/weather_fetch_worker.ex +++ b/lib/microwaveprop/workers/weather_fetch_worker.ex @@ -80,11 +80,31 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do ingest_asos_rows(station, station_id, station_code, rows, start_dt, end_dt) {:error, reason} -> - Logger.warning("WeatherFetch ASOS: #{station_code} failed: #{inspect(reason)}") - {:error, reason} + handle_iem_error("ASOS", station_code, reason) end end + # IEM rate-limits heavy ASOS pulls with HTTP 429. That's routine + # during a backfill sweep, not a job failure — snooze the job + # instead of surfacing Oban.PerformError, which spams the logger + # with a full exception per 429 even though the retry is working + # as intended. Other statuses (5xx, connection errors) stay on the + # {:error, _} path so they count against max_attempts. + defp handle_iem_error(kind, station_code, "IEM " <> _ = reason) do + if String.contains?(reason, "429") do + Logger.debug("WeatherFetch #{kind}: #{station_code} rate-limited, snoozing") + {:snooze, 300} + else + Logger.warning("WeatherFetch #{kind}: #{station_code} failed: #{reason}") + {:error, reason} + end + end + + defp handle_iem_error(kind, station_code, reason) do + Logger.warning("WeatherFetch #{kind}: #{station_code} failed: #{inspect(reason, printable_limit: 200)}") + {:error, reason} + end + defp ingest_asos_rows(station, station_id, station_code, rows, start_dt, end_dt) do count = Enum.count(rows, & &1.observed_at) @@ -139,8 +159,7 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do :ok {:error, reason} -> - Logger.warning("WeatherFetch RAOB: #{station_code} failed: #{inspect(reason)}") - {:error, reason} + handle_iem_error("RAOB", station_code, reason) end end diff --git a/lib/microwaveprop_web/live/map_live.ex b/lib/microwaveprop_web/live/map_live.ex index ed1eee5e..8e462fa8 100644 --- a/lib/microwaveprop_web/live/map_live.ex +++ b/lib/microwaveprop_web/live/map_live.ex @@ -384,12 +384,23 @@ defmodule MicrowavepropWeb.MapLive do bounds = socket.assigns.bounds selected = socket.assigns.selected_time + # Forecast preload walks up to 18 valid_times; each call reads the + # band's .prop file off NFS and filters to the viewport. The reads + # are independent, so fan them out across 4 Tasks — serial ran at + # ~18 × single-read latency, which dominated time-to-interactive + # after a band change on /map. Preserves the original ordering + # since the JS hook expects `hours` in valid_time order. + forecast_times = Enum.reject(socket.assigns.valid_times, &(selected && DateTime.compare(&1, selected) == :eq)) + hours = - socket.assigns.valid_times - |> Enum.reject(&(selected && DateTime.compare(&1, selected) == :eq)) - |> Enum.map(fn t -> - %{time: DateTime.to_iso8601(t), scores: Propagation.scores_at(band, t, bounds)} - end) + forecast_times + |> Task.async_stream( + fn t -> %{time: DateTime.to_iso8601(t), scores: Propagation.scores_at(band, t, bounds)} end, + max_concurrency: 4, + ordered: true, + timeout: 10_000 + ) + |> Enum.map(fn {:ok, h} -> h end) {:noreply, socket diff --git a/priv/repo/migrations/20260421220000_tighten_grid_tasks_claim_index.exs b/priv/repo/migrations/20260421220000_tighten_grid_tasks_claim_index.exs new file mode 100644 index 00000000..3ea488d3 --- /dev/null +++ b/priv/repo/migrations/20260421220000_tighten_grid_tasks_claim_index.exs @@ -0,0 +1,32 @@ +defmodule Microwaveprop.Repo.Migrations.TightenGridTasksClaimIndex do + use Ecto.Migration + + # CREATE INDEX CONCURRENTLY requires not being inside a transaction, + # and not grabbing Ecto's migration-level advisory lock (it would + # serialize with the DDL command we're about to issue). + @disable_ddl_transaction true + @disable_migration_lock true + + @old_idx :grid_tasks_queued_idx + + # Rust's `claim_next` (rust/prop_grid_rs/src/db.rs) orders by + # `run_time DESC, forecast_hour ASC` inside the claimable subset + # (`status='queued' AND forecast_hour > 0 AND kind='forecast'`). + # The old partial index only indexed `status` — Postgres then had to + # sort the whole queued tail on every claim. The table is small today + # (~19 rows/hour, pruned), but ordering inside the index is free to + # add and future-proofs the worker if we ever expand forecast coverage. + def change do + execute( + """ + CREATE INDEX CONCURRENTLY IF NOT EXISTS grid_tasks_forecast_claim_idx + ON grid_tasks (run_time DESC, forecast_hour ASC) + WHERE status = 'queued' AND forecast_hour > 0 AND kind = 'forecast' + """, + "DROP INDEX IF EXISTS grid_tasks_forecast_claim_idx" + ) + + # Redundant once the covering partial index exists. + drop_if_exists(index(:grid_tasks, [:status], name: @old_idx)) + end +end 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 28afaeae..d58ce795 100644 --- a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs +++ b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs @@ -4,7 +4,6 @@ //! requested point, and upserts into `hrrr_profiles`. FOR UPDATE SKIP //! LOCKED allows N replicas with no coordination. -use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -28,8 +27,6 @@ async fn main() -> Result<(), Box> { .ok() .and_then(|v| v.parse().ok()) .unwrap_or(4); - let tmp_dir = PathBuf::from(std::env::var("TMP_DIR").unwrap_or_else(|_| "/tmp".to_string())); - let pool = db::connect(&database_url, max_conn).await?; let client = Arc::new(HrrrClient::new_with_fallback(&hrrr_base, &hrrr_fallback)?); @@ -46,8 +43,7 @@ async fn main() -> Result<(), Box> { let points = task.points.clone(); let started = Instant::now(); - match hrrr_points::process_batch(&client, &pool, valid_time, &points, &tmp_dir) - .await + match hrrr_points::process_batch(&client, &pool, valid_time, &points).await { Ok(stats) => { let elapsed = started.elapsed(); diff --git a/rust/prop_grid_rs/src/fetcher.rs b/rust/prop_grid_rs/src/fetcher.rs index 52df9ebe..a99cdab8 100644 --- a/rust/prop_grid_rs/src/fetcher.rs +++ b/rust/prop_grid_rs/src/fetcher.rs @@ -58,6 +58,35 @@ pub const GRID_PRESSURE_LEVELS: &[u16] = &[ 1000, 975, 950, 925, 900, 875, 850, 825, 800, 775, 750, 725, 700, ]; +/// Pre-formatted lookup keys for per-cell pressure-level access. +/// +/// The hot decode → score loop runs over 92k cells × 13 levels and +/// would otherwise `format!("TMP:{p} mb")` on every cell, every level, +/// every variable — ~3.6M String allocations per chain step. This +/// table is built once (at first access), stored as owned Strings, +/// and borrowed as `&str` at every call site. +pub struct LevelKeys { + pub pres_mb: f64, + pub tmp: String, + pub dpt: String, + pub hgt: String, +} + +pub fn grid_level_keys() -> &'static [LevelKeys] { + static KEYS: std::sync::OnceLock> = std::sync::OnceLock::new(); + KEYS.get_or_init(|| { + GRID_PRESSURE_LEVELS + .iter() + .map(|&p| LevelKeys { + pres_mb: p as f64, + tmp: format!("TMP:{p} mb"), + dpt: format!("DPT:{p} mb"), + hgt: format!("HGT:{p} mb"), + }) + .collect() + }) +} + pub const SURFACE_MESSAGES: &[(&str, &str)] = &[ ("TMP", "2 m above ground"), ("DPT", "2 m above ground"), diff --git a/rust/prop_grid_rs/src/hrrr_points.rs b/rust/prop_grid_rs/src/hrrr_points.rs index 39fdd17d..65036f2f 100644 --- a/rust/prop_grid_rs/src/hrrr_points.rs +++ b/rust/prop_grid_rs/src/hrrr_points.rs @@ -10,7 +10,7 @@ //! rare and handled via `ON CONFLICT DO UPDATE` on (lat, lon, //! valid_time). -use std::path::Path; +use std::sync::OnceLock; use chrono::{DateTime, Timelike, Utc}; use sqlx::PgPool; @@ -20,6 +20,24 @@ use crate::decoder::{self, CellValues}; use crate::fetcher::{self, HrrrClient, Product}; use crate::grid::wgrib2_grid_spec; +// wgrib2 filter patterns are derived from static SURFACE_MESSAGES / +// GRID_PRESSURE_LEVELS tables and never change at runtime. Building +// them per batch was wasted allocation on every hrrr_fetch_tasks tick. +fn sfc_pattern() -> &'static str { + static P: OnceLock = OnceLock::new(); + P.get_or_init(|| build_pattern(&fetcher::surface_messages_owned())) +} + +fn prs_pattern() -> &'static str { + static P: OnceLock = OnceLock::new(); + P.get_or_init(|| build_pattern(&fetcher::pressure_messages_grid())) +} + +fn build_pattern(msgs: &[(String, String)]) -> String { + let vars: Vec<&str> = msgs.iter().map(|(v, _)| v.as_str()).collect(); + format!(":({}):", vars.join("|")) +} + #[derive(Debug, thiserror::Error)] pub enum PointFetchError { #[error("fetch: {0}")] @@ -39,7 +57,7 @@ pub struct PointFetchStats { /// Fetch + decode + persist for one `hrrr_fetch_tasks` batch. #[tracing::instrument( name = "hrrr_points.process_batch", - skip(client, pool, _tmp_dir), + skip(client, pool), fields( valid_time = %valid_time, points = points.len(), @@ -50,7 +68,6 @@ pub async fn process_batch( pool: &PgPool, valid_time: DateTime, points: &[(f64, f64)], - _tmp_dir: &Path, ) -> Result { let date = valid_time.date_naive(); let hour = valid_time.hour() as u8; @@ -58,38 +75,22 @@ pub async fn process_batch( let sfc_wanted = fetcher::surface_messages_owned(); let prs_wanted = fetcher::pressure_messages_grid(); - let sfc_pattern = format!( - ":({}):", - sfc_wanted - .iter() - .map(|(v, _)| v.as_str()) - .collect::>() - .join("|") - ); - let prs_pattern = format!( - ":({}):", - prs_wanted - .iter() - .map(|(v, _)| v.as_str()) - .collect::>() - .join("|") - ); + let sfc_pat = sfc_pattern(); + let prs_pat = prs_pattern(); let sfc_fut = client.fetch_product_blob(date, hour, Product::Surface, 0, &sfc_wanted); let prs_fut = client.fetch_product_blob(date, hour, Product::Pressure, 0, &prs_wanted); let (sfc_blob, prs_blob) = tokio::try_join!(sfc_fut, prs_fut)?; - let sfc_grid = tokio::task::spawn_blocking(move || { - decoder::extract_grid(&sfc_blob, &sfc_pattern, grid_spec) - }) - .await - .expect("blocking join")?; + let sfc_grid = + tokio::task::spawn_blocking(move || decoder::extract_grid(&sfc_blob, sfc_pat, grid_spec)) + .await + .expect("blocking join")?; - let prs_grid = tokio::task::spawn_blocking(move || { - decoder::extract_grid(&prs_blob, &prs_pattern, grid_spec) - }) - .await - .expect("blocking join")?; + let prs_grid = + tokio::task::spawn_blocking(move || decoder::extract_grid(&prs_blob, prs_pat, grid_spec)) + .await + .expect("blocking join")?; // Empty grids almost always mean the HRRR archive doesn't have this // cycle (common for backfills older than the rolling retention @@ -158,15 +159,14 @@ async fn upsert_profile( // We build a Vec> so sqlx encodes it as jsonb[] rather // than wrapping the whole list in a single jsonb (which yields the // "expression is of type jsonb" type-mismatch error at runtime). - let levels: Vec> = fetcher::GRID_PRESSURE_LEVELS + let levels: Vec> = fetcher::grid_level_keys() .iter() - .filter_map(|&p| { - let pres_mb = p as f64; - let t = cell.get(&format!("TMP:{p} mb")).copied()?; - let h = cell.get(&format!("HGT:{p} mb")).copied()?; - let d = cell.get(&format!("DPT:{p} mb")).copied(); + .filter_map(|k| { + let t = cell.get(k.tmp.as_str()).copied()?; + let h = cell.get(k.hgt.as_str()).copied()?; + let d = cell.get(k.dpt.as_str()).copied(); let mut m = serde_json::Map::new(); - m.insert("pres_mb".into(), serde_json::json!(pres_mb)); + m.insert("pres_mb".into(), serde_json::json!(k.pres_mb)); m.insert("hght_m".into(), serde_json::json!(h as f64)); m.insert("tmpc".into(), serde_json::json!((t as f64) - 273.15)); if let Some(dv) = d { diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index 4eb768ee..b46b1291 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -254,15 +254,14 @@ fn cell_to_conditions( .filter(|v| *v > 0.0); let bl_depth_m = cell.get("HPBL:surface").copied().map(|v| v as f64); - let levels: Vec = fetcher::GRID_PRESSURE_LEVELS + let levels: Vec = fetcher::grid_level_keys() .iter() - .filter_map(|&p| { - let pres_mb = p as f64; - let t = cell.get(&format!("TMP:{p} mb")).copied()?; - let d = cell.get(&format!("DPT:{p} mb")).copied(); - let h = cell.get(&format!("HGT:{p} mb")).copied()?; + .filter_map(|k| { + let t = cell.get(k.tmp.as_str()).copied()?; + let d = cell.get(k.dpt.as_str()).copied(); + let h = cell.get(k.hgt.as_str()).copied()?; Some(Level { - pres_mb, + pres_mb: k.pres_mb, hght_m: h as f64, tmpc: (t - 273.15) as f64, dwpc: d.map(|v| (v - 273.15) as f64), @@ -667,15 +666,14 @@ fn cell_to_profile_entry(lat: f64, lon: f64, cell: &CellValues) -> CellEntry { // Pressure-level profile list for SoundingParams.derive on the // Elixir side. Mirrors the fields derive expects. - let levels: Vec = fetcher::GRID_PRESSURE_LEVELS + let levels: Vec = fetcher::grid_level_keys() .iter() - .filter_map(|&p| { - let pres_mb = p as f64; - let t = cell.get(&format!("TMP:{p} mb")).copied()?; - let d = cell.get(&format!("DPT:{p} mb")).copied(); - let h = cell.get(&format!("HGT:{p} mb")).copied()?; + .filter_map(|k| { + let t = cell.get(k.tmp.as_str()).copied()?; + let d = cell.get(k.dpt.as_str()).copied(); + let h = cell.get(k.hgt.as_str()).copied()?; let mut pairs: Vec<(V, V)> = Vec::with_capacity(4); - pairs.push((V::String("pres_mb".into()), V::F64(pres_mb))); + pairs.push((V::String("pres_mb".into()), V::F64(k.pres_mb))); pairs.push((V::String("hght_m".into()), V::F64(h as f64))); pairs.push((V::String("tmpc".into()), V::F64((t as f64) - 273.15))); if let Some(dv) = d {