Add support for calling READONLY on replicas (#255)

* Add support for calling READONLY on replicas

Signed-off-by: Adam Fowler <adamfowler71@gmail.com>

* Add documentation fix

Signed-off-by: Adam Fowler <adamfowler71@gmail.com>

---------

Signed-off-by: Adam Fowler <adamfowler71@gmail.com>

Author: adam-fowler

Sources/Valkey/Cluster/ValkeyClusterClientStateMachine.swift

@@ -283,7 +283,6 @@ where
                 newShards.lazy.flatMap { $0.nodes.lazy.map { ValkeyNodeDescription(description: $0) } },
                 removeUnmentionedPools: true
             )
-            assert(poolUpdate.poolsToRun.isEmpty)
 
             var result = ClusterDiscoverySucceededAction(
                 createTimer: .init(

Sources/Valkey/Cluster/ValkeyClusterNodeClientFactory.swift

@@ -50,6 +50,7 @@ package struct ValkeyClusterNodeClientFactory: ValkeyNodeConnectionPoolFactory {
             address,
             connectionIDGenerator: self.connectionIDGenerator,
             connectionFactory: self.connectionFactory,
+            readOnly: nodeDescription.readOnly,
             eventLoopGroup: self.eventLoopGroup,
             logger: self.logger
         )

Sources/Valkey/Cluster/ValkeyNodeDescription.swift

@@ -41,6 +41,9 @@ package struct ValkeyNodeDescription: Identifiable, Hashable, Sendable {
     /// This property is required and is used as part of the node's unique identifier.
     package var port: Int
 
+    /// Is node a readonly replica
+    package var readOnly: Bool
+
     /// Creates a node description from any type conforming to the `ValkeyNodeDescriptionProtocol`.
     ///
     /// This initializer allows for easy conversion from various node description types
@@ -51,6 +54,7 @@ package struct ValkeyNodeDescription: Identifiable, Hashable, Sendable {
     package init(description: any ValkeyNodeDescriptionProtocol) {
         self.endpoint = description.endpoint
         self.port = description.port
+        self.readOnly = description.readOnly
     }
 
     /// Creates a node description from a cluster node description.
@@ -64,6 +68,7 @@ package struct ValkeyNodeDescription: Identifiable, Hashable, Sendable {
     package init(description: ValkeyClusterDescription.Node) {
         self.endpoint = description.endpoint
         self.port = description.tlsPort ?? description.port ?? 6379
+        self.readOnly = description.role == .replica
     }
 
     /// Creates a node description from a redirection error.
@@ -74,6 +79,7 @@ package struct ValkeyNodeDescription: Identifiable, Hashable, Sendable {
     package init(redirectionError: ValkeyClusterRedirectionError) {
         self.endpoint = redirectionError.endpoint
         self.port = redirectionError.port
+        self.readOnly = false
     }
 
     /// Determines whether this node description matches a given cluster node description.

Sources/Valkey/Cluster/ValkeyNodeDescriptionProtocol.swift

@@ -30,4 +30,7 @@ public protocol ValkeyNodeDescriptionProtocol: Sendable, Equatable {
 
     /// The port number on which the Valkey service is listening.
     var port: Int { get }
+
+    /// Is node a readonly replica.
+    var readOnly: Bool { get }
 }

Sources/Valkey/Cluster/ValkeyNodeDiscovery.swift

@@ -45,6 +45,7 @@ public struct ValkeyStaticNodeDiscovery: ValkeyNodeDiscovery {
     public struct NodeDescription: ValkeyNodeDescriptionProtocol {
         public var endpoint: String
         public var port: Int
+        public var readOnly: Bool { false }
 
         /// Initializes a `NodeDescription` with a host and optional IP.
         ///

Sources/Valkey/Connection/ValkeyChannelHandler.swift

@@ -57,6 +57,7 @@ final class ValkeyChannelHandler: ChannelInboundHandler {
         @usableFromInline
         let blockingCommandTimeout: TimeAmount
         let clientName: String?
+        let readOnly: Bool
     }
     @usableFromInline
     struct PendingCommand {
@@ -286,10 +287,15 @@ final class ValkeyChannelHandler: ChannelInboundHandler {
         let clientInfoLibName = CLIENT.SETINFO(attr: .libname(valkeySwiftLibraryName))
         let clientInfoLibVersion = CLIENT.SETINFO(attr: .libver(valkeySwiftLibraryVersion))
 
+        var numberOfPendingCommands = 2
         self.encoder.reset()
         helloCommand.encode(into: &self.encoder)
         clientInfoLibName.encode(into: &self.encoder)
         clientInfoLibVersion.encode(into: &self.encoder)
+        if self.configuration.readOnly {
+            numberOfPendingCommands += 1
+            READONLY().encode(into: &self.encoder)
+        }
 
         let promise = eventLoop.makePromise(of: RESPToken.self)
 
@@ -300,10 +306,7 @@ final class ValkeyChannelHandler: ChannelInboundHandler {
         self.stateMachine.setConnected(
             context: context,
             pendingHelloCommand: .init(promise: .nio(promise), requestID: 0, deadline: deadline),
-            pendingCommands: [
-                .init(promise: .forget, requestID: 0, deadline: deadline),  // CLIENT.SETINFO libname
-                .init(promise: .forget, requestID: 0, deadline: deadline),  // CLIENT.SETINFO libver
-            ]
+            pendingCommands: .init(repeating: .init(promise: .forget, requestID: 0, deadline: deadline), count: numberOfPendingCommands)
         )
     }
 
@@ -535,7 +538,8 @@ extension ValkeyChannelHandler.Configuration {
             authentication: other.authentication,
             commandTimeout: .init(other.commandTimeout),
             blockingCommandTimeout: .init(other.blockingCommandTimeout),
-            clientName: other.clientName
+            clientName: other.clientName,
+            readOnly: other.readOnly
         )
     }
 }

Sources/Valkey/Connection/ValkeyConnectionConfiguration.swift

@@ -116,6 +116,11 @@ public struct ValkeyConnectionConfiguration: Sendable {
     /// Default value is `nil` (no client name is set).
     public var clientName: String?
 
+    /// Is connection to be flagged as readonly.
+    ///
+    /// Readonly connections can run readonly commands on replica nodes
+    public var readOnly: Bool
+
     #if DistributedTracingSupport
     /// The distributed tracing configuration to use for this connection.
     /// Defaults to using the globally bootstrapped tracer with OpenTelemetry semantic conventions.
@@ -133,18 +138,21 @@ public struct ValkeyConnectionConfiguration: Sendable {
     ///   - blockingCommandTimeout: Maximum time to wait for a response to blocking commands. Defaults to 120 seconds.
     ///   - tls: TLS configuration for secure connections. Defaults to `.disable` for unencrypted connections.
     ///   - clientName: Optional name to identify this client connection on the server. Defaults to `nil`.
+    ///   - readOnly: Is the connection a readonly connection
     public init(
         authentication: Authentication? = nil,
         commandTimeout: Duration = .seconds(30),
         blockingCommandTimeout: Duration = .seconds(120),
         tls: TLS = .disable,
-        clientName: String? = nil
+        clientName: String? = nil,
+        readOnly: Bool = false
     ) {
         self.authentication = authentication
         self.commandTimeout = commandTimeout
         self.blockingCommandTimeout = blockingCommandTimeout
         self.tls = tls
         self.clientName = clientName
+        self.readOnly = readOnly
     }
 }
 

Sources/Valkey/Connection/ValkeyConnectionFactory.swift

@@ -54,13 +54,14 @@ package final class ValkeyConnectionFactory: Sendable {
 
     package func makeConnection(
         address: ValkeyServerAddress,
+        readOnly: Bool,
         connectionID: Int,
         eventLoop: any EventLoop,
         logger: Logger
     ) async throws -> ValkeyConnection {
         switch self.mode {
         case .default:
-            let connectionConfig = try await self.makeConnectionConfiguration()
+            let connectionConfig = try await self.makeConnectionConfiguration(readOnly: readOnly)
             return try await ValkeyConnection.connect(
                 address: address,
                 connectionID: connectionID,
@@ -70,7 +71,7 @@ package final class ValkeyConnectionFactory: Sendable {
             )
 
         case .custom(let customHandler):
-            async let connectionConfigPromise = self.makeConnectionConfiguration()
+            async let connectionConfigPromise = self.makeConnectionConfiguration(readOnly: readOnly)
             let channel = try await customHandler(address, eventLoop)
             let connectionConfig = try await connectionConfigPromise
 
@@ -94,7 +95,7 @@ package final class ValkeyConnectionFactory: Sendable {
         }
     }
 
-    func makeConnectionConfiguration() async throws -> ValkeyConnectionConfiguration {
+    func makeConnectionConfiguration(readOnly: Bool) async throws -> ValkeyConnectionConfiguration {
         let tls: ValkeyConnectionConfiguration.TLS =
             switch self.configuration.tls.base {
             case .disable:
@@ -110,7 +111,8 @@ package final class ValkeyConnectionFactory: Sendable {
             commandTimeout: self.configuration.commandTimeout,
             blockingCommandTimeout: self.configuration.blockingCommandTimeout,
             tls: tls,
-            clientName: nil
+            clientName: nil,
+            readOnly: readOnly
         )
 
         #if DistributedTracingSupport

Sources/Valkey/Node/ValkeyNodeClient.swift

@@ -46,8 +46,10 @@ package final class ValkeyNodeClient: Sendable {
     public let serverAddress: ValkeyServerAddress
     /// Connection pool
     let connectionPool: Pool
-
+    /// Connection factory
     let connectionFactory: ValkeyConnectionFactory
+    /// Should we create readonly connections
+    let readOnly: Bool
     /// configuration
     public var configuration: ValkeyClientConfiguration { self.connectionFactory.configuration }
     /// EventLoopGroup to use
@@ -70,6 +72,7 @@ package final class ValkeyNodeClient: Sendable {
         _ address: ValkeyServerAddress,
         connectionIDGenerator: ConnectionIDGenerator,
         connectionFactory: ValkeyConnectionFactory,
+        readOnly: Bool,
         eventLoopGroup: any EventLoopGroup,
         logger: Logger
     ) {
@@ -80,6 +83,7 @@ package final class ValkeyNodeClient: Sendable {
         poolConfiguration.maximumConnectionSoftLimit = connectionFactory.configuration.connectionPool.maximumConnectionCount
         poolConfiguration.maximumConnectionHardLimit = connectionFactory.configuration.connectionPool.maximumConnectionCount
 
+        self.readOnly = readOnly
         self.connectionPool = .init(
             configuration: poolConfiguration,
             idGenerator: connectionIDGenerator,
@@ -93,6 +97,7 @@ package final class ValkeyNodeClient: Sendable {
 
             let connection = try await connectionFactory.makeConnection(
                 address: address,
+                readOnly: readOnly,
                 connectionID: connectionID,
                 eventLoop: eventLoopGroup.any(),
                 logger: logger

Sources/Valkey/Node/ValkeyNodeClientFactory.swift

@@ -21,6 +21,12 @@ package struct ValkeyNodeClientFactory: ValkeyNodeConnectionPoolFactory {
     let connectionIDGenerator = ConnectionIDGenerator()
     let connectionFactory: ValkeyConnectionFactory
 
+    @usableFromInline
+    package struct NodeDescription {
+        let address: ValkeyServerAddress
+        let readOnly: Bool
+    }
+
     /// Creates a new `ValkeyClientFactory` instance.
     ///
     /// - Parameters:
@@ -44,11 +50,12 @@ package struct ValkeyNodeClientFactory: ValkeyNodeConnectionPoolFactory {
     /// - Parameter nodeDescription: Description of the node to connect to.
     /// - Returns: A configured `ValkeyNode` instance ready to connect to the specified node.
     @usableFromInline
-    package func makeConnectionPool(nodeDescription: ValkeyServerAddress) -> ValkeyNodeClient {
+    package func makeConnectionPool(nodeDescription: NodeDescription) -> ValkeyNodeClient {
         ValkeyNodeClient(
-            nodeDescription,
+            nodeDescription.address,
             connectionIDGenerator: self.connectionIDGenerator,
             connectionFactory: self.connectionFactory,
+            readOnly: nodeDescription.readOnly,
             eventLoopGroup: self.eventLoopGroup,
             logger: self.logger
         )