NIOSingleStepByteToMessageDecoder reentrancy safety (apple#2881)

### Motivation:

`NIOSingleStepByteToMessageDecoder` calls out part way through its
processing step to a user-provided closure which can cause re-entrant
behavior which violates the assumption made in the code that if the
buffer is non-empty at the start that will be true later in the method.

### Modifications:

`NIOSingleStepByteToMessageDecoder` no longer assumes a non-empty buffer
in its final phase of buffer management.

Further changes to protect against re-entrancy shouldn't be necessary
because the outside call to `messageReceiver` is the only one which is
permitted to be re-entrant (`decode` and `decodeLast` are not).

### Result:

`NIOSingleStepByteToMessageDecoder` can handle re-entrant processing
calls.

Sources/NIOCore/SingleStepByteToMessageDecoder.swift

@@ -233,8 +233,9 @@ public final class NIOSingleStepByteToMessageProcessor<Decoder: NIOSingleStepByt
         }
 
         self._buffer = nil  // To avoid CoW
+        defer { self._buffer = buffer }
+
         let result = try body(&buffer)
-        self._buffer = buffer
         return result
     }
 
@@ -272,8 +273,9 @@ public final class NIOSingleStepByteToMessageProcessor<Decoder: NIOSingleStepByt
             throw ByteToMessageDecoderError.PayloadTooLargeError()
         }
 
-        // force unwrapping is safe because nil case is handled already and asserted above
-        if self._buffer!.readerIndex > 0, self.decoder.shouldReclaimBytes(buffer: self._buffer!) {
+        if let readerIndex = self._buffer?.readerIndex, readerIndex > 0,
+            self.decoder.shouldReclaimBytes(buffer: self._buffer!)
+        {
             self._buffer!.discardReadBytes()
         }
     }

Tests/NIOCoreTests/SingleStepByteToMessageDecoderTest.swift

@@ -549,4 +549,44 @@ public final class NIOSingleStepByteToMessageDecoderTest: XCTestCase {
         XCTAssertEqual(2, messageReceiver.count)
         XCTAssertEqual(processor.unprocessedBytes, 0)
     }
+
+    /// Tests re-entrancy by having a nested decoding operation empty the buffer and exit part way
+    /// through the outer processing step
+    func testErrorDuringNestedDecoding() {
+        class ThrowingOnLastDecoder: NIOSingleStepByteToMessageDecoder {
+            /// `ByteBuffer` is the expected type passed in.
+            public typealias InboundIn = ByteBuffer
+            /// `ByteBuffer`s will be passed to the next stage.
+            public typealias InboundOut = ByteBuffer
+
+            public init() {}
+
+            struct DecodeLastError: Error {}
+
+            func decodeLast(buffer: inout NIOCore.ByteBuffer, seenEOF: Bool) throws -> NIOCore.ByteBuffer? {
+                buffer = ByteBuffer()  // to allow the decode loop to exit
+                throw DecodeLastError()
+            }
+
+            func decode(buffer: inout NIOCore.ByteBuffer) throws -> NIOCore.ByteBuffer? {
+                ByteBuffer()
+            }
+        }
+
+        let decoder = ThrowingOnLastDecoder()
+        let b2mp = NIOSingleStepByteToMessageProcessor(decoder)
+        var errorObserved = false
+        XCTAssertNoThrow(
+            try b2mp.process(buffer: ByteBuffer(string: "1\n\n2\n3\n")) { line in
+                // We will throw an error to exit the decoding within the nested process call prematurely.
+                // Unless this is carefully handled we can be left in an inconsistent state which the outer call will encounter
+                do {
+                    try b2mp.finishProcessing(seenEOF: true) { _ in }
+                } catch _ as ThrowingOnLastDecoder.DecodeLastError {
+                    errorObserved = true
+                }
+            }
+        )
+        XCTAssertTrue(errorObserved)
+    }
 }