From 1353745f582d847aaf0227531f57bf3d30a09747 Mon Sep 17 00:00:00 2001 From: anonsaber Date: Wed, 30 Sep 2026 09:13:38 +0800 Subject: [PATCH] fix: let the Redis connection reconnect instead of dying with the process RedisState built one MultiplexedConnection at startup and kept it for the life of the process; every call cloned the handle, which clones the handle and not the socket. When the underlying TCP connection dropped -- an Upstash idle timeout, a network blip, an idle-connection cutoff -- the handle was permanently dead. The process kept running, every Redis read failed, and only a restart recovered it. That is what had to be restarted. The crate says so in its own docs: redis 0.27.6 lib.rs:71 recommends ConnectionManager "for automatic reconnections" behind the connection-manager feature, which we had never enabled. RedisState now holds a ConnectionManager built from the Client. Both types derive Clone and implement ConnectionLike, so the five query_async call sites are untouched; the three helper signatures that took a cloned connection by value or by mutable reference change type only. MultiplexedConnection no longer appears anywhere in src/. Start-up semantics are unchanged: ConnectionManager::new awaits new_connection before returning, so an unreachable Redis still fails closed at boot rather than succeeding and erroring on first use. Reconnect is gated on is_io_error(), which is exactly kind() == IoError. A dropped socket is an IoError, so it reconnects with exponential backoff and jitter (base 2, 100ms factor, 6 retries); an authentication failure is AuthenticationFailed and is deliberately not retried, because that error does not fix itself. No Drop or cancel is needed: the live connection sits behind Arc>, the Client is retained for rebuilding, and no detached control task is spawned, so nothing leaks when RedisState is dropped and the existing #[derive(Clone)] still holds. /debug/redis collapsed every failure into the single string "redis_probe_failed", which is why the real error was invisible and the outage took a session to trace. StateError::Redis carries the underlying RedisError, so the class is readable here: authentication_failed, io_error, busy_loading, client_error, response_error, plus state_lock_poisoned and encryption_failed. ErrorKind is #[non_exhaustive], so the match keeps a fallback arm for future variants. No NotReachable variant exists in redis 0.27.6 -- a network outage surfaces there as IoError or ClientError -- so the classifier does not mention one. Four new transitive dependencies come in with the feature: backon 1.6.0, fastrand 2.4.1, futures 0.3.32, futures-executor 0.3.32. Not covered: there is no test for the automatic reconnect path, which would require tearing down a real TCP connection mid-flight. /debug/redis still probes through is_enabled() rather than issuing a bare PING. Gates: rust fmt clean, clippy --all-targets -D warnings clean, cargo test 87 passed 2 ignored, unchanged from the previous baseline. --- Cargo.lock | 45 +++++++++++++++++++++++++++++++++++++++++++++ Cargo.toml | 2 +- src/debug.rs | 25 ++++++++++++++++++++++++- src/state.rs | 10 +++++----- 4 files changed, 75 insertions(+), 7 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ef90320..33c46db 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -163,6 +163,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "backon" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef" +dependencies = [ + "fastrand", +] + [[package]] name = "base64" version = "0.22.1" @@ -329,6 +338,12 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "fastrand" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" + [[package]] name = "find-msvc-tools" version = "0.1.13" @@ -356,6 +371,21 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -363,6 +393,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -371,6 +402,17 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + [[package]] name = "futures-io" version = "0.3.34" @@ -406,6 +448,7 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", "futures-io", "futures-macro", @@ -1173,8 +1216,10 @@ checksum = "09d8f99a4090c89cc489a94833c901ead69bfbf3877b4867d5482e321ee875bc" dependencies = [ "arc-swap", "async-trait", + "backon", "bytes", "combine", + "futures", "futures-util", "itertools", "itoa", diff --git a/Cargo.toml b/Cargo.toml index b54bfa1..2579158 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,7 @@ axum = "0.8" chrono = "0.4" # Read-only HTTP JMAP only; keep WebSocket/EventSource out of the dependency graph. jmap-client = { version = "=0.4.2", default-features = false, features = ["async", "rustls"] } -redis = { version = "0.27", default-features = false, features = ["aio", "tokio-comp", "streams", "script", "tokio-rustls-comp"] } +redis = { version = "0.27", default-features = false, features = ["aio", "tokio-comp", "streams", "script", "tokio-rustls-comp", "connection-manager"] } reqwest = { version = "0.13", default-features = false, features = ["json", "rustls"] } ring = "0.17.14" rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } diff --git a/src/debug.rs b/src/debug.rs index e1e40ea..f5be33e 100644 --- a/src/debug.rs +++ b/src/debug.rs @@ -29,6 +29,7 @@ use crate::notify::{ business_client, chat_allowlist_snapshot, error_response, probe_jmap_session, probe_telegram_get_me, refresh_business_config, worker_authorized, AppState, }; +use crate::state::StateError; const DEFAULT_DEBUG_MESSAGE: &str = "messageweave debug: outbound test OK"; const MAX_DEBUG_MESSAGE_CHARS: usize = 1024; @@ -158,11 +159,33 @@ async fn debug_redis(State(app): State, headers: HeaderMap) -> Respons } let body = match app.state.is_enabled().await { Ok(enabled) => serde_json::json!({ "reachable": true, "global_enabled": enabled }), - Err(_) => serde_json::json!({ "reachable": false, "detail": "redis_probe_failed" }), + Err(err) => serde_json::json!({ "reachable": false, "detail": redis_failure_detail(&err) }), }; Json(body).into_response() } +/// Classify a Redis probe failure into a stable reason. +/// +/// `StateError::Redis` carries the underlying `redis::RedisError`, so the class +/// can be read here without losing information. Previously every failure was +/// collapsed to the single string `"redis_probe_failed"`, which made an +/// authentication mistake, a dead TCP connection, and a network outage +/// indistinguishable from `/debug/redis`. `ErrorKind` is `#[non_exhaustive]`, +/// so the fallback keeps this compiling across redis minor releases. +fn redis_failure_detail(err: &StateError) -> &'static str { + match err { + StateError::Redis(redis_error) => match redis_error.kind() { + redis::ErrorKind::AuthenticationFailed => "authentication_failed", + redis::ErrorKind::IoError => "io_error", + redis::ErrorKind::BusyLoadingError => "busy_loading", + redis::ErrorKind::ClientError => "client_error", + _ => "response_error", + }, + StateError::Poisoned => "state_lock_poisoned", + StateError::Encryption => "encryption_failed", + } +} + /// JMAP reachability. Reuses `probe_jmap_session` so `/ready` and debug can never drift. async fn debug_jmap(State(app): State, headers: HeaderMap) -> Response { if !debug_authorized(&app, &headers) { diff --git a/src/state.rs b/src/state.rs index e0a8025..f9abaf7 100644 --- a/src/state.rs +++ b/src/state.rs @@ -239,7 +239,7 @@ pub trait ReliableState: Send + Sync { } pub struct RedisState { - connection: redis::aio::MultiplexedConnection, + connection: redis::aio::ConnectionManager, encryption_key: Option<[u8; 32]>, } @@ -255,7 +255,7 @@ impl RedisState { ) -> Result { let client = redis::Client::open(redis_url)?; Ok(Self { - connection: client.get_multiplexed_async_connection().await?, + connection: redis::aio::ConnectionManager::new(client).await?, encryption_key, }) } @@ -827,7 +827,7 @@ fn stream_id_to_message(item: redis::streams::StreamId) -> Option /// can never be processed, so acknowledging them keeps the consumer group's PEL /// from filling up with unclaimable junk; the failure is logged for operators. async fn ack_malformed( - connection: &mut redis::aio::MultiplexedConnection, + connection: &mut redis::aio::ConnectionManager, stream: &str, group: &str, ids: &[String], @@ -854,7 +854,7 @@ async fn ack_malformed( } async fn set_nx_ex( - mut connection: redis::aio::MultiplexedConnection, + mut connection: redis::aio::ConnectionManager, key: &str, ttl_seconds: u64, ) -> Result { @@ -1371,7 +1371,7 @@ fn hex_decode(value: &str, max_len: usize) -> Result, StateError> { } async fn delete_key( - mut connection: redis::aio::MultiplexedConnection, + mut connection: redis::aio::ConnectionManager, key: &str, ) -> Result<(), StateError> { let _: i64 = redis::cmd("DEL")