From 3f3819333cfabfba76d2bec8c8d8a19bb250d1ab Mon Sep 17 00:00:00 2001 From: "Michael H." Date: Mon, 2 Nov 2026 19:47:23 +0100 Subject: [PATCH] fix(labrinth): bound Redis timeouts and add local-only healthz probe Redis commands and connection checkouts had no upper bound, so during a Valkey cluster failover requests to the vanished primary hung. The rate limiter runs on every route, including the one used by Kubernetes probes, which caused readiness failures and liveness kills. Cap every Redis command and connection checkout at 2s, set pool create/recycle timeouts, and set TCP_USER_TIMEOUT on the multiplexed cluster client so dead nodes trigger a slot refresh. Add a /healthz endpoint that touches no backing services, skips rate limiting and only answers requests from loopback, private or link-local peers. --- apps/labrinth/src/routes/index.rs | 41 ++++++++++++++++++++++- apps/labrinth/src/routes/mod.rs | 1 + apps/labrinth/src/util/ratelimit.rs | 4 +++ packages/xredis/src/connection.rs | 52 ++++++++++++++++++++++++++--- 4 files changed, 92 insertions(+), 6 deletions(-) diff --git a/apps/labrinth/src/routes/index.rs b/apps/labrinth/src/routes/index.rs index 343f28aaa45..ed4fa8d4e5e 100644 --- a/apps/labrinth/src/routes/index.rs +++ b/apps/labrinth/src/routes/index.rs @@ -1,4 +1,6 @@ -use actix_web::{HttpResponse, get}; +use std::net::IpAddr; + +use actix_web::{HttpRequest, HttpResponse, get}; use serde_json::json; fn build_info() -> serde_json::Value { @@ -25,3 +27,40 @@ pub async fn index_get() -> HttpResponse { pub async fn build_get() -> HttpResponse { HttpResponse::Ok().json(build_info()) } + +/// Liveness and readiness probe. Deliberately touches no backing services so a +/// Redis or Postgres failover cannot get healthy pods killed. +#[get("/healthz")] +pub async fn healthz_get(req: HttpRequest) -> HttpResponse { + if !is_local_request(&req) { + return HttpResponse::NotFound().finish(); + } + + HttpResponse::Ok().finish() +} + +/// Uses the TCP peer, never client-supplied headers. Requests carrying +/// forwarding headers came through a proxy, so their real origin is unknown. +fn is_local_request(req: &HttpRequest) -> bool { + let headers = req.headers(); + if headers.contains_key("X-Forwarded-For") + || headers.contains_key("CF-Connecting-IP") + { + return false; + } + + let Some(peer) = req.peer_addr() else { + return false; + }; + + match peer.ip().to_canonical() { + IpAddr::V4(ip) => { + ip.is_loopback() || ip.is_private() || ip.is_link_local() + } + IpAddr::V6(ip) => { + ip.is_loopback() + || ip.is_unique_local() + || ip.is_unicast_link_local() + } + } +} diff --git a/apps/labrinth/src/routes/mod.rs b/apps/labrinth/src/routes/mod.rs index 41e8b5b3d85..b0aeb43a515 100644 --- a/apps/labrinth/src/routes/mod.rs +++ b/apps/labrinth/src/routes/mod.rs @@ -534,6 +534,7 @@ pub fn root_config(cfg: &mut web::ServiceConfig) { .wrap(default_cors()) .service(index::index_get) .service(index::build_get) + .service(index::healthz_get) .service(Files::new("/", "assets/")), ); } diff --git a/apps/labrinth/src/util/ratelimit.rs b/apps/labrinth/src/util/ratelimit.rs index d9fa0b37bd1..a1d5cdd4e39 100644 --- a/apps/labrinth/src/util/ratelimit.rs +++ b/apps/labrinth/src/util/ratelimit.rs @@ -221,6 +221,10 @@ pub async fn rate_limit_middleware( .expect("Rate limiter not configured properly") .clone(); + if req.path() == "/healthz" { + return Ok(next.call(req).await?.map_into_left_body()); + } + if let Some(key) = req.headers().get("x-ratelimit-key") && key.to_str().ok() == Some(&ENV.RATE_LIMIT_IGNORE_KEY) { diff --git a/packages/xredis/src/connection.rs b/packages/xredis/src/connection.rs index 3632290d19e..a02c9f2991f 100644 --- a/packages/xredis/src/connection.rs +++ b/packages/xredis/src/connection.rs @@ -20,6 +20,13 @@ use super::metrics::{ const POOL_RETAIN_INTERVAL: Duration = Duration::from_secs(30); const MAX_IDLE_CONNECTION_AGE: Duration = Duration::from_secs(5 * 60); const MAX_STANDALONE_CONNECTION_AGE: Duration = Duration::from_secs(120); +const COMMAND_TIMEOUT: Duration = Duration::from_secs(2); +const CONNECT_TIMEOUT: Duration = Duration::from_secs(2); +/// Closes sockets whose sent data stays unacknowledged, so a primary that +/// vanished during failover surfaces as a dropped connection and the cluster +/// client refreshes its slot map instead of waiting on the kernel for minutes. +#[cfg(target_os = "linux")] +const TCP_USER_TIMEOUT: Duration = Duration::from_secs(10); /// The primary backing "connection provider" for a Redis backend implementation. #[derive(Clone)] @@ -77,6 +84,8 @@ impl RedisBackend { let pool = deadpool_redis::Pool::builder(manager) .max_size(pool_size.max()) .wait_timeout(Some(Duration::from_millis(config.wait_timeout_ms()))) + .create_timeout(Some(CONNECT_TIMEOUT)) + .recycle_timeout(Some(CONNECT_TIMEOUT)) .runtime(deadpool_redis::Runtime::Tokio1) .build() .wrap_err("building standalone Redis pool")?; @@ -101,6 +110,8 @@ impl RedisBackend { let pool = deadpool_redis::cluster::Pool::builder(manager) .max_size(pool_size.max()) .wait_timeout(Some(Duration::from_millis(config.wait_timeout_ms()))) + .create_timeout(Some(CONNECT_TIMEOUT)) + .recycle_timeout(Some(CONNECT_TIMEOUT)) .runtime(deadpool_redis::Runtime::Tokio1) .build() .wrap_err("building clustered Redis pool")?; @@ -122,7 +133,16 @@ impl RedisBackend { async fn cluster_multiplexed(config: &RedisConfig) -> Result { let mut builder = redis::cluster::ClusterClientBuilder::new( config.seed_urls().iter().map(String::as_str), - ); + ) + .connection_timeout(CONNECT_TIMEOUT); + + #[cfg(target_os = "linux")] + { + builder = builder.tcp_settings( + redis::io::tcp::TcpSettings::default() + .set_user_timeout(TCP_USER_TIMEOUT), + ); + } match config.read_replica_strategy() { ReadReplicaStrategy::Primary => {} @@ -146,7 +166,15 @@ impl RedisBackend { Ok(Self::ClusterMultiplexed(connection)) } + /// Bounds pool wait, creation and recycling together, since deadpool + /// applies its own timeouts to each phase separately. pub(crate) async fn connect(&self) -> Result { + tokio::time::timeout(CONNECT_TIMEOUT, self.checkout()) + .await + .wrap_err("timed out fetching Redis connection")? + } + + async fn checkout(&self) -> Result { let inner = match self { Self::StandalonePooled(pool) => { RedisConnectionInner::StandalonePooled( @@ -240,7 +268,7 @@ impl ConnectionLike for RedisConnection { &'a mut self, cmd: &'a redis::Cmd, ) -> redis::RedisFuture<'a, redis::Value> { - self.inner.req_packed_command(cmd) + with_command_timeout(self.inner.req_packed_command(cmd)) } fn req_packed_commands<'a>( @@ -249,7 +277,7 @@ impl ConnectionLike for RedisConnection { offset: usize, count: usize, ) -> redis::RedisFuture<'a, Vec> { - self.inner.req_packed_commands(cmd, offset, count) + with_command_timeout(self.inner.req_packed_commands(cmd, offset, count)) } fn get_db(&self) -> i64 { @@ -263,7 +291,7 @@ impl RoutableConnection for RedisConnection { command: redis::Cmd, routing: RoutingInfo, ) -> redis::RedisFuture<'a, redis::Value> { - Box::pin(async move { + with_command_timeout(Box::pin(async move { match &mut self.inner { RedisConnectionInner::StandalonePooled(connection) => { command.query_async(connection).await @@ -275,10 +303,24 @@ impl RoutableConnection for RedisConnection { connection.route_command(command, routing).await } } - }) + })) } } +/// Bounds a command so a node that stops responding mid-failover fails the +/// request instead of hanging it. Callers already treat Redis as fail-open. +fn with_command_timeout<'a, T: Send + 'a>( + command: redis::RedisFuture<'a, T>, +) -> redis::RedisFuture<'a, T> { + Box::pin(async move { + tokio::time::timeout(COMMAND_TIMEOUT, command) + .await + .unwrap_or_else(|_| { + Err(std::io::Error::from(std::io::ErrorKind::TimedOut).into()) + }) + }) +} + async fn warm_standalone_pool( pool: &deadpool_redis::Pool, min: usize,