fix(cluster): refuse an unroutable command before recording anything

CHANGELOG.md

@@ -20,6 +20,24 @@ Versions up to and including `0.19.3` are documented in the
   slots, including one passed to `PubSubStream::sunsubscribe`, is now split per
   slot and sent to the master owning each one.
 
+- **A command the cluster cannot route no longer poisons the connection.** A
+  command refused before being sent — keys spanning several slots where one
+  is required, or a command the cluster does not support — was refused only
+  when the network task fed it, after its bookkeeping had been recorded:
+  - a refused `SSUBSCRIBE` left its pending batch at the head of the
+    subscription queue, so the next subscription on the same `Client` was
+    matched against it and never answered;
+  - in a pipeline, the commands ahead of the refused one had already been
+    written, and their replies were handed to the next callers;
+  - the messages queued behind it in the same send batch were not sent until
+    another message woke the network task, so their callers could wait
+    forever on an idle client, and the refused message's byte charge was
+    never released from the send-queue budget.
+
+  Such a command is now refused before anything is recorded or written, and a
+  message whose feed fails for any other reason is failed on its own while
+  the rest of its batch goes out.
+
 ## [0.27.2] - 2026-10-20
 
 ### Fixed

src/network/cluster_connection.rs

@@ -44,6 +44,11 @@ pub(crate) struct ClusterTestHook {
     /// Counts every completed topology discovery, so a test can tell a refresh
     /// that happened on its own from one a redirection asked for.
     topology_refreshes: Arc<std::sync::atomic::AtomicUsize>,
+    /// When set, the next feed of a command by this name fails with
+    /// `ClusterConfig` before anything is written, reproducing a feed that
+    /// fails past the routing check: a slot no known node covers, or a node
+    /// connection that breaks on write.
+    feed_error_for: Arc<std::sync::Mutex<Option<String>>>,
 }
 
 #[cfg(test)]
@@ -113,6 +118,21 @@ impl ClusterTestHook {
         *self.transient_error.lock().unwrap() = Some(Bytes::from(format!("-{error}\r\n")));
     }
 
+    /// Arms a one-shot feed failure for the next command named `command_name`.
+    pub(crate) fn arm_feed_error_for(&self, command_name: &str) {
+        *self.feed_error_for.lock().unwrap() = Some(command_name.to_owned());
+    }
+
+    fn take_feed_error_for(&self, command_name: &[u8]) -> bool {
+        let mut armed = self.feed_error_for.lock().unwrap();
+        if armed.as_deref().map(str::as_bytes) == Some(command_name) {
+            *armed = None;
+            true
+        } else {
+            false
+        }
+    }
+
     fn take_transient_error(&self) -> Option<Bytes> {
         self.transient_error.lock().unwrap().take()
     }
@@ -152,6 +172,63 @@ fn is_shard_unsubscribe_spanning_slots(command: &Command) -> bool {
         .is_some_and(|first_slot| slots.any(|slot| slot != first_slot))
 }
 
+/// How the cluster connection serves a command, decided from the command alone.
+///
+/// The single place that tells a routable command from one that is not:
+/// [`ClusterConnection::check_routable`] and the feed path both read it, so
+/// what the handler accepts up front is exactly what the feed then serves.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Route {
+    /// A subscription command naming channels or patterns, hashed per name.
+    PubSubByChannel,
+    /// Every master.
+    AllShards,
+    /// Every node, masters and replicas alike.
+    AllNodes,
+    /// One slice per slot of its keys.
+    MultiShard,
+    /// The node owning the slot of its keys, or any node when it names none.
+    SingleShard,
+}
+
+fn route(command: &Command) -> Result<Route> {
+    if is_broadcast_pub_sub_command(command) {
+        // A channel-less UNSUBSCRIBE (or PUNSUBSCRIBE) names nothing to hash,
+        // and cancels every subscription the *connection* holds — which in a
+        // cluster is spread over the masters. Served by one node it cancels
+        // that node's share and silently leaves the rest, so every master
+        // hears it.
+        return Ok(if command.num_args() > 0 {
+            Route::PubSubByChannel
+        } else {
+            Route::AllShards
+        });
+    }
+
+    if is_shard_unsubscribe_spanning_slots(command) {
+        return Ok(Route::MultiShard);
+    }
+
+    match command.request_policy() {
+        Some(RequestPolicy::AllNodes) => Ok(Route::AllNodes),
+        Some(RequestPolicy::AllShards) => Ok(Route::AllShards),
+        Some(RequestPolicy::MultiShard) => Ok(Route::MultiShard),
+        Some(RequestPolicy::Special) => Err(Error::from(ClientError::CommandNotSupportedInCluster)
+            .with_command(command.name_bytes())),
+        None => {
+            let mut slots = command.slots();
+            if let Some(first_slot) = slots.next()
+                && !slots.all(|slot| slot == first_slot)
+            {
+                return Err(
+                    Error::from(ClientError::MismatchedKeySlots).with_command(command.name_bytes())
+                );
+            }
+            Ok(Route::SingleShard)
+        }
+    }
+}
+
 /// One shard's slice of a multi-key command: the keys of a single slot, the node
 /// that serves them, and whether the send must be prefixed with an `ASKING`.
 #[derive(Debug, PartialEq, Eq)]
@@ -383,6 +460,13 @@ impl ClusterConnection {
             return Ok(());
         }
 
+        #[cfg(test)]
+        if let Some(hook) = &self.test_hook
+            && hook.take_feed_error_for(command.name())
+        {
+            return Err(Error::from(ClientError::ClusterConfig));
+        }
+
         // The skip travels with the command it silences, on every node that command
         // reached. It applies to nothing further — including when the routing below
         // fails, where it never reached a node at all and the handler has already
@@ -554,46 +638,31 @@ impl ClusterConnection {
     ) -> Result<()> {
         trace!("Analyzing command {command:?}");
 
-        if is_broadcast_pub_sub_command(command) {
-            return if command.num_args() > 0 {
-                self.request_policy_pub_sub(command).await
-            } else {
-                // A channel-less UNSUBSCRIBE (or PUNSUBSCRIBE) names nothing to
-                // hash, and cancels every subscription the *connection* holds —
-                // which in a cluster is spread over the masters. Served by one
-                // node it cancels that node's share and silently leaves the rest,
-                // so every master hears it.
-                self.request_policy_all_shards(command).await
-            };
-        }
-
-        if is_shard_unsubscribe_spanning_slots(command) {
-            return self.request_policy_multi_shard(command, ask_reasons).await;
-        }
-
-        let request_policy = command.request_policy();
-
-        if let Some(request_policy) = request_policy {
-            match request_policy {
-                RequestPolicy::AllNodes => {
-                    self.request_policy_all_nodes(command).await?;
-                }
-                RequestPolicy::AllShards => {
-                    self.request_policy_all_shards(command).await?;
-                }
-                RequestPolicy::MultiShard => {
-                    self.request_policy_multi_shard(command, ask_reasons)
-                        .await?;
-                }
-                RequestPolicy::Special => {
-                    self.request_policy_special(command)?;
-                }
-            }
-        } else {
-            self.no_request_policy(command, ask_reasons).await?;
+        match route(command)? {
+            Route::PubSubByChannel => self.request_policy_pub_sub(command).await,
+            Route::AllShards => self.request_policy_all_shards(command).await,
+            Route::AllNodes => self.request_policy_all_nodes(command).await,
+            Route::MultiShard => self.request_policy_multi_shard(command, ask_reasons).await,
+            Route::SingleShard => self
+                .no_request_policy(command, ask_reasons)
+                .await
+                .map(|_| ()),
         }
+    }
 
-        Ok(())
+    /// Refuses, before anything is recorded or written, a command this
+    /// connection could never route: one whose keys span several slots where a
+    /// single slot is required, or one the cluster does not support at all.
+    ///
+    /// The refusal depends on the command alone, so the network handler asks
+    /// before it files any pub/sub bookkeeping or feeds any command of a
+    /// message. Refused at feed time instead, a SUBSCRIBE left its pending
+    /// batch at the head of the router's queue — where the confirmation of
+    /// every later subscription was matched against it — and the commands of a
+    /// pipeline fed ahead of the refused one were already written, their
+    /// replies due to callers that had been answered with the error.
+    pub(crate) fn check_routable(command: &Command) -> Result<()> {
+        route(command).map(|_| ())
     }
 
     #[inline]
@@ -821,10 +890,6 @@ impl ClusterConnection {
         Ok(())
     }
 
-    fn request_policy_special(&mut self, _command: &Command) -> Result<()> {
-        Err(Error::from(ClientError::CommandNotSupportedInCluster))
-    }
-
     /// A pending request is orphaned once one of its still-unresolved
     /// sub-requests targets a node that is no longer part of the cluster: a
     /// topology refresh removed that node, and its connection died with it, so

src/network/connection.rs

@@ -186,6 +186,16 @@ impl Connection {
         matches!(self, Connection::Cluster(_))
     }
 
+    /// Refuses a command this connection could never route, before anything
+    /// is recorded or written for it. Only a cluster refuses anything: see
+    /// [`ClusterConnection::check_routable`].
+    pub(crate) fn check_routable(&self, command: &Command) -> Result<()> {
+        match self {
+            Connection::Cluster(_) => ClusterConnection::check_routable(command),
+            Connection::Standalone(_) | Connection::Sentinel(_) => Ok(()),
+        }
+    }
+
     pub(crate) fn tag(&self) -> Arc<str> {
         match self {
             Connection::Standalone(connection) => connection.tag(),

src/network/network_handler.rs

@@ -349,6 +349,21 @@ impl NetworkHandler {
     fn handle_message(&mut self, mut msg: Message) {
         trace!("[{:?}] Will handle message: {msg:?}", self.mode);
 
+        // First of all: a command the connection can never route is refused
+        // before it registers any pub/sub bookkeeping or reaches the queue. Left
+        // to fail at feed time, a refused SUBSCRIBE kept its pending batch at
+        // the head of the router's queue, against which every later
+        // confirmation was then matched, and the commands of a pipeline fed
+        // ahead of the refused one were written with nobody left to read
+        // their replies.
+        if let Err(e) = msg
+            .commands()
+            .try_for_each(|command| self.connection.check_routable(command))
+        {
+            msg.send_error(e);
+            return;
+        }
+
         // Ahead of the shedding below: an UNSUBSCRIBE answered here is never
         // queued, so it costs the queue nothing and must not be refused.
         if self.mode.is_disconnected() && self.unsubscribe_while_disconnected(&mut msg) {
@@ -598,6 +613,7 @@ impl NetworkHandler {
             }
 
             let mut num_commands_to_receive: usize = 0;
+            let mut feed_error = None;
 
             // Commands are fed one by one on purpose. Batching them into a single
             // call (to hoist the stream-variant `match` out of the loop and issue
@@ -664,12 +680,28 @@ impl NetworkHandler {
                 }
 
                 if let Err(e) = self.connection.feed(command, &retry_reasons).await {
-                    error!("Feed error: {e}");
-                    msg.send_error(e);
-                    return;
+                    feed_error = Some(e);
+                    break;
                 }
             }
 
+            // The message is failed on its own: the rest of the batch is still
+            // fed and flushed below. Returning here instead stranded every
+            // message behind it in the send queue until some later message
+            // happened to wake the send path — a caller awaiting one of them
+            // waited forever on an otherwise idle client — and kept the failed
+            // message's byte charge, which nothing released any more.
+            //
+            // Routing refusals never get this far, see `check_routable`: what
+            // fails here is the connection itself, or a topology that covers no
+            // node for a slot.
+            if let Some(e) = feed_error {
+                error!("Feed error: {e}");
+                self.queue.release(cost);
+                msg.send_error(e);
+                continue;
+            }
+
             if num_commands_to_receive > 0 {
                 self.queue.await_reply(msg, num_commands_to_receive, cost);
             } else {

src/tests/cluster_server.rs

@@ -1460,16 +1460,18 @@ async fn a_read_goes_to_a_replica_when_the_read_preference_asks_for_it() -> Resu
     client.set("clu_read_replica_key", "value").await?;
 
     // The write is asynchronous, so the value may need a moment to reach the
-    // replica the read is now routed to.
-    let mut value = String::new();
+    // replica the read is now routed to. Read as an `Option`: until it does, the
+    // replica answers nil, which a `String` refuses with `UnexpectedNil` — and
+    // the `?` then ended the test on the first attempt instead of retrying.
+    let mut value = None;
     for _ in 0..20 {
-        value = client.get("clu_read_replica_key").await?;
-        if value == "value" {
+        value = client.get::<Option<String>>("clu_read_replica_key").await?;
+        if value.as_deref() == Some("value") {
             break;
         }
         sleep(Duration::from_millis(50)).await;
     }
-    assert_eq!("value", value);
+    assert_eq!(Some("value"), value.as_deref());
 
     assert!(
         any_node_ran(&replica_ports, name, "get").await?,
@@ -1706,3 +1708,77 @@ async fn a_redirection_spends_the_attempt_budget() -> Result<()> {
 
     Ok(())
 }
+
+/// A command whose feed fails is failed on its own: the messages queued behind
+/// it in the same send batch are still written. The handler used to return on
+/// the first feed error, which left them in the send queue until some later
+/// message woke the send path, so a caller awaiting one of them on an
+/// otherwise idle client waited forever.
+#[tokio::test]
+#[serial]
+async fn a_feed_error_does_not_strand_the_rest_of_the_batch() -> Result<()> {
+    crate::tests::log_try_init();
+
+    let cluster_hook = ClusterTestHook::new();
+
+    let host = get_default_host();
+    let mut config =
+        format!("redis+cluster://{host}:7000,{host}:7001,{host}:7002").into_config()?;
+    config.cluster_test_hook = Some(cluster_hook.clone());
+    let client = Client::connect(config).await?;
+
+    cluster_hook.arm_feed_error_for("GET");
+
+    // Queued synchronously, then the awaited SET right behind it: the test
+    // runtime is single-threaded, so the network task picks both up in one
+    // send batch, the failing GET first.
+    client.send_and_forget(cmd("GET").key("feed_error_batch{1}"), None)?;
+    timeout(
+        Duration::from_secs(5),
+        TimeoutKind::Command,
+        client.set("feed_error_batch{1}", "value").into_future(),
+    )
+    .await??;
+
+    let value: Option<String> = client.get("feed_error_batch{1}").await?;
+    assert_eq!(Some("value".to_owned()), value);
+
+    Ok(())
+}
+
+/// A pipeline holding a command the cluster cannot route is refused as a
+/// whole, before any of its commands is written. Refused at feed time, the
+/// commands ahead of the refused one had already been written: their replies
+/// came back with no caller waiting for them and were handed to the next one.
+#[tokio::test]
+#[serial]
+async fn a_pipeline_with_a_cross_slot_command_runs_nothing() -> Result<()> {
+    crate::tests::log_try_init();
+
+    let client = get_cluster_test_client().await?;
+    client.del("refused_pipeline{1}").await?;
+
+    let mut pipeline = client.create_pipeline();
+    pipeline.queue_command(cmd("SET").key("refused_pipeline{1}").arg("written"));
+    pipeline.queue_command(
+        cmd("SUNION")
+            .key("refused_pipeline{a}")
+            .key("refused_pipeline{b}"),
+    );
+    let refused: Result<()> = pipeline.execute().await;
+    assert!(matches!(
+        refused.unwrap_err().kind(),
+        ErrorKind::Client(ClientError::MismatchedKeySlots)
+    ));
+
+    // Nothing ran, and the next reply belongs to the next caller.
+    let value: Option<String> = timeout(
+        Duration::from_secs(5),
+        TimeoutKind::Command,
+        client.get("refused_pipeline{1}").into_future(),
+    )
+    .await??;
+    assert_eq!(None, value);
+
+    Ok(())
+}