From 264866bcbbbc8a6726452b965dcf5f5d6899130f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 27 Jul 2026 18:12:01 +0000 Subject: [PATCH] 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) + } +}