Fixed non key based commands not to be routed to a random node (#174)

Author: barshaul

redis/src/cluster_async/mod.rs

@@ -29,6 +29,7 @@ pub mod testing {
     pub use super::connections_logic::*;
 }
 use crate::{
+    cluster_routing::{Routable, RoutingInfo},
     cluster_slotmap::SlotMap,
     cluster_topology::SLOT_SIZE,
     cmd,
@@ -1765,7 +1766,7 @@ where
 
         // if we reached this point, we're sending the command only to single node, and we need to find the
         // right connection to the node.
-        let (address, mut conn) = Self::get_connection(routing, core)
+        let (address, mut conn) = Self::get_connection(routing, core, Some(cmd.clone()))
             .await
             .map_err(|err| (OperationTarget::NotFound, err))?;
         conn.req_packed_command(&cmd)
@@ -1801,7 +1802,7 @@ where
                     pipeline,
                     offset,
                     count,
-                    Self::get_connection(route, core),
+                    Self::get_connection(route, core, None),
                 )
                 .await
             }
@@ -1825,6 +1826,7 @@ where
     async fn get_connection(
         routing: InternalSingleNodeRouting<C>,
         core: Core<C>,
+        cmd: Option<Arc<Cmd>>,
     ) -> RedisResult<(ArcStr, C)> {
         let read_guard = core.conn_lock.read().await;
         let mut asking = false;
@@ -1849,15 +1851,32 @@ where
                     ConnectionCheck::Found,
                 )
             }
-            // This means that a request routed to a route without a matching connection will be sent to a random node, hopefully to be redirected afterwards.
             InternalSingleNodeRouting::SpecificNode(route) => {
-                read_guard.connection_for_route(&route).map_or_else(
-                    || {
-                        warn!("No connection found for route `{route:?}");
-                        ConnectionCheck::RandomConnection
-                    },
-                    ConnectionCheck::Found,
-                )
+                match read_guard.connection_for_route(&route) {
+                    Some((conn, address)) => ConnectionCheck::Found((conn, address)),
+                    None => {
+                        // No connection is found for the given route:
+                        // - For key-based commands, attempt redirection to a random node,
+                        //   hopefully to be redirected afterwards by a MOVED error.
+                        // - For non-key-based commands, avoid attempting redirection to a random node
+                        //   as it wouldn't result in MOVED hints and can lead to unwanted results
+                        //   (e.g., sending management command to a different node than the user asked for); instead, raise the error.
+                        let routable_cmd = cmd.and_then(|cmd| Routable::command(&*cmd));
+                        if routable_cmd.is_some()
+                            && !RoutingInfo::is_key_based_cmd(&routable_cmd.unwrap())
+                        {
+                            return Err((
+                                ErrorKind::ClusterConnectionNotFound,
+                                "Requested connection not found for route",
+                                format!("{route:?}"),
+                            )
+                                .into());
+                        } else {
+                            warn!("No connection found for route `{route:?}`. Attempting redirection to a random node.");
+                            ConnectionCheck::RandomConnection
+                        }
+                    }
+                }
             }
             InternalSingleNodeRouting::Random => ConnectionCheck::RandomConnection,
             InternalSingleNodeRouting::Connection { address, conn } => {

redis/tests/test_cluster_async.rs

@@ -4020,6 +4020,126 @@ mod cluster_async {
     .unwrap();
     }
 
+    #[test]
+    fn test_async_cluster_dont_route_to_a_random_on_non_key_based_cmd() {
+        // This test verifies that non-key-based commands do not get routed to a random node
+        // when no connection is found for the given route. Instead, the appropriate error
+        // should be raised.
+        let name = "test_async_cluster_dont_route_to_a_random_on_non_key_based_cmd";
+        let request_counter = Arc::new(AtomicU32::new(0));
+        let cloned_req_counter = request_counter.clone();
+        let MockEnv {
+            runtime,
+            async_connection: mut connection,
+            handler: _handler,
+            ..
+        } = MockEnv::with_client_builder(
+            ClusterClient::builder(vec![&*format!("redis://{name}")]),
+            name,
+            move |received_cmd: &[u8], _| {
+                let slots_config_vec = vec![
+                    MockSlotRange {
+                        primary_port: 6379,
+                        replica_ports: vec![],
+                        slot_range: (0_u16..8000_u16),
+                    },
+                    MockSlotRange {
+                        primary_port: 6380,
+                        replica_ports: vec![],
+                        // Don't cover all slots
+                        slot_range: (8001_u16..12000_u16),
+                    },
+                ];
+                respond_startup_with_config(name, received_cmd, Some(slots_config_vec), false)?;
+                // If requests are sent to random nodes, they will be caught and counted here.
+                request_counter.fetch_add(1, Ordering::Relaxed);
+                Err(Ok(Value::Nil))
+            },
+        );
+
+        runtime
+            .block_on(async move {
+                let uncovered_slot = 16000;
+                let route = redis::cluster_routing::Route::new(
+                    uncovered_slot,
+                    redis::cluster_routing::SlotAddr::Master,
+                );
+                let single_node_route =
+                    redis::cluster_routing::SingleNodeRoutingInfo::SpecificNode(route);
+                let routing = RoutingInfo::SingleNode(single_node_route);
+                let res = connection
+                    .route_command(&redis::cmd("FLUSHALL"), routing)
+                    .await;
+                assert!(res.is_err());
+                let res_err = res.unwrap_err();
+                assert_eq!(
+                    res_err.kind(),
+                    ErrorKind::ClusterConnectionNotFound,
+                    "{:?}",
+                    res_err
+                );
+                assert_eq!(cloned_req_counter.load(Ordering::Relaxed), 0);
+                Ok::<_, RedisError>(())
+            })
+            .unwrap();
+    }
+
+    #[test]
+    fn test_async_cluster_route_to_random_on_key_based_cmd() {
+        // This test verifies that key-based commands get routed to a random node
+        // when no connection is found for the given route. The command should
+        // then be redirected correctly by the server's MOVED error.
+        let name = "test_async_cluster_route_to_random_on_key_based_cmd";
+        let request_counter = Arc::new(AtomicU32::new(0));
+        let cloned_req_counter = request_counter.clone();
+        let MockEnv {
+            runtime,
+            async_connection: mut connection,
+            handler: _handler,
+            ..
+        } = MockEnv::with_client_builder(
+            ClusterClient::builder(vec![&*format!("redis://{name}")]),
+            name,
+            move |received_cmd: &[u8], _| {
+                let slots_config_vec = vec![
+                    MockSlotRange {
+                        primary_port: 6379,
+                        replica_ports: vec![],
+                        slot_range: (0_u16..8000_u16),
+                    },
+                    MockSlotRange {
+                        primary_port: 6380,
+                        replica_ports: vec![],
+                        // Don't cover all slots
+                        slot_range: (8001_u16..12000_u16),
+                    },
+                ];
+                respond_startup_with_config(name, received_cmd, Some(slots_config_vec), false)?;
+                if contains_slice(received_cmd, b"GET") {
+                    if request_counter.fetch_add(1, Ordering::Relaxed) == 0 {
+                        return Err(parse_redis_value(
+                            format!("-MOVED 12182 {name}:6380\r\n").as_bytes(),
+                        ));
+                    } else {
+                        return Err(Ok(Value::SimpleString("bar".into())));
+                    }
+                }
+                panic!("unexpected command {:?}", received_cmd);
+            },
+        );
+
+        runtime
+            .block_on(async move {
+                // The keyslot of "foo" is 12182 and it isn't covered by any node, so we expect the
+                // request to be routed to a random node and then to be redirected to the MOVED node (2 requests in total)
+                let res: String = connection.get("foo").await.unwrap();
+                assert_eq!(res, "bar".to_string());
+                assert_eq!(cloned_req_counter.load(Ordering::Relaxed), 2);
+                Ok::<_, RedisError>(())
+            })
+            .unwrap();
+    }
+
     #[cfg(feature = "tls-rustls")]
     mod mtls_test {
         use crate::support::mtls_test::create_cluster_client_from_cluster;