Fix Connection Creation Crash (#873)

### Motivation

When creating a connection, we wrongfully assumed that
`failedToCreateNewConnection` will always be called before
`http*ConnectionClosed` in the `HTTPConnectionPoolStateMachine`. However
this is far from correct. In NIO Futures are fulfilled before
`ChannelHandler` callbacks. Ordering in futures should not be assumed in
such a complex project.

### Change

We change the `http*ConnectionClosed` methods to be noops, if the
connection is in the starting state. We instead wait for the
`failedToCreateNewConnection` to create backoff timers and friends.

rdar://164674912

---------

Co-authored-by: George Barnett <gbarnett@apple.com>

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1Connections.swift

@@ -166,10 +166,22 @@ extension HTTPConnectionPool {
             }
         }
 
-        mutating func fail() {
+        enum FailAction {
+            case removeConnection
+            case none
+        }
+
+        mutating func fail() -> FailAction {
             switch self.state {
-            case .starting, .backingOff, .idle, .leased:
+            case .starting:
+                // If the connection fails while we are starting it, the fail call raced with
+                // `failedToConnect` (promises are succeeded or failed before channel handler
+                // callbacks). let's keep the state in `starting`, so that `failedToConnect` can
+                // create a backoff timer.
+                return .none
+            case .backingOff, .idle, .leased:
                 self.state = .closed
+                return .removeConnection
             case .closed:
                 preconditionFailure("Invalid state: \(self.state)")
             }
@@ -559,23 +571,28 @@ extension HTTPConnectionPool {
             }
 
             let use: ConnectionUse
-            self.connections[index].fail()
-            let eventLoop = self.connections[index].eventLoop
-            let starting: Int
-            if index < self.overflowIndex {
-                use = .generalPurpose
-                starting = self.startingGeneralPurposeConnections
-            } else {
-                use = .eventLoop(eventLoop)
-                starting = self.startingEventLoopConnections(on: eventLoop)
-            }
+            switch self.connections[index].fail() {
+            case .removeConnection:
+                let eventLoop = self.connections[index].eventLoop
+                let starting: Int
+                if index < self.overflowIndex {
+                    use = .generalPurpose
+                    starting = self.startingGeneralPurposeConnections
+                } else {
+                    use = .eventLoop(eventLoop)
+                    starting = self.startingEventLoopConnections(on: eventLoop)
+                }
 
-            let context = FailedConnectionContext(
-                eventLoop: eventLoop,
-                use: use,
-                connectionsStartingForUseCase: starting
-            )
-            return (index, context)
+                let context = FailedConnectionContext(
+                    eventLoop: eventLoop,
+                    use: use,
+                    connectionsStartingForUseCase: starting
+                )
+                return (index, context)
+
+            case .none:
+                return nil
+            }
         }
 
         // MARK: Migration

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift

@@ -250,6 +250,11 @@ extension HTTPConnectionPool {
             self.failedConsecutiveConnectionAttempts += 1
             self.lastConnectFailure = error
 
+            // We don't care how many waiting requests we have at this point, we will schedule a
+            // retry. More tasks, may appear until the backoff has completed. The final
+            // decision about the retry will be made in `connectionCreationBackoffDone(_:)`
+            let eventLoop = self.connections.backoffNextConnectionAttempt(connectionID)
+
             switch self.lifecycleState {
             case .running:
                 guard self.retryConnectionEstablishment else {
@@ -265,10 +270,6 @@ extension HTTPConnectionPool {
                         connection: .none
                     )
                 }
-                // We don't care how many waiting requests we have at this point, we will schedule a
-                // retry. More tasks, may appear until the backoff has completed. The final
-                // decision about the retry will be made in `connectionCreationBackoffDone(_:)`
-                let eventLoop = self.connections.backoffNextConnectionAttempt(connectionID)
 
                 let backoff = calculateBackoff(failedAttempt: self.failedConsecutiveConnectionAttempts)
                 return .init(

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift

@@ -187,10 +187,22 @@ extension HTTPConnectionPool {
             }
         }
 
-        mutating func fail() {
+        enum FailAction {
+            case removeConnection
+            case none
+        }
+
+        mutating func fail() -> FailAction {
             switch self.state {
-            case .starting, .active, .backingOff, .draining:
+            case .starting:
+                // If the connection fails while we are starting it, the fail call raced with
+                // `failedToConnect` (promises are succeeded or failed before channel handler
+                // callbacks). let's keep the state in `starting`, so that `failedToConnect` can
+                // create a backoff timer.
+                return .none
+            case .active, .backingOff, .draining:
                 self.state = .closed
+                return .removeConnection
             case .closed:
                 preconditionFailure("Invalid state: \(self.state)")
             }
@@ -749,10 +761,16 @@ extension HTTPConnectionPool {
                 // must ignore the event.
                 return nil
             }
-            self.connections[index].fail()
-            let eventLoop = self.connections[index].eventLoop
-            let context = FailedConnectionContext(eventLoop: eventLoop)
-            return (index, context)
+
+            switch self.connections[index].fail() {
+            case .none:
+                return nil
+
+            case .removeConnection:
+                let eventLoop = self.connections[index].eventLoop
+                let context = FailedConnectionContext(eventLoop: eventLoop)
+                return (index, context)
+            }
         }
 
         mutating func shutdown() -> CleanupContext {

Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift

@@ -226,20 +226,18 @@ extension HTTPConnectionPool {
         ) -> EstablishedAction {
             self.failedConsecutiveConnectionAttempts = 0
             self.lastConnectFailure = nil
-            if self.connections.hasActiveConnection(for: connection.eventLoop) {
-                guard let (index, _) = self.connections.failConnection(connection.id) else {
-                    preconditionFailure("we have established a new connection that we know nothing about?")
-                }
-                self.connections.removeConnection(at: index)
+            let doesConnectionExistsForEL = self.connections.hasActiveConnection(for: connection.eventLoop)
+            let (index, context) = self.connections.newHTTP2ConnectionEstablished(
+                connection,
+                maxConcurrentStreams: maxConcurrentStreams
+            )
+            if doesConnectionExistsForEL {
+                let connection = self.connections.closeConnection(at: index)
                 return .init(
                     request: .none,
                     connection: .closeConnection(connection, isShutdown: .no)
                 )
             } else {
-                let (index, context) = self.connections.newHTTP2ConnectionEstablished(
-                    connection,
-                    maxConcurrentStreams: maxConcurrentStreams
-                )
                 return self.nextActionForAvailableConnection(at: index, context: context)
             }
         }
@@ -424,6 +422,8 @@ extension HTTPConnectionPool {
             self.failedConsecutiveConnectionAttempts += 1
             self.lastConnectFailure = error
 
+            let eventLoop = self.connections.backoffNextConnectionAttempt(connectionID)
+
             switch self.lifecycleState {
             case .running:
                 guard self.retryConnectionEstablishment else {
@@ -440,7 +440,6 @@ extension HTTPConnectionPool {
                     )
                 }
 
-                let eventLoop = self.connections.backoffNextConnectionAttempt(connectionID)
                 let backoff = calculateBackoff(failedAttempt: self.failedConsecutiveConnectionAttempts)
                 return .init(
                     request: .none,