fix(quic): close inbound connection when handler rejects it (#499)

The inbound QUIC server path invokes the application connection handler
synchronously inside channelActive. When the handler rejects the
connection by throwing (e.g. PeerAlreadyConnectedException for a
duplicate/simultaneous peer), the handshake-waiter handler has already
removed itself, so the exception reaches the Netty pipeline tail and
logs a noisy 'exceptionCaught reached tail of pipeline' warning while
leaking the rejected QuicChannel until idle timeout.

Catch the exception in routeInboundHandshake and close the channel,
mirroring the dial paths which already close on handler failure. Narrow
the guard to catch(Exception) rather than Throwable so fatal Errors
(OutOfMemoryError, LinkageError, ...) propagate instead of being
downgraded to a routine connection close; logging stays at debug.

Author: lucassaldanha

libp2p/src/main/kotlin/io/libp2p/transport/quic/QuicTransport.kt

@@ -131,8 +131,24 @@ class QuicTransport(
             // Route this validated inbound connection to the waiting hole-punch caller.
             pending.future.complete(holePunchChannel())
         } else {
-            // Normal path: deliver to the connection handler.
-            exposeConnection()
+            // Normal path: deliver to the connection handler. The application handler may reject
+            // the connection by throwing (e.g. PeerAlreadyConnectedException for a duplicate or
+            // simultaneous connection to an already-connected peer). This runs inside the server
+            // pipeline's channelActive, and the handshake-waiter handler has already removed itself
+            // by this point, so an escaping exception would reach the Netty pipeline tail and log a
+            // noisy "exceptionCaught reached tail of pipeline" warning while leaking the channel.
+            // Close the rejected connection instead, mirroring the dial paths. Only Exception is
+            // caught — fatal Errors (OutOfMemoryError, LinkageError, ...) must propagate rather than
+            // be downgraded to a routine rejection. The library cannot tell a deliberate rejection
+            // from an application bug (the handler contract has no typed rejection), so the log stays
+            // at debug to avoid reintroducing the noise this guard removes; the application owns
+            // logging of its own connection decisions.
+            try {
+                exposeConnection()
+            } catch (e: Exception) {
+                logger.debug("Inbound connection rejected by handler; closing", e)
+                closeChannel()
+            }
         }
     }
 

libp2p/src/test/kotlin/io/libp2p/transport/quic/QuicInboundHandshakeRoutingTest.kt

@@ -6,6 +6,8 @@ import io.libp2p.security.tls.TlsPeerIdentity
 import io.mockk.mockk
 import io.netty.handler.codec.quic.QuicChannel
 import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatCode
+import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.jupiter.api.Test
 import java.util.concurrent.CompletableFuture
 
@@ -107,4 +109,56 @@ class QuicInboundHandshakeRoutingTest {
         assertThat(exposed).isTrue()
         assertThat(closed).isFalse()
     }
+
+    @Test
+    fun `closes the connection when the handler rejects a normal inbound connection`() {
+        val transport = transport()
+        val inbound = identity()
+
+        var closed = false
+
+        // The application connection handler may reject an inbound connection by throwing
+        // (e.g. PeerAlreadyConnectedException for a duplicate/simultaneous connection). That
+        // exception must NOT escape routeInboundHandshake: in the live server pipeline the
+        // handshake-waiter handler has already removed itself by the time the connection is
+        // exposed, so an escaping exception reaches the Netty pipeline tail and logs a noisy
+        // "exceptionCaught reached tail of pipeline" warning. Instead the rejected connection
+        // must be closed, mirroring the dial paths.
+        assertThatCode {
+            transport.routeInboundHandshake(
+                remoteIdentity = inbound,
+                pending = null,
+                prepareConnection = { },
+                closeChannel = { closed = true },
+                holePunchChannel = { error("no hole punch is pending") },
+                exposeConnection = { throw RuntimeException("Already connected to peer") }
+            )
+        }.doesNotThrowAnyException()
+
+        assertThat(closed).isTrue()
+    }
+
+    @Test
+    fun `does not swallow a fatal Error thrown by the handler`() {
+        val transport = transport()
+        val inbound = identity()
+
+        var closed = false
+
+        // A fatal Error (OutOfMemoryError, LinkageError, ...) is NOT a routine connection rejection
+        // and must propagate rather than be silently downgraded to a closed connection. Only
+        // Exception is caught by routeInboundHandshake.
+        assertThatThrownBy {
+            transport.routeInboundHandshake(
+                remoteIdentity = inbound,
+                pending = null,
+                prepareConnection = { },
+                closeChannel = { closed = true },
+                holePunchChannel = { error("no hole punch is pending") },
+                exposeConnection = { throw OutOfMemoryError("simulated fatal error") }
+            )
+        }.isInstanceOf(OutOfMemoryError::class.java)
+
+        assertThat(closed).isFalse()
+    }
 }