fix(quartz): close HashSet race in LiveEventStore.query dedupe
The previous design wrapped a mutable HashSet in an AtomicReference, which makes the *reference* thread-safe but not the set's internals. The historical-replay closure ran `seenIds.load()?.add(event.id)` from the subscriber's coroutine while `deliver` ran `seen.contains(event.id)` from the publisher's coroutine — concurrent add/contains can corrupt buckets or throw CME. The pre-index implementation didn't have this race because both operations ran on the same Flow collector coroutine. Index-driven dispatch put them on different coroutines. Switch to AtomicReference<Set<String>?> over an immutable Set with a CAS-loop on add. Reads in `deliver` always see a fully-constructed snapshot. Single-writer in steady state so the CAS typically succeeds first try; the loop only matters across the `seenIds.store(null)` post-EOSE handoff. Also: - hoist `NostrSignerSync(keys[target])` out of the per-iteration loop in LoadBenchmark.fanoutScaling. - add FilterIndex tests for `tagsAll` selection, multi-char tag fall-through, and re-register key-union semantics.
This commit is contained in:
@@ -500,6 +500,7 @@ class LoadBenchmark {
|
|||||||
// FilterIndex's byAuthor bucket.
|
// FilterIndex's byAuthor bucket.
|
||||||
val keys = List(subs) { KeyPair() }
|
val keys = List(subs) { KeyPair() }
|
||||||
val pubkeys = keys.map { it.pubKey.toHexKey() }
|
val pubkeys = keys.map { it.pubKey.toHexKey() }
|
||||||
|
val signers = keys.map { NostrSignerSync(it) }
|
||||||
|
|
||||||
// Per-subscriber arrival timestamp. We
|
// Per-subscriber arrival timestamp. We
|
||||||
// overwrite per round; the publish path waits
|
// overwrite per round; the publish path waits
|
||||||
@@ -579,8 +580,7 @@ class LoadBenchmark {
|
|||||||
val start = System.nanoTime()
|
val start = System.nanoTime()
|
||||||
for (i in 0 until totalEvents) {
|
for (i in 0 until totalEvents) {
|
||||||
val target = i % subs
|
val target = i % subs
|
||||||
val signer = NostrSignerSync(keys[target])
|
val event = signers[target].sign(TextNoteEvent.build("scale-$i"))
|
||||||
val event = signer.sign(TextNoteEvent.build("scale-$i"))
|
|
||||||
arrivalNs.set(target, 0)
|
arrivalNs.set(target, 0)
|
||||||
val pubStart = System.nanoTime()
|
val pubStart = System.nanoTime()
|
||||||
pubClient.publishAndConfirm(event, setOf(relayUrl))
|
pubClient.publishAndConfirm(event, setOf(relayUrl))
|
||||||
|
|||||||
+26
-12
@@ -78,22 +78,26 @@ class LiveEventStore(
|
|||||||
onEach: (Event) -> Unit,
|
onEach: (Event) -> Unit,
|
||||||
onEose: () -> 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
|
// emitted so the live path can dedupe. The index registers
|
||||||
// *before* the replay starts (otherwise an event accepted
|
// *before* the replay starts (otherwise an event accepted
|
||||||
// mid-replay would slip past the live path entirely — same
|
// mid-replay would slip past the live path entirely — same
|
||||||
// race the previous SharedFlow-based implementation closed
|
// race the previous SharedFlow-based implementation closed
|
||||||
// with `onSubscription`). Anything the store and the live
|
// with `onSubscription`).
|
||||||
// path both observe gets dropped here on the live side.
|
|
||||||
//
|
//
|
||||||
// Held in an AtomicReference because the live-dispatch
|
// The set is read from the publisher's coroutine (in
|
||||||
// coroutine (which calls `deliver` from `insert`) needs to
|
// `deliver`, called synchronously from `insert`) and written
|
||||||
// see the post-EOSE handoff promptly. Once cleared to null,
|
// from this coroutine (the historical-replay closure below).
|
||||||
// the dedupe check short-circuits and every live event is
|
// It must be a persistent / immutable Set under an
|
||||||
// forwarded. The set itself is mutated only from the
|
// AtomicReference — wrapping a mutable HashSet would race
|
||||||
// historical-replay closure below, which runs on the same
|
// because AtomicReference only protects the reference, not
|
||||||
// coroutine that owns `query` — no cross-thread mutation.
|
// the set's internal state. Each `add` is a CAS-loop that
|
||||||
val seenIds = AtomicReference<HashSet<String>?>(HashSet())
|
// 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<Set<String>?>(emptySet())
|
||||||
|
|
||||||
val sub =
|
val sub =
|
||||||
LiveSubscription(
|
LiveSubscription(
|
||||||
@@ -108,7 +112,17 @@ class LiveEventStore(
|
|||||||
index.register(filters, sub)
|
index.register(filters, sub)
|
||||||
try {
|
try {
|
||||||
store.query<Event>(filters) { event ->
|
store.query<Event>(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)
|
onEach(event)
|
||||||
}
|
}
|
||||||
onEose()
|
onEose()
|
||||||
|
|||||||
+56
@@ -233,6 +233,62 @@ class FilterIndexTest {
|
|||||||
assertEquals(setOf(s1, s2, s3), visited.toSet())
|
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<Sub>()
|
||||||
|
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<Sub>()
|
||||||
|
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<Sub>()
|
||||||
|
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
|
@Test
|
||||||
fun registerThenUnregisterLeavesNoStaleBuckets() {
|
fun registerThenUnregisterLeavesNoStaleBuckets() {
|
||||||
// Churn test — repeatedly add and remove a subscriber and
|
// Churn test — repeatedly add and remove a subscriber and
|
||||||
|
|||||||
Reference in New Issue
Block a user