perf(rust): disk-backed HRRR idx cache + scale workers off talos5

Shared NFS idx cache at /data/hrrr_idx lets all prop-grid-rs and
hrrr-point-rs replicas deduplicate redundant NOAA S3 idx fetches.
Opt-in via HRRR_IDX_CACHE_DIR; falls back to in-memory-only when unset.

Also scales prop-grid-rs to 3 replicas (no talos5 pinning),
hrrr-point-rs to 2 replicas, and drops hot pod replicas 4→3 so the
physical-host anti-affinity still fits with talos5 cordoned off for
Postgres.

k8s/deployment-grid-rs.yaml

@@ -9,10 +9,11 @@ metadata:
   # commercial-link degradation, ProfilesFile write). f01..f18 land
   # here from the Rust worker.
 spec:
-  # Two replicas for HA. FOR UPDATE SKIP LOCKED lets both pods claim
-  # from grid_tasks without coordination. Anti-affinity spreads them
-  # across hosts so talos5 going down doesn't halt the hourly chain.
-  replicas: 2
+  # Three replicas claim from grid_tasks via FOR UPDATE SKIP LOCKED
+  # with no coordination. Anti-affinity spreads them across distinct
+  # hosts so any single node going down leaves the hourly chain with
+  # two workers still draining.
+  replicas: 3
   minReadySeconds: 5
   strategy:
     type: RollingUpdate
@@ -34,10 +35,8 @@ spec:
         prometheus.io/port: "9100"
         prometheus.io/path: "/metrics"
     spec:
-      # Primary replica prefers talos5 (32 GB NUC with NVMe). Secondary
-      # replica lands anywhere else the scheduler allows — the
-      # required anti-affinity rule below guarantees the two don't
-      # stack on the same host.
+      # No node pinning — let the scheduler place replicas on any host
+      # the anti-affinity rule allows. talos5 is reserved for Postgres.
       affinity:
         podAntiAffinity:
           requiredDuringSchedulingIgnoredDuringExecution:
@@ -45,19 +44,6 @@ spec:
                 matchLabels:
                   app: prop-grid-rs
               topologyKey: kubernetes.io/hostname
-        nodeAffinity:
-          preferredDuringSchedulingIgnoredDuringExecution:
-            - weight: 100
-              preference:
-                matchExpressions:
-                  - key: prop-grid-rs
-                    operator: In
-                    values: ["primary"]
-      tolerations:
-        - key: workload
-          operator: Equal
-          value: grid-rs
-          effect: NoSchedule
       imagePullSecrets:
         - name: forgejo-registry
       securityContext:
@@ -80,6 +66,11 @@ spec:
               value: "http://skippy.w5isp.com:8080"
             - name: PROP_SCORES_DIR
               value: "/data/scores"
+            # Shared idx-file cache on NFS. All grid-rs replicas read/write
+            # through the same directory; idx bodies are immutable per HRRR
+            # run, so cross-replica hits eliminate redundant NOAA S3 calls.
+            - name: HRRR_IDX_CACHE_DIR
+              value: "/data/hrrr_idx"
             - name: RUST_LOG
               value: "info"
             # Single concurrent task per pod. 2 replicas → 2 concurrent

k8s/deployment-hrrr-point-rs.yaml

@@ -8,7 +8,9 @@ metadata:
   # points JSONB array); this deployment drains via FOR UPDATE SKIP
   # LOCKED, fetches each HRRR cycle once, and upserts hrrr_profiles.
 spec:
-  replicas: 1
+  # Two replicas for parallelism on QSO-submit spikes. FOR UPDATE SKIP
+  # LOCKED on hrrr_fetch_tasks lets both claim without coordination.
+  replicas: 2
   strategy:
     type: RollingUpdate
     rollingUpdate:
@@ -59,6 +61,10 @@ spec:
               value: "info"
             - name: PROP_GRID_RS_PG_CONNS
               value: "4"
+            # Shared NFS idx cache — same directory as prop-grid-rs so a
+            # chain-step idx fetch warms both workers.
+            - name: HRRR_IDX_CACHE_DIR
+              value: "/data/hrrr_idx"
             - name: OTEL_EXPORTER_OTLP_ENDPOINT
               value: "http://otel-collector.observability.svc.cluster.local:4317"
           envFrom:
@@ -87,3 +93,11 @@ spec:
                 - ALL
             seccompProfile:
               type: RuntimeDefault
+          volumeMounts:
+            - name: data
+              mountPath: /data
+      volumes:
+        - name: data
+          nfs:
+            server: 10.0.15.103
+            path: /data

k8s/deployment.yaml

@@ -4,7 +4,7 @@ metadata:
   name: prop
   namespace: prop
 spec:
-  replicas: 4
+  replicas: 3
   minReadySeconds: 5
   strategy:
     type: RollingUpdate

rust/prop_grid_rs/src/fetcher.rs

@@ -7,6 +7,9 @@
 //!   * retry on HTTP 429/5xx with jittered exponential backoff
 
 use std::collections::HashMap;
+use std::fs;
+use std::io::Write;
+use std::path::{Path, PathBuf};
 use std::sync::Mutex;
 use std::time::{Duration, Instant};
 
@@ -98,13 +101,15 @@ fn parse_idx_line(line: &str) -> Option<IdxEntry> {
     parts.next()?; // date
     let var = parts.next()?.to_string();
     let level = parts.next()?.to_string();
-    Some(IdxEntry { msg, offset, var, level })
+    Some(IdxEntry {
+        msg,
+        offset,
+        var,
+        level,
+    })
 }
 
-pub fn byte_ranges_for_messages(
-    idx: &[IdxEntry],
-    wanted: &[(String, String)],
-) -> Vec<ByteRange> {
+pub fn byte_ranges_for_messages(idx: &[IdxEntry], wanted: &[(String, String)]) -> Vec<ByteRange> {
     let mut out = Vec::new();
     for (var, level) in wanted {
         for (i, entry) in idx.iter().enumerate() {
@@ -113,7 +118,10 @@ pub fn byte_ranges_for_messages(
                     .get(i + 1)
                     .map(|nxt| nxt.offset.saturating_sub(1))
                     .unwrap_or(entry.offset + 10_000_000);
-                out.push(ByteRange { start: entry.offset, end });
+                out.push(ByteRange {
+                    start: entry.offset,
+                    end,
+                });
             }
         }
     }
@@ -166,15 +174,73 @@ pub fn surface_messages_owned() -> Vec<(String, String)> {
 
 // ── Idx cache ────────────────────────────────────────────────────────
 
+// Disk TTL is longer than the in-memory TTL because idx files are
+// immutable per HRRR run — cross-replica sharing on NFS benefits from
+// keeping hits alive across pod restarts and the next hourly cycle.
+pub const IDX_DISK_TTL: Duration = Duration::from_secs(24 * 3_600);
+
 #[derive(Debug, Clone)]
 struct IdxCacheEntry {
     body: String,
     inserted: Instant,
 }
 
+#[derive(Debug)]
+pub struct DiskIdxCache {
+    dir: PathBuf,
+    ttl: Duration,
+}
+
+impl DiskIdxCache {
+    pub fn new(dir: impl AsRef<Path>, ttl: Duration) -> Self {
+        let dir = dir.as_ref().to_path_buf();
+        let _ = fs::create_dir_all(&dir);
+        Self { dir, ttl }
+    }
+
+    fn path_for(&self, url: &str) -> PathBuf {
+        // Sanitize URL into a safe filename. Collisions would require
+        // two distinct URLs to differ only in unsafe characters mapped
+        // to '_' — not possible for HRRR S3 URLs.
+        let mut sanitized = String::with_capacity(url.len());
+        for c in url.chars() {
+            if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' {
+                sanitized.push(c);
+            } else {
+                sanitized.push('_');
+            }
+        }
+        self.dir.join(sanitized)
+    }
+
+    pub fn get(&self, url: &str) -> Option<String> {
+        let path = self.path_for(url);
+        let meta = fs::metadata(&path).ok()?;
+        let mtime = meta.modified().ok()?;
+        if mtime.elapsed().ok()? > self.ttl {
+            return None;
+        }
+        fs::read_to_string(&path).ok()
+    }
+
+    pub fn put(&self, url: &str, body: &str) {
+        // Atomic write: tmp file + rename so concurrent readers across
+        // replicas never observe a partial body.
+        let path = self.path_for(url);
+        let tmp = path.with_extension("tmp");
+        if let Ok(mut f) = fs::File::create(&tmp) {
+            if f.write_all(body.as_bytes()).is_ok() {
+                drop(f);
+                let _ = fs::rename(&tmp, &path);
+            }
+        }
+    }
+}
+
 #[derive(Debug, Default)]
 pub struct IdxCache {
     inner: Mutex<HashMap<String, IdxCacheEntry>>,
+    disk: Option<DiskIdxCache>,
 }
 
 impl IdxCache {
@@ -182,19 +248,50 @@ impl IdxCache {
         Self::default()
     }
 
-    pub fn get(&self, url: &str) -> Option<String> {
-        let map = self.inner.lock().ok()?;
-        let entry = map.get(url)?;
-        if entry.inserted.elapsed() < IDX_TTL {
-            Some(entry.body.clone())
-        } else {
-            None
+    pub fn with_disk(dir: impl AsRef<Path>) -> Self {
+        Self {
+            inner: Mutex::new(HashMap::new()),
+            disk: Some(DiskIdxCache::new(dir, IDX_DISK_TTL)),
         }
     }
 
+    pub fn get(&self, url: &str) -> Option<String> {
+        if let Ok(map) = self.inner.lock() {
+            if let Some(entry) = map.get(url) {
+                if entry.inserted.elapsed() < IDX_TTL {
+                    return Some(entry.body.clone());
+                }
+            }
+        }
+        if let Some(disk) = &self.disk {
+            if let Some(body) = disk.get(url) {
+                if let Ok(mut map) = self.inner.lock() {
+                    map.insert(
+                        url.to_string(),
+                        IdxCacheEntry {
+                            body: body.clone(),
+                            inserted: Instant::now(),
+                        },
+                    );
+                }
+                return Some(body);
+            }
+        }
+        None
+    }
+
     pub fn put(&self, url: String, body: String) {
+        if let Some(disk) = &self.disk {
+            disk.put(&url, &body);
+        }
         if let Ok(mut map) = self.inner.lock() {
-            map.insert(url, IdxCacheEntry { body, inserted: Instant::now() });
+            map.insert(
+                url,
+                IdxCacheEntry {
+                    body,
+                    inserted: Instant::now(),
+                },
+            );
         }
     }
 }
@@ -213,10 +310,17 @@ impl HrrrClient {
             .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(),
+        };
         Ok(Self {
             http,
             base: base.into(),
-            idx_cache: IdxCache::new(),
+            idx_cache,
         })
     }
 
@@ -397,8 +501,18 @@ mod tests {
     #[test]
     fn byte_ranges_cover_last_message_with_fallback() {
         let idx = vec![
-            IdxEntry { msg: 1, offset: 0, var: "A".into(), level: "x".into() },
-            IdxEntry { msg: 2, offset: 100, var: "B".into(), level: "y".into() },
+            IdxEntry {
+                msg: 1,
+                offset: 0,
+                var: "A".into(),
+                level: "x".into(),
+            },
+            IdxEntry {
+                msg: 2,
+                offset: 100,
+                var: "B".into(),
+                level: "y".into(),
+            },
         ];
         let wanted = vec![("A".into(), "x".into()), ("B".into(), "y".into())];
         let r = byte_ranges_for_messages(&idx, &wanted);
@@ -413,8 +527,14 @@ mod tests {
     fn adjacent_ranges_merge() {
         let merged = merge_ranges(vec![
             ByteRange { start: 0, end: 99 },
-            ByteRange { start: 100, end: 200 },
-            ByteRange { start: 500, end: 600 },
+            ByteRange {
+                start: 100,
+                end: 200,
+            },
+            ByteRange {
+                start: 500,
+                end: 600,
+            },
         ]);
         assert_eq!(merged.len(), 2);
         assert_eq!(merged[0].start, 0);
@@ -442,4 +562,38 @@ mod tests {
         assert_eq!(cache.get("u").as_deref(), Some("body"));
         assert!(cache.get("missing").is_none());
     }
+
+    #[test]
+    fn disk_idx_cache_round_trips() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let cache = DiskIdxCache::new(tmp.path(), Duration::from_secs(3_600));
+        cache.put("https://example.com/a.grib2.idx", "body1");
+        assert_eq!(
+            cache.get("https://example.com/a.grib2.idx").as_deref(),
+            Some("body1")
+        );
+        assert!(cache.get("https://example.com/missing.idx").is_none());
+    }
+
+    #[test]
+    fn disk_idx_cache_respects_ttl() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let cache = DiskIdxCache::new(tmp.path(), Duration::from_millis(10));
+        cache.put("u", "body");
+        std::thread::sleep(Duration::from_millis(30));
+        assert!(cache.get("u").is_none());
+    }
+
+    #[test]
+    fn idx_cache_disk_fallback_rewarms_memory() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        // Seed disk via one cache instance...
+        let c1 = IdxCache::with_disk(tmp.path());
+        c1.put("u".into(), "body".into());
+
+        // A fresh cache simulates a new pod — in-memory map is empty
+        // but the disk entry must still resolve.
+        let c2 = IdxCache::with_disk(tmp.path());
+        assert_eq!(c2.get("u").as_deref(), Some("body"));
+    }
 }