Add AZ_AFFINITY_ALL_NODES read policy

CHANGELOG.md

@@ -21,6 +21,7 @@
 
 ### Changes
 
+* Core, Python: Add support to AZAffinityAllNodes read strategy ([#6653](https://github.com/valkey-io/valkey-glide/issues/6653))
 * Java: Add `GlideString.asReadOnlyByteBuffer()` for zero-copy, read-only access to binary payloads  ([#6600](https://github.com/valkey-io/valkey-glide/issues/6600))
 * Core: Zero-copy receive path for GET/MGET ([#6559](https://github.com/valkey-io/valkey-glide/pull/6559))
 * Go: Expose `inflightRequestsLimit` configuration via `WithInflightRequestsLimit`, bringing the Go client to parity with Java, Python, and Node ([#6385](https://github.com/valkey-io/valkey-glide/issues/6385))

ffi/src/lib.rs

@@ -1636,7 +1636,7 @@ pub unsafe extern "C-unwind" fn create_client(
 ///   - `cluster_mode_enabled`: Boolean for cluster mode (bool)
 ///   - `refresh_topology_from_initial_nodes`: When cluster mode is enabled, refresh topology using only the initial seed nodes (bool)
 ///   - `protocol`: Protocol version - "RESP2" or "RESP3" (string)
-///   - `read_from`: Read routing - "Primary", "PreferReplica", "LowestLatency", "AZAffinity", or "AZAffinityReplicasAndPrimary" (string)
+///   - `read_from`: Read routing - "Primary", "PreferReplica", "LowestLatency", "AZAffinity", "AZAffinityReplicasAndPrimary", "AllNodes", or "AZAffinityAllNodes" (string)
 ///   - `connection_retry_strategy`: Retry configuration with `number_of_retries`, `factor`, `exponent_base`, and optional `jitter_percent` (object)
 ///   - `root_certs`: Array of PEM-encoded CA certificates for TLS (array of strings)
 ///   - `client_az`: Client availability zone for AZ affinity routing (string)
@@ -2149,6 +2149,8 @@ fn apply_json_options(
             "AZAffinityReplicasAndPrimary" => {
                 connection_request::ReadFrom::AZAffinityReplicasAndPrimary
             }
+            "AllNodes" => connection_request::ReadFrom::AllNodes,
+            "AZAffinityAllNodes" => connection_request::ReadFrom::AZAffinityAllNodes,
             _ => return Err(format!("Unknown read_from value: {}", read_from_str)),
         };
         request.read_from = ::protobuf::EnumOrUnknown::new(read_from_enum);

ffi/tests/test_create_client_from_uri.rs

@@ -786,6 +786,8 @@ fn test_create_client_from_uri_valid_formats(#[case] uri_format: &str) {
 // #[case("LowestLatency")] // TODO: Not yet implemented in glide-core
 #[case("AZAffinity")]
 #[case("AZAffinityReplicasAndPrimary")]
+#[case("AllNodes")]
+#[case("AZAffinityAllNodes")]
 fn test_create_client_from_uri_all_read_from_values(#[case] read_from: &str) {
     let server = Server::new();
     let uri = CString::new(format!("redis://127.0.0.1:{}", server.port)).unwrap();

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

@@ -147,7 +147,7 @@ where
 async fn setup_connection<C>(
     connection_info: &RedisConnectionInfo,
     con: &mut C,
-    // This parameter is set to 'true' if ReadFromReplica strategy is set to AZAffinity or AZAffinityReplicasAndPrimary.
+    // This parameter is set to 'true' if ReadFromReplica strategy is set to AZAffinity, AZAffinityReplicasAndPrimary, or AZAffinityAllNodes.
     // An INFO command will be triggered in the connection's setup to update the 'availability_zone' property.
     discover_az: bool,
 ) -> RedisResult<()>

glide-core/redis-rs/redis/src/client.rs

@@ -88,7 +88,7 @@ pub struct GlideConnectionOptions {
     #[cfg(feature = "aio")]
     /// Passive disconnect notifier
     pub disconnect_notifier: Option<Box<dyn DisconnectNotifier>>,
-    /// If ReadFromReplica strategy is set to AZAffinity or AZAffinityReplicasAndPrimary, this parameter will be set to 'true'.
+    /// If ReadFromReplica strategy is set to AZAffinity, AZAffinityReplicasAndPrimary, or AZAffinityAllNodes, this parameter will be set to 'true'.
     /// In this case, an INFO command will be triggered in the connection's setup to update the connection's 'availability_zone' property.
     pub discover_az: bool,
     /// Connection timeout duration.

glide-core/redis-rs/redis/src/cluster_async/connections_container.rs

@@ -464,6 +464,56 @@ where
         self.get_connection_by_az_affinity_strategy(slot_map_value, client_az, true)
     }
 
+    /// Returns a connection to a node (primary or replica) in the same availability zone
+    /// as `client_az`, rotating equally among all in-AZ nodes in a round robin manner.
+    /// Falls back to a round robin across all nodes if no in-AZ node is available.
+    pub(crate) fn round_robin_read_from_all_nodes_with_az_awareness(
+        &self,
+        slot_map_value: &SlotMapValue,
+        client_az: String,
+    ) -> Option<ConnectionAndAddress<Connection>> {
+        let addrs = &slot_map_value.addrs;
+        let total_nodes = addrs.replicas().len() + 1; // primary + replicas, index 0 = primary
+                                                      // `last_used_replica` is shared with the replica-only rotations, as in the AllNodes strategy.
+        let initial_index = slot_map_value.last_used_replica.load(Ordering::Relaxed);
+        let mut check_count = 0;
+
+        loop {
+            check_count += 1;
+
+            // Looped through all nodes; no connected node found in the same availability zone.
+            if check_count > total_nodes {
+                break;
+            }
+
+            let index = (initial_index + check_count) % total_nodes;
+            let node_address = if index == 0 {
+                addrs.primary()
+            } else {
+                addrs.replicas()[index - 1].clone()
+            };
+
+            // Check if this node's availability zone matches the user's availability zone.
+            if let Some((address, connection_details)) =
+                self.connection_details_for_address(node_address.as_str())
+            {
+                if self.az_for_address(&address) == Some(client_az.clone()) {
+                    // Attempt to update `last_used_replica` with the index of this node.
+                    let _ = slot_map_value.last_used_replica.compare_exchange_weak(
+                        initial_index,
+                        index,
+                        Ordering::Relaxed,
+                        Ordering::Relaxed,
+                    );
+                    return Some((address, connection_details.conn));
+                }
+            }
+        }
+
+        // Fall back to any available node (primary or replica) using round-robin.
+        self.round_robin_read_from_all_nodes(slot_map_value)
+    }
+
     fn get_connection_by_az_affinity_strategy(
         &self,
         slot_map_value: &SlotMapValue,
@@ -549,6 +599,11 @@ where
                         slot_map_value,
                         az.to_string(),
                     ),
+                ReadFromReplicaStrategy::AZAffinityAllNodes(az) => self
+                    .round_robin_read_from_all_nodes_with_az_awareness(
+                        slot_map_value,
+                        az.to_string(),
+                    ),
             },
             // when the user strategy per command is replica_preffered
             SlotAddr::ReplicaRequired => match &self.read_from_replica_strategy {
@@ -562,6 +617,13 @@ where
                         slot_map_value,
                         az.to_string(),
                     ),
+                // Explicit replica routes stay in the replica rotation for this strategy:
+                // in-AZ replicas first, then any replica, primary only if no replica is connected.
+                ReadFromReplicaStrategy::AZAffinityAllNodes(az) => self
+                    .round_robin_read_from_replica_with_az_awareness(
+                        slot_map_value,
+                        az.to_string(),
+                    ),
                 _ => self.round_robin_read_from_replica(slot_map_value),
             },
         }
@@ -1326,6 +1388,129 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn get_connection_for_az_affinity_all_nodes_route_round_robin() {
+        // Create a container with AZAffinityAllNodes strategy
+        let container: ConnectionsContainer<usize> = create_container_with_az_strategy(
+            false,
+            Some(ReadFromReplicaStrategy::AZAffinityAllNodes(
+                "use-1a".to_string(),
+            )),
+        );
+
+        // Set the primary of the slot to the client's AZ
+        container
+            .connection_map
+            .get_mut("primary3")
+            .unwrap()
+            .user_connection
+            .az = Some("use-1a".to_string());
+
+        // The primary and the two in-AZ replicas should be rotated equally
+        let mut addresses: Vec<usize> = (0..6)
+            .map(|_| {
+                container
+                    .connection_for_route(&Route::new(2001, SlotAddr::ReplicaOptional))
+                    .unwrap()
+                    .1
+            })
+            .collect();
+        addresses.sort();
+        assert_eq!(addresses, vec![3, 3, 31, 31, 33, 33]);
+    }
+
+    #[test]
+    fn get_connection_for_az_affinity_all_nodes_route() {
+        // Create a container with AZAffinityAllNodes strategy
+        let container: ConnectionsContainer<usize> = create_container_with_az_strategy(
+            false,
+            Some(ReadFromReplicaStrategy::AZAffinityAllNodes(
+                "use-1a".to_string(),
+            )),
+        );
+
+        // Set the primary of the slot to the client's AZ
+        container
+            .connection_map
+            .get_mut("primary3")
+            .unwrap()
+            .user_connection
+            .az = Some("use-1a".to_string());
+
+        // Slot number does not exist (slot 1001 wasn't assigned to any primary)
+        assert!(container
+            .connection_for_route(&Route::new(1001, SlotAddr::ReplicaOptional))
+            .is_none());
+
+        // Get one of the in-AZ nodes (primary or replica) for slot 2001
+        assert!(one_of(
+            container.connection_for_route(&Route::new(2001, SlotAddr::ReplicaOptional)),
+            &[3, 31, 33],
+        ));
+
+        // Explicitly replica-routed commands must not be served by the in-AZ primary
+        let mut addresses: Vec<usize> = (0..4)
+            .map(|_| {
+                container
+                    .connection_for_route(&Route::new(2001, SlotAddr::ReplicaRequired))
+                    .unwrap()
+                    .1
+            })
+            .collect();
+        addresses.sort();
+        assert_eq!(addresses, vec![31, 31, 33, 33]);
+
+        // Remove the in-AZ replicas; the in-AZ primary should now get all reads
+        remove_nodes(&container, &["replica3-1", "replica3-3"]);
+        for _ in 0..3 {
+            assert_eq!(
+                3,
+                container
+                    .connection_for_route(&Route::new(2001, SlotAddr::ReplicaOptional))
+                    .unwrap()
+                    .1
+            );
+        }
+
+        // Move the primary out of the client's AZ; with no in-AZ node left,
+        // fall back to a round robin across all remaining nodes
+        container
+            .connection_map
+            .get_mut("primary3")
+            .unwrap()
+            .user_connection
+            .az = Some("use-1b".to_string());
+        let mut addresses: Vec<usize> = (0..4)
+            .map(|_| {
+                container
+                    .connection_for_route(&Route::new(2001, SlotAddr::ReplicaOptional))
+                    .unwrap()
+                    .1
+            })
+            .collect();
+        addresses.sort();
+        assert_eq!(addresses, vec![3, 3, 32, 32]);
+
+        // Write commands should still be routed to the primary
+        assert_eq!(
+            3,
+            container
+                .connection_for_route(&Route::new(2001, SlotAddr::Master))
+                .unwrap()
+                .1
+        );
+
+        // With no replica connected, an explicit replica route falls back to the primary
+        remove_nodes(&container, &["replica3-2"]);
+        assert_eq!(
+            3,
+            container
+                .connection_for_route(&Route::new(2001, SlotAddr::ReplicaRequired))
+                .unwrap()
+                .1
+        );
+    }
+
     #[test]
     fn get_connection_by_address() {
         let container = create_container();

glide-core/redis-rs/redis/src/cluster_async/connections_logic.rs

@@ -180,6 +180,7 @@ where
         params.read_from_replicas,
         crate::cluster_slotmap::ReadFromReplicaStrategy::AZAffinity(_)
             | crate::cluster_slotmap::ReadFromReplicaStrategy::AZAffinityReplicasAndPrimary(_)
+            | crate::cluster_slotmap::ReadFromReplicaStrategy::AZAffinityAllNodes(_)
     );
 
     match create_connection::<C>(

glide-core/redis-rs/redis/src/cluster_async/mod.rs

@@ -1564,6 +1564,7 @@ where
             cluster_params.read_from_replicas,
             crate::cluster_slotmap::ReadFromReplicaStrategy::AZAffinity(_)
                 | crate::cluster_slotmap::ReadFromReplicaStrategy::AZAffinityReplicasAndPrimary(_)
+                | crate::cluster_slotmap::ReadFromReplicaStrategy::AZAffinityAllNodes(_)
         );
 
         let connection_retry_strategy = cluster_params.reconnect_retry_strategy.unwrap_or_default();

glide-core/redis-rs/redis/src/cluster_client.rs

@@ -474,6 +474,8 @@ impl ClusterClientBuilder {
     /// If no suitable replica is found (i.e. no replica could be found in the requested availability zone), choose any replica. Falling back to primary if needed.
     /// `ReadFromReplicaStrategy::AZAffinityReplicasAndPrimary(availability_zone)` - attempt to access nodes in the same availability zone.
     ///  prioritizing local replicas, then the local primary, and falling back to any replica or the primary if needed.
+    /// `ReadFromReplicaStrategy::AZAffinityAllNodes(availability_zone)` - spread read requests equally among all nodes (primary and replicas)
+    ///  in the same availability zone, falling back to a round robin across all nodes if no node is available in that zone.
     /// `ReadFromReplicaStrategy::RoundRobin` - reads are distributed across replicas for load balancing using round-robin algorithm. Falling back to primary if needed.
     /// `ReadFromReplicaStrategy::AlwaysFromPrimary` ensures all read and write queries are directed to the primary node.
     ///

glide-core/redis-rs/redis/src/cluster_slotmap.rs

@@ -44,6 +44,10 @@ pub enum ReadFromReplicaStrategy {
     AZAffinityReplicasAndPrimary(String),
     /// Spread the read requests between all nodes (primary and replicas) in a round robin manner.
     AllNodes,
+    /// Spread the read requests equally among all nodes (primary and replicas) within the client's
+    /// Availability Zone (AZ) in a round robin manner, falling back to a round robin across all
+    /// nodes if no node in the client's AZ is available.
+    AZAffinityAllNodes(String),
 }
 
 #[derive(Debug, Default)]
@@ -91,6 +95,14 @@ fn get_address_from_slot(
         // behavior of these strategies when no local node is known.
         ReadFromReplicaStrategy::AZAffinity(_az) => round_robin_replica(),
         ReadFromReplicaStrategy::AZAffinityReplicasAndPrimary(_az) => round_robin_all_nodes(),
+        // Explicit replica routes stay in the replica rotation, matching this
+        // strategy's `lookup_route` behavior.
+        ReadFromReplicaStrategy::AZAffinityAllNodes(_az)
+            if slot_addr == SlotAddr::ReplicaRequired =>
+        {
+            round_robin_replica()
+        }
+        ReadFromReplicaStrategy::AZAffinityAllNodes(_az) => round_robin_all_nodes(),
     }
 }
 
@@ -927,6 +939,54 @@ mod tests_cluster_slotmap {
         );
     }
 
+    #[test]
+    fn test_slot_map_az_affinity_all_nodes_falls_back_to_all_nodes() {
+        let slot_map = get_slot_map(ReadFromReplicaStrategy::AZAffinityAllNodes(
+            "zone-a".to_string(),
+        ));
+        let route = Route::new(2001, SlotAddr::ReplicaOptional);
+        let mut addresses = vec![
+            slot_map.slot_addr_for_route(&route).unwrap(),
+            slot_map.slot_addr_for_route(&route).unwrap(),
+            slot_map.slot_addr_for_route(&route).unwrap(),
+            slot_map.slot_addr_for_route(&route).unwrap(),
+        ];
+        addresses.sort();
+        assert_eq!(
+            addresses,
+            vec![
+                "node3:6379",
+                "replica4:6379",
+                "replica5:6379",
+                "replica6:6379"
+            ]
+            .into_iter()
+            .map(|s| Arc::new(s.to_string()))
+            .collect::<Vec<_>>()
+        );
+    }
+
+    #[test]
+    fn test_slot_map_az_affinity_all_nodes_replica_required_stays_on_replicas() {
+        let slot_map = get_slot_map(ReadFromReplicaStrategy::AZAffinityAllNodes(
+            "zone-a".to_string(),
+        ));
+        let route = Route::new(2001, SlotAddr::ReplicaRequired);
+        let mut addresses = vec![
+            slot_map.slot_addr_for_route(&route).unwrap(),
+            slot_map.slot_addr_for_route(&route).unwrap(),
+            slot_map.slot_addr_for_route(&route).unwrap(),
+        ];
+        addresses.sort();
+        assert_eq!(
+            addresses,
+            vec!["replica4:6379", "replica5:6379", "replica6:6379"]
+                .into_iter()
+                .map(|s| Arc::new(s.to_string()))
+                .collect::<Vec<_>>()
+        );
+    }
+
     #[test]
     fn test_get_slots_of_node() {
         let slot_map = get_slot_map(ReadFromReplicaStrategy::AlwaysFromPrimary);