laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-08 23:54:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3434 from vitorpamplona/claude/quartz-negentropy-sync-accessory-r92bue

Browse Source 
Add NIP-77 negentropy sync with streaming and windowing support
This commit is contained in:
Vitor Pamplona
2026-07-01 10:24:12 -04:00
committed by GitHub
parent ccba020012 322e33678a
commit 2cb058b323
9 changed files with 1574 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt
+12
View File
@@ -83,6 +83,18 @@ interface INostrClient : AutoCloseable {
fun removeConnectionListener(listener: RelayConnectionListener)
/**
* Returns the [IRelayClient] for [url], creating and registering it in the
* connection pool if it is not there yet.
*
* Most callers should never need this — [subscribe]/[count]/[publish] manage
* the pool for you. It exists for accessories that must drive a single relay
* directly, such as NIP-77 negentropy (which sends `NEG-OPEN` and walks the
* reconciliation rounds on one connection). The default implementation throws;
* only a real pool-backed client can hand out relay clients.
*/
fun getOrCreateRelay(url: NormalizedRelayUrl): IRelayClient = throw UnsupportedOperationException("This INostrClient does not expose relay clients")
fun getReqFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>?
fun getCountFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>?
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt
+2
View File
@@ -342,6 +342,8 @@ class NostrClient(
listeners.forEach { it.onCannotConnect(relay, errorMessage) }
}
override fun getOrCreateRelay(url: NormalizedRelayUrl): IRelayClient = relayPool.getOrCreateRelay(url)
override fun addConnectionListener(listener: RelayConnectionListener) {
listeners = listeners.plus(listener)
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropySyncException.kt
+69
View File
@@ -0,0 +1,69 @@
/*
* 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 com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
/**
* Thrown by [negentropySync] when a relay's matched set cannot be reconciled
* through NIP-77 for a given [window].
*
* The accessory does NOT silently fall back to plain paging — that is a heavier,
* non-delta transport and the choice belongs to the caller. Catch this and decide:
* page the filter yourself with [fetchAllPages], try another relay, narrow the
* filter, or give up. For the common "try negentropy, else page" shape use
* [negentropySyncOrFetch], which does exactly that (with id-dedup) for you.
*
* [window] is the specific `created_at` slice that failed. When negentropy fails
* on the very first reconcile (e.g. the relay does not speak NIP-77) it equals the
* filter you passed; after windowing it is a sub-range. Note that events from
* windows that DID reconcile before this failure may already have been delivered
* to your `onEvent`, so dedupe by event id if you then page the whole filter.
*
* @property relay the relay that could not reconcile.
* @property window the filter slice that failed.
* @property reason machine-readable category — branch on this to recover.
* @property detail the underlying specifics (a relay's `NEG-ERR` text, `timeout`, …).
*/
class NegentropySyncException(
val relay: NormalizedRelayUrl,
val window: Filter,
val reason: Reason,
val detail: String,
) : Exception("NIP-77 sync of $relay failed ($reason): $detail") {
enum class Reason {
/**
* The relay caps negentropy below the matched set and even a minimal
* `created_at` window still exceeds that cap (strfry's `max_sync_events`),
* so negentropy cannot enumerate the window at all. Paging is the only way
* to get these events.
*/
OVER_MAX_SYNC_EVENTS,
/**
* The relay did not complete reconciliation: no NIP-77 support, a
* non-overflow `NEG-ERR`, a disconnect, or a timeout. [detail] carries the
* specifics.
*/
UNAVAILABLE,
}
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientFetchAllPagesExt.kt
+13
View File
@@ -64,6 +64,19 @@ suspend fun INostrClient.fetchAllPages(
// Track how many matching events each filter has received so far.
val matchCountPerFilter = IntArray(filters.size)
// One subscription id reused for every page. Each page opens it (with the
// page's `until`), waits for EOSE, then closes it before the next page opens
// it again — so at most one subscription is ever live and the whole download
// occupies a single subscription slot on the connection (relays cap the
// number of concurrent subscriptions per connection, so churning through a
// fresh id per page is wasteful).
//
// Reusing the id is safe because the pool serializes the "send a REQ"
// decision: after each page's EOSE, the pool's auto-resend and this loop's
// unsubscribe+resubscribe can no longer both fire a REQ for the same id (see
// PoolRequests.decideCommandLocked / PoolRequestsConcurrencyTest). Without
// that fix the two raced and produced a duplicate REQ — two EOSEs, or an
// empty page that silently truncated large results.
val subId = newSubId()
while (true) {
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncEventsExt.kt
+141
View File
@@ -0,0 +1,141 @@
/*
* 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 com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.callbackFlow
/**
* Streaming form of [negentropySync]: emits each event **individually** as it
* arrives, then completes when the sync finishes. Nothing is accumulated, so it
* stays O(1) in memory regardless of how many events the relay holds.
*
* Events are buffered with [Channel.UNLIMITED] because [negentropySync] delivers
* them through a non-suspending callback on the relay reader thread: a bounded
* buffer would force the producer to drop events when the collector lags. A slow
* collector therefore lets the buffer grow — apply your own
* [kotlinx.coroutines.flow.buffer]/`conflate`/`collectLatest` downstream if you
* need a different policy. Cancelling the collector cancels the sync and tears
* down its subscriptions via [awaitClose].
*
* See [negentropySync] for the meaning of every parameter.
*/
fun INostrClient.negentropySyncEvents(
relay: NormalizedRelayUrl,
filter: Filter,
maxEvents: Int = 0,
maxConcurrentReqs: Int = 8,
fetchBatch: Int = 500,
idleTimeoutMs: Long = 120_000L,
): Flow<Event> =
callbackFlow {
negentropySync(
relay = relay,
filter = filter,
maxEvents = maxEvents,
maxConcurrentReqs = maxConcurrentReqs,
fetchBatch = fetchBatch,
idleTimeoutMs = idleTimeoutMs,
) { event ->
trySend(event)
}
close()
awaitClose { }
}.buffer(Channel.UNLIMITED)
fun INostrClient.negentropySyncEvents(
relay: String,
filter: Filter,
maxEvents: Int = 0,
maxConcurrentReqs: Int = 8,
fetchBatch: Int = 500,
idleTimeoutMs: Long = 120_000L,
): Flow<Event> =
negentropySyncEvents(
relay = RelayUrlNormalizer.normalize(relay),
filter = filter,
maxEvents = maxEvents,
maxConcurrentReqs = maxConcurrentReqs,
fetchBatch = fetchBatch,
idleTimeoutMs = idleTimeoutMs,
)
/**
* Streaming "try negentropy, else page" — the [Flow] form of
* [negentropySyncOrFetch]. Emits each event individually as it arrives from
* whichever transport delivered it, deduped by id across both phases, then
* completes. Unlike [negentropySyncEvents] it never throws on a relay that can't
* reconcile; it pages instead.
*
* See [negentropySyncEvents] for the buffering/backpressure note and
* [negentropySync] for the meaning of every parameter.
*/
fun INostrClient.negentropySyncOrFetchEvents(
relay: NormalizedRelayUrl,
filter: Filter,
maxEvents: Int = 0,
maxConcurrentReqs: Int = 8,
fetchBatch: Int = 500,
idleTimeoutMs: Long = 120_000L,
): Flow<Event> =
callbackFlow {
negentropySyncOrFetch(
relay = relay,
filter = filter,
maxEvents = maxEvents,
maxConcurrentReqs = maxConcurrentReqs,
fetchBatch = fetchBatch,
idleTimeoutMs = idleTimeoutMs,
) { event ->
trySend(event)
}
close()
awaitClose { }
}.buffer(Channel.UNLIMITED)
fun INostrClient.negentropySyncOrFetchEvents(
relay: String,
filter: Filter,
maxEvents: Int = 0,
maxConcurrentReqs: Int = 8,
fetchBatch: Int = 500,
idleTimeoutMs: Long = 120_000L,
): Flow<Event> =
negentropySyncOrFetchEvents(
relay = RelayUrlNormalizer.normalize(relay),
filter = filter,
maxEvents = maxEvents,
maxConcurrentReqs = maxConcurrentReqs,
fetchBatch = fetchBatch,
idleTimeoutMs = idleTimeoutMs,
)
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt
+726
View File
@@ -0,0 +1,726 @@
/*
* 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 com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
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.Message
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession
import com.vitorpamplona.quartz.utils.TimeUtils
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.joinAll
import kotlinx.coroutines.launch
import kotlinx.coroutines.withTimeoutOrNull
import kotlin.concurrent.Volatile
import kotlin.coroutines.coroutineContext
import kotlin.math.min
import kotlin.time.TimeSource
/**
* Outcome of a successful [negentropySync] run.
*
* @property needCount ids the relay had that we lacked (i.e. everything that
* matched [Filter] on the relay — this sync always reconciles against an empty
* local set, so it downloads the full matched set).
* @property haveCount ids we had that the relay lacked. Always `0` here because
* the local set is empty; kept so the result mirrors a full NIP-77 reconcile.
* @property downloaded distinct events actually delivered through `onEvent`.
* @property windows number of `created_at` windows the matched set was split
* into (`1` when the relay reconciled the whole filter in one shot).
*/
class NegentropySyncResult(
val needCount: Int,
val haveCount: Int,
val downloaded: Int,
val windows: Int,
)
/**
* Downloads every event a single [relay] holds matching [filter], delivering each
* one (deduped by id) through [onEvent]. A high-level wrapper over NIP-77
* negentropy that hides the parts that make the raw protocol painful to use:
*
* 1. Reconciles the relay's matched set against an empty local set, **streaming**
* the ids the relay has straight into the download pipeline as each NIP-77
* round arrives — the full id list is never materialised.
* 2. Downloads those ids through at most [maxConcurrentReqs] concurrent `REQ`
* subscriptions of [fetchBatch] ids each, refilling as each `EOSE` arrives. The
* reconciliation, the id queue and event delivery are all back-pressured, so a
* slow consumer throttles the whole chain and **peak memory is bounded by the
* pipeline depth, not by the window size** — a multi-million-event window
* streams through in roughly constant memory. No id/event dedup set is held:
* NIP-77 yields a distinct id set, so each event is requested (and returned)
* exactly once.
* 3. Handles the relay-side cap on negentropy (strfry's `max_sync_events`,
* observed as `NEG-ERR … "blocked: too many query results"`): the [filter] is
* split by `created_at` windows and each window reconciled on its own, a
* window that still overflows being halved and retried.
*
* This method is negentropy-only. It does NOT silently fall back to plain paging:
* if a window genuinely cannot be reconciled — a minimal `created_at` window still
* over the relay's cap, or a relay that does not speak NIP-77 / drops the session /
* times out — it throws [NegentropySyncException] so the caller chooses what to do.
* For the common "try negentropy, else page" shape, use [negentropySyncOrFetch].
*
* Scope is controlled entirely by [filter] — narrow it (kinds, authors, `since`,
* tags, …) to download a slice instead of everything. [maxEvents] additionally caps
* the delivered set.
*
* Coroutine-cancellable: on completion, cancel, reaching [maxEvents], or a thrown
* [NegentropySyncException], all `REQ` subscriptions are unsubscribed and the
* negentropy session is closed and its listener removed, so nothing leaks.
*
* @throws NegentropySyncException when a window cannot be reconciled via NIP-77.
*
* @param relay the relay to sync from.
* @param filter what to download. A single filter (NEG-OPEN is single-filter).
* @param maxEvents stop after delivering this many distinct events. `0` = unlimited.
* @param maxConcurrentReqs upper bound on simultaneously-open download `REQ`s. Keep
* it at or below the relay's per-connection subscription cap.
* @param fetchBatch ids per download `REQ`.
* @param idleTimeoutMs the idle watchdog: the maximum time the relay may go
* **completely silent** before the sync gives up. It is NOT a per-round deadline —
* it **resets on every message the relay sends** (each NIP-77 round, every download
* `EOSE`/event) and on connect. So a genuinely slow but progressing sync runs for as
* long as it needs: only true silence trips it. This matters because the relay
* builds its whole negentropy snapshot before the FIRST round responds — O(matched
* set), a minute or more for a multi-million-event filter — and that first wait is a
* real silence, so keep this comfortably above the largest expected first-round build.
* A dead/half-open socket does NOT depend on this: the WebSocket keep-alive detects
* it and the disconnect is turned into a clean abort. Pass `0` to disable the
* watchdog entirely and run until the socket drops (download batches keep a finite
* internal idle bound regardless, so a single stuck batch can't hang the pipeline).
* @param onProgress optional `(needSoFar, downloaded)` ticks as work proceeds.
* @param onEvent called once per distinct event, on the relay reader thread.
*/
suspend fun INostrClient.negentropySync(
relay: NormalizedRelayUrl,
filter: Filter,
maxEvents: Int = 0,
maxConcurrentReqs: Int = 8,
fetchBatch: Int = 500,
idleTimeoutMs: Long = 120_000L,
onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null,
onEvent: (Event) -> Unit,
): NegentropySyncResult {
var need = 0
var windows = 0
var downloaded = 0
// Pin the relay in the pool's "desired" set for the whole sync. A NEG-OPEN is not
// a REQ, so during a reconcile round (before that window's first download REQ
// exists) the relay would otherwise look unwanted and the pool would disconnect
// it — fatal mid-sync, and frequent when many small windows each have such a gap.
// A never-matching keep-alive subscription holds the connection open without
// delivering anything.
val keepAliveSubId = newSubId()
subscribe(keepAliveSubId, mapOf(relay to listOf(Filter(ids = listOf(KEEP_ALIVE_ID)))), null)
try {
coroutineScope {
// Bounded funnel: every delivered event passes through this one consumer
// (so onEvent + the maxEvents cap run single-threaded) and the bound
// back-pressures the download workers when the consumer can't keep up.
val events = Channel<Event>(DELIVERY_BUFFER)
val producer =
launch {
try {
syncWindow(
relay = relay,
filter = filter,
idleTimeoutMs = idleTimeoutMs,
fetchBatch = fetchBatch,
maxConcurrentReqs = maxConcurrentReqs,
onWindow = { windows++ },
// Only accumulate here; progress is reported from the
// single consumer loop below so the user callback is never
// invoked from two coroutines at once.
onNeed = { need += it },
deliver = { events.send(it) },
)
} finally {
events.close()
}
}
for (event in events) {
downloaded++
onEvent(event)
onProgress?.invoke(need, downloaded)
if (maxEvents in 1..downloaded) break
}
// If we broke out early (cap reached) the producer may still be working —
// stop it. If the producer finished normally this is a no-op.
producer.cancel()
}
} finally {
unsubscribe(keepAliveSubId)
}
return NegentropySyncResult(
needCount = need,
haveCount = 0,
downloaded = downloaded,
windows = windows,
)
}
suspend fun INostrClient.negentropySync(
relay: String,
filter: Filter,
maxEvents: Int = 0,
maxConcurrentReqs: Int = 8,
fetchBatch: Int = 500,
idleTimeoutMs: Long = 120_000L,
onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null,
onEvent: (Event) -> Unit,
): NegentropySyncResult =
negentropySync(
relay = RelayUrlNormalizer.normalize(relay),
filter = filter,
maxEvents = maxEvents,
maxConcurrentReqs = maxConcurrentReqs,
fetchBatch = fetchBatch,
idleTimeoutMs = idleTimeoutMs,
onProgress = onProgress,
onEvent = onEvent,
)
/**
* Result of [negentropySyncOrFetch].
*
* @property downloaded distinct events delivered through `onEvent` (across whichever
* path ran).
* @property pagedFallback `true` if negentropy could not reconcile and the events
* came from [fetchAllPages] instead.
* @property negentropy the negentropy outcome when it succeeded; `null` on fallback.
* @property fallbackCause why negentropy was abandoned; `null` when it succeeded.
*/
class NegentropyOrFetchResult(
val downloaded: Int,
val pagedFallback: Boolean,
val negentropy: NegentropySyncResult?,
val fallbackCause: NegentropySyncException?,
)
/**
* "Try negentropy, else page." Runs [negentropySync] and, if it throws
* [NegentropySyncException] (relay can't reconcile the set — no NIP-77 support, an
* over-cap minimal window, a disconnect, …), transparently falls back to
* [fetchAllPages] over the same [filter].
*
* This is the convenience combinator for the common case where you just want the
* events and don't care which transport delivered them. Events are deduped by id
* across both phases, so anything the negentropy attempt already delivered before
* failing is not delivered again by the paging phase. [maxEvents] is honored across
* both phases.
*
* Use [negentropySync] directly if you want to decide the fallback yourself (try
* another relay, narrow the filter, abort, …) instead of always paging.
*/
suspend fun INostrClient.negentropySyncOrFetch(
relay: NormalizedRelayUrl,
filter: Filter,
maxEvents: Int = 0,
maxConcurrentReqs: Int = 8,
fetchBatch: Int = 500,
idleTimeoutMs: Long = 120_000L,
onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null,
onEvent: (Event) -> Unit,
): NegentropyOrFetchResult {
val seen = HashSet<HexKey>()
var delivered = 0
// Shared dedup + cap across both phases. Returns true if the event was new and
// delivered. Both phases run sequentially, so no concurrent access.
fun accept(event: Event): Boolean {
if ((maxEvents <= 0 || delivered < maxEvents) && seen.add(event.id)) {
delivered++
onEvent(event)
return true
}
return false
}
return try {
val result =
negentropySync(
relay = relay,
filter = filter,
maxEvents = maxEvents,
maxConcurrentReqs = maxConcurrentReqs,
fetchBatch = fetchBatch,
idleTimeoutMs = idleTimeoutMs,
onProgress = onProgress,
) { accept(it) }
NegentropyOrFetchResult(delivered, pagedFallback = false, negentropy = result, fallbackCause = null)
} catch (e: NegentropySyncException) {
// Negentropy couldn't enumerate the set — page the whole filter instead,
// skipping anything the negentropy attempt already delivered. fetchAllPages
// has no "no timeout" mode, so a disabled watchdog maps to a finite page bound.
val pageFilter = if (maxEvents > 0) filter.copy(limit = maxEvents) else filter
val pageTimeoutMs = if (idleTimeoutMs > 0) idleTimeoutMs else DEFAULT_DOWNLOAD_IDLE_MS
fetchAllPages(relay, listOf(pageFilter), pageTimeoutMs) { event ->
if (accept(event)) onProgress?.invoke(delivered, delivered)
}
NegentropyOrFetchResult(delivered, pagedFallback = true, negentropy = null, fallbackCause = e)
}
}
suspend fun INostrClient.negentropySyncOrFetch(
relay: String,
filter: Filter,
maxEvents: Int = 0,
maxConcurrentReqs: Int = 8,
fetchBatch: Int = 500,
idleTimeoutMs: Long = 120_000L,
onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null,
onEvent: (Event) -> Unit,
): NegentropyOrFetchResult =
negentropySyncOrFetch(
relay = RelayUrlNormalizer.normalize(relay),
filter = filter,
maxEvents = maxEvents,
maxConcurrentReqs = maxConcurrentReqs,
fetchBatch = fetchBatch,
idleTimeoutMs = idleTimeoutMs,
onProgress = onProgress,
onEvent = onEvent,
)
/**
* Recursively reconciles [filter] for [relay], splitting by `created_at` windows
* whenever the relay rejects the set as too large, and downloading the ids of each
* window as it resolves. Runs on a single coroutine; [deliver] funnels events out.
*
* Throws [NegentropySyncException] for any window negentropy cannot reconcile (a
* minimal window still over the cap, or an unavailable/erroring relay).
*/
private suspend fun INostrClient.syncWindow(
relay: NormalizedRelayUrl,
filter: Filter,
idleTimeoutMs: Long,
fetchBatch: Int,
maxConcurrentReqs: Int,
onWindow: () -> Unit,
onNeed: (Int) -> Unit,
deliver: suspend (Event) -> Unit,
) {
coroutineContext.ensureActive()
when (val outcome = downloadWindow(relay, filter, idleTimeoutMs, fetchBatch, maxConcurrentReqs, onNeed, deliver)) {
is ReconcileOutcome.Complete -> onWindow()
is ReconcileOutcome.Overflow -> {
val lo = filter.since ?: 0L
val hi = filter.until ?: TimeUtils.now()
if (hi - lo <= MIN_WINDOW_SECONDS) {
// A minimal window that still overflows: negentropy genuinely can't
// enumerate this slice. Surface it — paging is the caller's call.
throw NegentropySyncException(
relay = relay,
window = filter,
reason = NegentropySyncException.Reason.OVER_MAX_SYNC_EVENTS,
detail = "created_at window [$lo, $hi] still exceeds the relay's max_sync_events",
)
} else {
val mid = lo + (hi - lo) / 2
syncWindow(relay, filter.copy(since = lo, until = mid), idleTimeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onNeed, deliver)
syncWindow(relay, filter.copy(since = mid + 1, until = hi), idleTimeoutMs, fetchBatch, maxConcurrentReqs, onWindow, onNeed, deliver)
}
}
is ReconcileOutcome.Failed ->
throw NegentropySyncException(
relay = relay,
window = filter,
reason = NegentropySyncException.Reason.UNAVAILABLE,
detail = outcome.detail,
)
}
}
private sealed interface ReconcileOutcome {
/** Reconciliation completed; every id was streamed to the downloader. */
object Complete : ReconcileOutcome
/** Relay rejected the set as too large (strfry `max_sync_events`). */
object Overflow : ReconcileOutcome
/** Reconciliation could not complete; [detail] says why. */
class Failed(
val detail: String,
) : ReconcileOutcome
}
/**
* Drives one NIP-77 reconciliation of [filter] against an EMPTY local set, sending
* `NEG-OPEN` and walking the rounds itself (rather than via [NegentropyManager]) so
* it can apply back-pressure: each round's `needIds` are handed to [sendBatch] —
* which suspends while the download queue is full — *before* the next round is
* acked, so the relay's id stream is paced to the downloader and never piles up.
*
* The ids are streamed, not returned; the result is only the terminal outcome.
* Always sends `NEG-CLOSE` and removes the listener on the way out.
*/
private suspend fun INostrClient.reconcileStreaming(
relay: NormalizedRelayUrl,
filter: Filter,
idleTimeoutMs: Long,
fetchBatch: Int,
onNeed: (Int) -> Unit,
sendBatch: suspend (List<HexKey>) -> Unit,
): ReconcileOutcome {
val targetUrl = relay
val relayClient = getOrCreateRelay(relay)
val subId = newSubId()
val session = NegentropySession(subId, filter, localEvents = emptyList())
// Reader-thread → driver hand-off. Holds at most one frame: the relay only sends
// 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.
val clock = IdleClock()
val listener =
object : RelayConnectionListener {
override fun onConnected(
relay: IRelayClient,
pingMillis: Int,
compressed: Boolean,
) {
if (relay.url == targetUrl) clock.bump()
}
override fun onIncomingMessage(
relay: IRelayClient,
msgStr: String,
msg: Message,
) {
if (relay.url == targetUrl) clock.bump()
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
}
}
override fun onDisconnected(relay: IRelayClient) {
if (relay.url == targetUrl) incoming.trySend(NegFrame.Err("closed: relay disconnected"))
}
}
addConnectionListener(listener)
try {
// NEG-OPEN is a one-shot command. Unlike a REQ — which the client replays
// from its active-request state every time a relay (re)connects — a dropped
// NEG-OPEN is never resent, so we must connect and wait until the relay is
// ready before sending it. The connect itself keeps a finite bound even when
// the watchdog is disabled, so an unreachable relay can't hang here forever.
relayClient.connect()
val connectBound = if (idleTimeoutMs > 0) idleTimeoutMs else DEFAULT_CONNECT_TIMEOUT_MS
val connected =
withTimeoutOrNull(connectBound) {
connectedRelaysFlow().first { targetUrl in it }
}
if (connected == null) return ReconcileOutcome.Failed("could not connect within ${connectBound}ms")
relayClient.sendIfConnected(session.open())
while (true) {
// Wait for the relay's next frame, giving up only after idleTimeoutMs of
// total silence (the wait resets whenever the relay sends anything —
// another round, or an event on a download REQ). A disconnect arrives as
// an Err frame, so a dead socket ends this promptly regardless.
val frame =
incoming.receiveWithinIdle(clock, idleTimeoutMs)
?: return ReconcileOutcome.Failed(
if (idleTimeoutMs > 0) {
"relay went silent for ${idleTimeoutMs}ms mid-reconcile"
} else {
"connection closed before reconcile completed"
},
)
when (frame) {
is NegFrame.Err ->
return if (isOverflow(frame.reason)) ReconcileOutcome.Overflow else ReconcileOutcome.Failed(frame.reason)
is NegFrame.Msg -> {
val result = session.processMessage(frame.payload)
val needIds = result.needIds
if (needIds.isNotEmpty()) {
onNeed(needIds.size)
var i = 0
while (i < needIds.size) {
val end = min(i + fetchBatch, needIds.size)
// Copy each batch so the frame's full id list can be freed
// as soon as it is chunked; suspends under back-pressure.
sendBatch(ArrayList(needIds.subList(i, end)))
i = end
}
}
val next = result.nextCmd
if (next != null) {
relayClient.sendIfConnected(next)
} else {
return ReconcileOutcome.Complete
}
}
}
}
} finally {
relayClient.sendIfConnected(session.close())
removeConnectionListener(listener)
incoming.close()
}
}
private sealed interface NegFrame {
class Msg(
val payload: String,
) : NegFrame
class Err(
val reason: String,
) : 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 so equivalent wording from other relays
* still triggers the window split rather than aborting.
*/
private fun isOverflow(reason: String): Boolean =
reason == "blocked: too many query results" ||
reason.contains("too many", ignoreCase = true) ||
reason.startsWith("blocked", ignoreCase = true)
/**
* Reconciles [filter] and streams its ids straight into a bounded download pool, so
* reconciliation and download overlap and peak memory stays independent of the
* window's size. At most [maxConcurrentReqs] `REQ`s of [fetchBatch] ids are open at
* once; the id queue is bounded so a slow download back-pressures reconciliation.
* Returns the terminal [ReconcileOutcome]; events go out through [deliver].
*/
private suspend fun INostrClient.downloadWindow(
relay: NormalizedRelayUrl,
filter: Filter,
idleTimeoutMs: Long,
fetchBatch: Int,
maxConcurrentReqs: Int,
onNeed: (Int) -> Unit,
deliver: suspend (Event) -> Unit,
): ReconcileOutcome =
coroutineScope {
val workerCount = maxConcurrentReqs.coerceAtLeast(1)
// Bounded: when full, reconcileStreaming suspends instead of letting the
// relay's id stream accumulate. This is what keeps memory O(pipeline), not
// O(window).
val idBatches = Channel<List<HexKey>>(workerCount)
val workers =
List(workerCount) {
launch {
for (batch in idBatches) {
coroutineContext.ensureActive()
for (event in fetchByIds(relay, batch, idleTimeoutMs)) {
deliver(event)
}
}
}
}
val outcome =
reconcileStreaming(relay, filter, idleTimeoutMs, fetchBatch, onNeed) { batch ->
idBatches.send(batch)
}
idBatches.close()
workers.joinAll()
outcome
}
/**
* 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
* thread, so collecting here needs no synchronisation.
*
* Events are deduped *within this batch* (a [HashSet] bounded by the batch size, so
* still O(pipeline) memory). A REQ-by-ids should return each id once, but the client
* may re-send the REQ on a reconnect/filter-sync mid-flight, which makes the relay
* replay the batch; without this the same event would be delivered twice. We rely on
* NIP-77 yielding a distinct id set across batches, so no global dedup is needed.
*/
private suspend fun INostrClient.fetchByIds(
relay: NormalizedRelayUrl,
batch: List<HexKey>,
idleTimeoutMs: Long,
): List<Event> {
val subId = newSubId()
val done = Channel<Unit>(Channel.CONFLATED)
val collected = ArrayList<Event>(batch.size)
val seen = HashSet<HexKey>(batch.size)
// Per-batch idle clock: each event resets it, so a batch that keeps streaming is
// never cut off, but a batch that stalls (relay stops mid-flight) unblocks after
// the idle bound instead of hanging a worker. A download batch always keeps a
// finite bound even when the caller disabled the whole-sync watchdog.
val clock = IdleClock()
val batchIdleMs = if (idleTimeoutMs > 0) idleTimeoutMs else DEFAULT_DOWNLOAD_IDLE_MS
val listener =
object : SubscriptionListener {
override fun onEvent(
event: Event,
isLive: Boolean,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
clock.bump()
if (seen.add(event.id)) collected.add(event)
}
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
done.trySend(Unit)
}
override fun onClosed(
message: String,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
done.trySend(Unit)
}
override fun onCannotConnect(
relay: NormalizedRelayUrl,
message: String,
forFilters: List<Filter>?,
) {
done.trySend(Unit)
}
}
try {
subscribe(subId, mapOf(relay to listOf(Filter(ids = batch))), listener)
done.receiveWithinIdle(clock, batchIdleMs)
} finally {
unsubscribe(subId)
done.close()
}
return collected
}
/** Seconds: a window this small that still overflows can't be split further. */
private const val MIN_WINDOW_SECONDS = 1L
/** Bounded buffer between the download workers and the single delivery consumer. */
private const val DELIVERY_BUFFER = 256
/**
* A 32-byte id that no real event can have (all `f`s), used only to hold a
* never-matching keep-alive subscription that keeps the relay connected for the
* duration of a sync. Synthetic/real event ids are SHA-256 digests, so this never
* collides with an actual event.
*/
private val KEEP_ALIVE_ID = "f".repeat(64)
/**
* Finite fallback bounds (ms) for the two waits that must stay bounded even when the
* whole-sync idle watchdog is disabled (`idleTimeoutMs = 0`): the initial connect,
* and each individual download batch. Keeping these finite means an unreachable relay
* or a single stuck batch can never hang the pipeline, while the reconcile rounds
* still honor "run until the socket drops".
*/
private const val DEFAULT_CONNECT_TIMEOUT_MS = 30_000L
private const val DEFAULT_DOWNLOAD_IDLE_MS = 60_000L
/**
* Monotonic "last activity" marker for the idle watchdog. [bump] on every sign of
* life from the relay; [elapsedMs] reports the silence since the last bump.
*
* [bump] is on the per-event hot path (the connection listener bumps for every
* message the relay sends — millions during a large download), so it must not
* allocate: a single [start] mark is taken once (unboxed field) and each bump only
* writes a `Long` of nanos-since-start into a `@Volatile` field. Reader threads
* write, the driver coroutine reads — visibility is all we need, so a plain volatile
* Long beats boxing a `ValueTimeMark` into an `AtomicReference` on every event.
*/
private class IdleClock {
private val start = TimeSource.Monotonic.markNow()
@Volatile
private var lastNanos = 0L
fun bump() {
lastNanos = start.elapsedNow().inWholeNanoseconds
}
fun elapsedMs(): Long = (start.elapsedNow().inWholeNanoseconds - lastNanos) / 1_000_000
}
/**
* Receives the next item, giving up (returning `null`) only after [idleMs] elapse with
* no activity on [clock]. Because [clock] is bumped by *any* relay message — not just
* items on this channel — unrelated progress (e.g. download events arriving during a
* reconcile wait) keeps pushing the deadline out. [idleMs] `<= 0` disables the
* watchdog: it waits until an item arrives (a disconnect is delivered as an item, so
* a dead socket still unblocks it).
*/
private suspend fun <T> Channel<T>.receiveWithinIdle(
clock: IdleClock,
idleMs: Long,
): T? {
if (idleMs <= 0) return receive()
while (true) {
val remaining = idleMs - clock.elapsedMs()
if (remaining <= 0) return null
val item = withTimeoutOrNull(remaining) { receive() }
if (item != null) return item
// Timed out with nothing on this channel. If other activity bumped the clock
// meanwhile, the next `remaining` is positive and we wait again; otherwise it
// is <= 0 on the next iteration and we give up.
}
}
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt
+156 -70
View File
@@ -35,6 +35,8 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.utils.cache.LargeCache
import kotlinx.coroutines.flow.MutableStateFlow
import kotlin.concurrent.atomics.AtomicBoolean
import kotlin.concurrent.atomics.ExperimentalAtomicApi
/**
* Manages relay subscriptions for the entire pool in a way that only
@@ -43,6 +45,7 @@ import kotlinx.coroutines.flow.MutableStateFlow
* This code also awaits a subscription to come to EOSE since many relays
* have through switching subs while they are processing the past.
*/
@OptIn(ExperimentalAtomicApi::class)
class PoolRequests {
/**
* Desired subs and listeners
@@ -64,6 +67,42 @@ class PoolRequests {
fun subState(subId: String): RequestSubscriptionState<NormalizedRelayUrl> = relayState.getOrCreate(subId) { RequestSubscriptionState() }
/**
* Serializes every access to the subscription state machine
* ([RequestSubscriptionState]) and the "should I send a REQ?" decision.
*
* A single subscription can span many relays, and each relay's
* socket-reader thread delivers messages into this class concurrently while
* the app thread adds/removes subscriptions — so the plain maps inside
* [RequestSubscriptionState] are written from several threads at once. That
* is both a memory hazard (concurrent map mutation) and, more importantly,
* a logic hazard: the check-then-send in [decideCommandLocked] must be
* atomic, otherwise two threads can both observe "no REQ in flight" and both
* send a REQ for the same sub id.
*
* This is a tiny non-reentrant spin lock (the same [AtomicBoolean] primitive
* used by BasicRelayClient's connecting mutex): the critical sections are a
* handful of map operations, never any I/O. Listener callbacks and the
* actual socket sends are ALWAYS performed outside the lock — they re-enter
* this class through [onSent], so holding the lock across them would
* self-deadlock.
*/
private val stateLock = AtomicBoolean(false)
private inline fun <R> withStateLock(block: () -> R): R {
while (stateLock.exchange(true)) {
// Another thread holds the lock. Spin-read until it looks free
// (test-and-test-and-set: cheaper on the cache line than hammering
// exchange) then retry the acquisition above.
while (stateLock.load()) { }
}
try {
return block()
} finally {
stateLock.store(false)
}
}
/**
* This is called when a sub is added or removed from this class and
* should update the desired relay list to get the pool to connect
@@ -150,8 +189,10 @@ class PoolRequests {
*/
fun onConnecting(url: NormalizedRelayUrl) {
// Change states to connecting.
relayState.forEach { subId, state ->
state.connecting(url)
withStateLock {
relayState.forEach { subId, state ->
state.connecting(url)
}
}
}
@@ -164,7 +205,9 @@ class PoolRequests {
) {
when (cmd) {
is ReqCmd -> {
subState(cmd.subId).onOpenReq(relay, cmd.filters)
withStateLock {
subState(cmd.subId).onOpenReq(relay, cmd.filters)
}
desiredSubListeners.get(cmd.subId)?.onSubscriptionStarted(
relay = relay.url,
forFilters = cmd.filters,
@@ -172,7 +215,9 @@ class PoolRequests {
}
is CloseCmd -> {
subState(cmd.subId).onSubscriptionClosed(relay)
withStateLock {
subState(cmd.subId).onSubscriptionClosed(relay)
}
desiredSubListeners.get(cmd.subId)?.onSubscriptionClosed(
relay = relay.url,
)
@@ -189,46 +234,63 @@ class PoolRequests {
) {
when (msg) {
is EventMessage -> {
val state = relayState.get(msg.subId)
state?.onNewEvent(relay.url)
var isLive = false
var forFilters: List<Filter>? = null
withStateLock {
val state = relayState.get(msg.subId)
state?.onNewEvent(relay.url)
isLive = state?.currentState(relay.url) == ReqSubStatus.LIVE
forFilters = state?.lastKnownFilterStates(relay.url)
}
desiredSubListeners.get(msg.subId)?.onEvent(
event = msg.event,
isLive = state?.currentState(relay.url) == ReqSubStatus.LIVE,
isLive = isLive,
relay = relay.url,
forFilters = state?.lastKnownFilterStates(relay.url),
forFilters = forFilters,
)
}
is EoseMessage -> {
val state = relayState.get(msg.subId)
state?.onEose(relay.url)
var forFilters: List<Filter>? = null
val cmd =
withStateLock {
val state = relayState.get(msg.subId)
state?.onEose(relay.url)
forFilters = state?.lastKnownFilterStates(relay.url)
// Decide (and pre-mark) the resend while still holding the
// lock, so a concurrent subscribe/unsubscribe on the app
// thread can't also decide to send a REQ for this sub.
decideCommandLocked(msg.subId, relay.url)
}
desiredSubListeners.get(msg.subId)?.onEose(
relay = relay.url,
forFilters = state?.lastKnownFilterStates(relay.url),
forFilters = forFilters,
)
// send a newer version when done
sendToRelayIfChanged(msg.subId, relay.url) { cmd ->
if (cmd != null) {
relay.sendOrConnectAndSync(cmd)
}
}
is ClosedMessage -> {
val state = relayState.get(msg.subId)
state?.onClosed(relay.url)
var forFilters: List<Filter>? = null
val cmd =
withStateLock {
val state = relayState.get(msg.subId)
state?.onClosed(relay.url)
forFilters = state?.lastKnownFilterStates(relay.url)
decideCommandLocked(msg.subId, relay.url)
}
desiredSubListeners.get(msg.subId)?.onClosed(
message = msg.message,
relay = relay.url,
forFilters = state?.lastKnownFilterStates(relay.url),
forFilters = forFilters,
)
// send a newer version when done
sendToRelayIfChanged(msg.subId, relay.url) { cmd ->
// don't send a close if just closed
if (cmd !is CloseCmd) {
relay.sendOrConnectAndSync(cmd)
}
// send a newer version when done, but don't send a close if just closed
if (cmd != null && cmd !is CloseCmd) {
relay.sendOrConnectAndSync(cmd)
}
}
}
@@ -238,8 +300,10 @@ class PoolRequests {
* When the relay disconnects
*/
fun onDisconnected(url: NormalizedRelayUrl) {
relayState.forEach { subId, state ->
state.disconnected(url)
withStateLock {
relayState.forEach { subId, state ->
state.disconnected(url)
}
}
}
@@ -262,16 +326,27 @@ class PoolRequests {
url: NormalizedRelayUrl,
errorMessage: String,
) {
relayState.forEach { subId, state ->
// These are all my subs.. need to figure out which relays have them
val subs = desiredSubs.get(subId)
if (subs != null && url in subs.keys) {
desiredSubListeners.get(subId)?.onCannotConnect(
relay = url,
message = errorMessage,
forFilters = state.lastKnownFilterStates(url),
)
// Snapshot the affected subs (and their last-known filters) under the
// lock, then notify listeners outside it.
val toNotify =
withStateLock {
val list = mutableListOf<Pair<String, List<Filter>?>>()
relayState.forEach { subId, state ->
// These are all my subs.. need to figure out which relays have them
val subs = desiredSubs.get(subId)
if (subs != null && url in subs.keys) {
list.add(subId to state.lastKnownFilterStates(url))
}
}
list
}
toNotify.forEach { (subId, forFilters) ->
desiredSubListeners.get(subId)?.onCannotConnect(
relay = url,
message = errorMessage,
forFilters = forFilters,
)
}
}
@@ -281,54 +356,65 @@ class PoolRequests {
sync: (NormalizedRelayUrl, Command) -> Unit,
) {
relaysToUpdate.forEach { relay ->
sendToRelayIfChanged(subId, relay) { cmd ->
if (cmd is ReqCmd) {
val currentState = relayState.get(subId)?.currentState(relay)
if (currentState == ReqSubStatus.SENT || currentState == ReqSubStatus.QUERYING_PAST) {
// sending multiple REQs triggers multiple EOSEs back and we then don't know which
// one is which.
} else {
sync(relay, cmd)
}
} else {
sync(relay, cmd)
}
// Decide + pre-mark atomically under the lock, then send outside it.
val cmd = withStateLock { decideCommandLocked(subId, relay) }
if (cmd != null) {
sync(relay, cmd)
}
}
}
fun sendToRelayIfChanged(
/**
* Decides which command (if any) must be sent to [relay] to bring it in line
* with the desired filters for [subId], and — for a REQ — pre-marks the
* subscription state as SENT before returning.
*
* Pre-marking is what makes the check-then-send atomic: a second thread that
* runs this method for the same sub sees the SENT state (or the
* already-updated filters) and declines to send a duplicate REQ. Two REQs on
* one sub id race on the wire and produce duplicate EOSEs/events (or, if a
* CLOSE interleaves, an empty result that silently truncates a paged
* download) — that is the bug this guards against.
*
* MUST be called while holding [withStateLock].
*/
private fun decideCommandLocked(
subId: String,
relay: NormalizedRelayUrl,
sync: (Command) -> Unit,
) {
): Command? {
val state = relayState.get(subId)
val oldFilters = state?.currentFilters(relay)
val newFilters = desiredSubs.get(subId)?.get(relay)
sendToRelayIfChanged(subId, oldFilters, newFilters, sync)
}
fun sendToRelayIfChanged(
subId: String,
oldFilters: List<Filter>?,
newFilters: List<Filter>?,
sync: (Command) -> Unit,
) {
if (newFilters.isNullOrEmpty()) {
// some relays are not in this sub anymore. Stop their subscriptions
if (!oldFilters.isNullOrEmpty()) {
// only update if the old filters are not already closed.
sync(CloseCmd(subId))
return when {
newFilters.isNullOrEmpty() -> {
// some relays are not in this sub anymore. Stop their subscriptions
// only if the old filters are not already closed.
if (!oldFilters.isNullOrEmpty()) CloseCmd(subId) else null
}
oldFilters.isNullOrEmpty() || FiltersChanged.needsToResendRequest(oldFilters, newFilters) -> {
// A REQ is warranted: a brand new sub, or the filters changed
// enough (not just a `since` bump) to need a resend. But if a REQ
// is already in flight, don't send another — multiple REQs on one
// sub id trigger multiple EOSEs and we can no longer tell which
// reply belongs to which REQ. The pending change is picked up
// later by the EOSE handler, which runs this method again once the
// sub reaches LIVE.
val current = state?.currentState(relay)
if (current == ReqSubStatus.SENT || current == ReqSubStatus.QUERYING_PAST) {
null
} else {
// Pre-mark SENT + filters so a concurrent decider skips.
subState(subId).onOpenReq(relay, newFilters)
ReqCmd(subId, newFilters)
}
}
else -> {
// Filters are effectively the same; nothing to do.
null
}
} else if (oldFilters.isNullOrEmpty()) {
// new relays were added. Start a new sub in them
sync(ReqCmd(subId, newFilters))
} else if (FiltersChanged.needsToResendRequest(oldFilters, newFilters)) {
// filters were changed enough (not only an update in since) to warn a new update
sync(ReqCmd(subId, newFilters))
} else {
// They are the same don't do anything.
}
}
quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt
+309
View File
@@ -0,0 +1,309 @@
/*
* 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
import com.vitorpamplona.geode.InProcessRelays
import com.vitorpamplona.geode.fixtures.SyntheticEvents
import com.vitorpamplona.geode.testing.RelayClientTest
import com.vitorpamplona.geode.testing.preload
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.NegentropySyncException
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPages
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySync
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySyncEvents
import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.negentropySyncOrFetch
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFailsWith
import kotlin.test.assertFalse
import kotlin.test.assertTrue
class NostrClientNegentropySyncTest : RelayClientTest() {
@Test
fun fullDownloadDeliversEveryEvent() =
runBlocking {
defaultRelay.preload(SyntheticEvents.batch(20, kind = 1))
val got = mutableListOf<Event>()
val result =
withTimeout(20_000) {
client.negentropySync(
relay = defaultRelayUrl,
filter = Filter(kinds = listOf(1)),
) { got.add(it) }
}
assertEquals(20, got.size, "every event should be delivered")
assertEquals(20, got.map { it.id }.toSet().size, "no duplicates")
assertEquals(20, result.needCount)
assertEquals(0, result.haveCount)
assertEquals(20, result.downloaded)
assertEquals(1, result.windows, "small set reconciles in a single window")
}
@Test
fun maxEventsCapsDelivery() =
runBlocking {
defaultRelay.preload(SyntheticEvents.batch(20, kind = 1))
val got = mutableListOf<Event>()
val result =
withTimeout(20_000) {
client.negentropySync(
relay = defaultRelayUrl,
filter = Filter(kinds = listOf(1)),
maxEvents = 10,
fetchBatch = 5,
) { got.add(it) }
}
assertEquals(10, got.size, "delivery stops at maxEvents")
assertEquals(10, result.downloaded)
}
@Test
fun cleanTeardownLeavesNoSubscriptions() =
runBlocking {
defaultRelay.preload(SyntheticEvents.batch(15, kind = 1))
withTimeout(20_000) {
client.negentropySync(
relay = defaultRelayUrl,
filter = Filter(kinds = listOf(1)),
fetchBatch = 4,
) { }
}
assertTrue(
client.activeRequests(defaultRelayUrl).isEmpty(),
"all download subscriptions must be closed after the sync",
)
}
@Test
fun flowVariantStreamsEachEvent() =
runBlocking {
defaultRelay.preload(SyntheticEvents.batch(12, kind = 1))
val events =
withTimeout(20_000) {
client
.negentropySyncEvents(
relay = defaultRelayUrl,
filter = Filter(kinds = listOf(1)),
).toList()
}
assertEquals(12, events.size)
assertEquals(12, events.map { it.id }.toSet().size)
}
/**
* Forces the relay to split its NEG-MSG responses into many small frames
* (`frameSizeLimit` at the library floor) so reconciliation spans many rounds,
* and downloads through a small, bounded pipeline (`fetchBatch`/`maxConcurrentReqs`).
* Exercises the streaming + back-pressure path end to end: every event must still
* be delivered exactly once with nothing accumulated.
*/
@Test
fun multiRoundReconcileStreamsEveryEventThrough() =
runBlocking {
val hub = InProcessRelays(negentropySettings = NegentropySettings(frameSizeLimit = 4096))
val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
val client = NostrClient(hub, scope)
try {
val url = RelayUrlNormalizer.normalize("ws://127.0.0.1:7784/")
hub.getOrCreate(url).preload(SyntheticEvents.batch(1500, kind = 1))
val got = mutableListOf<Event>()
val result =
withTimeout(60_000) {
client.negentropySync(
relay = url,
filter = Filter(kinds = listOf(1)),
fetchBatch = 50,
maxConcurrentReqs = 4,
) { got.add(it) }
}
assertEquals(1500, got.size, "every event delivered across many reconcile rounds")
assertEquals(1500, got.map { it.id }.toSet().size, "each exactly once")
assertEquals(1500, result.downloaded)
assertEquals(1500, result.needCount)
} finally {
client.disconnect()
scope.cancel()
hub.close()
}
}
/**
* A relay that caps negentropy below the matched-set size (strfry's
* `max_sync_events`) but whose events are spread across distinct `created_at`
* values. Windowing alone resolves the cap — each window ends up under it — so
* the sync completes purely via negentropy, no exception, no paging.
*/
@Test
fun overCapWithSpreadTimestampsSucceedsViaWindowing() =
runBlocking {
val hub = InProcessRelays(negentropySettings = NegentropySettings(maxSyncEvents = 3))
val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
val client = NostrClient(hub, scope)
try {
val url = RelayUrlNormalizer.normalize("ws://127.0.0.1:7781/")
// 12 events at distinct created_at: windowing can split until each
// window holds <= the cap.
hub.getOrCreate(url).preload(SyntheticEvents.batch(12, kind = 1))
val got = mutableListOf<Event>()
val result =
withTimeout(60_000) {
client.negentropySync(
relay = url,
filter = Filter(kinds = listOf(1)),
) { got.add(it) }
}
assertEquals(12, got.map { it.id }.toSet().size, "all events reconciled via windowing")
assertEquals(12, result.downloaded)
assertTrue(result.windows > 1, "the set must be split into multiple created_at windows")
} finally {
client.disconnect()
scope.cancel()
hub.close()
}
}
/**
* A relay that caps negentropy below the matched-set size AND whose events all
* share one `created_at`, so no `created_at` window can separate them. Even the
* minimal window stays over the cap, so [negentropySync] cannot reconcile it and
* throws [NegentropySyncException] (reason OVER_MAX_SYNC_EVENTS) rather than
* silently paging — the fallback is the caller's call.
*/
@Test
fun overCapMinimalWindowThrowsInsteadOfPaging() =
runBlocking {
val hub = InProcessRelays(negentropySettings = NegentropySettings(maxSyncEvents = 3))
val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
val client = NostrClient(hub, scope)
try {
val url = RelayUrlNormalizer.normalize("ws://127.0.0.1:7782/")
val events = (1..10).map { SyntheticEvents.fakeEvent(idSeed = it, kind = 1, createdAt = 1000L) }
hub.getOrCreate(url).preload(events)
val thrown =
assertFailsWith<NegentropySyncException> {
withTimeout(60_000) {
client.negentropySync(
relay = url,
filter = Filter(kinds = listOf(1)),
) { }
}
}
assertEquals(NegentropySyncException.Reason.OVER_MAX_SYNC_EVENTS, thrown.reason)
// And the caller can recover by paging it themselves.
val paged = mutableListOf<Event>()
withTimeout(60_000) {
client.fetchAllPages(url, listOf(Filter(kinds = listOf(1)))) { paged.add(it) }
}
assertEquals(10, paged.map { it.id }.toSet().size)
} finally {
client.disconnect()
scope.cancel()
hub.close()
}
}
/**
* The "try negentropy, else page" combinator: against the same over-cap relay
* where raw [negentropySync] throws, [negentropySyncOrFetch] transparently pages
* and delivers every event, reporting that it fell back.
*/
@Test
fun orFetchPagesWhenNegentropyCannotReconcile() =
runBlocking {
val hub = InProcessRelays(negentropySettings = NegentropySettings(maxSyncEvents = 3))
val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
val client = NostrClient(hub, scope)
try {
val url = RelayUrlNormalizer.normalize("ws://127.0.0.1:7783/")
val events = (1..10).map { SyntheticEvents.fakeEvent(idSeed = it, kind = 1, createdAt = 1000L) }
hub.getOrCreate(url).preload(events)
val got = mutableListOf<Event>()
val result =
withTimeout(60_000) {
client.negentropySyncOrFetch(
relay = url,
filter = Filter(kinds = listOf(1)),
) { got.add(it) }
}
assertEquals(10, got.map { it.id }.toSet().size, "all events delivered via the paging fallback")
assertEquals(10, result.downloaded)
assertTrue(result.pagedFallback, "it should have fallen back to paging")
assertEquals(
NegentropySyncException.Reason.OVER_MAX_SYNC_EVENTS,
result.fallbackCause?.reason,
)
} finally {
client.disconnect()
scope.cancel()
hub.close()
}
}
/**
* On a relay that reconciles fine, [negentropySyncOrFetch] uses negentropy and
* does not page.
*/
@Test
fun orFetchUsesNegentropyWhenItWorks() =
runBlocking {
defaultRelay.preload(SyntheticEvents.batch(8, kind = 1))
val got = mutableListOf<Event>()
val result =
withTimeout(20_000) {
client.negentropySyncOrFetch(
relay = defaultRelayUrl,
filter = Filter(kinds = listOf(1)),
) { got.add(it) }
}
assertEquals(8, got.size)
assertFalse(result.pagedFallback, "negentropy should have handled it")
assertEquals(8, result.negentropy?.downloaded)
}
}
quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/PoolRequestsConcurrencyTest.kt
+146
View File
@@ -0,0 +1,146 @@
/*
* 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
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.PoolRequests
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
import java.util.concurrent.CountDownLatch
import java.util.concurrent.atomic.AtomicInteger
import kotlin.concurrent.thread
import kotlin.test.Test
import kotlin.test.assertEquals
/**
* Regression guard for the shared-sub-id double-REQ race in [PoolRequests].
*
* A single subscription id is driven from two threads at once: the app thread
* (the subscribe path, [PoolRequests.sendToRelayIfChanged]) and the relay reader
* thread (an EOSE that triggers an auto-resend, [PoolRequests.onIncomingMessage]).
* The subscription is already LIVE and its desired filters have just changed, so
* both threads independently conclude "the filters changed, send a REQ".
*
* The bug: the decision (read state) and the send (mark state SENT via onSent)
* were not atomic, so the reader could read the pre-send state (filters still on
* the previous value) while the app had already moved the desired filters
* forward — and both would send a REQ for the same sub id. Two REQs on one id
* race on the wire: the relay answers with two EOSEs and duplicate events, or —
* if a CLOSE interleaves — an empty result that silently truncates a paged
* download (this is what broke `fetchAllPages` on large sets).
*
* The fix makes the "should I send a REQ?" decision pre-mark the state
* atomically, so exactly one REQ is ever produced. This test pins the exact
* interleaving the bug needs (app has produced its REQ but not yet run onSent)
* open and asserts only one REQ comes out.
*/
class PoolRequestsConcurrencyTest {
private class FakeRelay(
override val url: NormalizedRelayUrl,
val onCmd: (Command) -> Unit,
) : IRelayClient {
override fun connect() {}
override fun needsToReconnect() = false
override fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean) {}
override fun isConnected() = true
override fun sendOrConnectAndSync(cmd: Command) = onCmd(cmd)
override fun sendIfConnected(cmd: Command) = onCmd(cmd)
override fun disconnect() {}
}
@Test
fun concurrentEoseResendAndSubscribeSendExactlyOneReq() {
val url = RelayUrlNormalizer.normalize("ws://race/")
val subId = "shared-sub"
val filtersA = listOf(Filter(kinds = listOf(1)))
val filtersB = listOf(Filter(kinds = listOf(2)))
val listener = object : SubscriptionListener {}
// Many episodes so a regression that only sometimes doubles still trips.
repeat(300) { episode ->
val pool = PoolRequests()
val reqBCount = AtomicInteger(0)
fun countReqB(cmd: Command) {
if (cmd is ReqCmd && cmd.filters == filtersB) reqBCount.incrementAndGet()
}
val fakeRelay =
FakeRelay(url) { cmd ->
// relay-reader auto-resend send path
countReqB(cmd)
pool.onSent(url, cmd)
}
// Bring the sub to LIVE with filters A.
val setupRelays = pool.addOrUpdate(subId, mapOf(url to filtersA), listener)
pool.sendToRelayIfChanged(subId, setupRelays) { _, cmd -> pool.onSent(url, cmd) }
pool.onIncomingMessage(fakeRelay, EoseMessage(subId))
// The desired filters change to B (e.g. the next page of a paged download).
pool.addOrUpdate(subId, mapOf(url to filtersB), listener)
val appProducedReq = CountDownLatch(1)
val readerDone = CountDownLatch(1)
val appThread =
thread {
pool.sendToRelayIfChanged(subId, setOf(url)) { _, cmd ->
countReqB(cmd)
// App has produced its REQ(B); park before onSent so the
// subscription state is not yet advanced — the exact window
// the race needs.
appProducedReq.countDown()
readerDone.await()
pool.onSent(url, cmd)
}
}
val readerThread =
thread {
appProducedReq.await()
pool.onIncomingMessage(fakeRelay, EoseMessage(subId))
readerDone.countDown()
}
appThread.join()
readerThread.join()
assertEquals(
1,
reqBCount.get(),
"episode $episode: exactly one REQ must be sent for the changed filters, " +
"never a duplicate from the app + reader race",
)
}
}
}