fix(skewt): write profile files for f01..f18 forecast hours
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.
This commit is contained in:
parent
3a1e79fb67
commit
9e776a07d5
7 changed files with 65 additions and 22 deletions
|
|
@ -129,8 +129,13 @@ defmodule MicrowavepropWeb.SkewtLive do
|
||||||
end
|
end
|
||||||
|
|
||||||
defp available_valid_times do
|
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()
|
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)
|
future_cutoff = DateTime.add(now, 18 * 3600, :second)
|
||||||
|
|
||||||
ProfilesFile.list_valid_times()
|
ProfilesFile.list_valid_times()
|
||||||
|
|
|
||||||
|
|
@ -48,8 +48,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||||
let points = task.points.clone();
|
let points = task.points.clone();
|
||||||
let started = Instant::now();
|
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) => {
|
Ok(stats) => {
|
||||||
let elapsed = started.elapsed();
|
let elapsed = started.elapsed();
|
||||||
info!(
|
info!(
|
||||||
|
|
|
||||||
|
|
@ -169,6 +169,7 @@ async fn worker_loop(
|
||||||
forecast_hour = task.forecast_hour,
|
forecast_hour = task.forecast_hour,
|
||||||
files = stats.score_files_written,
|
files = stats.score_files_written,
|
||||||
points = stats.point_count,
|
points = stats.point_count,
|
||||||
|
profile_cells = stats.profile_cells_written,
|
||||||
duration_s = elapsed.as_secs_f64(),
|
duration_s = elapsed.as_secs_f64(),
|
||||||
"chain step done"
|
"chain step done"
|
||||||
),
|
),
|
||||||
|
|
|
||||||
|
|
@ -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;
|
let lat_key = (lat * 1000.0).round() as i32;
|
||||||
for i in 0..nx {
|
for i in 0..nx {
|
||||||
let cell_offset = (j * nx + i) * 4;
|
let cell_offset = (j * nx + i) * 4;
|
||||||
let value = f32::from_le_bytes(
|
let value =
|
||||||
chunk[cell_offset..cell_offset + 4].try_into().unwrap(),
|
f32::from_le_bytes(chunk[cell_offset..cell_offset + 4].try_into().unwrap());
|
||||||
);
|
|
||||||
if value > UNDEFINED_VALUE / 2.0 {
|
if value > UNDEFINED_VALUE / 2.0 {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -14,8 +14,8 @@ use std::sync::Mutex;
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use chrono::{DateTime, NaiveDate, Timelike, Utc};
|
use chrono::{DateTime, NaiveDate, Timelike, Utc};
|
||||||
use tokio::task::JoinSet;
|
|
||||||
use reqwest::header::{HeaderMap, HeaderValue, RANGE};
|
use reqwest::header::{HeaderMap, HeaderValue, RANGE};
|
||||||
|
use tokio::task::JoinSet;
|
||||||
|
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
pub enum Product {
|
pub enum Product {
|
||||||
|
|
@ -754,7 +754,11 @@ mod tests {
|
||||||
// accumulate each hour. Calling get() on an expired entry must
|
// accumulate each hour. Calling get() on an expired entry must
|
||||||
// remove it so the in-memory map bounds itself to the hot set.
|
// remove it so the in-memory map bounds itself to the hot set.
|
||||||
let cache = IdxCache::new();
|
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());
|
cache.put("fresh".into(), "new".into());
|
||||||
|
|
||||||
assert_eq!(cache.len(), 2);
|
assert_eq!(cache.len(), 2);
|
||||||
|
|
|
||||||
|
|
@ -229,7 +229,10 @@ mod tests {
|
||||||
let mut c = CellValues::new();
|
let mut c = CellValues::new();
|
||||||
for i in 1..=count {
|
for i in 1..=count {
|
||||||
c.insert(format!("HGT:{i} hybrid level").into(), (i as f32) * 50.0);
|
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(
|
c.insert(
|
||||||
format!("SPFH:{i} hybrid level").into(),
|
format!("SPFH:{i} hybrid level").into(),
|
||||||
0.008 - (i as f32) * 0.0001,
|
0.008 - (i as f32) * 0.0001,
|
||||||
|
|
|
||||||
|
|
@ -62,6 +62,13 @@ pub struct ChainStepStats {
|
||||||
pub score_files_written: u32,
|
pub score_files_written: u32,
|
||||||
pub point_count: u32,
|
pub point_count: u32,
|
||||||
pub band_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(
|
#[tracing::instrument(
|
||||||
|
|
@ -128,19 +135,40 @@ pub async fn run_chain_step(
|
||||||
let merged = merge_grids(sfc_grid, prs_grid);
|
let merged = merge_grids(sfc_grid, prs_grid);
|
||||||
let point_count = merged.len() as u32;
|
let point_count = merged.len() as u32;
|
||||||
|
|
||||||
// Step 1: consume `merged` into a compact `Vec<(lat, lon, Conditions,
|
// Build `prepared` (for scoring) and `profile_entries` (for the
|
||||||
// invariants)>`. This is the biggest win — `merged`'s inner HashMaps
|
// per-valid_time profile file consumed by /skewt and the LiveView
|
||||||
// of `String -> f32` account for ~200 MB of heap (92k × ~60 string
|
// point-detail panels) in a single pass over `merged`. We need a
|
||||||
// keys) and are dropped as soon as this loop finishes.
|
// borrowing iterator (not `into_iter`) because `cell_to_profile_entry`
|
||||||
let prepared: Vec<(f64, f64, Conditions, scorer::BandInvariants)> = merged
|
// also reads from the cell — `merged` is dropped immediately after
|
||||||
.into_iter()
|
// this loop finishes, so the ~200 MB string-keyed HashMaps don't
|
||||||
.filter_map(|(key, cell)| {
|
// outlive the scoring stage. The price is briefly retaining both
|
||||||
let (lat, lon) = decoder::key_to_latlon(key);
|
// vecs at the same time as `merged`, which is well inside the
|
||||||
let conditions = cell_to_conditions(&cell, lat, lon, &valid_time)?;
|
// 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<CellEntry> = 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);
|
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.
|
// 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<ScorePoint> is ~1.8 MB so
|
// path. Memory stays bounded — each Vec<ScorePoint> is ~1.8 MB so
|
||||||
// 23 concurrent ≈ ~40 MB peak.
|
// 23 concurrent ≈ ~40 MB peak.
|
||||||
let bands = band_config::all_bands();
|
let bands = band_config::all_bands();
|
||||||
let scores_dir = scores_dir.to_path_buf();
|
|
||||||
let prepared = Arc::new(prepared);
|
let prepared = Arc::new(prepared);
|
||||||
let scored: Vec<(u32, Vec<ScorePoint>)> = {
|
let scored: Vec<(u32, Vec<ScorePoint>)> = {
|
||||||
use rayon::prelude::*;
|
use rayon::prelude::*;
|
||||||
|
|
@ -179,7 +206,7 @@ pub async fn run_chain_step(
|
||||||
let write_handles: Vec<_> = scored
|
let write_handles: Vec<_> = scored
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|(band_mhz, scores)| {
|
.map(|(band_mhz, scores)| {
|
||||||
let dir = scores_dir.clone();
|
let dir = scores_dir_owned.clone();
|
||||||
tokio::task::spawn_blocking(move || {
|
tokio::task::spawn_blocking(move || {
|
||||||
scores_file::write_atomic(&dir, band_mhz, valid_time, &scores)
|
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 {
|
for h in write_handles {
|
||||||
h.await.expect("blocking join")?;
|
h.await.expect("blocking join")?;
|
||||||
}
|
}
|
||||||
|
profile_future
|
||||||
|
.await
|
||||||
|
.expect("blocking join")
|
||||||
|
.map_err(PipelineError::ProfileWrite)?;
|
||||||
let files_written = bands.len() as u32;
|
let files_written = bands.len() as u32;
|
||||||
|
|
||||||
Ok(ChainStepStats {
|
Ok(ChainStepStats {
|
||||||
score_files_written: files_written,
|
score_files_written: files_written,
|
||||||
point_count,
|
point_count,
|
||||||
band_count: bands.len() as u32,
|
band_count: bands.len() as u32,
|
||||||
|
profile_cells_written,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue