Bound the idle wait so a busy page cannot hang select()

core/src/commonMain/kotlin/dev/kdriver/core/browser/Config.kt

@@ -18,6 +18,10 @@ class Config(
     val browserConnectionTimeout: Long = Defaults.BROWSER_CONNECTION_TIMEOUT,
     val browserConnectionMaxTries: Int = Defaults.BROWSER_CONNECTION_MAX_TRIES,
     val commandTimeout: Long = Defaults.COMMAND_TIMEOUT,
+    /**
+     * How long, in milliseconds, the connection must receive nothing before it counts as idle.
+     */
+    val timeBeforeConsideredIdle: Long = Defaults.TIME_BEFORE_CONSIDERED_IDLE,
     val autoDiscoverTargets: Boolean = Defaults.AUTO_DISCOVER_TARGETS,
     val debugStringLimit: Int = Defaults.DEBUG_STRING_LIMIT,
 ) {
@@ -134,6 +138,7 @@ class Config(
         browserConnectionTimeout = browserConnectionTimeout,
         browserConnectionMaxTries = browserConnectionMaxTries,
         commandTimeout = commandTimeout,
+        timeBeforeConsideredIdle = timeBeforeConsideredIdle,
         autoDiscoverTargets = autoDiscoverTargets,
         debugStringLimit = debugStringLimit,
     ).also { copy ->
@@ -165,6 +170,21 @@ class Config(
          * the timeout (wait indefinitely).
          */
         const val COMMAND_TIMEOUT: Long = 30_000
+
+        /**
+         * How long, in milliseconds, the connection must receive nothing before it counts as idle.
+         */
+        const val TIME_BEFORE_CONSIDERED_IDLE: Long = 100
+
+        /**
+         * Default upper bound, in milliseconds, on how long
+         * [dev.kdriver.core.connection.Connection.wait] will watch for the connection to go idle.
+         *
+         * A page can stream events indefinitely (polling, server-sent events, ads, a refreshing
+         * interstitial), in which case idleness never arrives. That is not an error — waiting is
+         * best-effort — so past this bound `wait` simply stops watching and returns.
+         */
+        const val IDLE_WAIT_TIMEOUT: Long = 10_000
         const val AUTO_DISCOVER_TARGETS: Boolean = true
         const val DEBUG_STRING_LIMIT: Int = 128
     }

core/src/commonMain/kotlin/dev/kdriver/core/browser/ConfigBuilder.kt

@@ -30,6 +30,10 @@ class ConfigBuilder {
     var browserConnectionTimeout: Long = Defaults.BROWSER_CONNECTION_TIMEOUT
     var browserConnectionMaxTries: Int = Defaults.BROWSER_CONNECTION_MAX_TRIES
     var commandTimeout: Long = Defaults.COMMAND_TIMEOUT
+    /**
+     * How long, in milliseconds, the connection must receive nothing before it counts as idle.
+     */
+    var timeBeforeConsideredIdle: Long = Defaults.TIME_BEFORE_CONSIDERED_IDLE
     var autoDiscoverTargets: Boolean = Defaults.AUTO_DISCOVER_TARGETS
     var debugStringLimit: Int = Defaults.DEBUG_STRING_LIMIT
 
@@ -51,6 +55,7 @@ class ConfigBuilder {
             browserConnectionTimeout = browserConnectionTimeout,
             browserConnectionMaxTries = browserConnectionMaxTries,
             commandTimeout = commandTimeout,
+            timeBeforeConsideredIdle = timeBeforeConsideredIdle,
             autoDiscoverTargets = autoDiscoverTargets,
             debugStringLimit = debugStringLimit
         )

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

@@ -4,6 +4,7 @@ import dev.kdriver.cdp.CDP
 import dev.kdriver.cdp.CommandMode
 import dev.kdriver.cdp.InternalCdpApi
 import dev.kdriver.core.browser.BrowserTarget
+import dev.kdriver.core.browser.Config
 import kotlinx.serialization.json.JsonElement
 
 /**
@@ -56,12 +57,18 @@ interface Connection : BrowserTarget, CDP {
     suspend fun updateTarget()
 
     /**
-     * Waits until the event listener reports idle (no new events received in a certain timespan).
-     * When \`t\` is provided, ensures waiting for \`t\` milliseconds, no matter what.
+     * Waits for the connection to go idle (nothing received for a short while).
      *
-     * @param t Time in milliseconds to wait, or null to wait until idle.
+     * Waiting is best-effort: a page can stream events indefinitely — polling, server-sent events,
+     * ads, a refreshing interstitial — in which case idleness never arrives. That is not an error,
+     * so after [idleTimeout] this simply stops watching and returns rather than waiting forever.
+     *
+     * @param t Minimum time in milliseconds to wait, even if the connection settles sooner. This is
+     *   a floor, not a deadline: it never cuts the wait short. Null means no minimum.
+     * @param idleTimeout Upper bound in milliseconds on watching for idleness. It caps the watching
+     *   only — a larger [t] is still honoured.
      */
-    suspend fun wait(t: Long? = null)
+    suspend fun wait(t: Long? = null, idleTimeout: Long = Config.Defaults.IDLE_WAIT_TIMEOUT)
 
     /**
      * Suspends the coroutine for a specified time in milliseconds.

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

@@ -36,6 +36,10 @@ open class DefaultConnection(
         // One initial send plus one resend after a transparent reconnect. The resend is only ever reached
         // when a send fails (bytes never left the client), so it cannot double-execute a command.
         private const val SEND_ATTEMPTS = 2
+
+        // How often idleness is re-read while waiting. Only bounds how quickly `wait` notices the
+        // connection went quiet; it is not itself a wait.
+        private const val IDLE_POLL_INTERVAL_MS = 50L
     }
 
     private val logger = KtorSimpleLogger("Connection")
@@ -74,6 +78,15 @@ open class DefaultConnection(
      */
     protected open fun createTransport(): WebSocketTransport = KtorWebSocketTransport(websocketUrl)
 
+    /**
+     * Current time in epoch millis, used to measure how long the connection has been quiet.
+     *
+     * Overridable for the same reason as [createTransport]: it lets a test drive idleness off the
+     * scheduler's virtual clock, so "traffic kept arriving" and "nothing arrived" are decided
+     * deterministically instead of by how fast the machine happens to run.
+     */
+    protected open fun currentTimeMillis(): Long = Clock.System.now().toEpochMilliseconds()
+
     private var prepareHeadlessDone = false
     private var prepareExpertDone = false
 
@@ -85,6 +98,15 @@ open class DefaultConnection(
     @Volatile
     private var needsRestore = false
 
+    // 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.
+    @Volatile
+    private var lastMessageAt: Long? = null
+
     private val allMessages = MutableSharedFlow<Message>(extraBufferCapacity = Channel.UNLIMITED)
 
     @InternalCdpApi
@@ -139,6 +161,7 @@ open class DefaultConnection(
         socketSubscription = messageListeningScope.launch {
             try {
                 t.incoming().collect { text ->
+                    lastMessageAt = currentTimeMillis()
                     try {
                         logger.debug("WS < CDP: ${text.take(owner?.config?.debugStringLimit ?: Defaults.DEBUG_STRING_LIMIT)}")
                         val received = Serialization.json.decodeFromString<Message>(text)
@@ -334,33 +357,31 @@ open class DefaultConnection(
         this.targetInfo = targetInfo.targetInfo
     }
 
-    override suspend fun wait(t: Long?) {
+    override suspend fun wait(t: Long?, idleTimeout: Long) {
         updateTarget()
-        val idleEvent: suspend () -> Boolean = {
-            withTimeoutOrNull(100.milliseconds) { events.first() } == null
+
+        val start = currentTimeMillis()
+
+        // Watch for the connection to settle, but never longer than `idleTimeout`: a page can keep
+        // streaming forever, and that is not a failure to report — it just means we stop waiting.
+        withTimeoutOrNull(idleTimeout.milliseconds) {
+            while (!isIdle()) delay(IDLE_POLL_INTERVAL_MS.milliseconds)
         }
 
+        // `t` is a floor, not a deadline: callers that pass it want the page left alone for at
+        // least that long, whether or not it settled sooner.
         if (t != null) {
-            val start = Clock.System.now().toEpochMilliseconds()
-            withTimeoutOrNull(t.milliseconds) {
-                // Wait for idle event or timeout
-                while (true) {
-                    if (idleEvent()) break
-                    delay(50.milliseconds)
-                }
-            }
-            // Ensure total wait time is at least t milliseconds
-            val elapsed = Clock.System.now().toEpochMilliseconds() - start
+            val elapsed = currentTimeMillis() - start
             if (elapsed < t) delay((t - elapsed).milliseconds)
-        } else {
-            // Wait indefinitely for idle event
-            while (true) {
-                if (idleEvent()) break
-                delay(50.milliseconds)
-            }
         }
     }
 
+    private fun isIdle(): Boolean {
+        val last = lastMessageAt ?: return true
+        val quietFor = currentTimeMillis() - last
+        return quietFor >= (owner?.config?.timeBeforeConsideredIdle ?: Defaults.TIME_BEFORE_CONSIDERED_IDLE)
+    }
+
     override suspend fun sleep(t: Long) {
         updateTarget()
         delay(t.milliseconds)

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

@@ -0,0 +1,148 @@
+package dev.kdriver.core.connection
+
+import dev.kdriver.cdp.CommandMode
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.StandardTestDispatcher
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.test.Test
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+/**
+ * Covers [DefaultConnection.wait]'s idle detection.
+ *
+ * `wait()` does not wait for a *result*: it waits for the page to settle. A page that never settles
+ * is therefore not an error — the caller should carry on. Two properties matter:
+ *
+ * 1. it must always hand control back, even on a page that never goes quiet;
+ * 2. "quiet" must be a fact about the connection, not about whichever 100 ms window a poll happened
+ *    to open.
+ */
+class ConnectionIdleWaitTest {
+
+    private class FakeTransport : 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) = Unit
+        override fun incoming(): Flow<String> = channel.receiveAsFlow()
+        suspend fun deliver(frame: String) = channel.send(frame)
+
+        override suspend fun close() {
+            isActive = false
+            channel.close()
+        }
+    }
+
+    /**
+     * `updateTarget()` is stubbed out: it issues a CDP command, which is not what these tests are
+     * about, and a fake transport that answers nothing would hang there rather than in the idle loop.
+     */
+    private class TestConnection(
+        scope: TestScope,
+        private val transport: FakeTransport,
+    ) : DefaultConnection("ws://stub/devtools/page/stub", scope) {
+        private val scheduler = scope.testScheduler
+        override fun createTransport(): WebSocketTransport = transport
+        override suspend fun updateTarget() = Unit
+
+        /** Idleness is measured on the scheduler's clock, so these tests are not timing-dependent. */
+        override fun currentTimeMillis(): Long = scheduler.currentTime
+
+        /** Connecting is lazy, on the first command. The reply never comes and is not needed here. */
+        suspend fun open() {
+            withTimeoutOrNull(1) { callCommand("Stub.noop", null, CommandMode.ONE_SHOT) }
+        }
+    }
+
+    /**
+     * A page that never goes quiet must not trap the caller forever.
+     *
+     * The stream here is deliberately busier than the idle threshold, so the connection is never
+     * idle by any definition. `wait()` must still return, bounded by its idle timeout.
+     */
+    @Test
+    fun wait_returnsOnAPageThatNeverGoesQuiet() = runTest(StandardTestDispatcher()) {
+        val transport = FakeTransport()
+        val connection = TestConnection(this, transport)
+        connection.open()
+
+        val noise = launch {
+            while (true) {
+                transport.deliver("""{"method":"Network.dataReceived","params":{}}""")
+                delay(50)
+            }
+        }
+        // Let the first frame land, so the connection has actually seen traffic before we wait.
+        runCurrent()
+
+        try {
+            val before = testScheduler.currentTime
+            val returned = withTimeoutOrNull(60_000) { connection.wait(idleTimeout = 10_000) }
+            val spent = testScheduler.currentTime - before
+
+            assertNotNull(returned, "wait() must give control back on a page that never settles")
+            // Right at the bound: any sooner would mean the traffic went unnoticed and the
+            // connection was mistaken for idle.
+            assertTrue(
+                spent >= 10_000,
+                "traffic kept arriving, so wait() should have watched for the full bound, not ${spent}ms",
+            )
+        } finally {
+            noise.cancelAndJoin()
+            connection.close()
+        }
+    }
+
+    /**
+     * On a silent connection idleness is already a fact — nothing has arrived. `wait()` should say
+     * so at once instead of spending a polling window rediscovering it.
+     */
+    @Test
+    fun wait_onASilentConnection_doesNotSpendAPollingWindow() = runTest(StandardTestDispatcher()) {
+        val transport = FakeTransport()
+        val connection = TestConnection(this, transport)
+        connection.open()
+
+        val before = testScheduler.currentTime
+        connection.wait()
+        val spent = testScheduler.currentTime - before
+
+        assertTrue(
+            spent < 100,
+            "a silent connection is idle by definition; wait() spent ${spent}ms rediscovering it",
+        )
+        connection.close()
+    }
+
+    /**
+     * `t` is a floor, not a deadline. A connection that settles at once must still be left alone for
+     * the requested time — otherwise callers asking for breathing room silently stop getting it.
+     */
+    @Test
+    fun wait_withAMinimum_honoursItEvenWhenAlreadyIdle() = runTest(StandardTestDispatcher()) {
+        val transport = FakeTransport()
+        val connection = TestConnection(this, transport)
+        connection.open()
+
+        val before = testScheduler.currentTime
+        connection.wait(t = 500)
+        val spent = testScheduler.currentTime - before
+
+        assertTrue(spent >= 500, "wait(500) returned after only ${spent}ms")
+        connection.close()
+    }
+}

opentelemetry/src/commonMain/kotlin/dev/kdriver/opentelemetry/OpenTelemetryTab.kt

@@ -424,7 +424,7 @@ class OpenTelemetryTab(
 
     override suspend fun updateTarget() = tab.updateTarget()
 
-    override suspend fun wait(t: Long?) = tab.wait(t)
+    override suspend fun wait(t: Long?, idleTimeout: Long) = tab.wait(t, idleTimeout)
 
     override suspend fun sleep(t: Long) = tab.sleep(t)