diff --git a/k8s/deployment-hrrr-point-rs.yaml b/k8s/deployment-hrrr-point-rs.yaml index 63fcdaef..9579fb8e 100644 --- a/k8s/deployment-hrrr-point-rs.yaml +++ b/k8s/deployment-hrrr-point-rs.yaml @@ -69,10 +69,13 @@ spec: 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. + # the full 92k-cell merged CellValues map kept in memory + # until all requested points are drained). 1 Gi OOMKilled + # repeatedly — four restarts in ten minutes — so bumped + # to 2 Gi. The scheduler-on-worker2 placement has plenty + # of headroom. cpu: "2" - memory: 1Gi + memory: 2Gi securityContext: allowPrivilegeEscalation: false runAsNonRoot: true diff --git a/rust/prop_grid_rs/Cargo.toml b/rust/prop_grid_rs/Cargo.toml index bcfde348..692479cb 100644 --- a/rust/prop_grid_rs/Cargo.toml +++ b/rust/prop_grid_rs/Cargo.toml @@ -18,7 +18,7 @@ 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"] } +sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "uuid", "chrono", "macros", "json"] } reqwest = { version = "0.12", features = ["rustls-tls", "stream"], default-features = false } bytes = "1" anyhow = "1" diff --git a/rust/prop_grid_rs/src/hrrr_points.rs b/rust/prop_grid_rs/src/hrrr_points.rs index 7e8c8bd9..46126a74 100644 --- a/rust/prop_grid_rs/src/hrrr_points.rs +++ b/rust/prop_grid_rs/src/hrrr_points.rs @@ -75,6 +75,17 @@ pub async fn process_batch( .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 + // window). Log so "0 inserted" doesn't look like a silent success. + if sfc_grid.is_empty() && prs_grid.is_empty() { + tracing::warn!( + valid_time = %valid_time, + points = points.len(), + "hrrr fetch returned empty surface and pressure grids — upstream archive likely missing this cycle" + ); + } + // 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). @@ -124,10 +135,14 @@ async fn upsert_profile( .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 + // Pressure-level profile list. Postgres column `profile` is jsonb[] + // (ARRAY of jsonb), one element per level, matching Elixir's + // HrrrClient.build_profile output shape: + // [{"pres_mb", "hght_m", "tmpc", "dwpc"}, …] + // 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 .iter() .filter_map(|&p| { let pres_mb = p as f64; @@ -141,10 +156,9 @@ async fn upsert_profile( if let Some(dv) = d { m.insert("dwpc".into(), serde_json::json!((dv as f64) - 273.15)); } - Some(serde_json::Value::Object(m)) + Some(sqlx::types::Json(serde_json::Value::Object(m))) }) .collect(); - let profile_json = serde_json::Value::Array(levels); let id = Uuid::new_v4(); sqlx::query( @@ -154,7 +168,7 @@ async fn upsert_profile( 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()) + ($1, $2, $3, $4, $2, $5, $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, @@ -169,7 +183,7 @@ async fn upsert_profile( .bind(valid_time.naive_utc()) .bind(lat) .bind(lon) - .bind(serde_json::to_string(&profile_json).unwrap_or_else(|_| "[]".to_string())) + .bind(&levels) .bind(hpbl_m) .bind(pwat_mm) .bind(surface_temp_c)