mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-10-05 11:18:24 +00:00
Merge pull request #3747 from vitorpamplona/claude/quartz-negentropy-stalls-6g9ukr
Fix NIP-77 negentropy stall against relays that refuse via NOTICE
This commit is contained in:
Vitor Pamplona
@@ -0,0 +1,189 @@
|
||||
# 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.
|
||||
|
||||
## 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
|
||||
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.
|
||||
+6
-1
@@ -175,8 +175,13 @@ class NegentropyStoreSync(
|
||||
try {
|
||||
coroutineScope {
|
||||
// needIds = relay has, store lacks; haveIds = store has, relay lacks.
|
||||
// Both are bounded so a slow consumer back-pressures the reconcile
|
||||
// instead of buffering the whole residual set. haveBatches drains
|
||||
// through the single, network-bound uploader (publish + OK wait) far
|
||||
// slower than the reconcile produces have-ids, so leaving it UNLIMITED
|
||||
// let a first push of a large local store queue O(local-set) ids.
|
||||
val needBatches = Channel<List<HexKey>>(config.downloadWorkers * 2)
|
||||
val haveBatches = Channel<List<HexKey>>(Channel.UNLIMITED)
|
||||
val haveBatches = Channel<List<HexKey>>((config.downloadWorkers * 2).coerceAtLeast(2))
|
||||
|
||||
val downloaders =
|
||||
List(config.downloadWorkers.coerceAtLeast(1)) {
|
||||
|
||||
+154
-25
@@ -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
|
||||
@@ -147,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(
|
||||
@@ -285,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.
|
||||
*/
|
||||
@@ -484,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 {
|
||||
@@ -524,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 ->
|
||||
@@ -814,12 +851,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 +883,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()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -935,26 +1020,61 @@ 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)
|
||||
|
||||
/**
|
||||
* 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 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.
|
||||
*/
|
||||
internal fun isNegentropyRejectionNotice(reason: String): Boolean =
|
||||
reason.contains("negentropy", ignoreCase = true) ||
|
||||
reason.contains("unknown envelope", 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
|
||||
@@ -1032,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
|
||||
|
||||
|
||||
+74
@@ -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"))
|
||||
}
|
||||
}
|
||||
+164
@@ -0,0 +1,164 @@
|
||||
/*
|
||||
* 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 java.util.concurrent.atomic.AtomicInteger
|
||||
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 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: (subId: String) -> String,
|
||||
) : WebsocketBuilder {
|
||||
val io = Executors.newSingleThreadScheduledExecutor()
|
||||
val negOpens = AtomicInteger(0)
|
||||
|
||||
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.
|
||||
msg.startsWith("[\"NEG-OPEN\"") -> {
|
||||
negOpens.incrementAndGet()
|
||||
subIdOf(msg)?.let { out.onMessage(replyToNegOpen(it)) }
|
||||
}
|
||||
else -> Unit
|
||||
}
|
||||
}, 5, TimeUnit.MILLISECONDS)
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
fun shutdown() = io.shutdownNow()
|
||||
}
|
||||
|
||||
private fun negOpenRejectedBy(reply: (subId: String) -> String): ScriptedRelay {
|
||||
val relay = ScriptedRelay(reply)
|
||||
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 runtime negentropy refusal 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 the refusal")
|
||||
assertEquals(0, result.downloaded)
|
||||
}
|
||||
} finally {
|
||||
client.close()
|
||||
relay.shutdown()
|
||||
}
|
||||
return relay
|
||||
}
|
||||
|
||||
@Test
|
||||
fun strfryNegentropyDisabledFallsBackToPaging() {
|
||||
negOpenRejectedBy { "[\"NOTICE\",\"ERROR: bad msg: negentropy disabled\"]" }
|
||||
}
|
||||
|
||||
@Test
|
||||
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()}",
|
||||
)
|
||||
}
|
||||
}
|
||||
+235
@@ -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()
|
||||
}
|
||||
}
|
||||
}
|
||||
+254
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user