Introduce a fast reconnect process for async cluster connections.

The process is periodic and can be configured via ClusterParams.
This process ensures that all expected user connections exist and have not been passively closed.
The expected connections are calculated from the current slot map.
Additionally, for the Tokio runtime, an instant disconnect notification is available, allowing the reconnect process to be triggered instantly without waiting for the periodic check.
This process is especially important for pub/sub support, as passive disconnects can render a pub/sub subscriber inoperative.
Three integration tests are introduced with this feature: a generic fast reconnect test, pub/sub resilience to passive disconnects, and pub/sub resilience to scale-out.

Author: ikolomi

Diff for: redis-test/src/lib.rs

@@ -288,6 +288,10 @@ impl AioConnectionLike for MockRedisConnection {
     fn get_db(&self) -> i64 {
         0
     }
+
+    fn is_closed(&self) -> bool {
+        false
+    }
 }
 
 #[cfg(test)]

Diff for: redis/examples/async-await.rs

@@ -4,7 +4,7 @@ use redis::AsyncCommands;
 #[tokio::main]
 async fn main() -> redis::RedisResult<()> {
     let client = redis::Client::open("redis://127.0.0.1/").unwrap();
-    let mut con = client.get_multiplexed_async_connection(None).await?;
+    let mut con = client.get_multiplexed_async_connection(None, None).await?;
 
     con.set("key1", b"foo").await?;
 

Diff for: redis/examples/async-connection-loss.rs

@@ -80,7 +80,9 @@ async fn main() -> RedisResult<()> {
 
     let client = redis::Client::open("redis://127.0.0.1/").unwrap();
     match mode {
-        Mode::Default => run_multi(client.get_multiplexed_tokio_connection(None).await?).await?,
+        Mode::Default => {
+            run_multi(client.get_multiplexed_tokio_connection(None, None).await?).await?
+        }
         Mode::Reconnect => run_multi(client.get_connection_manager().await?).await?,
         #[allow(deprecated)]
         Mode::Deprecated => run_single(client.get_async_connection(None).await?).await?,

Diff for: redis/examples/async-multiplexed.rs

@@ -34,7 +34,10 @@ async fn test_cmd(con: &MultiplexedConnection, i: i32) -> RedisResult<()> {
 async fn main() {
     let client = redis::Client::open("redis://127.0.0.1/").unwrap();
 
-    let con = client.get_multiplexed_tokio_connection(None).await.unwrap();
+    let con = client
+        .get_multiplexed_tokio_connection(None, None)
+        .await
+        .unwrap();
 
     let cmds = (0..100).map(|i| test_cmd(&con, i));
     let result = future::try_join_all(cmds).await.unwrap();

Diff for: redis/examples/async-pub-sub.rs

@@ -5,7 +5,7 @@ use redis::AsyncCommands;
 #[tokio::main]
 async fn main() -> redis::RedisResult<()> {
     let client = redis::Client::open("redis://127.0.0.1/").unwrap();
-    let mut publish_conn = client.get_multiplexed_async_connection(None).await?;
+    let mut publish_conn = client.get_multiplexed_async_connection(None, None).await?;
     let mut pubsub_conn = client.get_async_pubsub().await?;
 
     pubsub_conn.subscribe("wavephone").await?;

Diff for: redis/examples/async-scan.rs

@@ -5,7 +5,7 @@ use redis::{AsyncCommands, AsyncIter};
 #[tokio::main]
 async fn main() -> redis::RedisResult<()> {
     let client = redis::Client::open("redis://127.0.0.1/").unwrap();
-    let mut con = client.get_multiplexed_async_connection(None).await?;
+    let mut con = client.get_multiplexed_async_connection(None, None).await?;
 
     con.set("async-key1", b"foo").await?;
     con.set("async-key2", b"foo").await?;

Diff for: redis/src/aio/connection.rs

@@ -305,6 +305,11 @@ where
     fn get_db(&self) -> i64 {
         self.db
     }
+
+    fn is_closed(&self) -> bool {
+        // always false for AsyncRead + AsyncWrite (cant do better)
+        false
+    }
 }
 
 /// Represents a `PubSub` connection.

Diff for: redis/src/aio/connection_manager.rs

@@ -196,6 +196,7 @@ impl ConnectionManager {
                 response_timeout,
                 connection_timeout,
                 None,
+                None,
             )
         })
         .await
@@ -301,4 +302,9 @@ impl ConnectionLike for ConnectionManager {
     fn get_db(&self) -> i64 {
         self.client.connection_info().redis.db
     }
+
+    fn is_closed(&self) -> bool {
+        // always return false due to automatic reconnect
+        false
+    }
 }

Diff for: redis/src/aio/mod.rs

@@ -85,6 +85,24 @@ pub trait ConnectionLike {
     /// also might be incorrect if the connection like object is not
     /// actually connected.
     fn get_db(&self) -> i64;
+
+    /// Returns the state of the connection
+    fn is_closed(&self) -> bool;
+}
+
+/// Implements ability to notify about disconnection events
+pub trait DisconnectNotifier: Send + Sync {
+    /// Notify about disconnect event
+    fn notify_disconnect(&mut self);
+
+    /// Inteded to be used with Box
+    fn clone_box(&self) -> Box<dyn DisconnectNotifier>;
+}
+
+impl Clone for Box<dyn DisconnectNotifier> {
+    fn clone(&self) -> Box<dyn DisconnectNotifier> {
+        self.clone_box()
+    }
 }
 
 // Initial setup for every connection.

Diff for: redis/src/aio/multiplexed_connection.rs

@@ -1,5 +1,6 @@
 use super::{ConnectionLike, Runtime};
 use crate::aio::setup_connection;
+use crate::aio::DisconnectNotifier;
 use crate::cmd::Cmd;
 #[cfg(any(feature = "tokio-comp", feature = "async-std-comp"))]
 use crate::parser::ValueCodec;
@@ -23,6 +24,7 @@ use std::fmt;
 use std::fmt::Debug;
 use std::io;
 use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::task::{self, Poll};
 use std::time::Duration;
@@ -77,13 +79,15 @@ struct Pipeline<SinkItem> {
     sender: mpsc::Sender<PipelineMessage<SinkItem>>,
 
     push_manager: Arc<ArcSwap<PushManager>>,
+    is_stream_closed: Arc<AtomicBool>,
 }
 
 impl<SinkItem> Clone for Pipeline<SinkItem> {
     fn clone(&self) -> Self {
         Pipeline {
             sender: self.sender.clone(),
             push_manager: self.push_manager.clone(),
+            is_stream_closed: self.is_stream_closed.clone(),
         }
     }
 }
@@ -104,14 +108,21 @@ pin_project! {
         in_flight: VecDeque<InFlight>,
         error: Option<RedisError>,
         push_manager: Arc<ArcSwap<PushManager>>,
+        disconnect_notifier: Option<Box<dyn DisconnectNotifier>>,
+        is_stream_closed: Arc<AtomicBool>,
     }
 }
 
 impl<T> PipelineSink<T>
 where
     T: Stream<Item = RedisResult<Value>> + 'static,
 {
-    fn new<SinkItem>(sink_stream: T, push_manager: Arc<ArcSwap<PushManager>>) -> Self
+    fn new<SinkItem>(
+        sink_stream: T,
+        push_manager: Arc<ArcSwap<PushManager>>,
+        disconnect_notifier: Option<Box<dyn DisconnectNotifier>>,
+        is_stream_closed: Arc<AtomicBool>,
+    ) -> Self
     where
         T: Sink<SinkItem, Error = RedisError> + Stream<Item = RedisResult<Value>> + 'static,
     {
@@ -120,6 +131,8 @@ where
             in_flight: VecDeque::new(),
             error: None,
             push_manager,
+            disconnect_notifier,
+            is_stream_closed,
         }
     }
 
@@ -130,7 +143,15 @@ where
                 Some(result) => result,
                 // The redis response stream is not going to produce any more items so we `Err`
                 // to break out of the `forward` combinator and stop handling requests
-                None => return Poll::Ready(Err(())),
+                None => {
+                    // this is the right place to notify about the passive TCP disconnect
+                    // In other places we cannot distinguish between the active destruction of MultiplexedConnection and passive disconnect
+                    if let Some(disconnect_notifier) = self.as_mut().project().disconnect_notifier {
+                        disconnect_notifier.notify_disconnect();
+                    }
+                    self.is_stream_closed.store(true, Ordering::Relaxed);
+                    return Poll::Ready(Err(()));
+                }
             };
             self.as_mut().send_result(item);
         }
@@ -296,7 +317,10 @@ impl<SinkItem> Pipeline<SinkItem>
 where
     SinkItem: Send + 'static,
 {
-    fn new<T>(sink_stream: T) -> (Self, impl Future<Output = ()>)
+    fn new<T>(
+        sink_stream: T,
+        disconnect_notifier: Option<Box<dyn DisconnectNotifier>>,
+    ) -> (Self, impl Future<Output = ()>)
     where
         T: Sink<SinkItem, Error = RedisError> + Stream<Item = RedisResult<Value>> + 'static,
         T: Send + 'static,
@@ -308,7 +332,13 @@ where
         let (sender, mut receiver) = mpsc::channel(BUFFER_SIZE);
         let push_manager: Arc<ArcSwap<PushManager>> =
             Arc::new(ArcSwap::new(Arc::new(PushManager::default())));
-        let sink = PipelineSink::new::<SinkItem>(sink_stream, push_manager.clone());
+        let is_stream_closed = Arc::new(AtomicBool::new(false));
+        let sink = PipelineSink::new::<SinkItem>(
+            sink_stream,
+            push_manager.clone(),
+            disconnect_notifier,
+            is_stream_closed.clone(),
+        );
         let f = stream::poll_fn(move |cx| receiver.poll_recv(cx))
             .map(Ok)
             .forward(sink)
@@ -317,6 +347,7 @@ where
             Pipeline {
                 sender,
                 push_manager,
+                is_stream_closed,
             },
             f,
         )
@@ -363,6 +394,10 @@ where
     async fn set_push_manager(&mut self, push_manager: PushManager) {
         self.push_manager.store(Arc::new(push_manager));
     }
+
+    pub fn is_closed(&self) -> bool {
+        self.is_stream_closed.load(Ordering::Relaxed)
+    }
 }
 
 /// A connection object which can be cloned, allowing requests to be be sent concurrently
@@ -392,6 +427,7 @@ impl MultiplexedConnection {
         connection_info: &ConnectionInfo,
         stream: C,
         push_sender: Option<mpsc::UnboundedSender<PushInfo>>,
+        disconnect_notifier: Option<Box<dyn DisconnectNotifier>>,
     ) -> RedisResult<(Self, impl Future<Output = ()>)>
     where
         C: Unpin + AsyncRead + AsyncWrite + Send + 'static,
@@ -401,6 +437,7 @@ impl MultiplexedConnection {
             stream,
             std::time::Duration::MAX,
             push_sender,
+            disconnect_notifier,
         )
         .await
     }
@@ -412,6 +449,7 @@ impl MultiplexedConnection {
         stream: C,
         response_timeout: std::time::Duration,
         push_sender: Option<mpsc::UnboundedSender<PushInfo>>,
+        disconnect_notifier: Option<Box<dyn DisconnectNotifier>>,
     ) -> RedisResult<(Self, impl Future<Output = ()>)>
     where
         C: Unpin + AsyncRead + AsyncWrite + Send + 'static,
@@ -429,7 +467,7 @@ impl MultiplexedConnection {
         let codec = ValueCodec::default()
             .framed(stream)
             .and_then(|msg| async move { msg });
-        let (mut pipeline, driver) = Pipeline::new(codec);
+        let (mut pipeline, driver) = Pipeline::new(codec, disconnect_notifier);
         let driver = boxed(driver);
         let pm = PushManager::default();
         if let Some(sender) = push_sender {
@@ -560,6 +598,10 @@ impl ConnectionLike for MultiplexedConnection {
     fn get_db(&self) -> i64 {
         self.db
     }
+
+    fn is_closed(&self) -> bool {
+        self.pipeline.is_closed()
+    }
 }
 impl MultiplexedConnection {
     /// Subscribes to a new channel.