fix(node): close() detaches a standalone client with a blocking command in flight

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Fixes
 
+* Node/Core: `close()` now detaches a standalone client with a blocking command in flight (`XREADGROUP ... BLOCK`, `BLPOP`, ...) from the server immediately. The connection stayed open until the server answered the blocked command, so entries added to a stream after `close()` were still delivered to the closed consumer's PEL, and an idle client lingered in `CLIENT LIST` for up to 3 s after `close()`. Closing a client, or dropping its last clone, now aborts the connection driver instead of waiting for in-flight replies ([#7242](https://github.com/valkey-io/valkey-glide/issues/7242), [#7244](https://github.com/valkey-io/valkey-glide/pull/7244))
 * Java: Clear jedis-compat-shared javadoc errors and deprecate substr ([#7191](https://github.com/valkey-io/valkey-glide/pull/7191))
 * Core: A scope pool no longer leaks capacity when connecting to a paused or unreachable shard. A reserved `max_total` slot is now held by an RAII guard that returns it on every creation outcome — including a creation cancelled mid-handshake, which previously leaked the slot until scopes to healthy shards could no longer be opened — and decrements saturate at zero so a give-back cannot wrap into permanent exhaustion. Concurrent retries of a single acquire are deduped to one in-flight creation, so a retry storm against a slow shard can no longer consume the whole pool, while distinct concurrent acquires each still open their own connection ([#6966](https://github.com/valkey-io/valkey-glide/issues/6966), [#7067](https://github.com/valkey-io/valkey-glide/issues/7067))
 * Core/All, Node: Fix ClientPool abandon monitor race for blocking commands ([#7063](https://github.com/valkey-io/valkey-glide/pull/7063))

glide-core/redis-rs/redis/src/aio/multiplexed_connection.rs

@@ -16,7 +16,7 @@ use ::tokio::{
 };
 use arc_swap::ArcSwap;
 use futures_util::{
-    future::{Future, FutureExt},
+    future::{AbortHandle, Abortable, Future, FutureExt},
     ready,
     sink::Sink,
     stream::{self, Stream, StreamExt, TryStreamExt as _},
@@ -273,6 +273,8 @@ pub(crate) struct Pipeline<SinkItem> {
     /// response reading, so a response-only signal would freeze. See
     /// [`Self::send_recv`].
     progress: Arc<AtomicU64>,
+    /// Aborts the driver task that owns the socket. See [`Pipeline::kill`].
+    abort_handle: AbortHandle,
 }
 
 impl<SinkItem> Debug for Pipeline<SinkItem>
@@ -786,16 +788,18 @@ where
             cache,
             progress.clone(),
         );
+        let (abort_handle, abort_registration) = AbortHandle::new_pair();
         let f = stream::poll_fn(move |cx| receiver.poll_recv(cx))
             .map(Ok)
-            .forward(sink)
-            .map(|_| ());
+            .forward(sink);
+        let f = Abortable::new(f, abort_registration).map(|_| ());
         (
             Pipeline {
                 sender,
                 push_manager,
                 is_stream_closed,
                 progress,
+                abort_handle,
             },
             f,
         )
@@ -964,6 +968,18 @@ where
     pub fn is_closed(&self) -> bool {
         self.is_stream_closed.load(Ordering::Relaxed)
     }
+
+    /// Tears the connection down immediately by aborting the driver task.
+    ///
+    /// Dropping all `Pipeline` clones is not enough to close the socket: the driver's
+    /// `poll_close` waits for every in-flight request to be answered first, so a blocking
+    /// command (`XREADGROUP ... BLOCK`, `BLPOP`, ...) keeps the socket open until the server
+    /// replies. Aborting the driver drops the sink and the underlying stream, which closes
+    /// the socket and fails every pending request with a `FatalReceiveError`.
+    pub fn kill(&self) {
+        self.is_stream_closed.store(true, Ordering::Relaxed);
+        self.abort_handle.abort();
+    }
 }
 
 /// A connection object which can be cloned, allowing requests to be be sent concurrently
@@ -1083,6 +1099,14 @@ impl MultiplexedConnection {
         self.response_timeout = timeout;
     }
 
+    /// Closes the socket immediately, without waiting for in-flight requests to complete.
+    ///
+    /// Pending requests fail with `FatalReceiveError`; later requests fail with `FatalSendError`.
+    /// The server sees the TCP close and discards any command that is blocked on this connection.
+    pub fn kill(&self) {
+        self.pipeline.kill();
+    }
+
     /// Sends an already encoded (packed) command into the TCP socket and
     /// reads the single response from it.
     pub async fn send_packed_command(&mut self, cmd: &Cmd) -> RedisResult<Value> {

glide-core/src/client/mod.rs

@@ -26,7 +26,7 @@ use redis::{
 use regex::Regex;
 pub use standalone_client::StandaloneClient;
 use std::io;
-use std::sync::atomic::{AtomicIsize, AtomicU32, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicIsize, AtomicU32, AtomicU64, Ordering};
 use std::sync::{Arc, LazyLock};
 use std::thread;
 use std::thread::JoinHandle;
@@ -389,6 +389,9 @@ pub struct ClientShared {
     current_database: Arc<AtomicU32>,
     // Whether this client is in cluster mode (immutable).
     is_cluster: bool,
+    /// Set by `kill()`. A lazy client checks it before connecting so a command queued
+    /// before `close()` cannot open a connection afterwards.
+    killed: AtomicBool,
 }
 
 /// Why [`Client::address_for_slot`] / [`Client::try_address_for_slot`] could not
@@ -1016,6 +1019,12 @@ impl Client {
         config.lazy_connect = false;
 
         let mut guard = self.internal_client.write().await;
+        if self.killed.load(Ordering::Acquire) {
+            return Err(RedisError::from((
+                ErrorKind::ClientError,
+                "Client was closed before the lazy connection was established",
+            )));
+        }
         let iam_manager_ref = self.iam_token_manager.as_ref();
         if let ClientWrapper::Lazy(_) = &*guard {
             // Create the appropriate client based on configuration
@@ -1075,6 +1084,24 @@ impl Client {
         Ok(guard.clone()) // ✅ Return clone of the now-initialized wrapper
     }
 
+    /// Closes the underlying connections immediately.
+    ///
+    /// Unlike dropping the client, this does not wait for in-flight requests: a blocking
+    /// command (`XREADGROUP ... BLOCK`, `BLPOP`, ...) is cut off and the server discards it.
+    /// Pending requests fail with a connection error. The client must not be used afterwards.
+    ///
+    /// Cluster connections are not torn down here; they close once every clone is dropped
+    /// and the in-flight requests have been answered.
+    pub async fn kill(&self) {
+        // Set first: a lazy client that has not connected yet checks this flag under the
+        // write lock in get_or_initialize_client and refuses to connect.
+        self.killed.store(true, Ordering::Release);
+        let guard = self.internal_client.read().await;
+        if let ClientWrapper::Standalone(client) = &*guard {
+            client.kill();
+        }
+    }
+
     /// Internal command execution logic. Takes owned data so the returned future
     /// is `Send + 'static`.
     async fn execute_command_owned(
@@ -2962,6 +2989,7 @@ impl Client {
                     }),
                     current_database: Arc::new(AtomicU32::new(request.database_id as u32)),
                     is_cluster: request.cluster_mode_enabled,
+                    killed: AtomicBool::new(false),
                 }),
                 iam_token_manager: None,
                 otel_metadata: Arc::new(otel_metadata),
@@ -3192,6 +3220,7 @@ impl Client {
                 circuit_breaker: None,
                 current_database: Arc::new(AtomicU32::new(0)),
                 is_cluster: false,
+                killed: AtomicBool::new(false),
             }),
             iam_token_manager: None,
             otel_metadata: Arc::new(OTelMetadata {
@@ -3259,6 +3288,7 @@ pub fn create_test_glide_client() -> Client {
             circuit_breaker: None,
             current_database: Arc::new(AtomicU32::new(0)),
             is_cluster: false,
+            killed: AtomicBool::new(false),
         }),
         iam_token_manager: None,
         otel_metadata: Arc::new(OTelMetadata {
@@ -3661,6 +3691,7 @@ mod tests {
     fn create_test_client() -> Client {
         use crate::pubsub::create_pubsub_synchronizer;
         use std::sync::Arc;
+        use std::sync::atomic::AtomicBool;
         use std::sync::atomic::AtomicIsize;
         use std::sync::atomic::AtomicU32;
         use tokio::sync::RwLock;
@@ -3707,6 +3738,7 @@ mod tests {
                 circuit_breaker: None,
                 current_database: Arc::new(AtomicU32::new(0)),
                 is_cluster: false,
+                killed: AtomicBool::new(false),
             }),
             iam_token_manager: None,
             otel_metadata: Arc::new(OTelMetadata {
@@ -3719,6 +3751,24 @@ mod tests {
         }
     }
 
+    #[test]
+    fn kill_prevents_lazy_client_from_connecting() {
+        // A command queued before close() must not open the lazy connection after kill().
+        // Without the killed flag this would connect to 127.0.0.1:6379 (or fail with a
+        // connection error), never with ClientError.
+        let mut client = create_test_client();
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        rt.block_on(async {
+            client.kill().await;
+            let mut cmd = redis::cmd("PING");
+            let err = client.send_command(&mut cmd, None).await.unwrap_err();
+            assert!(
+                matches!(err.kind(), redis::ErrorKind::ClientError),
+                "expected ClientError, got {err}"
+            );
+        });
+    }
+
     #[test]
     fn test_is_client_set_name_command() {
         // Create a mock client for testing

glide-core/src/client/reconnecting_connection.rs

@@ -121,6 +121,9 @@ struct ConnectionBackend {
     connection_info: RwLock<redis::Client>,
     /// Once this flag is set, the internal connection needs no longer try to reconnect to the server, because all the outer clients were dropped.
     client_dropped_flagged: AtomicBool,
+    /// Notified by [`ReconnectingConnection::kill`]. The reconnect task races its probe against
+    /// this so a connection opened during a reconnect is closed as soon as the client is killed.
+    kill_notify: Notify,
     /// Optional handle to the IAM token cache for refreshing the password before reconnection.
     iam_token_handle: Option<IAMTokenHandle>,
     /// Optional handle to the reloaded mTLS certificate cache for refreshing the
@@ -372,6 +375,7 @@ impl ReconnectingConnection {
             connection_info: RwLock::new(connection_info),
             connection_available_signal: ManualResetEvent::new(true),
             client_dropped_flagged: AtomicBool::new(false),
+            kill_notify: Notify::new(),
             iam_token_handle,
             cert_material_handle,
         };
@@ -404,13 +408,39 @@ impl ReconnectingConnection {
     }
 
     pub(super) fn mark_as_dropped(&self) {
-        // Update the telemetry for each connection that is dropped. A dropped connection
-        // will not be re-connected, so update the telemetry here
-        Telemetry::decr_total_connections(1);
-        self.inner
+        let was_dropped = self
+            .inner
             .backend
             .client_dropped_flagged
-            .store(true, Ordering::Relaxed)
+            .swap(true, Ordering::Relaxed);
+
+        if !was_dropped {
+            // Update the telemetry for each connection that is dropped. A dropped connection
+            // will not be re-connected, so update the telemetry here
+            Telemetry::decr_total_connections(1);
+        }
+    }
+
+    /// Marks the connection as dropped and closes the socket right away.
+    ///
+    /// `mark_as_dropped` alone only stops future reconnects; the socket stays open until the
+    /// last `MultiplexedConnection` clone is gone and every in-flight request has been answered.
+    /// A blocking command in flight therefore keeps a closed client attached to the server.
+    /// Killing the connection makes the server drop the client immediately.
+    pub(super) fn kill(&self) {
+        self.mark_as_dropped();
+        {
+            let guard = self.inner.state.lock().unwrap();
+            if let ConnectionState::Connected(connection) = &*guard {
+                connection.kill();
+            }
+        }
+        // A reconnect in progress owns its new connection until the probe completes; wake it
+        // so it closes that connection instead of installing it.
+        self.inner.backend.kill_notify.notify_one();
+        // Requests parked in get_connection wait for the reconnect that will never finish.
+        // Wake them so they fail instead of hanging.
+        self.inner.backend.connection_available_signal.set();
     }
 
     pub(super) async fn try_get_connection(&self) -> Option<MultiplexedConnection> {
@@ -428,13 +458,23 @@ impl ReconnectingConnection {
             if let Some(connection) = self.try_get_connection().await {
                 return Ok(connection);
             }
+            if self.is_dropped() {
+                return Err(RedisError::from((
+                    redis::ErrorKind::ClientError,
+                    "Client was closed while waiting for a connection",
+                )));
+            }
         }
     }
 
     /// Attempt to re-connect the connection.
     ///
     /// This function spawns a task to perform the reconnection in the background
     pub(super) fn reconnect(&self, reason: ReconnectReason) {
+        if self.is_dropped() {
+            log_debug("reconnect", "skipped, client was dropped");
+            return;
+        }
         {
             let mut guard = self.inner.state.lock().unwrap();
             if matches!(*guard, ConnectionState::Reconnecting) {
@@ -491,7 +531,13 @@ impl ReconnectingConnection {
                         "ReconnectingConnection",
                         "reconnect stopped after client was dropped",
                     );
-                    // Client was dropped, reconnection attempts can stop
+                    // Client was dropped, reconnection attempts can stop. Wake requests
+                    // waiting for this reconnect so they fail instead of hanging.
+                    connection_clone
+                        .inner
+                        .backend
+                        .connection_available_signal
+                        .set();
                     return;
                 }
 
@@ -545,16 +591,57 @@ impl ReconnectingConnection {
                     .await
                 {
                     Ok(mut connection) => {
-                        if connection
-                            .send_packed_command(&redis::cmd("PING"))
-                            .await
-                            .is_err()
-                        {
-                            tokio::time::sleep(sleep_duration).await;
-                            continue;
+                        // Until the probe completes, this task is the only owner of the new
+                        // connection and kill() cannot reach it. Race the probe against the
+                        // kill signal so a close during the probe closes the socket right away
+                        // instead of after the server answers the PING.
+                        let ping = redis::cmd("PING");
+                        let probe = tokio::select! {
+                            biased;
+                            _ = connection_clone.inner.backend.kill_notify.notified() => None,
+                            result = connection.send_packed_command(&ping) => Some(result),
+                        };
+                        match probe {
+                            None => {
+                                log_debug(
+                                    "reconnect",
+                                    "client was dropped during the probe, closing the new connection",
+                                );
+                                connection.kill();
+                                connection_clone
+                                    .inner
+                                    .backend
+                                    .connection_available_signal
+                                    .set();
+                                return;
+                            }
+                            Some(Err(_)) => {
+                                // The connection is discarded; abort its driver instead of
+                                // leaving the socket open until the unanswered PING returns.
+                                connection.kill();
+                                tokio::time::sleep(sleep_duration).await;
+                                continue;
+                            }
+                            Some(Ok(_)) => {}
                         }
                         {
                             let mut guard = connection_clone.inner.state.lock().unwrap();
+                            // kill() may have run after the probe completed. It found the
+                            // state Reconnecting and had nothing to close, so close the new
+                            // connection here instead of installing it.
+                            if connection_clone.is_dropped() {
+                                log_debug(
+                                    "reconnect",
+                                    "client was dropped during reconnect, closing the new connection",
+                                );
+                                connection.kill();
+                                connection_clone
+                                    .inner
+                                    .backend
+                                    .connection_available_signal
+                                    .set();
+                                return;
+                            }
                             log_debug("reconnect", "completed successfully");
                             connection_clone
                                 .inner