ByteToMessageProcessor: re-entrancy bugs

[weissi]

Expected behavior

ByteToMessageProcessor itself must be able to handle re-entrancy.

Actual behavior

Crash on re-entrancy.

        // buffer can only be nil if we're called from finishProcessing which is handled above
        assert(self._buffer != nil)   // <-- FINAL NIL CHECK

        func decodeOnce(buffer: inout ByteBuffer) throws -> Decoder.InboundOut? {
            if decodeMode == .normal {
                return try self.decoder.decode(buffer: &buffer)
            } else {
                return try self.decoder.decodeLast(buffer: &buffer, seenEOF: seenEOF)
            }
        }

        while let message = try self._withNonCoWBuffer(decodeOnce) {
            try messageReceiver(message) // <-- WHOOPSIE CALLOUT
        }

        if let maximumBufferSize = self.maximumBufferSize, self._buffer!.readableBytes > maximumBufferSize {
            throw ByteToMessageDecoderError.PayloadTooLargeError()
        }

        // vvvvv---- this is a lie, the user callout above may invalidate this.
        // force unwrapping is safe because nil case is handled already and asserted above
        if self._buffer!.readerIndex > 0, self.decoder.shouldReclaimBytes(buffer: self._buffer!) {

Steps to reproduce

For example one common case (even in B2MP's example code) is:

class Handler: ChannelInboundHandler {
    [...]

    func channelRead(...) {
        [...]
        try self.b2mp.process(buffer: data) { message in
            context.fireChannelRead(...(message)) // this may call `channelInactive` on the same stack
        }
    }

    func channelInactive(...) { // may be called re-entrantly
        [...]
        try self.b2mp.finishProcessing(...) { message in // this may be called within the `process` closure
            context.fireChannelRead(...(message))
        }
    }
}

The actual code I got this crash with was a unit test (the code isn't finished yet but it shows the crash):

    func testBasicSingleStepEOFDuringDecoding() {
        let decoder = LineBasedFrameDecoder()
        let b2mp = NIOSingleStepByteToMessageProcessor(decoder)
        var callCount = 0
        XCTAssertNoThrow(try b2mp.process(buffer: ByteBuffer(string: "1\n\n2\n3\n")) { line in
            callCount += 1
            switch callCount {
            case 1:
                XCTAssertEqual(ByteBuffer(string: "1"), line)
                XCTAssertNoThrow(try b2mp.finishProcessing(seenEOF: true) { line in
                    print(line)
                })
            case 2:
                XCTAssertEqual(ByteBuffer(string: ""), line)
            case 3:
                XCTAssertEqual(ByteBuffer(string: "2"), line)
            case 4:
                XCTAssertEqual(ByteBuffer(string: "3"), line)
            default:
                XCTFail("not expecting call no \(callCount)")
            }
        })
    }