Fix exclusive access violation in `NIOAsyncChannelOutboundWriterHandl…

…er` (#2580)

# Motivation
We were setting `self.sink = nil` in the `NIOAsyncChannelOutboundWriterHandler` twice in the same call stack which is an exclusivity violation. This happens because the first `self.sink = nil` triggers the `didTerminate` delegate call which again triggered `self.sink = nil`.

# Modification
This PR changes the code to only call `self.sink?.finish()` and only sets the `sink` to `nil` in the `didTerminate` implementation. This follows what we do for the inbound handler implementation. I also added a test that triggers this exclusivity violation.

# Result
No more exclusivity violations in our code.

Sources/NIOCore/AsyncChannel/AsyncChannelOutboundWriterHandler.swift

@@ -133,7 +133,7 @@ internal final class NIOAsyncChannelOutboundWriterHandler<OutboundOut: Sendable>
     @inlinable
     func handlerRemoved(context: ChannelHandlerContext) {
         self.context = nil
-        self.sink = nil
+        self.sink?.finish()
     }
 
     @inlinable

Tests/NIOCoreTests/AsyncChannel/AsyncChannelTests.swift

@@ -18,6 +18,25 @@ import NIOEmbedded
 import XCTest
 
 final class AsyncChannelTests: XCTestCase {
+    func testAsyncChannelCloseOnWrite() async throws {
+        guard #available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) else { return }
+        final class CloseOnWriteHandler: ChannelOutboundHandler {
+            typealias OutboundIn = String
+
+            func write(context: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
+                context.close(promise: promise)
+            }
+        }
+        let channel = NIOAsyncTestingChannel()
+        let wrapped = try await channel.testingEventLoop.executeInContext {
+            try channel.pipeline.syncOperations.addHandler(CloseOnWriteHandler())
+            return try NIOAsyncChannel<String, String>(synchronouslyWrapping: channel)
+        }
+
+        try await wrapped.outbound.write("Test")
+        try await channel.closeFuture.get()
+    }
+
     func testAsyncChannelBasicFunctionality() async throws {
         guard #available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) else { return }
         let channel = NIOAsyncTestingChannel()