From 264866bcbbbc8a6726452b965dcf5f5d6899130f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 27 Jul 2026 18:12:01 +0000 Subject: [PATCH 1/3] fix: detect NIP-77 negentropy refusal sent as a connection-level NOTICE Relays that advertise NIP-77 in NIP-11 but refuse it at runtime answer a NEG-OPEN with a connection-level NOTICE (which carries no subId) instead of a subId-addressed NEG-ERR: - strfry with negentropy disabled: "ERROR: bad msg: negentropy disabled" - purplepag.es (no NEG envelope): "failed to parse envelope: unknown envelope label" reconcileStreaming only routed NegMsg/NegErr for its exact subId into the driver channel, so the NOTICE was dropped and the driver blocked in receiveWithinIdle with no terminating frame. Worse, the connection-level idle watchdog was bumped by every relay message, so unrelated refusal chatter (a rejected keep-alive REQ being re-CLOSED on re-sync) reset it forever and it never fired. Net effect: negentropySync/negentropyReconcile hung against relay.primal.net and purplepag.es, and negentropySyncOrFetch never reached its paging fallback. Fix, in reconcileStreaming's connection listener: - route a CLOSED for our NEG subId into the driver as a terminal failure; - treat a negentropy-refusal NOTICE as terminal, bound to this session by phase (before the first valid NEG frame) + wording (isNegentropyRejectionNotice), so an unrelated NOTICE on a healthy relay mid-reconcile cannot abort a progressing sync; - stop bumping the idle clock on NOTICE/CLOSED so refusal chatter can no longer keep a dead sync alive; it now advances only on real progress (this session's NEG frames and the download REQs' events/EOSEs). The refusal now surfaces as NegentropySyncException(UNAVAILABLE); negentropySync throws promptly and negentropySyncOrFetch pages the same filter (both relays answer ordinary REQs fine). Verified live: primal 0-event hang -> pages 11.7k events in 6.4s; purplepag.es 0-event hang -> pages continuously. Tests: - NegentropyRejectionFallbackTest: offline, deterministic; a scripted fake relay answers NEG-OPEN with each observed NOTICE and asserts the sync throws UNAVAILABLE fast and negentropySyncOrFetch sets pagedFallback. - NegentropyStallRepro: gated live repro against the three real relays. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01B6ZVTixuc1ef8eGB6MQHRn --- ...07-27-negentropy-notice-rejection-stall.md | 113 ++++++++ .../NostrClientNegentropySyncExt.kt | 85 +++++- .../client/NegentropyRejectionFallbackTest.kt | 140 ++++++++++ .../relay/prodbench/NegentropyStallRepro.kt | 254 ++++++++++++++++++ 4 files changed, 584 insertions(+), 8 deletions(-) create mode 100644 quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NegentropyRejectionFallbackTest.kt create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyStallRepro.kt diff --git a/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md b/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md new file mode 100644 index 0000000000..3ea486849f --- /dev/null +++ b/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md @@ -0,0 +1,113 @@ +# NIP-77 client stalls against relays that refuse negentropy via NOTICE + +**Status: root-caused + fixed (reproduced live, regression-tested offline).** + +## Symptom + +`INostrClient.negentropySyncOrFetch` / `negentropyReconcile` hang forever against +some relays that advertise NIP-77 in NIP-11: + +- `wss://relay.ditto.pub` → works (events download, call returns). +- `wss://relay.primal.net` (strfry) → stalls: `onEvent` never fires, `downloaded` + stays 0, the suspend fun never returns and never hits the idle timeout. +- `wss://purplepag.es` → stalls, same shape. + +Only an external `withTimeout` wall clock unblocked the caller. + +## Reproduction + +`quartz/.../prodbench/NegentropyStallRepro.kt` (gated on `NEG_STALL_REPRO=1`) +builds the exact reported client — `NostrClient(BasicOkHttpWebSocket.Builder { okHttpClient })` +— wraps the socket to log every frame both ways, and runs +`negentropySyncOrFetch(relay, Filter(kinds=[0]), localEntries=emptyList())` against +all three relays under a 60 s external wall clock. + +Wire trace (the decisive lines): + +``` +DITTO -> NEG-OPEN {kinds:[0]} + <- NEG-ERR "blocked: query matches too many records (2988225 > 1000000)" # overflow -> window split + <- NEG-MSG <120 KB id frames> ... # reconciles, streams ids + => 53,811 events delivered in 60 s (working; would finish given time) + +PRIMAL -> NEG-OPEN {kinds:[0]} + <- NOTICE "ERROR: bad msg: negentropy disabled" # refusal, NO subId + => 0 events, STALLED (only the 60 s wall clock freed it) + +PURPLEPAGES -> NEG-OPEN {kinds:[0]} + <- CLOSED "blocked: filters must specify at least one kind" # rejects the keep-alive REQ + <- NOTICE "failed to parse envelope: unknown envelope label" # doesn't know NEG-OPEN + => 0 events, STALLED +``` + +So it is **not** strfry-specific, not a large-corpus reconcile-convergence +problem, and not the fetch stage: the relays never enter reconciliation at all. +They **refuse negentropy** — one has it switched off, the other never implemented +the envelope — and both signal the refusal with a connection-level `NOTICE` +(strfry) / `NOTICE`+`CLOSED` (purplepag.es). NIP-11 advertising NIP-77 is not a +runtime guarantee. + +## Root cause (a quartz client bug) + +`reconcileStreaming` (in `NostrClientNegentropySyncExt.kt`) installs a +`RelayConnectionListener` that only routes two message types into its driver +channel: + +```kotlin +is NegMsgMessage -> if (msg.subId == subId) incoming.trySend(NegFrame.Msg(...)) +is NegErrMessage -> if (msg.subId == subId) incoming.trySend(NegFrame.Err(...)) +else -> Unit +``` + +A `NOTICE` carries **no subId** (it is a connection-level message), so it can +never match and falls into `else -> Unit`. The driver sits in +`receiveWithinIdle(clock, idleTimeoutMs)` waiting for a frame that never comes. + +Why the idle watchdog didn't save it: the connection-level `IdleClock` was bumped +on **every** message the relay sent. In isolation the relay goes silent after the +`NOTICE`, so the 120 s idle *would* eventually fire (the repro just used a shorter +60 s wall clock) — but in concurrent/real use the same connection keeps chattering +(the rejected keep-alive REQ being re-`CLOSED` on re-sync, other subscriptions' +traffic), and each such frame reset the watchdog, so it never fired. Net effect: +`negentropySync` never throws → `negentropySyncOrFetch` never reaches its paging +fallback → caller hangs. + +## Fix + +In `reconcileStreaming`'s listener: + +1. **Route a terminal `CLOSED` for our NEG subId** into the driver as a failure. +2. **Treat a negentropy-refusal `NOTICE` as terminal**, bound to this session by + *phase + wording*: only before the first valid `NEG` frame (`sawNegFrame`), + and only when the text looks like a negentropy/parse refusal + (`isNegentropyRejectionNotice`: contains `negentropy` / `envelope` / + `NEG-OPEN` / `NEG-MSG`). Both conditions keep it narrow so an unrelated NOTICE + on a healthy relay mid-reconcile can never abort an otherwise-progressing sync. +3. **Stop bumping the idle clock on `NOTICE`/`CLOSED`.** The watchdog now advances + only on real progress — this session's own `NEG` frames and the download REQs' + `EVENT`/`EOSE` — so error chatter can no longer keep a dead sync alive. This is + the "make the idle timeout fire on no-download-progress" ask, scoped safely. + +The refusal surfaces as `NegFrame.Err` → `isOverflow` is false (no +`too many`/`too large`/`max_sync_events`) → `ReconcileOutcome.Failed` → +`NegentropySyncException(UNAVAILABLE)`. `negentropySync` throws promptly; +`negentropySyncOrFetch` catches it and pages the same filter (both primal and +purplepag.es answer ordinary REQs fine, so paging delivers the events). + +## Tests + +- `NegentropyRejectionFallbackTest` (offline, deterministic): a scripted fake + relay answers `NEG-OPEN` with each observed `NOTICE`; asserts `negentropySync` + throws `UNAVAILABLE` fast and `negentropySyncOrFetch` sets `pagedFallback`. +- `NegentropyStallRepro` (gated live): end-to-end proof against the real relays. + +## Not changed / follow-ups + +- The keep-alive subscription filter `Filter(ids=[f*64])` is `CLOSED` by relays + that require a `kinds` (purplepag.es). Harmless now that the reconcile fails + fast and unsubscribes it, but a keep-alive that every relay accepts would be + tidier. +- A relay that refuses via an *unrecognized* signal (neither NEG-ERR, nor a + 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. diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt index fc03bc762c..adc4f1661f 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt @@ -27,7 +27,9 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnection import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.ClosedMessage import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer @@ -814,12 +816,23 @@ private suspend fun INostrClient.reconcileStreaming( // the next one once we ack, and we ack only after this round's ids are queued. val incoming = Channel(Channel.UNLIMITED) - // Idle watchdog. Bumped on connect and on EVERY message this relay sends — - // including the download REQs' events, since this is a connection-level listener - // that sees all of them — so any progress anywhere in the pipeline pushes the - // reconcile deadline out. Only true silence trips it. + // Idle watchdog. Bumped on connect and on the messages that represent real + // progress on this connection — this session's own NEG frames and the download + // REQs' events/EOSEs (a connection-level listener sees them all) — so any + // progress anywhere in the pipeline pushes the reconcile deadline out. It is + // deliberately NOT bumped by NOTICE/CLOSED error chatter: a relay that keeps + // refusing our subscriptions would otherwise reset the watchdog forever, which + // is exactly how a rejected sync escaped the idle timeout. val clock = IdleClock() + // Have we received a single valid NEG frame for our subId yet? A relay that + // advertises NIP-77 but refuses it at runtime answers our NEG-OPEN with a + // connection-level NOTICE (which carries no subId) rather than a subId-addressed + // NEG-ERR. Before the first NEG frame arrives, such a NOTICE is the answer to + // our NEG-OPEN and must be treated as terminal. Only touched from the relay's + // single reader coroutine. + var sawNegFrame = false + val listener = object : RelayConnectionListener { override fun onConnected( @@ -835,11 +848,48 @@ private suspend fun INostrClient.reconcileStreaming( msgStr: String, msg: Message, ) { - if (relay.url == targetUrl) clock.bump() + if (relay.url != targetUrl) return when (msg) { - is NegMsgMessage -> if (msg.subId == subId) incoming.trySend(NegFrame.Msg(msg.message)) - is NegErrMessage -> if (msg.subId == subId) incoming.trySend(NegFrame.Err(msg.reason)) - else -> Unit + is NegMsgMessage -> { + clock.bump() + if (msg.subId == subId) { + sawNegFrame = true + incoming.trySend(NegFrame.Msg(msg.message)) + } + } + + is NegErrMessage -> { + clock.bump() + if (msg.subId == subId) { + sawNegFrame = true + incoming.trySend(NegFrame.Err(msg.reason)) + } + } + + is ClosedMessage -> + // A CLOSED addressed to our negentropy subscription is a + // terminal rejection of the NEG-OPEN (some relays answer a + // refused negentropy session this way instead of NEG-ERR). + if (msg.subId == subId) incoming.trySend(NegFrame.Err("closed: ${msg.message}")) + + is NoticeMessage -> + // NIP-77 says a relay SHOULD reject with NEG-ERR, but relays + // that advertise NIP-77 yet refuse it at runtime answer with a + // connection-level NOTICE instead (strfry: "ERROR: bad msg: + // negentropy disabled"; purplepag.es: "failed to parse + // envelope: unknown envelope label"). A NOTICE has no subId, so + // we bind it to this session by phase + wording: before the + // first valid NEG frame, a negentropy-looking NOTICE is the + // answer to our NEG-OPEN. Surface it as terminal so the caller + // fails over (paging) instead of hanging until the socket drops. + if (!sawNegFrame && isNegentropyRejectionNotice(msg.message)) { + incoming.trySend(NegFrame.Err("notice: ${msg.message}")) + } + + else -> + // EVENT/EOSE from the download REQs (and any other framing on + // this connection) = real progress; keep the reconcile alive. + clock.bump() } } @@ -955,6 +1005,25 @@ private fun isOverflow(reason: String): Boolean = reason.contains("too large", ignoreCase = true) || reason.contains("max_sync_events", ignoreCase = true) +/** + * A relay that advertises NIP-77 but refuses it at runtime signals the refusal with + * a connection-level `NOTICE` (which carries no subId) rather than a subId-addressed + * `NEG-ERR`. Observed against public relays that all list NIP-77 in NIP-11: + * - strfry with negentropy off: `"ERROR: bad msg: negentropy disabled"` + * - purplepag.es (no NEG envelope): `"failed to parse envelope: unknown envelope label"` + * + * 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. + */ +private 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) + /** * One `REQ` for [batch] ids; collects the matching events and returns them on * `EOSE`/close/timeout. All events for a single relay arrive on its one reader diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NegentropyRejectionFallbackTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NegentropyRejectionFallbackTest.kt new file mode 100644 index 0000000000..e27930325f --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NegentropyRejectionFallbackTest.kt @@ -0,0 +1,140 @@ +/* + * 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 + +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySync +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySyncOrFetch +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertTrue + +/** + * Regression guard for the primal.net / purplepag.es negentropy stall + * (`quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md`). + * + * These relays advertise NIP-77 in NIP-11 but refuse it at runtime, answering + * `NEG-OPEN` with a connection-level `NOTICE` (which carries no subId) instead of + * a subId-addressed `NEG-ERR`: + * - strfry, negentropy off: `"ERROR: bad msg: negentropy disabled"` + * - purplepag.es: `"failed to parse envelope: unknown envelope label"` + * + * Before the fix the reconcile driver only reacted to `NEG-MSG`/`NEG-ERR` for its + * exact subId, so the `NOTICE` was dropped and the call blocked until an external + * wall-clock timeout. These tests drive the same exchange against a scripted fake + * relay and assert the client now fails fast: [negentropySync] throws and + * [negentropySyncOrFetch] falls back to paging. + */ +class NegentropyRejectionFallbackTest { + private val url = NormalizedRelayUrl("wss://reject.example.com") + + /** `["REQ","",…]` → ``; also matches NEG-OPEN's subId slot. */ + 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. + */ + private inner class ScriptedRelay( + val replyToNegOpen: String, + ) : WebsocketBuilder { + val io = Executors.newSingleThreadScheduledExecutor() + + override fun build( + url: NormalizedRelayUrl, + out: WebSocketListener, + ): WebSocket = + object : WebSocket { + override fun needsReconnect() = false + + override fun connect() { + io.schedule({ out.onOpen(10, false) }, 5, TimeUnit.MILLISECONDS) + } + + override fun disconnect() {} + + override fun send(msg: String): Boolean { + io.schedule({ + when { + // 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) + else -> Unit + } + }, 5, TimeUnit.MILLISECONDS) + return true + } + } + + fun shutdown() = io.shutdownNow() + } + + private fun negOpenRejectedBy(notice: String) { + val relay = ScriptedRelay(notice) + val client = NostrClient(relay) + try { + runBlocking { + // negentropySync must fail fast (throw), not hang. + val thrown = + assertFailsWith { + withTimeout(8_000) { + client.negentropySync(url, Filter(kinds = listOf(0))) { } + } + } + assertTrue( + thrown.reason == NegentropySyncException.Reason.UNAVAILABLE, + "a NOTICE rejection should be UNAVAILABLE, was ${thrown.reason}", + ) + + // negentropySyncOrFetch must transparently fall back to paging. + val result = + withTimeout(8_000) { + client.negentropySyncOrFetch(url, Filter(kinds = listOf(0))) { } + } + assertTrue(result.pagedFallback, "expected paging fallback after NOTICE rejection") + assertEquals(0, result.downloaded) + } + } finally { + client.close() + relay.shutdown() + } + } + + @Test + fun strfryNegentropyDisabledFallsBackToPaging() = negOpenRejectedBy("[\"NOTICE\",\"ERROR: bad msg: negentropy disabled\"]") + + @Test + fun purplePagesUnknownEnvelopeFallsBackToPaging() = negOpenRejectedBy("[\"NOTICE\",\"failed to parse envelope: unknown envelope label\"]") +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyStallRepro.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyStallRepro.kt new file mode 100644 index 0000000000..04d6338582 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyStallRepro.kt @@ -0,0 +1,254 @@ +/* + * 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.negentropySyncOrFetch +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener +import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder +import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull +import okhttp3.OkHttpClient +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong +import kotlin.test.Test + +/** + * Isolated repro for the client-side NIP-77 negentropy stall against + * primal.net / purplepag.es (works against relay.ditto.pub). + * + * Gated: no-ops unless NEG_STALL_REPRO is set. + * + * NEG_STALL_REPRO=1 ./gradlew :quartz:jvmTest --tests "*.NegentropyStallRepro" --info + */ +class NegentropyStallRepro { + companion object { + const val WALL_CLOCK_MS = 60_000L + } + + /** Live, line-flushed log so progress is visible even under Gradle stdout capture. */ + private val logFile = System.getenv("NEG_STALL_LOG")?.let { java.io.File(it) } + + private fun log(line: String) { + println(line) + logFile?.appendText(line + "\n") + } + + /** A tag for a raw nostr frame so the log reads at a glance. */ + private fun tag(frame: String): String { + val head = frame.take(12) + return when { + head.contains("NEG-OPEN") -> "NEG-OPEN" + head.contains("NEG-MSG") -> "NEG-MSG" + head.contains("NEG-ERR") -> "NEG-ERR" + head.contains("NEG-CLOSE") -> "NEG-CLOSE" + head.contains("\"REQ\"") || head.startsWith("[\"REQ\"") -> "REQ" + head.contains("\"EVENT\"") || head.startsWith("[\"EVENT\"") -> "EVENT" + head.contains("\"EOSE\"") -> "EOSE" + head.contains("\"CLOSE\"") -> "CLOSE" + head.contains("\"CLOSED\"") -> "CLOSED" + head.contains("\"NOTICE\"") -> "NOTICE" + head.contains("\"COUNT\"") -> "COUNT" + else -> "?" + } + } + + /** + * Wraps a [WebsocketBuilder] to log every frame in both directions and + * tally per-type counts, so we can see the exact sequence and where it + * stops making progress. + */ + private class LoggingBuilder( + val delegate: WebsocketBuilder, + val sentCounts: ConcurrentHashMap, + val recvCounts: ConcurrentHashMap, + val negMsgBytesIn: AtomicLong, + val negMsgBytesOut: AtomicLong, + val verbose: Boolean, + val tagger: (String) -> String, + val log: (String) -> Unit, + ) : WebsocketBuilder { + override fun build( + url: NormalizedRelayUrl, + out: WebSocketListener, + ): WebSocket { + val loggingOut = + object : WebSocketListener { + override fun onOpen( + pingMillis: Int, + usingCompression: Boolean, + ) { + log(" [<-open] ${url.url} ping=${pingMillis}ms deflate=$usingCompression") + out.onOpen(pingMillis, usingCompression) + } + + override fun onMessage(text: String) { + val t = tagger(text) + recvCounts.getOrPut(t) { AtomicInteger() }.incrementAndGet() + if (t == "NEG-MSG") negMsgBytesIn.addAndGet(text.length.toLong()) + if (verbose && t != "EVENT") { + log(" [<-$t] len=${text.length} ${text.take(80)}") + } + out.onMessage(text) + } + + override fun onClosed( + code: Int, + reason: String, + ) { + log(" [<-closed] ${url.url} code=$code reason=$reason") + out.onClosed(code, reason) + } + + override fun onFailure( + t: Throwable, + code: Int?, + errorMessage: String?, + ) { + log(" [<-failure] ${url.url} code=$code msg=$errorMessage err=${t.message}") + out.onFailure(t, code, errorMessage) + } + } + + val socket = delegate.build(url, loggingOut) + return object : WebSocket { + override fun needsReconnect() = socket.needsReconnect() + + override fun connect() = socket.connect() + + override fun disconnect() = socket.disconnect() + + override fun send(msg: String): Boolean { + val t = tagger(msg) + sentCounts.getOrPut(t) { AtomicInteger() }.incrementAndGet() + if (t == "NEG-MSG") negMsgBytesOut.addAndGet(msg.length.toLong()) + if (verbose && t != "EVENT") { + log(" [->$t] len=${msg.length} ${msg.take(80)}") + } + return socket.send(msg) + } + } + } + } + + private fun run( + label: String, + relay: String, + filter: Filter, + httpClient: OkHttpClient, + verbose: Boolean, + ) { + log("\n=== $label -> $relay filter=$filter ===") + + val sent = ConcurrentHashMap() + val recv = ConcurrentHashMap() + val negIn = AtomicLong(0) + val negOut = AtomicLong(0) + + val builder = + LoggingBuilder( + BasicOkHttpWebSocket.Builder { httpClient }, + sent, + recv, + negIn, + negOut, + verbose, + ::tag, + ::log, + ) + + val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + val client = NostrClient(builder, scope) + + var lastProgress = -1L + val events = AtomicInteger(0) + val startMs = System.currentTimeMillis() + + runBlocking { + val result = + withTimeoutOrNull(WALL_CLOCK_MS) { + client.negentropySyncOrFetch( + relay = relay, + filter = filter, + localEntries = emptyList(), + onProgress = { need, downloaded -> + // Throttle to one line/sec so an endless reconcile + // doesn't flood the log. + val now = System.currentTimeMillis() + if (now - lastProgress > 1000) { + lastProgress = now + log(" [progress] need=$need downloaded=$downloaded (recv NEG-MSG=${recv["NEG-MSG"]?.get() ?: 0})") + } + }, + onEvent = { events.incrementAndGet() }, + ) + } + + val took = System.currentTimeMillis() - startMs + if (result == null) { + log(" RESULT: *** STALLED *** (externally timed out after ${took}ms)") + } else { + log(" RESULT: returned in ${took}ms downloaded=${result.downloaded} pagedFallback=${result.pagedFallback}") + } + } + + log(" events delivered: ${events.get()}") + log(" frames SENT: ${sent.entries.sortedBy { it.key }.joinToString { "${it.key}=${it.value.get()}" }}") + log(" frames RECV: ${recv.entries.sortedBy { it.key }.joinToString { "${it.key}=${it.value.get()}" }}") + log(" NEG-MSG bytes: out=${negOut.get()} in=${negIn.get()}") + + client.close() + } + + @Test + fun reproduce() { + if (System.getenv("NEG_STALL_REPRO") == null && System.getProperty("negStallRepro") == null) { + println("NegentropyStallRepro skipped. Set NEG_STALL_REPRO=1 to run against live relays.") + return + } + + val httpClient = + OkHttpClient + .Builder() + .connectTimeout(15, TimeUnit.SECONDS) + .readTimeout(120, TimeUnit.SECONDS) + .pingInterval(30, TimeUnit.SECONDS) + .build() + + val verbose = System.getenv("NEG_STALL_VERBOSE") != null + + // Control: known-good relay. + run("DITTO (control, expected to work)", "wss://relay.ditto.pub", Filter(kinds = listOf(0)), httpClient, verbose) + + // The two stalls. + run("PRIMAL (expected to stall)", "wss://relay.primal.net", Filter(kinds = listOf(0)), httpClient, verbose) + run("PURPLEPAGES (expected to stall)", "wss://purplepag.es", Filter(kinds = listOf(0)), httpClient, verbose) + } +} From 30b9e589d43d07ec186d737c89bf11f7ea34ee2b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 27 Jul 2026 19:13:14 +0000 Subject: [PATCH 2/3] 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 Claude-Session: https://claude.ai/code/session_01B6ZVTixuc1ef8eGB6MQHRn --- ...07-27-negentropy-notice-rejection-stall.md | 49 ++++++++ .../client/accessories/NegentropyStoreSync.kt | 7 +- .../NostrClientNegentropySyncExt.kt | 106 ++++++++++++++---- .../NegentropyErrorClassificationTest.kt | 74 ++++++++++++ .../client/NegentropyRejectionFallbackTest.kt | 50 ++++++--- 5 files changed, 249 insertions(+), 37 deletions(-) create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyErrorClassificationTest.kt diff --git a/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md b/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md index 3ea486849f..064d4b0d69 100644 --- a/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md +++ b/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md @@ -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. diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyStoreSync.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyStoreSync.kt index 0d2ce9df08..540411a522 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyStoreSync.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyStoreSync.kt @@ -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>(config.downloadWorkers * 2) - val haveBatches = Channel>(Channel.UNLIMITED) + val haveBatches = Channel>((config.downloadWorkers * 2).coerceAtLeast(2)) val downloaders = List(config.downloadWorkers.coerceAtLeast(1)) { diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt index adc4f1661f..0e93bcef1f 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt @@ -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 diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyErrorClassificationTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyErrorClassificationTest.kt new file mode 100644 index 0000000000..65d647de39 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyErrorClassificationTest.kt @@ -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")) + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NegentropyRejectionFallbackTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NegentropyRejectionFallbackTest.kt index e27930325f..230674d4d0 100644 --- a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NegentropyRejectionFallbackTest.kt +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NegentropyRejectionFallbackTest.kt @@ -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()}", + ) + } } From 9de8eae5fdc1e083ee2f674c751c3a80cbc850b8 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 27 Jul 2026 19:23:46 +0000 Subject: [PATCH 3/3] test: broad live NIP-77 validation across 30 public relays MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds NegentropyMultiRelayLiveTest (gated NEG_MULTI=1): runs negentropySyncOrFetch against 30 reachable public relays and asserts none hangs. Live run: 0 hangs, 0 errors — 10 reconciled via native negentropy, 20 fell over to paging, across 9+ relay softwares (strfry, ditto, purplepag.es, nostr.wine, nostr-rs-relay, NFDB, rockstr, wot-relay, nostrcheck). Confirms both fallback paths: the NOTICE fast-path (~1-4s) for relays whose refusal names negentropy / unknown-envelope, and the idle-watchdog backstop (~20s) for the rest (e.g. damus silently ignores NEG-OPEN, snort answers "Unknown message type: NEG-OPEN") — now reliable because NOTICE/CLOSED no longer reset the watchdog. Matrix recorded in the plan doc. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01B6ZVTixuc1ef8eGB6MQHRn --- ...07-27-negentropy-notice-rejection-stall.md | 27 ++ .../prodbench/NegentropyMultiRelayLiveTest.kt | 235 ++++++++++++++++++ 2 files changed, 262 insertions(+) create mode 100644 quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyMultiRelayLiveTest.kt diff --git a/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md b/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md index 064d4b0d69..54543c91e6 100644 --- a/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md +++ b/quartz/plans/2026-07-27-negentropy-notice-rejection-stall.md @@ -112,6 +112,33 @@ purplepag.es answer ordinary REQs fine, so paging delivers the events). watchdog — which now fires correctly because the refusal chatter no longer resets it. +## Live validation across 30 public relays + +`NegentropyMultiRelayLiveTest` (gated `NEG_MULTI=1`) runs `negentropySyncOrFetch` +(`kinds:[0]`, `maxEvents=100`, `idleTimeoutMs=20s`) against 30 reachable relays and +fails if any HANGs. Result: **0 hangs, 0 errors** — 10 reconciled via native +negentropy, 20 fell over to paging. Nine+ relay softwares (strfry, ditto, +purplepag.es, nostr.wine, nostr-rs-relay, NFDB, rockstr, wot-relay, nostrcheck). + +Both fallback mechanisms fire as designed: + +- **NOTICE fast-path (~1-4 s):** `relay.primal.net` (`negentropy disabled`), + `purplepag.es` / `wot.utxo.one` (`unknown envelope`), `relay.nostrplebs.com` / + `relay.0xchat.com` (`negentropy error …`). +- **Idle-watchdog backstop (~20 s):** relays whose refusal wording the (deliberately + narrow) matcher skips still fail over because NOTICE/CLOSED no longer bump the + clock — `relay.damus.io` (silently ignores NEG-OPEN, no error at all), + `relay.snort.social` (`Unknown message type: NEG-OPEN`), `relay.momostr.pink`, + `relay.wellorder.net` (`could not parse command`), `relay.nostrcheck.me`, … +- **Overflow window-split (native):** `nos.lol`, `relay.ditto.pub`, + `nostr.oxtr.dev` returned `NEG-ERR "too many … records/results"` → split → + downloaded natively. + +Takeaway: the narrow NOTICE matcher plus the un-defeated idle watchdog is the right +split — the fast-path speeds up the relays whose wording is unambiguous, and the +watchdog safely (if more slowly) catches everything else, with no false aborts on +the 10 relays that genuinely speak NIP-77. + ## Follow-up audit (same PR) A read-through of the whole negentropy accessories package surfaced a few more diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyMultiRelayLiveTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyMultiRelayLiveTest.kt new file mode 100644 index 0000000000..1c72b53690 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/NegentropyMultiRelayLiveTest.kt @@ -0,0 +1,235 @@ +/* + * 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.negentropySyncOrFetch +import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket +import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage +import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withPermit +import kotlinx.coroutines.withTimeoutOrNull +import okhttp3.OkHttpClient +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.Test +import kotlin.test.fail + +/** + * Broad live validation of the NIP-77 stall fix: run `negentropySyncOrFetch` against + * as many public relays as we can reach and assert NONE hangs — every relay must + * either reconcile via negentropy or fall over to paging, and the suspend fun must + * return without an external wall-clock rescue. + * + * Gated (opens live sockets to ~35 relays): + * NEG_MULTI=1 ./gradlew :quartz:jvmTest --tests "*.NegentropyMultiRelayLiveTest" --info + * + * The wall-clock [HARD_TIMEOUT_MS] is only a safety net set well above [IDLE_MS]: + * with the fix a refusing relay trips the idle watchdog (or the notice/closed + * fast-path) and returns; a relay that only returns because the wall clock fired is + * the bug this guards against and fails the test. + */ +class NegentropyMultiRelayLiveTest { + companion object { + const val MAX_EVENTS = 100 + const val IDLE_MS = 20_000L + const val HARD_TIMEOUT_MS = 90_000L + const val PARALLELISM = 6 + + val RELAYS = + listOf( + "wss://relay.damus.io", + "wss://nos.lol", + "wss://relay.primal.net", + "wss://relay.nostr.band", + "wss://nostr.wine", + "wss://purplepag.es", + "wss://relay.ditto.pub", + "wss://nostr.mom", + "wss://relay.nostr.bg", + "wss://offchain.pub", + "wss://nostr.oxtr.dev", + "wss://relay.nostrplebs.com", + "wss://nostr21.com", + "wss://relay.mostr.pub", + "wss://nostr.bitcoiner.social", + "wss://relay.nostrcheck.me", + "wss://nostr-pub.wellorder.net", + "wss://relay.snort.social", + "wss://relayable.org", + "wss://nostr.land", + "wss://relay.momostr.pink", + "wss://relay.0xchat.com", + "wss://nostr.fmt.wiz.biz", + "wss://wot.utxo.one", + "wss://nostr.data.haus", + "wss://eden.nostr.land", + "wss://atlas.nostr.land", + "wss://relay.nostr.com.au", + "wss://relay.wellorder.net", + "wss://relay.fountain.fm", + ) + } + + private class RelayProbe { + val negMsgs = AtomicInteger(0) + val negErrs = AtomicInteger(0) + + @Volatile var firstNegErr: String? = null + + @Volatile var firstNotice: String? = null + } + + private class Outcome( + val relay: String, + val label: String, + val downloaded: Int, + val ms: Long, + val detail: String, + ) + + @Test + fun everyReachableRelayReturnsNoHang() { + if (System.getenv("NEG_MULTI") == null && System.getProperty("negMulti") == null) { + println("NegentropyMultiRelayLiveTest skipped. Set NEG_MULTI=1 to run against live relays.") + return + } + + val httpClient = + OkHttpClient + .Builder() + .connectTimeout(15, TimeUnit.SECONDS) + .readTimeout(60, TimeUnit.SECONDS) + .pingInterval(30, TimeUnit.SECONDS) + .build() + + val outcomes = + runBlocking { + val gate = Semaphore(PARALLELISM) + RELAYS + .map { relay -> + async(Dispatchers.IO) { + gate.withPermit { probeRelay(relay, httpClient) } + } + }.awaitAll() + } + + println("\n================ NIP-77 multi-relay results ================") + println("%-26s %-14s %10s %8s %s".format("relay", "outcome", "downloaded", "ms", "detail")) + outcomes.sortedBy { it.label }.forEach { + println( + "%-26s %-14s %10d %8d %s".format( + it.relay.removePrefix("wss://"), + it.label, + it.downloaded, + it.ms, + it.detail, + ), + ) + } + + val byLabel = outcomes.groupingBy { it.label }.eachCount() + println("\nsummary: $byLabel") + + val hangs = outcomes.filter { it.label == "HANG" } + if (hangs.isNotEmpty()) { + fail("These relays HUNG (no return within ${HARD_TIMEOUT_MS}ms) — the stall is not fixed for them:\n" + hangs.joinToString("\n") { " ${it.relay}: ${it.detail}" }) + } + } + + private suspend fun probeRelay( + relay: String, + httpClient: OkHttpClient, + ): Outcome { + val probe = RelayProbe() + val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + val client = NostrClient(BasicOkHttpWebSocket.Builder { httpClient }, scope) + + val listener = + object : RelayConnectionListener { + override fun onIncomingMessage( + relay: IRelayClient, + msgStr: String, + msg: Message, + ) { + when (msg) { + is NegMsgMessage -> probe.negMsgs.incrementAndGet() + is NegErrMessage -> { + probe.negErrs.incrementAndGet() + if (probe.firstNegErr == null) probe.firstNegErr = msg.reason + } + is NoticeMessage -> if (probe.firstNotice == null) probe.firstNotice = msg.message + else -> Unit + } + } + } + client.addConnectionListener(listener) + + val started = System.currentTimeMillis() + return try { + val downloaded = AtomicInteger(0) + val result = + withTimeoutOrNull(HARD_TIMEOUT_MS) { + client.negentropySyncOrFetch( + relay = relay, + filter = Filter(kinds = listOf(0)), + maxEvents = MAX_EVENTS, + idleTimeoutMs = IDLE_MS, + localEntries = emptyList(), + onEvent = { downloaded.incrementAndGet() }, + ) + } + val ms = System.currentTimeMillis() - started + val wire = + "negMsg=${probe.negMsgs.get()} negErr=${probe.negErrs.get()}" + + (probe.firstNegErr?.let { " err=\"${it.take(48)}\"" } ?: "") + + (probe.firstNotice?.let { " notice=\"${it.take(48)}\"" } ?: "") + + when { + result == null -> + Outcome(relay, "HANG", downloaded.get(), ms, "no return; $wire") + result.pagedFallback -> + Outcome(relay, "OK-PAGED", result.downloaded, ms, "fell back to paging; $wire") + result.downloaded > 0 -> + Outcome(relay, "OK-NEG", result.downloaded, ms, "native negentropy; $wire") + else -> + Outcome(relay, "OK-EMPTY", result.downloaded, ms, "returned, 0 events; $wire") + } + } catch (e: Throwable) { + val ms = System.currentTimeMillis() - started + Outcome(relay, "ERROR", 0, ms, "${e::class.simpleName}: ${e.message?.take(60)}") + } finally { + client.close() + } + } +}