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;