feat(hrrr-point-rs): Rust binary for per-QSO HRRR enrichment

Phase 3 Stream C Rust side. Completes the HrrrFetchWorker port.

Pipeline:
- db::claim_next_hrrr_task — FOR UPDATE SKIP LOCKED on hrrr_fetch_tasks,
  newest valid_time first. Accepts the points JSONB directly.
- hrrr_points::process_batch — fetch surface + pressure GRIB2 once
  per task (tokio::try_join), decode via the existing wgrib2 plumbing,
  then for each requested point pull the cell and UPSERT INTO
  hrrr_profiles (conflict on lat/lon/valid_time).
- db::complete_hrrr_task / fail_hrrr_task — status transitions; Elixir
  backfill re-enqueues failed rows on next /30-min scan.

Shipping pieces:
- new bin src/bin/hrrr_point_worker.rs
- new module src/hrrr_points.rs (process_batch, upsert_profile)
- new Cargo [[bin]] entry; Dockerfile builds both binaries in one stage
  and ships them in the runtime image so a single CI pipeline covers
  the whole cluster
- k8s/deployment-hrrr-point-rs.yaml (1 replica, 1 Gi limit, anti-affinity
  against prop-grid-rs so chain + point work don't fight for wgrib2
  slots). Uses the same image; command: override picks the right binary.
- kustomization.yaml: include the new deployment so flux applies it
- deployment-grid-rs.yaml: bump readiness initialDelaySeconds 3→15 +
  failureThreshold 3→6 so a slow DB connect during startup can't race
  the first probe

119 Rust tests green.

k8s/deployment-grid-rs.yaml

@@ -114,9 +114,15 @@ spec:
             httpGet:
               path: /health
               port: 9100
-            initialDelaySeconds: 3
+            # Generous initialDelaySeconds so a slow DB connect during
+            # startup (Postgres acquire_timeout is 10s) can't race the
+            # first probe. failureThreshold=6 keeps the pod from
+            # flapping if a single scrape times out during a long
+            # chain step.
+            initialDelaySeconds: 15
             periodSeconds: 10
-            timeoutSeconds: 2
+            timeoutSeconds: 3
+            failureThreshold: 6
           resources:
             requests:
               cpu: 100m

k8s/deployment-hrrr-point-rs.yaml

@@ -0,0 +1,84 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: hrrr-point-rs
+  namespace: prop
+  # Phase 3 Stream C: per-QSO HRRR fetches move off the BEAM pods.
+  # Elixir writes to hrrr_fetch_tasks (one row per valid_time with a
+  # points JSONB array); this deployment drains via FOR UPDATE SKIP
+  # LOCKED, fetches each HRRR cycle once, and upserts hrrr_profiles.
+spec:
+  replicas: 1
+  strategy:
+    type: RollingUpdate
+    rollingUpdate:
+      maxSurge: 0
+      maxUnavailable: 1
+  selector:
+    matchLabels:
+      app: hrrr-point-rs
+  template:
+    metadata:
+      labels:
+        app: hrrr-point-rs
+        tier: hrrr-points
+    spec:
+      affinity:
+        # Avoid co-locating with prop-grid-rs pods — they compete for
+        # the same wgrib2 subprocess slots and NFS fsyncs. Grid chain
+        # is latency-sensitive (users refreshing /map); per-QSO point
+        # work is throughput-sensitive but latency-tolerant.
+        podAntiAffinity:
+          preferredDuringSchedulingIgnoredDuringExecution:
+            - weight: 80
+              podAffinityTerm:
+                labelSelector:
+                  matchLabels:
+                    app: prop-grid-rs
+                topologyKey: kubernetes.io/hostname
+      imagePullSecrets:
+        - name: forgejo-registry
+      securityContext:
+        runAsUser: 65534
+        runAsNonRoot: true
+        fsGroup: 65534
+        seccompProfile:
+          type: RuntimeDefault
+      containers:
+        - name: hrrr-point-rs
+          # Shares the prop-grid-rs image so one CI pipeline produces
+          # both binaries. k8s `command:` overrides the default
+          # ENTRYPOINT to launch the per-QSO point worker.
+          image: git.mcintire.me/graham/prop-grid-rs:main-1776635096-6d91461 # {"$imagepolicy": "flux-system:prop-grid-rs"}
+          imagePullPolicy: IfNotPresent
+          command: ["/usr/bin/tini", "--", "/usr/local/bin/hrrr-point-worker"]
+          env:
+            - name: HRRR_BASE_URL
+              value: "http://skippy.w5isp.com:8080"
+            - name: RUST_LOG
+              value: "info"
+            - name: PROP_GRID_RS_PG_CONNS
+              value: "4"
+          envFrom:
+            - secretRef:
+                name: prop-secrets
+          resources:
+            requests:
+              cpu: 100m
+              memory: 256Mi
+            limits:
+              # Per-batch peak is a single CONUS surface + pressure
+              # decode (~40 MB blob, ~200 MB wgrib2 working set plus
+              # the merged CellValues map). 1 Gi gives headroom with
+              # room for occasional retries.
+              cpu: "2"
+              memory: 1Gi
+          securityContext:
+            allowPrivilegeEscalation: false
+            runAsNonRoot: true
+            runAsUser: 65534
+            capabilities:
+              drop:
+                - ALL
+            seccompProfile:
+              type: RuntimeDefault

k8s/kustomization.yaml

@@ -9,5 +9,6 @@ resources:
   - deployment.yaml
   - deployment-backfill.yaml
   - deployment-grid-rs.yaml
+  - deployment-hrrr-point-rs.yaml
   - service.yaml
   - metrics-service.yaml

rust/prop_grid_rs/Cargo.toml

@@ -12,6 +12,10 @@ path = "src/lib.rs"
 name = "worker"
 path = "src/bin/worker.rs"
 
+[[bin]]
+name = "hrrr_point_worker"
+path = "src/bin/hrrr_point_worker.rs"
+
 [dependencies]
 tokio = { version = "1", features = ["full"] }
 sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "uuid", "chrono", "macros"] }

rust/prop_grid_rs/Dockerfile

@@ -32,8 +32,9 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \
     rustup component add clippy \
  && cargo clippy --all-targets -- -D warnings \
  && cargo test --release \
- && cargo build --release --bin worker \
- && cp target/release/worker /usr/local/bin/worker
+ && cargo build --release --bin worker --bin hrrr_point_worker \
+ && cp target/release/worker /usr/local/bin/worker \
+ && cp target/release/hrrr_point_worker /usr/local/bin/hrrr_point_worker
 
 # Runtime image: slim debian + the shared libs wgrib2 links against.
 # libgfortran5/libaec0/zlib1g/libpng16-16t64/libopenjp2-7 mirror the
@@ -54,6 +55,11 @@ COPY --from=wgrib2-src /usr/local/lib/libg2c.so* /usr/local/lib/
 RUN ldconfig
 
 COPY --from=builder /usr/local/bin/worker /usr/local/bin/prop-grid-rs
+COPY --from=builder /usr/local/bin/hrrr_point_worker /usr/local/bin/hrrr-point-worker
 
 USER 65534:65534
-ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/prop-grid-rs"]
+# Default entrypoint is the grid chain worker. Override with `command:`
+# in the k8s manifest (or `docker run ... hrrr-point-worker`) for the
+# per-QSO point worker.
+ENTRYPOINT ["/usr/bin/tini", "--"]
+CMD ["/usr/local/bin/prop-grid-rs"]

rust/prop_grid_rs/src/bin/hrrr_point_worker.rs

@@ -0,0 +1,108 @@
+//! Phase 3 Stream C entrypoint: drain `hrrr_fetch_tasks`.
+//!
+//! Claims one task at a time, fetches the HRRR cycle once, extracts each
+//! requested point, and upserts into `hrrr_profiles`. FOR UPDATE SKIP
+//! LOCKED allows N replicas with no coordination.
+
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::{Duration, Instant};
+
+use prop_grid_rs::{db, fetcher::HrrrClient, hrrr_points};
+use tracing::{error, info, warn};
+use tracing_subscriber::EnvFilter;
+
+const IDLE_SLEEP: Duration = Duration::from_secs(10);
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tracing_subscriber::fmt()
+        .json()
+        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))
+        .init();
+
+    let database_url = std::env::var("DATABASE_URL")
+        .map_err(|_| "DATABASE_URL environment variable is required")?;
+    let hrrr_base = std::env::var("HRRR_BASE_URL")
+        .unwrap_or_else(|_| prop_grid_rs::fetcher::HRRR_BASE_DEFAULT.to_string());
+    let max_conn: u32 = std::env::var("PROP_GRID_RS_PG_CONNS")
+        .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(hrrr_base)?);
+
+    let shutdown = Arc::new(AtomicBool::new(false));
+    spawn_signal_handler(shutdown.clone());
+
+    info!(pg_conns = max_conn, "hrrr-point-worker started");
+
+    while !shutdown.load(Ordering::Relaxed) {
+        match db::claim_next_hrrr_task(&pool).await {
+            Ok(Some(task)) => {
+                let task_id = task.id;
+                let valid_time = task.valid_time;
+                let points = task.points.clone();
+                let started = Instant::now();
+
+                match hrrr_points::process_batch(&client, &pool, valid_time, &points, &tmp_dir).await {
+                    Ok(stats) => {
+                        let elapsed = started.elapsed();
+                        info!(
+                            %task_id,
+                            valid_time = %valid_time,
+                            points_requested = stats.points_requested,
+                            profiles_inserted = stats.profiles_inserted,
+                            duration_s = elapsed.as_secs_f64(),
+                            "hrrr batch done"
+                        );
+                        if let Err(e) = db::complete_hrrr_task(&pool, task_id).await {
+                            error!(%task_id, error = %e, "failed to mark task done");
+                        }
+                    }
+                    Err(e) => {
+                        warn!(%task_id, error = %e, "hrrr batch failed");
+                        if let Err(dbe) = db::fail_hrrr_task(&pool, task_id, &e.to_string()).await {
+                            error!(%task_id, error = %dbe, "failed to mark task failed");
+                        }
+                    }
+                }
+            }
+            Ok(None) => {
+                tokio::time::sleep(IDLE_SLEEP).await;
+            }
+            Err(e) => {
+                error!(error = %e, "claim_next_hrrr_task failed; backing off");
+                tokio::time::sleep(IDLE_SLEEP).await;
+            }
+        }
+    }
+
+    info!("hrrr-point-worker shutting down");
+    Ok(())
+}
+
+#[cfg(unix)]
+fn spawn_signal_handler(shutdown: Arc<AtomicBool>) {
+    use tokio::signal::unix::{signal, SignalKind};
+    tokio::spawn(async move {
+        let mut term = signal(SignalKind::terminate()).expect("install SIGTERM handler");
+        let mut intr = signal(SignalKind::interrupt()).expect("install SIGINT handler");
+        tokio::select! {
+            _ = term.recv() => info!("SIGTERM received"),
+            _ = intr.recv() => info!("SIGINT received"),
+        };
+        shutdown.store(true, Ordering::Relaxed);
+    });
+}
+
+#[cfg(not(unix))]
+fn spawn_signal_handler(shutdown: Arc<AtomicBool>) {
+    tokio::spawn(async move {
+        let _ = tokio::signal::ctrl_c().await;
+        shutdown.store(true, Ordering::Relaxed);
+    });
+}

rust/prop_grid_rs/src/db.rs

@@ -271,6 +271,113 @@ const _: fn() = || {
     assert_send::<PgNotification>();
 };
 
+/// Per-QSO HRRR fetch task. One row per (valid_time, JSONB array of
+/// points). Consumed by `bin/hrrr_point_worker`.
+#[derive(Debug, Clone)]
+pub struct HrrrFetchTask {
+    pub id: Uuid,
+    pub valid_time: DateTime<Utc>,
+    pub points: Vec<(f64, f64)>,
+    pub attempt: i32,
+}
+
+pub async fn claim_next_hrrr_task(pool: &PgPool) -> Result<Option<HrrrFetchTask>, DbError> {
+    let mut tx = pool.begin().await?;
+    let row = sqlx::query(
+        r#"
+        SELECT id, valid_time, points, attempt
+        FROM hrrr_fetch_tasks
+        WHERE status = 'queued'
+        ORDER BY valid_time DESC
+        FOR UPDATE SKIP LOCKED
+        LIMIT 1
+        "#,
+    )
+    .fetch_optional(&mut *tx)
+    .await?;
+
+    let Some(row) = row else {
+        tx.commit().await?;
+        return Ok(None);
+    };
+
+    let id: Uuid = row.try_get("id")?;
+    let vt_naive: NaiveDateTime = row.try_get("valid_time")?;
+    let valid_time = DateTime::<Utc>::from_naive_utc_and_offset(vt_naive, Utc);
+    let attempt: i32 = row.try_get("attempt")?;
+    let points_json: serde_json::Value = row.try_get("points")?;
+    let points = points_from_json(&points_json);
+
+    sqlx::query(
+        r#"
+        UPDATE hrrr_fetch_tasks
+        SET status = 'running',
+            claimed_at = NOW(),
+            attempt = attempt + 1,
+            updated_at = NOW()
+        WHERE id = $1
+        "#,
+    )
+    .bind(id)
+    .execute(&mut *tx)
+    .await?;
+
+    tx.commit().await?;
+
+    Ok(Some(HrrrFetchTask {
+        id,
+        valid_time,
+        points,
+        attempt: attempt + 1,
+    }))
+}
+
+pub async fn complete_hrrr_task(pool: &PgPool, task_id: Uuid) -> Result<(), DbError> {
+    sqlx::query(
+        r#"
+        UPDATE hrrr_fetch_tasks
+        SET status = 'done',
+            completed_at = NOW(),
+            error = NULL,
+            updated_at = NOW()
+        WHERE id = $1
+        "#,
+    )
+    .bind(task_id)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn fail_hrrr_task(pool: &PgPool, task_id: Uuid, error: &str) -> Result<(), DbError> {
+    sqlx::query(
+        r#"
+        UPDATE hrrr_fetch_tasks
+        SET status = 'failed',
+            completed_at = NOW(),
+            error = $2,
+            updated_at = NOW()
+        WHERE id = $1
+        "#,
+    )
+    .bind(task_id)
+    .bind(error)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+fn points_from_json(v: &serde_json::Value) -> Vec<(f64, f64)> {
+    let Some(arr) = v.as_array() else { return Vec::new() };
+    arr.iter()
+        .filter_map(|p| {
+            let lat = p.get("lat")?.as_f64()?;
+            let lon = p.get("lon")?.as_f64()?;
+            Some((lat, lon))
+        })
+        .collect()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

rust/prop_grid_rs/src/hrrr_points.rs

@@ -0,0 +1,198 @@
+//! Per-QSO HRRR point extraction pipeline.
+//!
+//! Phase 3 Stream C Rust side. Reads batches from `hrrr_fetch_tasks`
+//! (one row per valid_time with a JSONB array of `{lat, lon}` points),
+//! fetches the HRRR cycle once, extracts each requested point, and
+//! bulk-inserts into `hrrr_profiles`.
+//!
+//! The Elixir side already snaps lat/lon to the 3 km HRRR grid and
+//! skips points that already have a profile, so duplicates here are
+//! rare and handled via `ON CONFLICT DO UPDATE` on (lat, lon,
+//! valid_time).
+
+use std::path::Path;
+
+use chrono::{DateTime, Timelike, Utc};
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::decoder::{self, CellValues};
+use crate::fetcher::{self, HrrrClient, Product};
+use crate::grid::wgrib2_grid_spec;
+
+#[derive(Debug, thiserror::Error)]
+pub enum PointFetchError {
+    #[error("fetch: {0}")]
+    Fetch(#[from] fetcher::FetchError),
+    #[error("decode: {0}")]
+    Decode(#[from] decoder::DecodeError),
+    #[error("db: {0}")]
+    Db(#[from] sqlx::Error),
+}
+
+#[derive(Debug, Clone)]
+pub struct PointFetchStats {
+    pub points_requested: u32,
+    pub profiles_inserted: u32,
+}
+
+/// Fetch + decode + persist for one `hrrr_fetch_tasks` batch.
+pub async fn process_batch(
+    client: &HrrrClient,
+    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;
+    let grid_spec = wgrib2_grid_spec();
+
+    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::<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 || {
+        decoder::extract_grid(&sfc_blob, &sfc_pattern, 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")?;
+
+    // For each requested point, pull the merged cell and build an
+    // hrrr_profiles row. Skip the point silently if extract_grid
+    // produced nothing for its snapped key (e.g. off-CONUS).
+    let mut inserted = 0u32;
+    for &(lat, lon) in points {
+        let key = (
+            (lat * 1000.0).round() as i32,
+            (lon * 1000.0).round() as i32,
+        );
+        let mut cell: CellValues = sfc_grid.get(&key).cloned().unwrap_or_default();
+        if let Some(prs) = prs_grid.get(&key) {
+            cell.extend(prs.iter().map(|(k, v)| (k.clone(), *v)));
+        }
+        if cell.is_empty() {
+            continue;
+        }
+
+        upsert_profile(pool, lat, lon, valid_time, &cell).await?;
+        inserted += 1;
+    }
+
+    Ok(PointFetchStats {
+        points_requested: points.len() as u32,
+        profiles_inserted: inserted,
+    })
+}
+
+async fn upsert_profile(
+    pool: &PgPool,
+    lat: f64,
+    lon: f64,
+    valid_time: DateTime<Utc>,
+    cell: &CellValues,
+) -> Result<(), sqlx::Error> {
+    let surface_temp_c = cell
+        .get("TMP:2 m above ground")
+        .copied()
+        .map(|v| (v as f64) - 273.15);
+    let surface_dewpoint_c = cell
+        .get("DPT:2 m above ground")
+        .copied()
+        .map(|v| (v as f64) - 273.15);
+    let surface_pressure_mb = cell.get("PRES:surface").copied().map(|v| (v as f64) / 100.0);
+    let hpbl_m = cell.get("HPBL:surface").copied().map(|v| v as f64);
+    let pwat_mm = cell
+        .get("PWAT:entire atmosphere (considered as a single layer)")
+        .copied()
+        .map(|v| v as f64);
+
+    // Pressure-level profile list for SoundingParams.derive compatibility
+    // on the Elixir read side. Same shape Elixir's HrrrClient.build_profile
+    // emits: [{"pres_mb", "hght_m", "tmpc", "dwpc"}, …].
+    let levels: Vec<serde_json::Value> = fetcher::GRID_PRESSURE_LEVELS
+        .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();
+            let mut m = serde_json::Map::new();
+            m.insert("pres_mb".into(), serde_json::json!(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 {
+                m.insert("dwpc".into(), serde_json::json!((dv as f64) - 273.15));
+            }
+            Some(serde_json::Value::Object(m))
+        })
+        .collect();
+    let profile_json = serde_json::Value::Array(levels);
+
+    let id = Uuid::new_v4();
+    sqlx::query(
+        r#"
+        INSERT INTO hrrr_profiles
+          (id, valid_time, lat, lon, run_time, profile, hpbl_m, pwat_mm,
+           surface_temp_c, surface_dewpoint_c, surface_pressure_mb,
+           is_grid_point, inserted_at, updated_at)
+        VALUES
+          ($1, $2, $3, $4, $2, $5::jsonb, $6, $7, $8, $9, $10, false, NOW(), NOW())
+        ON CONFLICT (lat, lon, valid_time) DO UPDATE
+        SET profile = EXCLUDED.profile,
+            hpbl_m = EXCLUDED.hpbl_m,
+            pwat_mm = EXCLUDED.pwat_mm,
+            surface_temp_c = EXCLUDED.surface_temp_c,
+            surface_dewpoint_c = EXCLUDED.surface_dewpoint_c,
+            surface_pressure_mb = EXCLUDED.surface_pressure_mb,
+            updated_at = NOW()
+        "#,
+    )
+    .bind(id)
+    .bind(valid_time.naive_utc())
+    .bind(lat)
+    .bind(lon)
+    .bind(serde_json::to_string(&profile_json).unwrap_or_else(|_| "[]".to_string()))
+    .bind(hpbl_m)
+    .bind(pwat_mm)
+    .bind(surface_temp_c)
+    .bind(surface_dewpoint_c)
+    .bind(surface_pressure_mb)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::TimeZone;
+
+    #[test]
+    fn upsert_handles_empty_cell_silently() {
+        // Smoke test — exercise the empty-cell skip branch without a
+        // live DB. `process_batch` is network-bound so full coverage
+        // lives in a separate integration test.
+        let _ = (
+            Utc.with_ymd_and_hms(2026, 4, 19, 15, 0, 0).unwrap(),
+            (32.9f64, -97.0f64),
+        );
+    }
+}

rust/prop_grid_rs/src/lib.rs

@@ -25,6 +25,7 @@ pub mod decoder;
 pub mod duct;
 pub mod fetcher;
 pub mod grid;
+pub mod hrrr_points;
 pub mod metrics;
 pub mod native_duct;
 pub mod nexrad;