Fix cluster moved error handling (#119)

Author: adam-fowler

Sources/Valkey/Cluster/ValkeyClusterClient.swift

@@ -155,26 +155,25 @@ public final class ValkeyClusterClient: Sendable {
     @inlinable
     public func send<Command: ValkeyCommand>(command: Command) async throws -> Command.Response {
         let hashSlots = command.keysAffected.map { HashSlot(key: $0) }
-        let clientSelector: () async throws -> ValkeyClient = {
+        var clientSelector: () async throws -> ValkeyClient = {
             try await self.client(for: hashSlots)
         }
 
-        var respToken = try await self.retryingSend(
-            clientSelector: clientSelector,
-            command: command,
-            logger: logger
-        )
-
-        // TODO: We might want to collect redirects here for diagnostic purposes.
-        while let movedError = respToken.parseMovedError() {
-            respToken = try await self.retryingSend(
-                clientSelector: { try await self.client(for: movedError) },
-                command: command,
-                logger: logger
-            )
+        while true {
+            do {
+                let respToken = try await self.retryingSend(
+                    clientSelector: clientSelector,
+                    command: command,
+                    logger: logger
+                )
+                return try Command.Response(fromRESP: respToken)
+            } catch let error as ValkeyClientError where error.errorCode == .commandError {
+                guard let errorMessage = error.message, let movedError = ValkeyMovedError(errorMessage) else {
+                    throw error
+                }
+                clientSelector = { try await self.client(for: movedError) }
+            }
         }
-
-        return try Command.Response(fromRESP: respToken)
     }
 
     /// Starts running the cluster client.
@@ -195,7 +194,7 @@ public final class ValkeyClusterClient: Sendable {
     ///     group.addTask {
     ///         await client.run()
     ///     }
-    ///     
+    ///
     ///     // use the client here
     ///     let foo = try await client.get(key: "foo")
     /// }
@@ -209,7 +208,7 @@ public final class ValkeyClusterClient: Sendable {
         self.actionStreamContinuation.yield(.runClusterDiscovery(runNodeDiscovery: true))
 
         await withTaskCancellationHandler {
-            await withDiscardingTaskGroup() { taskGroup in
+            await withDiscardingTaskGroup { taskGroup in
                 await self.runUsingTaskGroup(&taskGroup)
             }
         } onCancel: {
@@ -427,7 +426,6 @@ public final class ValkeyClusterClient: Sendable {
                     try await withCheckedThrowingContinuation {
                         (continuation: CheckedContinuation<Void, any Error>) in
 
-
                         let action = self.stateLock.withLock {
                             $0.waitForHealthy(waiterID: waiterID, successNotifier: continuation)
                         }
@@ -486,9 +484,6 @@ public final class ValkeyClusterClient: Sendable {
                 return try await client._send(command)
             } catch ValkeyClusterError.noNodeToTalkTo {
                 // TODO: Rerun node discovery!
-            } catch let error as ValkeyClientError {
-                fatalError("TODO: error received: \(error)")
-                break
             }
         }
 
@@ -528,13 +523,14 @@ public final class ValkeyClusterClient: Sendable {
     /// - Parameter runNodeDiscoveryFirst: Whether to run node discovery before querying for cluster topology.
     private func runClusterDiscovery(runNodeDiscoveryFirst: Bool) async {
         do {
-            let voters = if runNodeDiscoveryFirst {
-                try await self.runNodeDiscovery()
-            } else {
-                self.stateLock.withLock {
-                    $0.getInitialVoters()
+            let voters =
+                if runNodeDiscoveryFirst {
+                    try await self.runNodeDiscovery()
+                } else {
+                    self.stateLock.withLock {
+                        $0.getInitialVoters()
+                    }
                 }
-            }
 
             let clusterDescription = try await self.runClusterDiscoveryFindingConsensus(voters: voters)
             let action = self.stateLock.withLock {
@@ -543,9 +539,12 @@ public final class ValkeyClusterClient: Sendable {
 
             self.runClusterDiscoverySucceededAction(action)
         } catch {
-            self.logger.debug("Valkey cluster discovery failed", metadata: [
-                "error": "\(error)",
-            ])
+            self.logger.debug(
+                "Valkey cluster discovery failed",
+                metadata: [
+                    "error": "\(error)"
+                ]
+            )
             let action = self.stateLock.withLock {
                 $0.valkeyClusterDiscoveryFailed(error)
             }
@@ -567,30 +566,36 @@ public final class ValkeyClusterClient: Sendable {
             let actions = self.stateLock.withLock {
                 $0.updateValkeyServiceNodes(mapped)
             }
-            self.logger.debug("Discovered nodes", metadata: [
-                "node_count": "\(nodes.count)",
-            ])
+            self.logger.debug(
+                "Discovered nodes",
+                metadata: [
+                    "node_count": "\(nodes.count)"
+                ]
+            )
             self.runUpdateValkeyNodesAction(actions)
             return actions.voters
         } catch {
-            self.logger.debug("Failed to discover nodes", metadata: [
-                "error": "\(error)",
-            ])
+            self.logger.debug(
+                "Failed to discover nodes",
+                metadata: [
+                    "error": "\(error)"
+                ]
+            )
             throw error
         }
     }
 
     /// Establishes consensus on the cluster topology by querying multiple nodes.
     ///
-    /// This method uses a voting mechanism to establish consensus among multiple nodes 
+    /// This method uses a voting mechanism to establish consensus among multiple nodes
     /// about the current cluster topology. It requires a quorum of nodes to agree
     /// on the topology before accepting it.
     ///
     /// - Parameter voters: The list of nodes that can vote on cluster topology.
     /// - Returns: The agreed-upon cluster description.
     /// - Throws: `ValkeyClusterError.clusterIsUnavailable` if consensus cannot be reached.
     private func runClusterDiscoveryFindingConsensus(voters: [ValkeyClusterVoter<ValkeyClient>]) async throws -> ValkeyClusterDescription {
-        return try await withThrowingTaskGroup(of: (ValkeyClusterDescription, ValkeyNodeID).self) { taskGroup in
+        try await withThrowingTaskGroup(of: (ValkeyClusterDescription, ValkeyNodeID).self) { taskGroup in
             for voter in voters {
                 taskGroup.addTask {
                     (try await voter.client.clusterShards(), voter.nodeID)
@@ -606,17 +611,23 @@ public final class ValkeyClusterClient: Sendable {
                     do {
                         let metrics = try election.voteReceived(for: description, from: nodeID)
 
-                        self.logger.debug("Vote received", metadata: [
-                            "candidate_count": "\(metrics.candidateCount)",
-                            "candidate": "\(metrics.candidate)",
-                            "votes_received": "\(metrics.votesReceived)",
-                            "votes_needed": "\(metrics.votesNeeded)"
-                        ])
+                        self.logger.debug(
+                            "Vote received",
+                            metadata: [
+                                "candidate_count": "\(metrics.candidateCount)",
+                                "candidate": "\(metrics.candidate)",
+                                "votes_received": "\(metrics.votesReceived)",
+                                "votes_needed": "\(metrics.votesNeeded)",
+                            ]
+                        )
                     } catch let error as ValkeyClusterError {
-                        self.logger.debug("Vote invalid", metadata: [
-                            "nodeID": "\(nodeID)",
-                            "error": "\(error)",
-                        ])
+                        self.logger.debug(
+                            "Vote invalid",
+                            metadata: [
+                                "nodeID": "\(nodeID)",
+                                "error": "\(error)",
+                            ]
+                        )
                         continue
                     }
 
@@ -636,9 +647,12 @@ public final class ValkeyClusterClient: Sendable {
                     }
 
                 case .failure(let error):
-                    self.logger.debug("Received an error while asking for cluster topology", metadata: [
-                        "error": "\(error)"
-                    ])
+                    self.logger.debug(
+                        "Received an error while asking for cluster topology",
+                        metadata: [
+                            "error": "\(error)"
+                        ]
+                    )
                 }
             }
 

Sources/Valkey/Cluster/ValkeyMovedError.swift

@@ -45,46 +45,20 @@ package struct ValkeyMovedError: Hashable, Sendable {
     }
 }
 
-extension RESPToken {
+extension ValkeyMovedError {
     static let movedPrefix = "MOVED "
 
-    /// Attempts to parse a RESP error token as a Valkey MOVED error.
+    /// Attempts to parse a Valkey MOVED error from a String.
     ///
-    /// This method extracts the hash slot, endpoint, and port information from a RESP error
-    /// token if it represents a Valkey MOVED error. MOVED errors are returned by Valkey cluster
+    /// This method extracts the hash slot, endpoint, and port information from the string
+    /// if it represents a Valkey MOVED error. MOVED errors are returned by Valkey cluster
     /// nodes when a client attempts to access a key that belongs to a different node.
     ///
     /// The error format is expected to be: `"MOVED <slot> <endpoint>:<port>"`
     ///
     /// - Returns: A `ValkeyMovedError` if the token represents a valid MOVED error, or `nil` otherwise.
     @usableFromInline
-    func parseMovedError() -> ValkeyMovedError? {
-        let byteBuffer: ByteBuffer? =
-            switch self.value {
-            case .bulkError(let byteBuffer),
-                .simpleError(let byteBuffer):
-                byteBuffer
-
-            case .simpleString,
-                .bulkString,
-                .verbatimString,
-                .number,
-                .double,
-                .boolean,
-                .bigNumber,
-                .array,
-                .attribute,
-                .map,
-                .set,
-                .push,
-                .null:
-                nil
-            }
-
-        guard var byteBuffer else {
-            return nil
-        }
-        let errorMessage = byteBuffer.readString(length: byteBuffer.readableBytes)!
+    init?(_ errorMessage: String) {
         guard errorMessage.hasPrefix(Self.movedPrefix) else {
             return nil
         }
@@ -112,6 +86,6 @@ extension RESPToken {
             return nil
         }
 
-        return ValkeyMovedError(slot: slot, endpoint: Swift.String(endpoint), port: port)
+        self = ValkeyMovedError(slot: slot, endpoint: Swift.String(endpoint), port: port)
     }
 }

Tests/ValkeyTests/Cluster/ValkeyMovedErrorTests.swift

@@ -22,11 +22,8 @@ struct ValkeyMovedErrorTests {
 
     @Test("parseMovedError parses valid MOVED error")
     func testParseValidMovedError() async throws {
-        // Create a RESPToken with a MOVED error
-        let token = RESPToken(.simpleError("MOVED 1234 redis.example.com:6379"))
-
         // Parse the moved error
-        let movedError = token.parseMovedError()
+        let movedError = ValkeyMovedError("MOVED 1234 redis.example.com:6379")
 
         // Verify the moved error is parsed correctly
         #expect(movedError != nil)
@@ -39,11 +36,9 @@ struct ValkeyMovedErrorTests {
     func testParseValidMovedErrorFromBulkError() async throws {
         // Create a RESPToken with a MOVED error
         let errorMessage = "MOVED 5000 10.0.0.1:6380"
-        let byteBuffer = ByteBuffer(string: errorMessage)
-        let token = RESPToken(.bulkError(byteBuffer))
 
         // Parse the moved error
-        let movedError = token.parseMovedError()
+        let movedError = ValkeyMovedError(errorMessage)
 
         // Verify the moved error is parsed correctly
         #expect(movedError != nil)
@@ -52,54 +47,33 @@ struct ValkeyMovedErrorTests {
         #expect(movedError?.port == 6380)
     }
 
-    @Test("parseMovedError returns nil for non-error tokens")
-    func testParseNonErrorToken() async throws {
-        // Test with various non-error token types
-        let nullToken = RESPToken(.null)
-        #expect(nullToken.parseMovedError() == nil)
-
-        let stringToken = RESPToken(.simpleString("OK"))
-        #expect(stringToken.parseMovedError() == nil)
-
-        let numberToken = RESPToken(.number(42))
-        #expect(numberToken.parseMovedError() == nil)
-
-        let arrayToken = RESPToken(.array([.number(1), .number(2)]))
-        #expect(arrayToken.parseMovedError() == nil)
-    }
-
     @Test("parseMovedError returns nil for error tokens without MOVED prefix")
     func testParseNonMovedError() async throws {
-        let errorMessage = "ERR unknown command"
-        let byteBuffer = ByteBuffer(string: errorMessage)
-        let token = RESPToken(.simpleError(byteBuffer))
-
-        #expect(token.parseMovedError() == nil)
+        #expect(ValkeyMovedError("ERR unknown command") == nil)
     }
 
     @Test("parseMovedError returns nil for invalid MOVED format")
     func testParseInvalidMovedFormat() async throws {
         // Test with various invalid MOVED formats
 
         // Missing slot number
-        let missingSlot = RESPToken(.simpleError("MOVED redis.example.com:6379"))
-        #expect(missingSlot.parseMovedError() == nil)
+        #expect(ValkeyMovedError("MOVED redis.example.com:6379") == nil)
 
         // Missing port number
-        let missingPort = RESPToken(.simpleError("MOVED 1234 redis.example.com"))
-        #expect(missingPort.parseMovedError() == nil)
+        let missingPort = "MOVED 1234 redis.example.com"
+        #expect(ValkeyMovedError(missingPort) == nil)
 
         // Invalid slot number
-        let invalidSlot = RESPToken(.simpleError("MOVED abc redis.example.com:6379"))
-        #expect(invalidSlot.parseMovedError() == nil)
+        let invalidSlot = "MOVED abc redis.example.com:6379"
+        #expect(ValkeyMovedError(invalidSlot) == nil)
 
         // Invalid port number
-        let invalidPort = RESPToken(.simpleError("MOVED 1234 redis.example.com:port"))
-        #expect(invalidPort.parseMovedError() == nil)
+        let invalidPort = "MOVED 1234 redis.example.com:port"
+        #expect(ValkeyMovedError(invalidPort) == nil)
 
         // Slot number out of range
-        let outOfRangeSlot = RESPToken(.simpleError("MOVED 999999 redis.example.com:6379"))
-        #expect(outOfRangeSlot.parseMovedError() == nil)
+        let outOfRangeSlot = "MOVED 999999 redis.example.com:6379"
+        #expect(ValkeyMovedError(outOfRangeSlot) == nil)
     }
 
     @Test("ValkeyMovedError is Hashable")