From 42aaab43f9acfd078c7efb4afd7d1db1b5dcfed8 Mon Sep 17 00:00:00 2001 From: Alex Rehnby-Martin Date: Tue, 26 May 2026 08:32:09 -0700 Subject: [PATCH] Implement connection refresh on circular moved (#5893) * Implement connection refresh on circular moved Signed-off-by: Alex Rehnby-Martin * fix: format redirect_node.map() call on single line Signed-off-by: Alex Rehnby-Martin * fix(redis-rs): handle circular MOVED redirect in pipeline path Extract shared is_circular_moved_redirect() function to detect when a MOVED redirect points to the same address (circular redirect). Apply this detection to both single command and pipeline paths. For pipelines, circular MOVED errors now route through the reconnect logic instead of the normal redirect logic, matching the behavior of single commands. Added test to verify pipeline commands recover from circular MOVED + disconnect scenarios. Signed-off-by: Alex Rehnby-Martin * chore(deps-dev): bump maturin from 1.13.1 to 1.13.3 in /python in the patch-updates group across 1 directory (#5910) chore(deps-dev): bump maturin Bumps the patch-updates group with 1 update in the /python directory: [maturin](https://github.com/pyo3/maturin). Updates `maturin` from 1.13.1 to 1.13.3 - [Release notes](https://github.com/pyo3/maturin/releases) - [Changelog](https://github.com/PyO3/maturin/blob/main/Changelog.md) - [Commits](https://github.com/pyo3/maturin/compare/v1.13.1...v1.13.3) --- updated-dependencies: - dependency-name: maturin dependency-version: 1.13.3 dependency-type: direct:development update-type: version-update:semver-patch dependency-group: patch-updates ... Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Signed-off-by: Alex Rehnby-Martin * fix(redis-rs): improve circular MOVED detection with hostname/IP resolution - Add resolve_address parameter to is_circular_moved_redirect to handle cases where MOVED returns an IP but client connected via hostname - Fix resolve_address to preserve original port when resolving IP to hostname (the MOVED response port is authoritative) - Extend circular MOVED detection to pipeline path via handle_redirect_logic - Add unit tests for hostname vs IP detection scenarios - Add integration test for pipeline circular MOVED handling Signed-off-by: Alex Rehnby-Martin * style: fix cargo fmt formatting Signed-off-by: Alex Rehnby-Martin --------- Signed-off-by: Alex Rehnby-Martin Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> (cherry picked from commit 87c9991c1e77e3968f14f443e2f05222d4b9c410) Signed-off-by: James Xin --- .../redis-rs/redis/src/cluster_async/mod.rs | 167 ++++++- .../src/cluster_async/pipeline_routing.rs | 67 ++- .../redis/tests/support/mock_cluster.rs | 39 +- .../redis/tests/test_cluster_async.rs | 429 ++++++++++++++++++ 4 files changed, 681 insertions(+), 21 deletions(-) diff --git a/glide-core/redis-rs/redis/src/cluster_async/mod.rs b/glide-core/redis-rs/redis/src/cluster_async/mod.rs index 12a506544af..8805969495c 100644 --- a/glide-core/redis-rs/redis/src/cluster_async/mod.rs +++ b/glide-core/redis-rs/redis/src/cluster_async/mod.rs @@ -140,6 +140,51 @@ fn set_routed_node_on_span(span: &GlideSpan, address: &str) { } } +/// Checks if a MOVED redirect is circular (points to the same address we're already connected to). +/// +/// A circular MOVED can happen when: +/// 1. Client connects through a DNS endpoint (e.g., cluster.example.com) +/// 2. MOVED response points back to the same DNS endpoint +/// 3. The connection may be closed, causing retry to fail on read +/// +/// In this case, we need to reconnect before retrying to ensure we get +/// a fresh connection. Without this, the retry write may succeed (to buffer) +/// but the read will fail with FatalReceiveError, which doesn't trigger retry. +/// +/// The `resolve_address` parameter allows resolving IP addresses to their canonical +/// hostname form using the slot map's IP→address table. This handles cases where: +/// - Connected via hostname but MOVED returns an IP +/// - Different IP representations of the same machine +/// +/// Returns `true` if the redirect is circular and a reconnect should be triggered. +pub(crate) fn is_circular_moved_redirect( + redirect_node: Option<(&str, u16)>, + current_address: &str, + resolve_address: F, +) -> bool +where + F: Fn(&str) -> String, +{ + if let Some((redirect_addr, _slot)) = redirect_node { + // Resolve both addresses to canonical form for comparison + let resolved_redirect = resolve_address(redirect_addr); + let resolved_current = resolve_address(current_address); + + if resolved_redirect == resolved_current { + log_debug_lazy!( + "cluster", + format!( + "Detected circular MOVED redirect: {} -> {} (resolved: {} == {}). \ + Reconnecting before retry to avoid potential connection issues.", + current_address, redirect_addr, resolved_current, resolved_redirect + ) + ); + return true; + } + } + false +} + /// This represents an async Cluster connection. It stores the /// underlying connections maintained for each node in the cluster, as well /// as common parameters for connecting to nodes and executing commands. @@ -1152,6 +1197,7 @@ pin_project! { struct Request { retry_params: RetryParams, request: Option>, + core: Arc>, #[pin] future: RequestState>, } @@ -1184,7 +1230,10 @@ enum Next { Done, } -impl Future for Request { +impl Future for Request +where + C: ConnectionLike + Connect + Clone + Send + Sync + 'static, +{ type Output = Next; fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context) -> Poll { @@ -1343,9 +1392,25 @@ impl Future for Request { RetryMethod::MovedRedirect => { let mut request = this.request.take().unwrap(); let redirect_node = err.redirect_node(); + let core = this.core.clone(); + + // Check for circular MOVED and trigger reconnect if detected + // Use resolve_address to handle hostname vs IP mismatches + if is_circular_moved_redirect(redirect_node, &address, |addr| { + ClusterConnInner::resolve_address(&core, addr) + }) { + // Reset routing and reconnect with retry + request.info.reset_routing(); + return Next::Reconnect { + request: Some(request), + target: address, + } + .into(); + } + + // Normal MOVED handling: set redirect and refresh slots request.info.set_redirect( - err.redirect_node() - .map(|(node, _slot)| Redirect::Moved(node.to_string())), + redirect_node.map(|(node, _slot)| Redirect::Moved(node.to_string())), ); Next::RefreshSlots { request: Some(request), @@ -2740,14 +2805,20 @@ where /// Resolution order: /// 1. Reverse IP lookup: parse the host as an IP and look it up in the slot map's /// IP→address table (built from DNS resolution during CLUSTER SLOTS refresh). + /// If found, replace only the host portion, preserving the original port. /// 2. Raw address fallback: return the original address unchanged. pub(crate) fn resolve_address(inner: &Arc>, address: &str) -> String { let conn_lock = inner.conn_lock.read(); // Step 1: Reverse IP lookup via slot map. - if let Some((host, _port)) = address.rsplit_once(':') { + if let Some((host, port)) = address.rsplit_once(':') { if let Ok(ip) = host.parse::() { if let Some(node_addr) = conn_lock.slot_map.node_address_for_ip(ip) { + // Extract just the hostname from the resolved address and combine with original port + if let Some((resolved_host, _resolved_port)) = node_addr.rsplit_once(':') { + return format!("{}:{}", resolved_host, port); + } + // Fallback: if resolved address has no port (shouldn't happen), return as-is return (*node_addr).clone(); } } @@ -3650,6 +3721,7 @@ where self.in_flight_requests.push(Box::pin(Request { retry_params: retry_params.clone(), request: Some(request), + core: self.inner.clone(), future: RequestState::Future { future }, })); } @@ -3667,6 +3739,7 @@ where self.in_flight_requests.push(Box::pin(Request { retry_params: retry_params.clone(), request: Some(request), + core: self.inner.clone(), future: RequestState::Future { future: Box::pin(future), }, @@ -3684,6 +3757,7 @@ where self.in_flight_requests.push(Box::pin(Request { retry_params: retry_params.clone(), request: Some(request), + core: self.inner.clone(), future: RequestState::Future { future: Box::pin(future), }, @@ -3729,6 +3803,7 @@ where self.in_flight_requests.push(Box::pin(Request { retry_params, request, + core: self.inner.clone(), future, })); } @@ -4776,3 +4851,87 @@ mod parse_node_address_tests { assert_eq!(parse_node_address(":6379"), Some(("", 6379))); } } + +#[cfg(test)] +mod is_circular_moved_redirect_tests { + use super::is_circular_moved_redirect; + + // Identity resolver for simple tests (no address translation) + fn identity_resolver(addr: &str) -> String { + addr.to_string() + } + + #[test] + fn exact_match_is_circular() { + assert!(is_circular_moved_redirect( + Some(("127.0.0.1:6379", 5000)), + "127.0.0.1:6379", + identity_resolver + )); + } + + #[test] + fn different_port_is_not_circular() { + assert!(!is_circular_moved_redirect( + Some(("127.0.0.1:6379", 5000)), + "127.0.0.1:6380", + identity_resolver + )); + } + + #[test] + fn different_host_is_not_circular() { + assert!(!is_circular_moved_redirect( + Some(("10.0.0.1:6379", 5000)), + "10.0.0.2:6379", + identity_resolver + )); + } + + #[test] + fn none_redirect_is_not_circular() { + assert!(!is_circular_moved_redirect( + None, + "127.0.0.1:6379", + identity_resolver + )); + } + + #[test] + fn ip_vs_hostname_should_detect_circular() { + // Simulate a resolver that maps 127.0.0.1:6379 to localhost:6379 + // This is what the slot map's IP→address table would do + fn localhost_resolver(addr: &str) -> String { + if addr == "127.0.0.1:6379" { + "localhost:6379".to_string() + } else { + addr.to_string() + } + } + + // Connected via "localhost:6379" but MOVED returns "127.0.0.1:6379" + // With the resolver, both should resolve to "localhost:6379" + assert!( + is_circular_moved_redirect( + Some(("127.0.0.1:6379", 5000)), + "localhost:6379", + localhost_resolver + ), + "Should detect circular redirect when MOVED returns IP but connected via hostname" + ); + } + + #[test] + fn ip_vs_hostname_without_resolver_does_not_detect() { + // Without a proper resolver, IP vs hostname won't match + // This demonstrates the limitation when no IP mapping is available + assert!( + !is_circular_moved_redirect( + Some(("127.0.0.1:6379", 5000)), + "localhost:6379", + identity_resolver + ), + "Without resolver, IP vs hostname won't be detected as circular" + ); + } +} diff --git a/glide-core/redis-rs/redis/src/cluster_async/pipeline_routing.rs b/glide-core/redis-rs/redis/src/cluster_async/pipeline_routing.rs index f3f4d236767..ff9f3db5ca5 100644 --- a/glide-core/redis-rs/redis/src/cluster_async/pipeline_routing.rs +++ b/glide-core/redis-rs/redis/src/cluster_async/pipeline_routing.rs @@ -22,6 +22,7 @@ use tokio::sync::oneshot; use tokio::sync::oneshot::error::RecvError; use super::boxed_sleep; +use super::is_circular_moved_redirect; use super::testing::RefreshConnectionType; use super::CmdArg; use super::PendingRequest; @@ -363,9 +364,9 @@ where /// * `pipeline_map` - A map of node pipelines where the commands are grouped by their corresponding nodes. /// * `core` - The core object that provides access to connection locks and other resources. /// * `retry` - The retry counter. -/// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands. -/// - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering). -/// - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions). +/// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands. +/// - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering). +/// - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions). /// /// # Returns /// @@ -563,9 +564,9 @@ type RetryMap = HashMap>; /// `RedisResult` or a `RecvError`. /// - `addresses_and_indices`: A collection of pairs where each pair associates a node address with the indices /// of commands in the pipeline that were sent to that node. -/// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands. -/// - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering). -/// - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions). +/// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands. +/// - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering). +/// - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions). /// /// # Returns /// @@ -751,9 +752,9 @@ fn update_retry_map( /// * `pipeline` - A reference to the original pipeline containing the commands. /// * `core` - The core object that provides access to connection locks and other resources. /// * `response_policies` - A HashMap of routing info and response policies to the pipeline commands. -/// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands. -/// - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering). -/// - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions). +/// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands. +/// - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering). +/// - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions). pub(crate) async fn process_and_retry_pipeline_responses( mut responses: Vec, RecvError>>, mut addresses_and_indices: AddressAndIndices, @@ -833,9 +834,9 @@ where /// * `retry` - The retry counter. /// * `pipeline_responses` - A mutable reference to the collection of pipeline responses. /// * `response_policies` - A HashMap of routing info and response policies to the pipeline commands. -/// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands. -/// - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering). -/// - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions). +/// - `pipeline_retry_strategy`: Configures retry behavior for pipeline commands. +/// - `retry_server_error`: If `true`, retries commands on server errors (may cause reordering). +/// - `retry_connection_error`: If `true`, retries on connection errors (may lead to duplicate executions). /// /// # Returns /// @@ -1040,6 +1041,9 @@ where /// This function processes the retry map entries that indicate a redirection error (e.g., MOVED or ASK). /// It attempts to obtain a new connection based on the redirection information and reassigns the command /// to the appropriate node pipeline for execution. +/// +/// For circular MOVED redirects (where the redirect points to the same address), this function +/// triggers a reconnect to get a fresh connection before retrying. async fn handle_redirect_logic( retry_method: RetryMethod, core: Core, @@ -1052,7 +1056,44 @@ async fn handle_redirect_logic( where C: Clone + ConnectionLike + Connect + Send + Sync + 'static, { - for (indices, address, mut error) in indices_addresses_and_error { + // Separate circular MOVED redirects from normal redirects + // Circular MOVED needs reconnect handling, not normal redirect handling + let mut circular_moved_entries: Vec<((usize, Option), String, ServerError)> = Vec::new(); + let mut normal_redirect_entries: Vec<((usize, Option), String, ServerError)> = + Vec::new(); + + for (indices, address, error) in indices_addresses_and_error { + let redis_error: RedisError = error.clone().into(); + + // Check for circular MOVED redirect + // Use resolve_address to handle hostname vs IP mismatches + if matches!(retry_method, RetryMethod::MovedRedirect) + && is_circular_moved_redirect(redis_error.redirect_node(), &address, |addr| { + ClusterConnInner::resolve_address(&core, addr) + }) + { + circular_moved_entries.push((indices, address, error)); + } else { + normal_redirect_entries.push((indices, address, error)); + } + } + + // Handle circular MOVED redirects by triggering reconnect + if !circular_moved_entries.is_empty() { + handle_reconnect_logic( + circular_moved_entries, + core.clone(), + pipeline, + pipeline_responses, + true, // should_retry = true, we want to retry after reconnect + pipeline_map, + response_policies, + ) + .await?; + } + + // Handle normal redirects + for (indices, address, mut error) in normal_redirect_entries { // Convert the ServerError to a RedisError and try to extract redirect info. let redis_error: RedisError = error.clone().into(); let (index, inner_index) = indices; diff --git a/glide-core/redis-rs/redis/tests/support/mock_cluster.rs b/glide-core/redis-rs/redis/tests/support/mock_cluster.rs index 37cd568b8b3..db1b46623ab 100644 --- a/glide-core/redis-rs/redis/tests/support/mock_cluster.rs +++ b/glide-core/redis-rs/redis/tests/support/mock_cluster.rs @@ -358,12 +358,43 @@ impl aio::ConnectionLike for MockConnection { fn req_packed_commands<'a>( &'a mut self, - _pipeline: &'a redis::Pipeline, - _offset: usize, - _count: usize, + pipeline: &'a redis::Pipeline, + offset: usize, + count: usize, _pipeline_retry_strategy: Option, ) -> RedisFuture<'a, Vec> { - Box::pin(future::ok(vec![])) + // Process each command in the pipeline through the handler + let handler = self.handler.clone(); + let port = self.port; + + Box::pin(future::ready({ + // For atomic pipelines, we need to handle the MULTI/EXEC wrapper + // The packed pipeline contains all commands concatenated + // We'll process each command individually and collect results + let mut results = Vec::new(); + + // Iterate through commands in the pipeline + for cmd in pipeline.cmd_iter().skip(offset).take(count) { + let packed_cmd = cmd.get_packed_command(); + match (handler)(&packed_cmd, port) { + Ok(()) => { + // Handler didn't specify a response, use default + results.push(Value::Nil); + } + Err(result) => { + match result { + Ok(value) => results.push(value), + Err(err) => { + // Convert error to ServerError value + results.push(Value::ServerError(err.into())); + } + } + } + } + } + + Ok(results) + })) } fn get_db(&self) -> i64 { diff --git a/glide-core/redis-rs/redis/tests/test_cluster_async.rs b/glide-core/redis-rs/redis/tests/test_cluster_async.rs index 4b3738c7edc..fd89737a028 100644 --- a/glide-core/redis-rs/redis/tests/test_cluster_async.rs +++ b/glide-core/redis-rs/redis/tests/test_cluster_async.rs @@ -6707,6 +6707,435 @@ mod cluster_async { ); } + /// Test for circular MOVED detection and reconnect behavior. + /// + /// This test verifies that when a MOVED response points to the same address + /// (circular MOVED), the client triggers a reconnect before retrying. + /// + /// The test tracks: + /// 1. Connection attempts (via PING count) - to verify reconnect happened + /// 2. GET requests - to verify the retry flow + /// + /// Expected flow with the fix: + /// - Initial connection: PING (connection 1) + /// - GET request 0: returns MOVED to same address (circular) + /// - Fix detects circular MOVED, triggers reconnect + /// - Reconnect: PING (connection 2) + /// - GET request 1: returns success (on new connection) + /// + /// The key assertion is that we see more PINGs than the initial connection, + /// proving that a reconnect occurred before the retry succeeded. + /// + /// To run: cargo test --test test_cluster_async -- test_async_cluster_circular_moved_triggers_reconnect --nocapture + #[test] + #[serial_test::serial] + fn test_async_cluster_circular_moved_triggers_reconnect() { + let name = "test_circular_moved_reconnect"; + let get_requests = Arc::new(atomic::AtomicUsize::new(0)); + let get_requests_clone = get_requests.clone(); + let ping_count = Arc::new(atomic::AtomicUsize::new(0)); + let ping_count_clone = ping_count.clone(); + + // Track the ping count at the time of each GET request + // This lets us verify that a reconnect (new PING) happened between requests + let ping_at_get_0 = Arc::new(atomic::AtomicUsize::new(0)); + let ping_at_get_0_clone = ping_at_get_0.clone(); + let ping_at_get_1 = Arc::new(atomic::AtomicUsize::new(0)); + let ping_at_get_1_clone = ping_at_get_1.clone(); + + let MockEnv { + runtime, + async_connection: mut connection, + handler, + .. + } = MockEnv::with_client_builder( + ClusterClient::builder(vec![&*format!("redis://{name}")]) + .retries(5) + .slots_refresh_rate_limit(Duration::from_secs(0), 0), + name, + move |cmd: &[u8], port| { + // Track connection establishment via PING + if contains_slice(cmd, b"PING") { + ping_count_clone.fetch_add(1, atomic::Ordering::SeqCst); + return Err(Ok(Value::SimpleString("OK".into()))); + } + + if contains_slice(cmd, b"SETNAME") { + return Err(Ok(Value::SimpleString("OK".into()))); + } + + if contains_slice(cmd, b"CLUSTER") && contains_slice(cmd, b"SLOTS") { + return Err(Ok(Value::Array(vec![Value::Array(vec![ + Value::Int(0), + Value::Int(16383), + Value::Array(vec![ + Value::BulkString(name.as_bytes().to_vec()), + Value::Int(port as i64), + ]), + ])]))); + } + + if contains_slice(cmd, b"READONLY") { + return Err(Ok(Value::SimpleString("OK".into()))); + } + + if contains_slice(cmd, b"GET") { + let i = get_requests_clone.fetch_add(1, atomic::Ordering::SeqCst); + let current_pings = ping_count_clone.load(atomic::Ordering::SeqCst); + + match i { + 0 => { + // Record ping count at first GET + ping_at_get_0_clone.store(current_pings, atomic::Ordering::SeqCst); + // Return MOVED pointing to the SAME address (circular) + Err(parse_redis_value( + format!("-MOVED 12345 {name}:{port}\r\n").as_bytes(), + )) + } + _ => { + // Record ping count at retry GET + ping_at_get_1_clone.store(current_pings, atomic::Ordering::SeqCst); + // Return success + Err(Ok(Value::BulkString(b"success".to_vec()))) + } + } + } else { + Err(Ok(Value::SimpleString("OK".into()))) + } + }, + ); + + let result = runtime.block_on(async move { + cmd("GET") + .arg("test_key") + .query_async::<_, Option>(&mut connection) + .await + }); + + drop(handler); + + let total_gets = get_requests.load(atomic::Ordering::SeqCst); + let total_pings = ping_count.load(atomic::Ordering::SeqCst); + let pings_at_first_get = ping_at_get_0.load(atomic::Ordering::SeqCst); + let pings_at_retry_get = ping_at_get_1.load(atomic::Ordering::SeqCst); + + // Verify the command succeeded + match result { + Ok(Some(value)) => { + assert_eq!(value, "success", "Expected successful response"); + } + Ok(None) => { + panic!("Expected Some(value), got None"); + } + Err(e) => { + panic!( + "Request failed with error: {:?}. Total GETs: {}, Total PINGs: {}", + e, total_gets, total_pings + ); + } + } + + // Verify that at least 2 GET requests were made (original + retry) + assert!( + total_gets >= 2, + "Expected at least 2 GET requests, got {}", + total_gets + ); + + // Verify that a reconnect happened between the first GET and the retry + // The ping count at retry should be higher than at the first GET + assert!( + pings_at_retry_get > pings_at_first_get, + "Expected reconnect between GETs: pings at GET 0 = {}, pings at GET 1 = {}. \ + A reconnect should have added more PINGs before the retry.", + pings_at_first_get, + pings_at_retry_get + ); + + println!( + "Test PASSED: Circular MOVED triggered reconnect. \ + GETs: {}, PINGs: {} (at GET 0: {}, at GET 1: {})", + total_gets, total_pings, pings_at_first_get, pings_at_retry_get + ); + } + + /// Variant test: Circular MOVED with SET command. + /// Verifies the fix works for write commands as well as read commands. + /// + /// To run: cargo test --test test_cluster_async -- test_async_cluster_circular_moved_set_triggers_reconnect --nocapture + #[test] + #[serial_test::serial] + fn test_async_cluster_circular_moved_set_triggers_reconnect() { + let name = "test_circular_moved_set_reconnect"; + let set_requests = Arc::new(atomic::AtomicUsize::new(0)); + let set_requests_clone = set_requests.clone(); + let ping_count = Arc::new(atomic::AtomicUsize::new(0)); + let ping_count_clone = ping_count.clone(); + let ping_at_set_0 = Arc::new(atomic::AtomicUsize::new(0)); + let ping_at_set_0_clone = ping_at_set_0.clone(); + let ping_at_set_1 = Arc::new(atomic::AtomicUsize::new(0)); + let ping_at_set_1_clone = ping_at_set_1.clone(); + + let MockEnv { + runtime, + async_connection: mut connection, + handler, + .. + } = MockEnv::with_client_builder( + ClusterClient::builder(vec![&*format!("redis://{name}")]) + .retries(5) + .slots_refresh_rate_limit(Duration::from_secs(0), 0), + name, + move |cmd: &[u8], port| { + if contains_slice(cmd, b"PING") { + ping_count_clone.fetch_add(1, atomic::Ordering::SeqCst); + return Err(Ok(Value::SimpleString("OK".into()))); + } + + if contains_slice(cmd, b"SETNAME") { + return Err(Ok(Value::SimpleString("OK".into()))); + } + + if contains_slice(cmd, b"CLUSTER") && contains_slice(cmd, b"SLOTS") { + return Err(Ok(Value::Array(vec![Value::Array(vec![ + Value::Int(0), + Value::Int(16383), + Value::Array(vec![ + Value::BulkString(name.as_bytes().to_vec()), + Value::Int(port as i64), + ]), + ])]))); + } + + if contains_slice(cmd, b"READONLY") { + return Err(Ok(Value::SimpleString("OK".into()))); + } + + if contains_slice(cmd, b"SET") { + let i = set_requests_clone.fetch_add(1, atomic::Ordering::SeqCst); + let current_pings = ping_count_clone.load(atomic::Ordering::SeqCst); + + match i { + 0 => { + ping_at_set_0_clone.store(current_pings, atomic::Ordering::SeqCst); + Err(parse_redis_value( + format!("-MOVED 5000 {name}:{port}\r\n").as_bytes(), + )) + } + _ => { + ping_at_set_1_clone.store(current_pings, atomic::Ordering::SeqCst); + Err(Ok(Value::SimpleString("OK".into()))) + } + } + } else { + Err(Ok(Value::SimpleString("OK".into()))) + } + }, + ); + + let result = runtime.block_on(async move { + cmd("SET") + .arg("key") + .arg("value") + .query_async::<_, ()>(&mut connection) + .await + }); + + drop(handler); + + let total_sets = set_requests.load(atomic::Ordering::SeqCst); + let total_pings = ping_count.load(atomic::Ordering::SeqCst); + let pings_at_first_set = ping_at_set_0.load(atomic::Ordering::SeqCst); + let pings_at_retry_set = ping_at_set_1.load(atomic::Ordering::SeqCst); + + assert!(result.is_ok(), "SET command failed: {:?}", result.err()); + + assert!( + total_sets >= 2, + "Expected at least 2 SET requests, got {}", + total_sets + ); + + assert!( + pings_at_retry_set > pings_at_first_set, + "Expected reconnect between SETs: pings at SET 0 = {}, pings at SET 1 = {}", + pings_at_first_set, + pings_at_retry_set + ); + + println!( + "Test PASSED: Circular MOVED SET triggered reconnect. \ + SETs: {}, PINGs: {} (at SET 0: {}, at SET 1: {})", + total_sets, total_pings, pings_at_first_set, pings_at_retry_set + ); + } + + #[test] + #[serial_test::serial] + fn test_async_cluster_circular_moved_pipeline_triggers_reconnect() { + let name = "test_circular_moved_pipeline_reconnect"; + let set_requests = Arc::new(atomic::AtomicUsize::new(0)); + let set_requests_clone = set_requests.clone(); + let ping_count = Arc::new(atomic::AtomicUsize::new(0)); + let ping_count_clone = ping_count.clone(); + + // Track the ping count at the time of each SET request + // This lets us verify that a reconnect (new PING) happened between requests + let ping_at_set_0 = Arc::new(atomic::AtomicUsize::new(0)); + let ping_at_set_0_clone = ping_at_set_0.clone(); + let ping_at_set_1 = Arc::new(atomic::AtomicUsize::new(0)); + let ping_at_set_1_clone = ping_at_set_1.clone(); + + let MockEnv { + runtime, + async_connection: mut connection, + handler, + .. + } = MockEnv::with_client_builder( + ClusterClient::builder(vec![&*format!("redis://{name}")]) + .retries(5) + .slots_refresh_rate_limit(Duration::from_secs(0), 0), + name, + move |cmd: &[u8], port| { + // Track connection establishment via PING + if contains_slice(cmd, b"PING") { + ping_count_clone.fetch_add(1, atomic::Ordering::SeqCst); + return Err(Ok(Value::SimpleString("OK".into()))); + } + + if contains_slice(cmd, b"SETNAME") { + return Err(Ok(Value::SimpleString("OK".into()))); + } + + if contains_slice(cmd, b"CLUSTER") && contains_slice(cmd, b"SLOTS") { + return Err(Ok(Value::Array(vec![Value::Array(vec![ + Value::Int(0), + Value::Int(16383), + Value::Array(vec![ + Value::BulkString(name.as_bytes().to_vec()), + Value::Int(port as i64), + ]), + ])]))); + } + + if contains_slice(cmd, b"READONLY") { + return Err(Ok(Value::SimpleString("OK".into()))); + } + + if contains_slice(cmd, b"SET") { + let i = set_requests_clone.fetch_add(1, atomic::Ordering::SeqCst); + let current_pings = ping_count_clone.load(atomic::Ordering::SeqCst); + + match i { + 0 => { + // Record ping count at first SET + ping_at_set_0_clone.store(current_pings, atomic::Ordering::SeqCst); + // Return MOVED pointing to the SAME address (circular) + Err(parse_redis_value( + format!("-MOVED 5000 {name}:{port}\r\n").as_bytes(), + )) + } + _ => { + // Record ping count at retry SET + ping_at_set_1_clone.store(current_pings, atomic::Ordering::SeqCst); + // Return success + Err(Ok(Value::SimpleString("OK".into()))) + } + } + } else { + Err(Ok(Value::SimpleString("OK".into()))) + } + }, + ); + + let result = runtime.block_on(async move { + // Create a non-atomic pipeline with a single SET command + // Non-atomic pipelines go through the pipeline_routing path + let mut pipeline = redis::pipe(); + pipeline.set("test_key", "test_value"); + + connection + .route_pipeline( + &pipeline, + 0, + 1, + None, + Some(PipelineRetryStrategy { + retry_server_error: true, + retry_connection_error: false, + }), + ) + .await + }); + + drop(handler); + + let total_sets = set_requests.load(atomic::Ordering::SeqCst); + let total_pings = ping_count.load(atomic::Ordering::SeqCst); + let pings_at_first_set = ping_at_set_0.load(atomic::Ordering::SeqCst); + let pings_at_retry_set = ping_at_set_1.load(atomic::Ordering::SeqCst); + + // Verify the pipeline succeeded + match result { + Ok(values) => { + assert!( + !values.is_empty(), + "Expected at least one response from pipeline" + ); + // Check if the response is OK or an error + match &values[0] { + Value::SimpleString(s) if s == "OK" => { + // Success + } + Value::ServerError(err) => { + panic!( + "Pipeline command failed with error: {:?}. \ + Total SETs: {}, Total PINGs: {}", + err, total_sets, total_pings + ); + } + other => { + panic!( + "Unexpected response: {:?}. Total SETs: {}, Total PINGs: {}", + other, total_sets, total_pings + ); + } + } + } + Err(e) => { + panic!( + "Pipeline failed with error: {:?}. Total SETs: {}, Total PINGs: {}", + e, total_sets, total_pings + ); + } + } + + // Verify that at least 2 SET requests were made (original + retry) + assert!( + total_sets >= 2, + "Expected at least 2 SET requests, got {}", + total_sets + ); + + // Verify that a reconnect happened between the first SET and the retry + // The ping count at retry should be higher than at the first SET + // This assertion will FAIL until the circular MOVED fix is applied to the pipeline path + assert!( + pings_at_retry_set > pings_at_first_set, + "Expected reconnect between pipeline SETs: pings at SET 0 = {}, pings at SET 1 = {}. \ + A reconnect should have added more PINGs before the retry. \ + This indicates the circular MOVED fix is NOT applied to the pipeline path.", + pings_at_first_set, + pings_at_retry_set + ); + + println!( + "Test PASSED: Circular MOVED in pipeline triggered reconnect. \ + SETs: {}, PINGs: {} (at SET 0: {}, at SET 1: {})", + total_sets, total_pings, pings_at_first_set, pings_at_retry_set + ); + } + mod mtls_test { use crate::support::mtls_test::create_cluster_client_from_cluster; use redis::ConnectionInfo;