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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B6ZVTixuc1ef8eGB6MQHRn
2026-08-09 16:14:40 +00:00 · 2026-07-27 18:12:01 +00:00
parent 855435fcc1
commit 264866bcbb
4 changed files with 584 additions and 8 deletions
@@ -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.
@@ -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<NegFrame>(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
@@ -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","<subId>",…]` → `<subId>`; 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<NegentropySyncException> {
+                        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\"]")
+}
@@ -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<String, AtomicInteger>,
+        val recvCounts: ConcurrentHashMap<String, AtomicInteger>,
+        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<String, AtomicInteger>()
+        val recv = ConcurrentHashMap<String, AtomicInteger>()
+        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)
+    }
+}