Skip to content

Commit 2d7200f

Browse files
barshaulbarshaul
authored
Breaking change: reconnect_to_initial_nodes will be called only when AllConnectionsUnavailable (#183)
* Fixed reconnect_to_initial_nodes not to replace the current connection map but to extend it * Breaking changes: renamed ClusterConnectionNotFound to AllConnectionsUnavailable, added new error ConnectionNotFoundForRoute
1 parent efd61ff commit 2d7200f

File tree

6 files changed

+82
-26
lines changed

6 files changed

+82
-26
lines changed

‎redis/src/cluster_async/connections_container.rs‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -137,6 +137,14 @@ where
137137
}
138138
}
139139

140+
// Extends the current connection map with the provided one
141+
pub(crate) fn extend_connection_map(
142+
&mut self,
143+
other_connection_map: ConnectionsMap<Connection>,
144+
) {
145+
self.connection_map.extend(other_connection_map.0);
146+
}
147+
140148
/// Returns true if the address represents a known primary node.
141149
pub(crate) fn is_primary(&self, address: &String) -> bool {
142150
self.connection_for_address(address).is_some()
@@ -841,4 +849,33 @@ mod tests {
841849

842850
assert!(!container.is_primary(&address));
843851
}
852+
853+
#[test]
854+
fn test_extend_connection_map() {
855+
let mut container = create_container();
856+
let mut current_addresses: Vec<_> = container
857+
.all_node_connections()
858+
.map(|conn| conn.0)
859+
.collect();
860+
861+
let new_node = "new_primary1".to_string();
862+
// Check that `new_node` not exists in the current
863+
assert!(container.connection_for_address(&new_node).is_none());
864+
// Create new connection map
865+
let new_connection_map = DashMap::new();
866+
new_connection_map.insert(new_node.clone(), create_cluster_node(1, false));
867+
868+
// Extend the current connection map
869+
container.extend_connection_map(ConnectionsMap(new_connection_map));
870+
871+
// Check that the new addresses vector contains both the new node and all previous nodes
872+
let mut new_addresses: Vec<_> = container
873+
.all_node_connections()
874+
.map(|conn| conn.0)
875+
.collect();
876+
current_addresses.push(new_node);
877+
current_addresses.sort();
878+
new_addresses.sort();
879+
assert_eq!(current_addresses, new_addresses);
880+
}
844881
}

‎redis/src/cluster_async/mod.rs‎

Lines changed: 9 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -808,7 +808,7 @@ impl<C> Future for Request<C> {
808808
let request = this.request.as_mut().unwrap();
809809
// TODO - would be nice if we didn't need to repeat this code twice, with & without retries.
810810
if request.retry >= this.retry_params.number_of_retries {
811-
let next = if err.kind() == ErrorKind::ClusterConnectionNotFound {
811+
let next = if err.kind() == ErrorKind::AllConnectionsUnavailable {
812812
Next::ReconnectToInitialNodes { request: None }.into()
813813
} else if matches!(err.retry_method(), crate::types::RetryMethod::MovedRedirect)
814814
|| matches!(target, OperationTarget::NotFound)
@@ -836,7 +836,7 @@ impl<C> Future for Request<C> {
836836
}
837837
request.retry = request.retry.saturating_add(1);
838838

839-
if err.kind() == ErrorKind::ClusterConnectionNotFound {
839+
if err.kind() == ErrorKind::AllConnectionsUnavailable {
840840
return Next::ReconnectToInitialNodes {
841841
request: Some(this.request.take().unwrap()),
842842
}
@@ -1132,12 +1132,7 @@ where
11321132
}
11331133
};
11341134
let mut write_lock = inner.conn_lock.write().await;
1135-
*write_lock = ConnectionsContainer::new(
1136-
Default::default(),
1137-
connection_map,
1138-
inner.cluster_params.read_from_replicas,
1139-
0,
1140-
);
1135+
write_lock.extend_connection_map(connection_map);
11411136
drop(write_lock);
11421137
if let Err(err) = Self::refresh_slots_and_subscriptions_with_retries(
11431138
inner.clone(),
@@ -1260,7 +1255,7 @@ where
12601255
} else {
12611256
Err(last_err.unwrap_or_else(|| {
12621257
(
1263-
ErrorKind::ClusterConnectionNotFound,
1258+
ErrorKind::AllConnectionsUnavailable,
12641259
"Couldn't find any connection",
12651260
)
12661261
.into()
@@ -1656,7 +1651,7 @@ where
16561651
return OperationResult::Err((
16571652
OperationTarget::FanOut,
16581653
(
1659-
ErrorKind::ClusterConnectionNotFound,
1654+
ErrorKind::AllConnectionsUnavailable,
16601655
"No connections found for multi-node operation",
16611656
)
16621657
.into(),
@@ -1700,7 +1695,7 @@ where
17001695
)
17011696
} else {
17021697
let _ = sender.send(Err((
1703-
ErrorKind::ClusterConnectionNotFound,
1698+
ErrorKind::ConnectionNotFoundForRoute,
17041699
"Connection not found",
17051700
)
17061701
.into()));
@@ -1871,7 +1866,7 @@ where
18711866
&& !RoutingInfo::is_key_routing_command(&routable_cmd.unwrap())
18721867
{
18731868
return Err((
1874-
ErrorKind::ClusterConnectionNotFound,
1869+
ErrorKind::ConnectionNotFoundForRoute,
18751870
"Requested connection not found for route",
18761871
format!("{route:?}"),
18771872
)
@@ -1892,7 +1887,7 @@ where
18921887
return Ok((address, conn.await));
18931888
} else {
18941889
return Err((
1895-
ErrorKind::ClusterConnectionNotFound,
1890+
ErrorKind::ConnectionNotFoundForRoute,
18961891
"Requested connection not found",
18971892
address,
18981893
)
@@ -1938,7 +1933,7 @@ where
19381933
.random_connections(1, ConnectionType::User)
19391934
.next()
19401935
.ok_or(RedisError::from((
1941-
ErrorKind::ClusterConnectionNotFound,
1936+
ErrorKind::AllConnectionsUnavailable,
19421937
"No random connection found",
19431938
)))?;
19441939
return Ok((random_address, random_conn_future.await));

‎redis/src/commands/cluster_scan.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -451,7 +451,9 @@ where
451451
// the connection to the address cant be reached from different reasons, we will check we want to check if
452452
// the problem is problem that we can recover from like failover or scale down or some network issue
453453
// that we can retry the scan command to an address that own the next slot we are at.
454-
ErrorKind::IoError | ErrorKind::ClusterConnectionNotFound => {
454+
ErrorKind::IoError
455+
| ErrorKind::AllConnectionsUnavailable
456+
| ErrorKind::ConnectionNotFoundForRoute => {
455457
let retry =
456458
retry_scan(&scan_state, &core, match_pattern, count, object_type).await?;
457459
(from_redis_value(&retry.0?)?, retry.1)

‎redis/src/types.rs‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -133,8 +133,10 @@ pub enum ErrorKind {
133133
EmptySentinelList,
134134
/// Attempted to kill a script/function while they werent' executing
135135
NotBusy,
136-
/// Used when a cluster connection cannot find a connection to a valid node.
137-
ClusterConnectionNotFound,
136+
/// Used when no valid node connections remain in the cluster connection
137+
AllConnectionsUnavailable,
138+
/// Used when a connection is not found for the specified route.
139+
ConnectionNotFoundForRoute,
138140

139141
#[cfg(feature = "json")]
140142
/// Error Serializing a struct to JSON form
@@ -875,7 +877,8 @@ impl RedisError {
875877
ErrorKind::NoValidReplicasFoundBySentinel => "no valid replicas found by sentinel",
876878
ErrorKind::EmptySentinelList => "empty sentinel list",
877879
ErrorKind::NotBusy => "not busy",
878-
ErrorKind::ClusterConnectionNotFound => "connection to node in cluster not found",
880+
ErrorKind::AllConnectionsUnavailable => "no valid connections remain in the cluster",
881+
ErrorKind::ConnectionNotFoundForRoute => "No connection found for the requested route",
879882
#[cfg(feature = "json")]
880883
ErrorKind::Serialize => "serializing",
881884
ErrorKind::RESP3NotSupported => "resp3 is not supported by server",
@@ -1046,7 +1049,8 @@ impl RedisError {
10461049

10471050
ErrorKind::ParseError => RetryMethod::Reconnect,
10481051
ErrorKind::AuthenticationFailed => RetryMethod::Reconnect,
1049-
ErrorKind::ClusterConnectionNotFound => RetryMethod::Reconnect,
1052+
ErrorKind::AllConnectionsUnavailable => RetryMethod::Reconnect,
1053+
ErrorKind::ConnectionNotFoundForRoute => RetryMethod::Reconnect,
10501054

10511055
ErrorKind::IoError => match &self.repr {
10521056
ErrorRepr::IoError(err) => match err.kind() {

‎redis/tests/test_cluster_async.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2400,7 +2400,7 @@ mod cluster_async {
24002400
.block_on(cmd.query_async::<_, Vec<String>>(&mut connection))
24012401
.unwrap_err();
24022402
assert!(
2403-
matches!(result.kind(), ErrorKind::ClusterConnectionNotFound)
2403+
matches!(result.kind(), ErrorKind::ConnectionNotFoundForRoute)
24042404
|| result.is_connection_dropped()
24052405
);
24062406
}
@@ -4031,7 +4031,7 @@ mod cluster_async {
40314031
handler: _handler,
40324032
..
40334033
} = MockEnv::with_client_builder(
4034-
ClusterClient::builder(vec![&*format!("redis://{name}")]),
4034+
ClusterClient::builder(vec![&*format!("redis://{name}")]).retries(1),
40354035
name,
40364036
move |received_cmd: &[u8], _| {
40374037
let slots_config_vec = vec![
@@ -4071,7 +4071,7 @@ mod cluster_async {
40714071
let res_err = res.unwrap_err();
40724072
assert_eq!(
40734073
res_err.kind(),
4074-
ErrorKind::ClusterConnectionNotFound,
4074+
ErrorKind::ConnectionNotFoundForRoute,
40754075
"{:?}",
40764076
res_err
40774077
);

‎redis/tests/test_cluster_scan.rs‎

Lines changed: 22 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -164,7 +164,12 @@ mod test_cluster_scan_async {
164164

165165
#[tokio::test] // test cluster scan with node fail in the middle
166166
async fn test_async_cluster_scan_with_fail() {
167-
let cluster = TestClusterContext::new(3, 0);
167+
let cluster = TestClusterContext::new_with_cluster_client_builder(
168+
3,
169+
0,
170+
|builder| builder.retries(1),
171+
false,
172+
);
168173
let mut connection = cluster.async_connection(None).await;
169174
// Set some keys
170175
for i in 0..1000 {
@@ -224,7 +229,11 @@ mod test_cluster_scan_async {
224229
let cluster = TestClusterContext::new_with_cluster_client_builder(
225230
6,
226231
1,
227-
|builder| builder.slots_refresh_rate_limit(Duration::from_secs(0), 0),
232+
|builder| {
233+
builder
234+
.slots_refresh_rate_limit(Duration::from_secs(0), 0)
235+
.retries(1)
236+
},
228237
false,
229238
);
230239

@@ -374,7 +383,11 @@ mod test_cluster_scan_async {
374383
let cluster = TestClusterContext::new_with_cluster_client_builder(
375384
6,
376385
1,
377-
|builder| builder.slots_refresh_rate_limit(Duration::from_secs(0), 0),
386+
|builder| {
387+
builder
388+
.slots_refresh_rate_limit(Duration::from_secs(0), 0)
389+
.retries(1)
390+
},
378391
false,
379392
);
380393

@@ -772,7 +785,12 @@ mod test_cluster_scan_async {
772785
// Testing cluster scan when connection fails in the middle and we get an error
773786
// then cluster up again and scanning can continue without any problem
774787
async fn test_async_cluster_scan_failover() {
775-
let mut cluster = TestClusterContext::new(3, 0);
788+
let mut cluster = TestClusterContext::new_with_cluster_client_builder(
789+
3,
790+
0,
791+
|builder| builder.retries(1),
792+
false,
793+
);
776794
let mut connection = cluster.async_connection(None).await;
777795
let mut i = 0;
778796
loop {

0 commit comments

Comments
 (0)