fix: harden negentropy window-splitting and error classification

Audit follow-ups on the NIP-77 client, found while reviewing the notice-rejection fix: - isOverflow was too broad. A bare "too many"/"too large" match meant a NON-shrinking error ("too many requests", "too many concurrent subscriptions") was read as a set-too-large overflow. Such an error doesn't shrink with the window, so every created_at split re-triggers it and reconcileWindows walks toward 1-second leaves, queueing up to ~2^31 Filters (OOM + relay hammering). Tightened to result-set-qualified phrases (too many records / too many query results / result set too large / max_sync_events); rate/quota errors now fail over to paging. - Added a MAX_WINDOWS (100k) backstop in reconcileWindows: a wording-independent guard that bails to paging if a split ever fails to converge, so no novel overflow-looking-but-non-shrinking error can storm. - Window split dropped future-dated events. On overflow the upper child was copy(until = hi) with hi = until ?: now(), so once any split happened, events with created_at > now() (clock skew) were excluded though the un-split path included them. The upper child now keeps the window's original until (may be null = unbounded); the split math still uses now() so it converges. - Hardened the NOTICE rejection matcher. isNegentropyRejectionNotice matched bare "envelope"/"NEG-OPEN"/"NEG-MSG"; since a NOTICE has no subId and every connection listener sees it, an unrelated notice on a shared connection could abort a healthy reconcile mid-handshake. Narrowed to "negentropy"/"unknown envelope"; the now-un-defeated idle watchdog is the wording-independent backstop. - NegentropyStoreSync up-direction memory: haveBatches was an UNLIMITED channel drained by a single network-bound uploader, so a first push of a large store buffered O(local-set) ids. Bounded it like needBatches to back-pressure the reconcile. - Docs: flagged negentropySyncOrFetch's O(delivered) cross-phase dedup memory (steer bulk mirrors to negentropySync/negentropyReconcile); corrected the stale onEvent "reader thread" note (it runs on the delivery consumer). Tests: NegentropyErrorClassificationTest pins both wording classifiers; NegentropyRejectionFallbackTest adds a rate-limit NEG-ERR case asserting paging after exactly one NEG-OPEN per phase (no split storm). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B6ZVTixuc1ef8eGB6MQHRn
2026-08-11 16:57:39 +00:00 · 2026-07-27 19:13:14 +00:00
parent 264866bcbb
commit 30b9e589d4
5 changed files with 249 additions and 37 deletions
@@ -111,3 +111,52 @@ purplepag.es answer ordinary REQs fine, so paging delivers the events).
  matching NOTICE, nor CLOSED-for-subId, just silence) still relies on the idle
  watchdog — which now fires correctly because the refusal chatter no longer
  resets it.
+
+## Follow-up audit (same PR)
+
+A read-through of the whole negentropy accessories package surfaced a few more
+issues; the actionable ones are fixed here.
+
+- **`isOverflow` was too broad → split-storm (fixed).** It matched a bare
+  `"too many"` / `"too large"`, so a *non-shrinking* error — `"too many
+  requests"`, `"too many concurrent subscriptions"` — was read as a
+  set-too-large overflow. Because such an error doesn't shrink with the window,
+  every split re-triggers it and `reconcileWindows` walks toward 1-second leaves,
+  queueing up to ~2³¹ `Filter`s (OOM + relay hammering). Tightened to
+  result-set-qualified phrases (`too many records`, `too many query results`,
+  `result set too large`, `max_sync_events`), so a rate/quota error now fails
+  over to paging. Added a `MAX_WINDOWS` (100k) backstop in `reconcileWindows` —
+  wording-independent — that bails to paging if a split ever fails to converge.
+- **NOTICE matcher hardened (fixed).** The first-cut `isNegentropyRejectionNotice`
+  matched bare `"envelope"` / `"NEG-OPEN"` / `"NEG-MSG"`; since a `NOTICE` has no
+  subId and every connection listener sees it, an unrelated notice on a shared
+  connection could abort a healthy reconcile mid-handshake. Narrowed to
+  `"negentropy"` / `"unknown envelope"` (phrases a NIP-77-speaking relay never
+  emits for a well-formed client); the now-un-defeated idle watchdog is the
+  wording-independent backstop, so under-matching here is safe.
+- **Window split dropped future-dated events (fixed).** On overflow the upper
+  child was `copy(until = hi)` with `hi = until ?: now()`, so once any split
+  happened, events with `created_at > now()` (clock skew) were excluded though
+  the un-split path included them. The upper child now keeps the window's
+  original `until` (may be null = unbounded); the split *math* still uses `now()`
+  so it converges.
+- **`NegentropyStoreSync` up-direction memory (fixed).** `haveBatches` was an
+  UNLIMITED channel drained by a single network-bound uploader, so a first push
+  of a large store buffered O(local-set) ids. Bounded it like `needBatches` so
+  the have-direction back-pressures the reconcile.
+- **`negentropySyncOrFetch` O(delivered) memory (documented).** The cross-phase
+  dedup set is inherent to the combinator's contract; added a KDoc note steering
+  unbounded bulk mirrors to `negentropySync` / `negentropyReconcile` directly.
+
+Noted but not changed (low severity / would cost more than they save):
+
+- `fetchByIds` returns an `ArrayList` mutated on the relay reader thread; on the
+  idle-timeout path there's no channel happens-before, so a late in-flight event
+  could race the worker's iteration. Near-impossible for by-id filters (needs a
+  live event on a specific 32-byte id after the idle deadline); a fix would add
+  per-event synchronization on the download hot path.
+- Per-batch `ArrayList(needIds.subList(...))` copy and the fan-out's no-op
+  `sendHaveBatch` chunk-then-discard are minor allocation churn.
+- `negentropySync`'s "exactly once, no dedup" holds only because relays send the
+  overflow NEG-ERR up-front (before streaming any ids); a relay that streamed
+  partial rounds then overflowed would double-deliver. Latent, not triggered.
@@ -175,8 +175,13 @@ class NegentropyStoreSync(
            try {
                coroutineScope {
                    // needIds = relay has, store lacks; haveIds = store has, relay lacks.
+                    // Both are bounded so a slow consumer back-pressures the reconcile
+                    // instead of buffering the whole residual set. haveBatches drains
+                    // through the single, network-bound uploader (publish + OK wait) far
+                    // slower than the reconcile produces have-ids, so leaving it UNLIMITED
+                    // let a first push of a large local store queue O(local-set) ids.
                    val needBatches = Channel<List<HexKey>>(config.downloadWorkers * 2)
-                    val haveBatches = Channel<List<HexKey>>(Channel.UNLIMITED)
+                    val haveBatches = Channel<List<HexKey>>((config.downloadWorkers * 2).coerceAtLeast(2))

                    val downloaders =
                        List(config.downloadWorkers.coerceAtLeast(1)) {
@@ -149,7 +149,9 @@ class NegentropySyncResult(
 *   cap can wedge the connection, not just fail the extra REQ — size the two knobs
 *   to fit the target relay.
 * @param onProgress        optional `(needSoFar, downloaded)` ticks as work proceeds.
- * @param onEvent           called once per distinct event, on the relay reader thread.
+ * @param onEvent           called once per distinct event, serially, from the single
+ *   delivery consumer coroutine (not the relay reader thread) — so it never overlaps
+ *   itself and the [maxEvents] cap is exact.
 */
@OptIn(ExperimentalAtomicApi::class)
 suspend fun INostrClient.negentropySync(
@@ -287,6 +289,14 @@ class NegentropyOrFetchResult(
 * failing is not delivered again by the paging phase. [maxEvents] is honored across
 * both phases.
 *
+ * **Memory:** unlike bare [negentropySync] (which streams in memory bounded by the
+ * pipeline depth), this holds a set of every delivered id for the whole run — needed
+ * to dedup the paging phase against what negentropy already delivered — so peak heap
+ * is O(delivered ids) (~100 B each). Fine for bounded/filtered syncs; for an
+ * open-ended bulk mirror of a multi-million-event set prefer [negentropySync] (or
+ * [negentropyReconcile]) directly and handle the fallback yourself, to keep the
+ * streaming memory bound.
+ *
 * Use [negentropySync] directly if you want to decide the fallback yourself (try
 * another relay, narrow the filter, abort, …) instead of always paging.
 */
@@ -486,6 +496,15 @@ internal suspend fun reconcileWindows(
    val remaining = AtomicInt(1)
    pending.send(filter)

+    // Total windows ever created (never decremented). A genuine overflow shrinks the
+    // window until it fits, so it converges after a handful of splits; a
+    // non-shrinking error mislabeled as overflow (e.g. a rate limit) would instead
+    // split forever toward 1-second leaves and blow up `pending`. This cap is the
+    // wording-independent backstop [isOverflow]'s narrowing relies on: cross it and
+    // we fail over to paging instead of storming the relay. Legitimate splits are in
+    // the tens–hundreds, so the cap is orders of magnitude above any real sync.
+    val totalWindows = AtomicInt(1)
+
    val reconcilers =
        List(reconcileConcurrency.coerceAtLeast(1)) { reconcilerIndex ->
            launch {
@@ -526,10 +545,26 @@ internal suspend fun reconcileWindows(
                                    detail = "created_at window [$lo, $hi] still exceeds the relay's max_sync_events",
                                )
                            }
+                            if (totalWindows.addAndFetch(2) > MAX_WINDOWS) {
+                                // The split isn't converging — almost always a
+                                // non-shrinking error (rate limit, quota) misread as
+                                // overflow. Bail to paging rather than storm the relay.
+                                throw NegentropySyncException(
+                                    relay = relay,
+                                    window = window,
+                                    reason = NegentropySyncException.Reason.UNAVAILABLE,
+                                    detail = "created_at window split exceeded $MAX_WINDOWS windows without converging; the relay likely rejects negentropy with an overflow-looking error",
+                                )
+                            }
                            val mid = lo + (hi - lo) / 2
                            remaining.incrementAndFetch()
+                            // The lower child gets the finite midpoint; the upper child
+                            // KEEPS this window's original `until` (which may be null =
+                            // unbounded). Replacing null with `now()` here would drop
+                            // every event dated after now() (clock skew) once any split
+                            // happens, while the un-split path would have included them.
                            pending.send(window.copy(since = lo, until = mid))
-                            pending.send(window.copy(since = mid + 1, until = hi))
+                            pending.send(window.copy(since = mid + 1, until = window.until))
                        }

                        is ReconcileOutcome.Failed ->
@@ -985,24 +1020,35 @@ private sealed interface NegFrame {
 }

 /**
- * strfry sends `["NEG-ERR", subId, "blocked: too many query results"]` when a
- * NEG-OPEN matches more than `relay__negentropy__maxSyncEvents`. Match that
- * verbatim, plus a looser contains-check for equivalent "result set too large"
- * wording from other relays, so it still triggers the window split rather than
+ * strfry sends `["NEG-ERR", subId, "blocked: query matches too many records (N > M)"]`
+ * (and, older, `"too many query results"`) when a NEG-OPEN matches more than
+ * `relay__negentropy__maxSyncEvents`. Match that, plus equivalent "result set too
+ * large" wording from other relays, so it triggers the window split rather than
 * aborting.
 *
- * This MUST stay narrow: only a genuine *set-too-large* signal may be treated as
- * overflow, because overflow triggers `created_at` window-splitting. A NEG-ERR
- * that is really a hard refusal — negentropy disabled, `auth-required`, a ban —
- * must NOT match, or every split re-opens, is refused again, and the splitter
- * fans out across the whole `created_at` range (a ~2^31-window storm) instead of
- * failing over to paging. In particular a bare `blocked: …` prefix is such a
- * refusal (e.g. strfry-style `"blocked: Negentropy sync is disabled"`) and is
- * deliberately excluded — only the specific overflow wording counts.
+ * This MUST stay narrow, and specifically must key on the *result-set-size* meaning:
+ * only a genuine set-too-large signal may be treated as overflow, because overflow
+ * triggers `created_at` window-splitting. Two ways a too-lax matcher goes wrong:
+ *  - A hard refusal (negentropy disabled, `auth-required`, a ban) that happens to
+ *    contain a matched word would split, re-open, be refused again, and fan out
+ *    across the whole `created_at` range instead of failing over to paging.
+ *  - A *rate/quota* error — `"too many requests"`, `"too many concurrent
+ *    subscriptions"` — is especially dangerous: it does not shrink as the window
+ *    shrinks, so every split re-triggers it and the splitter walks toward 1-second
+ *    leaves, queueing up to ~2^31 windows (an OOM + relay-hammering storm) before
+ *    any window is small enough to give up on. That is why the bare `"too many"` /
+ *    `"too large"` substrings were replaced with result-set-qualified phrases:
+ *    `"too many requests"` no longer looks like overflow, so it fails over to paging.
+ *
+ * [reconcileWindows] also caps the total window count as a wording-independent
+ * backstop, so a novel overflow-looking-but-not-shrinking error can never storm.
 */
-private fun isOverflow(reason: String): Boolean =
-    reason.contains("too many", ignoreCase = true) ||
-        reason.contains("too large", ignoreCase = true) ||
+internal fun isOverflow(reason: String): Boolean =
+    reason.contains("too many records", ignoreCase = true) ||
+        reason.contains("too many results", ignoreCase = true) ||
+        reason.contains("too many query results", ignoreCase = true) ||
+        reason.contains("result set too large", ignoreCase = true) ||
+        reason.contains("results too large", ignoreCase = true) ||
        reason.contains("max_sync_events", ignoreCase = true)

 /**
@@ -1015,14 +1061,19 @@ private fun isOverflow(reason: String): Boolean =
 * We only treat a NOTICE as our negentropy rejection when it plausibly refers to the
 * NEG exchange (this matcher) AND it arrives before this session's first valid NEG
 * frame — so an unrelated NOTICE on a healthy relay mid-reconcile can never abort an
- * otherwise-progressing sync. This MUST stay narrow for the same reason [isOverflow]
- * must: a false positive fails the whole window over to paging.
+ * otherwise-progressing sync. This is only a *fast path*: it is deliberately narrow
+ * (a false positive fails the window over to paging), and anything it misses is still
+ * caught by the idle watchdog, which — since NOTICE/CLOSED no longer bump the clock —
+ * fires once a refusing relay goes silent after its notice. So prefer under-matching
+ * here. Both matched phrases are ones a relay that actually speaks NIP-77 would never
+ * emit for a well-formed client (quartz only sends valid frames): "negentropy" names
+ * the feature; "unknown envelope" is the parse failure of a relay that never
+ * implemented the NEG-OPEN envelope. Broad substrings like a bare "envelope" or the
+ * echoed command names are excluded — an unrelated parse/rate NOTICE could carry them.
 */
-private fun isNegentropyRejectionNotice(reason: String): Boolean =
+internal fun isNegentropyRejectionNotice(reason: String): Boolean =
    reason.contains("negentropy", ignoreCase = true) ||
-        reason.contains("envelope", ignoreCase = true) ||
-        reason.contains("NEG-OPEN", ignoreCase = true) ||
-        reason.contains("NEG-MSG", ignoreCase = true)
+        reason.contains("unknown envelope", ignoreCase = true)

 /**
 * One `REQ` for [batch] ids; collects the matching events and returns them on
@@ -1101,6 +1152,15 @@ internal suspend fun INostrClient.fetchByIds(
 /** Seconds: a window this small that still overflows can't be split further. */
 private const val MIN_WINDOW_SECONDS = 1L

+/**
+ * Hard cap on total `created_at` windows a single reconcile may split into, a
+ * wording-independent backstop against a non-shrinking error (rate limit, quota)
+ * being mistaken for a set-too-large overflow and splitting forever. A real sync
+ * against a huge relay converges in tens–hundreds of windows, so this is a wide
+ * margin; crossing it fails the sync over to paging instead of storming the relay.
+ */
+private const val MAX_WINDOWS = 100_000
+
 /** Bounded buffer between the download workers and the single delivery consumer. */
 private const val DELIVERY_BUFFER = 256

@@ -0,0 +1,74 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quartz.nip01Core.relay.client.accessories
+
+import kotlin.test.Test
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+/**
+ * Contract for the two narrow NEG-ERR/NOTICE wording classifiers. Both feed
+ * irreversible control flow — [isOverflow] triggers `created_at` window-splitting,
+ * [isNegentropyRejectionNotice] aborts a reconcile to paging — so a false positive is
+ * costly (a split storm, or a healthy sync abandoned). These lock the boundary.
+ */
+class NegentropyErrorClassificationTest {
+    @Test
+    fun overflowMatchesResultSetSizeErrors() {
+        // strfry, current + older wording, and equivalents from other relays.
+        assertTrue(isOverflow("blocked: query matches too many records (2988225 > 1000000)"))
+        assertTrue(isOverflow("too many query results"))
+        assertTrue(isOverflow("ERROR: too many results"))
+        assertTrue(isOverflow("result set too large"))
+        assertTrue(isOverflow("results too large to sync"))
+        assertTrue(isOverflow("exceeds max_sync_events"))
+    }
+
+    @Test
+    fun overflowRejectsRateAndRefusalErrors() {
+        // These do NOT shrink when the window shrinks — treating them as overflow
+        // would split forever toward 1-second leaves and storm the relay. They must
+        // fail over to paging instead.
+        assertFalse(isOverflow("rate-limited: too many requests"))
+        assertFalse(isOverflow("error: too many concurrent subscriptions"))
+        assertFalse(isOverflow("too many connections"))
+        assertFalse(isOverflow("blocked: message too large"))
+        assertFalse(isOverflow("blocked: negentropy disabled"))
+        assertFalse(isOverflow("auth-required: restricted"))
+    }
+
+    @Test
+    fun rejectionNoticeMatchesTheObservedRefusals() {
+        assertTrue(isNegentropyRejectionNotice("ERROR: bad msg: negentropy disabled"))
+        assertTrue(isNegentropyRejectionNotice("Negentropy sync is disabled"))
+        assertTrue(isNegentropyRejectionNotice("failed to parse envelope: unknown envelope label"))
+    }
+
+    @Test
+    fun rejectionNoticeIgnoresUnrelatedNotices() {
+        // A NOTICE has no subId, so an over-broad matcher would let unrelated traffic
+        // on the shared connection abort a healthy reconcile mid-handshake.
+        assertFalse(isNegentropyRejectionNotice("rate-limited: slow down"))
+        assertFalse(isNegentropyRejectionNotice("invalid: bad event envelope size"))
+        assertFalse(isNegentropyRejectionNotice("could not parse REQ"))
+        assertFalse(isNegentropyRejectionNotice("restricted: auth required"))
+    }
+}
@@ -32,6 +32,7 @@ import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
@@ -60,15 +61,17 @@ class NegentropyRejectionFallbackTest {
    private fun subIdOf(frame: String): String? = Regex("^\\[\"[A-Z-]+\",\"([^\"]+)\"").find(frame)?.groupValues?.get(1)

    /**
-     * A fake relay whose reply to each sent frame is decided by [replyToNegOpen].
-     * All replies (and the initial onOpen) are posted on a background executor so the
-     * socket behaves like a real async OkHttp socket — never re-entrant into the
-     * client's send path.
+     * A fake relay whose reply to a NEG-OPEN is produced by [replyToNegOpen] (given the
+     * NEG-OPEN's subId), counting NEG-OPENs so a test can assert the client did NOT
+     * split-storm. All replies (and the initial onOpen) are posted on a background
+     * executor so the socket behaves like a real async OkHttp socket — never re-entrant
+     * into the client's send path.
     */
    private inner class ScriptedRelay(
-        val replyToNegOpen: String,
+        val replyToNegOpen: (subId: String) -> String,
    ) : WebsocketBuilder {
        val io = Executors.newSingleThreadScheduledExecutor()
+        val negOpens = AtomicInteger(0)

        override fun build(
            url: NormalizedRelayUrl,
@@ -89,8 +92,11 @@ class NegentropyRejectionFallbackTest {
                            // The keep-alive REQ and any paging REQ: answer EOSE so the
                            // subscription settles (paging then completes with 0 events).
                            msg.startsWith("[\"REQ\"") -> subIdOf(msg)?.let { out.onMessage("[\"EOSE\",\"$it\"]") }
-                            // The negentropy handshake: the relay refuses via NOTICE.
-                            msg.startsWith("[\"NEG-OPEN\"") -> out.onMessage(replyToNegOpen)
+                            // The negentropy handshake: the relay refuses.
+                            msg.startsWith("[\"NEG-OPEN\"") -> {
+                                negOpens.incrementAndGet()
+                                subIdOf(msg)?.let { out.onMessage(replyToNegOpen(it)) }
+                            }
                            else -> Unit
                        }
                    }, 5, TimeUnit.MILLISECONDS)
@@ -101,8 +107,8 @@ class NegentropyRejectionFallbackTest {
        fun shutdown() = io.shutdownNow()
    }

-    private fun negOpenRejectedBy(notice: String) {
-        val relay = ScriptedRelay(notice)
+    private fun negOpenRejectedBy(reply: (subId: String) -> String): ScriptedRelay {
+        val relay = ScriptedRelay(reply)
        val client = NostrClient(relay)
        try {
            runBlocking {
@@ -115,7 +121,7 @@ class NegentropyRejectionFallbackTest {
                    }
                assertTrue(
                    thrown.reason == NegentropySyncException.Reason.UNAVAILABLE,
-                    "a NOTICE rejection should be UNAVAILABLE, was ${thrown.reason}",
+                    "a runtime negentropy refusal should be UNAVAILABLE, was ${thrown.reason}",
                )

                // negentropySyncOrFetch must transparently fall back to paging.
@@ -123,18 +129,36 @@ class NegentropyRejectionFallbackTest {
                    withTimeout(8_000) {
                        client.negentropySyncOrFetch(url, Filter(kinds = listOf(0))) { }
                    }
-                assertTrue(result.pagedFallback, "expected paging fallback after NOTICE rejection")
+                assertTrue(result.pagedFallback, "expected paging fallback after the refusal")
                assertEquals(0, result.downloaded)
            }
        } finally {
            client.close()
            relay.shutdown()
        }
+        return relay
    }

    @Test
-    fun strfryNegentropyDisabledFallsBackToPaging() = negOpenRejectedBy("[\"NOTICE\",\"ERROR: bad msg: negentropy disabled\"]")
+    fun strfryNegentropyDisabledFallsBackToPaging() {
+        negOpenRejectedBy { "[\"NOTICE\",\"ERROR: bad msg: negentropy disabled\"]" }
+    }

    @Test
-    fun purplePagesUnknownEnvelopeFallsBackToPaging() = negOpenRejectedBy("[\"NOTICE\",\"failed to parse envelope: unknown envelope label\"]")
+    fun purplePagesUnknownEnvelopeFallsBackToPaging() {
+        negOpenRejectedBy { "[\"NOTICE\",\"failed to parse envelope: unknown envelope label\"]" }
+    }
+
+    @Test
+    fun rateLimitNegErrPagesWithoutSplitStorm() {
+        // A NEG-ERR that does NOT shrink with the window ("too many requests") must not
+        // be mistaken for a set-too-large overflow: doing so would binary-split the
+        // created_at range forever. Assert we page after exactly ONE NEG-OPEN.
+        val relay = negOpenRejectedBy { subId -> "[\"NEG-ERR\",\"$subId\",\"rate-limited: too many requests\"]" }
+        assertEquals(
+            2,
+            relay.negOpens.get(),
+            "one NEG-OPEN per phase (sync + syncOrFetch), i.e. no window-split storm; got ${relay.negOpens.get()}",
+        )
+    }
 }