Delay connection attempts without addresses. (#64)

In some circumstances users may have connection pools configured without
any SocketAddresses ready to go. This is particularly likely in service
discovery configurations. Right now, the effect of attempting to use
such a pool is two fold. First, we'll emit a bunch of error level logs
telling users we have no addresses. Second, we'll fall into the
exponential backoff phase of connection establishment.

The first property is annoying, but the second one is actively harmful.
If your construction is timed incorrectly, we'll have the awkward
problem of burning a bunch of CPU trying to create connections we know
we cannot, and then a lengthy delay after the addresses are actually
configured before we start trying to use them. That's the worst of all
worlds.

This patch adds logic to detect the attempt to create connections when
we don't have any configured addresses and delays them. This path should
improve performance and ergonomics when in this mode.

Authored-by: Cory Benfield <lukasa@apple.com>

Author: fabianfett

Sources/RediStack/ConnectionPool/ConnectionPool.swift

@@ -385,7 +385,7 @@ extension ConnectionPool {
         }
         self.connectionWaiters.append(waiter)
 
-        // What are we going to wait for? Well, now we check. If the number of active connections is
+        // Ok, we have connection targets. If the number of active connections is
         // below the max, or the pool is leaky, we can create a new connection. Otherwise, we just have
         // to wait for a connection to come back.
         if self.activeConnectionCount < self.maximumConnectionCount || self.leaky {

Sources/RediStack/RedisConnectionPool.swift

@@ -47,6 +47,14 @@ public class RedisConnectionPool {
     private var serverConnectionAddresses: ConnectionAddresses
     // This needs to be a var because its value changes as the pool enters/leaves pubsub mode to reuse the same connection.
     private var pubsubConnection: RedisConnection?
+    // This array buffers any request for a connection that cannot be succeeded right away in the case where we have no target.
+    // We never allow this to get larger than a specific bound, to resist DoS attacks. Past that bound we will fast-fail.
+    private var requestsForConnections: [EventLoopPromise<RedisConnection>] = []
+
+    /// The maximum number of connection requests we'll buffer in `requestsForConnections` before we start fast-failing. These
+    /// are buffered only when there are no available addresses to connect to, so in practice it's highly unlikely this will be
+    /// hit, but either way, 100 concurrent connection requests ought to be plenty in this case.
+    private static let maximumBufferedConnectionRequests = 100
 
     public init(configuration: Configuration, boundEventLoop: EventLoop) {
         var config = configuration
@@ -109,6 +117,11 @@ extension RedisConnectionPool {
 
             // This breaks the cycle between us and the pool.
             self.pool = nil
+
+            // Drop all pending connection attempts. No need to empty this manually, it'll get dropped regardless.
+            for request in self.requestsForConnections {
+                request.fail(RedisConnectionPoolError.poolClosed)
+            }
         }
     }
 
@@ -171,6 +184,14 @@ extension RedisConnectionPool {
 
         self.loop.execute {
             self.serverConnectionAddresses.update(newAddresses)
+
+            // Shiny, we can unbuffer any pending connections and pass them on as they now have somewhere to go.
+            let unbufferedRequests = self.requestsForConnections
+            self.requestsForConnections = []
+
+            for request in unbufferedRequests {
+                request.completeWith(self.connectionFactory(self.loop))
+            }
         }
     }
 
@@ -182,8 +203,17 @@ extension RedisConnectionPool {
         let factoryConfig = self.configuration.factoryConfiguration
 
         guard let nextTarget = self.serverConnectionAddresses.nextTarget() else {
-            // No valid connection target, we'll fail.
-            return targetLoop.makeFailedFuture(RedisConnectionPoolError.noAvailableConnectionTargets)
+            // No valid connection target, we'll keep track of the request and attempt to satisfy it later.
+            // First, confirm we have space to keep track of this. If not, fast-fail.
+            guard self.requestsForConnections.count < RedisConnectionPool.maximumBufferedConnectionRequests else {
+                return targetLoop.makeFailedFuture(RedisConnectionPoolError.noAvailableConnectionTargets)
+            }
+
+            // Ok, we can buffer, let's do that.
+            self.prepareLoggerForUse(nil).notice("waiting for target addresses")
+            let promise = targetLoop.makePromise(of: RedisConnection.self)
+            self.requestsForConnections.append(promise)
+            return promise.futureResult
         }
 
         let connectionConfig: RedisConnection.Configuration

Sources/RediStackTestUtils/RedisConnectionPoolIntegrationTestCase.swift

@@ -74,18 +74,33 @@ open class RedisConnectionPoolIntegrationTestCase: XCTestCase {
         self.pool.close()
         self.pool = nil
     }
-    
+
     public func makeNewPool(
         connectionRetryTimeout: TimeAmount? = .seconds(5),
         minimumConnectionCount: Int = 0
     ) throws -> RedisConnectionPool {
-        let address = try SocketAddress.makeAddressResolvingHost(self.redisHostname, port: self.redisPort)
+        try self.makeNewPool(
+            initialAddresses: nil,
+            initialConnectionBackoffDelay: .milliseconds(100),
+            connectionRetryTimeout: connectionRetryTimeout,
+            minimumConnectionCount: minimumConnectionCount
+        )
+    }
+
+    public func makeNewPool(
+        initialAddresses: [SocketAddress]?,
+        initialConnectionBackoffDelay: TimeAmount,
+        connectionRetryTimeout: TimeAmount?,
+        minimumConnectionCount: Int
+    ) throws -> RedisConnectionPool {
+        let addresses = try initialAddresses ?? [SocketAddress.makeAddressResolvingHost(self.redisHostname, port: self.redisPort)]
         let pool = RedisConnectionPool(
             configuration: RedisConnectionPool.Configuration(
-                initialServerConnectionAddresses: [address],
+                initialServerConnectionAddresses: addresses,
                 maximumConnectionCount: .maximumActiveConnections(4),
                 connectionFactoryConfiguration: .init(connectionPassword: self.redisPassword),
                 minimumConnectionCount: minimumConnectionCount,
+                initialConnectionBackoffDelay: initialConnectionBackoffDelay,
                 connectionRetryTimeout: connectionRetryTimeout
             ),
             boundEventLoop: self.eventLoopGroup.next()

Tests/RediStackIntegrationTests/RedisConnectionPoolTests.swift

@@ -43,6 +43,56 @@ final class RedisConnectionPoolTests: RediStackConnectionPoolIntegrationTestCase
         defer { pool.close() }
         XCTAssertNoThrow(try pool.get(#function).wait())
     }
+
+    func test_noConnectionAttemptsUntilAddressesArePresent() throws {
+        // Note the config here: we have no initial addresses, the connecton backoff delay is 10 seconds, and the retry timeout is only 5 seconds.
+        // The effect of this config is that if we fail a connection attempt, we'll fail it forever.
+        let pool = try self.makeNewPool(initialAddresses: [], initialConnectionBackoffDelay: .seconds(10), connectionRetryTimeout: .seconds(5), minimumConnectionCount: 0)
+        defer { pool.close() }
+
+        // As above we're gonna try to insert a bunch of elements into a set. This time,
+        // the pool has no addresses yet. We expect that when we add an address later everything will work nicely.
+        // We do fewer here.
+        let operations = (0..<10).map { number in
+            pool.sadd([number], to: #function)
+        }
+
+        // Now that we've kicked those off, let's hand over a new address.
+        try pool.updateConnectionAddresses([SocketAddress.makeAddressResolvingHost(self.redisHostname, port: self.redisPort)])
+
+        // We should get the results.
+        let results = try EventLoopFuture<Int>.whenAllSucceed(operations, on: self.eventLoopGroup.next()).wait()
+        XCTAssertEqual(results, Array(repeating: 1, count: 10))
+    }
+
+    func testDelayedConnectionsFailOnClose() throws {
+        // Note the config here: we have no initial addresses, the connecton backoff delay is 10 seconds, and the retry timeout is only 5 seconds.
+        // The effect of this config is that if we fail a connection attempt, we'll fail it forever.
+        let pool = try self.makeNewPool(initialAddresses: [], initialConnectionBackoffDelay: .seconds(10), connectionRetryTimeout: .seconds(5), minimumConnectionCount: 0)
+        defer { pool.close() }
+
+        // As above we're gonna try to insert a bunch of elements into a set. This time,
+        // the pool has no addresses yet. We expect that when we add an address later everything will work nicely.
+        // We do fewer here.
+        let operations = (0..<10).map { number in
+            pool.sadd([number], to: #function)
+        }
+
+        // Now that we've kicked those off, let's close.
+        pool.close()
+
+        let results = try EventLoopFuture<Int>.whenAllComplete(operations, on: self.eventLoopGroup.next()).wait()
+        for result in results {
+            switch result {
+            case .success:
+                XCTFail("Request succeeded")
+            case .failure(let error) where error as? RedisConnectionPoolError == .poolClosed:
+                ()  // Pass
+            case .failure(let error):
+                XCTFail("Unexpected failure: \(error)")
+            }
+        }
+    }
 }
 
 // MARK: Leasing a connection