Add redisclusterunpooled cache type

Adds downstream changes to support unpooled redis cluster.
Simply put, we don't gain anything by using connection pooling
for clustered redis. The redis cluster client handles establishing
connections, handling errors, etc., internally. As long as we don't
make blocking redis requests, we are fine without connection pooling
in the redis caching layer.

Author: jaymell

server/Cargo.lock

@@ -50,7 +50,7 @@ echo "*********** RUN 3 ***********"
 echo "*********** RUN 4 ***********"
 (
     export SVIX_QUEUE_TYPE="rediscluster"
-    export SVIX_CACHE_TYPE="rediscluster"
+    export SVIX_CACHE_TYPE="redisclusterunpooled"
     export SVIX_REDIS_DSN="redis://localhost:6380"
     ${TEST_COMMAND}
     ${TEST_COMMAND} -- --ignored redis

server/run-tests.sh

@@ -48,8 +48,8 @@ ed25519-compact = "2.1.1"
 chrono = { version="0.4.26", features = ["serde"] }
 reqwest = { version = "0.11.27", features = ["json", "rustls-tls", "hickory-resolver"], default-features = false }
 bb8 = "0.8"
-bb8-redis = "0.14.0"
-redis = { version = "0.24.0", features = ["tokio-comp", "tokio-native-tls-comp", "streams", "cluster-async", "tcp_nodelay"] }
+bb8-redis = "0.15.0"
+redis = { version = "0.25.4", features = ["tokio-comp", "tokio-native-tls-comp", "streams", "cluster-async", "tcp_nodelay"] }
 thiserror = "1.0.30"
 bytes = "1.1.0"
 blake2 = "0.10.4"

server/svix-server/Cargo.toml

@@ -193,7 +193,7 @@ pub struct ConfigurationInner {
 fn validate_config_complete(config: &ConfigurationInner) -> Result<(), ValidationError> {
     match config.cache_type {
         CacheType::None | CacheType::Memory => {}
-        CacheType::Redis | CacheType::RedisCluster => {
+        CacheType::Redis | CacheType::RedisCluster | CacheType::RedisClusterUnpooled => {
             if config.cache_dsn().is_none() {
                 return Err(ValidationError {
                     code: Cow::from("missing field"),
@@ -267,6 +267,9 @@ impl ConfigurationInner {
             CacheType::Memory => CacheBackend::Memory,
             CacheType::Redis => CacheBackend::Redis(self.cache_dsn().expect(err)),
             CacheType::RedisCluster => CacheBackend::RedisCluster(self.cache_dsn().expect(err)),
+            CacheType::RedisClusterUnpooled => {
+                CacheBackend::RedisClusterUnpooled(self.cache_dsn().expect(err))
+            }
         }
     }
 }
@@ -304,6 +307,7 @@ pub enum CacheBackend<'a> {
     Memory,
     Redis(&'a str),
     RedisCluster(&'a str),
+    RedisClusterUnpooled(&'a str),
 }
 
 #[derive(Clone, Debug, Deserialize)]
@@ -336,6 +340,7 @@ pub enum CacheType {
     Memory,
     Redis,
     RedisCluster,
+    RedisClusterUnpooled,
     None,
 }
 

server/svix-server/src/cfg.rs

@@ -115,6 +115,9 @@ mod tests {
     async fn get_pool(redis_dsn: &str, cfg: &crate::cfg::Configuration) -> RedisPool {
         match cfg.cache_type {
             CacheType::RedisCluster => crate::redis::new_redis_pool_clustered(redis_dsn, cfg).await,
+            CacheType::RedisClusterUnpooled => {
+                crate::redis::new_redis_clustered_unpooled(redis_dsn).await
+            }
             CacheType::Redis => crate::redis::new_redis_pool(redis_dsn, cfg).await,
             _ => panic!(
                 "This test should only be run when redis is configured as the cache provider"

server/svix-server/src/core/cache/redis.rs

@@ -117,6 +117,10 @@ pub async fn run_with_prefix(
             let mgr = crate::redis::new_redis_pool_clustered(dsn, &cfg).await;
             cache::redis::new(mgr)
         }
+        CacheBackend::RedisClusterUnpooled(dsn) => {
+            let mgr = crate::redis::new_redis_clustered_unpooled(dsn).await;
+            cache::redis::new(mgr)
+        }
     };
     tracing::debug!("Cache: Started");
 

server/svix-server/src/lib.rs

@@ -369,6 +369,9 @@ pub mod tests {
             CacheType::Redis => {
                 crate::redis::new_redis_pool(cfg.redis_dsn.as_deref().unwrap(), cfg).await
             }
+            CacheType::RedisClusterUnpooled => {
+                crate::redis::new_redis_clustered_unpooled(cfg.redis_dsn.as_deref().unwrap()).await
+            }
             _ => {
                 panic!("This test should only be run when redis is configured as the queue backend")
             }

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

@@ -1,15 +1,20 @@
 mod cluster;
 
+use std::time::Duration;
+
 use bb8::{Pool, RunError};
-use bb8_redis::RedisMultiplexedConnectionManager;
+use bb8_redis::RedisConnectionManager;
 use redis::{FromRedisValue, RedisError, RedisResult};
 
 pub use self::cluster::RedisClusterConnectionManager;
 use crate::cfg::Configuration;
 
+pub const REDIS_CONN_TIMEOUT: Duration = Duration::from_secs(2);
+
 #[derive(Clone, Debug)]
 pub enum RedisPool {
     Clustered(ClusteredRedisPool),
+    ClusteredUnpooled(ClusteredRedisUnpooled),
     NonClustered(NonClusteredRedisPool),
 }
 
@@ -18,6 +23,7 @@ impl RedisPool {
         match self {
             Self::Clustered(pool) => pool.get().await,
             Self::NonClustered(pool) => pool.get().await,
+            Self::ClusteredUnpooled(pool) => pool.get().await,
         }
     }
 }
@@ -36,9 +42,30 @@ impl ClusteredRedisPool {
     }
 }
 
+#[derive(Clone)]
+pub struct ClusteredRedisUnpooled {
+    con: redis::cluster_async::ClusterConnection,
+}
+
+impl ClusteredRedisUnpooled {
+    pub async fn get(&self) -> Result<PooledConnection<'_>, RunError<RedisError>> {
+        Ok(PooledConnection::ClusteredUnpooled(
+            ClusteredUnpooledConnection {
+                con: self.con.clone(),
+            },
+        ))
+    }
+}
+
+impl std::fmt::Debug for ClusteredRedisUnpooled {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ClusteredRedisUnpooled").finish()
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct NonClusteredRedisPool {
-    pool: Pool<RedisMultiplexedConnectionManager>,
+    pool: Pool<RedisConnectionManager>,
 }
 
 impl NonClusteredRedisPool {
@@ -51,6 +78,7 @@ impl NonClusteredRedisPool {
 
 pub enum PooledConnection<'a> {
     Clustered(ClusteredPooledConnection<'a>),
+    ClusteredUnpooled(ClusteredUnpooledConnection),
     NonClustered(NonClusteredPooledConnection<'a>),
 }
 
@@ -75,6 +103,7 @@ impl redis::aio::ConnectionLike for PooledConnection<'_> {
         match self {
             PooledConnection::Clustered(conn) => conn.con.req_packed_command(cmd),
             PooledConnection::NonClustered(conn) => conn.con.req_packed_command(cmd),
+            PooledConnection::ClusteredUnpooled(conn) => conn.con.req_packed_command(cmd),
         }
     }
 
@@ -89,19 +118,23 @@ impl redis::aio::ConnectionLike for PooledConnection<'_> {
             PooledConnection::NonClustered(conn) => {
                 conn.con.req_packed_commands(cmd, offset, count)
             }
+            PooledConnection::ClusteredUnpooled(conn) => {
+                conn.con.req_packed_commands(cmd, offset, count)
+            }
         }
     }
 
     fn get_db(&self) -> i64 {
         match self {
             PooledConnection::Clustered(conn) => conn.con.get_db(),
             PooledConnection::NonClustered(conn) => conn.con.get_db(),
+            PooledConnection::ClusteredUnpooled(conn) => conn.con.get_db(),
         }
     }
 }
 
 pub struct NonClusteredPooledConnection<'a> {
-    con: bb8::PooledConnection<'a, RedisMultiplexedConnectionManager>,
+    con: bb8::PooledConnection<'a, RedisConnectionManager>,
 }
 
 impl<'a> NonClusteredPooledConnection<'a> {
@@ -134,28 +167,73 @@ impl<'a> ClusteredPooledConnection<'a> {
     }
 }
 
-pub async fn new_redis_pool_clustered(redis_dsn: &str, cfg: &Configuration) -> RedisPool {
-    let mgr = RedisClusterConnectionManager::new(redis_dsn)
-        .expect("Error initializing redis cluster client");
-    let pool = bb8::Pool::builder()
-        .max_size(cfg.redis_pool_max_size.into())
-        .build(mgr)
+pub struct ClusteredUnpooledConnection {
+    con: redis::cluster_async::ClusterConnection,
+}
+
+impl ClusteredUnpooledConnection {
+    pub async fn query_async<T: FromRedisValue>(&mut self, cmd: redis::Cmd) -> RedisResult<T> {
+        cmd.query_async(&mut self.con).await
+    }
+
+    pub async fn query_async_pipeline<T: FromRedisValue>(
+        &mut self,
+        pipe: redis::Pipeline,
+    ) -> RedisResult<T> {
+        pipe.query_async(&mut self.con).await
+    }
+}
+
+async fn new_redis_pool_helper(
+    redis_dsn: &str,
+    clustered: bool,
+    max_connections: u16,
+) -> RedisPool {
+    if clustered {
+        let mgr = RedisClusterConnectionManager::new(redis_dsn)
+            .expect("Error initializing redis cluster client");
+        let pool = bb8::Pool::builder()
+            .max_size(max_connections.into())
+            .build(mgr)
+            .await
+            .expect("Error initializing redis cluster connection pool");
+        let pool = ClusteredRedisPool { pool };
+        RedisPool::Clustered(pool)
+    } else {
+        let mgr = RedisConnectionManager::new(redis_dsn).expect("Error initializing redis client");
+        let pool = bb8::Pool::builder()
+            .max_size(max_connections.into())
+            .build(mgr)
+            .await
+            .expect("Error initializing redis connection pool");
+        let pool = NonClusteredRedisPool { pool };
+        RedisPool::NonClustered(pool)
+    }
+}
+
+async fn new_redis_unpooled_helper(redis_dsn: &str) -> RedisPool {
+    let cli = redis::cluster::ClusterClient::builder(vec![redis_dsn])
+        .retries(1)
+        .connection_timeout(REDIS_CONN_TIMEOUT)
+        .build()
+        .expect("Error initializing redis-unpooled client");
+    let con = cli
+        .get_async_connection()
         .await
-        .expect("Error initializing redis cluster connection pool");
-    let pool = ClusteredRedisPool { pool };
-    RedisPool::Clustered(pool)
+        .expect("Failed to get redis-cluster-unpooled connection");
+    RedisPool::ClusteredUnpooled(ClusteredRedisUnpooled { con })
+}
+
+pub async fn new_redis_pool_clustered(redis_dsn: &str, cfg: &Configuration) -> RedisPool {
+    new_redis_pool_helper(redis_dsn, true, cfg.redis_pool_max_size).await
+}
+
+pub async fn new_redis_clustered_unpooled(redis_dsn: &str) -> RedisPool {
+    new_redis_unpooled_helper(redis_dsn).await
 }
 
 pub async fn new_redis_pool(redis_dsn: &str, cfg: &Configuration) -> RedisPool {
-    let mgr =
-        RedisMultiplexedConnectionManager::new(redis_dsn).expect("Error initializing redis client");
-    let pool = bb8::Pool::builder()
-        .max_size(cfg.redis_pool_max_size.into())
-        .build(mgr)
-        .await
-        .expect("Error initializing redis connection pool");
-    let pool = NonClusteredRedisPool { pool };
-    RedisPool::NonClustered(pool)
+    new_redis_pool_helper(redis_dsn, false, cfg.redis_pool_max_size).await
 }
 
 #[cfg(test)]
@@ -168,6 +246,7 @@ mod tests {
     async fn get_pool(redis_dsn: &str, cfg: &Configuration) -> RedisPool {
         match cfg.cache_type {
             CacheType::RedisCluster => super::new_redis_pool_clustered(redis_dsn, cfg).await,
+            CacheType::RedisClusterUnpooled => super::new_redis_clustered_unpooled(redis_dsn).await,
             CacheType::Redis => super::new_redis_pool(redis_dsn, cfg).await,
             _ => panic!(
                 "This test should only be run when redis is configured as the cache provider"