diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyFanOutResult.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyFanOutResult.kt index 96392ace8a..a7cbb1678f 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyFanOutResult.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NegentropyFanOutResult.kt @@ -36,4 +36,11 @@ class NegentropyFanOutResult( val downloaded: Int, val windows: Int, val connections: Int, + /** + * Ids `wantId` declined, so no `REQ` was ever issued for them. `0` when no + * predicate was passed. Same accounting as + * [NegentropySyncResult.skipped]: apart from [downloaded], and never folded + * into [needCount], which stays the honest protocol diff. + */ + val skipped: Int = 0, ) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropyFanOutExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropyFanOutExt.kt index e941af4417..35583fe02c 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropyFanOutExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropyFanOutExt.kt @@ -84,6 +84,13 @@ suspend fun negentropySyncFanOut( fetchBatch: Int = 250, idleTimeoutMs: Long = 120_000L, reconcileConcurrency: Int = 2, + /** + * Same contract as [negentropySync]'s: consulted for every id the reconcile + * names, before the `REQ` that would fetch it. Declined ids are counted in + * [NegentropyFanOutResult.skipped]. Called from several reconciler + * coroutines at once, so it must be cheap and thread-safe. + */ + wantId: ((HexKey) -> Boolean)? = null, onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, onEvent: suspend (Event) -> Unit, ): NegentropyFanOutResult { @@ -91,6 +98,7 @@ suspend fun negentropySyncFanOut( val need = AtomicInt(0) val have = AtomicInt(0) + val skipped = AtomicInt(0) val windows = AtomicInt(0) val used = AtomicInt(0) var downloaded = 0 @@ -155,6 +163,7 @@ suspend fun negentropySyncFanOut( need.addAndFetch(it) }, onHave = { have.addAndFetch(it) }, + gate = NeedGate(wantId) { skipped.addAndFetch(it) }, sendNeedBatch = { batch -> idBatches.send(batch) }, sendHaveBatch = if (localEntries.isEmpty() && localIndex == null) null else { _ -> }, ) @@ -192,6 +201,7 @@ suspend fun negentropySyncFanOut( downloaded = downloaded, windows = windows.load(), connections = used.load(), + skipped = skipped.load(), ) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt index 51941e1856..570f56c147 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientNegentropySyncExt.kt @@ -77,6 +77,18 @@ class NegentropySyncResult( val downloaded: Int, val windows: Int, val peerCap: Long? = null, + /** + * Ids the reconcile named that `wantId` declined, so no `REQ` was ever + * issued for them. Always `0` when no predicate was passed. + * + * Reported apart from [downloaded] and from [needCount] on purpose: + * [needCount] stays the honest protocol diff (what the relay has that the + * local set lacks) whether or not the caller chose to fetch it, and a + * skipped id was never a download. Folding either way would leave a caller + * unable to tell "my predicate is doing nothing" from "my predicate is + * eating everything" — and both are silent. + */ + val skipped: Int = 0, ) /** @@ -151,6 +163,26 @@ class NegentropySyncResult( * `limitation.max_subscriptions`; e.g. strfry defaults to 20) and exceeding the * cap can wedge the connection, not just fail the extra REQ — size the two knobs * to fit the target relay. + * @param wantId optional gate consulted for every id the reconcile names, + * BEFORE the `REQ` that would download it. Return `false` and the id is dropped + * from the fetch queue and counted in [NegentropySyncResult.skipped]; the + * reconcile itself is untouched, so [NegentropySyncResult.needCount] still + * reports the true diff. Called from the reconciler coroutines (possibly + * several at once when `reconcileConcurrency > 1`), so it must be cheap and + * thread-safe — a membership test, not a query. + * + * The case this exists for: a caller whose store will refuse an id no matter + * how often it arrives. A mirror that keeps only the newest version of a + * replaceable event is offered every relay's older copy on every sync, and + * without a hook here the only place to decline is after the body is already + * on the wire. `onEvent` is too late to save the bytes. + * + * **It does not cover a window handed to [onUnreconcilableWindow].** That + * window is drained by the caller over `REQ`, and a `REQ` names no ids before + * it streams bodies, so declined events in such a window arrive anyway and are + * not counted in [NegentropySyncResult.skipped]. Rare — it takes a single + * second denser than the relay's cap — but a caller treating the gate as an + * absolute bound on what it can receive would be wrong. * @param onProgress optional `(needSoFar, downloaded)` ticks as work proceeds. * @param onEvent called once per distinct event, serially, from the single * delivery consumer coroutine (not the relay reader thread) — so it never overlaps @@ -170,10 +202,12 @@ suspend fun INostrClient.negentropySync( localIndex: NegentropyLocalIndex? = null, targetWindow: Int = 0, onUnreconcilableWindow: (suspend (Filter) -> Unit)? = null, + wantId: ((HexKey) -> Boolean)? = null, onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, onEvent: suspend (Event) -> Unit, ): NegentropySyncResult { val need = AtomicInt(0) + val skipped = AtomicInt(0) val windows = AtomicInt(0) var downloaded = 0 var peerCap: Long? = null @@ -213,6 +247,8 @@ suspend fun INostrClient.negentropySync( // single consumer loop below so the user callback is never // invoked from two coroutines at once. onNeed = { need.addAndFetch(it) }, + onSkipped = { skipped.addAndFetch(it) }, + wantId = wantId, deliver = { events.send(it) }, ) } finally { @@ -241,6 +277,7 @@ suspend fun INostrClient.negentropySync( downloaded = downloaded, windows = windows.load(), peerCap = peerCap, + skipped = skipped.load(), ) } @@ -257,6 +294,7 @@ suspend fun INostrClient.negentropySync( localIndex: NegentropyLocalIndex? = null, targetWindow: Int = 0, onUnreconcilableWindow: (suspend (Filter) -> Unit)? = null, + wantId: ((HexKey) -> Boolean)? = null, onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, onEvent: suspend (Event) -> Unit, ): NegentropySyncResult = @@ -273,6 +311,7 @@ suspend fun INostrClient.negentropySync( localIndex = localIndex, targetWindow = targetWindow, onUnreconcilableWindow = onUnreconcilableWindow, + wantId = wantId, onProgress = onProgress, onEvent = onEvent, ) @@ -342,6 +381,14 @@ suspend fun INostrClient.negentropySyncOrFetch( localEntries: List = emptyList(), localIndex: NegentropyLocalIndex? = null, targetWindow: Int = 0, + /** + * Passed straight to [negentropySync]. Note it does NOT apply to the paged + * fallback: a `REQ` names no ids before it streams bodies, so there is + * nothing to gate there. A caller relying on this to bound its downloads + * should read [NegentropyOrFetchResult.pagedFallback] and expect the + * suppressed ids to arrive after all when a window pages. + */ + wantId: ((HexKey) -> Boolean)? = null, onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, onEvent: suspend (Event) -> Unit, ): NegentropyOrFetchResult { @@ -397,6 +444,7 @@ suspend fun INostrClient.negentropySyncOrFetch( if (accept(event)) onProgress?.invoke(delivered, delivered) } }, + wantId = wantId, onProgress = onProgress, ) { accept(it) } NegentropyOrFetchResult( @@ -440,6 +488,7 @@ suspend fun INostrClient.negentropySyncOrFetch( localEntries: List = emptyList(), localIndex: NegentropyLocalIndex? = null, targetWindow: Int = 0, + wantId: ((HexKey) -> Boolean)? = null, onProgress: ((needSoFar: Int, downloaded: Int) -> Unit)? = null, onEvent: suspend (Event) -> Unit, ): NegentropyOrFetchResult = @@ -455,10 +504,53 @@ suspend fun INostrClient.negentropySyncOrFetch( localEntries = localEntries, localIndex = localIndex, targetWindow = targetWindow, + wantId = wantId, onProgress = onProgress, onEvent = onEvent, ) +/** + * Decides which of the ids a reconcile named are actually worth a `REQ`. + * + * Small and separate because it is the only place in the download path where + * an id can still be declined for free, and because the three things it does + * are each easy to get wrong in a lambda nobody can call from a test: apply + * the predicate, count what was dropped, and refuse to emit an empty batch. + * + * [keep] runs on the reconciler coroutines, so with `reconcileConcurrency > 1` + * it is called concurrently — hence the counting goes through [onSkipped], + * which the caller makes atomic, rather than a field here. + * + * **It is applied to a whole reconcile round's ids, before they are chunked + * into fetch batches.** Gating after the chunking instead would keep the batch + * COUNT and shrink every one of them: at the density this exists for (a mirror + * declining most of what it is offered) a `fetchBatch` of 500 would become a + * hundred `REQ`s of five ids apiece, each with its own EOSE round trip — + * turning a bandwidth saving into a latency regression. + */ +internal class NeedGate( + private val wantId: ((HexKey) -> Boolean)?, + private val onSkipped: (Int) -> Unit, +) { + /** + * The subset of [batch] to download. Empty when everything was declined — + * deliberately NOT null: the caller's chunk loop already does nothing with + * an empty list, and a nullable return here invites + * `gate?.keep(ids) ?: ids`, which reads as "no gate, keep everything" and + * silently means "everything was declined, so send everything". That exact + * elvis collapsed the fully-declining case into a full download once. + * + * With no predicate this returns [batch] itself — the unfiltered sync is + * the common case and must not pay a copy for a feature it is not using. + */ + fun keep(batch: List): List { + val wanted = if (wantId == null) batch else batch.filter(wantId) + val dropped = batch.size - wanted.size + if (dropped > 0) onSkipped(dropped) + return wanted + } +} + /** * The whole-sync pipeline: a single pool of [maxConcurrentReqs] download workers * fed by up to [reconcileConcurrency] concurrent window reconciliations through a @@ -494,6 +586,8 @@ private suspend fun INostrClient.syncPipeline( targetWindow: Int, onWindow: () -> Unit, onNeed: (Int) -> Unit, + onSkipped: (Int) -> Unit, + wantId: ((HexKey) -> Boolean)?, onPeerCap: ((Long) -> Unit)?, onUnreconcilableWindow: (suspend (Filter) -> Unit)?, deliver: suspend (Event) -> Unit, @@ -526,6 +620,9 @@ private suspend fun INostrClient.syncPipeline( onHave = {}, onPeerCap = onPeerCap, onUnreconcilableWindow = onUnreconcilableWindow, + // The gate is applied inside the reconcile, before these batches are + // cut — see NeedGate — so by here every id is one we want. + gate = NeedGate(wantId, onSkipped), sendNeedBatch = { batch -> idBatches.send(batch) }, sendHaveBatch = null, ) @@ -583,6 +680,9 @@ internal suspend fun reconcileWindows( // that hit the window, so a slow drain holds that reconciler — with // reconcileConcurrency = 1 the rest of the sweep waits for it. onUnreconcilableWindow: (suspend (Filter) -> Unit)? = null, + // Applied to each round's need ids before they are chunked. Null for the + // callers that hand the ids straight to their own consumer. + gate: NeedGate? = null, sendNeedBatch: suspend (List) -> Unit, sendHaveBatch: (suspend (List) -> Unit)?, ) = coroutineScope { @@ -703,6 +803,7 @@ internal suspend fun reconcileWindows( fetchBatch = batchSize, onNeed = onNeed, onHave = onHave, + gate = gate, sendNeedBatch = sendNeedBatch, sendHaveBatch = sendHaveBatch, ) @@ -1032,6 +1133,7 @@ private suspend fun INostrClient.reconcileStreaming( fetchBatch: Int, onNeed: (Int) -> Unit, onHave: (Int) -> Unit, + gate: NeedGate? = null, sendNeedBatch: suspend (List) -> Unit, sendHaveBatch: (suspend (List) -> Unit)?, ): ReconcileOutcome { @@ -1170,13 +1272,19 @@ private suspend fun INostrClient.reconcileStreaming( val result = session.processMessage(frame.payload) val needIds = result.needIds if (needIds.isNotEmpty()) { + // Counted before the gate: this is the protocol diff, and + // it is true whether or not the caller wants to fetch it. onNeed(needIds.size) + // Gated before the chunking, so a selective predicate + // yields FEWER full batches rather than the same number + // of nearly-empty ones — see NeedGate. + val wanted = if (gate == null) needIds else gate.keep(needIds) var i = 0 - while (i < needIds.size) { - val end = min(i + fetchBatch, needIds.size) + while (i < wanted.size) { + val end = min(i + fetchBatch, wanted.size) // Copy each batch so the frame's full id list can be freed // as soon as it is chunked; suspends under back-pressure. - sendNeedBatch(ArrayList(needIds.subList(i, end))) + sendNeedBatch(ArrayList(wanted.subList(i, end))) i = end } } diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NeedGateTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NeedGateTest.kt new file mode 100644 index 0000000000..f404e96887 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NeedGateTest.kt @@ -0,0 +1,137 @@ +/* + * 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.HexKey +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertSame +import kotlin.test.assertTrue + +/** + * The one point on the download path where an id can still be declined for + * free. Everything asserted here is a property the inline version of this code + * could break silently: a sync with no predicate must not start allocating, a + * fully-declined batch must not become an empty REQ, and the skip count must + * stay separate from the reconcile's own diff so an operator can tell a + * predicate that does nothing from one that eats everything. + */ +class NeedGateTest { + private fun id(n: Int): HexKey = n.toString().padStart(64, '0') + + private fun batch(range: IntRange) = range.map(::id) + + private class Skips { + var total = 0 + var calls = 0 + + val sink: (Int) -> Unit = { + total += it + calls++ + } + } + + @Test + fun `with no predicate the batch passes through untouched and uncopied`() { + // The unfiltered sync is the common case and must not pay a copy per + // batch for a feature it is not using. + val skips = Skips() + val input = batch(1..500) + val kept = NeedGate(null, skips.sink).keep(input) + + assertSame(input, kept, "an unfiltered batch must be the same list instance, not a copy") + assertEquals(0, skips.total) + assertEquals(0, skips.calls, "nothing was dropped so the counter should not even be touched") + } + + @Test + fun `a predicate keeps its subset in order`() { + val skips = Skips() + val wanted = setOf(id(2), id(4), id(6)) + val kept = NeedGate({ it in wanted }, skips.sink).keep(batch(1..6)) + + assertEquals(listOf(id(2), id(4), id(6)), kept) + assertEquals(3, skips.total) + } + + @Test + fun `a fully declined batch yields an empty list and never null`() { + // Non-null on purpose. A nullable return invites + // `gate?.keep(ids) ?: ids`, which reads as "no gate, keep everything" + // and silently means "everything declined, so send everything" — a + // fully-declining gate turning into a full download. + val skips = Skips() + val kept = NeedGate({ false }, skips.sink).keep(batch(1..10)) + + assertTrue(kept.isEmpty(), "nothing survived, so there is nothing to send") + assertEquals(10, skips.total) + } + + @Test + fun `an empty input yields empty and counts nothing`() { + val skips = Skips() + assertTrue(NeedGate({ true }, skips.sink).keep(emptyList()).isEmpty()) + assertEquals(0, skips.total) + } + + @Test + fun `a predicate that accepts everything drops nothing`() { + val skips = Skips() + val kept = NeedGate({ true }, skips.sink).keep(batch(1..20)) + + assertEquals(20, kept.size) + assertEquals(0, skips.total) + assertEquals(0, skips.calls) + } + + @Test + fun `skips accumulate across batches`() { + // One gate spans a whole sync, so the count has to survive more than + // the batch it was produced in. + val skips = Skips() + val gate = NeedGate({ it.endsWith("1") }, skips.sink) + gate.keep(batch(1..10)) + gate.keep(batch(11..20)) + + assertEquals(18, skips.total, "only ids 1 and 11 of the twenty end in 1") + } + + @Test + fun `the predicate sees every id in the batch exactly once`() { + val seen = mutableListOf() + NeedGate({ + seen.add(it) + true + }, Skips().sink).keep(batch(1..5)) + + assertEquals(batch(1..5), seen) + } + + @Test + fun `a sync result reports no skips unless a predicate declined something`() { + // Back-compat: every existing caller constructs this without the new + // field and must keep reading zero. + val result = NegentropySyncResult(needCount = 7, haveCount = 0, downloaded = 7, windows = 1) + + assertEquals(0, result.skipped) + assertTrue(result.needCount == 7, "the reconcile diff is unaffected by the gate") + } +} diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt index 6f4b121146..4524099b2c 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientNegentropySyncTest.kt @@ -360,6 +360,123 @@ class NostrClientNegentropySyncTest : RelayClientTest() { } } + /** + * `wantId` declines ids BEFORE the download REQ, so the events never cross + * the wire — the point of putting the hook here rather than at `onEvent`. + * + * The assertion that matters is the one on the recorded `REQ` filters: a + * gate that merely dropped events after delivery would satisfy every other + * check in this test while saving nothing. + */ + @Test + fun wantIdSkipsIdsBeforeTheyAreEverRequested() = + runBlocking { + // Every id the relay was actually asked for, straight off the wire. + val requested = java.util.Collections.synchronizedList(mutableListOf()) + val recording = + object : PassThroughPolicy() { + override fun accept(cmd: ReqCmd): PolicyResult { + cmd.filters.forEach { f -> f.ids?.let { requested.addAll(it) } } + return PolicyResult.Accepted(cmd) + } + } + + val hub = InProcessRelays(defaultPolicy = { recording }) + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(hub, scope) + try { + val url = RelayUrlNormalizer.normalize("ws://127.0.0.1:7791/") + val events = (1..20).map { SyntheticEvents.fakeEvent(idSeed = it, kind = 1) } + hub.getOrCreate(url).preload(events) + + // Decline the first ten by id. + val unwanted = events.take(10).map { it.id }.toSet() + + val got = mutableListOf() + val result = + withTimeout(30_000) { + client.negentropySync( + relay = url, + filter = Filter(kinds = listOf(1)), + idleTimeoutMs = 10_000L, + wantId = { it !in unwanted }, + ) { got.add(it) } + } + + assertEquals(10, got.size, "only the wanted half is delivered") + assertTrue(got.none { it.id in unwanted }, "no declined event was delivered") + assertEquals(10, result.downloaded) + assertEquals(10, result.skipped, "the declined ids are reported apart from the download count") + assertEquals( + 20, + result.needCount, + "needCount stays the honest protocol diff — the relay really did have all 20 that we lacked", + ) + + // The whole point: the declined ids were never asked for. + assertTrue( + requested.none { it in unwanted }, + "a declined id must never reach a REQ; requested = ${requested.filter { it in unwanted }}", + ) + // Every wanted id WAS asked for — so the gate declined the right + // half rather than simply starving the download. Asserted as a + // subset rather than a count because negentropySync also opens a + // keep-alive subscription carrying a sentinel id. + assertTrue( + requested.containsAll(events.drop(10).map { it.id }), + "every wanted id should still have been requested", + ) + } finally { + client.disconnect() + scope.cancel() + hub.close() + } + } + + /** With no `wantId` nothing changes: every id is fetched and `skipped` is 0. */ + @Test + fun withoutWantIdEveryIdIsStillRequested() = + runBlocking { + defaultRelay.preload(SyntheticEvents.batch(12, kind = 1)) + + val got = mutableListOf() + val result = + withTimeout(20_000) { + client.negentropySync( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + ) { got.add(it) } + } + + assertEquals(12, got.size) + assertEquals(0, result.skipped, "no predicate means nothing is ever skipped") + } + + /** + * A gate that declines everything must finish cleanly rather than hang: the + * empty batches are dropped instead of being queued as REQs for no ids. + */ + @Test + fun wantIdDecliningEverythingDownloadsNothingAndStillCompletes() = + runBlocking { + defaultRelay.preload(SyntheticEvents.batch(15, kind = 1)) + + val got = mutableListOf() + val result = + withTimeout(20_000) { + client.negentropySync( + relay = defaultRelayUrl, + filter = Filter(kinds = listOf(1)), + wantId = { false }, + ) { got.add(it) } + } + + assertTrue(got.isEmpty(), "nothing was wanted, so nothing is delivered") + assertEquals(0, result.downloaded) + assertEquals(15, result.skipped) + assertEquals(15, result.needCount, "the reconcile still saw the full diff") + } + /** * On a relay that reconciles fine, [negentropySyncOrFetch] uses negentropy and * does not page.