From 989d310447d7c155e44b88049b99bb9d0f30d3c9 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 21 Apr 2026 15:54:02 -0500 Subject: [PATCH] fix(grid-rs): fall back to NOAA S3 when the LAN HRRR proxy is down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of the blank /map analysis breakdown: the skippy.w5isp.com caching proxy was unreachable from the cluster, so every f00 analysis step failed at the idx fetch and Rust never wrote a ProfilesFile. The /map point-detail panel silently degraded to empty factors because no profile ever lands on `/data/scores/profiles/`. HrrrClient now takes an optional fallback base. When the primary base exhausts its retries (5× exponential backoff), fetches retry once against the fallback — defaulted to the NOAA S3 public bucket. Forecast, analysis, and native-duct paths all go through the same fetch_blob_with_fallback helper so a proxy outage degrades to direct S3 instead of stalling. Moved the native-URL builder out of native_duct into fetcher so the single helper can build both primary and fallback URLs for the native hybrid-sigma product. --- k8s/deployment-grid-rs.yaml | 6 + k8s/deployment-hrrr-point-rs.yaml | 2 + .../prop_grid_rs/src/bin/hrrr_point_worker.rs | 4 +- rust/prop_grid_rs/src/bin/worker.rs | 8 +- rust/prop_grid_rs/src/fetcher.rs | 144 +++++++++++++++++- rust/prop_grid_rs/src/native_duct.rs | 48 +----- 6 files changed, 162 insertions(+), 50 deletions(-) diff --git a/k8s/deployment-grid-rs.yaml b/k8s/deployment-grid-rs.yaml index 8cdf65c6..3ffa547c 100644 --- a/k8s/deployment-grid-rs.yaml +++ b/k8s/deployment-grid-rs.yaml @@ -64,6 +64,12 @@ spec: # reads it directly (sqlx connects with the same URL). - name: HRRR_BASE_URL value: "http://skippy.w5isp.com:8080" + # When the LAN caching proxy at skippy is unreachable, fetches + # retry once against the NOAA S3 public bucket so a proxy + # outage degrades to direct S3 reads instead of stalling the + # hourly pipeline. + - name: HRRR_FALLBACK_BASE_URL + value: "https://noaa-hrrr-bdp-pds.s3.amazonaws.com" - name: PROP_SCORES_DIR value: "/data/scores" # Shared idx-file cache on NFS. All grid-rs replicas read/write diff --git a/k8s/deployment-hrrr-point-rs.yaml b/k8s/deployment-hrrr-point-rs.yaml index 83f978da..b15c3411 100644 --- a/k8s/deployment-hrrr-point-rs.yaml +++ b/k8s/deployment-hrrr-point-rs.yaml @@ -57,6 +57,8 @@ spec: env: - name: HRRR_BASE_URL value: "http://skippy.w5isp.com:8080" + - name: HRRR_FALLBACK_BASE_URL + value: "https://noaa-hrrr-bdp-pds.s3.amazonaws.com" - name: RUST_LOG value: "info" - name: PROP_GRID_RS_PG_CONNS 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 f4e94985..28afaeae 100644 --- a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs +++ b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs @@ -22,6 +22,8 @@ async fn main() -> Result<(), Box> { .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 hrrr_fallback = std::env::var("HRRR_FALLBACK_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()) @@ -29,7 +31,7 @@ async fn main() -> Result<(), Box> { 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 client = Arc::new(HrrrClient::new_with_fallback(&hrrr_base, &hrrr_fallback)?); let shutdown = Arc::new(AtomicBool::new(false)); spawn_signal_handler(shutdown.clone()); diff --git a/rust/prop_grid_rs/src/bin/worker.rs b/rust/prop_grid_rs/src/bin/worker.rs index cc0b2957..6fdec80a 100644 --- a/rust/prop_grid_rs/src/bin/worker.rs +++ b/rust/prop_grid_rs/src/bin/worker.rs @@ -38,6 +38,12 @@ async fn main() -> Result<(), Box> { .unwrap_or_else(|_| PathBuf::from("/data/scores")); let hrrr_base = std::env::var("HRRR_BASE_URL") .unwrap_or_else(|_| prop_grid_rs::fetcher::HRRR_BASE_DEFAULT.to_string()); + // When the primary HRRR base is a LAN caching proxy, an outage on + // that box would otherwise stall every hourly cycle. The fallback + // defaults to the NOAA S3 public bucket so the pipeline degrades + // gracefully instead. + let hrrr_fallback = std::env::var("HRRR_FALLBACK_BASE_URL") + .unwrap_or_else(|_| prop_grid_rs::fetcher::HRRR_BASE_DEFAULT.to_string()); let parallelism: usize = std::env::var("PROP_GRID_RS_PARALLELISM") .ok() .and_then(|v| v.parse().ok()) @@ -52,7 +58,7 @@ async fn main() -> Result<(), Box> { .unwrap_or((parallelism as u32) + 2); let pool = db::connect(&database_url, max_conn).await?; - let client = Arc::new(HrrrClient::new(hrrr_base)?); + let client = Arc::new(HrrrClient::new_with_fallback(&hrrr_base, &hrrr_fallback)?); let shutdown = Arc::new(AtomicBool::new(false)); spawn_signal_handler(shutdown.clone()); diff --git a/rust/prop_grid_rs/src/fetcher.rs b/rust/prop_grid_rs/src/fetcher.rs index e5fd1bd3..52df9ebe 100644 --- a/rust/prop_grid_rs/src/fetcher.rs +++ b/rust/prop_grid_rs/src/fetcher.rs @@ -84,7 +84,16 @@ pub fn hrrr_url( Product::Surface => format!("hrrr.t{hh}z.wrfsfcf{fh}.grib2"), Product::Pressure => format!("hrrr.t{hh}z.wrfprsf{fh}.grib2"), }; - format!("{base}/hrrr.{date_str}/conus/{file}") + let trimmed = base.trim_end_matches('/'); + format!("{trimmed}/hrrr.{date_str}/conus/{file}") +} + +pub fn hrrr_native_url(base: &str, date: NaiveDate, hour: u8, forecast_hour: u8) -> String { + let date_str = date.format("%Y%m%d").to_string(); + let hh = format!("{hour:02}"); + let fh = format!("{forecast_hour:02}"); + let trimmed = base.trim_end_matches('/'); + format!("{trimmed}/hrrr.{date_str}/conus/hrrr.t{hh}z.wrfnatf{fh}.grib2") } pub fn parse_idx(text: &str) -> Vec { @@ -301,25 +310,40 @@ impl IdxCache { pub struct HrrrClient { http: reqwest::Client, base: String, + // When the primary base fails every retry (e.g. an on-cluster + // caching proxy is unreachable), fetches retry once against this + // base. Production sets it to the NOAA S3 bucket so a proxy outage + // degrades to direct S3 reads instead of a hung pipeline. + fallback_base: Option, idx_cache: IdxCache, } impl HrrrClient { pub fn new(base: impl Into) -> Result { + Self::build(base.into(), None) + } + + pub fn new_with_fallback( + base: impl Into, + fallback: impl Into, + ) -> Result { + Self::build(base.into(), Some(fallback.into())) + } + + fn build(base: String, fallback: Option) -> Result { let http = reqwest::Client::builder() .timeout(REQUEST_TIMEOUT) .pool_max_idle_per_host(MAX_PARALLEL_RANGES) .build()?; - // HRRR_IDX_CACHE_DIR opts into a disk-backed idx cache shared - // across replicas via NFS (e.g. `/data/hrrr_idx`). Unset leaves - // the in-memory-only behaviour intact. let idx_cache = match std::env::var("HRRR_IDX_CACHE_DIR") { Ok(dir) if !dir.is_empty() => IdxCache::with_disk(dir), _ => IdxCache::new(), }; + let fallback = fallback.filter(|f| f != &base); Ok(Self { http, - base: base.into(), + base, + fallback_base: fallback, idx_cache, }) } @@ -332,6 +356,10 @@ impl HrrrClient { &self.base } + pub fn fallback_base(&self) -> Option<&str> { + self.fallback_base.as_deref() + } + pub async fn fetch_idx(&self, url: &str) -> Result { if let Some(body) = self.idx_cache.get(url) { return Ok(body); @@ -383,6 +411,8 @@ impl HrrrClient { } /// End-to-end: fetch idx, compute ranges for `wanted`, download grib. + /// Tries the primary base first, then the fallback base (if configured) + /// when the primary exhausts its retries. pub async fn fetch_product_blob( &self, date: NaiveDate, @@ -391,12 +421,67 @@ impl HrrrClient { forecast_hour: u8, wanted: &[(String, String)], ) -> Result, FetchError> { - let url = hrrr_url(&self.base, date, hour, product, forecast_hour); + let primary = hrrr_url(&self.base, date, hour, product, forecast_hour); + let fallback = self + .fallback_base + .as_deref() + .map(|fb| hrrr_url(fb, date, hour, product, forecast_hour)); + self.fetch_blob_with_fallback(&primary, fallback.as_deref(), wanted) + .await + } + + /// End-to-end native-level fetch (hrrr.tHHz.wrfnatfFH.grib2) with the + /// same primary-then-fallback logic as `fetch_product_blob`. Callers + /// (native_duct) supply the duct-variable `wanted` list. + pub async fn fetch_native_blob( + &self, + date: NaiveDate, + hour: u8, + forecast_hour: u8, + wanted: &[(String, String)], + ) -> Result, FetchError> { + let primary = hrrr_native_url(&self.base, date, hour, forecast_hour); + let fallback = self + .fallback_base + .as_deref() + .map(|fb| hrrr_native_url(fb, date, hour, forecast_hour)); + self.fetch_blob_with_fallback(&primary, fallback.as_deref(), wanted) + .await + } + + async fn fetch_blob_with_fallback( + &self, + primary_url: &str, + fallback_url: Option<&str>, + wanted: &[(String, String)], + ) -> Result, FetchError> { + match self.fetch_blob_at(primary_url, wanted).await { + Ok(blob) => Ok(blob), + Err(primary_err) => match fallback_url { + Some(fb) => { + tracing::warn!( + error = %primary_err, + primary = primary_url, + fallback = fb, + "hrrr primary failed; retrying against fallback" + ); + self.fetch_blob_at(fb, wanted).await + } + None => Err(primary_err), + }, + } + } + + async fn fetch_blob_at( + &self, + url: &str, + wanted: &[(String, String)], + ) -> Result, FetchError> { let idx_url = format!("{url}.idx"); let idx_body = self.fetch_idx(&idx_url).await?; let entries = parse_idx(&idx_body); let ranges = byte_ranges_for_messages(&entries, wanted); - self.download_grib_ranges(&url, ranges).await + self.download_grib_ranges(url, ranges).await } } @@ -489,6 +574,51 @@ mod tests { ); } + #[test] + fn hrrr_url_trims_trailing_slash() { + let d = NaiveDate::from_ymd_opt(2026, 4, 19).unwrap(); + assert_eq!( + hrrr_url("https://x.y/", d, 14, Product::Surface, 0), + "https://x.y/hrrr.20260419/conus/hrrr.t14z.wrfsfcf00.grib2" + ); + } + + #[test] + fn hrrr_native_url_format() { + let d = NaiveDate::from_ymd_opt(2026, 4, 19).unwrap(); + assert_eq!( + hrrr_native_url("https://x.y", d, 14, 0), + "https://x.y/hrrr.20260419/conus/hrrr.t14z.wrfnatf00.grib2" + ); + assert_eq!( + hrrr_native_url("https://x.y/", d, 4, 7), + "https://x.y/hrrr.20260419/conus/hrrr.t04z.wrfnatf07.grib2" + ); + } + + #[test] + fn new_stores_no_fallback() { + let c = HrrrClient::new("https://primary").unwrap(); + assert_eq!(c.base(), "https://primary"); + assert!(c.fallback_base().is_none()); + } + + #[test] + fn new_with_fallback_stores_fallback() { + let c = HrrrClient::new_with_fallback("http://primary", "https://fallback").unwrap(); + assert_eq!(c.base(), "http://primary"); + assert_eq!(c.fallback_base(), Some("https://fallback")); + } + + #[test] + fn fallback_equal_to_primary_is_dropped() { + // Pointing fallback at the same base adds no resilience and + // would double the wait on every failure. Treat it as "no + // fallback" so retries stop after the primary exhausts. + let c = HrrrClient::new_with_fallback("https://same", "https://same").unwrap(); + assert!(c.fallback_base().is_none()); + } + #[test] fn idx_parsing_drops_bad_lines() { let idx = "1:0:d=2024:TMP:2 m above ground:anl:\nbad\n2:12345:d=2024:DPT:2 m above ground:anl:"; diff --git a/rust/prop_grid_rs/src/native_duct.rs b/rust/prop_grid_rs/src/native_duct.rs index 7a1cb360..dd54a0a2 100644 --- a/rust/prop_grid_rs/src/native_duct.rs +++ b/rust/prop_grid_rs/src/native_duct.rs @@ -20,7 +20,7 @@ use chrono::NaiveDate; use crate::decoder::{self, CellValues, DecodeError, PointGrid}; use crate::duct::{self, NativeProfile}; -use crate::fetcher::{self, byte_ranges_for_messages, parse_idx, HrrrClient}; +use crate::fetcher::{self, HrrrClient}; use crate::grid::GridSpec; /// The 4 variables needed for duct detection. UGRD/VGRD/TKE were @@ -66,27 +66,9 @@ pub fn match_pattern() -> String { format!(":({}):.*hybrid level:", DUCT_VARIABLES.join("|")) } -/// URL builder for native-level HRRR files. -/// -/// ``` -/// use chrono::NaiveDate; -/// use prop_grid_rs::native_duct::hrrr_native_url; -/// assert_eq!( -/// hrrr_native_url("https://example", NaiveDate::from_ymd_opt(2026, 4, 19).unwrap(), 14, 0), -/// "https://example/hrrr.20260419/conus/hrrr.t14z.wrfnatf00.grib2" -/// ); -/// ``` -pub fn hrrr_native_url(base: &str, date: NaiveDate, hour: u8, forecast_hour: u8) -> String { - format!( - "{}/hrrr.{:04}{:02}{:02}/conus/hrrr.t{:02}z.wrfnatf{:02}.grib2", - base.trim_end_matches('/'), - date.format("%Y").to_string().parse::().unwrap_or(0), - date.format("%m").to_string().parse::().unwrap_or(0), - date.format("%d").to_string().parse::().unwrap_or(0), - hour, - forecast_hour - ) -} +/// URL builder for native-level HRRR files. Re-export of +/// `fetcher::hrrr_native_url` kept for external callers. +pub use crate::fetcher::hrrr_native_url; /// Fetch the native-level duct grid for a (date, hour, forecast_hour) /// triple. Downloads the byte-range subset, decodes via wgrib2, and @@ -98,13 +80,10 @@ pub async fn fetch_native_duct_grid( forecast_hour: u8, grid_spec: GridSpec, ) -> Result, NativeDuctError> { - let url = hrrr_native_url(client.base(), date, hour, forecast_hour); - let idx_url = format!("{url}.idx"); - let idx_body = client.fetch_idx(&idx_url).await?; - let entries = parse_idx(&idx_body); let messages = duct_messages(); - let ranges = byte_ranges_for_messages(&entries, &messages); - let blob = client.download_grib_ranges(&url, ranges).await?; + let blob = client + .fetch_native_blob(date, hour, forecast_hour, &messages) + .await?; // wgrib2 decode must run on the blocking pool — it forks a // subprocess and the stdout read is synchronous IO. @@ -239,19 +218,6 @@ mod tests { assert_eq!(p, ":(TMP|SPFH|HGT|PRES):.*hybrid level:"); } - #[test] - fn hrrr_native_url_format() { - let d = NaiveDate::from_ymd_opt(2026, 4, 19).unwrap(); - assert_eq!( - hrrr_native_url("https://x.y", d, 14, 0), - "https://x.y/hrrr.20260419/conus/hrrr.t14z.wrfnatf00.grib2" - ); - assert_eq!( - hrrr_native_url("https://x.y/", d, 4, 7), - "https://x.y/hrrr.20260419/conus/hrrr.t04z.wrfnatf07.grib2" - ); - } - fn cell_with_levels(count: usize) -> CellValues { let mut c = CellValues::new(); for i in 1..=count {