[kyo-net] ConnectionPool: linearize release against close so a raced release cannot orphan a connection

release() and close() were not linearizable on the per-host ring. A connection
returned by release at the same instant as close could land in a ring close had
already drained, or in a fresh HostPool getPool re-created after close cleared
the map. Nothing drains that ring again, so the connection never reaches the
discard callback and its socket stays open with its read armed. On the shared
io_uring transport this surfaced as a flaky end-of-run fd leak (pendingCloses=0)
in the kyo-sql container suites.

release now re-reads closed after publishing and drains-and-discards its HostPool
when the pool closed under it. The HostPool drain, shared by close and the new
drainDiscard, consumes every claimed slot up to tail and spins over a slot that
is mid-publish rather than stopping at it, so a release that claimed a slot but
has not yet stored its sequence is waited for. The ring head CAS keeps disposal
exactly-once across both drainers.

ConnectionPoolTest reproduces the orphan deterministically through a default
no-op test seam and asserts the connection is disposed exactly once, plus a
concurrency smoke that races n releases against one close.

Author: fwbrasil

kyo-net/shared/src/main/scala/kyo/net/internal/ConnectionPool.scala

@@ -33,6 +33,11 @@ final private[kyo] class ConnectionPool[K, C](
     /** True once `close()` has run. For testing the client's close/release path only. */
     private[kyo] def isClosed(using AllowUnsafe): Boolean = closed
 
+    // Test seam (default no-op): a deterministic interleaving point for the release-vs-close linearizability regression in
+    // ConnectionPoolTest. It runs after release() has observed the pool open but before it publishes, so a test can drive
+    // close() into exactly the window the shared-transport fd leak lives in. Never set outside that test.
+    private[internal] var raceProbe: () => Unit = ConnectionPool.noRaceProbe
+
     /** Try to get a live idle connection for the given host. */
     def poll(key: K)(using AllowUnsafe): Maybe[C] =
         if closed then Maybe.empty
@@ -41,7 +46,21 @@ final private[kyo] class ConnectionPool[K, C](
     /** Return a connection to the idle pool. If the ring is full, discard it. */
     def release(key: K, conn: C)(using AllowUnsafe): Unit =
         if closed then discardConn(conn)
-        else getPool(key).release(conn, discardConn)
+        else
+            raceProbe()
+            val hostPool = getPool(key)
+            hostPool.release(conn, discardConn)
+            // close() can race this release: it sets `closed`, drains every host pool, and clears the map, any of which may
+            // fall between the `closed` read above and the publish just done. A connection published into a ring close()
+            // already drained (or a fresh pool getPool re-created after pools.clear()) would otherwise never be drained again
+            // and its socket never closed. Re-read `closed`. If it is now set, drain and discard this host pool ourselves. The
+            // ring's head CAS makes disposal exactly-once against close()'s own drain.
+            if closed then
+                hostPool.drainDiscard(discardConn)
+                // Drop the entry we may have re-created after close()'s pools.clear() so it does not linger. The two-arg remove
+                // unmaps only this exact instance, so a fresh pool another releaser inserted for the same key is left alone.
+                kyo.discard(pools.remove(key, hostPool))
+            end if
 
     /** Discard a connection without returning it to the pool. */
     def discard(conn: C)(using AllowUnsafe): Unit =
@@ -81,6 +100,9 @@ end ConnectionPool
 
 private[kyo] object ConnectionPool:
 
+    // The shared default for `raceProbe`: a single no-op instance so a production pool allocates no per-instance lambda.
+    private[internal] val noRaceProbe: () => Unit = () => ()
+
     def init[K, C](
         maxConnectionsPerHost: Int,
         idleConnectionTimeout: Duration,
@@ -178,28 +200,51 @@ private[kyo] object ConnectionPool:
         def unreserve(): Unit =
             kyo.discard(inFlight.decrementAndGet())
 
-        /** Close the pool. Drains idle connections for the caller to close. */
-        def close[C](into: ChunkBuilder[C]): Unit =
+        /** Drain every slot claimed for release, from `head` up to `tail`, applying `sink` to each connection.
+          *
+          * A concurrent `release` publishes in two steps: it CASes `tail` to claim a slot, then stores the connection and
+          * `lazySet`s the slot's sequence to mark it readable. A drain that stopped at the first slot whose sequence is not yet
+          * visible would treat a slot being published right now as the end of the ring and leave that connection behind, never
+          * drained again. So while `head` is below `tail` a stale sequence means a claim is mid-publish: spin until its store
+          * lands rather than terminate. A claimer between its CAS and its store is running on its own carrier (release never
+          * suspends), so the wait is bounded. A single-threaded runtime has no release in flight while this runs, so the spin
+          * is never taken. Ends when `head == tail`.
+          */
+        private def drainClaimed[C](sink: C => Unit): Unit =
             @tailrec def loop(): Unit =
                 val currentHead = head.get()
-                val idx         = (currentHead % capacity).toInt
-                val seq         = sequences.get(idx)
-                if seq < currentHead + 1 then ()
-                else if head.compareAndSet(currentHead, currentHead + 1) then
-                    connections(idx) match
-                        case Present(conn) =>
-                            connections(idx) = Absent
-                            kyo.discard(into += conn.asInstanceOf[C])
-                        case Absent =>
-                            connections(idx) = Absent
-                    end match
-                    sequences.lazySet(idx, currentHead + capacity)
-                    loop()
-                else loop()
+                val currentTail = tail.get()
+                if currentHead >= currentTail then ()
+                else
+                    val idx = (currentHead % capacity).toInt
+                    val seq = sequences.get(idx)
+                    if seq < currentHead + 1 then loop()
+                    else if head.compareAndSet(currentHead, currentHead + 1) then
+                        connections(idx) match
+                            case Present(conn) =>
+                                connections(idx) = Absent
+                                sink(conn.asInstanceOf[C])
+                            case Absent =>
+                                connections(idx) = Absent
+                        end match
+                        sequences.lazySet(idx, currentHead + capacity)
+                        loop()
+                    else loop()
+                    end if
                 end if
             end loop
             loop()
-        end close
+        end drainClaimed
+
+        /** Close the pool. Drains idle connections for the caller to close. */
+        def close[C](into: ChunkBuilder[C]): Unit =
+            drainClaimed[C](conn => kyo.discard(into += conn))
+
+        /** Drain and discard every connection still in the ring, for a `release` that observed the pool closed after it had
+          * already published. Shares [[drainClaimed]] with [[close]], so the same wait-for-a-mid-publish-claim rule applies.
+          */
+        def drainDiscard[C](discardConn: C => Unit): Unit =
+            drainClaimed(discardConn)
 
     end HostPool
 

kyo-net/shared/src/test/scala/kyo/net/internal/ConnectionPoolTest.scala

@@ -8,6 +8,7 @@ import kyo.net.Test
 class ConnectionPoolTest extends Test:
 
     import AllowUnsafe.embrace.danger
+    given Frame = Frame.internal
 
     val key1 = NetAddress.Tcp("host1", 80)
     val key2 = NetAddress.Tcp("host2", 80)
@@ -133,4 +134,62 @@ class ConnectionPoolTest extends Test:
         assert(discardCount.get() == 2)
     }
 
+    "release that observes close mid-publish disposes the connection, never orphans it (fd-leak race regression, CI #1837)" in {
+        // The shared-transport fd leak: release(key, conn) passes its `closed` check, then close() runs (drains every host
+        // pool, sets closed, clears the map). Release then re-creates a host pool via computeIfAbsent and publishes into a
+        // ring nothing else will ever drain, so the connection's socket is never closed (the CI dump: pendingCloses=0, recv
+        // still armed). The raceProbe seam fires close() in exactly that window, so the interleaving is deterministic on
+        // every platform. The connection must still be disposed exactly once.
+        val discardCount = AtomicInt.Unsafe.init(0)
+        val pool =
+            ConnectionPool.init[NetAddress, String](2, kyo.Duration.Infinity, _ => true, _ => discard(discardCount.incrementAndGet()))
+        pool.raceProbe = () => discard(pool.close())
+        pool.release(key1, "c")
+        assert(
+            discardCount.get() == 1,
+            s"the connection must be disposed exactly once (1); got ${discardCount.get()} (0 = orphaned/leaked)"
+        )
+    }
+
+    "close drains concurrently with releases without orphaning or double-disposing (concurrency smoke)" in {
+        // The linearizable-drain path under real preemption: `n` concurrent releases race one close on a ring sized to hold
+        // them, so close catches slots mid-publish and its drain must spin over them rather than stop. Every released
+        // connection ends up extracted by close or discarded, exactly once. This also guards the drain's spin against
+        // deadlock under contention. JS has no in-method preemption, so it passes by construction.
+        val n         = 32
+        val scenarios = 1000
+        val expected  = (0 until n).map("c" + _).toSet
+        Loop(0) { s =>
+            if s >= scenarios then Loop.done(assert(true))
+            else
+                val discarded = AtomicRef.Unsafe.init(Chunk.empty[String])
+                val pool =
+                    ConnectionPool.init[NetAddress, String](
+                        n,
+                        kyo.Duration.Infinity,
+                        _ => true,
+                        c => discard(discarded.updateAndGet(_ :+ c))
+                    )
+                for
+                    latch     <- Latch.init(1)
+                    releasers <- Fiber.init(Async.foreach(0 until n, n)(j => latch.await.map(_ => Sync.defer(pool.release(key1, "c" + j)))))
+                    closer    <- Fiber.init(latch.await.map(_ => Sync.defer(pool.close())))
+                    _         <- latch.release
+                    _         <- releasers.get
+                    extracted <- closer.get
+                yield
+                    // Exactly-once per connection: every released id lands in exactly one of the two sets, none orphaned, none doubled.
+                    val ext = extracted.toArray.toSet
+                    val dis = discarded.get().toArray.toSet
+                    if (ext ++ dis) == expected && ext.intersect(dis).isEmpty then Loop.continue(s + 1)
+                    else
+                        Loop.done(assert(
+                            (ext ++ dis) == expected && ext.intersect(dis).isEmpty,
+                            s"scenario $s: exactly-once violated. orphaned=${expected -- ext -- dis}, double-disposed=${ext.intersect(dis)}"
+                        ))
+                    end if
+                end for
+        }
+    }
+
 end ConnectionPoolTest