From 4fefb81c246c50a08b214457da80001355de5f85 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 19 Apr 2026 18:27:55 -0500 Subject: [PATCH] feat(hrrr-point-rs): Rust binary for per-QSO HRRR enrichment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 10 +- k8s/deployment-hrrr-point-rs.yaml | 84 ++++++++ k8s/kustomization.yaml | 1 + rust/prop_grid_rs/Cargo.toml | 4 + rust/prop_grid_rs/Dockerfile | 12 +- .../prop_grid_rs/src/bin/hrrr_point_worker.rs | 108 ++++++++++ rust/prop_grid_rs/src/db.rs | 107 ++++++++++ rust/prop_grid_rs/src/hrrr_points.rs | 198 ++++++++++++++++++ rust/prop_grid_rs/src/lib.rs | 1 + 9 files changed, 520 insertions(+), 5 deletions(-) create mode 100644 k8s/deployment-hrrr-point-rs.yaml create mode 100644 rust/prop_grid_rs/src/bin/hrrr_point_worker.rs create mode 100644 rust/prop_grid_rs/src/hrrr_points.rs diff --git a/k8s/deployment-grid-rs.yaml b/k8s/deployment-grid-rs.yaml index 32a5568e..7bc871c2 100644 --- a/k8s/deployment-grid-rs.yaml +++ b/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 diff --git a/k8s/deployment-hrrr-point-rs.yaml b/k8s/deployment-hrrr-point-rs.yaml new file mode 100644 index 00000000..90e57ba3 --- /dev/null +++ b/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 diff --git a/k8s/kustomization.yaml b/k8s/kustomization.yaml index 0d59a70f..ea616ff3 100644 --- a/k8s/kustomization.yaml +++ b/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 diff --git a/rust/prop_grid_rs/Cargo.toml b/rust/prop_grid_rs/Cargo.toml index e026e65d..bcfde348 100644 --- a/rust/prop_grid_rs/Cargo.toml +++ b/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"] } diff --git a/rust/prop_grid_rs/Dockerfile b/rust/prop_grid_rs/Dockerfile index 9b12a1c2..cf5c6c01 100644 --- a/rust/prop_grid_rs/Dockerfile +++ b/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"] diff --git a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs new file mode 100644 index 00000000..2dba04ae --- /dev/null +++ b/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> { + 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) { + 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) { + tokio::spawn(async move { + let _ = tokio::signal::ctrl_c().await; + shutdown.store(true, Ordering::Relaxed); + }); +} diff --git a/rust/prop_grid_rs/src/db.rs b/rust/prop_grid_rs/src/db.rs index 38832361..68ab2234 100644 --- a/rust/prop_grid_rs/src/db.rs +++ b/rust/prop_grid_rs/src/db.rs @@ -271,6 +271,113 @@ const _: fn() = || { assert_send::(); }; +/// 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, + pub points: Vec<(f64, f64)>, + pub attempt: i32, +} + +pub async fn claim_next_hrrr_task(pool: &PgPool) -> Result, 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::::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::*; diff --git a/rust/prop_grid_rs/src/hrrr_points.rs b/rust/prop_grid_rs/src/hrrr_points.rs new file mode 100644 index 00000000..a9a14525 --- /dev/null +++ b/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, + points: &[(f64, f64)], + _tmp_dir: &Path, +) -> Result { + 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::>().join("|") + ); + let prs_pattern = format!( + ":({}):", + prs_wanted.iter().map(|(v, _)| v.as_str()).collect::>().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, + 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 = 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), + ); + } +} diff --git a/rust/prop_grid_rs/src/lib.rs b/rust/prop_grid_rs/src/lib.rs index c0e7f4d9..49178ddd 100644 --- a/rust/prop_grid_rs/src/lib.rs +++ b/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;