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")