Fix ConnectionClosedException propagating to Netty event loop from MuxChannel.doWrite (libp2p#455)

In Netty 4.2.x the original code's throw+catch pattern caused the exception to
escape doWrite() uncaught, surfacing as spurious "PLEASE FIX OR REPORT" noise in
Teku logs. The throw was caught locally, but buf.remove(cause) — called from the
catch block without its own guard — can propagate exceptions back through
promise-listener callbacks in certain Netty versions.

Fix: move the localDisconnected check outside the try/catch and wrap the
buf.remove(cause) call defensively so nothing can escape doWrite(). Pending write
promises are still properly failed with ConnectionClosedException.

Author: lucassaldanha

libp2p/src/main/kotlin/io/libp2p/etc/util/netty/mux/MuxChannel.kt

@@ -32,13 +32,20 @@ class MuxChannel<TData>(
         initializer(this)
     }
 
+    @Suppress("SwallowedException")
     override fun doWrite(buf: ChannelOutboundBuffer) {
         while (true) {
             val msg = buf.current() ?: break
+            if (localDisconnected) {
+                // Must not throw from doWrite — exceptions escape uncaught to the Netty event loop.
+                // Wrap buf.remove() defensively: in some Netty versions promise listeners triggered
+                // by buf.remove() can propagate back through it.
+                try {
+                    buf.remove(ConnectionClosedException("The stream was closed for writing locally: $id"))
+                } catch (e: Throwable) { }
+                continue
+            }
             try {
-                if (localDisconnected) {
-                    throw ConnectionClosedException("The stream was closed for writing locally: $id")
-                }
                 // the msg is released by both onChildWrite and buf.remove() so we need to retain
                 // however it is still to be confirmed that no buf leaks happen here TODO
                 ReferenceCountUtil.retain(msg)

libp2p/src/test/kotlin/io/libp2p/mux/MuxHandlerAbstractTest.kt

@@ -6,6 +6,7 @@ import io.libp2p.core.StreamHandler
 import io.libp2p.etc.types.fromHex
 import io.libp2p.etc.types.getX
 import io.libp2p.etc.types.toHex
+import io.libp2p.etc.util.netty.mux.MuxChannel
 import io.libp2p.etc.util.netty.mux.RemoteWriteClosed
 import io.libp2p.etc.util.netty.nettyInitializer
 import io.libp2p.mux.MuxHandlerAbstractTest.AbstractTestMuxFrame.Flag.*
@@ -442,6 +443,26 @@ abstract class MuxHandlerAbstractTest {
         }
     }
 
+    @Test
+    fun `write with localDisconnected should fail promise without throwing from doWrite`() {
+        val handler = openStreamLocal()
+        readFrameOrThrow()
+
+        // Simulate the state between localDisconnected=true and deactivate() in doDisconnect(),
+        // which is when a queued WriteTask can reach doWrite with localDisconnected=true while
+        // the channel is still active (flush0 would take the "not-yet-connected" path otherwise).
+        @Suppress("UNCHECKED_CAST")
+        (handler.ctx.channel() as MuxChannel<ByteBuf>).localDisconnected = true
+
+        val writeFuture = handler.ctx.writeAndFlush(allocateMessage("42"))
+        ech.runPendingTasks()
+
+        assertTrue(writeFuture.isDone)
+        assertThrows(ConnectionClosedException::class.java) {
+            writeFuture.sync()
+        }
+    }
+
     @Test
     fun `should throw when writing to reset stream`() {
         val handler = openStreamLocal()