     _ part: GRPCServerResponsePart<ResponsePayload>,
     promise: EventLoopPromise<Void>?
   ) {
-    self.act(on: self.state.sendResponsePartFromObserver(part, promise: promise))
+    let forward: Bool
+
+    switch part {
+    case .metadata:
+      forward = self.state.sendResponsePartFromObserver(.metadata)
+    case .message:
+      forward = self.state.sendResponsePartFromObserver(.message)
+    case .end:
+      forward = self.state.sendResponsePartFromObserver(.end)
+    }
+
+    if forward {
+      self.sendResponsePartToInterceptors(part, promise: promise)
+    } else {
+      promise?.fail(GRPCError.AlreadyComplete())
+    }
   }
 
   /// Processes a library error to form a `GRPCStatus` and trailers to send back to the client.

    case requestClosedResponseOpen
    case requestClosedResponseClosed

    enum Filter {
      case allow
      case drop
    }

    mutating func receiveHeaders() -> Filter {
      switch self {
      case .requestIdleResponseIdle:
        self = .requestOpenResponseIdle
        return .allow

      case .requestOpenResponseIdle,
           .requestOpenResponseOpen,
           .requestClosedResponseIdle,
           .requestClosedResponseOpen,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    func receiveMessage() -> Filter {
      switch self {
      case .requestOpenResponseIdle,
           .requestOpenResponseOpen:
        return .allow

      case .requestIdleResponseIdle,
           .requestClosedResponseIdle,
           .requestClosedResponseOpen,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    mutating func receiveEnd() -> Filter {
      switch self {
      case .requestOpenResponseIdle:
        self = .requestClosedResponseIdle
        return .allow

      case .requestOpenResponseOpen:
        self = .requestClosedResponseOpen
        return .allow

      case .requestIdleResponseIdle,
           .requestClosedResponseIdle,
           .requestClosedResponseOpen,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    mutating func sendHeaders() -> Filter {
      switch self {
      case .requestOpenResponseIdle:
        self = .requestOpenResponseOpen
        return .allow

      case .requestClosedResponseIdle:
        self = .requestClosedResponseOpen
        return .allow

      case .requestIdleResponseIdle,
           .requestOpenResponseOpen,
           .requestClosedResponseOpen,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    func sendMessage() -> Filter {
      switch self {
      case .requestOpenResponseOpen,
           .requestClosedResponseOpen:
        return .allow

      case .requestIdleResponseIdle,
           .requestOpenResponseIdle,
           .requestClosedResponseIdle,
           .requestClosedResponseClosed:
        return .drop
      }
    }

    mutating func sendEnd() -> Filter {
      switch self {
      case .requestIdleResponseIdle:
        return .drop

      case .requestOpenResponseIdle,
           .requestOpenResponseOpen,
           .requestClosedResponseIdle,
           .requestClosedResponseOpen:
        self = .requestClosedResponseClosed
        return .allow

      case .requestClosedResponseClosed:
        return .drop
      }
    }
  }

  fileprivate struct ActiveState {
    var context: ChannelHandlerContext

    /// The stream state between the 'Channel' and interceptor pipeline.
    var channelStreamState: StreamState

    /// The stream state between the interceptor pipeline and event observer.
    var observerStreamState: StreamState

    init(context: ChannelHandlerContext) {
      self.context = context
      self.channelStreamState = .requestIdleResponseIdle
      self.observerStreamState = .requestIdleResponseIdle
    }
  }
}

extension _BaseCallHandler.State {
  fileprivate enum Action {
    /// Do nothing.
    case none

    /// Receive the request part in the interceptor pipeline.
    case receiveRequestPartInInterceptors(GRPCServerRequestPart<_BaseCallHandler.RequestPayload>)

    /// Receive the request part in the observer.
    case receiveRequestPartInObserver(GRPCServerRequestPart<_BaseCallHandler.RequestPayload>)

    /// Receive an error in the observer.
    case receiveLibraryErrorInObserver(Error)

    /// Send a response part to the interceptor pipeline.
    case sendResponsePartToInterceptors(
      GRPCServerResponsePart<_BaseCallHandler.ResponsePayload>,
      EventLoopPromise<Void>?
    )

    /// Write the response part to the `Channel`.
    case writeResponsePartToChannel(
      ChannelHandlerContext,
      GRPCServerResponsePart<_BaseCallHandler.ResponsePayload>,
      promise: EventLoopPromise<Void>?
    )

    /// Complete the promise with the result.
    case completePromise(EventLoopPromise<Void>?, Result<Void, Error>)

    /// Perform multiple actions.
    indirect case multiple([Action])
  }
}

extension _BaseCallHandler.State {
  /// The handler was added to the `ChannelPipeline`: this is the only way to move from the `.idle`
  /// state. We only expect this to be called once.
  internal mutating func handlerAdded(context: ChannelHandlerContext) -> Action {
    switch self {
    case .idle:
      // This is the only way we can become active.
      self = .active(.init(context: context))
      return .none

    case .active:
      preconditionFailure("Invalid state: already active")

    case .closed:
      return .none
    }
  }

  /// Received an error from the `Channel`.
  internal mutating func errorCaught(_ error: Error) -> Action {
    switch self {
    case .active:
      return .receiveLibraryErrorInObserver(error)

    case .idle, .closed:
      return .none
    }
  }

  /// Receive a request part from the `Channel`. If we're active we just forward these through the
  /// pipeline. We validate at the other end.
  internal mutating func channelRead(
    _ requestPart: GRPCServerRequestPart<_BaseCallHandler.RequestPayload>
  ) -> Action {
    switch self {
    case .idle:
      preconditionFailure("Invalid state: the handler isn't in the pipeline yet")

    case var .active(state):
      // Avoid CoW-ing state.
      self = .idle

      let filter: StreamState.Filter
      let part: GRPCServerRequestPart<_BaseCallHandler.RequestPayload>

      switch requestPart {
      case let .metadata(headers):
        filter = state.channelStreamState.receiveHeaders()
        part = .metadata(headers)
      case let .message(message):
        filter = state.channelStreamState.receiveMessage()
        part = .message(message)
      case .end:
        filter = state.channelStreamState.receiveEnd()
        part = .end
      }

      // Restore state.
      self = .active(state)

      switch filter {
      case .allow:
        return .receiveRequestPartInInterceptors(part)
      case .drop:
        return .none
      }

    case .closed:
      return .none
    }
  }

  /// Send a response part from the observer to the interceptors.
  internal mutating func sendResponsePartFromObserver(
    _ part: GRPCServerResponsePart<_BaseCallHandler.ResponsePayload>,
    promise: EventLoopPromise<Void>?
  ) -> Action {
    switch self {
    case .idle:
      preconditionFailure("Invalid state: the handler isn't in the pipeline yet")

    case var .active(state):
      // Avoid CoW-ing 'state'.
      self = .idle

      let filter: StreamState.Filter

      switch part {
      case .metadata:
        filter = state.observerStreamState.sendHeaders()
      case .message:
        filter = state.observerStreamState.sendMessage()
      case .end:
        filter = state.observerStreamState.sendEnd()
      }

      // Restore the state.
      self = .active(state)

      switch filter {
      case .allow:
        return .sendResponsePartToInterceptors(part, promise)
      case .drop:
        return .completePromise(promise, .failure(GRPCError.AlreadyComplete()))
      }

    case .closed:
      return .completePromise(promise, .failure(GRPCError.AlreadyComplete()))
    }
  }

  /// Send a response part from the interceptors to the `Channel`.
  internal mutating func sendResponsePartFromInterceptors(
    _ part: GRPCServerResponsePart<_BaseCallHandler.ResponsePayload>,
    promise: EventLoopPromise<Void>?
  ) -> Action {
    switch self {
    case .idle:
      preconditionFailure("Invalid state: can't send response on idle call")

      // Avoid CoW-ing `state`.
      self = .idle

      let allow: Bool

      // Does the active state allow us to send this?
      switch part {
      case .metadata:
        allow = state.observerStreamState.receiveHeaders()
      case .message:
        allow = state.observerStreamState.receiveMessage()
      case .end:
        allow = state.observerStreamState.receiveEnd()
      }

      // Put `state` back.
      self = .active(state)
      return allow

    case .closed:
      // We're closed, just ignore this.
      return false
    }
  }
}
