Merge branch 'main' into claude/review-fanout-index-e1Uqq

Conflicts: - quartz/.../LiveEventStore.kt: main added IngestQueue (group-commit) and a fire-and-forget submit() path; this branch replaced the SharedFlow live fanout with FilterIndex-driven dispatch. Resolved by keeping main's submit/insert pipeline shape (IngestQueue ingest ctor param, submit() callback, insert() wrapping submit via CompletableDeferred) but routing the on-Accepted fanout through FilterIndex.candidatesFor instead of newEventStream.tryEmit. Added private fanout(event) helper. Kept main's snapshotIdsForNegentropy method. - geode/.../LoadBenchmark.kt: both sides added a new @Test method. Kept fanoutScaling (this branch) and publishPipelinedSingleClient (main).
2026-05-07 23:55:00 +00:00
parent 70afcd10ca 90fac6f4e3
commit 44a8aeea28
59 changed files with 5111 additions and 357 deletions
@@ -28,9 +28,11 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.KindAllowDenyPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.PubkeyAllowDenyPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.RejectFutureEventsPolicy
+import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyAuthOnlyPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.server.policies.VerifyPolicy
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
 import java.io.File

 /**
@@ -83,6 +85,12 @@ fun main(args: Array<String>) {
    // opts out (CLI `--no-verify` or `[options].verify_signatures = false`
    // in the config).
    val verifySigs = !a.flag("--no-verify") && config.options.verify_signatures
+    // Parallel verify is on whenever signature checking is on; the
+    // IngestQueue handles it instead of VerifyPolicy. Operators can
+    // force the legacy in-policy path with `--no-parallel-verify` or
+    // `[options].parallel_verify = false`.
+    val parallelVerify =
+        verifySigs && !a.flag("--no-parallel-verify") && config.options.parallel_verify

    // Advertised URL: explicit `info.relay_url` wins, then build from
    // host/port/path. 0.0.0.0 bind → 127.0.0.1 in the URL so NIP-42
@@ -98,11 +106,26 @@ fun main(args: Array<String>) {
    val store: IEventStore = EventStore(dbName = dbFile, relay = advertisedUrl)

    val policyBuilder: () -> IRelayPolicy = {
-        composePolicy(config, advertisedUrl, requireAuth, verifySigs)
+        composePolicy(config, advertisedUrl, requireAuth, verifySigs, parallelVerify)
    }

    val stateFile = config.admin.state_file?.let { File(it) }
-    val relay = Relay(advertisedUrl, store, info, policyBuilder, stateFile = stateFile)
+    val negentropySettings =
+        NegentropySettings(
+            frameSizeLimit = config.negentropy.frame_size_limit,
+            maxSyncEvents = config.negentropy.max_sync_events,
+            maxSessionsPerConnection = config.negentropy.max_sessions_per_connection,
+        )
+    val relay =
+        Relay(
+            advertisedUrl,
+            store,
+            info,
+            policyBuilder,
+            stateFile = stateFile,
+            parallelVerify = parallelVerify,
+            negentropySettings = negentropySettings,
+        )
    // Frame cap honors max_ws_frame_bytes when set; max_ws_message_bytes
    // is treated as the same cap (Ktor's WebSockets plugin only exposes
    // a single per-frame limit; multi-frame messages remain unbounded).
@@ -151,6 +174,7 @@ private fun composePolicy(
    advertisedUrl: com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl,
    requireAuth: Boolean,
    verifySigs: Boolean,
+    parallelVerify: Boolean,
 ): IRelayPolicy {
    val pieces = mutableListOf<IRelayPolicy>()

@@ -171,7 +195,11 @@ private fun composePolicy(
    }

    if (verifySigs) {
-        pieces += VerifyPolicy
+        // When parallel verify is on, the IngestQueue handles EVENT
+        // verification on the writer's CPU fan-out — but AUTH events
+        // bypass the queue, so we still need the policy chain to
+        // verify those. `VerifyAuthOnlyPolicy` does exactly that.
+        pieces += if (parallelVerify) VerifyAuthOnlyPolicy else VerifyPolicy
    }

    return pieces.fold<IRelayPolicy, IRelayPolicy>(EmptyPolicy) { acc, p ->
@@ -33,6 +33,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
 import com.vitorpamplona.quartz.nip01Core.store.IEventStore
 import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
 import com.vitorpamplona.quartz.nip11RelayInfo.Nip11RelayInformation
+import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
 import com.vitorpamplona.quartz.nip86RelayManagement.server.BanListPolicy
 import com.vitorpamplona.quartz.nip86RelayManagement.server.BanStore
 import kotlinx.coroutines.SupervisorJob
@@ -73,6 +74,22 @@ class Relay(
     * everything in memory only — fine for tests.
     */
    stateFile: File? = null,
+    /**
+     * Run Schnorr signature verification in parallel inside the
+     * [com.vitorpamplona.quartz.nip01Core.relay.server.IngestQueue]
+     * instead of serially in the policy chain. Enables the Tier-3
+     * win in `geode/plans/2026-05-07-event-ingestion-batching.md`.
+     *
+     * When set, callers MUST omit `VerifyPolicy` from [policyBuilder]
+     * — having both verifies the same event twice for no benefit.
+     * `Main.kt` skips `VerifyPolicy` when this flag is on.
+     */
+    parallelVerify: Boolean = false,
+    /**
+     * NIP-77 server-side tuning (frame cap, snapshot cap,
+     * per-connection session cap). Defaults to strfry-parity values.
+     */
+    negentropySettings: NegentropySettings = NegentropySettings.Default,
 ) : AutoCloseable {
    private val stateStore: RelayStateStore? = stateFile?.let { RelayStateStore(it) }

@@ -157,7 +174,9 @@ class Relay(
                val user = policyBuilder()
                if (user === EmptyPolicy) BanListPolicy(banStore) else user + BanListPolicy(banStore)
            },
-            parentContext,
+            parentContext = parentContext,
+            parallelVerify = parallelVerify,
+            negentropySettings = negentropySettings,
        )

    /**
@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.server.policies.EmptyPolicy
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
+import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
 import java.util.concurrent.ConcurrentHashMap

 /**
@@ -50,6 +51,7 @@ import java.util.concurrent.ConcurrentHashMap
 */
 class RelayHub(
    private val defaultPolicy: () -> IRelayPolicy = { EmptyPolicy },
+    private val negentropySettings: NegentropySettings = NegentropySettings.Default,
 ) : WebsocketBuilder,
    AutoCloseable {
    private val relays = ConcurrentHashMap<NormalizedRelayUrl, Relay>()
@@ -60,7 +62,11 @@ class RelayHub(
    fun getOrCreate(url: NormalizedRelayUrl): Relay {
        check(!closed) { "RelayHub has been closed" }
        return relays.getOrPut(url) {
-            Relay(url = url, policyBuilder = defaultPolicy)
+            Relay(
+                url = url,
+                policyBuilder = defaultPolicy,
+                negentropySettings = negentropySettings,
+            )
        }
    }

@@ -43,6 +43,7 @@ data class RelayConfig(
    val limits: LimitsSection = LimitsSection(),
    val authorization: AuthorizationSection = AuthorizationSection(),
    val admin: AdminSection = AdminSection(),
+    val negentropy: NegentropySection = NegentropySection(),
 ) {
    /**
     * Maps the `[info]` section into a [RelayInfo] used by the NIP-11
@@ -141,6 +142,18 @@ data class RelayConfig(
         * for trusted-input scenarios (test fixtures, mirror replays).
         */
        val verify_signatures: Boolean = true,
+        /**
+         * Run signature verification in parallel inside the IngestQueue
+         * (CPU fan-out across `Dispatchers.Default`) instead of serially
+         * on each connection's WebSocket pump. Tier-3 of the
+         * `event-ingestion-batching` plan. Wins scale with how many
+         * EVENTs a single connection sends back-to-back: ~CPU_COUNT×
+         * verify-step speed-up on burst publishes. Set false to keep
+         * the legacy in-policy verify path.
+         *
+         * Only takes effect when [verify_signatures] is also true.
+         */
+        val parallel_verify: Boolean = true,
    )

    data class LimitsSection(
@@ -148,6 +161,34 @@ data class RelayConfig(
        val max_ws_frame_bytes: Int? = null,
    )

+    /**
+     * NIP-77 negentropy tuning. Defaults track strfry
+     * (`hoytech/strfry`) so a Geode relay accepts the same workload
+     * shape and exchanges the same NEG-MSG round-trip size as
+     * strfry — the de-facto reference implementation.
+     *
+     * - [frame_size_limit] mirrors strfry's hard-coded
+     *   `Negentropy ne(storage, 500'000)` in `RelayNegentropy.cpp`.
+     *   Hex-encoded that's ~1 MB on the wire per NEG-MSG; ensure
+     *   `[limits].max_ws_frame_bytes` (when set) is at least double
+     *   this or NEG-MSGs get truncated by the WS layer.
+     * - [max_sync_events] mirrors strfry's
+     *   `relay__negentropy__maxSyncEvents`. NEG-OPEN whose snapshot
+     *   exceeds this returns
+     *   `["NEG-ERR", "<subId>", "blocked: too many query results"]`.
+     * - [max_sessions_per_connection] caps concurrent NEG-OPEN
+     *   sessions held by a single connection. strfry shares its
+     *   200-cap with REQ subs via `relay__maxSubsPerConnection`;
+     *   Geode counts NEG independently for now (REQ has no cap yet).
+     *   Overflow returns NOTICE
+     *   `"too many concurrent NEG requests"` (matches strfry).
+     */
+    data class NegentropySection(
+        val frame_size_limit: Long = 500_000L,
+        val max_sync_events: Int = 1_000_000,
+        val max_sessions_per_connection: Int = 200,
+    )
+
    data class AuthorizationSection(
        val pubkey_whitelist: List<String> = emptyList(),
        val pubkey_blacklist: List<String> = emptyList(),
@@ -24,6 +24,7 @@ import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
 import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
@@ -33,6 +34,7 @@ import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
 import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
 import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage
 import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession
+import com.vitorpamplona.quartz.nip77Negentropy.NegentropySettings
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.runBlocking
@@ -237,12 +239,82 @@ class Nip77NegentropyTest {
                val response = client.nextMessage()
                assertTrue(response is NegErrMessage, "expected NEG-ERR, got ${response::class.simpleName}")
                assertEquals("ghost-sub", response.subId)
-                assertTrue(response.reason.contains("no negentropy session"))
+                // strfry-parity wording — clients in the wild string-match this.
+                assertEquals("closed: unknown subscription handle", response.reason)
            } finally {
                client.close()
            }
        }

+    @Test
+    fun negOpenSnapshotOverflowReturnsStrFryNegErr() =
+        runBlocking {
+            // Tiny cap so the test is fast. Preload more events than the
+            // cap so NEG-OPEN must reject — strfry's parity behaviour for
+            // `relay__negentropy__maxSyncEvents`.
+            val capped = RelayHub(negentropySettings = NegentropySettings(maxSyncEvents = 5))
+            try {
+                val capUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7771/")
+                val events = makeEvents(20)
+                capped.getOrCreate(capUrl).preload(events)
+
+                val client = WireClient(capped, capUrl)
+                try {
+                    val session =
+                        NegentropySession(
+                            subId = "neg-overflow",
+                            filter = Filter(kinds = listOf(1)),
+                            localEvents = emptyList(),
+                        )
+                    client.send(OptimizedJsonMapper.toJson(session.open()))
+
+                    val response = client.nextMessage()
+                    assertTrue(response is NegErrMessage, "expected NEG-ERR, got ${response::class.simpleName}")
+                    assertEquals("neg-overflow", response.subId)
+                    // strfry-parity wording.
+                    assertEquals("blocked: too many query results", response.reason)
+                } finally {
+                    client.close()
+                }
+            } finally {
+                capped.close()
+            }
+        }
+
+    @Test
+    fun negOpenPerConnectionCapEmitsNotice() =
+        runBlocking {
+            // Cap = 2, so the third NEG-OPEN on one connection should
+            // be rejected with a NOTICE (matching strfry's wording).
+            val capped = RelayHub(negentropySettings = NegentropySettings(maxSessionsPerConnection = 2))
+            try {
+                val capUrl = RelayUrlNormalizer.normalize("ws://127.0.0.1:7772/")
+                capped.getOrCreate(capUrl).preload(makeEvents(3))
+
+                val client = WireClient(capped, capUrl)
+                try {
+                    repeat(2) { i ->
+                        val s = NegentropySession("ok-$i", Filter(kinds = listOf(1)), localEvents = emptyList())
+                        client.send(OptimizedJsonMapper.toJson(s.open()))
+                        // Drain the NEG-MSG response so the next OPEN goes
+                        // through cleanly.
+                        client.nextMessage() as NegMsgMessage
+                    }
+
+                    // Third OPEN — should be rejected with a NOTICE.
+                    val third = NegentropySession("third", Filter(kinds = listOf(1)), localEvents = emptyList())
+                    client.send(OptimizedJsonMapper.toJson(third.open()))
+                    val response = client.nextMessage()
+                    assertTrue(response is NoticeMessage, "expected NOTICE, got ${response::class.simpleName}")
+                    assertEquals("too many concurrent NEG requests", response.message)
+                } finally {
+                    client.close()
+                }
+            } finally {
+                capped.close()
+            }
+        }
+
    @Test
    fun negOpenWithSameSubIdReplacesPriorSession() =
        runBlocking {
@@ -0,0 +1,186 @@
+/*
+ * 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.geode.interop
+
+import com.vitorpamplona.geode.LocalRelayServer
+import com.vitorpamplona.geode.Relay
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll
+import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndConfirm
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
+import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.runBlocking
+import okhttp3.OkHttpClient
+import kotlin.test.AfterTest
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+/**
+ * The Kotlin counterpart to strfry's `test/syncTest.pl`: stand up two
+ * Geode relays on real WebSocket endpoints, give each a partially
+ * overlapping corpus, and converge them via NIP-77.
+ *
+ * The driver mirrors `strfry sync ws://other --dir both`:
+ *
+ *  1. Read the local relay's snapshot for the negotiated filter.
+ *  2. Open NEG-OPEN against the remote with that snapshot.
+ *  3. Drive NEG-MSG round trips until the client-side
+ *     [com.vitorpamplona.quartz.nip77Negentropy.NegentropySession]
+ *     reports completion.
+ *  4. `needIds`: REQ them from the remote, insert into the local relay.
+ *  5. `haveIds`: fetch from the local relay, publish to the remote.
+ *
+ * After the round, both relays must hold the union of the original
+ * corpora. We assert via REQ on each side; an `id`-filter that returns
+ * every event we expect, and nothing more, proves convergence
+ * end-to-end through the NIP-77 server pipeline (`NegSessionRegistry`
+ * → `NegentropyServerSession` → `IEventStore.snapshotIdsForNegentropy`).
+ *
+ * Equivalent to strfry's `runSyncTests.pl` "full DB sync" case at
+ * small scale. Larger corpora belong in `LoadBenchmark`.
+ */
+class GeodeVsGeodeNegentropySyncTest {
+    private lateinit var relayA: Relay
+    private lateinit var relayB: Relay
+    private lateinit var serverA: LocalRelayServer
+    private lateinit var serverB: LocalRelayServer
+    private lateinit var scope: CoroutineScope
+    private lateinit var client: NostrClient
+    private val httpClient = OkHttpClient.Builder().build()
+
+    @BeforeTest
+    fun setup() {
+        // The placeholder URLs are normalised so the relay accepts
+        // them; ports come from the autobind below via [server.url].
+        relayA = Relay(url = "ws://127.0.0.1:7771/".normalizeRelayUrl())
+        relayB = Relay(url = "ws://127.0.0.1:7772/".normalizeRelayUrl())
+        serverA = LocalRelayServer(relayA, host = "127.0.0.1", port = 0).start()
+        serverB = LocalRelayServer(relayB, host = "127.0.0.1", port = 0).start()
+        scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+        client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> httpClient }, scope)
+    }
+
+    @AfterTest
+    fun teardown() {
+        client.disconnect()
+        scope.cancel()
+        serverA.stop()
+        serverB.stop()
+        relayA.close()
+        relayB.close()
+    }
+
+    /** Generates [count] signed text notes with monotonic createdAt. */
+    private fun makeEvents(
+        count: Int,
+        seed: Long = 1_700_000_000L,
+    ): List<Event> {
+        val signer = NostrSignerSync(KeyPair())
+        return List(count) { i ->
+            signer.sign(TextNoteEvent.build("event-$seed-$i", createdAt = seed + i))
+        }
+    }
+
+    private val driver by lazy { InteropSyncDriver(httpClient) }
+
+    @Test
+    fun bidirectionalSyncConvergesTwoRelays() =
+        runBlocking {
+            // Universe of 20 events. Relay A holds [0..14], Relay B
+            // holds [5..19]. Overlap [5..14], A-only [0..4], B-only
+            // [15..19]. After bidirectional sync both must hold [0..19].
+            val all = makeEvents(20)
+            val aEvents = all.subList(0, 15)
+            val bEvents = all.subList(5, 20)
+            relayA.preload(aEvents)
+            relayB.preload(bEvents)
+
+            val filter = Filter(kinds = listOf(1))
+            val urlA = serverA.url.normalizeRelayUrl()
+            val urlB = serverB.url.normalizeRelayUrl()
+
+            // B negotiates the symmetric difference with A.
+            val diff = driver.negotiate(serverA.url, filter, bEvents)
+            assertNull(diff.error, "negotiation must not error: ${diff.error}")
+            // From B's perspective: needs A-only, has B-only.
+            assertEquals(
+                aEvents.subList(0, 5).map { it.id }.toSet(),
+                diff.needIds,
+                "B should NEED [0..4] from A",
+            )
+            assertEquals(
+                bEvents.subList(10, 15).map { it.id }.toSet(),
+                diff.haveIds,
+                "B should announce HAVE for [15..19]",
+            )
+
+            // Close the loop: fetch needs from A, push haves to A.
+            val needFromA =
+                client.fetchAll(relay = urlA, filter = Filter(ids = diff.needIds.toList()))
+            relayB.preload(needFromA)
+
+            for (id in diff.haveIds) {
+                client.publishAndConfirm(bEvents.first { it.id == id }, setOf(urlA))
+            }
+
+            // --- Verify convergence ---
+            val expected = all.map { it.id }.toSet()
+            assertEquals(expected, idsOnRelay(urlA, filter), "Relay A should hold every event")
+            assertEquals(expected, idsOnRelay(urlB, filter), "Relay B should hold every event")
+        }
+
+    @Test
+    fun negentropyConvergesInBoundedRoundsOnSmallCorpus() =
+        runBlocking {
+            val all = makeEvents(200)
+            relayA.preload(all.subList(0, 150))
+            val bEvents = all.subList(50, 200)
+            relayB.preload(bEvents)
+
+            val res = driver.negotiate(serverA.url, Filter(kinds = listOf(1)), bEvents)
+            assertNull(res.error)
+            assertEquals(50, res.needIds.size, "B should need [0..49]")
+            assertEquals(50, res.haveIds.size, "B should have [150..199]")
+            // strfry typically converges these in ≤5 rounds; ≤16 is
+            // generous headroom that still catches regressions.
+            assertTrue(res.rounds <= 16, "expected ≤16 NEG-MSG rounds, got ${res.rounds}")
+        }
+
+    /** Every event id matching [filter] visible on the relay at [url] via REQ. */
+    private suspend fun idsOnRelay(
+        url: NormalizedRelayUrl,
+        filter: Filter,
+    ): Set<HexKey> = client.fetchAll(relay = url, filter = filter).map { it.id }.toSet()
+}
@@ -0,0 +1,220 @@
+/*
+ * 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.geode.interop
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
+import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.Response
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import java.io.File
+import java.net.ServerSocket
+import java.net.Socket
+import kotlin.io.path.createTempDirectory
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+/**
+ * Reciprocal interop test: Geode's NIP-77 client driving a real
+ * `strfry` instance.
+ *
+ * **Opt-in.** Skipped unless `STRFRY_BIN` env var (or
+ * `-Dstrfry.bin=...`) points at a `strfry` binary. When unset, the
+ * test prints a `[skip]` line and returns. Mirrors the gate
+ * `LoadBenchmark` uses for `-DrunLoadBenchmark`:
+ *
+ *   STRFRY_BIN=/usr/local/bin/strfry ./gradlew :geode:test \
+ *       --tests "*GeodeVsStrfryNegentropySyncTest*"
+ *
+ * The strfry process is booted on a free loopback port with a fresh
+ * temp LMDB dir. We feed events into it via the NIP-01 EVENT wire
+ * (no `strfry import`), then run [InteropSyncDriver] against it.
+ * Strfry's `RelayNegentropy.cpp` answers the same NIP-77 wire we test
+ * against Geode — passing both is byte-shape interop, not just
+ * "passes our own tests".
+ */
+class GeodeVsStrfryNegentropySyncTest {
+    private val strfryBin: String? =
+        System.getenv("STRFRY_BIN") ?: System.getProperty("strfry.bin")
+    private val enabled = strfryBin != null
+
+    private lateinit var strfryDir: File
+    private var strfryProcess: Process? = null
+    private val httpClient by lazy { OkHttpClient.Builder().build() }
+
+    @AfterTest
+    fun teardown() {
+        strfryProcess?.destroy()
+        strfryProcess?.waitFor()
+        if (::strfryDir.isInitialized) strfryDir.deleteRecursively()
+    }
+
+    /**
+     * Boots a strfry instance in a temp LMDB dir on a free port.
+     * Writes the smallest config strfry accepts, starts the
+     * subprocess, and polls until the WebSocket port is reachable.
+     *
+     * The config is intentionally minimal — strfry's defaults
+     * (negentropy on, sane limits) are what we want to test against.
+     * Adding speculative config keys risks failing on schema drift.
+     */
+    private fun startStrfry(): String {
+        val port = ServerSocket(0).use { it.localPort }
+        strfryDir = createTempDirectory(prefix = "strfry-interop-").toFile()
+        val configFile = File(strfryDir, "strfry.conf")
+        configFile.writeText(
+            """
+            db = "${strfryDir.absolutePath}/strfry-db"
+            relay {
+                bind = "127.0.0.1"
+                port = $port
+            }
+            """.trimIndent(),
+        )
+        File(strfryDir, "strfry-db").mkdirs()
+        strfryProcess =
+            ProcessBuilder(strfryBin, "--config", configFile.absolutePath, "relay")
+                .redirectErrorStream(true)
+                .redirectOutput(File(strfryDir, "strfry.log"))
+                .start()
+
+        val deadline = System.currentTimeMillis() + 5_000
+        while (System.currentTimeMillis() < deadline) {
+            runCatching {
+                Socket("127.0.0.1", port).close()
+                return "ws://127.0.0.1:$port/"
+            }
+            Thread.sleep(100)
+        }
+        throw IllegalStateException(
+            "strfry did not start within 5s; log: " +
+                File(strfryDir, "strfry.log").readText(),
+        )
+    }
+
+    private fun makeEvents(count: Int): List<Event> {
+        val signer = NostrSignerSync(KeyPair())
+        val now = 1_700_000_000L
+        return List(count) { i ->
+            signer.sign(TextNoteEvent.build("strfry-interop-$i", createdAt = now + i))
+        }
+    }
+
+    /**
+     * Push every event in [events] to the relay at [wsUrl] over a
+     * one-shot WebSocket, waiting for an `OK` response per event.
+     *
+     * Used to seed strfry from the same `Event` objects Geode's
+     * `Relay.preload` accepts — that way both sides start from a
+     * byte-identical corpus.
+     */
+    private suspend fun publishToStrfry(
+        wsUrl: String,
+        events: List<Event>,
+    ) {
+        val incoming = Channel<String>(UNLIMITED)
+        val ws =
+            httpClient.newWebSocket(
+                Request.Builder().url(wsUrl.replace("ws://", "http://")).build(),
+                object : WebSocketListener() {
+                    override fun onMessage(
+                        webSocket: WebSocket,
+                        text: String,
+                    ) {
+                        incoming.trySend(text)
+                    }
+
+                    override fun onFailure(
+                        webSocket: WebSocket,
+                        t: Throwable,
+                        response: Response?,
+                    ) {
+                        incoming.close(t)
+                    }
+                },
+            )
+        try {
+            for (e in events) {
+                check(ws.send("""["EVENT",${OptimizedJsonMapper.toJson(e)}]""")) {
+                    "publish to strfry failed"
+                }
+            }
+            var oks = 0
+            withTimeout(30_000) {
+                while (oks < events.size) {
+                    if (incoming.receive().startsWith("[\"OK\"")) oks++
+                }
+            }
+        } finally {
+            ws.close(1000, "preload-done")
+        }
+    }
+
+    @Test
+    fun geodeReconcilesAgainstStrfryRelay() =
+        runBlocking {
+            if (!enabled) {
+                println("[skip] GeodeVsStrfryNegentropySyncTest — set STRFRY_BIN=/path/to/strfry to enable")
+                return@runBlocking
+            }
+            val strfryWs = startStrfry()
+
+            // Same overlap shape as GeodeVsGeodeNegentropySyncTest so
+            // results are directly comparable: A=[0..14], local=[5..19].
+            val all = makeEvents(20)
+            val strfryEvents = all.subList(0, 15)
+            val localEvents = all.subList(5, 20)
+
+            publishToStrfry(strfryWs, strfryEvents)
+
+            val filter = Filter(kinds = listOf(1))
+
+            // The wire we care about: kmp-negentropy (client) talking
+            // to strfry's `Negentropy ne(storage, 500'000)` (server).
+            // Symmetric difference must match the Geode-vs-Geode case.
+            val res = InteropSyncDriver(httpClient).negotiate(strfryWs, filter, localEvents)
+            assertNull(res.error, "Geode↔strfry NEG must not error: ${res.error}")
+            assertEquals(
+                strfryEvents.subList(0, 5).map { it.id }.toSet(),
+                res.needIds,
+                "client should NEED [0..4] from strfry",
+            )
+            assertEquals(
+                localEvents.subList(10, 15).map { it.id }.toSet(),
+                res.haveIds,
+                "client should announce HAVE for [15..19]",
+            )
+            assertTrue(res.rounds <= 16, "expected ≤16 NEG-MSG rounds, got ${res.rounds}")
+        }
+}
@@ -0,0 +1,191 @@
+/*
+ * 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.geode.interop
+
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
+import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.NoticeMessage
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip77Negentropy.NegErrMessage
+import com.vitorpamplona.quartz.nip77Negentropy.NegMsgMessage
+import com.vitorpamplona.quartz.nip77Negentropy.NegentropySession
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
+import kotlinx.coroutines.withTimeout
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.Response
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import kotlin.test.fail
+
+/**
+ * Equivalent of strfry's `test/syncTest.pl` driver for our interop
+ * tests. Drives one round of NIP-77 *negotiation* (NEG-OPEN /
+ * NEG-MSG / NEG-CLOSE) against a real `ws://` endpoint — a Geode
+ * `LocalRelayServer`, a strfry process, or any other NIP-77 relay.
+ *
+ * The driver opens a raw WebSocket — no `NostrClient` overhead —
+ * to keep the wire format under direct control, the same way
+ * `strfry sync` does. That way we exercise the server's NEG-OPEN /
+ * NEG-MSG / NEG-CLOSE handling with no client-side framing or
+ * filter-management indirection.
+ *
+ * Note: this driver only computes the symmetric difference. The
+ * actual *sync* (REQ for `needIds`, EVENT for `haveIds`) is the
+ * caller's job; that's a NIP-01 follow-up, not part of NIP-77.
+ */
+class InteropSyncDriver(
+    private val httpClient: OkHttpClient = OkHttpClient.Builder().build(),
+) {
+    /**
+     * Negotiates the symmetric difference between `localEvents` and
+     * the relay at [wsUrl] under [filter]. Returns the id sets so
+     * the caller can close the loop with REQ / EVENT.
+     *
+     * @param wsUrl the source relay's `ws://…` URL.
+     * @param filter NEG-OPEN filter — usually the broadest filter the
+     *   sync should cover (e.g. `Filter(kinds = listOf(1))`).
+     * @param localEvents events the caller already has; the relay
+     *   reconciles these against its own snapshot.
+     * @param frameSizeLimit `0` lets the relay choose. We pass `0`
+     *   here so the relay's configured cap (500_000 by default) is
+     *   what governs framing — same shape as `strfry sync`.
+     * @param timeoutMs hard timeout on a single NEG-MSG round trip.
+     * @param maxRounds upper bound on round trips. Strfry typically
+     *   converges in ≤5 rounds for 100 k corpora; 64 is a generous
+     *   safety net that catches pathological splits without hanging
+     *   tests forever.
+     */
+    suspend fun negotiate(
+        wsUrl: String,
+        filter: Filter,
+        localEvents: List<Event>,
+        subId: String = "interop-sync",
+        frameSizeLimit: Long = 0,
+        timeoutMs: Long = 30_000L,
+        maxRounds: Int = 64,
+    ): Result {
+        val incoming = Channel<String>(UNLIMITED)
+        val ws =
+            httpClient.newWebSocket(
+                Request.Builder().url(wsUrl.replace("ws://", "http://")).build(),
+                object : WebSocketListener() {
+                    override fun onMessage(
+                        webSocket: WebSocket,
+                        text: String,
+                    ) {
+                        incoming.trySend(text)
+                    }
+
+                    override fun onClosing(
+                        webSocket: WebSocket,
+                        code: Int,
+                        reason: String,
+                    ) {
+                        incoming.close()
+                    }
+
+                    override fun onFailure(
+                        webSocket: WebSocket,
+                        t: Throwable,
+                        response: Response?,
+                    ) {
+                        incoming.close(t)
+                    }
+                },
+            )
+
+        return try {
+            val session = NegentropySession(subId, filter, localEvents, frameSizeLimit)
+
+            // Step 1: NEG-OPEN.
+            check(ws.send(OptimizedJsonMapper.toJson(session.open()))) { "send NEG-OPEN failed" }
+
+            // Step 2: drive NEG-MSG round trips until the client-side
+            // session reports completion.
+            val haveIds = mutableSetOf<HexKey>()
+            val needIds = mutableSetOf<HexKey>()
+            var rounds = 0
+            while (rounds < maxRounds) {
+                val raw = withTimeout(timeoutMs) { incoming.receive() }
+                val msg = OptimizedJsonMapper.fromJsonToMessage(raw)
+                rounds++
+                when (msg) {
+                    is NegErrMessage -> {
+                        return Result(
+                            haveIds = haveIds,
+                            needIds = needIds,
+                            rounds = rounds,
+                            error = "${msg.subId}: ${msg.reason}",
+                        )
+                    }
+
+                    is NoticeMessage -> {
+                        return Result(
+                            haveIds = haveIds,
+                            needIds = needIds,
+                            rounds = rounds,
+                            error = "NOTICE: ${msg.message}",
+                        )
+                    }
+
+                    is NegMsgMessage -> {
+                        val r = session.processMessage(msg.message)
+                        haveIds += r.haveIds
+                        needIds += r.needIds
+                        if (r.isComplete()) {
+                            return Result(haveIds, needIds, rounds, error = null)
+                        }
+                        check(ws.send(OptimizedJsonMapper.toJson(r.nextCmd!!))) {
+                            "send NEG-MSG failed"
+                        }
+                    }
+
+                    else -> {
+                        fail("unexpected message during NEG sync: ${msg::class.simpleName}")
+                    }
+                }
+            }
+            Result(haveIds, needIds, rounds, error = "did not converge in $maxRounds rounds")
+        } finally {
+            ws.close(1000, "interop-test-done")
+        }
+    }
+
+    /**
+     * Result of a reconciliation. `error` is non-null on
+     * NEG-ERR/NOTICE/timeout; otherwise the id-set fields are
+     * authoritative.
+     *
+     * @param haveIds events the client (us) had that the relay did not.
+     * @param needIds events the relay had that the client (us) lacked.
+     * @param rounds NEG-MSG round trips, including the one carrying
+     *   the terminator.
+     */
+    data class Result(
+        val haveIds: Set<HexKey>,
+        val needIds: Set<HexKey>,
+        val rounds: Int,
+        val error: String?,
+    )
+}
@@ -365,6 +365,52 @@ class LoadBenchmark {
            }
        }

+    /**
+     * Same workload as [publishThroughputSingleClient] (sequential
+     * publish-and-confirm on one connection) — kept as a regression
+     * floor for the group-commit code path. Synchronous publishes
+     * never coalesce in the writer (batch size is always 1), so the
+     * EPS here measures per-event SQLite tx cost. The pipelined win
+     * shows up in [publishPipelinedSingleClient].
+     */
+    @Test
+    fun publishGroupCommitSingleClient() =
+        benchmark("publish group-commit single client") {
+            runBenchmarkServer { server, http ->
+                val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+                val client = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope)
+                try {
+                    val signer = NostrSignerSync(KeyPair())
+                    val relayUrl = server.url.normalizeRelayUrl()
+
+                    val n = 10_000
+                    var ok = 0
+                    val elapsed =
+                        measureTime {
+                            runBlocking {
+                                repeat(n) { i ->
+                                    val event = signer.sign(TextNoteEvent.build("group-commit $i"))
+                                    if (client.publishAndConfirm(event, setOf(relayUrl))) ok++
+                                }
+                            }
+                        }
+                    val eps = (n * 1000.0) / elapsed.inWholeMilliseconds
+                    println(
+                        "events=$n ok=$ok elapsedMs=${elapsed.inWholeMilliseconds} eps=${"%.0f".format(eps)}",
+                    )
+                    check(ok == n) { "expected all $n events accepted, got $ok" }
+                    // Floor: the pre-batching baseline was ~760 EPS
+                    // single-client (see plan). Anything below 500
+                    // means the group-commit / ingest-queue rewrite
+                    // regressed the synchronous path.
+                    check(eps > 500) { "synchronous EPS $eps fell below the 500 floor" }
+                } finally {
+                    client.disconnect()
+                    scope.cancel()
+                }
+            }
+        }
+
    /**
     * One publisher, N subscribers. Publishes one EVENT and measures
     * fan-out latency: time from publish to last subscriber receiving.
@@ -614,6 +660,101 @@ class LoadBenchmark {
            }
        }

+    /**
+     * One publisher fires N EVENTs back-to-back without awaiting
+     * intermediate OKs, then collects all OKs by event id. This is
+     * the workload that exercises Tier 2 (per-connection ingest
+     * pipeline) + Tier 1 (group commit) together — multiple events
+     * are in flight on the same connection, so the writer can batch.
+     *
+     * Verifies the relaxed OK contract: every event id receives
+     * exactly one OK frame, in any order.
+     */
+    @Test
+    fun publishPipelinedSingleClient() =
+        benchmark("publish pipelined single client") {
+            runBenchmarkServer { server, http ->
+                val n = 10_000
+                val signer = NostrSignerSync(KeyPair())
+                val events =
+                    runBlocking {
+                        (0 until n).map { i ->
+                            signer.sign(TextNoteEvent.build("pipe $i"))
+                        }
+                    }
+                val ids = events.mapTo(HashSet()) { it.id }
+
+                val httpUrl =
+                    okhttp3.Request
+                        .Builder()
+                        .url(server.url.replace("ws://", "http://"))
+                        .build()
+                val okSeen = AtomicLong()
+                val okFailures = AtomicLong()
+                val unknownIds = AtomicLong()
+                val seenIds =
+                    java.util.concurrent.ConcurrentHashMap
+                        .newKeySet<String>()
+                val done = java.util.concurrent.CountDownLatch(1)
+
+                val ws =
+                    http.newWebSocket(
+                        httpUrl,
+                        object : okhttp3.WebSocketListener() {
+                            override fun onMessage(
+                                webSocket: okhttp3.WebSocket,
+                                text: String,
+                            ) {
+                                if (!text.startsWith("[\"OK\"")) return
+                                // ["OK","<id>",true|false,"<reason>"] —
+                                // a tiny string scan is enough for a
+                                // bench. Index 6 is past `["OK","`.
+                                val idStart = 7
+                                val idEnd = text.indexOf('"', idStart)
+                                if (idEnd <= idStart) return
+                                val id = text.substring(idStart, idEnd)
+                                if (!ids.contains(id)) {
+                                    unknownIds.incrementAndGet()
+                                    return
+                                }
+                                if (!seenIds.add(id)) return
+                                if (text.contains(",true,")) {
+                                    okSeen.incrementAndGet()
+                                } else {
+                                    okFailures.incrementAndGet()
+                                }
+                                if (okSeen.get() + okFailures.get() == n.toLong()) done.countDown()
+                            }
+                        },
+                    )
+
+                val elapsed =
+                    measureTime {
+                        // Burst-send: queue every EVENT to OkHttp's
+                        // outbound buffer without any await, then
+                        // wait for the corresponding OK frames.
+                        for (event in events) {
+                            ws.send("""["EVENT",${event.toJson()}]""")
+                        }
+                        check(done.await(60, java.util.concurrent.TimeUnit.SECONDS)) {
+                            "timed out waiting for OKs: ok=${okSeen.get()} rej=${okFailures.get()} unknown=${unknownIds.get()}"
+                        }
+                    }
+                val eps = (n * 1000.0) / elapsed.inWholeMilliseconds
+                println(
+                    "events=$n ok=${okSeen.get()} rejected=${okFailures.get()} " +
+                        "unknownIds=${unknownIds.get()} elapsedMs=${elapsed.inWholeMilliseconds} eps=${"%.0f".format(eps)}",
+                )
+                check(okSeen.get() == n.toLong()) {
+                    "expected $n accepted OKs, got ${okSeen.get()} (rejected ${okFailures.get()})"
+                }
+                check(seenIds.size == n) {
+                    "expected $n unique OK ids, got ${seenIds.size} — duplicate or missing OKs"
+                }
+                ws.cancel()
+            }
+        }
+
    /**
     * Many concurrent publishers, each on their own WebSocket. Tells
     * us whether the SQLite single-writer bottleneck is the floor or