Merge pull request #3892 from vitorpamplona/feat/negentropy-want-id-predicate

negentropy: let a caller decline an id before the download REQ
2026-10-06 03:38:23 +00:00 · 2026-08-10 18:21:30 -04:00
parent b259382652 7041980996
commit b6eb610749
5 changed files with 382 additions and 3 deletions
@@ -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,
 )
@@ -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(),
    )
 }

@@ -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<IdAndTime> = 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<IdAndTime> = 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<HexKey>): List<HexKey> {
+        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<HexKey>) -> Unit,
    sendHaveBatch: (suspend (List<HexKey>) -> 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<HexKey>) -> Unit,
    sendHaveBatch: (suspend (List<HexKey>) -> 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
                        }
                    }
@@ -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<HexKey>()
+        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")
+    }
+}
@@ -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<String>())
+            val recording =
+                object : PassThroughPolicy() {
+                    override fun accept(cmd: ReqCmd): PolicyResult<ReqCmd> {
+                        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<Event>()
+                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<Event>()
+            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<Event>()
+            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.