[kyo-net] Closed exceptions that name the failing connection, not the driver

## Problem

A user reported that Kyo's shared HTTP client wedges under load ([full report](https://gist.github.com/baldram/5c21566eacb4a5a361b6a60ead5af76d)). Under sustained concurrent traffic with connection churn (16 threads each running 200 requests against a server that restarted every 25ms, on the NIO backend), requests started failing with `NioIoDriver[sel=...] ... is closed. finishConnect failed`. The IO driver is a process-wide singleton shared by every `HttpClient`, every failure in the run named the same selector, and the failures did not stop, so the report concluded the shared driver had gone permanently closed for the life of the process, one failure taking down every later call.

That conclusion was wrong, and the error message is why. What happened during the churn is ordinary: an individual `finishConnect` lost its race against a server that had just restarted, and that one connection failed. But the failure was reported as `Closed` naming the driver, at a frame pointing into driver internals, so a routine per-connection failure read as the whole shared driver dying. The message turned a single connection failing into an apparent process-wide outage.

## Solution

`Closed` now names the resource that actually closed and carries that resource's own creation frame: a `connection` at the site it was opened, a `listener` at the site it began listening. A driver-internal frame remains only for a genuine whole-driver shutdown, the one case where the driver is the resource.

Failures that are not closures stop pretending to be. An OS-level read, connect, or accept error is now a typed `NetException` carrying the errno, which the transport still wraps into the public `NetConnectException` the caller already handles. Carrying the typed connect and accept errors meant widening an internal promise's error row from `Closed` to `Closed | NetException`, threaded through the real promise storage with no cast that narrows it back. The public exception types are unchanged.

The change adds two cross-backend resilience suites that did not exist before, `TransportResilienceTest` and `HttpServerResilienceTest`. They drive NIO, the kqueue and epoll poller, and io_uring through the failure modes the report implicated: connection churn, in-flight cancellation, RST and FIN and half-close, and connection-pool integrity, each time asserting that a co-tenant connection on the same shared driver keeps working while another one fails. They were the reproduction that showed the driver never wedges, and they stay as the guard that a per-connection failure can never again read as a driver-wide outage.

## Notes

- The reported permanent wedge does not exist. The resilience suites reproduce the report's load and never produce one: the driver stays live and keeps serving connections under the churn. The only defect the report actually surfaced is the misleading message, and that is what the repro validated.
- A separate io_uring bug, unrelated to the report, is fixed here too. A single transient accept error (`EINTR`, `ECONNABORTED`, or momentary file-descriptor exhaustion) made the driver fail the accept promise, and the accept loop reads any accept failure as the listener closing, so one transient errno permanently wedged the listener. It now re-arms the accept instead of failing it.
- That re-arm is deferred by a backoff for the two file-descriptor-exhaustion errnos. On `EMFILE`/`ENFILE` the kernel leaves the pending connection in the backlog, so re-arming immediately re-reaps the same errno at once and spins the reap carrier at full CPU. Those re-arm after a backoff instead, matching the poller, tunable through the new `-Dkyo.net.acceptResourceBackoff` flag (milliseconds, default 50), and guarded by a reproduction on a real io_uring ring.
- Validated on a real io_uring ring in a Linux container, including the end-of-run file-descriptor leak check, plus the NIO, poller, and JS drivers on the host, across JVM, JS, and Native.

Author: fwbrasil

build.sbt

@@ -2134,7 +2134,7 @@ lazy val `kyo-http` =
         .crossType(CrossType.Full)
         .in(file("kyo-http"))
         .dependsOn(`kyo-core`, `kyo-config`, `kyo-schema-json`)
-        .dependsOn(`kyo-net`)
+        .dependsOn(`kyo-net` % "compile->compile;test->test")
         .withKyoTest
         .settings(
             `kyo-settings`

kyo-http/jvm/src/test/scala/kyo/HttpServerResilienceTest.scala

@@ -1,6 +1,7 @@
 package kyo.net.internal
 
 import kyo.*
+import kyo.net.NetConnectionIoException
 import kyo.net.internal.transport.*
 import kyo.net.internal.util.HandleId
 import scala.scalajs.js
@@ -13,7 +14,7 @@ import scala.scalajs.js
   * Leftover bytes are stored when a `"data"` chunk arrives before `awaitRead` has been called (or when a chunk contains more data than the
   * pending read can consume). They are delivered on the next `awaitRead` call without resuming the socket.
   */
-final private[kyo] class JsHandle private[kyo] (val socket: js.Dynamic, val id: HandleId):
+final private[kyo] class JsHandle private[kyo] (val socket: js.Dynamic, val id: HandleId, val createdAt: Frame):
     // Pending read promise (at most one). Absent when no read is pending.
     var pendingRead: Maybe[Promise.Unsafe[ReadOutcome, Abort[Closed]]] = Absent
 
@@ -55,9 +56,9 @@ private[kyo] object JsHandle:
     private[kyo] case class Leftover(buf: Array[Byte], off: Int, len: Int)
 
     /** Create a `JsHandle` from a connected, paused Node.js socket and attach permanent `data`, `end`, `close`, and `error` listeners. */
-    def init(socket: js.Dynamic, driver: IoDriver[JsHandle])(using AllowUnsafe, Frame): JsHandle =
+    def init(socket: js.Dynamic, driver: IoDriver[JsHandle], createdAt: Frame)(using AllowUnsafe): JsHandle =
         // JS has no file-descriptor concept; use 0 as the fd placeholder so HandleId.next produces a process-unique id.
-        val handle = new JsHandle(socket, HandleId.next(0))
+        val handle = new JsHandle(socket, HandleId.next(0), createdAt)
 
         // Permanent "data" listener
         discard(socket.on(
@@ -91,11 +92,20 @@ private[kyo] object JsHandle:
         discard(socket.on("close", signalEof))
         discard(socket.on(
             "error",
-            { (_: js.Dynamic) =>
+            { (err: js.Dynamic) =>
                 handle.pendingRead match
                     case Present(pending) =>
                         handle.clearPendingRead()
-                        pending.completeDiscard(Result.fail(Closed(driver.label, summon[Frame], "socket error")))
+                        // A Node "error" is a hard receive error on a live read: surface it as a typed receive failure on the read outcome (not a
+                        // closure), carrying the socket's own creation frame and the error's message as the cause.
+                        val cause = if js.typeOf(err.message) == "string" then err.message.toString else ""
+                        pending.completeDiscard(Result.succeed(ReadOutcome.Failed(
+                            NetConnectionIoException(
+                                s"connection socket#${handle.id}",
+                                NetConnectionIoException.Operation.Receive,
+                                cause
+                            )(using handle.createdAt)
+                        )))
                     case Absent => ()
             }: js.Function1[js.Dynamic, Unit]
         ))

kyo-net/js-wasm/src/main/scala/kyo/net/internal/JsHandle.scala

@@ -1,6 +1,8 @@
 package kyo.net.internal
 
 import kyo.*
+import kyo.net.NetDriverUnsupportedException
+import kyo.net.NetException
 import kyo.net.internal.transport.*
 import kyo.scheduler.IOPromise
 import scala.scalajs.js
@@ -64,7 +66,7 @@ final private[kyo] class JsIoDriver private (
             promise.completeDiscard(Result.succeed(ReadOutcome.PeerFin))
         else if handle.socket.destroyed.asInstanceOf[Boolean] then
             // (iii) Socket destroyed (RST, or a completed close) with nothing staged: fail Closed.
-            promise.completeDiscard(Result.fail(Closed(label, summon[Frame], s"socket destroyed")))
+            promise.completeDiscard(Result.fail(Closed(s"connection ${handleLabel(handle)}", handle.createdAt, "socket destroyed")))
         else
             // (iv) Request the next chunk: the permanent 'data' listener delivers it (or 'end'/'error' the EOF/failure).
             handle.pendingRead = Present(promise)
@@ -89,20 +91,20 @@ final private[kyo] class JsIoDriver private (
         end if
     end isPeerClosed
 
-    def awaitConnect(handle: JsHandle, promise: Promise.Unsafe[Unit, Abort[Closed]])(using AllowUnsafe, Frame): Unit =
+    def awaitConnect(handle: JsHandle, promise: Promise.Unsafe[Unit, Abort[Closed | NetException]])(using AllowUnsafe, Frame): Unit =
         // JS connect is handled via Node.js 'connect' event callback, not via the driver
         promise.completeDiscard(Result.succeed(()))
 
-    def awaitAccept(handle: JsHandle, promise: Promise.Unsafe[Int, Abort[Closed]])(using AllowUnsafe, Frame): Unit =
+    def awaitAccept(handle: JsHandle, promise: Promise.Unsafe[Int, Abort[Closed | NetException]])(using AllowUnsafe, Frame): Unit =
         // JS does not run the PosixTransport accept loop; fail fast so a caller that accidentally uses this path gets an immediate error.
-        promise.completeDiscard(Result.fail(Closed(label, summon[Frame], s"awaitAccept not supported on JsIoDriver")))
+        promise.completeDiscard(Result.Panic(NetDriverUnsupportedException(label, "awaitAccept")))
 
-    def awaitWritable(handle: JsHandle, promise: Promise.Unsafe[Unit, Abort[Closed]])(using AllowUnsafe, Frame): Unit =
+    def awaitWritable(handle: JsHandle, promise: Promise.Unsafe[Unit, Abort[Closed | NetException]])(using AllowUnsafe, Frame): Unit =
         // Node's net.Socket#destroyed is a documented boolean property; js.Dynamic erases that to an untyped JS value, so recovering the typed
         // Boolean needs this narrowing cast. Safe per Node's documented property type; it cannot dissolve without a typed facade for Node's
         // net.Socket.
         if handle.socket.destroyed.asInstanceOf[Boolean] then
-            promise.completeDiscard(Result.fail(Closed(label, summon[Frame], s"socket destroyed")))
+            promise.completeDiscard(Result.fail(Closed(s"connection ${handleLabel(handle)}", handle.createdAt, "socket destroyed")))
         else
             // Register one-shot listeners for drain/close/error
             var drainFn: js.Function0[Unit]             = null
@@ -125,7 +127,7 @@ final private[kyo] class JsIoDriver private (
 
             def completeFailure(reason: String): Unit =
                 removeAll()
-                promise.completeDiscard(Result.fail(Closed(label, summon[Frame], reason)))
+                promise.completeDiscard(Result.fail(Closed(s"connection ${handleLabel(handle)}", handle.createdAt, reason)))
 
             drainFn = (() => completeSuccess()): js.Function0[Unit]
             closeFn = (() => completeFailure("socket closed before writable")): js.Function0[Unit]
@@ -156,7 +158,7 @@ final private[kyo] class JsIoDriver private (
 
     def cancel(handle: JsHandle)(using AllowUnsafe, Frame): Unit =
         discard(handle.socket.pause())
-        val closed = Closed(label, summon[Frame], s"${handleLabel(handle)} canceled")
+        val closed = Closed(s"connection ${handleLabel(handle)}", handle.createdAt, "canceled")
         handle.pendingRead match
             case Present(pending) =>
                 pending.completeDiscard(Result.fail(closed))

kyo-net/js-wasm/src/main/scala/kyo/net/internal/JsIoDriver.scala

@@ -375,7 +375,7 @@ final private[kyo] class JsTransport private (
         host: String,
         port: Int,
         connectTimeout: Duration
-    )(using AllowUnsafe, Frame): () => Unit =
+    )(using allow: AllowUnsafe, frame: Frame): () => Unit =
         if connectTimeout.isFinite then
             // Disarming INTERRUPTS the timer, which completes it and fires the callback below, so that callback must distinguish "the deadline
             // elapsed" from "the deadline was called off". Harmless while the only disarm ran after `promise` had already settled (the
@@ -394,7 +394,7 @@ final private[kyo] class JsTransport private (
             def disarm(): Unit =
                 // Set BEFORE the interrupt, so the callback the interrupt triggers observes it.
                 disarmed.set(true)
-                timer.interruptDiscard(Result.Panic(Closed("JsTransport", summon[Frame], "connect completed before deadline")))
+                timer.interruptDiscard(Result.Panic(Interrupted(frame, "connect completed before deadline")))
             end disarm
             // Backstop for every way this connect can end before its TCP phase completes.
             promise.onComplete(_ => disarm())
@@ -412,8 +412,8 @@ final private[kyo] class JsTransport private (
         config: kyo.net.NetConfig,
         handshakeTimeout: Duration = Duration.Infinity
     )(using
-        AllowUnsafe,
-        Frame
+        allow: AllowUnsafe,
+        frame: Frame
     ): Fiber.Unsafe[NetConnection, Abort[NetException]] =
         val promise = new IOPromise[NetException, Connection[JsHandle]]
         val driver  = pool.next()
@@ -453,11 +453,7 @@ final private[kyo] class JsTransport private (
                                 discard(socket.destroy())
                         }
                         promise.onComplete { _ =>
-                            deadline.interruptDiscard(Result.Panic(Closed(
-                                "JsTransport",
-                                summon[Frame],
-                                "handshake settled before deadline"
-                            )))
+                            deadline.interruptDiscard(Result.Panic(Interrupted(frame, "handshake settled before deadline")))
                         }
                     end if
                 }: js.Function0[Unit]
@@ -471,7 +467,7 @@ final private[kyo] class JsTransport private (
             connectEvent,
             { () =>
                 if tcpNoDelay then discard(socket.setNoDelay(true))
-                val handle = JsHandle.init(socket, driver)
+                val handle = JsHandle.init(socket, driver, frame)
                 handle.peerCloseGrace = config.peerCloseGrace
                 val connection = Connection.init(handle, driver, config.channelCapacity, config.peerCloseGrace)
                 // Wire upgrade function so upgradeToTls dispatches to this transport.
@@ -550,7 +546,7 @@ final private[kyo] class JsTransport private (
         config: kyo.net.NetConfig,
         onClose: Maybe[() => Unit] = Absent,
         acceptHandshakeCount: Maybe[() => Int] = Absent
-    )(using AllowUnsafe, Frame): Fiber.Unsafe[NetListener, Abort[NetException]] =
+    )(using allow: AllowUnsafe, frame: Frame): Fiber.Unsafe[NetListener, Abort[NetException]] =
         val promise = new IOPromise[NetException, NetListener]
         rejectUnsupportedBuffers(config) match
             case Present(e) =>
@@ -559,7 +555,7 @@ final private[kyo] class JsTransport private (
                 return promise.asInstanceOf[Fiber.Unsafe[NetListener, Abort[NetException]]]
             case Absent => ()
         end match
-        val listener = new JsListener(server, NetAddress.Tcp(host, port))
+        val listener = new JsListener(server, NetAddress.Tcp(host, port), frame)
         onClose.foreach(listener.onClose)
         acceptHandshakeCount.foreach(listener.acceptHandshakeCount)
 
@@ -570,7 +566,7 @@ final private[kyo] class JsTransport private (
                 if tcpNoDelay then discard(socket.setNoDelay(true))
 
                 val connDriver = pool.next()
-                val handle     = JsHandle.init(socket, connDriver)
+                val handle     = JsHandle.init(socket, connDriver, listener.createdAt)
                 handle.peerCloseGrace = config.peerCloseGrace
                 val connection = Connection.init(handle, connDriver, config.channelCapacity, config.peerCloseGrace)
                 // Accepted connection: a STARTTLS upgrade through the public upgradeToTls runs in the TLS server role (upgradeToTls reads
@@ -630,8 +626,8 @@ final private[kyo] class JsTransport private (
     end listenServer
 
     def connectUnix(path: String, connectTimeout: Duration, config: kyo.net.NetConfig)(using
-        AllowUnsafe,
-        Frame
+        allow: AllowUnsafe,
+        frame: Frame
     ): Fiber.Unsafe[NetConnection, Abort[NetException]] =
         kyo.net.Transport.checkConnectTimeout(connectTimeout)
         val promise = new IOPromise[NetException, Connection[JsHandle]]
@@ -661,7 +657,7 @@ final private[kyo] class JsTransport private (
             "connect",
             { () =>
                 // Unix sockets do not support TCP_NODELAY: skip setNoDelay
-                val handle = JsHandle.init(socket, driver)
+                val handle = JsHandle.init(socket, driver, frame)
                 handle.peerCloseGrace = config.peerCloseGrace
                 val connection = Connection.init(handle, driver, config.channelCapacity, config.peerCloseGrace)
                 // Wire upgrade function so upgradeToTls dispatches to this transport.
@@ -695,8 +691,8 @@ final private[kyo] class JsTransport private (
     end connectUnix
 
     override def stdio(channelCapacity: Int, readChunkSize: Int)(using
-        AllowUnsafe,
-        Frame
+        allow: AllowUnsafe,
+        frame: Frame
     ): Fiber.Unsafe[NetConnection, Abort[NetException]] =
         if !stdioClaimed.compareAndSet(false, true) then
             // Exactly one stdio per process: fds 0/1 are process-global, so double-ownership is rejected.
@@ -707,7 +703,7 @@ final private[kyo] class JsTransport private (
             // JsHandle/JsIoDriver expect a single socket-like object, so the shim presents one whose read events
             // come from stdin and whose write goes to stdout. destroy() is a no-op: the process owns fds 0/1.
             val shim   = stdioShim()
-            val handle = JsHandle.init(shim, driver)
+            val handle = JsHandle.init(shim, driver, frame)
             // stdio keeps peerCloseGrace = Infinity: no TCP peer to reclaim against.
             val connection = Connection.init(handle, driver, channelCapacity)
             if connection.start() then
@@ -771,7 +767,7 @@ final private[kyo] class JsTransport private (
 
     def listenUnix(path: String, backlog: Int, config: kyo.net.NetConfig)(
         handler: NetConnection => Unit
-    )(using AllowUnsafe, Frame): Fiber.Unsafe[NetListener, Abort[NetException]] =
+    )(using allow: AllowUnsafe, frame: Frame): Fiber.Unsafe[NetListener, Abort[NetException]] =
         val promise = new IOPromise[NetException, NetListener]
         rejectUnsupportedBuffers(config) match
             case Present(e) =>
@@ -783,7 +779,7 @@ final private[kyo] class JsTransport private (
         val net    = js.Dynamic.global.require("net")
         val server = net.createServer()
 
-        val listener = new JsListener(server, NetAddress.Unix(path))
+        val listener = new JsListener(server, NetAddress.Unix(path), frame)
 
         discard(server.on(
             "connection",
@@ -792,7 +788,7 @@ final private[kyo] class JsTransport private (
                 // Unix sockets do not support TCP_NODELAY: skip setNoDelay
 
                 val connDriver = pool.next()
-                val handle     = JsHandle.init(socket, connDriver)
+                val handle     = JsHandle.init(socket, connDriver, listener.createdAt)
                 handle.peerCloseGrace = config.peerCloseGrace
                 val connection = Connection.init(handle, connDriver, config.channelCapacity, config.peerCloseGrace)
                 // Accepted connection: a STARTTLS upgrade through the public upgradeToTls runs in the TLS server role (upgradeToTls reads
@@ -846,7 +842,7 @@ final private[kyo] class JsTransport private (
         conn: NetConnection,
         tls: kyo.net.NetTlsConfig,
         channelCapacity: Int
-    )(using AllowUnsafe, Frame): Fiber.Unsafe[NetConnection, Abort[NetException]] =
+    )(using allow: AllowUnsafe, frame: Frame): Fiber.Unsafe[NetConnection, Abort[NetException]] =
         // The SNI host the upgrade engine verifies against; also the host reported by any handshake failure (an upgrade has no fresh port, so -1).
         val upgradeHost = tls.sniHostname.getOrElse("")
         // Honor a NetTlsConfig.tlsProvider pin: JS upgrades via Node's tls module, so a pin to any non-"node" provider fails closed.
@@ -1067,7 +1063,7 @@ final private[kyo] class JsTransport private (
                     discard(tlsSocket.destroy())
             }
             promise.onComplete { _ =>
-                deadline.interruptDiscard(Result.Panic(Closed("JsTransport", summon[Frame], "upgrade settled before deadline")))
+                deadline.interruptDiscard(Result.Panic(Interrupted(frame, "upgrade settled before deadline")))
             }
         end if
 
@@ -1077,7 +1073,7 @@ final private[kyo] class JsTransport private (
                 // Pause the TLS socket now that the handshake is done: kyo controls data flow.
                 // (We cannot pause before the handshake as that blocks TLS record delivery.)
                 discard(tlsSocket.pause())
-                val newHandle = JsHandle.init(tlsSocket, driver)
+                val newHandle = JsHandle.init(tlsSocket, driver, frame)
                 newHandle.peerCloseGrace = handle.peerCloseGrace // the upgraded connection inherits the original connection's reclaim grace
                 val newConn = Connection.init(newHandle, driver, channelCapacity, handle.peerCloseGrace)
                 // Preserve the upgrade role on the new TLS connection so a further upgrade does not silently flip client/server.
@@ -1147,7 +1143,8 @@ end JsTransport
   */
 final private[net] class JsListener(
     private val server: js.Dynamic,
-    private var _address: NetAddress
+    private var _address: NetAddress,
+    val createdAt: Frame
 ) extends NetListener:
 
     /** Extra teardown the transport attaches, currently reclaiming the accepted sockets whose TLS handshake never settled. Written once at

kyo-net/js-wasm/src/main/scala/kyo/net/internal/JsTransport.scala

@@ -57,7 +57,7 @@ class JsIoDriverTest extends kyo.net.Test:
         given Frame = Frame.internal
         val driver  = JsIoDriver.init()
         openPair().map { case (serverSock, clientSock) =>
-            val handle = JsHandle.init(serverSock, driver)
+            val handle = JsHandle.init(serverSock, driver, Frame.internal)
             // Backpressured: no awaitRead is armed. The peer writes 3 bytes then half-closes (FIN via end()).
             discard(clientSock.write(buffer(Array[Byte](10, 20, 30))))
             discard(clientSock.end())
@@ -77,7 +77,7 @@ class JsIoDriverTest extends kyo.net.Test:
         given Frame = Frame.internal
         val driver  = JsIoDriver.init()
         openPair().map { case (serverSock, clientSock) =>
-            val handle = JsHandle.init(serverSock, driver)
+            val handle = JsHandle.init(serverSock, driver, Frame.internal)
             // The peer stays connected and sends nothing: resuming yields no data and no 'end', so readableEnded/destroyed stay false.
             assert(!driver.isPeerClosed(handle), "first isPeerClosed returns false")
             Async.sleep(300.millis).map { _ =>