From 001d5f0eb77f7ce9429b61bc9a8eaaba91930f80 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 3 Jul 2026 14:50:08 +0000
Subject: [PATCH] =?UTF-8?q?fix:=20port=20the=20O(n=C2=B2)=20replay-dedup?=
 =?UTF-8?q?=20fix=20into=20queryRaw;=20adopt=20IdAndTime=20negentropy=20AP?=
 =?UTF-8?q?I?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Main's giant-REQ fix (18bd3600) replaced query()'s copy-on-add immutable
dedupe set with a spin-locked HashSet, but the rebase left queryRaw —
now the default REQ path — on the old pattern, which would have
reintroduced the O(n²) crawl for large replays. Both paths now share
the mutable-set-under-spinlock shape.

relayBench's sync driver moves to NegentropySession.fromEvents(),
following the session's new List<IdAndTime> constructor.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NeoCvXnTxsKzqurkmjdC46
---
 .../relay/server/backend/LiveEventStore.kt    | 37 ++++++++++++-------
 .../relaybench/bench/SyncBenchmark.kt         |  2 +-
 2 files changed, 24 insertions(+), 15 deletions(-)

diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt
index 199d024a17..c4e8d93590 100644
--- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt
+++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/backend/LiveEventStore.kt
@@ -208,11 +208,12 @@ class LiveEventStore(
 
     /**
      * Zero-decode variant of [query]: the historical replay streams
-     * [com.vitorpamplona.quartz.nip01Core.store.RawEvent] rows straight
-     * from storage (no tags parse, no Event materialization, no
-     * re-serialization), while the live path after EOSE is identical to
-     * [query]'s. Same registration-before-replay ordering and the same
-     * immutable-set dedupe against events accepted mid-replay.
+     * [RawEvent] rows straight from storage (no tags parse, no Event
+     * materialization, no re-serialization), while the live path after
+     * EOSE is identical to [query]'s. Same registration-before-replay
+     * ordering, and the same spin-locked mutable dedupe set — NOT a
+     * copy-on-add immutable set, which made giant replays O(n²) (see
+     * the dedup comment in [query]).
      */
     override suspend fun queryRaw(
         ctx: RequestContext,
@@ -221,14 +222,26 @@ class LiveEventStore(
         onEachLive: (Event) -> Unit,
         onEose: () -> Unit,
     ) {
-        val seenIds = AtomicReference<Set<String>?>(emptySet())
+        val seenLock = AtomicBoolean(false)
+        var seenIds: HashSet<String>? = HashSet(1024)
+
+        fun <R> seenLocked(block: () -> R): R {
+            while (seenLock.exchange(true)) {
+                while (seenLock.load()) { }
+            }
+            try {
+                return block()
+            } finally {
+                seenLock.store(false)
+            }
+        }
 
         val sub =
             LiveSubscription(
                 filters = filters,
                 deliver = { event ->
-                    val seen = seenIds.load()
-                    if (seen != null && seen.contains(event.id)) return@LiveSubscription
+                    val duplicate = seenLocked { seenIds?.contains(event.id) ?: false }
+                    if (duplicate) return@LiveSubscription
                     onEachLive(event)
                 },
             )
@@ -236,15 +249,11 @@ class LiveEventStore(
         index.register(filters, sub)
         try {
             store.rawQuery(filters) { raw ->
-                while (true) {
-                    val current = seenIds.load() ?: break
-                    if (raw.id in current) break
-                    if (seenIds.compareAndSet(current, current + raw.id)) break
-                }
+                seenLocked { seenIds?.add(raw.id) }
                 onEachStored(raw)
             }
             onEose()
-            seenIds.store(null)
+            seenLocked { seenIds = null }
             awaitCancellation()
         } finally {
             index.unregister(sub)
diff --git a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/SyncBenchmark.kt b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/SyncBenchmark.kt
index a002ddaf37..683f854b6b 100644
--- a/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/SyncBenchmark.kt
+++ b/relayBench/src/main/kotlin/com/vitorpamplona/relaybench/bench/SyncBenchmark.kt
@@ -130,7 +130,7 @@ object SyncBenchmark {
         fun elapsedMs() = (System.nanoTime() - start) / 1_000_000.0
 
         try {
-            val session = NegentropySession("bench-sync", Filter(), localEvents, frameSizeLimit = 0)
+            val session = NegentropySession.fromEvents("bench-sync", Filter(), localEvents, frameSizeLimit = 0)
             val open = OptimizedJsonMapper.toJson(session.open())
             bytes += open.length
             check(socket.send(open)) { "send NEG-OPEN failed" }