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 203311e8e4..e5e98116f8 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 @@ -134,6 +134,28 @@ data class PagedFetchResult( * cap, so a larger value is clamped to the same page). Stepping past at least keeps * the download progressing to older events instead of stalling forever. * + * **Two guards keep that step from becoming a walk that never ends**, both learned + * from a relay in production rather than from reasoning: + * + * - **The cursor floors at zero.** `created_at` is an unsigned timestamp, so nothing + * can exist below epoch 0: a cursor that would step under it has reached the bottom + * of the time axis and the walk is [PagedFetchResult.End.DRAINED]. `until = 0` + * itself is still asked — it is a legal query, and the boundary re-fetch for events + * stamped at the epoch — it is only going *below* it that ends the walk. This also + * keeps a negative `until` off the wire, which relays disagree violently about: + * measured across five, one CLOSEs the subscription with a parse error, three + * answer a `NOTICE` and then never EOSE, and one drops the bound and serves its + * NEWEST events. + * - **A relay that ignores the cursor is [PagedFetchResult.End.UNPAGEABLE].** If a + * page delivered nothing and every event it received was NEWER than the `until` it + * asked for, the relay is not paging at all, and stepping one second lower just + * asks the same unanswered question again. That is exactly how the first guard's + * relay behaves — it treats `until <= 0` as no `until` — and without this the walk + * ran ~5.5 pages a second, 500 events fetched and discarded on each, EOSE on every + * one, for as long as the process lived. UNPAGEABLE is deliberate and conservative: + * it proves nothing about what the relay holds, so no coverage claim can be built + * on a page the relay never really answered. + * * A `search` ([Filter.search]) filter is the exception: NIP-50 results are ranked by * relevance, not `created_at`, so paging one by a `until` cursor is meaningless — it * would silently turn a top-N search into a time-walk, and never terminate against a @@ -259,6 +281,14 @@ suspend fun INostrClient.fetchAllPages( val boundary = until var received = 0 var delivered = 0 + + /** + * Events that came back NEWER than the `until` this page asked for — which an + * honest relay never sends. Counted because it is the only way to tell a relay + * that ignored the cursor apart from a boundary second too dense to page: both + * deliver nothing, and only one of them can be fixed by stepping past. + */ + var aboveBoundary = 0 var pageMinTs = Long.MAX_VALUE val idsAtPageMin = HashSet() @@ -289,6 +319,9 @@ suspend fun INostrClient.fetchAllPages( // early) or an unsafely published `idsAtPageMin`. try { received++ + // Before the dedup return, so it is counted for every event + // the page received, not just the ones that reach the match. + if (boundary != null && event.createdAt > boundary) aboveBoundary++ // 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 @@ -414,6 +447,35 @@ suspend fun INostrClient.fetchAllPages( // are resolved by stepping strictly past it. `boundary` is null only on // the first page, which has no dedup and so can't be all-duplicate. val step = boundary ?: break // first page, all-duplicate: impossible, and `end` stays UNPAGEABLE + + // The relay is not honouring `until`: every event it sent was NEWER than + // the cursor this page asked for. Stepping past cannot help — the next + // page repeats the same ask one second lower and gets the same answer, + // forever. Measured on a live relay (purplepag.es, which treats + // `until <= 0` as no `until` and answers with its newest page): ~5.5 + // pages a second, 500 events fetched and discarded on each, `until` + // marching one second further negative every time, an EOSE on every + // single page, for as long as the process ran. This is the ONE reading + // that ends it, and it is safely conservative — UNPAGEABLE proves + // nothing about what the relay holds, so no coverage claim is built on + // a page the relay never actually answered. + if (aboveBoundary == received) { + end = PagedFetchResult.End.UNPAGEABLE + break + } + + // Below the boundary there is nothing left to ask for: `created_at` is an + // unsigned timestamp, so no event can exist under epoch 0 and a cursor + // stepping past it has reached the bottom of the time axis. Ending here + // rather than sending `until = -1` also keeps a value off the wire that + // relays disagree violently about — measured across five: one CLOSEs the + // subscription with a parse error, three answer a NOTICE and then never + // EOSE (so every page burns a whole idle timeout), one drops the bound + // and serves its newest events. + if (step <= 0L) { + end = PagedFetchResult.End.DRAINED + break + } until = step - 1 seenAtBoundary = HashSet() continue @@ -432,6 +494,17 @@ suspend fun INostrClient.fetchAllPages( // termination both rely on `until` never increasing. Honest relays only // return events at-or-below `until`, so this is a no-op for them. val nextUntil = if (boundary != null) minOf(pageMinTs, boundary) else pageMinTs + + // The same floor as the step above, on the other way the cursor moves. It is + // reachable here too, and not only through a bug: `pageMinTs` is an event's + // own `created_at`, so one relay serving a negative timestamp is enough to + // put the cursor under zero. Clamping to 0 instead of stopping would not + // help — such an event never equals the boundary, so it dodges the dedup and + // comes back on every page, pinning the walk there for good. + if (nextUntil < 0L) { + end = PagedFetchResult.End.DRAINED + break + } if (boundary != null && nextUntil == boundary) { seenAtBoundary.addAll(idsAtPageMin) } else { diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt index f2de98b505..bc949a4826 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientFetchAllPagesDrainTest.kt @@ -224,4 +224,115 @@ class NostrClientFetchAllPagesDrainTest { assertEquals(PagedFetchResult.End.LIMIT_REACHED, result.end, "a fulfilled limit is the caller stopping, not the corpus ending") assertFalse(result.drained) } + + // ---- termination: the walk must END, whatever the relay does ------------- + + @Test + fun aRelayThatIgnoresTheCursorEndsTheWalkInsteadOfSteppingForever() = + runBlocking { + // The production bug, scripted. purplepag.es holds events stamped + // `created_at = 0` and treats `until <= 0` as NO `until`, so the page + // below them comes back with its NEWEST events instead. None of those + // matches the filter's own `until`, so the page delivers nothing — + // which used to read as "the boundary second is too dense", step one + // second lower, and ask the identical unanswerable question again. + // Measured against the live relay: ~5.5 pages a second, 500 events + // discarded on each, an EOSE on every one, for as long as the process + // ran. `aboveBoundary == received` is what tells the two apart. + val client = ScriptedClient() + val feeder = + launch { + client.awaitPage(1) + client.listener!!.onEvent(event(2000), false, relay, null) + client.listener!!.onEvent(event(1000), false, relay, null) + client.listener!!.onEose(relay, null) + + // Page two asks for `until = 1000` and gets events from the top + // of the corpus — the answer to a query nobody made. + client.awaitPage(2) + client.listener!!.onEvent(event(9000), false, relay, null) + client.listener!!.onEvent(event(8000), false, relay, null) + client.listener!!.onEose(relay, null) + } + + val result = + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1))), + idleTimeoutMs = 2_000, + ) { } + feeder.join() + + assertEquals(2, result.downloaded, "only the two events the relay actually answered for") + assertEquals(2, client.subscribeCount, "and it stops on the FIRST page the relay refused to page") + assertEquals(PagedFetchResult.End.UNPAGEABLE, result.end, "a relay ignoring `until` is not paging, and cannot be stepped past") + assertFalse(result.drained, "which proves nothing about what it holds, so no coverage may be claimed") + } + + @Test + fun aCursorSteppingUnderTheEpochDrainsInsteadOfGoingNegative() = + runBlocking { + // `created_at` is unsigned, so nothing exists below epoch 0. A boundary + // second AT the epoch that only ever returns duplicates has reached the + // bottom of the time axis: the walk is done, and `until = -1` must never + // reach a relay — one of the five indexers CLOSEs the subscription over + // it, three answer a NOTICE and then never EOSE. + val client = ScriptedClient() + val feeder = + launch { + client.awaitPage(1) + client.listener!!.onEvent(event(0), false, relay, null) + client.listener!!.onEose(relay, null) + + // Page two re-asks the boundary inclusively and gets back only + // the event page one already delivered: nothing new, and nowhere + // left below to step to. + client.awaitPage(2) + client.listener!!.onEvent(event(0), false, relay, null) + client.listener!!.onEose(relay, null) + } + + val result = + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1))), + idleTimeoutMs = 2_000, + ) { } + feeder.join() + + assertEquals(1, result.downloaded, "the epoch event, delivered once") + assertEquals(2, client.subscribeCount, "no third page: there is nothing under zero to ask for") + assertEquals(PagedFetchResult.End.DRAINED, result.end, "the bottom of the time axis is an end, not a stall") + assertTrue(result.drained) + } + + @Test + fun anEventStampedBeforeTheEpochCannotPinTheWalk() = + runBlocking { + // `pageMinTs` is an event's own `created_at`, so one relay serving a + // negative timestamp drives the cursor under zero on the ADVANCE path + // rather than the step path. Clamping to 0 would not save it: such an + // event never equals the boundary, so it dodges the dedup and comes + // back on every page, pinning the walk at 0 for good. + val client = ScriptedClient() + val feeder = + launch { + client.awaitPage(1) + client.listener!!.onEvent(event(2000), false, relay, null) + client.listener!!.onEvent(event(-5), false, relay, null) + client.listener!!.onEose(relay, null) + } + + val result = + client.fetchAllPages( + relay = relay, + filters = listOf(Filter(kinds = listOf(1))), + idleTimeoutMs = 2_000, + ) { } + feeder.join() + + assertEquals(2, result.downloaded, "both events are still delivered — they were received") + assertEquals(1, client.subscribeCount, "but there is no second page to ask") + assertEquals(PagedFetchResult.End.DRAINED, result.end, "below the epoch there is nothing left to walk") + } } diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/CursorTerminationProbe.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/CursorTerminationProbe.kt new file mode 100644 index 0000000000..a0efae418e --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/CursorTerminationProbe.kt @@ -0,0 +1,167 @@ +/* + * 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.prodbench + +import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull +import okhttp3.OkHttpClient +import java.time.Duration +import kotlin.test.Test + +/** + * The live half of the paging termination guards, against the relay that found + * them. [NostrClientFetchAllPagesDrainTest] scripts this behaviour, so it pins our + * INTERPRETATION of a relay; only dialling one can say whether the interpretation + * matches anything real. + * + * ## What it walks, and why that relay + * + * purplepag.es holds twelve `kind 10002` events stamped `created_at = 0` and treats + * `until <= 0` as *no* `until`, answering with its five hundred NEWEST events. A + * cursor walk therefore reaches zero, gets a page it never asked for, delivers none + * of it, and — before the guards — stepped one second lower and asked again. Measured + * against the live relay: ~5.5 pages a second, 500 events fetched and discarded on + * each, an EOSE on *every* page, `until` marching one second further negative every + * time, for as long as the process ran. A cold walk pulled 1,490,010 real events in + * ~10.8 minutes and then never returned. + * + * The ceiling is set just above the epoch-stamped events rather than `now` on + * purpose: this relay serves ~2,300 kind 0/10002 events a second and holds years of + * them, so starting at the top would spend a quarter of an hour on history that is + * not what this measures. One page from [TRAP_CEILING] already carries them. + * + * ## Reading it + * + * `lowest until` is the whole tell. A walk that ends leaves it at a real timestamp; a + * walk that cannot end leaves it below zero. With the guards in place the expected + * report is `UNPAGEABLE` with the cursor never going under `0`. + * + * OFF by default and not a gate: it dials the public internet, so it is neither + * hermetic nor reproducible, and a relay being down is not a code regression. It + * asserts nothing for that reason — it REPORTS, and a human reads it. + * + * ``` + * ./gradlew :quartz:jvmTest --tests "*.CursorTerminationProbe" -PprodRelayBench=1 -i + * ``` + */ +class CursorTerminationProbe { + @Test + fun reportWhetherAPagedWalkTerminates() { + if (System.getenv("PROD_RELAY_BENCH") == null && System.getProperty("prodRelayBench") == null) { + println("reportWhetherAPagedWalkTerminates skipped. Run with -PprodRelayBench=1 to enable.") + return + } + val okhttp = + OkHttpClient + .Builder() + .connectTimeout(Duration.ofSeconds(20)) + .pingInterval(Duration.ofSeconds(120)) + .build() + val scope = CoroutineScope(SupervisorJob()) + val client = NostrClient(BasicOkHttpWebSocket.Builder { okhttp }, scope) + + println("=".repeat(78)) + println("Does a paged walk TERMINATE? kinds [0, 10002], from $TRAP_CEILING down") + println("=".repeat(78)) + try { + for (url in RELAYS) { + val relay = RelayUrlNormalizer.normalize(url) + var events = 0 + var pages = 0 + var lowest = Long.MAX_VALUE + val startedAt = System.currentTimeMillis() + val outcome = + runCatching { + runBlocking { + // A hard ceiling, which `fetchAllPages` deliberately does + // not have: its own doc says a walk is bounded by a + // `limit` or by cancelling the caller, and this is the + // caller cancelling. Without it a relay with no guard + // hangs the probe — which is exactly what it is here to + // detect, so it must be detected rather than suffered. + withTimeoutOrNull(TERMINATION_MS) { + client.fetchAllPages( + relay, + listOf(Filter(kinds = listOf(0, 10002), until = TRAP_CEILING)), + idleTimeoutMs = 20_000L, + onNewPage = { until -> + pages++ + if (until < lowest) lowest = until + }, + ) { events++ } + } + } + } + val took = System.currentTimeMillis() - startedAt + val verdict = + outcome.fold( + onSuccess = { r -> + when (r) { + null -> "NEVER ENDED in ${TERMINATION_MS / 1000}s — THE GUARD IS NOT WORKING" + else -> "${r.end} (${r.downloaded} event(s), drained=${r.drained})" + } + }, + onFailure = { "threw ${it::class.simpleName}: ${it.message}" }, + ) + val reached = if (lowest == Long.MAX_VALUE) "no page after the first" else "$lowest" + println(" %-26s %-52s".format(url.removePrefix("wss://"), verdict)) + println(" %-26s %d page(s), %d event(s), %dms, lowest until=%s".format("", pages, events, took, reached)) + } + } finally { + runCatching { client.disconnect() } + scope.cancel() + } + println("=".repeat(78)) + } + + companion object { + /** + * purplepag.es is the one that found this. The other four are controls: they + * hold nothing at all below `1.5e9`, so they drain in a single page and prove + * the guards did not change an ordinary walk. + */ + private val RELAYS = + listOf( + "wss://purplepag.es", + "wss://user.kindpag.es", + "wss://directory.yabu.me", + "wss://profiles.nostr1.com", + "wss://indexer.coracle.social", + ) + + /** Just above the `created_at = 0` events, so one page reaches the cursor that matters. */ + private const val TRAP_CEILING = 1_600_000_000L + + /** + * Not an idle timeout — the relay answers, with an EOSE, the entire time. + * This is how long a walk gets to prove it can END. + */ + private const val TERMINATION_MS = 45_000L + } +}