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.

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

@@ -758,7 +758,23 @@ 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",
+                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.

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
         }
 
@@ -750,12 +750,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.
+    for state in states {
+      guard case .transientFailure = state else {
+        return .shutdown
+      }
     }
 
-    return .shutdown
+    return .transientFailure(cause: RPCError(
+      code: .internalError,
+      message: "All subchannels are in TRANSIENT_FAILURE state."
+    ))
   }
 }

Sources/GRPCNIOTransportCore/Client/Connection/LoadBalancers/Subchannel.swift

@@ -291,7 +291,14 @@ extension Subchannel {
 
     case .backoff(let duration):
       // All addresses have been tried, backoff for some time.
-      self.event.continuation.yield(.connectivityStateChanged(.transientFailure))
+      self.event.continuation.yield(
+        .connectivityStateChanged(
+          .transientFailure(cause: RPCError(
+            code: .unavailable, 
+            message: "All addresses have been tried: backing off."
+          ))
+        )
+      )
       group.addTask {
         do {
           try await Task.sleep(for: duration)
@@ -334,9 +341,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 +639,7 @@ extension Subchannel {
     enum OnClosed {
       case nothing
       case emitIdle
-      case emitTransientFailureAndReconnect
+      case emitTransientFailureAndReconnect(cause: RPCError)
       case finish(emitShutdown: Bool)
     }
 
@@ -646,9 +653,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/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),