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<ArcSwap<SharedRedisFuture>>, 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

@@ -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",

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"] }

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<AppState>, 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<AppState>, headers: HeaderMap) -> Response {
     if !debug_authorized(&app, &headers) {

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<Self, StateError> {
         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<StreamMessage>
 /// 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<bool, StateError> {
@@ -1371,7 +1371,7 @@ fn hex_decode(value: &str, max_len: usize) -> Result<Vec<u8>, StateError> {
 }
 
 async fn delete_key(
-    mut connection: redis::aio::MultiplexedConnection,
+    mut connection: redis::aio::ConnectionManager,
     key: &str,
 ) -> Result<(), StateError> {
     let _: i64 = redis::cmd("DEL")