Convert errors thrown from interceptor inbound or outbound stream

Author: bauer-andreas

Sources/GRPCCore/Call/Client/Internal/ClientResponse+Convenience.swift

@@ -71,6 +71,8 @@ extension ClientResponse {
       } catch let error as RPCError {
         // Known error type.
         self.accepted = .success(Contents(metadata: contents.metadata, error: error))
+      } catch let error as any RPCErrorConvertible {
+        self.accepted = .success(Contents(metadata: contents.metadata, error: RPCError(error)))
       } catch {
         // Unexpected, but should be handled nonetheless.
         self.accepted = .failure(RPCError(code: .unknown, message: String(describing: error)))

Sources/GRPCCore/Call/Client/Internal/ClientStreamExecutor.swift

@@ -96,7 +96,11 @@ internal enum ClientStreamExecutor {
       try await stream.write(.metadata(request.metadata))
       try await request.producer(.map(into: stream) { .message(try serializer.serialize($0)) })
     }.castError(to: RPCError.self) { other in
-      RPCError(code: .unknown, message: "Write failed.", cause: other)
+      if let convertible = other as? any RPCErrorConvertible {
+        RPCError(convertible)
+      } else {
+        RPCError(code: .unknown, message: "Write failed.", cause: other)
+      }
     }
 
     switch result {

Sources/GRPCCore/Call/Server/Internal/ServerRPCExecutor.swift

@@ -214,7 +214,11 @@ struct ServerRPCExecutor {
           .serializingToRPCResponsePart(into: outbound, with: serializer)
         )
       }.castError(to: RPCError.self) { error in
-        RPCError(code: .unknown, message: "", cause: error)
+        if let convertible = error as? (any RPCErrorConvertible) {
+          return RPCError(convertible)
+        } else {
+          return RPCError(code: .unknown, message: "", cause: error)
+        }
       }
 
       switch result {

Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTestSupport/ClientRPCExecutorTestHarness.swift

@@ -132,7 +132,11 @@ struct ClientRPCExecutorTestHarness {
     try await withThrowingTaskGroup(of: Void.self) { group in
       group.addTask {
         try await self.serverTransport.listen { stream, context in
-          try? await self.server.handle(stream: stream)
+          do {
+            try await self.server.handle(stream: stream)
+          } catch {
+            await stream.outbound.finish(throwing: error)
+          }
         }
       }
 

Tests/GRPCCoreTests/Call/Client/Internal/ClientRPCExecutorTests.swift

@@ -290,4 +290,46 @@ final class ClientRPCExecutorTests: XCTestCase {
       }
     }
   }
+
+  func testInterceptorProducerErrorConversion() async throws {
+    struct CustomError: RPCErrorConvertible, Error {
+      var rpcErrorCode: RPCError.Code { .alreadyExists }
+      var rpcErrorMessage: String { "foobar" }
+      var rpcErrorMetadata: Metadata { ["error": "yes"] }
+    }
+
+    let tester = ClientRPCExecutorTestHarness(
+      server: .echo,
+      interceptors: [.throwInProducer(CustomError())]
+    )
+
+    try await tester.unary(request: ClientRequest(message: [])) { response in
+      XCTAssertThrowsError(ofType: RPCError.self, try response.message) { error in
+        XCTAssertEqual(error.code, .alreadyExists)
+        XCTAssertEqual(error.message, "foobar")
+        XCTAssertEqual(error.metadata, ["error": "yes"])
+      }
+    }
+  }
+
+  func testInterceptorBodyPartsErrorConversion() async throws {
+    struct CustomError: RPCErrorConvertible, Error {
+      var rpcErrorCode: RPCError.Code { .alreadyExists }
+      var rpcErrorMessage: String { "foobar" }
+      var rpcErrorMetadata: Metadata { ["error": "yes"] }
+    }
+
+    let tester = ClientRPCExecutorTestHarness(
+      server: .echo,
+      interceptors: [.throwInBodyParts(CustomError())]
+    )
+
+    try await tester.unary(request: ClientRequest(message: [])) { response in
+      XCTAssertThrowsError(ofType: RPCError.self, try response.message) { error in
+        XCTAssertEqual(error.code, .alreadyExists)
+        XCTAssertEqual(error.message, "foobar")
+        XCTAssertEqual(error.metadata, ["error": "yes"])
+      }
+    }
+  }
 }

Tests/GRPCCoreTests/Call/Server/Internal/ServerRPCExecutorTests.swift

@@ -394,4 +394,48 @@ final class ServerRPCExecutorTests: XCTestCase {
       XCTAssertEqual(parts, [.status(status, metadata)])
     }
   }
+
+  func testInterceptorProducerErrorConversion() async throws {
+    struct CustomError: RPCErrorConvertible, Error {
+      var rpcErrorCode: RPCError.Code { .alreadyExists }
+      var rpcErrorMessage: String { "foobar" }
+      var rpcErrorMetadata: Metadata { ["error": "yes"] }
+    }
+
+    let harness = ServerRPCExecutorTestHarness(
+      interceptors: [.throwInProducer(CustomError(), after: .milliseconds(10))]
+    )
+    try await harness.execute(handler: .echo) { inbound in
+      try await inbound.write(.metadata(["foo": "bar"]))
+      try await inbound.write(.message([0]))
+      try await Task.sleep(for: .milliseconds(50))
+      try await inbound.write(.message([1]))
+      await inbound.finish()
+    } consumer: { outbound in
+      let parts = try await outbound.collect()
+      let status = Status(code: .alreadyExists, message: "foobar")
+      let metadata: Metadata = ["error": "yes"]
+      XCTAssertEqual(parts, [.metadata(["foo": "bar"]), .message([0]), .status(status, metadata)])
+    }
+  }
+
+  func testInterceptorMessagesErrorConversion() async throws {
+    struct CustomError: RPCErrorConvertible, Error {
+      var rpcErrorCode: RPCError.Code { .alreadyExists }
+      var rpcErrorMessage: String { "foobar" }
+      var rpcErrorMetadata: Metadata { ["error": "yes"] }
+    }
+
+    let harness = ServerRPCExecutorTestHarness(interceptors: [.throwInMessageSequence(CustomError())])
+    try await harness.execute(handler: .echo) { inbound in
+      try await inbound.write(.metadata(["foo": "bar"]))
+      try await inbound.write(.message([0])) // the sequence throws instantly, this should not arrive
+      await inbound.finish()
+    } consumer: { outbound in
+      let parts = try await outbound.collect()
+      let status = Status(code: .alreadyExists, message: "foobar")
+      let metadata: Metadata = ["error": "yes"]
+      XCTAssertEqual(parts, [.metadata(["foo": "bar"]), .status(status, metadata)])
+    }
+  }
 }

Tests/GRPCCoreTests/Test Utilities/Call/Client/ClientInterceptors.swift

@@ -26,6 +26,13 @@ extension ClientInterceptor where Self == RejectAllClientInterceptor {
     return RejectAllClientInterceptor(throw: error)
   }
 
+  static func throwInBodyParts(_ error: any Error) -> Self {
+    return RejectAllClientInterceptor(throwInBodyParts: error)
+  }
+
+  static func throwInProducer(_ error: any Error) -> Self {
+    return RejectAllClientInterceptor(throwInProducer: error)
+  }
 }
 
 @available(gRPCSwift 2.0, *)
@@ -43,6 +50,10 @@ struct RejectAllClientInterceptor: ClientInterceptor {
     case `throw`(any Error)
     /// Reject the RPC with a given error.
     case reject(RPCError)
+    /// Throw an error in the body parts sequence.
+    case throwInBodyParts(any Error)
+    /// Throw an error in the message producer closure.
+    case throwInProducer(any Error)
   }
 
   let mode: Mode
@@ -55,6 +66,14 @@ struct RejectAllClientInterceptor: ClientInterceptor {
     self.mode = .reject(error)
   }
 
+  init(throwInBodyParts error: any Error) {
+    self.mode = .throwInBodyParts(error)
+  }
+
+  init(throwInProducer error: any Error) {
+    self.mode = .throwInProducer(error)
+  }
+
   func intercept<Input: Sendable, Output: Sendable>(
     request: StreamingClientRequest<Input>,
     context: ClientContext,
@@ -68,6 +87,29 @@ struct RejectAllClientInterceptor: ClientInterceptor {
       throw error
     case .reject(let error):
       return StreamingClientResponse(error: error)
+    case .throwInBodyParts(let error):
+      var response = try await next(request, context)
+      switch response.accepted {
+      case .success(var success):
+        let stream = AsyncThrowingStream<StreamingClientResponse<Output>.Contents.BodyPart, any Error>.makeStream()
+        stream.continuation.finish(throwing: error)
+
+        success.bodyParts = RPCAsyncSequence(wrapping: stream.stream)
+        response.accepted = .success(success)
+        return response
+      case .failure:
+        return response
+      }
+    case .throwInProducer(let error):
+      let wrappedProducer = request.producer
+
+      var request = request
+      request.producer = { writer in
+        try await wrappedProducer(writer)
+        throw error
+      }
+
+      return try await next(request, context)
     }
   }
 }

Tests/GRPCCoreTests/Test Utilities/Call/Server/ServerInterceptors.swift

@@ -25,6 +25,14 @@ extension ServerInterceptor where Self == RejectAllServerInterceptor {
   static func throwError(_ error: any Error) -> Self {
     RejectAllServerInterceptor(throw: error)
   }
+
+  static func throwInProducer(_ error: any Error, after duration: Duration) -> Self {
+    RejectAllServerInterceptor(throwInProducer: error, after: duration)
+  }
+
+  static func throwInMessageSequence(_ error: any Error) -> Self {
+    RejectAllServerInterceptor(throwInMessageSequence: error)
+  }
 }
 
 @available(gRPCSwift 2.0, *)
@@ -42,6 +50,16 @@ struct RejectAllServerInterceptor: ServerInterceptor {
     case `throw`(any Error)
     /// Reject the RPC with a given error.
     case reject(RPCError)
+    /// Throw in the producer closure returned.
+    case throwInProducer(any Error, after: Duration)
+    /// Throw in the async sequence that stream inbound messages.
+    case throwInMessageSequence(any Error)
+  }
+
+  private enum TimeoutResult {
+    case `throw`(any Error)
+    case cancelled
+    case result(Metadata)
   }
 
   let mode: Mode
@@ -54,6 +72,14 @@ struct RejectAllServerInterceptor: ServerInterceptor {
     self.mode = .reject(error)
   }
 
+  init(throwInProducer error: any Error, after duration: Duration) {
+    self.mode = .throwInProducer(error, after: duration)
+  }
+
+  init(throwInMessageSequence error: any Error) {
+    self.mode = .throwInMessageSequence(error)
+  }
+
   func intercept<Input: Sendable, Output: Sendable>(
     request: StreamingServerRequest<Input>,
     context: ServerContext,
@@ -67,6 +93,60 @@ struct RejectAllServerInterceptor: ServerInterceptor {
       throw error
     case .reject(let error):
       return StreamingServerResponse(error: error)
+    case .throwInProducer(let error, let duration):
+      var response = try await next(request, context)
+      switch response.accepted {
+      case .success(var success):
+        let wrappedProducer = success.producer
+        success.producer = { writer in
+          let result: Result<Metadata, any Error> = await withTaskGroup(of: TimeoutResult.self) { group in
+            group.addTask {
+              do {
+                try await Task.sleep(for: duration, tolerance: .nanoseconds(1))
+              } catch {
+                return .cancelled
+              }
+              return .throw(error)
+            }
+
+            group.addTask {
+              do {
+                return .result(try await wrappedProducer(writer))
+              } catch {
+                return .throw(error)
+              }
+            }
+
+            let first = await group.next()!
+            group.cancelAll()
+            let second = await group.next()!
+
+            switch (first, second) {
+            case (.throw(let error), _):
+              return .failure(error)
+            case (.result(let metadata), _):
+              return .success(metadata)
+            case (.cancelled, _):
+              return .failure(CancellationError())
+            }
+          }
+
+          return try result.get()
+        }
+
+        response.accepted = .success(success)
+        return response
+      case .failure:
+        return response
+      }
+    case .throwInMessageSequence(let error):
+      let stream = AsyncThrowingStream<Input, any Error>.makeStream()
+      stream.continuation.finish(throwing: error)
+
+      var request = request
+      request.messages = RPCAsyncSequence(wrapping: stream.stream)
+
+      return try await next(request, context)
     }
   }
 }