diff --git a/rust/prop_grid_rs/src/fetcher.rs b/rust/prop_grid_rs/src/fetcher.rs index 5b725bd8..1af50ab0 100644 --- a/rust/prop_grid_rs/src/fetcher.rs +++ b/rust/prop_grid_rs/src/fetcher.rs @@ -600,7 +600,7 @@ async fn retry_get_range( let mut headers = HeaderMap::new(); headers.insert(RANGE, value); - for attempt in 0..5 { + for attempt in 0..3 { match client.get(url).headers(headers.clone()).send().await { Ok(resp) => { let status = resp.status().as_u16(); @@ -622,6 +622,9 @@ async fn retry_get_range( } fn is_retryable_status(status: u16) -> bool { + // 404 means the forecast hour isn't published yet — a permanent + // failure that retrying won't fix. Fail fast so the caller can + // move on instead of burning 31 s in exponential backoff. matches!(status, 429 | 500 | 502 | 503 | 504) } diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index aa470394..11e90fab 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -104,47 +104,54 @@ pub async fn run_chain_step( let sfc_wanted = fetcher::surface_messages_owned(); let prs_wanted = fetcher::pressure_messages_grid(); - let sfc_fut = client.fetch_product_blob( + + // Pin so we can drive them sequentially without losing the prs + // future when sfc resolves first. + let mut sfc_fut = Box::pin(client.fetch_product_blob( date, hour, Product::Surface, step.forecast_hour, &sfc_wanted, - ); - let prs_fut = client.fetch_product_blob( + )); + let mut prs_fut = Box::pin(client.fetch_product_blob( date, hour, Product::Pressure, step.forecast_hour, &prs_wanted, - ); - let fetch_started = std::time::Instant::now(); - let (sfc_blob, prs_blob) = tokio::try_join!(sfc_fut, prs_fut)?; - metrics::record_stage("fetch", fetch_started.elapsed()); - if sfc_blob.is_empty() { - return Err(PipelineError::EmptyGrib); - } + )); let grid_spec = wgrib2_grid_spec(); let sfc_pattern = match_pattern(&sfc_wanted); let prs_pattern = match_pattern(&prs_wanted); - // Decoder runs wgrib2 subprocess — block so we don't hog the tokio - // reactor. `spawn_blocking` lifts it onto the dedicated pool. - let decode_started = std::time::Instant::now(); - let grid_spec_sfc = grid_spec; - let sfc_grid = tokio::task::spawn_blocking(move || { - decoder::extract_grid(&sfc_blob, &sfc_pattern, grid_spec_sfc) - }) - .await - .expect("blocking join")?; + let fetch_started = std::time::Instant::now(); + // sfc (9 messages) resolves before prs (39 messages). Start sfc + // decode as soon as it lands so the wgrib2 subprocess overlaps + // with the remaining prs byte-range downloads. + let sfc_blob = sfc_fut.as_mut().await?; + if sfc_blob.is_empty() { + return Err(PipelineError::EmptyGrib); + } + let grid_spec_sfc = grid_spec; + let sfc_decode = tokio::task::spawn_blocking(move || { + decoder::extract_grid(&sfc_blob, &sfc_pattern, grid_spec_sfc) + }); + + // Now await prs while sfc decodes in the background. + let prs_blob = prs_fut.as_mut().await?; + metrics::record_stage("fetch", fetch_started.elapsed()); + + // Both decodes run concurrently on the blocking pool. + let decode_started = std::time::Instant::now(); let grid_spec_prs = grid_spec; - let prs_grid = tokio::task::spawn_blocking(move || { + let prs_decode = tokio::task::spawn_blocking(move || { decoder::extract_grid(&prs_blob, &prs_pattern, grid_spec_prs) - }) - .await - .expect("blocking join")?; + }); + let sfc_grid = sfc_decode.await.expect("blocking join")?; + let prs_grid = prs_decode.await.expect("blocking join")?; metrics::record_stage("decode", decode_started.elapsed()); let mut merged = sfc_grid; @@ -179,36 +186,19 @@ pub async fn run_chain_step( }) }; - // Dense `.sgrid` body built from the scalar rows while they're - // still available. The old chunked write below moves the rows; this - // must come first. + // Dense `.sgrid` body built from the scalar rows. This is the + // primary weather-scalar artifact; the Elixir reader prefers it + // over the legacy chunked `.mp.gz` format, which is only written + // as a cold-miss fallback by the Elixir side. let sgrid_body = sgrid::build_body(&spec, &fused.scalar_rows); - // Persist the derived scalar artifact alongside the raw profile so - // Elixir's `/weather` reads land on a pre-derived chunk file - // instead of falling back to the cold ProfilesFile decode + per-cell - // SoundingParams + WeatherLayers derivation. Identical wire format - // to what the Elixir writer produces — the reader doesn't care which - // side wrote the bytes. - let scores_dir_for_scalars = scores_dir_owned.clone(); - let scalar_future = { - let rows = fused.scalar_rows; - tokio::task::spawn_blocking(move || { - metrics::observe_stage("write_scalar", || { - weather_scalar_file::write_atomic(&scores_dir_for_scalars, valid_time, &rows) - .map(|_| 0u32) - }) - }) - }; - - // Write the `.sgrid` in parallel with the chunked write. During - // the transition window both formats coexist; the Elixir reader - // prefers `.sgrid` when it exists. let scores_dir_for_sgrid = scores_dir_owned.clone(); let sgrid_future = { tokio::task::spawn_blocking(move || { - sgrid::write_atomic(&scores_dir_for_sgrid, valid_time, &spec, &sgrid_body, false) - .map(|_| 0u32) + metrics::observe_stage("write_sgrid", || { + sgrid::write_atomic(&scores_dir_for_sgrid, valid_time, &spec, &sgrid_body, false) + .map(|_| 0u32) + }) }) }; @@ -226,10 +216,6 @@ pub async fn run_chain_step( .await .expect("blocking join") .map_err(PipelineError::ProfileWrite)?; - scalar_future - .await - .expect("blocking join") - .map_err(PipelineError::ScalarWrite)?; sgrid_future .await .expect("blocking join") @@ -1433,31 +1419,17 @@ pub async fn run_analysis_step( }) }; - // Dense `.sgrid` body built from the scalar rows while they're - // still available (the chunked write below moves them). + // Dense `.sgrid` body built from the scalar rows. Primary + // weather-scalar artifact; see run_chain_step for rationale. let sgrid_body = sgrid::build_body(&spec, &fused.scalar_rows); - // Same wire format as Elixir's `ScalarFile`. See run_chain_step - // (forecast hours) for the full rationale — both code paths go - // through the same writer so f00 and f01..f48 land identical - // artifacts on NFS. - let scores_dir_for_scalars = scores_dir_owned.clone(); - let scalar_future = { - let rows = fused.scalar_rows; - tokio::task::spawn_blocking(move || { - metrics::observe_stage("write_scalar", || { - weather_scalar_file::write_atomic(&scores_dir_for_scalars, valid_time, &rows) - .map(|_| 0u32) - }) - }) - }; - - // Write the `.sgrid` alongside the chunked scalar artifact. let scores_dir_for_sgrid = scores_dir_owned.clone(); let sgrid_future = { tokio::task::spawn_blocking(move || { - sgrid::write_atomic(&scores_dir_for_sgrid, valid_time, &spec, &sgrid_body, false) - .map(|_| 0u32) + metrics::observe_stage("write_sgrid", || { + sgrid::write_atomic(&scores_dir_for_sgrid, valid_time, &spec, &sgrid_body, false) + .map(|_| 0u32) + }) }) }; @@ -1475,10 +1447,6 @@ pub async fn run_analysis_step( .await .expect("blocking join") .map_err(PipelineError::ProfileWrite)?; - scalar_future - .await - .expect("blocking join") - .map_err(PipelineError::ScalarWrite)?; sgrid_future .await .expect("blocking join")