fix: Use Redis ConnectionManager to reconnect on failures

CHANGELOG.md

@@ -4,18 +4,23 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
-### Changed
+### Fixed
 
-- Bump Rust dependency versions ([#19]).
+- Use redis [`ConnectionManager`](https://docs.rs/redis/latest/redis/aio/struct.ConnectionManager.html) to reconnect on
+  Redis connection failures. Previously trino-lb would stop working once the Redis Pod restarted. This change only
+  affects the single Redis instance connection, *not* the cluster mode connection, as a
+  [ClusterConnection](https://docs.rs/redis/latest/redis/cluster/struct.ClusterConnection.html) does not seem to support
+  a [`ConnectionManager`](https://docs.rs/redis/latest/redis/aio/struct.ConnectionManager.html) ([#34]).
 
-[#19]: https://github.com/stackabletech/trino-lb/pull/19
+[#34]: https://github.com/stackabletech/trino-lb/pull/34
 
 ## [0.2.0] - 2024-04-07
 
 ### Added
 
 - BREAKING: Add support for single instance redis persistence (without Redis cluster mode).
-  This is breaking, because you need to set `trinoLb.persistence.redis.clusterMode: true` in your config to keep using the cluster mode ([#15]).
+  This is breaking, because you need to set `trinoLb.persistence.redis.clusterMode: true` in your config to keep using
+  the cluster mode ([#15]).
 
 [#15]: https://github.com/stackabletech/trino-lb/pull/15
 

trino-lb-persistence/src/lib.rs

@@ -88,7 +88,7 @@ pub trait Persistence {
 
 #[enum_dispatch]
 pub enum PersistenceImplementation {
-    Redis(redis::RedisPersistence<::redis::aio::MultiplexedConnection>),
+    Redis(redis::RedisPersistence<::redis::aio::ConnectionManager>),
     RedisCluster(
         redis::RedisPersistence<
             ::redis::cluster_async::ClusterConnection<::redis::aio::MultiplexedConnection>,

trino-lb-persistence/src/redis/mod.rs

@@ -6,7 +6,9 @@ use std::{
 
 use futures::{future::try_join_all, TryFutureExt};
 use redis::{
-    aio::MultiplexedConnection, cluster::ClusterClientBuilder, cluster_async::ClusterConnection,
+    aio::{ConnectionManager, MultiplexedConnection},
+    cluster::ClusterClientBuilder,
+    cluster_async::ClusterConnection,
     AsyncCommands, Client, RedisError, Script,
 };
 use snafu::{OptionExt, ResultExt, Snafu};
@@ -124,7 +126,7 @@ where
     cluster_groups: Vec<String>,
 }
 
-impl RedisPersistence<MultiplexedConnection> {
+impl RedisPersistence<ConnectionManager> {
     pub async fn new(config: &RedisConfig, cluster_groups: Vec<String>) -> Result<Self, Error> {
         let redis_host = config.endpoint.host_str().context(ExtractRedisHostSnafu {
             endpoint: config.endpoint.clone(),
@@ -133,7 +135,7 @@ impl RedisPersistence<MultiplexedConnection> {
 
         let client = Client::open(config.endpoint.as_str()).context(CreateClientSnafu)?;
         let connection = client
-            .get_multiplexed_async_connection()
+            .get_connection_manager()
             .await
             .context(CreateClientSnafu)?;
 

trino-lb/src/main.rs

@@ -116,7 +116,7 @@ async fn start() -> Result<(), MainError> {
                     .context(CreateRedisPersistenceClientSnafu)?
                     .into()
                 } else {
-                    RedisPersistence::<::redis::aio::MultiplexedConnection>::new(
+                    RedisPersistence::<::redis::aio::ConnectionManager>::new(
                         redis_config,
                         cluster_groups,
                     )