From 9e776a07d5a26c2faea04a88853bbc5f6239faca Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sat, 25 Apr 2026 12:58:10 -0500 Subject: [PATCH] fix(skewt): write profile files for f01..f18 forecast hours MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The /skewt page was empty in prod because Phase 2 of the Rust cutover moved f01..f18 grid scoring to Rust but left profile-file writing only on the analysis (f00) path. Result: /data/scores/profiles/ accumulated one file per chain run instead of nineteen, so SkewtLive only ever saw the analysis hour — and that hour falls outside the 1 h past cutoff within an hour of the chain finishing, leaving the page empty most of the time. Two surgical fixes: * `pipeline::run_chain_step` now builds `profile_entries` alongside `prepared` from the merged grid in a single iter pass, then spawns `profiles_file::write_atomic` on the blocking pool to overlap the NFS write with the score-band scoring/write fan-out — same pattern `run_analysis_step` already uses. `ChainStepStats` gains `profile_cells_written` so the worker log line is symmetric with the analysis step's existing field and a future failure mode that drops the profile would show up as a divergence. * `SkewtLive.available_valid_times/0` widens the past cutoff from 1 h to 3 h. The chain's analysis hour can be ~70 min old by the time the next chain finishes (and ~2 h on a missed cycle), so the 1 h cutoff was leaving the page blank during normal transients. 18 h forward window unchanged. Cosmetic: cargo fmt picked up trivial rewrites in decoder/fetcher/ native_duct/hrrr_point_worker that had drifted; rolled in. Verification: cargo build --release, cargo clippy --lib --bins -D warnings, cargo test --lib (134/134), mix test test/microwaveprop_web/live/skewt_live_test.exs (3/3) all green. --- lib/microwaveprop_web/live/skewt_live.ex | 7 ++- .../prop_grid_rs/src/bin/hrrr_point_worker.rs | 3 +- rust/prop_grid_rs/src/bin/worker.rs | 1 + rust/prop_grid_rs/src/decoder.rs | 5 +- rust/prop_grid_rs/src/fetcher.rs | 8 ++- rust/prop_grid_rs/src/native_duct.rs | 5 +- rust/prop_grid_rs/src/pipeline.rs | 58 ++++++++++++++----- 7 files changed, 65 insertions(+), 22 deletions(-) diff --git a/lib/microwaveprop_web/live/skewt_live.ex b/lib/microwaveprop_web/live/skewt_live.ex index faffd049..bce7ea84 100644 --- a/lib/microwaveprop_web/live/skewt_live.ex +++ b/lib/microwaveprop_web/live/skewt_live.ex @@ -129,8 +129,13 @@ defmodule MicrowavepropWeb.SkewtLive do end defp available_valid_times do + # The chain produces analysis + 18 forecast hours every wall-clock + # hour. By the time the next chain finishes the previous chain's + # analysis is up to 70 min old, and during a missed cycle it can be + # ~2 h old. A 1-hour past cutoff would leave the page empty while + # the chain catches up — keep 3 h so a stale analysis still shows. now = DateTime.utc_now() - past_cutoff = DateTime.add(now, -3600, :second) + past_cutoff = DateTime.add(now, -3 * 3600, :second) future_cutoff = DateTime.add(now, 18 * 3600, :second) ProfilesFile.list_valid_times() 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 fff65ad2..7de137f2 100644 --- a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs +++ b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs @@ -48,8 +48,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).await - { + match hrrr_points::process_batch(&client, &pool, valid_time, &points).await { Ok(stats) => { let elapsed = started.elapsed(); info!( diff --git a/rust/prop_grid_rs/src/bin/worker.rs b/rust/prop_grid_rs/src/bin/worker.rs index 6d7d4252..9dfad794 100644 --- a/rust/prop_grid_rs/src/bin/worker.rs +++ b/rust/prop_grid_rs/src/bin/worker.rs @@ -169,6 +169,7 @@ async fn worker_loop( forecast_hour = task.forecast_hour, files = stats.score_files_written, points = stats.point_count, + profile_cells = stats.profile_cells_written, duration_s = elapsed.as_secs_f64(), "chain step done" ), diff --git a/rust/prop_grid_rs/src/decoder.rs b/rust/prop_grid_rs/src/decoder.rs index f65c04b7..358c514f 100644 --- a/rust/prop_grid_rs/src/decoder.rs +++ b/rust/prop_grid_rs/src/decoder.rs @@ -242,9 +242,8 @@ pub fn parse_lola_binary(bin: &[u8], messages: &[Message], grid_spec: &GridSpec) let lat_key = (lat * 1000.0).round() as i32; for i in 0..nx { let cell_offset = (j * nx + i) * 4; - let value = f32::from_le_bytes( - chunk[cell_offset..cell_offset + 4].try_into().unwrap(), - ); + let value = + f32::from_le_bytes(chunk[cell_offset..cell_offset + 4].try_into().unwrap()); if value > UNDEFINED_VALUE / 2.0 { continue; } diff --git a/rust/prop_grid_rs/src/fetcher.rs b/rust/prop_grid_rs/src/fetcher.rs index c17fead5..83bae1aa 100644 --- a/rust/prop_grid_rs/src/fetcher.rs +++ b/rust/prop_grid_rs/src/fetcher.rs @@ -14,8 +14,8 @@ use std::sync::Mutex; use std::time::{Duration, Instant}; use chrono::{DateTime, NaiveDate, Timelike, Utc}; -use tokio::task::JoinSet; use reqwest::header::{HeaderMap, HeaderValue, RANGE}; +use tokio::task::JoinSet; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Product { @@ -754,7 +754,11 @@ mod tests { // accumulate each hour. Calling get() on an expired entry must // remove it so the in-memory map bounds itself to the hot set. let cache = IdxCache::new(); - cache.put_with_age("stale".into(), "old".into(), IDX_TTL + Duration::from_secs(1)); + cache.put_with_age( + "stale".into(), + "old".into(), + IDX_TTL + Duration::from_secs(1), + ); cache.put("fresh".into(), "new".into()); assert_eq!(cache.len(), 2); diff --git a/rust/prop_grid_rs/src/native_duct.rs b/rust/prop_grid_rs/src/native_duct.rs index bbfd7539..fda2b53b 100644 --- a/rust/prop_grid_rs/src/native_duct.rs +++ b/rust/prop_grid_rs/src/native_duct.rs @@ -229,7 +229,10 @@ mod tests { let mut c = CellValues::new(); for i in 1..=count { c.insert(format!("HGT:{i} hybrid level").into(), (i as f32) * 50.0); - c.insert(format!("TMP:{i} hybrid level").into(), 290.0 - (i as f32) * 0.5); + c.insert( + format!("TMP:{i} hybrid level").into(), + 290.0 - (i as f32) * 0.5, + ); c.insert( format!("SPFH:{i} hybrid level").into(), 0.008 - (i as f32) * 0.0001, diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index ae7a3543..ef1154b6 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -62,6 +62,13 @@ pub struct ChainStepStats { pub score_files_written: u32, pub point_count: u32, pub band_count: u32, + /// Number of cells written to the per-valid_time profile file. + /// Always `point_count` after the f01..f18 profile-file write was + /// added on 2026-04-25 — kept as a separate field so the log line + /// is symmetric with the analysis step's `profile_cells_written` + /// in `RunStepStats`, and so a future failure mode that writes + /// scores but skips the profile file shows up as a divergence. + pub profile_cells_written: u32, } #[tracing::instrument( @@ -128,19 +135,40 @@ pub async fn run_chain_step( let merged = merge_grids(sfc_grid, prs_grid); let point_count = merged.len() as u32; - // Step 1: consume `merged` into a compact `Vec<(lat, lon, Conditions, - // invariants)>`. This is the biggest win — `merged`'s inner HashMaps - // of `String -> f32` account for ~200 MB of heap (92k × ~60 string - // keys) and are dropped as soon as this loop finishes. - let prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> = merged - .into_iter() - .filter_map(|(key, cell)| { - let (lat, lon) = decoder::key_to_latlon(key); - let conditions = cell_to_conditions(&cell, lat, lon, &valid_time)?; + // Build `prepared` (for scoring) and `profile_entries` (for the + // per-valid_time profile file consumed by /skewt and the LiveView + // point-detail panels) in a single pass over `merged`. We need a + // borrowing iterator (not `into_iter`) because `cell_to_profile_entry` + // also reads from the cell — `merged` is dropped immediately after + // this loop finishes, so the ~200 MB string-keyed HashMaps don't + // outlive the scoring stage. The price is briefly retaining both + // vecs at the same time as `merged`, which is well inside the + // 512 MiB cgroup ceiling on talos5. + let mut prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> = + Vec::with_capacity(merged.len()); + let mut profile_entries: Vec = Vec::with_capacity(merged.len()); + for (key, cell) in merged.iter() { + let (lat, lon) = decoder::key_to_latlon(*key); + if let Some(conditions) = cell_to_conditions(cell, lat, lon, &valid_time) { let invariants = scorer::precompute_band_invariants(&conditions); - Some((lat, lon, conditions, invariants)) + prepared.push((lat, lon, conditions, invariants)); + profile_entries.push(cell_to_profile_entry(lat, lon, cell)); + } + } + drop(merged); + let profile_cells_written = profile_entries.len() as u32; + + // Spawn the profile-file write on the blocking pool so it overlaps + // with the scoring loop instead of serialising on the hot path. + let scores_dir_owned = scores_dir.to_path_buf(); + let scores_dir_for_profiles = scores_dir_owned.clone(); + let profile_future = { + let profiles = profile_entries; + tokio::task::spawn_blocking(move || { + profiles_file::write_atomic(&scores_dir_for_profiles, valid_time, &profiles) + .map(|_| 0u32) }) - .collect(); + }; // Step 2: score and write every band in parallel. // @@ -155,7 +183,6 @@ pub async fn run_chain_step( // path. Memory stays bounded — each Vec is ~1.8 MB so // 23 concurrent ≈ ~40 MB peak. let bands = band_config::all_bands(); - let scores_dir = scores_dir.to_path_buf(); let prepared = Arc::new(prepared); let scored: Vec<(u32, Vec)> = { use rayon::prelude::*; @@ -179,7 +206,7 @@ pub async fn run_chain_step( let write_handles: Vec<_> = scored .into_iter() .map(|(band_mhz, scores)| { - let dir = scores_dir.clone(); + let dir = scores_dir_owned.clone(); tokio::task::spawn_blocking(move || { scores_file::write_atomic(&dir, band_mhz, valid_time, &scores) }) @@ -188,12 +215,17 @@ pub async fn run_chain_step( for h in write_handles { h.await.expect("blocking join")?; } + profile_future + .await + .expect("blocking join") + .map_err(PipelineError::ProfileWrite)?; let files_written = bands.len() as u32; Ok(ChainStepStats { score_files_written: files_written, point_count, band_count: bands.len() as u32, + profile_cells_written, }) }