From 53be8d1755ff6784491d70b1efcfc1cfdd6ec5ee Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 02:48:15 +0000 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01KZe1Pq2ejoHehdufhPDRgb --- .../client/accessories/NostrClientCountExt.kt | 46 ++--- .../NostrClientFetchAllPagesExt.kt | 7 +- .../NostrClientFetchAllWithHooksExt.kt | 6 +- .../accessories/NostrClientFetchFirstExt.kt | 7 +- .../NostrClientNegentropySyncExt.kt | 4 - .../relay/client/accessories/README.md | 6 +- .../accessories/FetchFirstIdleTimeoutTest.kt | 158 ++++++++++++++++++ 7 files changed, 198 insertions(+), 36 deletions(-) create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/FetchFirstIdleTimeoutTest.kt diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientCountExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientCountExt.kt index 4fced89a6e..e9125e602b 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientCountExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientCountExt.kt @@ -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() - // 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 } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt index 2a940c342e..a1886854b5 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt @@ -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) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllWithHooksExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllWithHooksExt.kt index 9473d24daf..bc4bff4d44 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllWithHooksExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllWithHooksExt.kt @@ -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 { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchFirstExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchFirstExt.kt index 75a03704c2..fb25ae4672 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchFirstExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchFirstExt.kt @@ -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) { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt index 19afb00934..89360facbf 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt @@ -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. diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/README.md b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/README.md index 5d2034bc88..246f83fe2d 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/README.md +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/README.md @@ -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. | diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/FetchFirstIdleTimeoutTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/FetchFirstIdleTimeoutTest.kt new file mode 100644 index 0000000000..d918396017 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/FetchFirstIdleTimeoutTest.kt @@ -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>, + 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") + } +}