From ba35a87a7c06fbde6db0bbd207a74fa034c02e3d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 03:26:58 +0000 Subject: [PATCH] refactor: drop the ceiling params; bound waits by progress instead Follow-up audit on the timeout normalization. Three findings. 1. The maxTotalMs I added to fetchFirst and multi-relay count was the wrong shape twice over. A hard wall-clock bound already composes at the call site -- withTimeoutOrNull(ms) { fetchFirst(...) } -- so putting it in the signature duplicates what the caller has for free. And it was papering over the real defect: repeat chatter from a relay already accounted for (a CLOSED/reconnect loop, a duplicate COUNT) was treated as activity and restarted the idle window, so a flapping relay could hold the call open indefinitely. Both now reset the window only on genuine progress -- an event, or the first terminal signal from a relay still being waited on -- which is the rule the negentropy watchdog already applies to NOTICE/CLOSED chatter, and which makes both calls self-bounding at one window per relay. Ceiling params removed; the overflow guard they needed goes with them. 2. fetchFirst could drop a match: an event landing after the last terminal signal but before unsubscribe was left unread in the channel and the fetch reported nothing found. Added the post-loop drain that fetchAllWithHooks already does. Covered by a test. 3. fetchAllPages published its per-page counters across threads without a barrier on the idle path. received/delivered/pageMinTs/idsAtPageMin are written on the relay reader thread and read by the driver once the wait ends; the EOSE path gets happens-before from the channel, the idle path had none, so the driver could read a stale pageMinTs (ending the walk early) or an unsafely published idsAtPageMin. The volatile IdleClock bump now runs in a finally, so it covers every event including the early-returning duplicate and orders after the counters. Also folded the single-relay count channel close into its finally, so a throw mid-wait cleans up like every sibling accessory. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KZe1Pq2ejoHehdufhPDRgb --- .../client/accessories/NostrClientCountExt.kt | 69 ++++++++------- .../NostrClientFetchAllPagesExt.kt | 75 +++++++++------- .../accessories/NostrClientFetchFirstExt.kt | 86 +++++++++++-------- .../relay/client/accessories/README.md | 46 ++++++---- .../accessories/FetchFirstIdleTimeoutTest.kt | 68 +++++++++++---- 5 files changed, 209 insertions(+), 135 deletions(-) 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 e9125e602b..c8b8e65233 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 @@ -68,23 +68,22 @@ suspend fun INostrClient.count( } } - addConnectionListener(listener) + return try { + addConnectionListener(listener) - val result = - try { - count(subId = subId, filters = mapOf(relay to listOf(filter))) + count(subId = subId, filters = mapOf(relay to listOf(filter))) - withTimeoutOrNull(timeoutMs) { - resultChannel.receive() - } - } finally { - unsubscribe(subId) - removeConnectionListener(listener) + withTimeoutOrNull(timeoutMs) { + resultChannel.receive() } - - resultChannel.close() - - return result + } finally { + // Every cleanup step belongs in the finally: closing the channel used to + // sit after it, so a throw (or cancellation) mid-wait skipped it while the + // sibling accessories all cleaned up fully. + unsubscribe(subId) + removeConnectionListener(listener) + resultChannel.close() + } } /** @@ -92,23 +91,22 @@ suspend fun INostrClient.count( * (one filter per relay) and suspends until all results arrive * or the timeout expires. * - * [timeoutMs] is an **idle window measured from the most recent message**, not a + * [timeoutMs] is an **idle window measured from the most recent progress**, not a * wall-clock deadline for the whole batch — the package-wide accessory - * convention: each arriving COUNT result restarts it, so a large fan-out where - * 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; a non-positive value means uncapped (which also absorbs a - * `timeoutMs * 10` overflow from an effectively-infinite idle window). + * convention: each *new* relay's COUNT result restarts it, so a large fan-out + * where results keep trickling in is never cut short. A relay re-sending a result + * it already gave is not progress and does not restart the window, which makes + * the call self-bounding (at most one window per relay). A caller wanting a hard + * wall-clock bound has `withTimeoutOrNull(ms) { count(...) }` — at the cost of + * discarding the partial map, which is why this returns whatever arrived instead. * * @param filters Map of relay -> filter to count. - * @param timeoutMs Idle window between responses (default 15 s). + * @param timeoutMs Idle window between new responses (default 15 s). * @return Map of relay -> [CountResult] for every relay that responded in time. */ suspend fun INostrClient.count( filters: Map>, timeoutMs: Long = 15_000, - maxTotalMs: Long = timeoutMs * 10, ): Map { if (filters.isEmpty()) return emptyMap() @@ -140,15 +138,22 @@ suspend fun INostrClient.count( 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 - } + // One idle window per new relay result. The inner loop absorbs repeats + // (a relay answering twice) inside the SAME window, so only genuinely + // new information pushes the deadline out — bounding the call at one + // window per relay without needing a wall-clock ceiling. + while (results.size < filters.size) { + val progressed = + withTimeoutOrNull(timeoutMs) { + while (true) { + val (relay, result) = resultChannel.receive() + // put() returns the previous value: null means this relay + // had not answered yet, i.e. real progress. + if (results.put(relay, result) == null) break + } + true + } + if (progressed == null) break } } finally { subIdToRelay.keys.forEach { unsubscribe(it) } 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 52aa48affd..8688815352 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 @@ -178,40 +178,53 @@ suspend fun INostrClient.fetchAllPages( relay: NormalizedRelayUrl, forFilters: List?, ) { - clock.bump() - received++ - // Drop a boundary-second event we already delivered on an - // earlier page (the inclusive re-fetch returns it again). - if (boundary != null && event.createdAt == boundary && event.id in seenAtBoundary) return + // The bump is in a finally so it runs for EVERY event — + // including the duplicate that returns early below, which is + // still a sign of life — and, being a volatile write, runs + // AFTER the counters below. That ordering matters: these + // counters are written on the relay's reader thread and read + // by the driver coroutine once the wait ends. The EOSE path + // gets its happens-before from the channel, but the idle + // path has no such edge, so without the release write the + // driver could read a stale `pageMinTs` (ending the walk + // early) or an unsafely published `idsAtPageMin`. + try { + received++ + // Drop a boundary-second event we already delivered on an + // earlier page (the inclusive re-fetch returns it again). + if (boundary != null && event.createdAt == boundary && event.id in seenAtBoundary) return - // Count this event against every active filter it satisfies - // (one event can match more than one). Only a non-search filter - // may advance the `until` cursor: a search hit — possibly old, - // relevance-ranked — must not drag the cursor back and make the - // next page skip events a co-resident normal filter still needs. - var atLeastOne = false - var advancesCursor = false - for ((index, filter) in activeFilters) { - if (matchCountPerFilter[index] < (filter.limit ?: Int.MAX_VALUE) && filter.match(event)) { - matchCountPerFilter[index]++ - atLeastOne = true - if (filter.search == null) advancesCursor = true - } - } - if (atLeastOne) { - onEvent(event) - delivered++ - // Track the oldest advancing second and the ids delivered - // in it — that becomes the next boundary and its dedup set. - if (advancesCursor) { - if (event.createdAt < pageMinTs) { - pageMinTs = event.createdAt - idsAtPageMin.clear() - idsAtPageMin.add(event.id) - } else if (event.createdAt == pageMinTs) { - idsAtPageMin.add(event.id) + // Count this event against every active filter it satisfies + // (one event can match more than one). Only a non-search filter + // may advance the `until` cursor: a search hit — possibly old, + // relevance-ranked — must not drag the cursor back and make the + // next page skip events a co-resident normal filter still needs. + var atLeastOne = false + var advancesCursor = false + for ((index, filter) in activeFilters) { + if (matchCountPerFilter[index] < (filter.limit ?: Int.MAX_VALUE) && filter.match(event)) { + matchCountPerFilter[index]++ + atLeastOne = true + if (filter.search == null) advancesCursor = true } } + if (atLeastOne) { + onEvent(event) + delivered++ + // Track the oldest advancing second and the ids delivered + // in it — that becomes the next boundary and its dedup set. + if (advancesCursor) { + if (event.createdAt < pageMinTs) { + pageMinTs = event.createdAt + idsAtPageMin.clear() + idsAtPageMin.add(event.id) + } else if (event.createdAt == pageMinTs) { + idsAtPageMin.add(event.id) + } + } + } + } finally { + clock.bump() } } 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 fb25ae4672..45e28e7072 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 @@ -69,20 +69,24 @@ suspend fun INostrClient.fetchFirst( * every relay reached a terminal state — EOSE, CLOSED, or cannot-connect — with * nothing matching, or the line went quiet). * - * [timeoutMs] is an **idle window measured from the most recent message**, not a - * wall-clock deadline — the package-wide accessory convention: every arriving - * 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; a non-positive value means uncapped (which also absorbs a - * `timeoutMs * 10` overflow from an effectively-infinite idle window). + * [timeoutMs] is an **idle window measured from the most recent progress**, not a + * wall-clock deadline — the package-wide accessory convention. Progress means a + * signal that actually advances the fetch: an event, or the first terminal state + * from a relay still being waited on. Repeat chatter from a relay already + * accounted for (a CLOSED/reconnect loop) is *not* progress and does not restart + * the window — the same rule the negentropy watchdog applies to NOTICE/CLOSED + * error chatter, and what keeps a flapping relay from holding this open forever. + * + * That makes the call self-bounding: at most one progress signal per relay, each + * granting a fresh window. There is deliberately no ceiling parameter — a caller + * who wants a hard wall-clock bound already has one in + * `withTimeoutOrNull(ms) { fetchFirst(...) }`, which costs nothing here since a + * timed-out fetch returns `null` either way. */ suspend fun INostrClient.fetchFirst( subscriptionId: String = newSubId(), filters: Map>, timeoutMs: Long = 30_000L, - maxTotalMs: Long = timeoutMs * 10, ): Event? { val eventChannel = Channel(UNLIMITED) val doneChannel = Channel(UNLIMITED) @@ -127,37 +131,49 @@ suspend fun INostrClient.fetchFirst( try { subscribe(subscriptionId, filters, listener) - // 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. - val ceiling = if (maxTotalMs <= 0) Long.MAX_VALUE else maxTotalMs - withTimeoutOrNull(ceiling) { - while (remaining.isNotEmpty()) { - val progressed = - withTimeoutOrNull(timeoutMs) { - select { - eventChannel.onReceive { event -> - result = event - remaining.clear() - } - doneChannel.onReceive { relay -> - // A relay sends its matching events before its EOSE, so an event may - // already be buffered when this completion fires. select() picks a ready - // clause at random, so without this drain we could treat the relay as done - // and exit while its event still sits unread in the channel. - val buffered = eventChannel.tryReceive().getOrNull() - if (buffered != null) { - result = buffered + // One idle window per unit of progress. The inner loop keeps consuming + // non-progress signals INSIDE the same window, so repeat chatter from an + // already-accounted-for relay cannot push the deadline out; only a real + // advance escapes to the outer loop and earns a fresh window. + while (remaining.isNotEmpty()) { + val progressed = + withTimeoutOrNull(timeoutMs) { + while (true) { + val advanced = + select { + eventChannel.onReceive { event -> + result = event remaining.clear() - } else { - remaining.remove(relay) + true + } + doneChannel.onReceive { relay -> + // A relay sends its matching events before its EOSE, so an event may + // already be buffered when this completion fires. select() picks a ready + // clause at random, so without this drain we could treat the relay as done + // and exit while its event still sits unread in the channel. + val buffered = eventChannel.tryReceive().getOrNull() + if (buffered != null) { + result = buffered + remaining.clear() + true + } else { + // Only the FIRST terminal signal from a relay we are still + // waiting on advances the fetch; a repeat is chatter. + remaining.remove(relay) + } } } - } + if (advanced) break } - if (progressed == null) break - } + true + } + if (progressed == null) break } + + // An event can land after the last terminal signal but before we + // unsubscribe; without this drain it would be dropped and the fetch + // would report "nothing found" while holding a match. + if (result == null) result = eventChannel.tryReceive().getOrNull() } finally { unsubscribe(subscriptionId) eventChannel.close() 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 06c56d6be2..8933529397 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 @@ -15,26 +15,36 @@ Import as `com.vitorpamplona.quartz.nip01Core.relay.client.accessories.` ( ## Timeout convention Every `timeoutMs` / `idleTimeoutMs` in this package is an **idle window measured -from the relay's most recent message**, never a wall-clock deadline: each arriving -event / result / terminal signal resets it, so an actively streaming relay is never -cut off mid-delivery — the operation only gives up after a full window of silence. -The shared primitives are in `IdleWatchdog.kt` (`IdleClock` + `receiveWithinIdle`); -use them when writing a new accessory. +from the relay's most recent progress**, never a wall-clock deadline: real progress +resets it, so an actively streaming relay is never cut off mid-delivery — the +operation only gives up after a full window of silence. The shared primitives are in +`IdleWatchdog.kt` (`IdleClock` + `receiveWithinIdle`); use them in a new accessory. -An idle window alone never expires against a relay that trickles messages forever, -so the accessories that wait in **one** loop and then return — `fetchAll` / -`fetchAllWithHooks`, `fetchFirst`, multi-relay `count` — also take a wall-clock -ceiling (`maxTotalMs`, default 10x the idle window; non-positive means uncapped). -There the ceiling genuinely ends the call. +**Progress, not merely traffic.** A message that tells us nothing new — a relay +re-CLOSEing after we already recorded it as done, a duplicate COUNT — must not +restart the window, or a flapping relay keeps the call alive indefinitely. This is +the rule the negentropy watchdog already applies to `NOTICE`/`CLOSED` chatter, and +it is what makes `fetchFirst` and multi-relay `count` self-bounding: at most one +window per relay. -`fetchAllPages` deliberately has **no** ceiling. A per-page ceiling would bound a -page, not the call: the paging loop reacts to a page ending by advancing the cursor -and issuing the next `REQ`, so an endless trickle just gets re-paged forever (a -ceiling of 400 ms against one measured 8 `REQ`s and no return). It also makes -truncation unsafe — cutting a page mid-stream advances `until` to the oldest event -received *so far*, which only preserves the set if the relay streams strictly -newest-first, which NIP-01 recommends but does not require. Bound a paged download -with the filter's `limit`, or by cancelling the caller. +**No accessory takes a wall-clock ceiling parameter.** A hard bound composes at the +call site — `withTimeoutOrNull(ms) { fetchFirst(...) }` — so duplicating it in every +signature buys nothing. Prefer the idle window inside (the caller cannot implement +it; it needs the message stream) and the wall clock outside. Two consequences worth +knowing: + +- `fetchAllPages` has no ceiling and could not usefully have one. A per-page cap + bounds a *page*, not the call: the loop reacts to a page ending by advancing the + cursor and issuing the next `REQ`, so an endless trickle is just re-paged (a + 400 ms cap measured 8 `REQ`s and no return). It also makes truncation unsafe — + cutting a page mid-stream advances `until` to the oldest event received *so far*, + which only preserves the set if the relay streams strictly newest-first, which + NIP-01 recommends but does not require. Bound a paged download with the filter's + `limit`, or by cancelling. +- `fetchAll` / `fetchAllWithHooks` keep a pre-existing `maxTotalMs`, and it earns + its place: an endless *event* trickle there is genuine progress, so the call never + self-terminates, and the internal cap returns the events collected so far where an + external `withTimeoutOrNull` would discard them. The write side is its own case: `publishAndConfirm`'s `timeoutInSeconds` is a fixed window to collect the `OK`s — a bounded confirmation round-trip, not a stream. 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 index d918396017..d52656fc62 100644 --- 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 @@ -32,6 +32,7 @@ import kotlinx.coroutines.delay import kotlinx.coroutines.launch import kotlinx.coroutines.test.currentTime import kotlinx.coroutines.test.runTest +import kotlinx.coroutines.withTimeoutOrNull import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertNull @@ -59,6 +60,8 @@ class FetchFirstIdleTimeoutTest { private val relayA = RelayUrlNormalizer.normalize("wss://a.example.com") private val relayB = RelayUrlNormalizer.normalize("wss://b.example.com") + private val relayC = RelayUrlNormalizer.normalize("wss://c.example.com") + private val relayD = RelayUrlNormalizer.normalize("wss://d.example.com") private fun event(i: Int) = Event( @@ -74,28 +77,29 @@ class FetchFirstIdleTimeoutTest { private fun filters(vararg relays: NormalizedRelayUrl) = relays.associateWith { listOf(Filter(kinds = listOf(1))) } @Test - fun arrivingSignalsRestartTheIdleWindow() = + fun genuineProgressRestartsTheIdleWindow() = 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 + // Each relay's FIRST terminal signal is real progress and buys a + // fresh window, carrying the fetch well past a single 300ms window + // so the slow relay's event at 900ms still lands. 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) + client.listener!!.onClosed("rate limited", relayB, null) delay(250) - client.listener!!.onClosed("rate limited", relayA, null) + client.listener!!.onClosed("rate limited", relayC, null) delay(150) - client.listener!!.onEvent(event(1), false, relayB, null) + client.listener!!.onEvent(event(1), false, relayD, null) } val result = client.fetchFirst( - filters = filters(relayA, relayB), + filters = filters(relayA, relayB, relayC, relayD), timeoutMs = 300, ) - assertEquals(event(1).id, result?.id, "signals must restart the window; the slow relay's event still lands") + assertEquals(event(1).id, result?.id, "progress must restart the window; the slow relay's event still lands") } @Test @@ -113,14 +117,16 @@ class FetchFirstIdleTimeoutTest { } @Test - fun wallClockCeilingStopsEndlessTerminalChatter() = + fun repeatTerminalChatterDoesNotRestartTheIdleWindow() = 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. + // answers. Only relayA's FIRST CLOSED is progress — it removes + // relayA from `remaining`. The repeats say nothing new, so they + // must not push the deadline out (the rule the negentropy + // watchdog already uses for NOTICE/CLOSED chatter). while (true) { delay(200) client.listener!!.onClosed("auth-required: again", relayA, null) @@ -131,28 +137,52 @@ class FetchFirstIdleTimeoutTest { 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") + // First CLOSED at 200ms is the only progress; the window then expires + // 300ms later despite chatter at 400/600/800… + assertEquals(500L, currentTime - start, "repeat chatter must not keep the wait alive") } @Test - fun effectivelyInfiniteIdleWindowDoesNotOverflowTheCeiling() = + fun anEventArrivingAfterTheLastTerminalSignalIsStillReturned() = runTest { val client = ScriptedClient() launch { delay(100) - client.listener!!.onEvent(event(1), false, relayA, null) + // The only relay EOSEs, emptying `remaining` and ending the loop — + // then its matching event lands before we unsubscribe. Without the + // post-loop drain this returns null while holding a match. + client.listener!!.onEose(relayA, null) + client.listener!!.onEvent(event(7), 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, + timeoutMs = 300, ) - assertEquals(event(1).id, result?.id, "an overflowed default ceiling must mean uncapped, not instant timeout") + assertEquals(event(7).id, result?.id, "an event racing the final EOSE must not be dropped") + } + + @Test + fun aHardWallClockBoundIsTheCallersToApply() = + runTest { + val client = ScriptedClient() + val chatter = + launch { + while (true) { + delay(50) + client.listener!!.onClosed("flapping", relayA, null) + } + } + // No ceiling parameter: composing withTimeoutOrNull at the call site + // is the wall-clock bound, and costs nothing because a timed-out + // fetchFirst yields null either way. + val start = currentTime + val result = withTimeoutOrNull(120) { client.fetchFirst(filters = filters(relayA, relayB), timeoutMs = 10_000) } + chatter.cancel() + assertNull(result) + assertEquals(120L, currentTime - start, "the caller's timeout bounds the call") } }