diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt index e1d490c6cf..f7cf053614 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt @@ -500,6 +500,7 @@ class LoadBenchmark { // FilterIndex's byAuthor bucket. val keys = List(subs) { KeyPair() } val pubkeys = keys.map { it.pubKey.toHexKey() } + val signers = keys.map { NostrSignerSync(it) } // Per-subscriber arrival timestamp. We // overwrite per round; the publish path waits @@ -579,8 +580,7 @@ class LoadBenchmark { val start = System.nanoTime() for (i in 0 until totalEvents) { val target = i % subs - val signer = NostrSignerSync(keys[target]) - val event = signer.sign(TextNoteEvent.build("scale-$i")) + val event = signers[target].sign(TextNoteEvent.build("scale-$i")) arrivalNs.set(target, 0) val pubStart = System.nanoTime() pubClient.publishAndConfirm(event, setOf(relayUrl)) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt index f8835d9ad3..22424d0dfe 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/LiveEventStore.kt @@ -78,22 +78,26 @@ class LiveEventStore( onEach: (Event) -> Unit, onEose: () -> Unit, ) { - // During the historical replay, mark ids the store has + // During the historical replay, record ids the store has // emitted so the live path can dedupe. The index registers // *before* the replay starts (otherwise an event accepted // mid-replay would slip past the live path entirely — same // race the previous SharedFlow-based implementation closed - // with `onSubscription`). Anything the store and the live - // path both observe gets dropped here on the live side. + // with `onSubscription`). // - // Held in an AtomicReference because the live-dispatch - // coroutine (which calls `deliver` from `insert`) needs to - // see the post-EOSE handoff promptly. Once cleared to null, - // the dedupe check short-circuits and every live event is - // forwarded. The set itself is mutated only from the - // historical-replay closure below, which runs on the same - // coroutine that owns `query` — no cross-thread mutation. - val seenIds = AtomicReference?>(HashSet()) + // The set is read from the publisher's coroutine (in + // `deliver`, called synchronously from `insert`) and written + // from this coroutine (the historical-replay closure below). + // It must be a persistent / immutable Set under an + // AtomicReference — wrapping a mutable HashSet would race + // because AtomicReference only protects the reference, not + // the set's internal state. Each `add` is a CAS-loop that + // publishes a new immutable Set; `deliver`'s `load()` always + // sees a fully-constructed snapshot. + // + // Once cleared to null after EOSE, `deliver` short-circuits + // and every live event is forwarded. + val seenIds = AtomicReference?>(emptySet()) val sub = LiveSubscription( @@ -108,7 +112,17 @@ class LiveEventStore( index.register(filters, sub) try { store.query(filters) { event -> - seenIds.load()?.add(event.id) + // CAS-loop on an immutable Set. The historical + // replay is single-writer in steady state, so the + // CAS typically succeeds first try; the loop only + // matters if we lose a race on `seenIds.store(null)` + // (post-EOSE), in which case `current == null` and + // we exit cleanly. + while (true) { + val current = seenIds.load() ?: break + if (event.id in current) break + if (seenIds.compareAndSet(current, current + event.id)) break + } onEach(event) } onEose() diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndexTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndexTest.kt index a8e2690441..eebe45b03b 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndexTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterIndexTest.kt @@ -233,6 +233,62 @@ class FilterIndexTest { assertEquals(setOf(s1, s2, s3), visited.toSet()) } + @Test + fun tagsAllFilterMatchesEventsCarryingAllTagValues() { + // tagsAll keys the index on the first single-letter entry + // exactly like `tags`. Events that match the index lookup + // still have to pass `Filter.match` for the AND-of-values + // semantics — the index is a super-set. + val index = FilterIndex() + val s = Sub("s") + index.register(Filter(tagsAll = mapOf("p" to listOf(pTag1))), s) + + assertTrue(s in index.candidatesFor(event(tags = arrayOf(arrayOf("p", pTag1))))) + assertFalse(s in index.candidatesFor(event(tags = arrayOf(arrayOf("p", pTag2))))) + } + + @Test + fun multiCharTagKeyFallsThroughToKindThenUnindexed() { + // The index only buckets single-letter tag keys (NIP-12). + // A filter with only multi-char tag keys should fall through + // to the next dimension. + val index = FilterIndex() + val withKind = Sub("withKind") + val withoutKind = Sub("withoutKind") + + index.register(Filter(tags = mapOf("alt" to listOf("foo")), kinds = listOf(7)), withKind) + index.register(Filter(tags = mapOf("alt" to listOf("foo"))), withoutKind) + + // withKind goes to KindKey(7); withoutKind to Unindexed. + assertTrue(withKind in index.candidatesFor(event(kind = 7))) + assertFalse(withKind in index.candidatesFor(event(kind = 1))) + assertTrue(withoutKind in index.candidatesFor(event(kind = 1))) + assertTrue(withoutKind in index.candidatesFor(event(kind = 7))) + } + + @Test + fun reRegisterAddsAdditionalKeysWithoutDuplicating() { + // Calling register twice on the same subscriber unions the + // bucket assignments — useful when an observer wants to + // listen on multiple narrowing dimensions accumulated over + // time. The bucket sets must dedupe so the subscriber + // appears once in `candidatesFor`. + val index = FilterIndex() + val s = Sub("s") + index.register(Filter(authors = listOf(authorA)), s) + index.register(Filter(kinds = listOf(7)), s) + + // Author hit + kind hit = same subscriber, returned once. + val cands = index.candidatesFor(event(pubkey = authorA, kind = 7)) + assertEquals(1, cands.size) + assertTrue(s in cands) + + // Unregister cleans both dimensions. + index.unregister(s) + assertTrue(index.candidatesFor(event(pubkey = authorA)).isEmpty()) + assertTrue(index.candidatesFor(event(kind = 7)).isEmpty()) + } + @Test fun registerThenUnregisterLeavesNoStaleBuckets() { // Churn test — repeatedly add and remove a subscriber and