diff --git a/core/src/commonMain/kotlin/dev/kdriver/core/connection/DefaultConnection.kt b/core/src/commonMain/kotlin/dev/kdriver/core/connection/DefaultConnection.kt index 17ae0b066..cfa1bf68d 100644 --- a/core/src/commonMain/kotlin/dev/kdriver/core/connection/DefaultConnection.kt +++ b/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, 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). diff --git a/core/src/jvmTest/kotlin/dev/kdriver/core/connection/ReceiveLoopLivenessTest.kt b/core/src/jvmTest/kotlin/dev/kdriver/core/connection/ReceiveLoopLivenessTest.kt new file mode 100644 index 000000000..dd140d2de --- /dev/null +++ b/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(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 = channel.receiveAsFlow() + + override suspend fun close() { + isActive = false + channel.close() + } + } + + private class TestConnection( + scope: CoroutineScope, + private val transports: MutableList, + ) : 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() + // 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() + 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() + } +}