Surface transient connection errors

Author: gjcairo

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.

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

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:

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>) -> 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)
   }
 }

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,16 +282,25 @@ 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):
       // Try the next address.
       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.

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

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),