Make a paged walk terminate: floor the cursor at 0, stop when a relay ignores it

`fetchAllPages` could not end against a relay that does not honour `until`.
Found in production on purplepag.es, which holds twelve `kind 10002` events
stamped `created_at = 0` and treats `until <= 0` as *no* `until` — so the page
below them comes back with its five hundred NEWEST events. None of those
matches the filter's own `until`, so the page delivers nothing, which read as
"the boundary second is too dense to page", stepped one second lower, and
asked the identical unanswerable question again.

Measured against the live relay: ~5.5 pages a second, 500 events fetched and
discarded on each, an EOSE on every single 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, so the caller's
coverage was never recorded and the next boot re-walked all of it. Not a rate
limit: ~1,000 consecutive pages drew no NOTICE, no CLOSED and no throttling.

Two guards, both at the points where the cursor moves:

  - The cursor floors at zero. `created_at` is unsigned, 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 DRAINED. `until = 0` is still asked — it is a
    legal query and the boundary re-fetch for epoch-stamped events — only going
    BELOW it 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. The floor is applied
    on the advance path too, not just the step: `pageMinTs` is an event's own
    `created_at`, so one relay serving a negative timestamp is enough to drive
    the cursor under zero, and clamping rather than stopping would not help —
    such an event never equals the boundary, so it dodges the dedup and returns
    on every page.

  - A relay that ignores the cursor is UNPAGEABLE. When a page delivers 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 repeats the
    question. `aboveBoundary == received` is what tells this apart from a
    genuinely dense boundary second, whose events are AT the boundary rather
    than above it. 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.

This second guard is the structural fix. Giving the filter a `since` does not
substitute for it: the step path decrements `until` without regard to `since`,
so a cursor-ignoring relay still walks from the window floor down to 0 — up to
~1.5 billion pages. A `since` only helps when the relay honours it, and then
only because the empty page arrives as a drain.

Three scripted tests cover both guards, and CursorTerminationProbe dials the
five relays that found this (opt-in, `-PprodRelayBench=1`, asserts nothing).
Against the live relays after the change: purplepag.es ends UNPAGEABLE in one
page and 2.4s with the cursor never going under 0, where it previously ran 244
pages to `until = -243` without ending; the other four still DRAIN unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HPSzniNdvJxkhRsCe1QcyT

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<HexKey>()
 
@@ -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 {

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")
+        }
 }

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
+    }
+}