Remove connection pool for redis cache variant

Replace connection pooling with unpooled
`redis::aio::ConnectionManager`. The ConnectionManager uses a single
multiplexed connection to handle all requests and also handles
reconnections directly, making connection pooling unnecessary.

Note that these changes applying *only* to redis caching, not queuing.
This is safe in the redis caching layer because we do not make
blocking calls there. However, we *do* make blocking calls currently
in our redis queue implementation, so we must continue to use
connection pooling there to avoid blocking concurrent requests.

Also note that names for our many methods of acquiring a `RedisPool`
have been standardized for less ambiguity.

Author: jaymell

server/Cargo.lock

@@ -49,7 +49,7 @@ 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.15.0"
-redis = { version = "0.25.4", features = ["tokio-comp", "tokio-native-tls-comp", "streams", "cluster-async", "tcp_nodelay"] }
+redis = { version = "0.25.4", features = ["tokio-comp", "tokio-native-tls-comp", "streams", "cluster-async", "tcp_nodelay", "connection-manager"] }
 thiserror = "1.0.30"
 bytes = "1.1.0"
 blake2 = "0.10.4"

server/svix-server/Cargo.toml

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

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

@@ -110,7 +110,7 @@ pub async fn run_with_prefix(
         CacheBackend::None => cache::none::new(),
         CacheBackend::Memory => cache::memory::new(),
         CacheBackend::Redis(dsn) => {
-            let mgr = crate::redis::new_redis_pool(dsn, &cfg).await;
+            let mgr = crate::redis::new_redis_unpooled(dsn).await;
             cache::redis::new(mgr)
         }
         CacheBackend::RedisCluster(dsn) => {

server/svix-server/src/lib.rs

@@ -134,8 +134,8 @@ async fn new_pair_inner(
     // - redis tests that only makes sense to run with the DSN set
     let dsn = cfg.redis_dsn.as_deref().unwrap();
     let pool = match &cfg.queue_type {
-        QueueType::RedisCluster => crate::redis::new_redis_pool_clustered(dsn, cfg).await,
-        _ => crate::redis::new_redis_pool(dsn, cfg).await,
+        QueueType::RedisCluster => crate::redis::new_redis_clustered_pooled(dsn, cfg).await,
+        _ => crate::redis::new_redis_pooled(dsn, cfg).await,
     };
 
     // Create the stream and consumer group for the MAIN queue should it not already exist. The
@@ -364,10 +364,11 @@ pub mod tests {
     async fn get_pool(cfg: &Configuration) -> RedisPool {
         match cfg.queue_type {
             QueueType::RedisCluster => {
-                crate::redis::new_redis_pool_clustered(cfg.redis_dsn.as_deref().unwrap(), cfg).await
+                crate::redis::new_redis_clustered_pooled(cfg.redis_dsn.as_deref().unwrap(), cfg)
+                    .await
             }
             QueueType::Redis => {
-                crate::redis::new_redis_pool(cfg.redis_dsn.as_deref().unwrap(), cfg).await
+                crate::redis::new_redis_pooled(cfg.redis_dsn.as_deref().unwrap(), cfg).await
             }
             _ => {
                 panic!("This test should only be run when redis is configured as the queue backend")

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

@@ -16,6 +16,7 @@ pub enum RedisPool {
     Clustered(ClusteredRedisPool),
     ClusteredUnpooled(ClusteredRedisUnpooled),
     NonClustered(NonClusteredRedisPool),
+    NonClusteredUnpooled(NonClusteredRedisUnpooled),
 }
 
 impl RedisPool {
@@ -24,6 +25,7 @@ impl RedisPool {
             Self::Clustered(pool) => pool.get().await,
             Self::NonClustered(pool) => pool.get().await,
             Self::ClusteredUnpooled(pool) => pool.get().await,
+            Self::NonClusteredUnpooled(pool) => pool.get().await,
         }
     }
 }
@@ -63,6 +65,27 @@ impl std::fmt::Debug for ClusteredRedisUnpooled {
     }
 }
 
+#[derive(Clone)]
+pub struct NonClusteredRedisUnpooled {
+    con: redis::aio::ConnectionManager,
+}
+
+impl NonClusteredRedisUnpooled {
+    pub async fn get(&self) -> Result<PooledConnection<'_>, RunError<RedisError>> {
+        Ok(PooledConnection::NonClusteredUnpooled(
+            NonClusteredUnpooledConnection {
+                con: self.con.clone(),
+            },
+        ))
+    }
+}
+
+impl std::fmt::Debug for NonClusteredRedisUnpooled {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("NonClusteredRedisUnpooled").finish()
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct NonClusteredRedisPool {
     pool: Pool<RedisConnectionManager>,
@@ -80,6 +103,7 @@ pub enum PooledConnection<'a> {
     Clustered(ClusteredPooledConnection<'a>),
     ClusteredUnpooled(ClusteredUnpooledConnection),
     NonClustered(NonClusteredPooledConnection<'a>),
+    NonClusteredUnpooled(NonClusteredUnpooledConnection),
 }
 
 impl PooledConnection<'_> {
@@ -104,6 +128,7 @@ impl redis::aio::ConnectionLike for PooledConnection<'_> {
             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),
+            PooledConnection::NonClusteredUnpooled(conn) => conn.con.req_packed_command(cmd),
         }
     }
 
@@ -121,6 +146,9 @@ impl redis::aio::ConnectionLike for PooledConnection<'_> {
             PooledConnection::ClusteredUnpooled(conn) => {
                 conn.con.req_packed_commands(cmd, offset, count)
             }
+            PooledConnection::NonClusteredUnpooled(conn) => {
+                conn.con.req_packed_commands(cmd, offset, count)
+            }
         }
     }
 
@@ -129,6 +157,7 @@ impl redis::aio::ConnectionLike for PooledConnection<'_> {
             PooledConnection::Clustered(conn) => conn.con.get_db(),
             PooledConnection::NonClustered(conn) => conn.con.get_db(),
             PooledConnection::ClusteredUnpooled(conn) => conn.con.get_db(),
+            PooledConnection::NonClusteredUnpooled(conn) => conn.con.get_db(),
         }
     }
 }
@@ -150,6 +179,23 @@ impl<'a> NonClusteredPooledConnection<'a> {
     }
 }
 
+pub struct NonClusteredUnpooledConnection {
+    con: redis::aio::ConnectionManager,
+}
+
+impl NonClusteredUnpooledConnection {
+    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
+    }
+}
+
 pub struct ClusteredPooledConnection<'a> {
     con: bb8::PooledConnection<'a, RedisClusterConnectionManager>,
 }
@@ -211,31 +257,50 @@ async fn new_redis_pool_helper(
     }
 }
 
-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()
+async fn new_redis_unpooled_helper(redis_dsn: &str, clustered: bool) -> RedisPool {
+    if clustered {
+        let cli = redis::cluster::ClusterClient::builder(vec![redis_dsn])
+            .retries(1)
+            .connection_timeout(REDIS_CONN_TIMEOUT)
+            .build()
+            .expect("Error initializing redis-unpooled cluster client");
+        let con = cli
+            .get_async_connection()
+            .await
+            .expect("Failed to get redis-cluster-unpooled connection");
+        RedisPool::ClusteredUnpooled(ClusteredRedisUnpooled { con })
+    } else {
+        let cli = redis::Client::open(redis_dsn).expect("Error initializing redis unpooled client");
+        let con = redis::aio::ConnectionManager::new_with_backoff_and_timeouts(
+            cli,
+            2,
+            100,
+            1,
+            Duration::MAX,
+            REDIS_CONN_TIMEOUT,
+        )
         .await
-        .expect("Failed to get redis-cluster-unpooled connection");
-    RedisPool::ClusteredUnpooled(ClusteredRedisUnpooled { con })
+        .expect("Failed to get redis-unpooled connection manager");
+        RedisPool::NonClusteredUnpooled(NonClusteredRedisUnpooled { con })
+    }
 }
 
-pub async fn new_redis_pool_clustered(redis_dsn: &str, cfg: &Configuration) -> RedisPool {
+pub async fn new_redis_clustered_pooled(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
+    new_redis_unpooled_helper(redis_dsn, true).await
 }
 
-pub async fn new_redis_pool(redis_dsn: &str, cfg: &Configuration) -> RedisPool {
+pub async fn new_redis_pooled(redis_dsn: &str, cfg: &Configuration) -> RedisPool {
     new_redis_pool_helper(redis_dsn, false, cfg.redis_pool_max_size).await
 }
 
+pub async fn new_redis_unpooled(redis_dsn: &str) -> RedisPool {
+    new_redis_unpooled_helper(redis_dsn, false).await
+}
+
 #[cfg(test)]
 mod tests {
     use redis::AsyncCommands;
@@ -246,7 +311,7 @@ mod tests {
     async fn get_pool(redis_dsn: &str, cfg: &Configuration) -> RedisPool {
         match cfg.cache_type {
             CacheType::RedisCluster => super::new_redis_clustered_unpooled(redis_dsn).await,
-            CacheType::Redis => super::new_redis_pool(redis_dsn, cfg).await,
+            CacheType::Redis => super::new_redis_unpooled(redis_dsn).await,
             _ => panic!(
                 "This test should only be run when redis is configured as the cache provider"
             ),

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

@@ -12,7 +12,7 @@ use svix_server::{
         message_app::AppEndpointKey,
         types::{BaseId, OrganizationId},
     },
-    redis::{new_redis_clustered_unpooled, new_redis_pool},
+    redis::{new_redis_clustered_unpooled, new_redis_unpooled},
 };
 
 use crate::utils::{
@@ -66,7 +66,7 @@ async fn test_app_deletion() {
     let cache = match cfg.cache_backend() {
         CacheBackend::None => cache::none::new(),
         CacheBackend::Redis(dsn) => {
-            let mgr = new_redis_pool(dsn, &cfg).await;
+            let mgr = new_redis_unpooled(dsn).await;
             cache::redis::new(mgr)
         }
         CacheBackend::RedisCluster(dsn) => {
@@ -150,7 +150,7 @@ async fn test_endp_deletion() {
     let cache = match cfg.cache_backend() {
         CacheBackend::None => cache::none::new(),
         CacheBackend::Redis(dsn) => {
-            let mgr = new_redis_pool(dsn, &cfg).await;
+            let mgr = new_redis_unpooled(dsn).await;
             cache::redis::new(mgr)
         }
         CacheBackend::RedisCluster(dsn) => {

server/svix-server/tests/it/message_app.rs

@@ -13,16 +13,16 @@ use svix_server::{
     queue::{
         new_pair, MessageTask, QueueTask, TaskQueueConsumer, TaskQueueDelivery, TaskQueueProducer,
     },
-    redis::{new_redis_pool, new_redis_pool_clustered, RedisPool},
+    redis::{new_redis_clustered_pooled, new_redis_pooled, RedisPool},
 };
 
 // TODO: Don't copy this from the Redis queue test directly, place the fn somewhere both can access
 async fn get_pool(cfg: &Configuration) -> RedisPool {
     match cfg.queue_type {
         QueueType::RedisCluster => {
-            new_redis_pool_clustered(cfg.redis_dsn.as_deref().unwrap(), cfg).await
+            new_redis_clustered_pooled(cfg.redis_dsn.as_deref().unwrap(), cfg).await
         }
-        QueueType::Redis => new_redis_pool(cfg.redis_dsn.as_deref().unwrap(), cfg).await,
+        QueueType::Redis => new_redis_pooled(cfg.redis_dsn.as_deref().unwrap(), cfg).await,
         _ => {
             panic!("This test should only be run when redis is configured as the queue backend")
         }