Update redis health check

Due to recent changes in the redis library, a PING command is now
routed to all main nodes by default, rather than a single random node,
and all must succeed. This makes connections particularly fragile if
a single node is having transient issues. So, in an effort to
maximize uptime, let's require a successful response from only one
node.

server/svix-server/Cargo.toml

@@ -72,7 +72,7 @@ urlencoding = "2.1.2"
 form_urlencoded = "1.1.0"
 lapin = "2.1.1"
 sentry = { version = "0.32.2", features = ["tracing"] }
-omniqueue = { version = "0.2.0", default-features = false, features = ["in_memory", "rabbitmq-with-message-ids", "redis_cluster"] }
+omniqueue = { git = "https://github.com/svix/omniqueue-rs", rev = "5ae22000e2ea214ba707cac81657f098e5785a76", default-features = false, features = ["in_memory", "rabbitmq-with-message-ids", "redis_cluster"] }
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemallocator = { version = "0.5", optional = true }

server/svix-server/src/queue/redis.rs

@@ -29,8 +29,7 @@
 
 use std::{num::NonZeroUsize, sync::Arc, time::Duration};
 
-use bb8_redis::RedisMultiplexedConnectionManager;
-use omniqueue::backends::{redis::RedisClusterConnectionManager, RedisBackend, RedisConfig};
+use omniqueue::backends::{RedisBackend, RedisConfig};
 use redis::{AsyncCommands as _, RedisResult};
 
 use super::{QueueTask, TaskQueueConsumer, TaskQueueProducer};
@@ -213,22 +212,20 @@ async fn new_pair_inner(
 
     match &cfg.queue_type {
         QueueType::RedisCluster => {
-            let (producer, consumer) =
-                RedisBackend::<RedisClusterConnectionManager>::builder(config)
-                    .build_pair()
-                    .await
-                    .expect("Error initializing redis-cluster queue");
+            let (producer, consumer) = RedisBackend::cluster_builder(config)
+                .build_pair()
+                .await
+                .expect("Error initializing redis-cluster queue");
 
             let producer = TaskQueueProducer::new(producer);
             let consumer = TaskQueueConsumer::new(consumer);
             (producer, consumer)
         }
         _ => {
-            let (producer, consumer) =
-                RedisBackend::<RedisMultiplexedConnectionManager>::builder(config)
-                    .build_pair()
-                    .await
-                    .expect("Error initializing redis queue");
+            let (producer, consumer) = RedisBackend::builder(config)
+                .build_pair()
+                .await
+                .expect("Error initializing redis queue");
 
             let producer = TaskQueueProducer::new(producer);
             let consumer = TaskQueueConsumer::new(consumer);

server/svix-server/src/redis/cluster.rs

@@ -1,7 +1,8 @@
 use axum::async_trait;
 use redis::{
     cluster::{ClusterClient, ClusterClientBuilder},
-    ErrorKind, IntoConnectionInfo, RedisError,
+    cluster_routing::{MultipleNodeRoutingInfo, ResponsePolicy, RoutingInfo},
+    ErrorKind, FromRedisValue, IntoConnectionInfo, RedisError,
 };
 
 /// ConnectionManager that implements `bb8::ManageConnection` and supports
@@ -31,7 +32,16 @@ impl bb8::ManageConnection for RedisClusterConnectionManager {
     }
 
     async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
-        let pong: String = redis::cmd("PING").query_async(&mut *conn).await?;
+        let pong = conn
+            .route_command(
+                &redis::cmd("PING"),
+                RoutingInfo::MultiNode((
+                    MultipleNodeRoutingInfo::AllMasters,
+                    Some(ResponsePolicy::OneSucceeded),
+                )),
+            )
+            .await
+            .and_then(|v| String::from_redis_value(&v))?;
         match pong.as_str() {
             "PONG" => Ok(()),
             _ => Err((ErrorKind::ResponseError, "ping request").into()),