Reconnect when the receive loop died but the socket stayed open

core/src/commonMain/kotlin/dev/kdriver/core/connection/DefaultConnection.kt

@@ -101,9 +101,9 @@ open class DefaultConnection(
     // Epoch millis of the last frame the receive loop saw, so idleness can be *read* instead of
     // guessed. The previous check re-subscribed to `events` on every poll and only observed the
     // 100 ms window it had just opened, so a connection that had genuinely been quiet for a while
-    // still looked busy, and a busy one could look quiet. 0 means "nothing ever arrived", which is
-    // idle by definition. Null means nothing has ever arrived, which is idle too — a sentinel
-    // rather than an epoch value, so it stays correct whatever the clock starts at.
+    // still looked busy, and a busy one could look quiet. Null means nothing has ever arrived,
+    // which is idle too — a sentinel rather than an epoch value, so it stays correct whatever the
+    // clock starts at.
     @Volatile
     private var lastMessageAt: Long? = null
 
@@ -118,19 +118,31 @@ open class DefaultConnection(
     @InternalCdpApi
     override val generatedDomains: MutableMap<KClass<out Domain>, Domain> = mutableMapOf()
 
+    /**
+     * Whether [t] can still carry a command.
+     *
+     * An open socket is not enough: the receive loop has to be alive too, because it is what
+     * completes the waiters. The loop runs in a scope owned by the caller, so it can be cancelled
+     * while the socket stays open — leaving a transport that reports `isActive`, no disconnect ever
+     * observed, and every later command waiting out its timeout for a reply nobody will deliver.
+     */
+    private fun isUsable(t: WebSocketTransport): Boolean =
+        t.isActive && !disconnected && socketSubscription?.isActive == true
+
     private suspend fun connect(): WebSocketTransport {
         if (closed) throw ConnectionClosedException("Connection closed")
-        transport?.let { if (it.isActive && !disconnected) return it }
+        transport?.let { if (isUsable(it)) return it }
         // Guard so concurrent first commands don't each open a session (which would leak the
         // duplicate sockets/listeners). Double-checked: skip the lock once connected (ISSUE-4).
         return connectMutex.withLock {
             if (closed) throw ConnectionClosedException("Connection closed")
-            transport?.let { if (it.isActive && !disconnected) return@withLock it }
-            // Once the receive loop has reported a disconnect we don't trust the old transport's
-            // `isActive` (it can lag the real socket close), so dispose it and build a fresh one. This
-            // makes a command issued after a drop reconnect deterministically instead of reusing a
-            // dead socket and failing (or hanging until the command timeout).
-            if (disconnected) {
+            transport?.let { if (isUsable(it)) return@withLock it }
+            // Reached only when the current transport is unusable, so it is never a healthy socket
+            // being thrown away. Two ways to get here: the receive loop reported a disconnect — whose
+            // `isActive` can lag the real socket close, so it is not to be trusted — or the loop died
+            // without reporting anything. Either way the fix is the same: dispose and rebuild, so a
+            // command lands on a working socket instead of failing or waiting out its timeout.
+            if (transport != null) {
                 // Clear `transport` and cancel the old receive loop *before* closing the old transport,
                 // so the loop ending on that close sees `transport !== old` and no-ops instead of
                 // spuriously re-flagging `disconnected` (which would cascade into extra reconnects).

core/src/jvmTest/kotlin/dev/kdriver/core/connection/ReceiveLoopLivenessTest.kt

@@ -0,0 +1,115 @@
+package dev.kdriver.core.connection
+
+import dev.kdriver.cdp.CommandMode
+import dev.kdriver.cdp.Serialization
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.cancelChildren
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.StandardTestDispatcher
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withTimeoutOrNull
+import kotlinx.serialization.json.jsonObject
+import kotlinx.serialization.json.jsonPrimitive
+import kotlinx.serialization.json.long
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+
+/**
+ * A socket that is still open but whose receive loop has died is not a healthy connection: nobody
+ * delivers replies any more, so commands sent on it can only time out.
+ *
+ * The loop runs in a scope handed in from outside, so its lifetime is not the connection's to
+ * control — it can be cancelled without the socket ever closing. That leaves the transport
+ * reporting `isActive`, no disconnect ever observed, and every later command waiting on its own
+ * timeout for an answer that cannot arrive.
+ */
+class ReceiveLoopLivenessTest {
+
+    /** Answers every request immediately, so a live receive loop is the only thing in question. */
+    private class EchoTransport : WebSocketTransport {
+        private val channel = Channel<String>(Channel.RENDEZVOUS)
+        override var isActive: Boolean = false
+            private set
+
+        override suspend fun connect() {
+            isActive = true
+        }
+
+        override suspend fun send(message: String) {
+            val id = Serialization.json.parseToJsonElement(message).jsonObject["id"]!!.jsonPrimitive.long
+            channel.send("""{"id":$id,"result":{"value":1}}""")
+        }
+
+        override fun incoming(): Flow<String> = channel.receiveAsFlow()
+
+        override suspend fun close() {
+            isActive = false
+            channel.close()
+        }
+    }
+
+    private class TestConnection(
+        scope: CoroutineScope,
+        private val transports: MutableList<EchoTransport>,
+    ) : DefaultConnection("ws://stub/devtools/page/stub", scope) {
+        override fun createTransport(): WebSocketTransport = EchoTransport().also { transports += it }
+        override suspend fun updateTarget() = Unit
+    }
+
+    @Test
+    fun command_afterTheReceiveLoopWasCancelled_reconnectsInsteadOfTimingOut() =
+        runTest(StandardTestDispatcher()) {
+            val transports = mutableListOf<EchoTransport>()
+            // A scope of its own, so it can be cancelled the way an owner's scope would be.
+            val listeningScope = CoroutineScope(coroutineContext + Job())
+            val connection = TestConnection(listeningScope, transports)
+
+            // First command opens the socket and starts the receive loop.
+            val first = withTimeoutOrNull(5_000) {
+                connection.callCommand("Some.method", null, CommandMode.ONE_SHOT)
+            }
+            assertNotNull(first, "the connection should work before its receive loop is cancelled")
+
+            // The loop dies, the socket does not: this is the state that used to go unnoticed.
+            listeningScope.coroutineContext[Job]!!.cancelChildren()
+            runCurrent()
+
+            val second = withTimeoutOrNull(5_000) {
+                connection.callCommand("Some.method", null, CommandMode.ONE_SHOT)
+            }
+
+            assertNotNull(second, "a socket nobody reads must be replaced, not reused")
+            assertEquals(2, transports.size, "the dead connection should have been re-established")
+
+            connection.close()
+            listeningScope.coroutineContext[Job]!!.cancel()
+        }
+
+    /** A healthy connection must not be torn down and rebuilt on every command. */
+    @Test
+    fun consecutiveCommands_reuseTheSameLiveConnection() = runTest(StandardTestDispatcher()) {
+        val transports = mutableListOf<EchoTransport>()
+        val listeningScope = CoroutineScope(coroutineContext + Job())
+        val connection = TestConnection(listeningScope, transports)
+
+        repeat(3) {
+            assertNotNull(
+                withTimeoutOrNull(5_000) {
+                    connection.callCommand("Some.method", null, CommandMode.ONE_SHOT)
+                },
+                "command $it should have been answered",
+            )
+        }
+
+        assertEquals(1, transports.size, "a live connection must be reused, not rebuilt each time")
+
+        connection.close()
+        listeningScope.coroutineContext[Job]!!.cancel()
+    }
+}