laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-09 16:14:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: drop the ceiling params; bound waits by progress instead

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KZe1Pq2ejoHehdufhPDRgb
This commit is contained in:
Claude
2026-08-03 03:26:58 +00:00
parent ac6034f764
commit ba35a87a7c
5 changed files with 209 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientCountExt.kt
+37 -32
View File
@@ -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<NormalizedRelayUrl, List<Filter>>,
timeoutMs: Long = 15_000,
maxTotalMs: Long = timeoutMs * 10,
): Map<NormalizedRelayUrl, CountResult> {
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) }
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt
+44 -31
View File
@@ -178,40 +178,53 @@ suspend fun INostrClient.fetchAllPages(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
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()
}
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchFirstExt.kt
+51 -35
View File
@@ -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<NormalizedRelayUrl, List<Filter>>,
timeoutMs: Long = 30_000L,
maxTotalMs: Long = timeoutMs * 10,
): Event? {
val eventChannel = Channel<Event>(UNLIMITED)
val doneChannel = Channel<NormalizedRelayUrl>(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<Unit> {
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<Boolean> {
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()
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/README.md
+28 -18
View File
@@ -15,26 +15,36 @@ Import as `com.vitorpamplona.quartz.nip01Core.relay.client.accessories.<name>` (
## 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.
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/FetchFirstIdleTimeoutTest.kt
+49 -19
View File
@@ -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")
}
}