Tolerate zeroReturnError in closed/unwrapped (#323)

Motivation:

In some rare cases we've seen production applications crash due to us
getting a zeroReturnError in either closed or perhaps unwrapped. This is
pretty hard to get to trigger in realistic situations, but I've found a
contrived-but-possible way to hit it in tests. In both of these cases we
are able to simply ignore the error and move on.

Modifications:

- Added a test to replicate the issue
- Tolerate .unwrapped and .closed on zeroReturnError.

Result:

Fewer weird crashes

Sources/NIOSSL/NIOSSLHandler.swift

@@ -379,8 +379,13 @@ public class NIOSSLHandler : ChannelInboundHandler, ChannelOutboundHandler, Remo
 
             case .failed(BoringSSLError.zeroReturn):
                 switch self.state {
-                case .idle, .closed, .unwrapped, .handshaking:
+                case .idle, .handshaking:
                     preconditionFailure("Should not get zeroReturn in \(self.state)")
+                case .closed, .unwrapped:
+                    // This is an unexpected place to be, but it's not totally impossible. Assume this
+                    // is the result of a wonky I/O pattern and just ignore it.
+                    self.plaintextReadBuffer = receiveBuffer
+                    break readLoop
                 case .active:
                     self.state = .closing(self.scheduleTimedOutShutdown(context: context))
                 case .unwrapping, .closing:

Tests/NIOSSLTests/NIOSSLIntegrationTest+XCTest.swift

@@ -63,6 +63,7 @@ extension NIOSSLIntegrationTest {
                 ("testKeyLoggingClientAndServer", testKeyLoggingClientAndServer),
                 ("testLoadsOfCloses", testLoadsOfCloses),
                 ("testWriteFromFailureOfWrite", testWriteFromFailureOfWrite),
+                ("testChannelInactiveDuringHandshakeSucceeded", testChannelInactiveDuringHandshakeSucceeded),
                 ("testTrustedFirst", testTrustedFirst),
            ]
    }

Tests/NIOSSLTests/NIOSSLIntegrationTest.swift

@@ -2046,6 +2046,83 @@ class NIOSSLIntegrationTest: XCTestCase {
         serverChannel.pipeline.fireChannelInactive()
     }
 
+    func testChannelInactiveDuringHandshakeSucceeded() throws {
+        // This test aims to reproduce a very unusual crash. I've never been able to come up with a clear justification of
+        // how we managed to hit it, but it goes a bit like this:
+        //
+        // 1. During a TLS handshake, a server performs a `channelRead` that triggers a handshakeCompleted message.
+        // 2. Synchronously, during that pipeline traversal, we end up with a read buffer that also contains a CLOSE_NOTIFY alert.
+        //     This may have arrived in the same packet with the handshake completion message, or as a result of something we did.
+        // 3. Additionally, we manage to synchronously enter the .closed or .unwrapped state. This is hard to imagine, but it can
+        //     happen in a few ways: channelInactive forces this transition, and managing to do a shutdown reentrantly due to extra
+        //     I/O can trigger it as well.
+        // 4. We then progress through the handshake and crash.
+        //
+        // To make this manifest in the test we use a pair of promises and finagle the I/O such that everything goes wrong at once.
+        // This can indicate how unusual the circumstance is in which this happens. Nonetheless, we've seen it happen on production
+        // systems.
+        let serverChannel = EmbeddedChannel()
+        let clientChannel = EmbeddedChannel()
+        defer {
+            // Both were closed uncleanly in the test, but the server error was already
+            // consumed.
+            XCTAssertNoThrow(try serverChannel.finish())
+            XCTAssertThrowsError(try clientChannel.finish())
+        }
+
+        let context = try configuredSSLContext()
+        let clientChannelCompletedPromise = clientChannel.eventLoop.makePromise(of: Void.self)
+        let clientChannelCompletedHandler = WaitForHandshakeHandler(handshakeResultPromise: clientChannelCompletedPromise)
+        let serverChannelCompletedPromise = serverChannel.eventLoop.makePromise(of: Void.self)
+        let serverChannelCompletedHandler = WaitForHandshakeHandler(handshakeResultPromise: serverChannelCompletedPromise)
+
+        clientChannelCompletedPromise.futureResult.whenSuccess {
+            // Here we need to immediately (and _recursively_) ask the client channel to shutdown. This should force a CLOSE_NOTIFY
+            // message out in the same tick as the handshake message.
+            clientChannel.close(promise: nil)
+
+            // Now deliver all the client messages to the server channel _in one go_.
+            var flattenedBytes = clientChannel.allocator.buffer(capacity: 1024)
+            while let clientDatum = try! clientChannel.readOutbound(as: ByteBuffer.self) {
+                flattenedBytes.writeImmutableBuffer(clientDatum)
+            }
+
+            // Can't use XCTAssertThrowsError here, this function call isn't allowed to throw.
+            do {
+                try serverChannel.writeInbound(flattenedBytes)
+                XCTFail("Expected to throw")
+            } catch {
+                guard case .some(.uncleanShutdown) = error as? NIOSSLError else {
+                    XCTFail("Unexpected error \(error)")
+                    return
+                }
+            }
+        }
+
+        serverChannelCompletedPromise.futureResult.whenSuccess {
+            // Here we do something very, very dangerous: we call fireChannelInactive on our own channel.
+            // This simulates us hitting a close condition in some other form.
+            serverChannel.pipeline.fireChannelInactive()
+        }
+
+        XCTAssertNoThrow(
+            try serverChannel.pipeline.syncOperations.addHandlers([NIOSSLServerHandler(context: context), serverChannelCompletedHandler])
+        )
+        XCTAssertNoThrow(
+            try clientChannel.pipeline.syncOperations.addHandlers([try NIOSSLClientHandler(context: context, serverHostname: nil), clientChannelCompletedHandler])
+        )
+
+        // Do the handshake.
+        let addr: SocketAddress = try SocketAddress(unixDomainSocketPath: "/tmp/whatever")
+        let connectFuture = clientChannel.connect(to: addr)
+        serverChannel.pipeline.fireChannelActive()
+        try interactInMemory(clientChannel: clientChannel, serverChannel: serverChannel)
+        try connectFuture.wait()
+
+        // We now need to forcibly shutdown the client channel, as otherwise it'll hang waiting for a server that never comes back.
+        clientChannel.pipeline.fireChannelInactive()
+    }
+
     func testTrustedFirst() throws {
         // We need to explain this test a bit.
         //