fix: port the O(n²) replay-dedup fix into queryRaw; adopt IdAndTime negentropy API

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

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)

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" }