From 0b3af53aa0d2f5660fff25e755b8333bc8291d66 Mon Sep 17 00:00:00 2001 From: mcatanzariti Date: Sat, 14 Nov 2026 04:29:28 +0000 Subject: [PATCH 1/2] fix(cluster): refuse an unroutable command before recording anything A command the cluster connection cannot route (keys spanning several slots where one is required, or a command with the `special` request policy) was refused only when the network task fed it, after the handler had already recorded its bookkeeping and fed the commands ahead of it: - a refused SSUBSCRIBE left its pending batch at the head of the router's subscription queue; the next subscription's confirmation was matched against it and its caller was never answered; - in a pipeline, the commands fed ahead of the refused one were written and their replies were handed to the next callers; - `send_messages` returned on the error, stranding every message behind it in the send queue until something else woke the send path, and leaking the refused message's byte charge from the send-queue budget. The routing decision now lives in one classifier, `route`, read both by the feed path and by `check_routable`, which the handler calls before anything is recorded or queued. A feed that still fails (a broken node connection, a slot no node covers) fails its own message, releases its charge, and lets the rest of the batch be fed and flushed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Jo8jBqM8KTsnpJz9oRkK5G --- CHANGELOG.md | 18 ++++ src/network/cluster_connection.rs | 149 +++++++++++++++++++++--------- src/network/connection.rs | 10 ++ src/network/network_handler.rs | 38 +++++++- src/tests/cluster_server.rs | 74 +++++++++++++++ src/tests/pub_sub_commands.rs | 37 ++++++++ 6 files changed, 281 insertions(+), 45 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d5cff12..310f5e3 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/network/cluster_connection.rs b/src/network/cluster_connection.rs index 5346f48..7d4284e 100644 --- a/src/network/cluster_connection.rs +++ b/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, + /// 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>>, } #[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 { 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 { + 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 diff --git a/src/network/connection.rs b/src/network/connection.rs index 124ea7e..6f2963d 100644 --- a/src/network/connection.rs +++ b/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 { match self { Connection::Standalone(connection) => connection.tag(), diff --git a/src/network/network_handler.rs b/src/network/network_handler.rs index 5cf1c7d..200b7da 100644 --- a/src/network/network_handler.rs +++ b/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 { diff --git a/src/tests/cluster_server.rs b/src/tests/cluster_server.rs index 8a12073..ebe2bfb 100644 --- a/src/tests/cluster_server.rs +++ b/src/tests/cluster_server.rs @@ -1706,3 +1706,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 = 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 = timeout( + Duration::from_secs(5), + TimeoutKind::Command, + client.get("refused_pipeline{1}").into_future(), + ) + .await??; + assert_eq!(None, value); + + Ok(()) +} diff --git a/src/tests/pub_sub_commands.rs b/src/tests/pub_sub_commands.rs index 8980bae..34ce55f 100644 --- a/src/tests/pub_sub_commands.rs +++ b/src/tests/pub_sub_commands.rs @@ -1596,3 +1596,40 @@ async fn closing_a_shard_stream_spanning_slots_succeeds_in_cluster() -> Result<( Ok(()) } + +/// A SSUBSCRIBE refused before being sent -- its shard channels span two +/// slots -- must leave nothing behind: the client's next subscriptions are +/// confirmed and receive their messages. +#[tokio::test] +#[serial] +async fn a_refused_cross_slot_ssubscribe_leaves_the_client_usable() -> Result<()> { + log_try_init(); + + let client = get_cluster_test_client().await?; + let publisher = get_cluster_test_client().await?; + + let refused = client + .ssubscribe(["refused_ssubscribe{a}", "refused_ssubscribe{b}"]) + .await; + assert!(matches!( + refused.map(|_| ()).unwrap_err().kind(), + ErrorKind::Client(ClientError::MismatchedKeySlots) + )); + + let mut stream = timeout( + Duration::from_secs(5), + TimeoutKind::Command, + client.ssubscribe("refused_ssubscribe{a}"), + ) + .await??; + publisher + .spublish("refused_ssubscribe{a}", "payload") + .await?; + let message = timeout(Duration::from_secs(5), TimeoutKind::Command, stream.next()) + .await? + .unwrap()?; + assert_eq!(b"payload", message.payload()); + + stream.close().await?; + Ok(()) +} From b2d29ecd3a355b3cde8211511e2fa5c1d66ce514 Mon Sep 17 00:00:00 2001 From: mcatanzariti Date: Sat, 14 Nov 2026 04:44:17 +0000 Subject: [PATCH 2/2] test(cluster): let the replica-read test actually wait for replication `a_read_goes_to_a_replica_when_the_read_preference_asks_for_it` polls the replica until the value written to the master has reached it, but read it as a `String`. Until replication lands, the replica answers nil, which a `String` refuses with `UnexpectedNil`, and the `?` ended the test on the first attempt: the retry loop never ran. Seen on CI as `Received response Null` followed by the failure. The value is now read as an `Option`. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Jo8jBqM8KTsnpJz9oRkK5G --- src/tests/cluster_server.rs | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/src/tests/cluster_server.rs b/src/tests/cluster_server.rs index ebe2bfb..7583010 100644 --- a/src/tests/cluster_server.rs +++ b/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::>("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?,