Ask `ServerErrorDelegate` to transform status/response promise failures before returning them to the client. (#423)

* Ask `ServerErrorDelegate` to transform status/response promise failures before returning them to the client.

See https://github.com/grpc/grpc-swift/pull/422#issuecomment-480174616 for discussion.

* More observe/transform tweaks.

* Another try with `transform`.

* A few more renames and test fixes.

Sources/SwiftGRPCNIO/CallHandlers/BaseCallHandler.swift

@@ -30,7 +30,7 @@ public class BaseCallHandler<RequestMessage: Message, ResponseMessage: Message>:
   /// Whether this handler can still write messages to the client.
   private var serverCanWrite = true
 
-  /// Called for each error recieved in `errorCaught(context:error:)`.
+  /// Called for each error received in `errorCaught(context:error:)`.
   private weak var errorDelegate: ServerErrorDelegate?
 
   public init(errorDelegate: ServerErrorDelegate?) {
@@ -51,10 +51,11 @@ extension BaseCallHandler: ChannelInboundHandler {
   /// appropriate status is written. Errors which don't conform to `GRPCStatusTransformable`
   /// return a status with code `.internalError`.
   public func errorCaught(context: ChannelHandlerContext, error: Error) {
-    errorDelegate?.observe(error)
+    errorDelegate?.observeLibraryError(error)
 
-    let transformed = errorDelegate?.transform(error) ?? error
-    let status = (transformed as? GRPCStatusTransformable)?.asGRPCStatus() ?? GRPCStatus.processingError
+    let status = errorDelegate?.transformLibraryError(error)
+      ?? (error as? GRPCStatusTransformable)?.asGRPCStatus()
+      ?? .processingError
     sendErrorStatus(status)
   }
 

Sources/SwiftGRPCNIO/CallHandlers/BidirectionalStreamingCallHandler.swift

@@ -19,7 +19,7 @@ public class BidirectionalStreamingCallHandler<RequestMessage: Message, Response
   // If authentication fails, they can simply fail the observer future, which causes the call to be terminated.
   public init(channel: Channel, request: HTTPRequestHead, errorDelegate: ServerErrorDelegate?, eventObserverFactory: (StreamingResponseCallContext<ResponseMessage>) -> EventLoopFuture<EventObserver>) {
     super.init(errorDelegate: errorDelegate)
-    let context = StreamingResponseCallContextImpl<ResponseMessage>(channel: channel, request: request)
+    let context = StreamingResponseCallContextImpl<ResponseMessage>(channel: channel, request: request, errorDelegate: errorDelegate)
     self.callContext = context
     let eventObserver = eventObserverFactory(context)
     self.eventObserver = eventObserver

Sources/SwiftGRPCNIO/CallHandlers/ClientStreamingCallHandler.swift

@@ -19,7 +19,7 @@ public class ClientStreamingCallHandler<RequestMessage: Message, ResponseMessage
   // If authentication fails, they can simply fail the observer future, which causes the call to be terminated.
   public init(channel: Channel, request: HTTPRequestHead, errorDelegate: ServerErrorDelegate?, eventObserverFactory: (UnaryResponseCallContext<ResponseMessage>) -> EventLoopFuture<EventObserver>) {
     super.init(errorDelegate: errorDelegate)
-    let callContext = UnaryResponseCallContextImpl<ResponseMessage>(channel: channel, request: request)
+    let callContext = UnaryResponseCallContextImpl<ResponseMessage>(channel: channel, request: request, errorDelegate: errorDelegate)
     self.callContext = callContext
     let eventObserver = eventObserverFactory(callContext)
     self.eventObserver = eventObserver

Sources/SwiftGRPCNIO/CallHandlers/ServerStreamingCallHandler.swift

@@ -15,7 +15,7 @@ public class ServerStreamingCallHandler<RequestMessage: Message, ResponseMessage
 
   public init(channel: Channel, request: HTTPRequestHead, errorDelegate: ServerErrorDelegate?, eventObserverFactory: (StreamingResponseCallContext<ResponseMessage>) -> EventObserver) {
     super.init(errorDelegate: errorDelegate)
-    let callContext = StreamingResponseCallContextImpl<ResponseMessage>(channel: channel, request: request)
+    let callContext = StreamingResponseCallContextImpl<ResponseMessage>(channel: channel, request: request, errorDelegate: errorDelegate)
     self.callContext = callContext
     self.eventObserver = eventObserverFactory(callContext)
     callContext.statusPromise.futureResult.whenComplete { _ in

Sources/SwiftGRPCNIO/CallHandlers/UnaryCallHandler.swift

@@ -16,7 +16,7 @@ public class UnaryCallHandler<RequestMessage: Message, ResponseMessage: Message>
 
   public init(channel: Channel, request: HTTPRequestHead, errorDelegate: ServerErrorDelegate?, eventObserverFactory: (UnaryResponseCallContext<ResponseMessage>) -> EventObserver) {
     super.init(errorDelegate: errorDelegate)
-    let callContext = UnaryResponseCallContextImpl<ResponseMessage>(channel: channel, request: request)
+    let callContext = UnaryResponseCallContextImpl<ResponseMessage>(channel: channel, request: request, errorDelegate: errorDelegate)
     self.callContext = callContext
     self.eventObserver = eventObserverFactory(callContext)
     callContext.responsePromise.futureResult.whenComplete { _ in

Sources/SwiftGRPCNIO/GRPCChannelHandler.swift

@@ -41,10 +41,11 @@ extension GRPCChannelHandler: ChannelInboundHandler, RemovableChannelHandler {
   public typealias OutboundOut = RawGRPCServerResponsePart
 
   public func errorCaught(context: ChannelHandlerContext, error: Error) {
-    errorDelegate?.observe(error)
+    errorDelegate?.observeLibraryError(error)
 
-    let transformedError = errorDelegate?.transform(error) ?? error
-    let status = (transformedError as? GRPCStatusTransformable)?.asGRPCStatus() ?? GRPCStatus.processingError
+    let status = errorDelegate?.transformLibraryError(error)
+      ?? (error as? GRPCStatusTransformable)?.asGRPCStatus()
+      ?? .processingError
     context.writeAndFlush(wrapOutboundOut(.status(status)), promise: nil)
   }
 

Sources/SwiftGRPCNIO/HTTPProtocolSwitcher.swift

@@ -121,7 +121,7 @@ extension HTTPProtocolSwitcher: ChannelInboundHandler, RemovableChannelHandler {
   public func errorCaught(context: ChannelHandlerContext, error: Error) {
     switch self.state {
     case .notConfigured, .configuring:
-      errorDelegate?.observe(error)
+      errorDelegate?.observeLibraryError(error)
       context.close(mode: .all, promise: nil)
 
     case .configured:

Sources/SwiftGRPCNIO/LoggingServerErrorDelegate.swift

@@ -18,7 +18,11 @@ import Foundation
 public class LoggingServerErrorDelegate: ServerErrorDelegate {
   public init() {}
 
-  public func observe(_ error: Error) {
-    print("[grpc-server][\(Date())] \(error)")
+  public func observeLibraryError(_ error: Error) {
+    print("[grpc-server][\(Date())] library: \(error)")
+  }
+
+  public func observeRequestHandlerError(_ error: Error) {
+    print("[grpc-server][\(Date())] request handler: \(error)")
   }
 }

Sources/SwiftGRPCNIO/ServerCallContexts/StreamingResponseCallContext.swift

@@ -28,15 +28,25 @@ open class StreamingResponseCallContext<ResponseMessage: Message>: ServerCallCon
 open class StreamingResponseCallContextImpl<ResponseMessage: Message>: StreamingResponseCallContext<ResponseMessage> {
   public let channel: Channel
 
-  public init(channel: Channel, request: HTTPRequestHead) {
+  /// - Parameters:
+  ///   - channel: The NIO channel the call is handled on.
+  ///   - request: The headers provided with this call.
+  ///   - errorDelegate: Provides a means for transforming status promise failures to `GRPCStatusTransformable` before
+  ///     sending them to the client.
+  ///
+  ///     Note: `errorDelegate` is not called for status promise that are `succeeded` with a non-OK status.
+  public init(channel: Channel, request: HTTPRequestHead, errorDelegate: ServerErrorDelegate?) {
     self.channel = channel
 
     super.init(eventLoop: channel.eventLoop, request: request)
 
     statusPromise.futureResult
       // Ensure that any error provided can be transformed to `GRPCStatus`, using "internal server error" as a fallback.
-      .recover { error in
-        (error as? GRPCStatusTransformable)?.asGRPCStatus() ?? .processingError
+      .recover { [weak errorDelegate] error in
+        errorDelegate?.observeRequestHandlerError(error, request: request)
+        return errorDelegate?.transformRequestHandlerError(error, request: request)
+          ?? (error as? GRPCStatusTransformable)?.asGRPCStatus()
+          ?? .processingError
       }
       // Finish the call by returning the final status.
       .whenSuccess {

Sources/SwiftGRPCNIO/ServerCallContexts/UnaryResponseCallContext.swift

@@ -41,7 +41,12 @@ public protocol StatusOnlyCallContext: ServerCallContext {
 open class UnaryResponseCallContextImpl<ResponseMessage: Message>: UnaryResponseCallContext<ResponseMessage> {
   public let channel: Channel
 
-  public init(channel: Channel, request: HTTPRequestHead) {
+  /// - Parameters:
+  ///   - channel: The NIO channel the call is handled on.
+  ///   - request: The headers provided with this call.
+  ///   - errorDelegate: Provides a means for transforming response promise failures to `GRPCStatusTransformable` before
+  ///     sending them to the client.
+  public init(channel: Channel, request: HTTPRequestHead, errorDelegate: ServerErrorDelegate?) {
     self.channel = channel
 
     super.init(eventLoop: channel.eventLoop, request: request)
@@ -55,8 +60,11 @@ open class UnaryResponseCallContextImpl<ResponseMessage: Message>: UnaryResponse
         self.responseStatus
       }
       // Ensure that any error provided can be transformed to `GRPCStatus`, using "internal server error" as a fallback.
-      .recover { error in
-        (error as? GRPCStatusTransformable)?.asGRPCStatus() ?? .processingError
+      .recover { [weak errorDelegate] error in
+        errorDelegate?.observeRequestHandlerError(error, request: request)
+        return errorDelegate?.transformRequestHandlerError(error, request: request)
+          ?? (error as? GRPCStatusTransformable)?.asGRPCStatus()
+          ?? .processingError
       }
       // Finish the call by returning the final status.
       .whenSuccess { status in

Sources/SwiftGRPCNIO/ServerErrorDelegate.swift

@@ -15,26 +15,56 @@
  */
 import Foundation
 import NIO
+import NIOHTTP1
 
 public protocol ServerErrorDelegate: class {
   //! FIXME: Provide more context about where the error was thrown, i.e. using `GRPCError`.
   /// Called when an error is thrown in the channel pipeline.
-  func observe(_ error: Error)
+  func observeLibraryError(_ error: Error)
 
-  /// Transforms the given error into a new error.
+  /// Transforms the given error (thrown somewhere inside the gRPC library) into a new error.
   ///
-  /// This allows framework users to transform errors which may be out of their control
-  /// due to third-party libraries, for example, into more meaningful errors or
-  /// `GRPCStatus` errors. Errors returned from this protocol are not passed to
-  /// `observe`.
+  /// This allows library users to transform errors which may be out of their control
+  /// into more meaningful `GRPCStatus` errors before they are sent to the user.
   ///
   /// - note:
-  /// This defaults to returning the provided error.
-  func transform(_ error: Error) -> Error
+  /// Errors returned by this method are not passed to `observe` again.
+  ///
+  /// - note:
+  /// This defaults to returning `nil`. In that case, if the original error conforms to `GRPCStatusTransformable`,
+  /// that error's `asGRPCStatus()` result will be sent to the user. If that's not the case, either,
+  /// `GRPCStatus.processingError` is returned.
+  func transformLibraryError(_ error: Error) -> GRPCStatus?
+
+  /// Called when a request's status or response promise is failed somewhere in the user-provided request handler code.
+  /// - Parameters:
+  ///   - error: The original error the status/response promise was failed with.
+  ///   - request: The headers of the request whose status/response promise was failed.
+  func observeRequestHandlerError(_ error: Error, request: HTTPRequestHead)
+
+  /// Transforms the given status or response promise failure into a new error.
+  ///
+  /// This allows library users to transform errors which happen during their handling of the request
+  /// into more meaningful `GRPCStatus` errors before they are sent to the user.
+  ///
+  /// - note:
+  /// Errors returned by this method are not passed to `observe` again.
+  ///
+  /// - note:
+  /// This defaults to returning `nil`. In that case, if the original error conforms to `GRPCStatusTransformable`,
+  /// that error's `asGRPCStatus()` result will be sent to the user. If that's not the case, either,
+  /// `GRPCStatus.processingError` is returned.
+  ///
+  /// - Parameters:
+  ///   - error: The original error the status/response promise was failed with.
+  ///   - request: The headers of the request whose status/response promise was failed.
+  func transformRequestHandlerError(_ error: Error, request: HTTPRequestHead) -> GRPCStatus?
 }
 
 public extension ServerErrorDelegate {
-  func transform(_ error: Error) -> Error {
-    return error
-  }
+  func observeLibraryError(_ error: Error) { }
+  func transformLibraryError(_ error: Error) -> GRPCStatus? { return nil }
+
+  func observeRequestHandlerError(_ error: Error, request: HTTPRequestHead) { }
+  func transformRequestHandlerError(_ error: Error, request: HTTPRequestHead) -> GRPCStatus? { return nil }
 }

Tests/SwiftGRPCNIOTests/GRPCChannelHandlerResponseCapturingTestCase.swift

@@ -28,7 +28,7 @@ class CollectingServerErrorDelegate: ServerErrorDelegate {
     return (self.asGRPCErrors?.map { $0.error }) as? [GRPCCommonError]
   }
 
-  func observe(_ error: Error) {
+  func observeLibraryError(_ error: Error) {
     self.errors.append(error)
   }
 }

Tests/SwiftGRPCNIOTests/NIOBasicEchoTestCase.swift

@@ -104,9 +104,7 @@ class NIOEchoTestCaseBase: XCTestCase {
   let serverEventLoopGroup: EventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
   let clientEventLoopGroup: EventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
 
-  var transportSecurity: TransportSecurity {
-    return .none
-  }
+  var transportSecurity: TransportSecurity { return .none }
 
   var server: GRPCServer!
   var client: Echo_EchoService_NIOClient!
@@ -117,6 +115,7 @@ class NIOEchoTestCaseBase: XCTestCase {
       port: 5050,
       eventLoopGroup: self.serverEventLoopGroup,
       serviceProviders: [makeEchoProvider()],
+      errorDelegate: makeErrorDelegate(),
       tls: try self.transportSecurity.makeServerTLS()
     ).wait()
   }
@@ -130,9 +129,9 @@ class NIOEchoTestCaseBase: XCTestCase {
     ).wait()
   }
 
-  func makeEchoProvider() -> Echo_EchoProvider_NIO {
-    return EchoProviderNIO()
-  }
+  func makeEchoProvider() -> Echo_EchoProvider_NIO { return EchoProviderNIO() }
+
+  func makeErrorDelegate() -> ServerErrorDelegate? { return nil }
 
   func makeEchoClient() throws -> Echo_EchoService_NIOClient {
     return Echo_EchoService_NIOClient(connection: try self.makeClientConnection())

Tests/SwiftGRPCNIOTests/ServerThrowingTests.swift

@@ -21,31 +21,32 @@ import NIOHTTP2
 @testable import SwiftGRPCNIO
 import XCTest
 
-private let expectedError = GRPCStatus(code: .internalError, message: "expected error")
+let thrownError = GRPCStatus(code: .internalError, message: "expected error")
+let transformedError = GRPCStatus(code: .aborted, message: "transformed error")
 
 // Motivation for two different providers: Throwing immediately causes the event observer future (in the
 // client-streaming and bidi-streaming cases) to throw immediately, _before_ the corresponding handler has even added
 // to the channel. We want to test that case as well as the one where we throw only _after_ the handler has been added
 // to the channel.
-private class ImmediateThrowingEchoProviderNIO: Echo_EchoProvider_NIO {
+class ImmediateThrowingEchoProviderNIO: Echo_EchoProvider_NIO {
   func get(request: Echo_EchoRequest, context: StatusOnlyCallContext) -> EventLoopFuture<Echo_EchoResponse> {
-    return context.eventLoop.makeFailedFuture(expectedError)
+    return context.eventLoop.makeFailedFuture(thrownError)
   }
 
   func expand(request: Echo_EchoRequest, context: StreamingResponseCallContext<Echo_EchoResponse>) -> EventLoopFuture<GRPCStatus> {
-    return context.eventLoop.makeFailedFuture(expectedError)
+    return context.eventLoop.makeFailedFuture(thrownError)
   }
 
   func collect(context: UnaryResponseCallContext<Echo_EchoResponse>) -> EventLoopFuture<(StreamEvent<Echo_EchoRequest>) -> Void> {
-    return context.eventLoop.makeFailedFuture(expectedError)
+    return context.eventLoop.makeFailedFuture(thrownError)
   }
 
   func update(context: StreamingResponseCallContext<Echo_EchoResponse>) -> EventLoopFuture<(StreamEvent<Echo_EchoRequest>) -> Void> {
-    return context.eventLoop.makeFailedFuture(expectedError)
+    return context.eventLoop.makeFailedFuture(thrownError)
   }
 }
 
-private extension EventLoop {
+extension EventLoop {
   func makeFailedFuture<T>(_ error: Error, delay: TimeInterval) -> EventLoopFuture<T> {
     return self.scheduleTask(in: .nanoseconds(TimeAmount.Value(delay * 1000 * 1000 * 1000))) { () }.futureResult
       .flatMapThrowing { _ -> T in throw error }
@@ -53,47 +54,53 @@ private extension EventLoop {
 }
 
 /// See `ImmediateThrowingEchoProviderNIO`.
-private class DelayedThrowingEchoProviderNIO: Echo_EchoProvider_NIO {
+class DelayedThrowingEchoProviderNIO: Echo_EchoProvider_NIO {
   func get(request: Echo_EchoRequest, context: StatusOnlyCallContext) -> EventLoopFuture<Echo_EchoResponse> {
-    return context.eventLoop.makeFailedFuture(expectedError, delay: 0.01)
+    return context.eventLoop.makeFailedFuture(thrownError, delay: 0.01)
   }
 
   func expand(request: Echo_EchoRequest, context: StreamingResponseCallContext<Echo_EchoResponse>) -> EventLoopFuture<GRPCStatus> {
-    return context.eventLoop.makeFailedFuture(expectedError, delay: 0.01)
+    return context.eventLoop.makeFailedFuture(thrownError, delay: 0.01)
   }
 
   func collect(context: UnaryResponseCallContext<Echo_EchoResponse>) -> EventLoopFuture<(StreamEvent<Echo_EchoRequest>) -> Void> {
-    return context.eventLoop.makeFailedFuture(expectedError, delay: 0.01)
+    return context.eventLoop.makeFailedFuture(thrownError, delay: 0.01)
   }
 
   func update(context: StreamingResponseCallContext<Echo_EchoResponse>) -> EventLoopFuture<(StreamEvent<Echo_EchoRequest>) -> Void> {
-    return context.eventLoop.makeFailedFuture(expectedError, delay: 0.01)
+    return context.eventLoop.makeFailedFuture(thrownError, delay: 0.01)
   }
 }
 
 /// Ensures that fulfilling the status promise (where possible) with an error yields the same result as failing the future.
-private class ErrorReturningEchoProviderNIO: ImmediateThrowingEchoProviderNIO {
+class ErrorReturningEchoProviderNIO: ImmediateThrowingEchoProviderNIO {
   // There's no status promise to fulfill for unary calls (only the response promise), so that case is omitted.
 
   override func expand(request: Echo_EchoRequest, context: StreamingResponseCallContext<Echo_EchoResponse>) -> EventLoopFuture<GRPCStatus> {
-    return context.eventLoop.makeSucceededFuture(expectedError)
+    return context.eventLoop.makeSucceededFuture(thrownError)
   }
 
   override func collect(context: UnaryResponseCallContext<Echo_EchoResponse>) -> EventLoopFuture<(StreamEvent<Echo_EchoRequest>) -> Void> {
     return context.eventLoop.makeSucceededFuture({ _ in
-      context.responseStatus = expectedError
+      context.responseStatus = thrownError
       context.responsePromise.succeed(Echo_EchoResponse())
     })
   }
 
   override func update(context: StreamingResponseCallContext<Echo_EchoResponse>) -> EventLoopFuture<(StreamEvent<Echo_EchoRequest>) -> Void> {
     return context.eventLoop.makeSucceededFuture({ _ in
-      context.statusPromise.succeed(expectedError)
+      context.statusPromise.succeed(thrownError)
     })
   }
 }
 
+private class ErrorTransformingDelegate: ServerErrorDelegate {
+  func transformRequestHandlerError(_ error: Error, request: HTTPRequestHead) -> GRPCStatus? { return transformedError }
+}
+
 class ServerThrowingTests: NIOEchoTestCaseBase {
+  var expectedError: GRPCStatus { return thrownError }
+
   override func makeEchoProvider() -> Echo_EchoProvider_NIO { return ImmediateThrowingEchoProviderNIO() }
 }
 
@@ -105,6 +112,12 @@ class ClientThrowingWhenServerReturningErrorTests: ServerThrowingTests {
   override func makeEchoProvider() -> Echo_EchoProvider_NIO { return ErrorReturningEchoProviderNIO() }
 }
 
+class ServerErrorTransformingTests: ServerThrowingTests {
+  override var expectedError: GRPCStatus { return transformedError }
+
+  override func makeErrorDelegate() -> ServerErrorDelegate? { return ErrorTransformingDelegate() }
+}
+
 extension ServerThrowingTests {
   func testUnary() throws {
     let call = client.get(Echo_EchoRequest(text: "foo"))

Tests/SwiftGRPCNIOTests/XCTestManifests.swift

@@ -168,6 +168,18 @@ extension ServerDelayedThrowingTests {
     ]
 }
 
+extension ServerErrorTransformingTests {
+    // DO NOT MODIFY: This is autogenerated, use:
+    //   `swift test --generate-linuxmain`
+    // to regenerate.
+    static let __allTests__ServerErrorTransformingTests = [
+        ("testBidirectionalStreaming", testBidirectionalStreaming),
+        ("testClientStreaming", testClientStreaming),
+        ("testServerStreaming", testServerStreaming),
+        ("testUnary", testUnary),
+    ]
+}
+
 extension ServerThrowingTests {
     // DO NOT MODIFY: This is autogenerated, use:
     //   `swift test --generate-linuxmain`
@@ -193,6 +205,7 @@ public func __allTests() -> [XCTestCaseEntry] {
         testCase(NIOFunctionalTestsMutualAuthentication.__allTests__NIOFunctionalTestsMutualAuthentication),
         testCase(NIOServerWebTests.__allTests__NIOServerWebTests),
         testCase(ServerDelayedThrowingTests.__allTests__ServerDelayedThrowingTests),
+        testCase(ServerErrorTransformingTests.__allTests__ServerErrorTransformingTests),
         testCase(ServerThrowingTests.__allTests__ServerThrowingTests),
     ]
 }