Merge branch 'main' into client-subscribe

Author: adam-fowler

Sources/Valkey/Cluster/ValkeyClusterClient.swift

@@ -157,21 +157,38 @@ public final class ValkeyClusterClient: Sendable {
             try await self.client(for: hashSlots)
         }
 
-        while true {
+        while !Task.isCancelled {
             do {
-                let respToken = try await self.retryingSend(
-                    clientSelector: clientSelector,
-                    command: command,
-                    logger: logger
-                )
-                return try Command.Response(fromRESP: respToken)
+                let client = try await clientSelector()
+                return try await client.send(command: command)
+            } catch ValkeyClusterError.noNodeToTalkTo {
+                // TODO: Rerun node discovery!
             } 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) }
             }
         }
+        throw CancellationError()
+    }
+
+    /// Get connection from cluster and run operation using connection
+    ///
+    /// - Parameters:
+    ///   - keys: Keys affected by operation. This is used to choose the cluster node
+    ///   - isolation: Actor isolation
+    ///   - operation: Closure handling Valkey connection
+    /// - Returns: Value returned by closure
+    @inlinable
+    public func withConnection<Value>(
+        forKeys keys: some Collection<ValkeyKey>,
+        isolation: isolated (any Actor)? = #isolation,
+        operation: (ValkeyConnection) async throws -> sending Value
+    ) async throws -> Value {
+        let hashSlots = keys.map { HashSlot(key: $0) }
+        let client = try await self.client(for: hashSlots)
+        return try await client.withConnection(isolation: isolation, operation: operation)
     }
 
     /// Starts running the cluster client.
@@ -455,39 +472,6 @@ public final class ValkeyClusterClient: Sendable {
         throw ValkeyClusterError.clusterIsMissingSlotAssignment
     }
 
-    /// Sends a command to the Valkey cluster with automatic retries and error handling.
-    ///
-    /// This internal method handles:
-    /// - Client selection strategy using the provided selector function
-    /// - Automatic retries for certain types of failures
-    /// - Task cancellation
-    ///
-    /// - Parameters:
-    ///   - clientSelector: A function that resolves to the appropriate client for this command.
-    ///   - command: The command to send to the cluster.
-    ///   - logger: The logger to use for diagnostic information.
-    /// - Returns: The raw RESP token response.
-    /// - Throws:
-    ///   - `CancellationError` if the task is cancelled
-    ///   - Other errors if command execution fails after retries
-    @inlinable
-    func retryingSend(
-        clientSelector: () async throws -> ValkeyClient,
-        command: some ValkeyCommand,
-        logger: Logger
-    ) async throws -> RESPToken {
-        while !Task.isCancelled {
-            do {
-                let client = try await clientSelector()
-                return try await client._send(command)
-            } catch ValkeyClusterError.noNodeToTalkTo {
-                // TODO: Rerun node discovery!
-            }
-        }
-
-        throw CancellationError()
-    }
-
     /// Generates a new unique request ID for tracking internal operations.
     ///
     /// - Returns: A unique integer ID for tracking requests and waiters.

Sources/Valkey/Commands/BitmapCommands.swift

@@ -492,6 +492,7 @@ extension ValkeyConnectionProtocol {
     ///     * [Array]: The result of the subcommand at the same position
     ///     * [Array]: In case OVERFLOW FAIL was given and overflows or underflows detected
     @inlinable
+    @discardableResult
     public func bitfield(_ key: ValkeyKey, operations: [BITFIELD.Operation] = []) async throws -> RESPToken.Array {
         try await send(command: BITFIELD(key, operations: operations))
     }
@@ -514,6 +515,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N)
     /// - Response: [Integer]: The size of the string stored in the destination key, that is equal to the size of the longest input string.
     @inlinable
+    @discardableResult
     public func bitop(operation: BITOP.Operation, destkey: ValkeyKey, keys: [ValkeyKey]) async throws -> Int {
         try await send(command: BITOP(operation: operation, destkey: destkey, keys: keys))
     }
@@ -551,6 +553,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: The original bit value stored at offset.
     @inlinable
+    @discardableResult
     public func setbit(_ key: ValkeyKey, offset: Int, value: Int) async throws -> Int {
         try await send(command: SETBIT(key, offset: offset, value: value))
     }

Sources/Valkey/Commands/ClusterCommands.swift

@@ -687,6 +687,7 @@ extension ValkeyConnectionProtocol {
     ///     * [String]: If the epoch was incremented.
     ///     * [String]: If the node already has the greatest config epoch in the cluster.
     @inlinable
+    @discardableResult
     public func clusterBumpepoch() async throws -> ByteBuffer {
         try await send(command: CLUSTER.BUMPEPOCH())
     }
@@ -698,6 +699,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the number of failure reports
     /// - Response: [Integer]: The number of active failure reports for the node.
     @inlinable
+    @discardableResult
     public func clusterCountFailureReports<NodeId: RESPStringRenderable>(nodeId: NodeId) async throws -> Int {
         try await send(command: CLUSTER.COUNTFAILUREREPORTS(nodeId: nodeId))
     }
@@ -709,6 +711,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: [Integer]: The number of keys in the specified hash slot.
     @inlinable
+    @discardableResult
     public func clusterCountkeysinslot(slot: Int) async throws -> Int {
         try await send(command: CLUSTER.COUNTKEYSINSLOT(slot: slot))
     }
@@ -770,6 +773,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the number of requested keys
     /// - Response: [Array]: An array with up to count elements.
     @inlinable
+    @discardableResult
     public func clusterGetkeysinslot(slot: Int, count: Int) async throws -> RESPToken.Array {
         try await send(command: CLUSTER.GETKEYSINSLOT(slot: slot, count: count))
     }
@@ -781,6 +785,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: [Array]: Helpful text about subcommands.
     @inlinable
+    @discardableResult
     public func clusterHelp() async throws -> RESPToken.Array {
         try await send(command: CLUSTER.HELP())
     }
@@ -792,6 +797,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: [String]: A map between named fields and values in the form of <field>:<value> lines separated by newlines composed by the two bytes CRLF
     @inlinable
+    @discardableResult
     public func clusterInfo() async throws -> ByteBuffer {
         try await send(command: CLUSTER.INFO())
     }
@@ -803,6 +809,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the number of bytes in the key
     /// - Response: [Integer]: The hash slot number for the specified key
     @inlinable
+    @discardableResult
     public func clusterKeyslot<Key: RESPStringRenderable>(_ key: Key) async throws -> Int {
         try await send(command: CLUSTER.KEYSLOT(key))
     }
@@ -814,6 +821,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the total number of Cluster nodes
     /// - Response: [Array]: An array of cluster links and their attributes.
     @inlinable
+    @discardableResult
     public func clusterLinks() async throws -> RESPToken.Array {
         try await send(command: CLUSTER.LINKS())
     }
@@ -837,6 +845,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: [String]: The node id.
     @inlinable
+    @discardableResult
     public func clusterMyid() async throws -> ByteBuffer {
         try await send(command: CLUSTER.MYID())
     }
@@ -848,6 +857,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: [String]: The node's shard id.
     @inlinable
+    @discardableResult
     public func clusterMyshardid() async throws -> ByteBuffer {
         try await send(command: CLUSTER.MYSHARDID())
     }
@@ -859,6 +869,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the total number of Cluster nodes
     /// - Response: [String]: The serialized cluster configuration.
     @inlinable
+    @discardableResult
     public func clusterNodes() async throws -> ByteBuffer {
         try await send(command: CLUSTER.NODES())
     }
@@ -870,6 +881,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the number of replicas.
     /// - Response: [Array]: A list of replica nodes replicating from the specified primary node provided in the same format used by CLUSTER NODES.
     @inlinable
+    @discardableResult
     public func clusterReplicas<NodeId: RESPStringRenderable>(nodeId: NodeId) async throws -> RESPToken.Array {
         try await send(command: CLUSTER.REPLICAS(nodeId: nodeId))
     }
@@ -933,6 +945,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the total number of cluster nodes
     /// - Response: [Array]: A nested list of a map of hash ranges and shard nodes describing individual shards.
     @inlinable
+    @discardableResult
     public func clusterShards() async throws -> CLUSTER.SHARDS.Response {
         try await send(command: CLUSTER.SHARDS())
     }
@@ -945,6 +958,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the number of replicas.
     /// - Response: [Array]: A list of replica nodes replicating from the specified primary node provided in the same format used by CLUSTER NODES.
     @inlinable
+    @discardableResult
     public func clusterSlaves<NodeId: RESPStringRenderable>(nodeId: NodeId) async throws -> RESPToken.Array {
         try await send(command: CLUSTER.SLAVES(nodeId: nodeId))
     }
@@ -956,6 +970,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the total number of slots based on arguments. O(N*log(N)) with ORDERBY subcommand.
     /// - Response: [Array]: Array of nested arrays, where the inner array element represents a slot and its respective usage statistics.
     @inlinable
+    @discardableResult
     public func clusterSlotStats(filter: CLUSTER.SLOTSTATS.Filter) async throws -> RESPToken.Array {
         try await send(command: CLUSTER.SLOTSTATS(filter: filter))
     }
@@ -970,6 +985,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the total number of Cluster nodes
     /// - Response: [Array]: Nested list of slot ranges with networking information.
     @inlinable
+    @discardableResult
     public func clusterSlots() async throws -> RESPToken.Array {
         try await send(command: CLUSTER.SLOTS())
     }

Sources/Valkey/Commands/ConnectionCommands.swift

@@ -808,6 +808,7 @@ extension ValkeyConnectionProtocol {
     /// - Available: 2.4.0
     /// - Complexity: Depends on subcommand.
     @inlinable
+    @discardableResult
     public func client() async throws -> CLIENT.Response {
         try await send(command: CLIENT())
     }
@@ -841,6 +842,7 @@ extension ValkeyConnectionProtocol {
     ///     * [String]: The connection name of the current connection
     ///     * [Null]: Connection name was not set
     @inlinable
+    @discardableResult
     public func clientGetname() async throws -> ByteBuffer? {
         try await send(command: CLIENT.GETNAME())
     }
@@ -855,6 +857,7 @@ extension ValkeyConnectionProtocol {
     ///     * -1: Client tracking is not enabled.
     ///     * [Integer]: ID of the client we are redirecting the notifications to.
     @inlinable
+    @discardableResult
     public func clientGetredir() async throws -> Int {
         try await send(command: CLIENT.GETREDIR())
     }
@@ -866,6 +869,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: [Array]: Helpful text about subcommands.
     @inlinable
+    @discardableResult
     public func clientHelp() async throws -> RESPToken.Array {
         try await send(command: CLIENT.HELP())
     }
@@ -877,6 +881,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: [Integer]: The id of the client
     @inlinable
+    @discardableResult
     public func clientId() async throws -> Int {
         try await send(command: CLIENT.ID())
     }
@@ -898,6 +903,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: [String]: A unique string, as described at the CLIENT LIST page, for the current client.
     @inlinable
+    @discardableResult
     public func clientInfo() async throws -> ByteBuffer {
         try await send(command: CLIENT.INFO())
     }
@@ -920,6 +926,7 @@ extension ValkeyConnectionProtocol {
     ///     * "OK": When called in 3 argument format.
     ///     * [Integer]: When called in filter/value format, the number of clients killed.
     @inlinable
+    @discardableResult
     public func clientKill(filter: CLIENT.KILL.Filter) async throws -> Int? {
         try await send(command: CLIENT.KILL(filter: filter))
     }
@@ -940,6 +947,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(N) where N is the number of client connections
     /// - Response: [String]: Information and statistics about client connections
     @inlinable
+    @discardableResult
     public func clientList(
         clientType: CLIENT.LIST.ClientType? = nil,
         clientIds: [Int] = [],
@@ -1052,6 +1060,7 @@ extension ValkeyConnectionProtocol {
     /// - Available: 6.2.0
     /// - Complexity: O(1)
     @inlinable
+    @discardableResult
     public func clientTrackinginfo() async throws -> RESPToken.Map {
         try await send(command: CLIENT.TRACKINGINFO())
     }
@@ -1065,6 +1074,7 @@ extension ValkeyConnectionProtocol {
     ///     * 0: If the client was unblocked successfully.
     ///     * 1: If the client wasn't unblocked.
     @inlinable
+    @discardableResult
     public func clientUnblock(clientId: Int, unblockType: CLIENT.UNBLOCK.UnblockType? = nil) async throws -> Int {
         try await send(command: CLIENT.UNBLOCK(clientId: clientId, unblockType: unblockType))
     }
@@ -1086,6 +1096,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1)
     /// - Response: [String]: The given string
     @inlinable
+    @discardableResult
     public func echo<Message: RESPStringRenderable>(message: Message) async throws -> ByteBuffer {
         try await send(command: ECHO(message: message))
     }
@@ -1099,6 +1110,7 @@ extension ValkeyConnectionProtocol {
     ///     * 6.2.0: `protover` made optional; when called without arguments the command reports the current connection's context.
     /// - Complexity: O(1)
     @inlinable
+    @discardableResult
     public func hello(arguments: HELLO.Arguments? = nil) async throws -> RESPToken.Map {
         try await send(command: HELLO(arguments: arguments))
     }
@@ -1112,6 +1124,7 @@ extension ValkeyConnectionProtocol {
     ///     * "PONG": Default reply.
     ///     * [String]: Relay of given `message`.
     @inlinable
+    @discardableResult
     public func ping(message: String? = nil) async throws -> PING.Response {
         try await send(command: PING(message: message))
     }
@@ -1133,6 +1146,7 @@ extension ValkeyConnectionProtocol {
     /// - Available: 6.2.0
     /// - Complexity: O(1)
     @inlinable
+    @discardableResult
     public func reset() async throws -> String {
         try await send(command: RESET())
     }