fix: harden accessory timeout ceilings and count() listener cleanup

Audit follow-ups on the idle-window normalization: - maxTotalMs/maxPageMs defaults are timeoutMs * 10, which silently overflows to a negative Long for an effectively-infinite idle window (Long.MAX_VALUE * 10 wraps to -10). withTimeoutOrNull then expired immediately, inverting 'wait forever' into 'never wait'. All three ceilings (fetchAllPages, fetchFirst, fetchAllWithHooks) now treat a non-positive ceiling as uncapped, matching the idle window's <= 0 convention. Regression-tested via fetchFirst. - count(filters) leaked its RelayConnectionListener and left COUNT subs open if the caller cancelled mid-wait or a listener threw: the cleanup ran as straight-line code with no try/finally, unlike every sibling accessory. Wrapped so unsubscribe + removeConnectionListener + channel close always run. - New FetchFirstIdleTimeoutTest pins fetchFirst's idle-window semantics (signals restart the window, silence costs exactly one window, the ceiling stops endless terminal chatter, overflow means uncapped). - README: note publishAndConfirm's fixed window as the deliberate exception, and correct the fetchAllPagesFromPool row, which claimed cross-relay dedup the function explicitly does not do. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KZe1Pq2ejoHehdufhPDRgb
2026-08-08 23:54:39 +00:00 · 2026-08-03 02:48:15 +00:00
parent 431ad153bc
commit 53be8d1755
7 changed files with 198 additions and 36 deletions
@@ -98,7 +98,8 @@ suspend fun INostrClient.count(
 * results keep trickling in is never cut short; the wait only gives up after a
 * full window with no relay answering. [maxTotalMs] (default 10x the idle
 * window) is the wall-clock ceiling against a misbehaving relay re-sending
- * results forever.
+ * results forever; a non-positive value means uncapped (which also absorbs a
+ * `timeoutMs * 10` overflow from an effectively-infinite idle window).
 *
 * @param filters Map of relay -> filter to count.
 * @param timeoutMs Idle window between responses (default 15 s).
@@ -128,29 +129,32 @@ suspend fun INostrClient.count(
            }
        }

-    addConnectionListener(listener)
-
-    filters.forEach { (relay, filterList) ->
-        val subId = newSubId()
-        subIdToRelay[subId] = relay
-        count(subId = subId, filters = mapOf(relay to filterList))
-    }
-
    val results = mutableMapOf<NormalizedRelayUrl, CountResult>()

-    // Each receive is bounded by the idle window alone; every arriving result
-    // restarts it on the next loop iteration. The outer ceiling bounds the whole
-    // wait against a relay that keeps re-sending results.
-    withTimeoutOrNull(maxTotalMs) {
-        while (results.size < filters.size) {
-            val next = withTimeoutOrNull(timeoutMs) { resultChannel.receive() } ?: break
-            results[next.first] = next.second
-        }
-    }
+    try {
+        addConnectionListener(listener)

-    subIdToRelay.keys.forEach { unsubscribe(it) }
-    removeConnectionListener(listener)
-    resultChannel.close()
+        filters.forEach { (relay, filterList) ->
+            val subId = newSubId()
+            subIdToRelay[subId] = relay
+            count(subId = subId, filters = mapOf(relay to filterList))
+        }
+
+        // Each receive is bounded by the idle window alone; every arriving result
+        // restarts it on the next loop iteration. The outer ceiling bounds the whole
+        // wait against a relay that keeps re-sending results.
+        val ceiling = if (maxTotalMs <= 0) Long.MAX_VALUE else maxTotalMs
+        withTimeoutOrNull(ceiling) {
+            while (results.size < filters.size) {
+                val next = withTimeoutOrNull(timeoutMs) { resultChannel.receive() } ?: break
+                results[next.first] = next.second
+            }
+        }
+    } finally {
+        subIdToRelay.keys.forEach { unsubscribe(it) }
+        removeConnectionListener(listener)
+        resultChannel.close()
+    }

    return results
 }
@@ -241,12 +241,9 @@ suspend fun INostrClient.fetchAllPages(
            // ceiling means uncapped, mirroring the idle window's `<= 0` = disabled
            // convention (it also absorbs a `timeoutMs * 10` overflow from a caller
            // passing an effectively-infinite idle window).
-            if (maxPageMs <= 0 || maxPageMs == Long.MAX_VALUE) {
+            val ceiling = if (maxPageMs <= 0) Long.MAX_VALUE else maxPageMs
+            withTimeoutOrNull(ceiling) {
                doneChannel.receiveWithinIdle(clock, timeoutMs)
-            } else {
-                withTimeoutOrNull(maxPageMs) {
-                    doneChannel.receiveWithinIdle(clock, timeoutMs)
-                }
            }

            unsubscribe(subId)
@@ -86,7 +86,9 @@ suspend fun INostrClient.fetchAllWithHooks(
     * adversarial or misbehaving relay could pin the caller forever. The cap
     * restores an upper bound while staying far above the idle window, so a
     * legitimately streaming relay still finishes its backlog. Pass
-     * [Long.MAX_VALUE] for a deliberately uncapped drain.
+     * [Long.MAX_VALUE] for a deliberately uncapped drain; a non-positive value
+     * also uncaps (absorbing a `timeoutMs * 10` overflow from an
+     * effectively-infinite idle window).
     */
    maxTotalMs: Long = timeoutMs * 10,
    onEvent: suspend (relay: NormalizedRelayUrl, event: Event) -> Boolean,
@@ -144,7 +146,7 @@ suspend fun INostrClient.fetchAllWithHooks(
        coroutineScope {
            subscribe(subscriptionId, filters, listener)
            val watchdog =
-                if (maxTotalMs == Long.MAX_VALUE) {
+                if (maxTotalMs <= 0 || maxTotalMs == Long.MAX_VALUE) {
                    null
                } else {
                    launch {
@@ -74,7 +74,9 @@ suspend fun INostrClient.fetchFirst(
 * signal (a terminal state from one relay of many) restarts it, so the fetch only
 * gives up after a full window of total silence. [maxTotalMs] (default 10x the
 * idle window) is the wall-clock ceiling that bounds a relay emitting endless
- * terminal chatter (e.g. a CLOSED/reconnect loop) without ever delivering an event.
+ * terminal chatter (e.g. a CLOSED/reconnect loop) without ever delivering an
+ * event; a non-positive value means uncapped (which also absorbs a
+ * `timeoutMs * 10` overflow from an effectively-infinite idle window).
 */
 suspend fun INostrClient.fetchFirst(
    subscriptionId: String = newSubId(),
@@ -128,7 +130,8 @@ suspend fun INostrClient.fetchFirst(
        // Each wait is bounded by the idle window alone; any arriving signal
        // restarts it on the next loop iteration. The outer ceiling stays far
        // above the window so legitimate multi-relay stragglers still land.
-        withTimeoutOrNull(maxTotalMs) {
+        val ceiling = if (maxTotalMs <= 0) Long.MAX_VALUE else maxTotalMs
+        withTimeoutOrNull(ceiling) {
            while (remaining.isNotEmpty()) {
                val progressed =
                    withTimeoutOrNull(timeoutMs) {
@@ -1179,7 +1179,3 @@ internal val KEEP_ALIVE_ID = "f".repeat(64)
 */
 private const val DEFAULT_CONNECT_TIMEOUT_MS = 30_000L
 internal const val DEFAULT_DOWNLOAD_IDLE_MS = 60_000L
-
-// IdleClock and receiveWithinIdle — the idle-watchdog primitives this file
-// introduced — now live in IdleWatchdog.kt, shared by every accessory whose
-// timeout is measured from the relay's most recent message.
@@ -22,7 +22,9 @@ The shared primitives are in `IdleWatchdog.kt` (`IdleClock` + `receiveWithinIdle
 use them when writing a new accessory. Because an idle window alone is unbounded
 against a relay that trickles messages forever, the loops that could run away also
 take a hard wall-clock ceiling (`maxTotalMs` / `maxPageMs`, default 10x the idle
-window) as a backstop.
+window) as a backstop; a non-positive ceiling means uncapped. The one deliberate
+exception is the write side: `publishAndConfirm`'s `timeoutInSeconds` is a fixed
+window to collect the `OK`s — a bounded confirmation round-trip, not a stream.

 ## One-shot reads (subscribe → collect → return)

@@ -31,7 +33,7 @@ window) as a backstop.
 | `fetchAll(relay, filter, timeoutMs)` | `NostrClientFetchAllExt` | Get every event matching a filter in one REQ, deduped by id, until EOSE or a full idle window of silence. **No verify, no store** — just the events. |
 | `fetchFirst(relay, filter, timeoutMs)` | `NostrClientFetchFirstExt` | Get the first matching event and stop (returns `null` on none/timeout). |
 | `fetchAllPages(relay, filters, timeoutMs)` | `NostrClientFetchAllPagesExt` | Fully retrieve a result set larger than the relay's per-REQ cap (strfry `limit`, ~500) by walking a `created_at` cursor. Bound it with the filter's `limit`. |
-| `fetchAllPagesFromPool(filters, ...)` | `NostrClientFetchAllPagesPoolExt` | Same paging, across several relays at once, deduped across them. |
+| `fetchAllPagesFromPool(filters, ...)` | `NostrClientFetchAllPagesPoolExt` | Same paging, across several relays at once. No cross-relay dedup — the `WithHooks` variant below dedups. |
 | `fetchAllWithHooks(filters, ...)` | `NostrClientFetchAllWithHooksExt` | `fetchAll` with a suspending per-`(relay, event)` accept hook (verify+store as events arrive), per-relay terminal-reason tracking, optional dead-relay collection (`deadOut` + `classifyDrainFailure`), keep-pending-on-`auth-required` CLOSED (NIP-42 re-fire), and a timeout diagnostic hook. |
 | `fetchAllPagesFromPoolWithHooks(filters, ...)` | `NostrClientFetchAllWithHooksExt` | `fetchAllPagesFromPool` with the same suspending accept hook, run single-threaded in one consumer; deduped across relays by `SeenIds` before the hook. |

@@ -0,0 +1,158 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.currentTime
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+
+/**
+ * Pins [fetchFirst]'s timeout to the package-wide idle-window convention:
+ * [timeoutMs] is silence measured from the most recent relay signal, not an
+ * absolute deadline across the whole multi-relay wait — with [maxTotalMs] as
+ * the wall-clock ceiling.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class FetchFirstIdleTimeoutTest {
+    /** Captures the subscription listener so the test can play the relays. */
+    private class ScriptedClient : INostrClient by EmptyNostrClient() {
+        var listener: SubscriptionListener? = null
+
+        override fun subscribe(
+            subId: String,
+            filters: Map<NormalizedRelayUrl, List<Filter>>,
+            listener: SubscriptionListener?,
+        ) {
+            this.listener = listener
+        }
+    }
+
+    private val relayA = RelayUrlNormalizer.normalize("wss://a.example.com")
+    private val relayB = RelayUrlNormalizer.normalize("wss://b.example.com")
+
+    private fun event(i: Int) =
+        Event(
+            id = i.toString(16).padStart(64, '0'),
+            pubKey = "f".repeat(64),
+            createdAt = i.toLong(),
+            kind = 1,
+            tags = emptyArray(),
+            content = "e$i",
+            sig = "0".repeat(128),
+        )
+
+    private fun filters(vararg relays: NormalizedRelayUrl) = relays.associateWith { listOf(Filter(kinds = listOf(1))) }
+
+    @Test
+    fun arrivingSignalsRestartTheIdleWindow() =
+        runTest {
+            val client = ScriptedClient()
+            launch {
+                // Terminal chatter every 250ms keeps the 300ms window alive long
+                // enough for the slow relay's event at 900ms — an absolute
+                // deadline would have returned null at 300ms.
+                delay(250)
+                client.listener!!.onClosed("rate limited", relayA, null)
+                delay(250)
+                client.listener!!.onClosed("rate limited", relayA, null)
+                delay(250)
+                client.listener!!.onClosed("rate limited", relayA, null)
+                delay(150)
+                client.listener!!.onEvent(event(1), false, relayB, null)
+            }
+            val result =
+                client.fetchFirst(
+                    filters = filters(relayA, relayB),
+                    timeoutMs = 300,
+                )
+            assertEquals(event(1).id, result?.id, "signals must restart the window; the slow relay's event still lands")
+        }
+
+    @Test
+    fun totalSilenceReturnsNullAfterOneIdleWindow() =
+        runTest {
+            val client = ScriptedClient()
+            val start = currentTime
+            val result =
+                client.fetchFirst(
+                    filters = filters(relayA),
+                    timeoutMs = 300,
+                )
+            assertNull(result)
+            assertEquals(300L, currentTime - start, "a silent relay costs exactly one idle window")
+        }
+
+    @Test
+    fun wallClockCeilingStopsEndlessTerminalChatter() =
+        runTest {
+            val client = ScriptedClient()
+            val chatter =
+                launch {
+                    // relayA re-CLOSEs forever (a reconnect loop); relayB never
+                    // answers. Every signal restarts the window, so only the
+                    // ceiling can end the wait.
+                    while (true) {
+                        delay(200)
+                        client.listener!!.onClosed("auth-required: again", relayA, null)
+                    }
+                }
+            val start = currentTime
+            val result =
+                client.fetchFirst(
+                    filters = filters(relayA, relayB),
+                    timeoutMs = 300,
+                    maxTotalMs = 1_000,
+                )
+            chatter.cancel()
+            assertNull(result)
+            assertEquals(1_000L, currentTime - start, "the ceiling must end an endlessly-restarted wait")
+        }
+
+    @Test
+    fun effectivelyInfiniteIdleWindowDoesNotOverflowTheCeiling() =
+        runTest {
+            val client = ScriptedClient()
+            launch {
+                delay(100)
+                client.listener!!.onEvent(event(1), false, relayA, null)
+            }
+            // Long.MAX_VALUE * 10 wraps to -10; the default ceiling must
+            // degrade to "uncapped", not to an instantly-expired wait.
+            val result =
+                client.fetchFirst(
+                    filters = filters(relayA),
+                    timeoutMs = Long.MAX_VALUE,
+                )
+            assertEquals(event(1).id, result?.id, "an overflowed default ceiling must mean uncapped, not instant timeout")
+        }
+}