diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 42c74a7ce..d2fdc3809 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -1,5 +1,8 @@ name: Main +permissions: + contents: read + on: push: branches: [main] diff --git a/.github/workflows/pull_request.yml b/.github/workflows/pull_request.yml index 4efd046b0..7423cd3b2 100644 --- a/.github/workflows/pull_request.yml +++ b/.github/workflows/pull_request.yml @@ -1,5 +1,8 @@ name: PR +permissions: + contents: read + on: pull_request: types: [opened, reopened, synchronize] diff --git a/.github/workflows/pull_request_label.yml b/.github/workflows/pull_request_label.yml index 8fd47c13f..d2da2f1ac 100644 --- a/.github/workflows/pull_request_label.yml +++ b/.github/workflows/pull_request_label.yml @@ -1,5 +1,8 @@ name: PR label +permissions: + contents: read + on: pull_request: types: [labeled, unlabeled, opened, reopened, synchronize] diff --git a/Package.swift b/Package.swift index 3cff98089..aad0c1c53 100644 --- a/Package.swift +++ b/Package.swift @@ -1,4 +1,4 @@ -// swift-tools-version:5.10 +// swift-tools-version:6.0 //===----------------------------------------------------------------------===// // // This source file is part of the AsyncHTTPClient open source project @@ -19,10 +19,6 @@ let strictConcurrencyDevelopment = false let strictConcurrencySettings: [SwiftSetting] = { var initialSettings: [SwiftSetting] = [] - initialSettings.append(contentsOf: [ - .enableUpcomingFeature("StrictConcurrency"), - .enableUpcomingFeature("InferSendableFromCaptures"), - ]) if strictConcurrencyDevelopment { // -warnings-as-errors here is a workaround so that IDE-based development can @@ -44,7 +40,7 @@ let package = Package( .package(url: "https://github.com/apple/swift-nio-http2.git", from: "1.36.0"), .package(url: "https://github.com/apple/swift-nio-extras.git", from: "1.26.0"), .package(url: "https://github.com/apple/swift-nio-transport-services.git", from: "1.24.0"), - .package(url: "https://github.com/apple/swift-log.git", from: "1.6.0"), + .package(url: "https://github.com/apple/swift-log.git", from: "1.7.1"), .package(url: "https://github.com/apple/swift-atomics.git", from: "1.0.2"), .package(url: "https://github.com/apple/swift-algorithms.git", from: "1.0.0"), .package(url: "https://github.com/apple/swift-distributed-tracing.git", from: "1.3.0"), @@ -96,6 +92,7 @@ let package = Package( .product(name: "Algorithms", package: "swift-algorithms"), // Observability support .product(name: "Logging", package: "swift-log"), + .product(name: "InMemoryLogging", package: "swift-log"), .product(name: "Tracing", package: "swift-distributed-tracing"), .product(name: "InMemoryTracing", package: "swift-distributed-tracing"), ], diff --git a/README.md b/README.md index a4f49c8c8..b557e58fa 100644 --- a/README.md +++ b/README.md @@ -306,7 +306,7 @@ Please have a look at [SECURITY.md](SECURITY.md) for AsyncHTTPClient's security ## Supported Versions -The most recent versions of AsyncHTTPClient support Swift 5.10 and newer. The minimum Swift version supported by AsyncHTTPClient releases are detailed below: +The most recent versions of AsyncHTTPClient support Swift 6.0 and newer. The minimum Swift version supported by AsyncHTTPClient releases are detailed below: AsyncHTTPClient | Minimum Swift Version --------------------|---------------------- @@ -318,4 +318,5 @@ AsyncHTTPClient | Minimum Swift Version `1.20.0 ..< 1.21.0` | 5.7 `1.21.0 ..< 1.26.0` | 5.8 `1.26.0 ..< 1.27.0` | 5.9 -`1.27.0 ...` | 5.10 +`1.27.0 ..< 1.30.0` | 5.10 +`1.30.0 ...` | 6.0 diff --git a/Sources/AsyncHTTPClient/AsyncAwait/HTTPClient+execute.swift b/Sources/AsyncHTTPClient/AsyncAwait/HTTPClient+execute.swift index 0437211c6..bbf8c948c 100644 --- a/Sources/AsyncHTTPClient/AsyncAwait/HTTPClient+execute.swift +++ b/Sources/AsyncHTTPClient/AsyncAwait/HTTPClient+execute.swift @@ -92,7 +92,12 @@ extension HTTPClient { // this loop is there to follow potential redirects while true { - let preparedRequest = try HTTPClientRequest.Prepared(currentRequest, dnsOverride: configuration.dnsOverride) + let preparedRequest = + try HTTPClientRequest.Prepared( + currentRequest, + dnsOverride: configuration.dnsOverride, + tracing: self.configuration.tracing + ) let response = try await { var response = try await self.executeCancellable(preparedRequest, deadline: deadline, logger: logger) diff --git a/Sources/AsyncHTTPClient/AsyncAwait/HTTPClientRequest+Prepared.swift b/Sources/AsyncHTTPClient/AsyncAwait/HTTPClientRequest+Prepared.swift index c39452897..b5649cf90 100644 --- a/Sources/AsyncHTTPClient/AsyncAwait/HTTPClientRequest+Prepared.swift +++ b/Sources/AsyncHTTPClient/AsyncAwait/HTTPClientRequest+Prepared.swift @@ -12,9 +12,11 @@ // //===----------------------------------------------------------------------===// +import Instrumentation import NIOCore import NIOHTTP1 import NIOSSL +import ServiceContextModule import struct Foundation.URL @@ -45,7 +47,11 @@ extension HTTPClientRequest { @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) extension HTTPClientRequest.Prepared { - init(_ request: HTTPClientRequest, dnsOverride: [String: String] = [:]) throws { + init( + _ request: HTTPClientRequest, + dnsOverride: [String: String] = [:], + tracing: HTTPClient.TracingConfiguration? = nil + ) throws { guard !request.url.isEmpty, let url = URL(string: request.url) else { throw HTTPClientError.invalidURL } @@ -54,6 +60,12 @@ extension HTTPClientRequest.Prepared { var headers = request.headers headers.addHostIfNeeded(for: deconstructedURL) + if let tracer = tracing?.tracer, + let context = ServiceContext.current + { + tracer.inject(context, into: &headers, using: HTTPHeadersInjector.shared) + } + let metadata = try headers.validateAndSetTransportFraming( method: request.method, bodyLength: .init(request.body) diff --git a/Sources/AsyncHTTPClient/AsyncAwait/Transaction.swift b/Sources/AsyncHTTPClient/AsyncAwait/Transaction.swift index c0c22bfee..a25c92e80 100644 --- a/Sources/AsyncHTTPClient/AsyncAwait/Transaction.swift +++ b/Sources/AsyncHTTPClient/AsyncAwait/Transaction.swift @@ -105,26 +105,43 @@ final class Transaction: struct BreakTheWriteLoopError: Swift.Error {} - // FIXME: Refactor this to not use `self.state.unsafe`. private func writeRequestBodyPart(_ part: ByteBuffer) async throws { - self.state.unsafe.lock() - switch self.state.unsafe.withValueAssumingLockIsAcquired({ state in state.writeNextRequestPart() }) { + let action = self.state.withLockedValue { state in + state.writeNextRequestPart() + } + + switch action { case .writeAndContinue(let executor): - self.state.unsafe.unlock() executor.writeRequestBodyPart(.byteBuffer(part), request: self, promise: nil) - - case .writeAndWait(let executor): + case .writeAndWait: + // Holding the lock here *should* be safe but because of a bug in the runtime + // it isn't, so drop the lock, create the continuation and try again. + // + // See https://github.com/swiftlang/swift/issues/85668 try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in - self.state.unsafe.withValueAssumingLockIsAcquired({ state in - state.waitForRequestBodyDemand(continuation: continuation) - }) - self.state.unsafe.unlock() + let action = self.state.withLockedValue { state in + // Check that nothing has changed between dropping and re-acquiring the lock. + let action = state.writeNextRequestPart() + switch action { + case .writeAndContinue, .fail: + () + case .writeAndWait: + state.waitForRequestBodyDemand(continuation: continuation) + } + return action + } - executor.writeRequestBodyPart(.byteBuffer(part), request: self, promise: nil) + switch action { + case .writeAndContinue(let executor): + executor.writeRequestBodyPart(.byteBuffer(part), request: self, promise: nil) + continuation.resume() + case .writeAndWait(let executor): + executor.writeRequestBodyPart(.byteBuffer(part), request: self, promise: nil) + case .fail: + continuation.resume(throwing: BreakTheWriteLoopError()) + } } - case .fail: - self.state.unsafe.unlock() throw BreakTheWriteLoopError() } } diff --git a/Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool+Factory.swift b/Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool+Factory.swift index c896791cf..3dc47c5ae 100644 --- a/Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool+Factory.swift +++ b/Sources/AsyncHTTPClient/ConnectionPool/HTTPConnectionPool+Factory.swift @@ -68,13 +68,8 @@ extension HTTPConnectionPool.ConnectionFactory { var logger = logger logger[metadataKey: "ahc-connection-id"] = "\(connectionID)" - self.makeChannel( - requester: requester, - connectionID: connectionID, - deadline: deadline, - eventLoop: eventLoop, - logger: logger - ).whenComplete { [logger] result in + let promise = eventLoop.makePromise(of: NegotiatedProtocol.self) + promise.futureResult.whenComplete { [logger] result in switch result { case .success(.http1_1(let channel)): do { @@ -143,10 +138,26 @@ extension HTTPConnectionPool.ConnectionFactory { } } - case .failure(let error): + case .failure(var error): + // let's map `ChannelError.connectTimeout` into a `HTTPClientError.connectTimeout` + switch error { + case ChannelError.connectTimeout: + error = HTTPClientError.connectTimeout + default: + () + } requester.failedToCreateHTTPConnection(connectionID, error: error) } } + + self.makeChannel( + requester: requester, + connectionID: connectionID, + deadline: deadline, + eventLoop: eventLoop, + logger: logger, + promise: promise + ) } enum NegotiatedProtocol { @@ -159,50 +170,42 @@ extension HTTPConnectionPool.ConnectionFactory { connectionID: HTTPConnectionPool.Connection.ID, deadline: NIODeadline, eventLoop: EventLoop, - logger: Logger - ) -> EventLoopFuture { - let channelFuture: EventLoopFuture - + logger: Logger, + promise: EventLoopPromise + ) { if self.key.scheme.isProxyable, let proxy = self.clientConfiguration.proxy { switch proxy.type { case .socks: - channelFuture = self.makeSOCKSProxyChannel( + self.makeSOCKSProxyChannel( proxy, requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop, - logger: logger + logger: logger, + promise: promise ) case .http: - channelFuture = self.makeHTTPProxyChannel( + self.makeHTTPProxyChannel( proxy, requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop, - logger: logger + logger: logger, + promise: promise ) } } else { - channelFuture = self.makeNonProxiedChannel( + self.makeNonProxiedChannel( requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop, - logger: logger + logger: logger, + promise: promise ) } - - // let's map `ChannelError.connectTimeout` into a `HTTPClientError.connectTimeout` - return channelFuture.flatMapErrorThrowing { error throws -> NegotiatedProtocol in - switch error { - case ChannelError.connectTimeout: - throw HTTPClientError.connectTimeout - default: - throw error - } - } } private func makeNonProxiedChannel( @@ -210,29 +213,27 @@ extension HTTPConnectionPool.ConnectionFactory { connectionID: HTTPConnectionPool.Connection.ID, deadline: NIODeadline, eventLoop: EventLoop, - logger: Logger - ) -> EventLoopFuture { + logger: Logger, + promise: EventLoopPromise + ) { switch self.key.scheme { case .http, .httpUnix, .unix: - return self.makePlainChannel( + self.makePlainChannel( requester: requester, connectionID: connectionID, deadline: deadline, - eventLoop: eventLoop - ).map { .http1_1($0) } + eventLoop: eventLoop, + promise: promise + ) case .https, .httpsUnix: - return self.makeTLSChannel( + self.makeTLSChannel( requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop, - logger: logger - ).flatMapThrowing { - channel, - negotiated in - - try self.matchALPNToHTTPVersion(negotiated, channel: channel) - } + logger: logger, + promise: promise + ) } } @@ -240,15 +241,18 @@ extension HTTPConnectionPool.ConnectionFactory { requester: Requester, connectionID: HTTPConnectionPool.Connection.ID, deadline: NIODeadline, - eventLoop: EventLoop - ) -> EventLoopFuture { + eventLoop: EventLoop, + promise: EventLoopPromise + ) { precondition(!self.key.scheme.usesTLS, "Unexpected scheme") return self.makePlainBootstrap( requester: requester, connectionID: connectionID, deadline: deadline, eventLoop: eventLoop - ).connect(target: self.key.connectionTarget) + ).connect(target: self.key.connectionTarget).map { + .http1_1($0) + }.cascade(to: promise) } private func makeHTTPProxyChannel( @@ -257,8 +261,9 @@ extension HTTPConnectionPool.ConnectionFactory { connectionID: HTTPConnectionPool.Connection.ID, deadline: NIODeadline, eventLoop: EventLoop, - logger: Logger - ) -> EventLoopFuture { + logger: Logger, + promise: EventLoopPromise + ) { // A proxy connection starts with a plain text connection to the proxy server. After // the connection has been established with the proxy server, the connection might be // upgraded to TLS before we send our first request. @@ -268,34 +273,39 @@ extension HTTPConnectionPool.ConnectionFactory { deadline: deadline, eventLoop: eventLoop ) - return bootstrap.connect(host: proxy.host, port: proxy.port).flatMap { channel in - let encoder = HTTPRequestEncoder() - let decoder = ByteToMessageHandler(HTTPResponseDecoder(leftOverBytesStrategy: .dropBytes)) - let proxyHandler = HTTP1ProxyConnectHandler( - target: self.key.connectionTarget, - proxyAuthorization: proxy.authorization, - deadline: deadline - ) + bootstrap.connect(host: proxy.host, port: proxy.port).whenComplete { result in + switch result { + case .success(let channel): + let encoder = HTTPRequestEncoder() + let decoder = ByteToMessageHandler(HTTPResponseDecoder(leftOverBytesStrategy: .dropBytes)) + let proxyHandler = HTTP1ProxyConnectHandler( + target: self.key.connectionTarget, + proxyAuthorization: proxy.authorization, + deadline: deadline + ) - do { - try channel.pipeline.syncOperations.addHandler(encoder) - try channel.pipeline.syncOperations.addHandler(decoder) - try channel.pipeline.syncOperations.addHandler(proxyHandler) - } catch { - return channel.eventLoop.makeFailedFuture(error) - } + do { + try channel.pipeline.syncOperations.addHandler(encoder) + try channel.pipeline.syncOperations.addHandler(decoder) + try channel.pipeline.syncOperations.addHandler(proxyHandler) + } catch { + return promise.fail(error) + } - // The proxyEstablishedFuture is set as soon as the HTTP1ProxyConnectHandler is in a - // pipeline. It is created in HTTP1ProxyConnectHandler's handlerAdded method. - return proxyHandler.proxyEstablishedFuture!.assumeIsolated().flatMap { - channel.pipeline.syncOperations.removeHandler(proxyHandler).assumeIsolated().flatMap { - channel.pipeline.syncOperations.removeHandler(decoder).assumeIsolated().flatMap { - channel.pipeline.syncOperations.removeHandler(encoder) + // The proxyEstablishedFuture is set as soon as the HTTP1ProxyConnectHandler is in a + // pipeline. It is created in HTTP1ProxyConnectHandler's handlerAdded method. + return proxyHandler.proxyEstablishedFuture!.assumeIsolated().flatMap { + channel.pipeline.syncOperations.removeHandler(proxyHandler).assumeIsolated().flatMap { + channel.pipeline.syncOperations.removeHandler(decoder).assumeIsolated().flatMap { + channel.pipeline.syncOperations.removeHandler(encoder) + }.nonisolated() }.nonisolated() - }.nonisolated() - }.flatMap { - self.setupTLSInProxyConnectionIfNeeded(channel, deadline: deadline, logger: logger) - }.nonisolated() + }.flatMap { + self.setupTLSInProxyConnectionIfNeeded(channel, deadline: deadline, logger: logger) + }.nonisolated().cascade(to: promise) + case .failure(let error): + promise.fail(error) + } } } @@ -305,8 +315,9 @@ extension HTTPConnectionPool.ConnectionFactory { connectionID: HTTPConnectionPool.Connection.ID, deadline: NIODeadline, eventLoop: EventLoop, - logger: Logger - ) -> EventLoopFuture { + logger: Logger, + promise: EventLoopPromise + ) { // A proxy connection starts with a plain text connection to the proxy server. After // the connection has been established with the proxy server, the connection might be // upgraded to TLS before we send our first request. @@ -316,26 +327,32 @@ extension HTTPConnectionPool.ConnectionFactory { deadline: deadline, eventLoop: eventLoop ) - return bootstrap.connect(host: proxy.host, port: proxy.port).flatMap { channel in - let socksConnectHandler = SOCKSClientHandler(targetAddress: SOCKSAddress(self.key.connectionTarget)) - let socksEventHandler = SOCKSEventsHandler(deadline: deadline) - - do { - try channel.pipeline.syncOperations.addHandler(socksConnectHandler) - try channel.pipeline.syncOperations.addHandler(socksEventHandler) - } catch { - return channel.eventLoop.makeFailedFuture(error) + bootstrap.connect(host: proxy.host, port: proxy.port).whenComplete { result in + switch result { + case .success(let channel): + let socksConnectHandler = SOCKSClientHandler(targetAddress: SOCKSAddress(self.key.connectionTarget)) + let socksEventHandler = SOCKSEventsHandler(deadline: deadline) + + do { + try channel.pipeline.syncOperations.addHandler(socksConnectHandler) + try channel.pipeline.syncOperations.addHandler(socksEventHandler) + } catch { + return promise.fail(error) + } + + // The socksEstablishedFuture is set as soon as the SOCKSEventsHandler is in a + // pipeline. It is created in SOCKSEventsHandler's handlerAdded method. + socksEventHandler.socksEstablishedFuture!.assumeIsolated().flatMap { + channel.pipeline.syncOperations.removeHandler(socksEventHandler).assumeIsolated().flatMap { + channel.pipeline.syncOperations.removeHandler(socksConnectHandler) + }.nonisolated() + }.flatMap { + self.setupTLSInProxyConnectionIfNeeded(channel, deadline: deadline, logger: logger) + }.nonisolated().cascade(to: promise) + case .failure(let error): + promise.fail(error) } - // The socksEstablishedFuture is set as soon as the SOCKSEventsHandler is in a - // pipeline. It is created in SOCKSEventsHandler's handlerAdded method. - return socksEventHandler.socksEstablishedFuture!.assumeIsolated().flatMap { - channel.pipeline.syncOperations.removeHandler(socksEventHandler).assumeIsolated().flatMap { - channel.pipeline.syncOperations.removeHandler(socksConnectHandler) - }.nonisolated() - }.flatMap { - self.setupTLSInProxyConnectionIfNeeded(channel, deadline: deadline, logger: logger) - }.nonisolated() } } @@ -390,7 +407,7 @@ extension HTTPConnectionPool.ConnectionFactory { let sync = channel.pipeline.syncOperations let context = try sync.context(handlerType: TLSEventsHandler.self) return sync.removeHandler(context: context).flatMapThrowing { - try self.matchALPNToHTTPVersion(negotiated, channel: channel) + try Self.matchALPNToHTTPVersion(negotiated, channel: channel) } } catch { return channel.eventLoop.makeFailedFuture(error) @@ -449,8 +466,9 @@ extension HTTPConnectionPool.ConnectionFactory { connectionID: HTTPConnectionPool.Connection.ID, deadline: NIODeadline, eventLoop: EventLoop, - logger: Logger - ) -> EventLoopFuture<(Channel, String?)> { + logger: Logger, + promise: EventLoopPromise + ) { precondition(self.key.scheme.usesTLS, "Unexpected scheme") let bootstrapFuture = self.makeTLSBootstrap( requester: requester, @@ -460,36 +478,42 @@ extension HTTPConnectionPool.ConnectionFactory { logger: logger ) - var channelFuture = bootstrapFuture.flatMap { bootstrap -> EventLoopFuture in - bootstrap.connect(target: self.key.connectionTarget) - }.flatMap { channel -> EventLoopFuture<(Channel, String?)> in - do { - // if the channel is closed before flatMap is executed, all ChannelHandler are removed - // and TLSEventsHandler is therefore not present either - let tlsEventHandler = try channel.pipeline.syncOperations.handler(type: TLSEventsHandler.self) - - // The tlsEstablishedFuture is set as soon as the TLSEventsHandler is in a - // pipeline. It is created in TLSEventsHandler's handlerAdded method. - return tlsEventHandler.tlsEstablishedFuture!.assumeIsolated().flatMap { negotiated in - channel.pipeline.syncOperations.removeHandler(tlsEventHandler).map { (channel, negotiated) } - }.nonisolated() - } catch { - assert( - channel.isActive == false, - "if the channel is still active then TLSEventsHandler must be present but got error \(error)" - ) - return channel.eventLoop.makeFailedFuture(HTTPClientError.remoteConnectionClosed) + bootstrapFuture.whenComplete { result in + switch result { + case .success(let bootstrap): + bootstrap.connect(target: self.key.connectionTarget).flatMap { + channel -> EventLoopFuture<(Channel, String?)> in + do { + // if the channel is closed before flatMap is executed, all ChannelHandler are removed + // and TLSEventsHandler is therefore not present either + let tlsEventHandler = try channel.pipeline.syncOperations.handler(type: TLSEventsHandler.self) + + // The tlsEstablishedFuture is set as soon as the TLSEventsHandler is in a + // pipeline. It is created in TLSEventsHandler's handlerAdded method. + return tlsEventHandler.tlsEstablishedFuture!.assumeIsolated().flatMap { negotiated in + channel.pipeline.syncOperations.removeHandler(tlsEventHandler).map { (channel, negotiated) } + }.nonisolated() + } catch { + assert( + channel.isActive == false, + "if the channel is still active then TLSEventsHandler must be present but got error \(error)" + ) + return channel.eventLoop.makeFailedFuture(HTTPClientError.remoteConnectionClosed) + } + }.flatMapThrowing { channel, alpn in + try Self.matchALPNToHTTPVersion(alpn, channel: channel) + }.flatMapErrorThrowing { error in + // If NIOTransportSecurity is used, we want to map NWErrors into NWPOsixErrors or NWTLSError. + #if canImport(Network) + throw HTTPClient.NWErrorHandler.translateError(error) + #else + throw error + #endif + }.cascade(to: promise) + case .failure(let error): + promise.fail(error) } } - - #if canImport(Network) - // If NIOTransportSecurity is used, we want to map NWErrors into NWPOsixErrors or NWTLSError. - channelFuture = channelFuture.flatMapErrorThrowing { error in - throw HTTPClient.NWErrorHandler.translateError(error) - } - #endif - - return channelFuture } private func makeTLSBootstrap( @@ -582,7 +606,7 @@ extension HTTPConnectionPool.ConnectionFactory { } } - private func matchALPNToHTTPVersion(_ negotiated: String?, channel: Channel) throws -> NegotiatedProtocol { + private static func matchALPNToHTTPVersion(_ negotiated: String?, channel: Channel) throws -> NegotiatedProtocol { switch negotiated { case .none, .some("http/1.1"): return .http1_1(channel) diff --git a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1Connections.swift b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1Connections.swift index 87dcc03b6..3cdf51869 100644 --- a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1Connections.swift +++ b/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 diff --git a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift index 34c8027e9..395064377 100644 --- a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP1StateMachine.swift +++ b/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( diff --git a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift index dbb6b2d30..2a0e0cc80 100644 --- a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2Connections.swift +++ b/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 { diff --git a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift b/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift index 2372cab4b..67a07e6dd 100644 --- a/Sources/AsyncHTTPClient/ConnectionPool/State Machine/HTTPConnectionPool+HTTP2StateMachine.swift +++ b/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, diff --git a/Sources/AsyncHTTPClient/DeconstructedURL.swift b/Sources/AsyncHTTPClient/DeconstructedURL.swift index 52042bce3..f7d0b1977 100644 --- a/Sources/AsyncHTTPClient/DeconstructedURL.swift +++ b/Sources/AsyncHTTPClient/DeconstructedURL.swift @@ -48,7 +48,7 @@ extension DeconstructedURL { switch scheme { case .http, .https: - #if !canImport(Darwin) && compiler(>=6.0) + #if !canImport(Darwin) guard let urlHost = url.host, !urlHost.isEmpty else { throw HTTPClientError.emptyHost } @@ -89,7 +89,7 @@ extension DeconstructedURL { } } -#if !canImport(Darwin) && compiler(>=6.0) +#if !canImport(Darwin) extension String { @inlinable internal func trimIPv6Brackets() -> String { var utf8View = self.utf8[...] diff --git a/Sources/AsyncHTTPClient/HTTPClient+StructuredConcurrency.swift b/Sources/AsyncHTTPClient/HTTPClient+StructuredConcurrency.swift index f7d471f10..98c6555da 100644 --- a/Sources/AsyncHTTPClient/HTTPClient+StructuredConcurrency.swift +++ b/Sources/AsyncHTTPClient/HTTPClient+StructuredConcurrency.swift @@ -17,7 +17,6 @@ import NIO @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) extension HTTPClient { - #if compiler(>=6.0) /// Start & automatically shut down a new ``HTTPClient``. /// /// This method allows to start & automatically dispose of a ``HTTPClient`` following the principle of Structured Concurrency. @@ -43,30 +42,4 @@ extension HTTPClient { try await httpClient.shutdown() } } - #else - /// Start & automatically shut down a new ``HTTPClient``. - /// - /// This method allows to start & automatically dispose of a ``HTTPClient`` following the principle of Structured Concurrency. - /// The ``HTTPClient`` is guaranteed to be shut down upon return, whether `body` throws or not. - /// - /// This may be particularly useful if you cannot use the shared singleton (``HTTPClient/shared``). - public static func withHTTPClient( - eventLoopGroup: any EventLoopGroup = HTTPClient.defaultEventLoopGroup, - configuration: Configuration = Configuration(), - backgroundActivityLogger: Logger? = nil, - _ body: (HTTPClient) async throws -> Return - ) async throws -> Return { - let logger = (backgroundActivityLogger ?? HTTPClient.loggingDisabled) - let httpClient = HTTPClient( - eventLoopGroup: eventLoopGroup, - configuration: configuration, - backgroundActivityLogger: logger - ) - return try await asyncDo { - try await body(httpClient) - } finally: { _ in - try await httpClient.shutdown() - } - } - #endif } diff --git a/Sources/AsyncHTTPClient/HTTPClient.swift b/Sources/AsyncHTTPClient/HTTPClient.swift index fdb453e7e..80df3b946 100644 --- a/Sources/AsyncHTTPClient/HTTPClient.swift +++ b/Sources/AsyncHTTPClient/HTTPClient.swift @@ -1095,7 +1095,11 @@ public final class HTTPClient: Sendable { package var attributeKeys: AttributeKeys public init() { - self._tracer = nil + if #available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *) { + self._tracer = InstrumentationSystem.tracer + } else { + self._tracer = nil + } self.attributeKeys = .init() } diff --git a/Sources/AsyncHTTPClient/HTTPHandler.swift b/Sources/AsyncHTTPClient/HTTPHandler.swift index f9b337565..20df597ca 100644 --- a/Sources/AsyncHTTPClient/HTTPHandler.swift +++ b/Sources/AsyncHTTPClient/HTTPHandler.swift @@ -13,6 +13,7 @@ //===----------------------------------------------------------------------===// import Algorithms +import Foundation import Logging import NIOConcurrencyHelpers import NIOCore @@ -21,12 +22,6 @@ import NIOPosix import NIOSSL import Tracing -#if compiler(>=6.0) -import Foundation -#else -@preconcurrency import Foundation -#endif - extension HTTPClient { /// A request body. public struct Body: Sendable { diff --git a/Sources/AsyncHTTPClient/StructuredConcurrencyHelpers.swift b/Sources/AsyncHTTPClient/StructuredConcurrencyHelpers.swift index 25f1225e0..071e93d36 100644 --- a/Sources/AsyncHTTPClient/StructuredConcurrencyHelpers.swift +++ b/Sources/AsyncHTTPClient/StructuredConcurrencyHelpers.swift @@ -15,14 +15,13 @@ // Note: Whitespace changes are used to workaround compiler bug // https://github.com/swiftlang/swift/issues/79285 -#if compiler(>=6.0) @inlinable @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) internal func asyncDo( isolation: isolated (any Actor)? = #isolation, - // DO NOT FIX THE WHITESPACE IN THE NEXT LINE UNTIL 5.10 IS UNSUPPORTED - // https://github.com/swiftlang/swift/issues/79285 - _ body: () async throws -> sending R, finally: sending @escaping ((any Error)?) async throws -> Void) async throws -> sending R { + _ body: () async throws -> sending R, + finally: sending @escaping ((any Error)?) async throws -> Void +) async throws -> sending R { let result: R do { result = try await body() @@ -48,36 +47,3 @@ internal func asyncDo( }.value return result } -#else -@inlinable -@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) -internal func asyncDo( - _ body: () async throws -> R, - finally: @escaping @Sendable ((any Error)?) async throws -> Void -) async throws -> R { - let result: R - do { - result = try await body() - } catch { - // `body` failed, we need to invoke `finally` with the `error`. - - // This _looks_ unstructured but isn't really because we unconditionally always await the return. - // We need to have an uncancelled task here to assure this is actually running in case we hit a - // cancellation error. - try await Task { - try await finally(error) - }.value - throw error - } - - // `body` succeeded, we need to invoke `finally` with `nil` (no error). - - // This _looks_ unstructured but isn't really because we unconditionally always await the return. - // We need to have an uncancelled task here to assure this is actually running in case we hit a - // cancellation error. - try await Task { - try await finally(nil) - }.value - return result -} -#endif diff --git a/Tests/AsyncHTTPClientTests/AsyncTestHelpers.swift b/Tests/AsyncHTTPClientTests/AsyncTestHelpers.swift index 4a5c8d486..5e063be81 100644 --- a/Tests/AsyncHTTPClientTests/AsyncTestHelpers.swift +++ b/Tests/AsyncHTTPClientTests/AsyncTestHelpers.swift @@ -43,12 +43,11 @@ final class AsyncSequenceWriter: AsyncSequence, @unchecked Se case failed(Error, CheckedContinuation?) } - private var _state = State.buffering(.init(), nil) - private let lock = NIOLock() + private let state = NIOLockedValueBox(.buffering([], nil)) public var hasDemand: Bool { - self.lock.withLock { - switch self._state { + self.state.withLockedValue { state in + switch state { case .failed, .finished, .buffering: return false case .waiting: @@ -59,67 +58,132 @@ final class AsyncSequenceWriter: AsyncSequence, @unchecked Se /// Wait until a downstream consumer has issued more demand by calling `next`. public func demand() async { - self.lock.lock() + let shouldBuffer = self.state.withLockedValue { state in + switch state { + case .buffering(_, .none): + return true + case .waiting: + return false + case .buffering(_, .some), .failed(_, .some): + preconditionFailure("Already waiting for demand. Invalid state: \(state)") + case .finished, .failed: + preconditionFailure("Invalid state: \(state)") + } + } - switch self._state { - case .buffering(let buffer, .none): + if shouldBuffer { await withCheckedContinuation { (continuation: CheckedContinuation) in - self._state = .buffering(buffer, continuation) - self.lock.unlock() + let shouldResumeContinuation = self.state.withLockedValue { state in + switch state { + case .buffering(let buffer, .none): + state = .buffering(buffer, continuation) + return false + case .waiting: + return true + case .buffering(_, .some), .failed(_, .some): + preconditionFailure("Already waiting for demand. Invalid state: \(state)") + case .finished, .failed: + preconditionFailure("Invalid state: \(state)") + } + } + + if shouldResumeContinuation { + continuation.resume() + } } - - case .waiting: - self.lock.unlock() - return - - case .buffering(_, .some), .failed(_, .some): - let state = self._state - self.lock.unlock() - preconditionFailure("Already waiting for demand. Invalid state: \(state)") - - case .finished, .failed: - let state = self._state - self.lock.unlock() - preconditionFailure("Invalid state: \(state)") } } + private enum NextAction { + /// Resume the continuation if present, and return the result if present. + case resumeAndReturn(CheckedContinuation?, Result?) + /// Suspend the current task and wait for the next value. + case suspend + } + private func next() async throws -> Element? { - self.lock.lock() - switch self._state { - case .buffering(let buffer, let demandContinuation) where buffer.isEmpty: - return try await withCheckedThrowingContinuation { continuation in - self._state = .waiting(continuation) - self.lock.unlock() - demandContinuation?.resume(returning: ()) - } + let action: NextAction = self.state.withLockedValue { state in + switch state { + case .buffering(var buffer, let demandContinuation): + if buffer.isEmpty { + return .suspend + } else { + let first = buffer.removeFirst() + if first != nil { + state = .buffering(buffer, demandContinuation) + } else { + state = .finished + } + return .resumeAndReturn(nil, .success(first)) + } + + case .failed(let error, let demandContinuation): + state = .finished + return .resumeAndReturn(demandContinuation, .failure(error)) + + case .finished: + return .resumeAndReturn(nil, .success(nil)) - case .buffering(var buffer, let demandContinuation): - let first = buffer.removeFirst() - if first != nil { - self._state = .buffering(buffer, demandContinuation) - } else { - self._state = .finished + case .waiting: + preconditionFailure( + "Expected that there is always only one concurrent call to next. Invalid state: \(state)" + ) } - self.lock.unlock() - return first + } - case .failed(let error, let demandContinuation): - self._state = .finished - self.lock.unlock() + switch action { + case .resumeAndReturn(let demandContinuation, let result): demandContinuation?.resume() - throw error - - case .finished: - self.lock.unlock() - return nil - - case .waiting: - let state = self._state - self.lock.unlock() - preconditionFailure( - "Expected that there is always only one concurrent call to next. Invalid state: \(state)" - ) + return try result?.get() + + case .suspend: + // Holding the lock here *should* be safe but because of a bug in the runtime + // it isn't, so drop the lock, create the continuation and then try again. + // + // See https://github.com/swiftlang/swift/issues/85668 + return try await withCheckedThrowingContinuation { + (continuation: CheckedContinuation) in + let action: NextAction = self.state.withLockedValue { state in + switch state { + case .buffering(var buffer, let demandContinuation): + if buffer.isEmpty { + state = .waiting(continuation) + return .resumeAndReturn(demandContinuation, nil) + } else { + let first = buffer.removeFirst() + if first != nil { + state = .buffering(buffer, demandContinuation) + } else { + state = .finished + } + return .resumeAndReturn(nil, .success(first)) + } + + case .failed(let error, let demandContinuation): + state = .finished + return .resumeAndReturn(demandContinuation, .failure(error)) + + case .finished: + return .resumeAndReturn(nil, .success(nil)) + + case .waiting: + preconditionFailure( + "Expected that there is always only one concurrent call to next. Invalid state: \(state)" + ) + } + } + + switch action { + case .resumeAndReturn(let demandContinuation, let result): + demandContinuation?.resume() + // Resume the continuation rather than returning th result. + if let result { + continuation.resume(with: result) + } + case .suspend: + preconditionFailure() // Not returned from the code above. + } + } } } @@ -137,19 +201,19 @@ final class AsyncSequenceWriter: AsyncSequence, @unchecked Se } private func writeBufferOrEnd(_ element: Element?) { - let writeAction = self.lock.withLock { () -> WriteAction in - switch self._state { + let writeAction = self.state.withLockedValue { state -> WriteAction in + switch state { case .buffering(var buffer, let continuation): buffer.append(element) - self._state = .buffering(buffer, continuation) + state = .buffering(buffer, continuation) return .none case .waiting(let continuation): - self._state = .buffering(.init(), nil) + state = .buffering(.init(), nil) return .succeedContinuation(continuation, element) case .finished, .failed: - preconditionFailure("Invalid state: \(self._state)") + preconditionFailure("Invalid state: \(state)") } } @@ -170,17 +234,17 @@ final class AsyncSequenceWriter: AsyncSequence, @unchecked Se /// Drops all buffered writes and emits an error on the waiting `next`. If there is no call to `next` /// waiting, will emit the error on the next call to `next`. public func fail(_ error: Error) { - let errorAction = self.lock.withLock { () -> ErrorAction in - switch self._state { + let errorAction = self.state.withLockedValue { state -> ErrorAction in + switch state { case .buffering(_, let demandContinuation): - self._state = .failed(error, demandContinuation) + state = .failed(error, demandContinuation) return .none case .failed, .finished: return .none case .waiting(let continuation): - self._state = .finished + state = .finished return .failContinuation(continuation, error) } } diff --git a/Tests/AsyncHTTPClientTests/HTTPClient+SOCKSTests.swift b/Tests/AsyncHTTPClientTests/HTTPClient+SOCKSTests.swift index af32284b0..d5e1c895b 100644 --- a/Tests/AsyncHTTPClientTests/HTTPClient+SOCKSTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTPClient+SOCKSTests.swift @@ -13,6 +13,7 @@ //===----------------------------------------------------------------------===// import AsyncHTTPClient // NOT @testable - tests that need @testable go into HTTPClientInternalTests.swift +import InMemoryLogging import Logging import NIOCore import NIOHTTP1 @@ -27,7 +28,7 @@ class HTTPClientSOCKSTests: XCTestCase { var serverGroup: EventLoopGroup! var defaultHTTPBin: HTTPBin! var defaultClient: HTTPClient! - var backgroundLogStore: CollectEverythingLogHandler.LogStore! + var backgroundLogStore: InMemoryLogHandler! var defaultHTTPBinURLPrefix: String { "http://localhost:\(self.defaultHTTPBin.port)/" @@ -43,14 +44,8 @@ class HTTPClientSOCKSTests: XCTestCase { self.clientGroup = getDefaultEventLoopGroup(numberOfThreads: 1) self.serverGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1) self.defaultHTTPBin = HTTPBin() - self.backgroundLogStore = CollectEverythingLogHandler.LogStore() - var backgroundLogger = Logger( - label: "\(#function)", - factory: { _ in - CollectEverythingLogHandler(logStore: self.backgroundLogStore!) - } - ) - backgroundLogger.logLevel = .trace + let (backgroundLogStore, backgroundLogger) = InMemoryLogHandler.makeLogger(logLevel: .trace) + self.backgroundLogStore = backgroundLogStore self.defaultClient = HTTPClient( eventLoopGroupProvider: .shared(self.clientGroup), backgroundActivityLogger: backgroundLogger diff --git a/Tests/AsyncHTTPClientTests/HTTPClientBase.swift b/Tests/AsyncHTTPClientTests/HTTPClientBase.swift index 15620dd24..90ab12fe5 100644 --- a/Tests/AsyncHTTPClientTests/HTTPClientBase.swift +++ b/Tests/AsyncHTTPClientTests/HTTPClientBase.swift @@ -14,6 +14,7 @@ import AsyncHTTPClient import Atomics +import InMemoryLogging import Logging import NIOConcurrencyHelpers import NIOCore @@ -37,7 +38,7 @@ class XCTestCaseHTTPClientTestsBaseClass: XCTestCase { var serverGroup: EventLoopGroup! var defaultHTTPBin: HTTPBin! var defaultClient: HTTPClient! - var backgroundLogStore: CollectEverythingLogHandler.LogStore! + var backgroundLogStore: InMemoryLogHandler! var defaultHTTPBinURLPrefix: String { self.defaultHTTPBin.baseURL @@ -53,14 +54,8 @@ class XCTestCaseHTTPClientTestsBaseClass: XCTestCase { self.clientGroup = getDefaultEventLoopGroup(numberOfThreads: 1) self.serverGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1) self.defaultHTTPBin = HTTPBin() - self.backgroundLogStore = CollectEverythingLogHandler.LogStore() - var backgroundLogger = Logger( - label: "\(#function)", - factory: { _ in - CollectEverythingLogHandler(logStore: self.backgroundLogStore!) - } - ) - backgroundLogger.logLevel = .trace + let (backgroundLogStore, backgroundLogger) = InMemoryLogHandler.makeLogger(logLevel: .trace) + self.backgroundLogStore = backgroundLogStore self.defaultClient = HTTPClient( eventLoopGroupProvider: .shared(self.clientGroup), diff --git a/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift b/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift index f9917c885..689b4358e 100644 --- a/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift +++ b/Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift @@ -14,6 +14,7 @@ import Atomics import Foundation +import InMemoryLogging import Logging import NIOConcurrencyHelpers import NIOCore @@ -359,7 +360,13 @@ enum TestTLS { ) } -internal final class HTTPBin: Sendable +#if compiler(>=6.2) +typealias AHCTestSendableMetatype = SendableMetatype +#else +typealias AHCTestSendableMetatype = Any +#endif + +internal final class HTTPBin: Sendable where RequestHandler.InboundIn == HTTPServerRequestPart, RequestHandler.OutboundOut == HTTPServerResponsePart @@ -1284,65 +1291,21 @@ extension EventLoopFuture where Value: Sendable { } } -struct CollectEverythingLogHandler: LogHandler { - var metadata: Logger.Metadata = [:] - var logLevel: Logger.Level = .info - let logStore: LogStore - - final class LogStore: Sendable { - struct Entry { - var level: Logger.Level - var message: String - var metadata: [String: String] - } - - private let logs = NIOLockedValueBox<[Entry]>([]) - - var allEntries: [Entry] { - get { - self.logs.withLockedValue { $0 } - } - set { - self.logs.withLockedValue { $0 = newValue } - } - } - - func append(level: Logger.Level, message: Logger.Message, metadata: Logger.Metadata?) { - self.logs.withLockedValue { - $0.append( - Entry( - level: level, - message: message.description, - metadata: metadata?.mapValues { $0.description } ?? [:] - ) - ) +extension InMemoryLogHandler { + static func makeLogger( + logLevel: Logger.Level = .info, + function: String = #function + ) -> (InMemoryLogHandler, Logger) { + let handler = InMemoryLogHandler() + + var logger = Logger( + label: "\(function)", + factory: { _ in + handler } - } - } - - init(logStore: LogStore) { - self.logStore = logStore - } - - func log( - level: Logger.Level, - message: Logger.Message, - metadata: Logger.Metadata?, - source: String, - file: String, - function: String, - line: UInt - ) { - self.logStore.append(level: level, message: message, metadata: self.metadata.merging(metadata ?? [:]) { $1 }) - } - - subscript(metadataKey key: String) -> Logger.Metadata.Value? { - get { - self.metadata[key] - } - set { - self.metadata[key] = newValue - } + ) + logger.logLevel = logLevel + return (handler, logger) } } diff --git a/Tests/AsyncHTTPClientTests/HTTPClientTests.swift b/Tests/AsyncHTTPClientTests/HTTPClientTests.swift index 50c3ecb9d..054cf3487 100644 --- a/Tests/AsyncHTTPClientTests/HTTPClientTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTPClientTests.swift @@ -14,6 +14,7 @@ import AsyncHTTPClient // NOT @testable - tests that need @testable go into HTTPClientInternalTests.swift import Atomics +import InMemoryLogging import Logging import NIOConcurrencyHelpers import NIOCore @@ -46,7 +47,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { let request3 = try Request(url: "unix:///tmp/file") XCTAssertEqual(request3.host, "") - #if os(Linux) && compiler(>=6.0) && compiler(<6.1) + #if os(Linux) && compiler(<6.1) XCTAssertEqual(request3.url.host, "") #else XCTAssertNil(request3.url.host) @@ -2751,15 +2752,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { } func testLoggingCorrectlyAttachesRequestInformationEvenAfterDuringRedirect() { - let logStore = CollectEverythingLogHandler.LogStore() - - var logger = Logger( - label: "\(#function)", - factory: { _ in - CollectEverythingLogHandler(logStore: logStore) - } - ) - logger.logLevel = .trace + var (logStore, logger) = InMemoryLogHandler.makeLogger(logLevel: .trace) logger[metadataKey: "custom-request-id"] = "abcd" var maybeRequest: HTTPClient.Request? @@ -2782,7 +2775,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { logger: logger ).wait() ) - let logs = logStore.allEntries + let logs = logStore.entries XCTAssertTrue(logs.allSatisfy { $0.metadata["custom-request-id"] == "abcd" }) @@ -2804,12 +2797,12 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { } func testLoggingCorrectlyAttachesRequestInformation() { - let logStore = CollectEverythingLogHandler.LogStore() + let logStore = InMemoryLogHandler() var loggerYolo001 = Logger( label: "\(#function)", factory: { _ in - CollectEverythingLogHandler(logStore: logStore) + logStore } ) loggerYolo001.logLevel = .trace @@ -2817,7 +2810,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { var loggerACME002 = Logger( label: "\(#function)", factory: { _ in - CollectEverythingLogHandler(logStore: logStore) + logStore } ) loggerACME002.logLevel = .trace @@ -2840,8 +2833,8 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { logger: loggerYolo001 ).wait() ) - let logsAfterReq1 = logStore.allEntries - logStore.allEntries = [] + let logsAfterReq1 = logStore.entries + logStore.clear() // === Request 2 (Yolo001) XCTAssertNoThrow( @@ -2852,8 +2845,8 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { logger: loggerYolo001 ).wait() ) - let logsAfterReq2 = logStore.allEntries - logStore.allEntries = [] + let logsAfterReq2 = logStore.entries + logStore.clear() // === Request 3 (ACME002) XCTAssertNoThrow( @@ -2864,8 +2857,8 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { logger: loggerACME002 ).wait() ) - let logsAfterReq3 = logStore.allEntries - logStore.allEntries = [] + let logsAfterReq3 = logStore.entries + logStore.clear() // === Assertions XCTAssertGreaterThan(logsAfterReq1.count, 0) @@ -2879,7 +2872,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { { XCTAssertNil(entry.metadata["acme-request-id"]) XCTAssertEqual("yolo-001", yoloRequestID) - XCTAssertNotNil(Int(httpRequestMetadata)) + XCTAssertNotNil(Int("\(httpRequestMetadata)")) return true } else { XCTFail("log message doesn't contain the right IDs: \(entry)") @@ -2912,7 +2905,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { { XCTAssertNil(entry.metadata["acme-request-id"]) XCTAssertEqual("yolo-001", yoloRequestID) - XCTAssertNotNil(Int(httpRequestMetadata)) + XCTAssertNotNil(Int("\(httpRequestMetadata)")) return true } else { XCTFail("log message doesn't contain the right IDs: \(entry)") @@ -2940,7 +2933,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { { XCTAssertNil(entry.metadata["yolo-request-id"]) XCTAssertEqual("acme-002", acmeRequestID) - XCTAssertNotNil(Int(httpRequestMetadata)) + XCTAssertNotNil(Int("\(httpRequestMetadata)")) return true } else { XCTFail("log message doesn't contain the right IDs: \(entry)") @@ -2963,15 +2956,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { } func testNothingIsLoggedAtInfoOrHigher() { - let logStore = CollectEverythingLogHandler.LogStore() - - var logger = Logger( - label: "\(#function)", - factory: { _ in - CollectEverythingLogHandler(logStore: logStore) - } - ) - logger.logLevel = .info + let (logStore, logger) = InMemoryLogHandler.makeLogger(logLevel: .info) guard let request1 = try? HTTPClient.Request(url: self.defaultHTTPBinURLPrefix + "get"), let request2 = try? HTTPClient.Request(url: self.defaultHTTPBinURLPrefix + "stats") @@ -2989,7 +2974,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { logger: logger ).wait() ) - XCTAssertEqual(0, logStore.allEntries.count) + XCTAssertEqual(0, logStore.entries.count) // === Request 2 XCTAssertNoThrow( @@ -3000,7 +2985,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { logger: logger ).wait() ) - XCTAssertEqual(0, logStore.allEntries.count) + XCTAssertEqual(0, logStore.entries.count) // === Synthesized Request XCTAssertNoThrow( @@ -3012,21 +2997,14 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { logger: logger ).wait() ) - XCTAssertEqual(0, logStore.allEntries.count) + XCTAssertEqual(0, logStore.entries.count) - XCTAssertEqual(0, self.backgroundLogStore.allEntries.filter { $0.level >= .info }.count) + XCTAssertEqual(0, self.backgroundLogStore.entries.filter { $0.level >= .info }.count) // === Synthesized Socket Path Request XCTAssertNoThrow( try TemporaryFileHelpers.withTemporaryUnixDomainSocketPathName { path in - let backgroundLogStore = CollectEverythingLogHandler.LogStore() - var backgroundLogger = Logger( - label: "\(#function)", - factory: { _ in - CollectEverythingLogHandler(logStore: backgroundLogStore) - } - ) - backgroundLogger.logLevel = .trace + let (backgroundLogStore, backgroundLogger) = InMemoryLogHandler.makeLogger(logLevel: .trace) let localSocketPathHTTPBin = HTTPBin(bindTarget: .unixDomainSocket(path)) let localClient = HTTPClient( @@ -3048,23 +3026,16 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { logger: logger ).wait() ) - XCTAssertEqual(0, logStore.allEntries.count) + XCTAssertEqual(0, logStore.entries.count) - XCTAssertEqual(0, backgroundLogStore.allEntries.filter { $0.level >= .info }.count) + XCTAssertEqual(0, backgroundLogStore.entries.filter { $0.level >= .info }.count) } ) // === Synthesized Secure Socket Path Request XCTAssertNoThrow( try TemporaryFileHelpers.withTemporaryUnixDomainSocketPathName { path in - let backgroundLogStore = CollectEverythingLogHandler.LogStore() - var backgroundLogger = Logger( - label: "\(#function)", - factory: { _ in - CollectEverythingLogHandler(logStore: backgroundLogStore) - } - ) - backgroundLogger.logLevel = .trace + let (backgroundLogStore, backgroundLogger) = InMemoryLogHandler.makeLogger(logLevel: .trace) let localSocketPathHTTPBin = HTTPBin(.http1_1(ssl: true), bindTarget: .unixDomainSocket(path)) let localClient = HTTPClient( @@ -3087,33 +3058,25 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { logger: logger ).wait() ) - XCTAssertEqual(0, logStore.allEntries.count) + XCTAssertEqual(0, logStore.entries.count) - XCTAssertEqual(0, backgroundLogStore.allEntries.filter { $0.level >= .info }.count) + XCTAssertEqual(0, backgroundLogStore.entries.filter { $0.level >= .info }.count) } ) } func testAllMethodsLog() { func checkExpectationsWithLogger(type: String, _ body: (Logger, String) throws -> T) throws -> T { - let logStore = CollectEverythingLogHandler.LogStore() - - var logger = Logger( - label: "\(#function)", - factory: { _ in - CollectEverythingLogHandler(logStore: logStore) - } - ) - logger.logLevel = .trace + var (logStore, logger) = InMemoryLogHandler.makeLogger(logLevel: .trace) logger[metadataKey: "req"] = "yo-\(type)" let url = "not-found/request/\(type))" let result = try body(logger, url) - XCTAssertGreaterThan(logStore.allEntries.count, 0) - for entry in logStore.allEntries { + XCTAssertGreaterThan(logStore.entries.count, 0) + for entry in logStore.entries { XCTAssertEqual("yo-\(type)", entry.metadata["req"] ?? "n/a") - XCTAssertNotNil(Int(entry.metadata["ahc-request-id"] ?? "n/a")) + XCTAssertNotNil(Int(entry.metadata["ahc-request-id"]?.description ?? "n/a")) } return result } @@ -3162,18 +3125,11 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { ) // No background activity expected here. - XCTAssertEqual(0, self.backgroundLogStore.allEntries.filter { $0.level >= .debug }.count) + XCTAssertEqual(0, self.backgroundLogStore.entries.filter { $0.level >= .debug }.count) XCTAssertNoThrow( try TemporaryFileHelpers.withTemporaryUnixDomainSocketPathName { path in - let backgroundLogStore = CollectEverythingLogHandler.LogStore() - var backgroundLogger = Logger( - label: "\(#function)", - factory: { _ in - CollectEverythingLogHandler(logStore: backgroundLogStore) - } - ) - backgroundLogger.logLevel = .trace + let (backgroundLogStore, backgroundLogger) = InMemoryLogHandler.makeLogger(logLevel: .trace) let localSocketPathHTTPBin = HTTPBin(bindTarget: .unixDomainSocket(path)) let localClient = HTTPClient( @@ -3193,20 +3149,13 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { ) // No background activity expected here. - XCTAssertEqual(0, backgroundLogStore.allEntries.filter { $0.level >= .debug }.count) + XCTAssertEqual(0, backgroundLogStore.entries.filter { $0.level >= .debug }.count) } ) XCTAssertNoThrow( try TemporaryFileHelpers.withTemporaryUnixDomainSocketPathName { path in - let backgroundLogStore = CollectEverythingLogHandler.LogStore() - var backgroundLogger = Logger( - label: "\(#function)", - factory: { _ in - CollectEverythingLogHandler(logStore: backgroundLogStore) - } - ) - backgroundLogger.logLevel = .trace + let (backgroundLogStore, backgroundLogger) = InMemoryLogHandler.makeLogger(logLevel: .trace) let localSocketPathHTTPBin = HTTPBin(.http1_1(ssl: true), bindTarget: .unixDomainSocket(path)) let localClient = HTTPClient( @@ -3227,7 +3176,7 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { ) // No background activity expected here. - XCTAssertEqual(0, backgroundLogStore.allEntries.filter { $0.level >= .debug }.count) + XCTAssertEqual(0, backgroundLogStore.entries.filter { $0.level >= .debug }.count) } ) } @@ -3237,14 +3186,14 @@ final class HTTPClientTests: XCTestCaseHTTPClientTestsBaseClass { XCTAssertNoThrow(try self.defaultClient.syncShutdown()) - XCTAssertGreaterThanOrEqual(self.backgroundLogStore.allEntries.count, 0) + XCTAssertGreaterThanOrEqual(self.backgroundLogStore.entries.count, 0) XCTAssert( - self.backgroundLogStore.allEntries.contains { entry in + self.backgroundLogStore.entries.contains { entry in entry.message == "Shutting down connection pool" } ) XCTAssert( - self.backgroundLogStore.allEntries.allSatisfy { entry in + self.backgroundLogStore.entries.allSatisfy { entry in entry.metadata["ahc-request-id"] == nil && entry.metadata["ahc-request"] == nil && entry.metadata["ahc-pool-key"] != nil } diff --git a/Tests/AsyncHTTPClientTests/HTTPClientTracingInternalTests.swift b/Tests/AsyncHTTPClientTests/HTTPClientTracingInternalTests.swift new file mode 100644 index 000000000..53f1138ba --- /dev/null +++ b/Tests/AsyncHTTPClientTests/HTTPClientTracingInternalTests.swift @@ -0,0 +1,77 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the AsyncHTTPClient open source project +// +// Copyright (c) 2025 Apple Inc. and the AsyncHTTPClient project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of AsyncHTTPClient project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +import Atomics +import InMemoryTracing +import Logging +import NIOConcurrencyHelpers +import NIOCore +import NIOEmbedded +import NIOFoundationCompat +import NIOHTTP1 +import NIOHTTPCompression +import NIOPosix +import NIOSSL +import NIOTestUtils +import NIOTransportServices +import Tracing +import XCTest + +@testable @_spi(Tracing) import AsyncHTTPClient + +#if canImport(Network) +import Network +#endif + +private func makeTracedHTTPClient(tracer: InMemoryTracer) -> HTTPClient { + var config = HTTPClient.Configuration() + config.httpVersion = .automatic + config.tracing.tracer = tracer + return HTTPClient( + eventLoopGroupProvider: .singleton, + configuration: config + ) +} + +final class HTTPClientTracingInternalTests: XCTestCaseHTTPClientTestsBaseClass { + + var tracer: InMemoryTracer! + var client: HTTPClient! + + override func setUp() { + super.setUp() + self.tracer = InMemoryTracer() + self.client = makeTracedHTTPClient(tracer: tracer) + } + + override func tearDown() { + if let client = self.client { + XCTAssertNoThrow(try client.syncShutdown()) + self.client = nil + } + tracer = nil + } + + func testTrace_preparedHeaders_include_fromSpan() async throws { + let url = self.defaultHTTPBinURLPrefix + "404-does-not-exist" + let request = HTTPClientRequest(url: url) + + try tracer.withSpan("operation") { span in + let prepared = try HTTPClientRequest.Prepared(request, tracing: self.client.tracing) + XCTAssertTrue(prepared.head.headers.count > 2) + XCTAssertTrue(prepared.head.headers.contains(name: "in-memory-trace-id")) + XCTAssertTrue(prepared.head.headers.contains(name: "in-memory-span-id")) + } + } +} diff --git a/Tests/AsyncHTTPClientTests/HTTPClientTracingTests.swift b/Tests/AsyncHTTPClientTests/HTTPClientTracingTests.swift index dd342c2db..047c66e6d 100644 --- a/Tests/AsyncHTTPClientTests/HTTPClientTracingTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTPClientTracingTests.swift @@ -12,7 +12,7 @@ // //===----------------------------------------------------------------------===// -@_spi(Tracing) import AsyncHTTPClient // NOT @testable - tests that need @testable go into HTTPClientInternalTests.swift +@_spi(Tracing) import AsyncHTTPClient // NOT @testable - tests that need @testable go into HTTPClientTracingInternalTests.swift import Atomics import InMemoryTracing import Logging diff --git a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+FactoryTests.swift b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+FactoryTests.swift index 15cc9e7e9..37ff3a1ef 100644 --- a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+FactoryTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+FactoryTests.swift @@ -57,7 +57,10 @@ class HTTPConnectionPool_FactoryTests: XCTestCase { logger: .init(label: "test") ).wait() ) { - XCTAssertEqual($0 as? HTTPClientError, .connectTimeout) + guard let error = $0 as? ChannelError, case .connectTimeout = error else { + XCTFail("Unexpected error: \($0)") + return + } } } @@ -210,3 +213,24 @@ final class ExplodingRequester: HTTPConnectionRequester { XCTFail("waitingForConnectivity called unexpectedly") } } + +extension HTTPConnectionPool.ConnectionFactory { + fileprivate func makeChannel( + requester: Requester, + connectionID: HTTPConnectionPool.Connection.ID, + deadline: NIODeadline, + eventLoop: EventLoop, + logger: Logger + ) -> EventLoopFuture { + let promise = eventLoop.makePromise(of: NegotiatedProtocol.self) + self.makeChannel( + requester: requester, + connectionID: connectionID, + deadline: deadline, + eventLoop: eventLoop, + logger: logger, + promise: promise + ) + return promise.futureResult + } +} diff --git a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1ConnectionsTest.swift b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1ConnectionsTest.swift index f225307ea..89f3bf7b5 100644 --- a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1ConnectionsTest.swift +++ b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1ConnectionsTest.swift @@ -384,6 +384,8 @@ class HTTPConnectionPool_HTTP1ConnectionsTests: XCTestCase { XCTAssertEqual(connections.closeConnection(at: releaseIndex), lease) XCTAssertFalse(connections.isEmpty) + let backoffEL = connections.backoffNextConnectionAttempt(startingID) + XCTAssertIdentical(backoffEL, el2) guard let (failIndex, _) = connections.failConnection(startingID) else { return XCTFail("Expected that the connection is remembered") } diff --git a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1StateTests.swift b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1StateTests.swift index 8dd59baaf..9146f0593 100644 --- a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1StateTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP1StateTests.swift @@ -1493,4 +1493,48 @@ class HTTPConnectionPool_HTTP1StateMachineTests: XCTestCase { // We won't bother doing it though, it's enough that it asked. } + + func testFailConnectionRacesAgainstConnectionCreationFailed() { + let elg = EmbeddedEventLoopGroup(loops: 4) + defer { XCTAssertNoThrow(try elg.syncShutdownGracefully()) } + + var state = HTTPConnectionPool.StateMachine( + idGenerator: .init(), + maximumConcurrentHTTP1Connections: 2, + retryConnectionEstablishment: true, + preferHTTP1: true, + maximumConnectionUses: nil, + preWarmedHTTP1ConnectionCount: 0 + ) + + let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next()) + let request = HTTPConnectionPool.Request(mockRequest) + + let executeAction = state.executeRequest(request) + XCTAssertEqual(.scheduleRequestTimeout(for: request, on: mockRequest.eventLoop), executeAction.request) + + // 1. connection attempt + guard case .createConnection(let connectionID, on: let connectionEL) = executeAction.connection else { + return XCTFail("Unexpected connection action: \(executeAction.connection)") + } + XCTAssert(connectionEL === mockRequest.eventLoop) // XCTAssertIdentical not available on Linux + + // 2. connection fails – first with closed callback + + XCTAssertEqual(state.http1ConnectionClosed(connectionID), .none) + + // 3. connection fails – with make connection callback + + let action = state.failedToCreateNewConnection( + IOError(errnoCode: -1, reason: "Test failure"), + connectionID: connectionID + ) + XCTAssertEqual(action.request, .none) + guard case .scheduleBackoffTimer(connectionID, _, on: let backoffTimerEL) = action.connection else { + XCTFail("Unexpected connection action: \(action.connection)") + return + } + XCTAssertIdentical(connectionEL, backoffTimerEL) + + } } diff --git a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2ConnectionsTest.swift b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2ConnectionsTest.swift index dd56a9102..dbfe90ff9 100644 --- a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2ConnectionsTest.swift +++ b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2ConnectionsTest.swift @@ -331,6 +331,8 @@ class HTTPConnectionPool_HTTP2ConnectionsTests: XCTestCase { XCTAssertEqual(connections.closeConnection(at: releaseIndex), leasedConn) XCTAssertFalse(connections.isEmpty) + let backoffEL = connections.backoffNextConnectionAttempt(startingID) + XCTAssertIdentical(el6, backoffEL) guard let (failIndex, _) = connections.failConnection(startingID) else { return XCTFail("Expected that the connection is remembered") } diff --git a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2StateMachineTests.swift b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2StateMachineTests.swift index d59dae796..8fead4f4d 100644 --- a/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2StateMachineTests.swift +++ b/Tests/AsyncHTTPClientTests/HTTPConnectionPool+HTTP2StateMachineTests.swift @@ -1527,6 +1527,50 @@ class HTTPConnectionPool_HTTP2StateMachineTests: XCTestCase { XCTAssertEqual(state.http2ConnectionClosed(connection.id), .none) } + + func testFailConnectionRacesAgainstConnectionCreationFailed() { + let elg = EmbeddedEventLoopGroup(loops: 4) + defer { XCTAssertNoThrow(try elg.syncShutdownGracefully()) } + + var state = HTTPConnectionPool.StateMachine( + idGenerator: .init(), + maximumConcurrentHTTP1Connections: 2, + retryConnectionEstablishment: true, + preferHTTP1: false, + maximumConnectionUses: nil, + preWarmedHTTP1ConnectionCount: 0 + ) + + let mockRequest = MockHTTPScheduableRequest(eventLoop: elg.next()) + let request = HTTPConnectionPool.Request(mockRequest) + + let executeAction = state.executeRequest(request) + XCTAssertEqual(.scheduleRequestTimeout(for: request, on: mockRequest.eventLoop), executeAction.request) + + // 1. connection attempt + guard case .createConnection(let connectionID, on: let connectionEL) = executeAction.connection else { + return XCTFail("Unexpected connection action: \(executeAction.connection)") + } + XCTAssert(connectionEL === mockRequest.eventLoop) // XCTAssertIdentical not available on Linux + + // 2. connection fails – first with closed callback + + XCTAssertEqual(state.http2ConnectionClosed(connectionID), .none) + + // 3. connection fails – with make connection callback + + let action = state.failedToCreateNewConnection( + IOError(errnoCode: -1, reason: "Test failure"), + connectionID: connectionID + ) + XCTAssertEqual(action.request, .none) + guard case .scheduleBackoffTimer(connectionID, _, on: let backoffTimerEL) = action.connection else { + XCTFail("Unexpected connection action: \(action.connection)") + return + } + XCTAssertIdentical(connectionEL, backoffTimerEL) + } + } /// Should be used if you have a value of statically unknown type and want to compare its value to another `Equatable` value. diff --git a/Tests/AsyncHTTPClientTests/TransactionTests.swift b/Tests/AsyncHTTPClientTests/TransactionTests.swift index 3316de370..8e6464a5b 100644 --- a/Tests/AsyncHTTPClientTests/TransactionTests.swift +++ b/Tests/AsyncHTTPClientTests/TransactionTests.swift @@ -657,7 +657,6 @@ private actor Promise { @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) extension Transaction { - #if compiler(>=6.0) fileprivate static func makeWithResultTask( request: sending PreparedRequest, requestOptions: RequestOptions = .forTests(), @@ -685,40 +684,4 @@ extension Transaction { return (await transactionPromise.value, task) } - #else - fileprivate static func makeWithResultTask( - request: PreparedRequest, - requestOptions: RequestOptions = .forTests(), - logger: Logger = Logger(label: "test"), - connectionDeadline: NIODeadline = .distantFuture, - preferredEventLoop: EventLoop - ) async -> (Transaction, _Concurrency.Task) { - // It isn't sendable ... but on 6.0 and later we use 'sending'. - struct UnsafePrepareRequest: @unchecked Sendable { - var value: PreparedRequest - } - - let transactionPromise = Promise() - let unsafe = UnsafePrepareRequest(value: request) - let task = Task { - try await withCheckedThrowingContinuation { - (continuation: CheckedContinuation) in - let request = unsafe.value - let transaction = Transaction( - request: request, - requestOptions: requestOptions, - logger: logger, - connectionDeadline: connectionDeadline, - preferredEventLoop: preferredEventLoop, - responseContinuation: continuation - ) - Task { - await transactionPromise.fulfil(transaction) - } - } - } - - return (await transactionPromise.value, task) - } - #endif }