graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(grid-rs): fall back to NOAA S3 when the LAN HRRR proxy is down

Browse source 
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.
This commit is contained in:
Graham McIntire 2026-04-21 15:54:02 -05:00
parent d0dbb5c5b4
commit 989d310447
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
6 changed files with 162 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
k8s/deployment-grid-rs.yaml
View file

@ -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

2
k8s/deployment-hrrr-point-rs.yaml
View file

@ -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

4
rust/prop_grid_rs/src/bin/hrrr_point_worker.rs
View file

@ -22,6 +22,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
.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<dyn std::error::Error>> {
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());

8
rust/prop_grid_rs/src/bin/worker.rs
View file

@ -38,6 +38,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
.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<dyn std::error::Error>> {
.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());

144
rust/prop_grid_rs/src/fetcher.rs
View file

@ -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<IdxEntry> {
@ -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<String>,
idx_cache: IdxCache,
}
impl HrrrClient {
pub fn new(base: impl Into<String>) -> Result<Self, FetchError> {
Self::build(base.into(), None)
}
pub fn new_with_fallback(
base: impl Into<String>,
fallback: impl Into<String>,
) -> Result<Self, FetchError> {
Self::build(base.into(), Some(fallback.into()))
}
fn build(base: String, fallback: Option<String>) -> Result<Self, FetchError> {
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<String, FetchError> {
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<Vec<u8>, 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<Vec<u8>, 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<Vec<u8>, 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<Vec<u8>, 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:\n<html>bad</html>\n2:12345:d=2024:DPT:2 m above ground:anl:";

48
rust/prop_grid_rs/src/native_duct.rs
View file

@ -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::<u32>().unwrap_or(0),
date.format("%m").to_string().parse::<u32>().unwrap_or(0),
date.format("%d").to_string().parse::<u32>().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<HashMap<(i32, i32), DuctMetrics>, 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 {