diff --git a/k8s/deployment-grid-rs.yaml b/k8s/deployment-grid-rs.yaml index a2c3d952..0baf530a 100644 --- a/k8s/deployment-grid-rs.yaml +++ b/k8s/deployment-grid-rs.yaml @@ -9,7 +9,10 @@ metadata: # commercial-link degradation, ProfilesFile write). f01..f18 land # here from the Rust worker. spec: - replicas: 1 + # 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 minReadySeconds: 5 strategy: type: RollingUpdate @@ -24,14 +27,32 @@ spec: labels: app: prop-grid-rs tier: grid-rs + annotations: + # Scraped by the Prometheus at 10.0.15.25. Metrics endpoint is + # on container port 9100 (see METRICS_ADDR env below). + prometheus.io/scrape: "true" + prometheus.io/port: "9100" + prometheus.io/path: "/metrics" spec: - # Pin to talos5 — the 32 GB NUC with NVMe. The worker peaks at ~500 Mi - # steady-state but benefits from local NVMe for the NFS scores-dir - # writes (faster fsync) and from never sharing the node with a hot - # Elixir pod. Label talos5 with `prop-grid-rs=primary` and taint it - # `workload=grid-rs:NoSchedule` to enforce this. - nodeSelector: - prop-grid-rs: "primary" + # 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. + affinity: + podAntiAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + - labelSelector: + 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 @@ -61,24 +82,41 @@ spec: value: "/data/scores" - name: RUST_LOG value: "info" - # talos5 is 4c/4t (i5-7260U). Claiming 4 forecast hours in - # parallel drops full-chain wall time from ~18 min serial to - # ~5 min. Each concurrent task peaks around 250 Mi after the - # streamed-bands refactor; 4×250 ≈ 1 Gi is comfortably under - # the 2 Gi memory limit. + # talos5 is 4c/4t (i5-7260U). Each replica claims up to this + # many forecast hours in parallel. With 2 replicas × 3 → 6 + # concurrent tasks, the 18-hour chain completes in ~3 min. + # Per-task peak RAM is ~250 Mi post-streaming refactor, so + # 3×250 ≈ 750 Mi per pod, well under the 2 Gi limit. The + # secondary replica may land on a smaller node — 3 keeps it + # honest. - name: PROP_GRID_RS_PARALLELISM - value: "4" + value: "3" # Pool size = parallelism + 2 (NOTIFY + retry slack). Kept in # sync with PROP_GRID_RS_PARALLELISM when either changes. - name: PROP_GRID_RS_PG_CONNS - value: "6" + value: "5" + - name: METRICS_ADDR + value: "0.0.0.0:9100" envFrom: - secretRef: name: prop-secrets - # No HTTP surface. Liveness is "the process is still running"; - # let the kubelet restart us if we crash. Claiming work from - # grid_tasks is the implicit readiness signal — an idle pod is - # still healthy (the queue is simply empty). + ports: + - name: metrics + containerPort: 9100 + protocol: TCP + # Liveness is "the process is still running"; let the kubelet + # restart us if we crash. Claiming work from grid_tasks is the + # implicit readiness signal — an idle pod is still healthy + # (the queue is simply empty). A minimal readiness check on + # /health confirms the metrics listener is up, which doubles + # as evidence the tokio runtime is alive. + readinessProbe: + httpGet: + path: /health + port: 9100 + initialDelaySeconds: 3 + periodSeconds: 10 + timeoutSeconds: 2 resources: requests: cpu: 100m diff --git a/lib/microwaveprop/propagation/grid_task_enqueuer.ex b/lib/microwaveprop/propagation/grid_task_enqueuer.ex index b264923c..b1013b5e 100644 --- a/lib/microwaveprop/propagation/grid_task_enqueuer.ex +++ b/lib/microwaveprop/propagation/grid_task_enqueuer.ex @@ -31,6 +31,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do valid_time: DateTime.add(run_time, fh * 3600, :second), status: "queued", attempt: 0, + kind: "forecast", claimed_at: nil, completed_at: nil, error: nil, @@ -42,7 +43,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do {count, _} = Repo.insert_all("grid_tasks", rows, on_conflict: :nothing, - conflict_target: [:run_time, :forecast_hour] + conflict_target: [:run_time, :forecast_hour, :kind] ) if count > 0 do diff --git a/rust/prop_grid_rs/Cargo.lock b/rust/prop_grid_rs/Cargo.lock index cb46021a..fca3dc1c 100644 --- a/rust/prop_grid_rs/Cargo.lock +++ b/rust/prop_grid_rs/Cargo.lock @@ -32,6 +32,17 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "async-trait" +version = "0.1.89" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "atoi" version = "2.0.0" @@ -53,6 +64,56 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "axum" +version = "0.7.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" +dependencies = [ + "async-trait", + "axum-core", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "rustversion", + "serde", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09f2bd6146b97ae3359fa0cc6d6b376d9539582c7b4220f041a33ec24c226199" +dependencies = [ + "async-trait", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "rustversion", + "sync_wrapper", + "tower-layer", + "tower-service", +] + [[package]] name = "base64" version = "0.22.1" @@ -191,6 +252,25 @@ version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" +[[package]] +name = "crossbeam-deque" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-queue" version = "0.3.12" @@ -332,6 +412,12 @@ dependencies = [ "spin", ] +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "foldhash" version = "0.1.5" @@ -606,6 +692,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hyper" version = "1.9.0" @@ -619,6 +711,7 @@ dependencies = [ "http", "http-body", "httparse", + "httpdate", "itoa", "pin-project-lite", "smallvec", @@ -935,6 +1028,12 @@ dependencies = [ "regex-automata", ] +[[package]] +name = "matchit" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" + [[package]] name = "md-5" version = "0.10.6" @@ -951,6 +1050,12 @@ version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "mio" version = "1.2.0" @@ -1163,11 +1268,26 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prometheus" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1" +dependencies = [ + "cfg-if", + "fnv", + "lazy_static", + "memchr", + "parking_lot", + "thiserror 1.0.69", +] + [[package]] name = "prop_grid_rs" version = "0.1.0" dependencies = [ "anyhow", + "axum", "bincode", "byteorder", "bytes", @@ -1175,12 +1295,14 @@ dependencies = [ "futures", "num_cpus", "pretty_assertions", + "prometheus", + "rayon", "reqwest", "serde", "serde_json", "sqlx", "tempfile", - "thiserror", + "thiserror 2.0.18", "tokio", "tracing", "tracing-subscriber", @@ -1201,7 +1323,7 @@ dependencies = [ "rustc-hash", "rustls", "socket2", - "thiserror", + "thiserror 2.0.18", "tokio", "tracing", "web-time", @@ -1222,7 +1344,7 @@ dependencies = [ "rustls", "rustls-pki-types", "slab", - "thiserror", + "thiserror 2.0.18", "tinyvec", "tracing", "web-time", @@ -1322,6 +1444,26 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1708,7 +1850,7 @@ dependencies = [ "serde_json", "sha2", "smallvec", - "thiserror", + "thiserror 2.0.18", "tokio", "tokio-stream", "tracing", @@ -1793,7 +1935,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.18", "tracing", "uuid", "whoami", @@ -1832,7 +1974,7 @@ dependencies = [ "smallvec", "sqlx-core", "stringprep", - "thiserror", + "thiserror 2.0.18", "tracing", "uuid", "whoami", @@ -1858,7 +2000,7 @@ dependencies = [ "serde", "serde_urlencoded", "sqlx-core", - "thiserror", + "thiserror 2.0.18", "tracing", "url", "uuid", @@ -1931,13 +2073,33 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", +] + [[package]] name = "thiserror" version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl", + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn", ] [[package]] diff --git a/rust/prop_grid_rs/Cargo.toml b/rust/prop_grid_rs/Cargo.toml index 759b285e..26c01db9 100644 --- a/rust/prop_grid_rs/Cargo.toml +++ b/rust/prop_grid_rs/Cargo.toml @@ -29,6 +29,9 @@ bincode = "1.3" num_cpus = "1" byteorder = "1" futures = "0.3" +rayon = "1.10" +axum = { version = "0.7", default-features = false, features = ["http1", "tokio"] } +prometheus = { version = "0.13", default-features = false } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/rust/prop_grid_rs/src/bin/worker.rs b/rust/prop_grid_rs/src/bin/worker.rs index e8db59e6..a0372bae 100644 --- a/rust/prop_grid_rs/src/bin/worker.rs +++ b/rust/prop_grid_rs/src/bin/worker.rs @@ -10,12 +10,13 @@ //! failed; hourly cron reseeds. //! 3. Graceful shutdown on SIGTERM: workers exit after their current task. +use std::net::SocketAddr; use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::time::Duration; +use std::time::{Duration, Instant}; -use prop_grid_rs::{db, fetcher::HrrrClient, pipeline}; +use prop_grid_rs::{db, fetcher::HrrrClient, metrics, pipeline}; use tokio::task::JoinSet; use tracing::{error, info, warn}; use tracing_subscriber::EnvFilter; @@ -55,10 +56,21 @@ async fn main() -> Result<(), Box> { let shutdown = Arc::new(AtomicBool::new(false)); spawn_signal_handler(shutdown.clone()); + // Metrics endpoint binds on a separate port so scrape traffic + // never interferes with the worker's NFS writes or Postgres calls. + let metrics_addr: SocketAddr = std::env::var("METRICS_ADDR") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or_else(|| "0.0.0.0:9100".parse().expect("valid default addr")); + tokio::spawn(async move { + metrics::serve(metrics_addr).await; + }); + info!( scores_dir = %scores_dir.display(), parallelism, pg_conns = max_conn, + metrics_addr = %metrics_addr, "prop-grid-rs started" ); @@ -90,6 +102,8 @@ async fn worker_loop( match db::claim_next(&pool).await { Ok(Some(task)) => { let task_id = task.id; + let _guard = metrics::InFlightGuard::new(); + let started = Instant::now(); match pipeline::run_chain_step( &client, &scores_dir, @@ -101,6 +115,8 @@ async fn worker_loop( .await { Ok(stats) => { + let elapsed = started.elapsed(); + metrics::record_chain_step(elapsed, true); info!( worker_id, task_id = %task.id, @@ -108,6 +124,7 @@ async fn worker_loop( forecast_hour = task.forecast_hour, files = stats.score_files_written, points = stats.point_count, + duration_s = elapsed.as_secs_f64(), "chain step done" ); if let Err(e) = db::complete(&pool, &task).await { @@ -115,6 +132,7 @@ async fn worker_loop( } } Err(e) => { + metrics::record_chain_step(started.elapsed(), false); warn!(worker_id, %task_id, error = %e, "chain step failed"); if let Err(dbe) = db::fail(&pool, task_id, &e.to_string()).await { error!(worker_id, %task_id, error = %dbe, "failed to mark task failed"); diff --git a/rust/prop_grid_rs/src/db.rs b/rust/prop_grid_rs/src/db.rs index 04614389..9d6ef4b4 100644 --- a/rust/prop_grid_rs/src/db.rs +++ b/rust/prop_grid_rs/src/db.rs @@ -74,12 +74,18 @@ pub async fn claim_next(pool: &PgPool) -> Result, DbError> { // keeping them out of the forecast lane means a half-deployed cluster // (Elixir still owns f00, Rust pulls forecasts) doesn't deadlock on // an unfulfillable analysis row. + // + // Order: newest run first, then smallest forecast_hour first. Users + // loading /map always want the most recent cycle's near-hours before + // anything else. A stale run whose tasks never drained (broken HRRR + // publish, etc.) shouldn't block the current hour — the hourly cron + // reseeds the new run and the prune cron clears rotted rows. let row = sqlx::query( r#" SELECT id, run_time, forecast_hour, valid_time, attempt, kind FROM grid_tasks WHERE status = 'queued' AND forecast_hour > 0 AND kind = 'forecast' - ORDER BY run_time ASC, forecast_hour ASC + ORDER BY run_time DESC, forecast_hour ASC FOR UPDATE SKIP LOCKED LIMIT 1 "#, diff --git a/rust/prop_grid_rs/src/lib.rs b/rust/prop_grid_rs/src/lib.rs index 30d35bc5..05cd6cf8 100644 --- a/rust/prop_grid_rs/src/lib.rs +++ b/rust/prop_grid_rs/src/lib.rs @@ -23,6 +23,7 @@ pub mod db; pub mod decoder; pub mod fetcher; pub mod grid; +pub mod metrics; pub mod pipeline; pub mod region; pub mod scores_file; diff --git a/rust/prop_grid_rs/src/metrics.rs b/rust/prop_grid_rs/src/metrics.rs new file mode 100644 index 00000000..ea69f5b9 --- /dev/null +++ b/rust/prop_grid_rs/src/metrics.rs @@ -0,0 +1,146 @@ +//! Prometheus metrics + `/metrics` HTTP endpoint. +//! +//! Scraped by the existing Prometheus server at 10.0.15.25. Exposed on +//! `METRICS_ADDR` (default `0.0.0.0:9100`). + +use std::sync::LazyLock; +use std::time::Duration; + +use axum::{ + extract::State, + http::StatusCode, + response::IntoResponse, + routing::get, + Router, +}; +use prometheus::{ + register_histogram, register_histogram_vec, register_int_counter_vec, register_int_gauge, + Encoder, Histogram, HistogramVec, IntCounterVec, IntGauge, TextEncoder, +}; +use tracing::{info, warn}; + +/// End-to-end duration of one forecast-hour chain step (fetch → decode +/// → score → write). Labelled by outcome so p99 of successful steps can +/// be tracked independently from failure tails. +pub static CHAIN_STEP_DURATION: LazyLock = LazyLock::new(|| { + register_histogram_vec!( + "prop_grid_rs_chain_step_duration_seconds", + "Wall time per grid_tasks chain step", + &["outcome"], + vec![5.0, 10.0, 20.0, 30.0, 45.0, 60.0, 90.0, 120.0, 180.0, 300.0, 600.0] + ) + .expect("register histogram") +}); + +/// Count of chain steps by outcome. Makes the success/failure ratio +/// visible without having to derive it from the histogram. +pub static CHAIN_STEPS_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "prop_grid_rs_chain_steps_total", + "Total grid_tasks chain steps processed", + &["outcome"] + ) + .expect("register counter") +}); + +/// Tasks currently in flight. A gauge so both "idle pod" (0) and +/// "fully saturated" (== PROP_GRID_RS_PARALLELISM) are distinguishable. +pub static TASKS_IN_FLIGHT: LazyLock = LazyLock::new(|| { + register_int_gauge!( + "prop_grid_rs_tasks_in_flight", + "Number of grid_tasks rows currently being processed" + ) + .expect("register gauge") +}); + +/// Decode-only duration (the `spawn_blocking(wgrib2)` portion). Helps +/// answer "is wgrib2 fork overhead dominant?" without instrumenting the +/// full step. +pub static DECODE_DURATION: LazyLock = LazyLock::new(|| { + register_histogram!( + "prop_grid_rs_decode_duration_seconds", + "wgrib2 subprocess duration per product (surface or pressure)", + vec![0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 60.0] + ) + .expect("register histogram") +}); + +/// Install the lazy statics so the `/metrics` endpoint lists every +/// series from the first scrape, even before any chain step runs. +pub fn init() { + LazyLock::force(&CHAIN_STEP_DURATION); + LazyLock::force(&CHAIN_STEPS_TOTAL); + LazyLock::force(&TASKS_IN_FLIGHT); + LazyLock::force(&DECODE_DURATION); +} + +/// RAII helper: increment on construction, decrement on drop. Guarantees +/// the in-flight gauge returns to 0 even if the worker panics. +pub struct InFlightGuard; + +impl InFlightGuard { + pub fn new() -> Self { + TASKS_IN_FLIGHT.inc(); + Self + } +} + +impl Drop for InFlightGuard { + fn drop(&mut self) { + TASKS_IN_FLIGHT.dec(); + } +} + +impl Default for InFlightGuard { + fn default() -> Self { + Self::new() + } +} + +pub fn record_chain_step(duration: Duration, success: bool) { + let outcome = if success { "success" } else { "failure" }; + CHAIN_STEP_DURATION + .with_label_values(&[outcome]) + .observe(duration.as_secs_f64()); + CHAIN_STEPS_TOTAL.with_label_values(&[outcome]).inc(); +} + +async fn metrics_handler(State(()): State<()>) -> impl IntoResponse { + let mut buf = Vec::new(); + let encoder = TextEncoder::new(); + let metric_families = prometheus::gather(); + match encoder.encode(&metric_families, &mut buf) { + Ok(()) => ( + StatusCode::OK, + [("content-type", TextEncoder::new().format_type().to_string())], + buf, + ), + Err(e) => { + warn!(error = %e, "metrics encode failed"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + [("content-type", "text/plain".to_string())], + Vec::new(), + ) + } + } +} + +pub async fn serve(addr: std::net::SocketAddr) { + init(); + let app = Router::new() + .route("/metrics", get(metrics_handler)) + .route("/health", get(|| async { "ok" })) + .with_state(()); + match tokio::net::TcpListener::bind(addr).await { + Ok(listener) => { + info!(%addr, "metrics server listening"); + if let Err(e) = axum::serve(listener, app).await { + warn!(error = %e, "metrics server exited"); + } + } + Err(e) => { + warn!(error = %e, %addr, "failed to bind metrics listener; continuing without /metrics"); + } + } +} diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index de4aa05c..d7345bd2 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -7,6 +7,7 @@ //! boost, ProfilesFile write) — Elixir retains f00. use std::path::Path; +use std::sync::Arc; use chrono::{DateTime, Datelike, Timelike, Utc}; @@ -121,28 +122,47 @@ pub async fn run_chain_step( }) .collect(); - // Step 2: for each band, score → write → drop. One band's score - // vector (~92k × 20 B ≈ 1.8 MB) lives at a time instead of holding - // all 23 bands at once. The write runs on the blocking pool so the - // reactor stays responsive. + // Step 2: score and write every band in parallel. + // + // Scoring: rayon's thread pool over the 23 bands. Each band is + // independent pure math across the ~92k-cell `prepared` slice; + // sharing is read-only so no locking. This cuts per-step scoring + // wall from ~20s serial to ~5-8s on a 4-core box. + // + // Writing: each band file is an independent NFS fsync. Launching + // them as parallel `spawn_blocking` tasks lets them overlap on the + // blocking pool instead of serializing one-at-a-time on the hot + // path. Memory stays bounded — each Vec is ~1.8 MB so + // 23 concurrent ≈ ~40 MB peak. let bands = band_config::all_bands(); let scores_dir = scores_dir.to_path_buf(); - let mut files_written = 0; - for band in bands { - let mut scores: Vec = Vec::with_capacity(prepared.len()); - for (lat, lon, conditions, invariants) in &prepared { - let r = scorer::composite_score_with(conditions, band, Some(*invariants)); - scores.push(ScorePoint { lat: *lat, lon: *lon, score: r.score }); - } + let prepared = Arc::new(prepared); + let scored: Vec<(u32, Vec)> = { + use rayon::prelude::*; + bands + .par_iter() + .map(|band| { + let mut scores: Vec = Vec::with_capacity(prepared.len()); + for (lat, lon, conditions, invariants) in prepared.iter() { + let r = scorer::composite_score_with(conditions, band, Some(*invariants)); + scores.push(ScorePoint { lat: *lat, lon: *lon, score: r.score }); + } + (band.freq_mhz, scores) + }) + .collect() + }; + + let write_futs = scored.into_iter().map(|(band_mhz, scores)| { let dir = scores_dir.clone(); - let band_mhz = band.freq_mhz; tokio::task::spawn_blocking(move || { scores_file::write_atomic(&dir, band_mhz, valid_time, &scores) }) - .await - .expect("blocking join")?; - files_written += 1; + }); + let results = futures::future::try_join_all(write_futs).await.expect("blocking join"); + for r in results { + r?; } + let files_written = bands.len() as u32; Ok(ChainStepStats { score_files_written: files_written,