really no double handlerRemoved calls (#1094)

Motivation:

In #1091 we fixed the most common case of handlerRemoved being called
multiple times. There was still another (less likely) problem left which
is if a manual removal is triggered from _within_ the handlerRemoved
call from a pipeline teardown, then we would still call handlerRemoved
another time.

Modifications:

Properly guard against handlerRemoved being called twice.

Result:

Fewer bugs.

Sources/NIO/ChannelPipeline.swift

@@ -488,23 +488,14 @@ public final class ChannelPipeline: ChannelInvoker {
 
         let nextCtx = context.next
         let prevCtx = context.prev
-        var inThePipeline = false
 
         if let prevCtx = prevCtx {
-            inThePipeline = true
             prevCtx.next = nextCtx
         }
         if let nextCtx = nextCtx {
-            inThePipeline = true
             nextCtx.prev = prevCtx
         }
 
-        guard inThePipeline else {
-            // if both next and prev are nil already, then we were previously removed from the pipeline
-            promise?.succeed(())
-            return
-        }
-
         do {
             try context.invokeHandlerRemoved()
             promise?.succeed(())
@@ -1108,6 +1099,7 @@ public final class ChannelHandlerContext: ChannelInvoker {
     public let name: String
     private let inboundHandler: _ChannelInboundHandler?
     private let outboundHandler: _ChannelOutboundHandler?
+    private var removeHandlerInvoked = false
 
     // Only created from within ChannelPipeline
     fileprivate init(name: String, handler: ChannelHandler, pipeline: ChannelPipeline) {
@@ -1487,6 +1479,10 @@ public final class ChannelHandlerContext: ChannelInvoker {
 
     fileprivate func invokeHandlerRemoved() throws {
         self.eventLoop.assertInEventLoop()
+        guard !self.removeHandlerInvoked else {
+            return
+        }
+        self.removeHandlerInvoked = true
 
         handler.handlerRemoved(context: self)
     }

Tests/NIOTests/ChannelPipelineTest+XCTest.swift

@@ -60,6 +60,7 @@ extension ChannelPipelineTest {
                 ("testNonRemovableChannelHandlerIsNotRemovable", testNonRemovableChannelHandlerIsNotRemovable),
                 ("testAddMultipleHandlers", testAddMultipleHandlers),
                 ("testPipelineDebugDescription", testPipelineDebugDescription),
+                ("testWeDontCallHandlerRemovedTwiceIfAHandlerCompletesRemovalOnlyAfterChannelTeardown", testWeDontCallHandlerRemovedTwiceIfAHandlerCompletesRemovalOnlyAfterChannelTeardown),
            ]
    }
 }

Tests/NIOTests/ChannelPipelineTest.swift

@@ -1171,6 +1171,123 @@ class ChannelPipelineTest: XCTestCase {
                HTTPHandler ↓↑ HTTPHandler            [handler2]
         """)
     }
+
+    func testWeDontCallHandlerRemovedTwiceIfAHandlerCompletesRemovalOnlyAfterChannelTeardown() {
+        enum State: Int {
+            // When we start the test,
+            case testStarted = 0
+            // we send a trigger event,
+            case triggerEventRead = 1
+            // when receiving the trigger event, we start the manual removal (which won't complete).
+            case manualRemovalStarted = 2
+            // Instead, we now close the channel to force a pipeline teardown,
+            case pipelineTeardown = 3
+            // which will make `handlerRemoved` called, from where
+            case handlerRemovedCalled = 4
+            // we also complete the manual removal.
+            case manualRemovalCompleted = 5
+            // And hopefully we never reach the error state.
+            case error = 999
+
+            mutating func next() {
+                if let newState = State(rawValue: self.rawValue + 1) {
+                    self = newState
+                } else {
+                    XCTFail("there's no next state starting from \(self)")
+                    self = .error
+                }
+            }
+        }
+        class Handler: ChannelInboundHandler, RemovableChannelHandler {
+            typealias InboundIn = State
+            typealias InboundOut = State
+
+            private(set) var state: State = .testStarted
+            private var removalToken: ChannelHandlerContext.RemovalToken? = nil
+            private let allDonePromise: EventLoopPromise<Void>
+
+            init(allDonePromise: EventLoopPromise<Void>) {
+                self.allDonePromise = allDonePromise
+            }
+
+            func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+                let step = self.unwrapInboundIn(data)
+                self.state.next()
+                XCTAssertEqual(self.state, step)
+
+                // just to communicate to the outside where we are in our state machine
+                context.fireChannelRead(self.wrapInboundOut(self.state))
+
+                switch step {
+                case .triggerEventRead:
+                    // Step 1: Okay, let's kick off the manual removal (it won't complete)
+                    context.pipeline.removeHandler(self).map {
+                        // When the manual removal completes, we advance the state.
+                        self.state.next()
+                    }.cascade(to: self.allDonePromise)
+                default:
+                    XCTFail("channelRead called in state \(self.state)")
+                }
+            }
+
+            func removeHandler(context: ChannelHandlerContext, removalToken: ChannelHandlerContext.RemovalToken) {
+                self.state.next()
+                XCTAssertEqual(.manualRemovalStarted, self.state)
+
+                // Step 2: Save the removal token that we got from kicking off the manual removal (in step 1)
+                self.removalToken = removalToken
+            }
+
+            func handlerRemoved(context: ChannelHandlerContext) {
+                self.state.next()
+                XCTAssertEqual(.pipelineTeardown, self.state)
+
+                // Step 3: We'll call our own channelRead which will advance the state.
+                self.completeTheManualRemoval(context: context)
+            }
+
+            func completeTheManualRemoval(context: ChannelHandlerContext) {
+                self.state.next()
+                XCTAssertEqual(.handlerRemovedCalled, self.state)
+
+                // just to communicate to the outside where we are in our state machine
+                context.fireChannelRead(self.wrapInboundOut(self.state))
+
+                // Step 4: This happens when the pipeline is being torn down, so let's now also finish the manual
+                // removal process.
+                self.removalToken.map(context.leavePipeline(removalToken:))
+            }
+        }
+
+        let eventLoop = EmbeddedEventLoop()
+        let allDonePromise = eventLoop.makePromise(of: Void.self)
+        let handler = Handler(allDonePromise: allDonePromise)
+        let channel = EmbeddedChannel(handler: handler, loop: eventLoop)
+        XCTAssertNoThrow(try channel.connect(to: .init(ipAddress: "1.2.3.4", port: 5)).wait())
+
+        XCTAssertEqual(.testStarted, handler.state)
+        XCTAssertNoThrow(try channel.writeInbound(State.triggerEventRead))
+        XCTAssertNoThrow(XCTAssertEqual(State.triggerEventRead, try channel.readInbound()))
+        XCTAssertNoThrow(XCTAssertNil(try channel.readInbound()))
+
+        XCTAssertNoThrow(try {
+            // we'll get a left-over event on close which triggers the pipeline teardown and therefore continues the
+            // process.
+            switch try channel.finish() {
+            case .clean:
+                XCTFail("expected output")
+            case .leftOvers(inbound: let inbound, outbound: let outbound, pendingOutbound: let pendingOutbound):
+                XCTAssertEqual(0, outbound.count)
+                XCTAssertEqual(0, pendingOutbound.count)
+                XCTAssertEqual(1, inbound.count)
+                XCTAssertEqual(.handlerRemovedCalled, inbound.first?.tryAs(type: State.self))
+            }
+        }())
+
+        XCTAssertEqual(.manualRemovalCompleted, handler.state)
+
+        XCTAssertNoThrow(try allDonePromise.futureResult.wait())
+    }
 }
 
 // this should be within `testAddMultipleHandlers` but https://bugs.swift.org/browse/SR-9956