perf+hygiene: batch 1 of system-review fixes

Batched from a system-wide review pass.

**Rate-limit + transient-error log hygiene**
- WeatherFetchWorker: classify IEM HTTP 429 as {:snooze, 300} instead
  of {:error}. Stops Oban.PerformError stack-trace spam on every
  routine rate-limit during backfill, keeps the retry semantics.
- IonosphereFetchWorker: compress GIRO TLS :unknown_ca error from a
  ~400-char inspect blob to 'TLS unknown_ca (CA bundle missing)'.
- FreshnessMonitor: log only when an enqueue actually lands (the
  Oban unique conflict case was silently dropping jobs but still
  producing 'enqueuing grid worker' info lines every 5 minutes).

**Perf**
- Rust grid_level_keys(): pre-format 'TMP:{p} mb' / DPT / HGT keys
  once via OnceLock instead of format!()'ing per-cell. Removes ~3.6M
  String allocations per f01..f18 chain step across pipeline.rs,
  hrrr_points.rs. Same for the wgrib2 :(TMP|DPT|HGT): pattern in
  hrrr_points.process_batch (sfc_pattern / prs_pattern OnceLock).
- MapLive preload_forecast: Task.async_stream with max_concurrency=4
  replaces the 18-wide Enum.map serial walk. Forecast cache warms
  ~4× faster after a band change, with ordering preserved.
- grid_tasks: partial composite index on (run_time DESC,
  forecast_hour ASC) WHERE status='queued' AND kind='forecast',
  matching the Rust claim query's ORDER BY. Drops the old
  status-only partial that forced a sort per claim.

**Correctness**
- ContactImportWorker: add Oban unique:[keys: [:import_run_id,
  :offset]]. Was missing on a worker whose perform() does a
  non-idempotent atomic counter increment — a retry would double-
  count imported rows.
- CommonVolumeRadarWorker: x_min..x_max default step is -1 when
  x_min > x_max, triggering a runtime deprecation warning. Force
  Range.new(_, _, 1) explicitly.

**Cleanup**
- Drop unused tmp_dir parameter from hrrr_point_worker + its
  process_batch signature.

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 ->
+      nil -> maybe_enqueue(reason: "no scores found", age: nil)
-        Logger.info("FreshnessMonitor: no scores found, enqueuing grid worker")
+      latest -> check_stale(latest)
-        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
     end
   end
 
-  defp enqueue_if_not_queued do
+  defp check_stale(latest) do
-    # PropagationGridWorker declares `unique:` over a 1-hour window on
+    age = DateTime.diff(DateTime.utc_now(), latest, :minute)
-    # the seed args (`%{}`), so repeated 5-minute ticks during a long
+
-    # outage collapse into a single seed job — not one stacked chain
+    if age > @stale_threshold_minutes do
-    # per tick. `Oban.insert` returns `{:ok, job}` either way; the
+      maybe_enqueue(reason: "scores #{age}m old", age: age)
-    # `conflict?` field on the returned job distinguishes the two.
+    end
-    Oban.insert(PropagationGridWorker.new(%{}))
+  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

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,
+          Range.new(x_min, x_max, 1),
-          y_min..y_max,
+          Range.new(y_min, y_max, 1),
           pos1,
           pos2,
           step

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
 

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

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)}")
+        handle_iem_error("ASOS", station_code, reason)
-        {:error, 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)}")
+        handle_iem_error("RAOB", station_code, reason)
-        {:error, reason}
     end
   end
 

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
+      forecast_times
-      |> Enum.reject(&(selected && DateTime.compare(&1, selected) == :eq))
+      |> Task.async_stream(
-      |> Enum.map(fn t ->
+        fn t -> %{time: DateTime.to_iso8601(t), scores: Propagation.scores_at(band, t, bounds)} end,
-        %{time: DateTime.to_iso8601(t), scores: Propagation.scores_at(band, t, bounds)}
+        max_concurrency: 4,
-      end)
+        ordered: true,
+        timeout: 10_000
+      )
+      |> Enum.map(fn {:ok, h} -> h end)
 
     {:noreply,
      socket

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

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<dyn std::error::Error>> {
         .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<dyn std::error::Error>> {
                 let points = task.points.clone();
                 let started = Instant::now();
 
-                match hrrr_points::process_batch(&client, &pool, valid_time, &points, &tmp_dir)
+                match hrrr_points::process_batch(&client, &pool, valid_time, &points).await
-                    .await
                 {
                     Ok(stats) => {
                         let elapsed = started.elapsed();

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<Vec<LevelKeys>> = 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"),

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<String> = OnceLock::new();
+    P.get_or_init(|| build_pattern(&fetcher::surface_messages_owned()))
+}
+
+fn prs_pattern() -> &'static str {
+    static P: OnceLock<String> = 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<Utc>,
     points: &[(f64, f64)],
-    _tmp_dir: &Path,
 ) -> Result<PointFetchStats, PointFetchError> {
     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!(
+    let sfc_pat = sfc_pattern();
-        ":({}):",
+    let prs_pat = prs_pattern();
-        sfc_wanted
-            .iter()
-            .map(|(v, _)| v.as_str())
-            .collect::<Vec<_>>()
-            .join("|")
-    );
-    let prs_pattern = format!(
-        ":({}):",
-        prs_wanted
-            .iter()
-            .map(|(v, _)| v.as_str())
-            .collect::<Vec<_>>()
-            .join("|")
-    );
 
     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 || {
+    let sfc_grid =
-        decoder::extract_grid(&sfc_blob, &sfc_pattern, grid_spec)
+        tokio::task::spawn_blocking(move || decoder::extract_grid(&sfc_blob, sfc_pat, grid_spec))
-    })
+            .await
-    .await
+            .expect("blocking join")?;
-    .expect("blocking join")?;
 
-    let prs_grid = tokio::task::spawn_blocking(move || {
+    let prs_grid =
-        decoder::extract_grid(&prs_blob, &prs_pattern, grid_spec)
+        tokio::task::spawn_blocking(move || decoder::extract_grid(&prs_blob, prs_pat, grid_spec))
-    })
+            .await
-    .await
+            .expect("blocking join")?;
-    .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<Json<Value>> 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<sqlx::types::Json<serde_json::Value>> = fetcher::GRID_PRESSURE_LEVELS
+    let levels: Vec<sqlx::types::Json<serde_json::Value>> = fetcher::grid_level_keys()
         .iter()
-        .filter_map(|&p| {
+        .filter_map(|k| {
-            let pres_mb = p as f64;
+            let t = cell.get(k.tmp.as_str()).copied()?;
-            let t = cell.get(&format!("TMP:{p} mb")).copied()?;
+            let h = cell.get(k.hgt.as_str()).copied()?;
-            let h = cell.get(&format!("HGT:{p} mb")).copied()?;
+            let d = cell.get(k.dpt.as_str()).copied();
-            let d = cell.get(&format!("DPT:{p} mb")).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 {

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<Level> = fetcher::GRID_PRESSURE_LEVELS
+    let levels: Vec<Level> = fetcher::grid_level_keys()
         .iter()
-        .filter_map(|&p| {
+        .filter_map(|k| {
-            let pres_mb = p as f64;
+            let t = cell.get(k.tmp.as_str()).copied()?;
-            let t = cell.get(&format!("TMP:{p} mb")).copied()?;
+            let d = cell.get(k.dpt.as_str()).copied();
-            let d = cell.get(&format!("DPT:{p} mb")).copied();
+            let h = cell.get(k.hgt.as_str()).copied()?;
-            let h = cell.get(&format!("HGT:{p} mb")).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<V> = fetcher::GRID_PRESSURE_LEVELS
+    let levels: Vec<V> = fetcher::grid_level_keys()
         .iter()
-        .filter_map(|&p| {
+        .filter_map(|k| {
-            let pres_mb = p as f64;
+            let t = cell.get(k.tmp.as_str()).copied()?;
-            let t = cell.get(&format!("TMP:{p} mb")).copied()?;
+            let d = cell.get(k.dpt.as_str()).copied();
-            let d = cell.get(&format!("DPT:{p} mb")).copied();
+            let h = cell.get(k.hgt.as_str()).copied()?;
-            let h = cell.get(&format!("HGT:{p} mb")).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 {