From 119238cad630efd71af1af09f6a3640ba8237d75 Mon Sep 17 00:00:00 2001 From: mike-programs <123466677+mike-programs@users.noreply.github.com> Date: Tue, 20 Oct 2026 07:11:46 +0000 Subject: [PATCH] fix(api): harden shutdown, email worker restarts, CORS parsing and body-size middleware Closes: #1717 Closes: #1718 Closes: #1719 Closes: #1721 --- services/api/src/cors.rs | 28 ++++++++++++-- services/api/src/main.rs | 67 +++++++++++++++------------------- services/api/src/validation.rs | 16 ++++++-- 3 files changed, 67 insertions(+), 44 deletions(-) diff --git a/services/api/src/cors.rs b/services/api/src/cors.rs index 1aa891b0..e38df577 100644 --- a/services/api/src/cors.rs +++ b/services/api/src/cors.rs @@ -37,19 +37,37 @@ pub fn build_cors_layer(cfg: &CorsConfig) -> CorsLayer { let origins: Vec = cfg .allowed_origins .iter() - .filter_map(|o| o.parse::().ok()) + .filter_map(|o| match o.parse::() { + Ok(v) => Some(v), + Err(_) => { + tracing::warn!(value = %o, "CORS: ignoring unparsable allowed origin"); + None + } + }) .collect(); let methods: Vec = cfg .allowed_methods .iter() - .filter_map(|m| m.parse::().ok()) + .filter_map(|m| match m.parse::() { + Ok(v) => Some(v), + Err(_) => { + tracing::warn!(value = %m, "CORS: ignoring unparsable allowed method"); + None + } + }) .collect(); let headers: Vec = cfg .allowed_headers .iter() - .filter_map(|h| h.parse::().ok()) + .filter_map(|h| match h.parse::() { + Ok(v) => Some(v), + Err(_) => { + tracing::warn!(value = %h, "CORS: ignoring unparsable allowed header"); + None + } + }) .collect(); let layer = CorsLayer::new() @@ -59,6 +77,10 @@ pub fn build_cors_layer(cfg: &CorsConfig) -> CorsLayer { .max_age(Duration::from_secs(cfg.max_age_secs)); if cfg.allow_credentials { + if cfg.allowed_origins.iter().any(|o| o.trim() == "*") { + tracing::warn!("CORS: wildcard origin is not allowed with credentials — credentials disabled"); + return layer; + } layer.allow_credentials(true) } else { layer diff --git a/services/api/src/main.rs b/services/api/src/main.rs index b6af3562..7193a9a2 100644 --- a/services/api/src/main.rs +++ b/services/api/src/main.rs @@ -233,11 +233,13 @@ async fn main() -> anyhow::Result<()> { // ── Email queue worker (monitored restart loop) ──────────────────────────── // Wraps the worker in a panic-catching JoinHandle. If the task panics or // exits unexpectedly it is restarted with exponential backoff (1s, 2s, 4s, - // 8s, 16s). After MAX_EMAIL_WORKER_RESTARTS consecutive crashes without a + // 8s, 16s). The counter resets once a worker has run healthily for + // EMAIL_WORKER_HEALTHY_RESET. After MAX_EMAIL_WORKER_RESTARTS consecutive crashes without a // clean recovery the loop logs FATAL and increments worker_crash_total so // the Prometheus alert fires. { const MAX_EMAIL_WORKER_RESTARTS: u32 = 5; + const EMAIL_WORKER_HEALTHY_RESET: Duration = Duration::from_secs(300); let queue_worker = email_queue.clone(); let service_worker = email_service.clone(); @@ -250,6 +252,7 @@ async fn main() -> anyhow::Result<()> { tokio::spawn(async move { let mut restarts: u32 = 0; loop { + let started = std::time::Instant::now(); let q = queue_worker.clone(); let s = service_worker.clone(); let token = email_token.clone(); @@ -267,6 +270,10 @@ async fn main() -> anyhow::Result<()> { break; } Err(e) => { + // A worker that stayed healthy for a while is not part of a crash loop. + if started.elapsed() >= EMAIL_WORKER_HEALTHY_RESET { + restarts = 0; + } restarts += 1; crash_counter.with_label_values(&["email_queue_worker"]).inc(); @@ -361,7 +368,6 @@ async fn main() -> anyhow::Result<()> { .layer(TraceLayer::new_for_http()) .layer(middleware::from_fn_with_state(state.clone(), idempotency::idempotency_middleware)) .layer(middleware::from_fn(validation::content_type_validation_middleware)) - .layer(middleware::from_fn(validation::request_size_validation_middleware)) // CSRF defense-in-depth: validate Origin/Referer on state-changing requests. .layer(middleware::from_fn_with_state(csrf_config, csrf_protection_middleware)) .layer(middleware::from_fn_with_state( @@ -390,7 +396,6 @@ async fn main() -> anyhow::Result<()> { // - audit_logging_middleware: webhook events are tracked via email_events │ let webhook_routes = Router::new() .route("/webhooks/sendgrid", post(handlers::sendgrid_webhook)) - .layer(middleware::from_fn(validation::request_size_validation_middleware)) .layer(middleware::from_fn(security::security_headers_middleware)) .layer(middleware::from_fn_with_state( ( @@ -461,7 +466,6 @@ async fn main() -> anyhow::Result<()> { idempotency::idempotency_middleware, )) .layer(middleware::from_fn(validation::content_type_validation_middleware)) - .layer(middleware::from_fn(validation::request_size_validation_middleware)) .layer(middleware::from_fn_with_state( (ip_whitelist.clone(), security::TrustProxy(config_trust_proxy)), security::ip_whitelist_middleware, @@ -503,43 +507,32 @@ async fn main() -> anyhow::Result<()> { let listener = TcpListener::bind(bind_addr).await?; tracing::info!("API listening on {bind_addr}"); - // Axum's built-in graceful shutdown drains in-flight HTTP requests. - // After the server stops accepting connections we then drain background workers. - let axum_shutdown = { - let coord = coordinator.clone(); - let email_coord = email_coordinator.clone(); - async move { + // Stop accepting connections and drain in-flight requests first; only then + // drain background workers so no request can enqueue into a stopped worker. + axum::serve(listener, app) + .with_graceful_shutdown(async { wait_for_signal().await; tracing::info!("Shutdown signal received — stopping HTTP server"); + }) + .await?; - // Drain email queue first with its own timeout to avoid losing in-flight emails. - let email_drain = shutdown::email_queue_drain_timeout(); - tracing::info!( - timeout_secs = email_drain.as_secs(), - "Draining email queue worker" - ); - match email_coord.shutdown(email_drain).await { - Ok(_) => tracing::info!("Email queue worker drained cleanly"), - Err(e) => tracing::warn!("Email queue drain timeout — in-flight email may be lost: {e}"), - } - - // Then drain remaining background workers (blockchain sync, tx-monitor). - let timeout_dur = shutdown_timeout(); - tracing::info!( - timeout_secs = timeout_dur.as_secs(), - "Waiting for remaining background workers to drain" - ); - match coord.shutdown(timeout_dur).await { - Ok(_) => tracing::info!("All background workers stopped cleanly"), - Err(e) => tracing::warn!("Shutdown timeout — forcing exit: {e}"), - } - tracing_config::shutdown_tracing(); - } - }; + let email_drain = shutdown::email_queue_drain_timeout(); + tracing::info!(timeout_secs = email_drain.as_secs(), "Draining email queue worker"); + match email_coordinator.shutdown(email_drain).await { + Ok(_) => tracing::info!("Email queue worker drained cleanly"), + Err(e) => tracing::warn!("Email queue drain timeout — in-flight email may be lost: {e}"), + } - axum::serve(listener, app) - .with_graceful_shutdown(axum_shutdown) - .await?; + let timeout_dur = shutdown_timeout(); + tracing::info!( + timeout_secs = timeout_dur.as_secs(), + "Waiting for remaining background workers to drain" + ); + match coordinator.shutdown(timeout_dur).await { + Ok(_) => tracing::info!("All background workers stopped cleanly"), + Err(e) => tracing::warn!("Shutdown timeout — forcing exit: {e}"), + } + tracing_config::shutdown_tracing(); Ok(()) } diff --git a/services/api/src/validation.rs b/services/api/src/validation.rs index 3a9e3229..26710119 100644 --- a/services/api/src/validation.rs +++ b/services/api/src/validation.rs @@ -29,10 +29,17 @@ pub fn parse_request_body_max_bytes(val: Option<&str>) -> usize { .unwrap_or(DEFAULT_REQUEST_BODY_MAX_BYTES) } +/// Read once at first use and cached for the process lifetime. fn body_limit() -> usize { - parse_request_body_max_bytes(std::env::var("REQUEST_BODY_MAX_BYTES").ok().as_deref()) + static LIMIT: std::sync::OnceLock = std::sync::OnceLock::new(); + *LIMIT.get_or_init(|| { + parse_request_body_max_bytes(std::env::var("REQUEST_BODY_MAX_BYTES").ok().as_deref()) + }) } +/// Maximum time allowed to read a request body. +const BODY_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + #[derive(Serialize)] struct PayloadTooLargeError { error: &'static str, @@ -64,9 +71,10 @@ pub async fn request_size_validation_middleware( // Slow path: buffer stream up to limit+1 bytes. // axum::body::to_bytes returns Err when body exceeds the cap — treat that as 413. let (parts, body) = req.into_parts(); - let bytes = match axum::body::to_bytes(body, limit + 1).await { - Ok(b) => b, - Err(_) => return payload_too_large(limit), + let bytes = match tokio::time::timeout(BODY_READ_TIMEOUT, axum::body::to_bytes(body, limit + 1)).await { + Ok(Ok(b)) => b, + Ok(Err(_)) => return payload_too_large(limit), + Err(_) => return StatusCode::REQUEST_TIMEOUT.into_response(), }; if bytes.len() > limit { return payload_too_large(limit);