From 45bc5dca054f24bfcf75a7baae24e640f8088481 Mon Sep 17 00:00:00 2001 From: Gus Cairo Date: Mon, 30 Sep 2024 11:19:06 +0100 Subject: [PATCH 1/5] Surface transient connection errors --- .../Client/Connection/Connection.swift | 12 +++++-- .../Client/Connection/ConnectivityState.swift | 4 ++- .../Client/Connection/GRPCChannel.swift | 24 +++++++++++--- .../RoundRobinLoadBalancer.swift | 20 +++++++---- .../Connection/LoadBalancers/Subchannel.swift | 33 ++++++++++++++----- .../Connection/Connection+Equatable.swift | 8 ++--- .../LoadBalancers/SubchannelTests.swift | 9 +++-- 7 files changed, 79 insertions(+), 31 deletions(-) diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift b/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift index 6d10774..22a603d 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift @@ -68,7 +68,7 @@ package final class Connection: Sendable { /// Closed because the remote peer initiate shutdown (i.e. sent a GOAWAY frame). case remote /// Closed because the connection encountered an unexpected error. - case error(any Error, wasIdle: Bool) + case error(RPCError, wasIdle: Bool) } /// Inputs to the 'run' method. @@ -236,6 +236,7 @@ package final class Connection: Sendable { // This state is tracked here so that if the connection events sequence finishes and the // connection never became ready then the connection can report that the connect failed. var isReady = false + var unexpectedCloseError: (any Error)? func makeNeverReadyError(cause: (any Error)?) -> RPCError { return RPCError( @@ -265,10 +266,15 @@ package final class Connection: Sendable { // The connection will close at some point soon, yield a notification for this // because the close might not be imminent and this could result in address resolution. self.event.continuation.yield(.goingAway(errorCode, reason)) - case .idle, .keepaliveExpired, .initiatedLocally, .unexpected: + case .idle, .keepaliveExpired, .initiatedLocally: // The connection will be closed imminently in these cases there's no need to do // anything. () + case .unexpected(let error, _): + // The connection will be closed imminently in this case. + // We'll store the error that caused the unexpected closure so we + // can surface it. + unexpectedCloseError = error } // Take the reason with the highest precedence. A GOAWAY may be superseded by user @@ -318,7 +324,7 @@ package final class Connection: Sendable { finalEvent = .closed(connectionCloseReason) } else { // The connection never became ready, this therefore counts as a failed connect attempt. - finalEvent = .connectFailed(makeNeverReadyError(cause: nil)) + finalEvent = .connectFailed(makeNeverReadyError(cause: unexpectedCloseError)) } // The connection events sequence has finished: the connection is now closed. diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/ConnectivityState.swift b/Sources/GRPCNIOTransportCore/Client/Connection/ConnectivityState.swift index 6f4b000..8ef3a89 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/ConnectivityState.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/ConnectivityState.swift @@ -14,6 +14,8 @@ * limitations under the License. */ +package import GRPCCore + package enum ConnectivityState: Sendable, Hashable { /// This channel isn't trying to create a connection because of a lack of new or pending RPCs. /// @@ -34,7 +36,7 @@ package enum ConnectivityState: Sendable, Hashable { /// establish a connection again. Since retries are done with exponential backoff, channels that /// fail to connect will start out spending very little time in this state but as the attempts /// fail repeatedly, the channel will spend increasingly large amounts of time in this state. - case transientFailure + case transientFailure(cause: RPCError) /// This channel has started shutting down. Any new RPCs should fail immediately. Pending RPCs /// may continue running until the application cancels them. Channels may enter this state either diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/GRPCChannel.swift b/Sources/GRPCNIOTransportCore/Client/Connection/GRPCChannel.swift index 9f3c1fe..6ce0fd1 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/GRPCChannel.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/GRPCChannel.swift @@ -290,7 +290,7 @@ extension GRPCChannel { } case .failRPC: - return .stopTrying(RPCError(code: .unavailable, message: "channel isn't ready")) + return .stopTrying(RPCError(code: .unavailable, message: "Channel isn't ready.")) } } @@ -300,7 +300,7 @@ extension GRPCChannel { loadBalancer: LoadBalancer ) async -> MakeStreamResult { guard let subchannel = loadBalancer.pickSubchannel() else { - return .tryAgain(RPCError(code: .unavailable, message: "channel isn't ready")) + return .tryAgain(RPCError(code: .unavailable, message: "Channel isn't ready.")) } let methodConfig = self.config(forMethod: descriptor) @@ -758,14 +758,30 @@ extension GRPCChannel.StateMachine { result: .success(state.current) ) - case .transientFailure, .shutdown: // shutdown includes shutting down + case .transientFailure(let cause): // Current load-balancer failed. Remove all the 'fast-failing' continuations in the // queue, these are RPCs which set the 'wait for ready' option to false. The rest of // the entries in the queue will wait for a load-balancer to become ready. let continuations = state.queue.removeFastFailingEntries() actions.resumeContinuations = ConnectivityStateChangeActions.ResumableContinuations( continuations: continuations, - result: .failure(RPCError(code: .unavailable, message: "channel isn't ready")) + result: .failure( + RPCError( + code: .unavailable, + message: "Channel isn't ready.", + cause: cause + ) + ) + ) + + case .shutdown: // shutdown includes shutting down + // Current load-balancer failed. Remove all the 'fast-failing' continuations in the + // queue, these are RPCs which set the 'wait for ready' option to false. The rest of + // the entries in the queue will wait for a load-balancer to become ready. + let continuations = state.queue.removeFastFailingEntries() + actions.resumeContinuations = ConnectivityStateChangeActions.ResumableContinuations( + continuations: continuations, + result: .failure(RPCError(code: .unavailable, message: "Channel isn't ready.")) ) case .idle, .connecting: diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/RoundRobinLoadBalancer.swift b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/RoundRobinLoadBalancer.swift index f0185ef..178f2ef 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/RoundRobinLoadBalancer.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/RoundRobinLoadBalancer.swift @@ -530,7 +530,7 @@ extension RoundRobinLoadBalancer { // The transition from transient failure to connecting is ignored. // // See: https://github.com/grpc/grpc/blob/master/doc/load-balancing.md - if self.state == .transientFailure, newState == .connecting { + if case .transientFailure = self.state, newState == .connecting { return false } @@ -735,6 +735,10 @@ extension ConnectivityState { static func aggregate(_ states: some Collection) -> ConnectivityState { // See https://github.com/grpc/grpc/blob/master/doc/load-balancing.md + if states.isEmpty { + return .shutdown + } + // If any one subchannel is in READY state, the channel's state is READY. if states.contains(where: { $0 == .ready }) { return .ready @@ -750,12 +754,16 @@ extension ConnectivityState { return .idle } - // Otherwise, if all subchannels are in state TRANSIENT_FAILURE, the channel's state - // is TRANSIENT_FAILURE. - if states.allSatisfy({ $0 == .transientFailure }) { - return .transientFailure + // Otherwise, if all subchannels are in state TRANSIENT_FAILURE, the channel's state is TRANSIENT_FAILURE. + // Pick one of the errors to surface, as we can't surface all of them. + var cause: RPCError! + for state in states { + guard case .transientFailure(let error) = state else { + return .shutdown + } + cause = error } - return .shutdown + return .transientFailure(cause: cause) } } diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift index a9947cb..0135725 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift @@ -256,8 +256,8 @@ extension Subchannel { switch event { case .connectSucceeded: self.handleConnectSucceededEvent() - case .connectFailed: - self.handleConnectFailedEvent(in: &group) + case .connectFailed(let cause): + self.handleConnectFailedEvent(in: &group, error: cause) case .goingAway: self.handleGoingAwayEvent() case .closed(let reason): @@ -282,7 +282,7 @@ extension Subchannel { } } - private func handleConnectFailedEvent(in group: inout DiscardingTaskGroup) { + private func handleConnectFailedEvent(in group: inout DiscardingTaskGroup, error: any Error) { let onConnectFailed = self.state.withLock { $0.connectFailed(connector: self.connector) } switch onConnectFailed { case .connect(let connection): @@ -290,8 +290,17 @@ extension Subchannel { self.runConnection(connection, in: &group) case .backoff(let duration): + let transientFailureCause = (error as? RPCError) ?? RPCError( + code: .unavailable, + message: "All addresses have been tried: backing off.", + cause: error + ) // All addresses have been tried, backoff for some time. - self.event.continuation.yield(.connectivityStateChanged(.transientFailure)) + self.event.continuation.yield( + .connectivityStateChanged( + .transientFailure(cause: transientFailureCause) + ) + ) group.addTask { do { try await Task.sleep(for: duration) @@ -334,9 +343,9 @@ extension Subchannel { case .emitIdle: self.event.continuation.yield(.connectivityStateChanged(.idle)) - case .emitTransientFailureAndReconnect: + case .emitTransientFailureAndReconnect(let cause): // Unclean closes trigger a transient failure state change and a name resolution. - self.event.continuation.yield(.connectivityStateChanged(.transientFailure)) + self.event.continuation.yield(.connectivityStateChanged(.transientFailure(cause: cause))) self.event.continuation.yield(.requiresNameResolution) // Attempt to reconnect. self.handleConnectInput(in: &group) @@ -632,7 +641,7 @@ extension Subchannel { enum OnClosed { case nothing case emitIdle - case emitTransientFailureAndReconnect + case emitTransientFailureAndReconnect(cause: RPCError) case finish(emitShutdown: Bool) } @@ -646,9 +655,15 @@ extension Subchannel { self = .notConnected(NotConnected(from: state)) onClosed = .emitIdle - case .keepaliveTimeout, .error(_, wasIdle: false): + case .keepaliveTimeout: + self = .notConnected(NotConnected(from: state)) + onClosed = .emitTransientFailureAndReconnect( + cause: RPCError(code: .unavailable, message: "The keepalive timed out.") + ) + + case .error(let error, wasIdle: false): self = .notConnected(NotConnected(from: state)) - onClosed = .emitTransientFailureAndReconnect + onClosed = .emitTransientFailureAndReconnect(cause: error) case .initiatedLocally: // Should be in the 'shuttingDown' state. diff --git a/Tests/GRPCNIOTransportCoreTests/Client/Connection/Connection+Equatable.swift b/Tests/GRPCNIOTransportCoreTests/Client/Connection/Connection+Equatable.swift index 393e2f8..bab073a 100644 --- a/Tests/GRPCNIOTransportCoreTests/Client/Connection/Connection+Equatable.swift +++ b/Tests/GRPCNIOTransportCoreTests/Client/Connection/Connection+Equatable.swift @@ -53,12 +53,8 @@ extension Connection.CloseReason { (.remote, .remote): return true - case (.error(let lhsError, let lhsStreams), .error(let rhsError, let rhsStreams)): - if let lhs = lhsError as? RPCError, let rhs = rhsError as? RPCError { - return lhs == rhs && lhsStreams == rhsStreams - } else { - return lhsStreams == rhsStreams - } + case (.error(_, let lhsStreams), .error(_, let rhsStreams)): + return lhs == rhs && lhsStreams == rhsStreams default: return false diff --git a/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift b/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift index 5ef50ca..f7d4b23 100644 --- a/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift +++ b/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift @@ -161,7 +161,10 @@ final class SubchannelTests: XCTestCase { [ .connectivityStateChanged(.idle), .connectivityStateChanged(.connecting), - .connectivityStateChanged(.transientFailure), + .connectivityStateChanged(.transientFailure(cause: RPCError( + code: .unavailable, + message: "All addresses have been tried: backing off." + ))), .connectivityStateChanged(.connecting), ] ) @@ -440,7 +443,9 @@ final class SubchannelTests: XCTestCase { .connectivityStateChanged(.idle), .connectivityStateChanged(.connecting), .connectivityStateChanged(.ready), - .connectivityStateChanged(.transientFailure), + .connectivityStateChanged(.transientFailure(cause: RPCError( + code: .unavailable, message: "The TCP connection was dropped unexpectedly." + ))), .requiresNameResolution, .connectivityStateChanged(.connecting), .connectivityStateChanged(.ready), From 6edc785ee631a020bb76bcaf66f84770f6f4a0eb Mon Sep 17 00:00:00 2001 From: Gus Cairo Date: Tue, 8 Oct 2024 22:52:40 +0000 Subject: [PATCH 2/5] Formatting --- .../Connection/LoadBalancers/Subchannel.swift | 12 ++++++---- .../LoadBalancers/SubchannelTests.swift | 23 +++++++++++++------ 2 files changed, 23 insertions(+), 12 deletions(-) diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift index 0135725..8fa3599 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift @@ -290,11 +290,13 @@ extension Subchannel { self.runConnection(connection, in: &group) case .backoff(let duration): - let transientFailureCause = (error as? RPCError) ?? RPCError( - code: .unavailable, - message: "All addresses have been tried: backing off.", - cause: error - ) + let transientFailureCause = + (error as? RPCError) + ?? RPCError( + code: .unavailable, + message: "All addresses have been tried: backing off.", + cause: error + ) // All addresses have been tried, backoff for some time. self.event.continuation.yield( .connectivityStateChanged( diff --git a/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift b/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift index f7d4b23..c76f62c 100644 --- a/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift +++ b/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift @@ -161,10 +161,14 @@ final class SubchannelTests: XCTestCase { [ .connectivityStateChanged(.idle), .connectivityStateChanged(.connecting), - .connectivityStateChanged(.transientFailure(cause: RPCError( - code: .unavailable, - message: "All addresses have been tried: backing off." - ))), + .connectivityStateChanged( + .transientFailure( + cause: RPCError( + code: .unavailable, + message: "All addresses have been tried: backing off." + ) + ) + ), .connectivityStateChanged(.connecting), ] ) @@ -443,9 +447,14 @@ final class SubchannelTests: XCTestCase { .connectivityStateChanged(.idle), .connectivityStateChanged(.connecting), .connectivityStateChanged(.ready), - .connectivityStateChanged(.transientFailure(cause: RPCError( - code: .unavailable, message: "The TCP connection was dropped unexpectedly." - ))), + .connectivityStateChanged( + .transientFailure( + cause: RPCError( + code: .unavailable, + message: "The TCP connection was dropped unexpectedly." + ) + ) + ), .requiresNameResolution, .connectivityStateChanged(.connecting), .connectivityStateChanged(.ready), From 818f7ca2a096060616e1c68c7db9d0df7069cc01 Mon Sep 17 00:00:00 2001 From: Gus Cairo Date: Wed, 9 Oct 2024 11:53:00 +0000 Subject: [PATCH 3/5] PR changes --- .../Client/Connection/Connection.swift | 16 ++++++++++--- .../Client/Connection/GRPCChannel.swift | 8 +------ .../RoundRobinLoadBalancer.swift | 24 ++++++++++++------- .../Connection/LoadBalancers/Subchannel.swift | 19 +++++++-------- .../Internal/Result+Catching.swift | 4 ++-- .../Connection/Connection+Equatable.swift | 4 ++-- 6 files changed, 42 insertions(+), 33 deletions(-) diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift b/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift index 22a603d..117f3da 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift @@ -49,7 +49,7 @@ package final class Connection: Sendable { /// The connect attempt succeeded and the connection is ready to use. case connectSucceeded /// The connect attempt failed. - case connectFailed(any Error) + case connectFailed(RPCError) /// The connection received a GOAWAY and will close soon. No new streams /// should be opened on this connection. case goingAway(HTTP2ErrorCode, String) @@ -127,9 +127,19 @@ package final class Connection: Sendable { /// This function returns when the connection has closed. You can observe connection events /// by consuming the ``events`` sequence. package func run() async { - let connectResult = await Result { - try await self.http2Connector.establishConnection(to: self.address) + func establishConnectionOrThrow() async throws(RPCError) -> HTTP2Connection { + do { + return try await self.http2Connector.establishConnection(to: self.address) + } catch { + throw (error as? RPCError) + ?? RPCError( + code: .unavailable, + message: "Could not establish a connection to \(self.address).", + cause: error + ) + } } + let connectResult = await Result(catching: establishConnectionOrThrow) switch connectResult { case .success(let connected): diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/GRPCChannel.swift b/Sources/GRPCNIOTransportCore/Client/Connection/GRPCChannel.swift index 6ce0fd1..8bd5e46 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/GRPCChannel.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/GRPCChannel.swift @@ -765,13 +765,7 @@ extension GRPCChannel.StateMachine { let continuations = state.queue.removeFastFailingEntries() actions.resumeContinuations = ConnectivityStateChangeActions.ResumableContinuations( continuations: continuations, - result: .failure( - RPCError( - code: .unavailable, - message: "Channel isn't ready.", - cause: cause - ) - ) + result: .failure(cause) ) case .shutdown: // shutdown includes shutting down diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/RoundRobinLoadBalancer.swift b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/RoundRobinLoadBalancer.swift index 178f2ef..abdacb1 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/RoundRobinLoadBalancer.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/RoundRobinLoadBalancer.swift @@ -735,10 +735,6 @@ extension ConnectivityState { static func aggregate(_ states: some Collection) -> ConnectivityState { // See https://github.com/grpc/grpc/blob/master/doc/load-balancing.md - if states.isEmpty { - return .shutdown - } - // If any one subchannel is in READY state, the channel's state is READY. if states.contains(where: { $0 == .ready }) { return .ready @@ -755,15 +751,25 @@ extension ConnectivityState { } // Otherwise, if all subchannels are in state TRANSIENT_FAILURE, the channel's state is TRANSIENT_FAILURE. - // Pick one of the errors to surface, as we can't surface all of them. - var cause: RPCError! + var cause: RPCError? for state in states { - guard case .transientFailure(let error) = state else { + switch state { + case .transientFailure(let error): + // Pick one of the errors to surface, as we can't surface all of them. + cause = error + case .shutdown: return .shutdown + case .idle, .connecting, .ready: + fatalError("Unreachable state: these should have been handled above.") } - cause = error } - return .transientFailure(cause: cause) + if let cause { + return .transientFailure(cause: cause) + } else { + // We can only reach this point without a `cause` if `states` was empty. + // Fall back to shutdown: we have nothing better to do. + return .shutdown + } } } diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift index 8fa3599..b2be017 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift @@ -282,7 +282,7 @@ extension Subchannel { } } - private func handleConnectFailedEvent(in group: inout DiscardingTaskGroup, error: any Error) { + private func handleConnectFailedEvent(in group: inout DiscardingTaskGroup, error: RPCError) { let onConnectFailed = self.state.withLock { $0.connectFailed(connector: self.connector) } switch onConnectFailed { case .connect(let connection): @@ -290,17 +290,10 @@ extension Subchannel { self.runConnection(connection, in: &group) case .backoff(let duration): - let transientFailureCause = - (error as? RPCError) - ?? RPCError( - code: .unavailable, - message: "All addresses have been tried: backing off.", - cause: error - ) // All addresses have been tried, backoff for some time. self.event.continuation.yield( .connectivityStateChanged( - .transientFailure(cause: transientFailureCause) + .transientFailure(cause: error) ) ) group.addTask { @@ -660,7 +653,13 @@ extension Subchannel { case .keepaliveTimeout: self = .notConnected(NotConnected(from: state)) onClosed = .emitTransientFailureAndReconnect( - cause: RPCError(code: .unavailable, message: "The keepalive timed out.") + cause: RPCError( + code: .unavailable, + message: """ + The connection became unresponsive and was closed because the \ + keepalive timeout fired. + """ + ) ) case .error(let error, wasIdle: false): diff --git a/Sources/GRPCNIOTransportCore/Internal/Result+Catching.swift b/Sources/GRPCNIOTransportCore/Internal/Result+Catching.swift index 5c5060e..6b5bbf2 100644 --- a/Sources/GRPCNIOTransportCore/Internal/Result+Catching.swift +++ b/Sources/GRPCNIOTransportCore/Internal/Result+Catching.swift @@ -14,12 +14,12 @@ * limitations under the License. */ -extension Result where Failure == any Error { +extension Result { /// Like `Result(catching:)`, but `async`. /// /// - Parameter body: An `async` closure to catch the result of. @inlinable - init(catching body: () async throws -> Success) async { + init(catching body: () async throws(Failure) -> Success) async { do { self = .success(try await body()) } catch { diff --git a/Tests/GRPCNIOTransportCoreTests/Client/Connection/Connection+Equatable.swift b/Tests/GRPCNIOTransportCoreTests/Client/Connection/Connection+Equatable.swift index bab073a..8d09c60 100644 --- a/Tests/GRPCNIOTransportCoreTests/Client/Connection/Connection+Equatable.swift +++ b/Tests/GRPCNIOTransportCoreTests/Client/Connection/Connection+Equatable.swift @@ -53,8 +53,8 @@ extension Connection.CloseReason { (.remote, .remote): return true - case (.error(_, let lhsStreams), .error(_, let rhsStreams)): - return lhs == rhs && lhsStreams == rhsStreams + case (.error(let lhsError, let lhsStreams), .error(let rhsError, let rhsStreams)): + return lhsError == rhsError && lhsStreams == rhsStreams default: return false From 003ac5cf7b7e1c0071e603b320e0fea9fd7a25c3 Mon Sep 17 00:00:00 2001 From: Gus Cairo Date: Wed, 9 Oct 2024 12:58:56 +0000 Subject: [PATCH 4/5] PR change --- .../Client/Connection/Connection.swift | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift b/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift index 117f3da..1a4330a 100644 --- a/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift +++ b/Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift @@ -130,13 +130,14 @@ package final class Connection: Sendable { func establishConnectionOrThrow() async throws(RPCError) -> HTTP2Connection { do { return try await self.http2Connector.establishConnection(to: self.address) + } catch let error as RPCError { + throw error } catch { - throw (error as? RPCError) - ?? RPCError( - code: .unavailable, - message: "Could not establish a connection to \(self.address).", - cause: error - ) + throw RPCError( + code: .unavailable, + message: "Could not establish a connection to \(self.address).", + cause: error + ) } } let connectResult = await Result(catching: establishConnectionOrThrow) From 93751f093846b981b194db52b9baa7a04a60203e Mon Sep 17 00:00:00 2001 From: Gus Cairo Date: Wed, 9 Oct 2024 14:50:26 +0000 Subject: [PATCH 5/5] Fix test --- .../Client/Connection/LoadBalancers/SubchannelTests.swift | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift b/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift index c76f62c..069a03a 100644 --- a/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift +++ b/Tests/GRPCNIOTransportCoreTests/Client/Connection/LoadBalancers/SubchannelTests.swift @@ -165,7 +165,8 @@ final class SubchannelTests: XCTestCase { .transientFailure( cause: RPCError( code: .unavailable, - message: "All addresses have been tried: backing off." + message: + "Could not establish a connection to [unix]test-connect-eventually-succeeds." ) ) ),