From 0b0abeebbbe594c8afab974943a30ed148d75b86 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 28 Jul 2026 17:23:50 +0000 Subject: [PATCH 1/5] fix: prevent duplicate LazyColumn key from observeNotes on addressable updates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NoteListMatchingFilter (backing LocalCache.observeNotes) stored notes in a ConcurrentSkipListSet ordered by CreatedAtIdHexComparator. AddressableNotes are mutable: when a newer replaceable event arrives, LocalCache swaps the event on the SAME note instance (consumeBaseReplaceable -> loadEvent), changing its createdAt in place, then re-notifies observers. A sorted set cannot survive a member's sort key mutating underneath it — the moved node is no longer found on the add() search path, so the same note gets inserted a second time and the emitted list carries a duplicate idHex. The App Recommendations screen keys its LazyColumn on note.idHex (an AddressableNote's address, e.g. 31990::nostr-dvm-labeler), so the duplicate crashed with IllegalArgumentException: "Key ... was already used". Dedupe by the immutable idHex instead of a createdAt-ordered set; ordering is computed fresh on each emission. Adds a regression test reproducing the multi-item corruption path. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ah1aCniyjnzc27x4pwq2Df --- .../observables/NoteListMatchingFilter.kt | 49 +++++-- .../observables/NoteListMatchingFilterTest.kt | 121 ++++++++++++++++++ 2 files changed, 156 insertions(+), 14 deletions(-) create mode 100644 commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt index e938fb2882..07c82c1dc1 100644 --- a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt @@ -24,22 +24,34 @@ import com.vitorpamplona.amethyst.commons.model.AddressableNote import com.vitorpamplona.amethyst.commons.model.Note import com.vitorpamplona.quartz.nip01Core.core.AddressableEvent import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import java.util.SortedSet -import java.util.concurrent.ConcurrentSkipListSet +import java.util.concurrent.ConcurrentHashMap /** * Creates a list of notes (regular and addressable) * that only gets updated when a new note appears. * - * New versions of addressables do not update the list + * New versions of addressables do not update the list. + * + * Membership is keyed by the immutable [Note.idHex] rather than kept in a + * sorted set ordered by createdAt. AddressableNotes are mutable: when a newer + * version of a replaceable event arrives, LocalCache swaps the event on the + * SAME note instance, changing its createdAt in place. A + * ConcurrentSkipListSet ordered on that createdAt cannot survive the change — + * the moved node is no longer found by add()/remove(), so the note ends up + * inserted twice and the emitted list carries a duplicate idHex, crashing any + * LazyColumn keyed on it. Deduping by idHex keeps membership correct + * regardless of createdAt changes; the display order is computed fresh on each + * emission. */ class NoteListMatchingFilter( private val filter: Filter, private val atOnce: (filter: Filter) -> SortedSet, private val update: (List) -> Unit, ) : Observable { - var currentResults: ConcurrentSkipListSet = ConcurrentSkipListSet(CreatedAtIdHexComparator) + val currentResults: ConcurrentHashMap = ConcurrentHashMap() override fun new( event: Event, @@ -47,26 +59,35 @@ class NoteListMatchingFilter( ) { if (event is AddressableEvent && note !is AddressableNote) return - if (filter.match(event)) { - if (currentResults.add(note)) { - val limit = filter.limit - if (limit != null && currentResults.size > limit) { - currentResults.remove(currentResults.last()) - } + // New versions of addressables do not update the list. + if (currentResults.containsKey(note.idHex)) return - update(currentResults.toList()) + if (filter.match(event)) { + currentResults[note.idHex] = note + + val limit = filter.limit + if (limit != null && currentResults.size > limit) { + // Drop the oldest (sorts last under CreatedAtIdHexComparator). + currentResults.values.maxWithOrNull(CreatedAtIdHexComparator)?.let { + currentResults.remove(it.idHex) + } } + + update(snapshot()) } } override fun remove(note: Note) { - if (currentResults.remove(note)) { - update(currentResults.toList()) + if (currentResults.remove(note.idHex) != null) { + update(snapshot()) } } fun init() { - currentResults = ConcurrentSkipListSet(atOnce(filter)) - update(currentResults.toList()) + currentResults.clear() + atOnce(filter).forEach { currentResults[it.idHex] = it } + update(snapshot()) } + + private fun snapshot(): List = currentResults.values.sortedWith(CreatedAtIdHexComparator) } diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt new file mode 100644 index 0000000000..cbdfdd6bfc --- /dev/null +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt @@ -0,0 +1,121 @@ +/* + * 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.amethyst.commons.model.observables + +import com.vitorpamplona.amethyst.commons.model.AddressableNote +import com.vitorpamplona.amethyst.commons.model.Note +import com.vitorpamplona.quartz.nip01Core.core.Address +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip89AppHandlers.definition.AppDefinitionEvent +import com.vitorpamplona.quartz.utils.EventFactory +import java.util.TreeSet +import kotlin.test.Test +import kotlin.test.assertEquals + +class NoteListMatchingFilterTest { + private val author = "d0d0a746b44c9de8422165aef520b1fe041eedf5794f7592505477eeac122c18" + + private val filter = Filter(kinds = listOf(AppDefinitionEvent.KIND)) + + // Amethyst reuses a single AddressableNote instance per address (LocalCache), + // so a newer replaceable event mutates createdAt on the SAME object. + private fun noteFor(dTag: String) = AddressableNote(Address(AppDefinitionEvent.KIND, author, dTag)) + + private fun appDefinition( + dTag: String, + createdAt: Long, + ): Event = + EventFactory.create( + id = "%064x".format(createdAt), + pubKey = author, + createdAt = createdAt, + kind = AppDefinitionEvent.KIND, + tags = arrayOf(arrayOf("d", dTag)), + content = "{}", + sig = "00".repeat(64), + ) + + private fun AddressableNote.load(createdAt: Long) { + event = appDefinition(dTag(), createdAt) + } + + private fun newFilter(sink: (List) -> Unit) = + NoteListMatchingFilter( + filter = filter, + atOnce = { TreeSet(CreatedAtIdHexComparator) }, + update = sink, + ) + + @Test + fun newerVersionOfAnAddressableDoesNotDuplicateTheKey() { + var last: List = emptyList() + val subject = newFilter { last = it } + subject.init() + + // Three app definitions arrive. Their createdAt spread matters: after the + // target moves, the sorted set's search path for the new key must be able + // to bypass the stale node, which is what corrupts a createdAt-ordered set. + val target = noteFor("nostr-dvm-labeler") + val newer = noteFor("other-app") + val newest = noteFor("top-app") + + target.load(1000) + subject.new(target.event!!, target) + newer.load(2000) + subject.new(newer.event!!, newer) + newest.load(4000) + subject.new(newest.event!!, newest) + assertEquals(3, last.size) + + // A newer definition replaces the event on the SAME target instance, + // moving its createdAt from 1000 to 3000 (now between 2000 and 4000). + // LocalCache then notifies the observer again. This must NOT insert the + // note a second time. + target.load(3000) + subject.new(target.event!!, target) + + assertEquals( + listOf(newest.idHex, newer.idHex, target.idHex).sorted(), + last.map { it.idHex }.sorted(), + "each addressable must appear exactly once", + ) + assertEquals(last.size, last.map { it.idHex }.toSet().size, "no duplicate keys") + } + + @Test + fun removeDropsTheNoteEvenAfterCreatedAtChanged() { + var last: List = emptyList() + val subject = newFilter { last = it } + subject.init() + + val target = noteFor("nostr-dvm-labeler") + target.load(1000) + subject.new(target.event!!, target) + assertEquals(1, last.size) + + // The createdAt sort key moves before the delete arrives. + target.load(2000) + subject.remove(target) + + assertEquals(0, last.size, "remove must find the note despite the createdAt change") + } +} From 65e30c0acd32f77c8cd8783fe9e5bee76dee30c2 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 28 Jul 2026 18:03:26 +0000 Subject: [PATCH 2/5] fix: keep observeNotes list sorted while deduping addressables by idHex MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback: keep the incrementally-maintained, created_at-sorted structure (a feed must stay sorted like a relay) instead of re-sorting a hash map on every emission. The root cause is unchanged: there is one Note instance per id/address (LocalCache owns creation), but a note's sort key is mutable — a newer replaceable event swaps the event on the SAME AddressableNote instance, changing created_at in place. A sorted set ordered on that live value corrupts: the moved node leaves the add()/remove() search path, so the same instance is inserted twice and the emitted list carries a duplicate idHex, crashing the App Recommendations LazyColumn (keyed on idHex). Fix: snapshot the sort key into an immutable Entry when the note first enters, order a ConcurrentSkipListSet on that snapshot (never read live again), and index entries by the stable idHex (ConcurrentHashMap + putIfAbsent) so membership stays unique and removal is reliable regardless of later created_at changes. Ordering and "new versions do not update the list" are preserved. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ah1aCniyjnzc27x4pwq2Df --- .../observables/NoteListMatchingFilter.kt | 104 ++++++++++++------ .../observables/NoteListMatchingFilterTest.kt | 21 ++++ 2 files changed, 89 insertions(+), 36 deletions(-) diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt index 07c82c1dc1..29bd6f3147 100644 --- a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt @@ -28,30 +28,56 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import java.util.SortedSet import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.ConcurrentSkipListSet /** - * Creates a list of notes (regular and addressable) - * that only gets updated when a new note appears. + * Creates a list of notes (regular and addressable), sorted by created_at like a + * relay, that only grows when a new note appears. * * New versions of addressables do not update the list. * - * Membership is keyed by the immutable [Note.idHex] rather than kept in a - * sorted set ordered by createdAt. AddressableNotes are mutable: when a newer - * version of a replaceable event arrives, LocalCache swaps the event on the - * SAME note instance, changing its createdAt in place. A - * ConcurrentSkipListSet ordered on that createdAt cannot survive the change — - * the moved node is no longer found by add()/remove(), so the note ends up - * inserted twice and the emitted list carries a duplicate idHex, crashing any - * LazyColumn keyed on it. Deduping by idHex keeps membership correct - * regardless of createdAt changes; the display order is computed fresh on each - * emission. + * There is exactly one [Note] instance per id/address (LocalCache owns their + * creation), so uniqueness is a non-issue in principle — except a note's sort + * key is mutable: a newer replaceable event swaps the event on the SAME + * [AddressableNote] instance, changing its created_at in place. A sorted set + * ordered on that live value cannot survive it — the moved node is no longer on + * the search path of add()/remove(), so the same instance gets inserted twice + * and the emitted list carries a duplicate idHex, crashing any LazyColumn keyed + * on it. + * + * So the sort key is snapshotted into an immutable [Entry] when the note first + * enters and never read live again; the ordered set is keyed on that snapshot + * (stable), and an idHex index keeps membership unique and makes removal reliable + * regardless of later created_at changes. */ class NoteListMatchingFilter( private val filter: Filter, private val atOnce: (filter: Filter) -> SortedSet, private val update: (List) -> Unit, ) : Observable { - val currentResults: ConcurrentHashMap = ConcurrentHashMap() + /** A note plus the sort key captured at insertion time, so ordering never depends on mutable state. */ + private class Entry( + val note: Note, + val createdAt: Long, + val id: HexKey, + ) + + // created_at descending, id ascending as a stable tiebreak. Both fields are + // immutable snapshots, so an Entry never moves once inserted. + private val order = + Comparator { a, b -> + val byCreatedAt = b.createdAt.compareTo(a.createdAt) + if (byCreatedAt != 0) byCreatedAt else a.id.compareTo(b.id) + } + + private val sorted = ConcurrentSkipListSet(order) + private val byId = ConcurrentHashMap() + + private fun entryFor(note: Note): Entry { + // A null event (unresolved note) sorts last, matching CreatedAtIdHexComparator. + val event = note.event + return Entry(note, note.createdAt() ?: Long.MIN_VALUE, event?.id ?: note.idHex) + } override fun new( event: Event, @@ -59,35 +85,41 @@ class NoteListMatchingFilter( ) { if (event is AddressableEvent && note !is AddressableNote) return - // New versions of addressables do not update the list. - if (currentResults.containsKey(note.idHex)) return + if (!filter.match(event)) return - if (filter.match(event)) { - currentResults[note.idHex] = note + val entry = entryFor(note) - val limit = filter.limit - if (limit != null && currentResults.size > limit) { - // Drop the oldest (sorts last under CreatedAtIdHexComparator). - currentResults.values.maxWithOrNull(CreatedAtIdHexComparator)?.let { - currentResults.remove(it.idHex) - } - } + // putIfAbsent gates uniqueness atomically: new versions of an already + // listed addressable return here without touching the sorted set. + if (byId.putIfAbsent(note.idHex, entry) != null) return - update(snapshot()) + sorted.add(entry) + + val limit = filter.limit + if (limit != null && sorted.size > limit) { + sorted.pollLast()?.let { byId.remove(it.note.idHex, it) } } - } - override fun remove(note: Note) { - if (currentResults.remove(note.idHex) != null) { - update(snapshot()) - } - } - - fun init() { - currentResults.clear() - atOnce(filter).forEach { currentResults[it.idHex] = it } update(snapshot()) } - private fun snapshot(): List = currentResults.values.sortedWith(CreatedAtIdHexComparator) + override fun remove(note: Note) { + val entry = byId.remove(note.idHex) ?: return + sorted.remove(entry) + update(snapshot()) + } + + fun init() { + sorted.clear() + byId.clear() + atOnce(filter).forEach { note -> + val entry = entryFor(note) + if (byId.putIfAbsent(note.idHex, entry) == null) { + sorted.add(entry) + } + } + update(snapshot()) + } + + private fun snapshot(): List = sorted.map { it.note } } diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt index cbdfdd6bfc..91457f08df 100644 --- a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt @@ -101,6 +101,27 @@ class NoteListMatchingFilterTest { assertEquals(last.size, last.map { it.idHex }.toSet().size, "no duplicate keys") } + @Test + fun listStaysSortedByCreatedAtDescendingAsNotesArriveOutOfOrder() { + var last: List = emptyList() + val subject = newFilter { last = it } + subject.init() + + val a = noteFor("app-a") + val b = noteFor("app-b") + val c = noteFor("app-c") + + // Arrive out of order; the emitted list must always be newest-first. + a.load(2000) + subject.new(a.event!!, a) + b.load(4000) + subject.new(b.event!!, b) + c.load(1000) + subject.new(c.event!!, c) + + assertEquals(listOf(b.idHex, a.idHex, c.idHex), last.map { it.idHex }) + } + @Test fun removeDropsTheNoteEvenAfterCreatedAtChanged() { var last: List = emptyList() From 636331487abcfd9bbf10c394560834ab40c51d94 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 28 Jul 2026 18:24:37 +0000 Subject: [PATCH 3/5] fix: make observeNotes dedup lock-free and race-safe under concurrent consume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit follow-up. The previous fix used two independent concurrent structures (ConcurrentSkipListSet + ConcurrentHashMap) coordinated with putIfAbsent, but observer callbacks fire from multiple consume threads at once (relay ingest + UI-side justConsume). A new()/remove() interleaving for the same idHex could desync the two structures — remove() clears byId and no-ops on the sorted set before new() has added the entry — leaving an orphan that a later new() duplicates, reintroducing the duplicate-key crash. Keep it lock-free (this observer is used everywhere and needs the throughput): every write to the sorted index for a given idHex now happens inside that key's ConcurrentHashMap.compute critical section, so the sorted set and membership map move together. ConcurrentHashMap stripes per key, so same-idHex ops serialize while different keys stay fully parallel. Invariant: an entry is added to the sorted set only while its key is absent from byId, and every path that frees a key removes its sorted entry first, so the set can never hold two entries for one idHex. Adds concurrency stress tests (with and without a relay limit) that fan out 8 threads hammering new/remove while created_at churns; both fail against the non-atomic version and pass here. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ah1aCniyjnzc27x4pwq2Df --- .../observables/NoteListMatchingFilter.kt | 58 +++++++++----- .../observables/NoteListMatchingFilterTest.kt | 76 +++++++++++++++++-- 2 files changed, 107 insertions(+), 27 deletions(-) diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt index 29bd6f3147..d4de963e52 100644 --- a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt @@ -46,9 +46,19 @@ import java.util.concurrent.ConcurrentSkipListSet * on it. * * So the sort key is snapshotted into an immutable [Entry] when the note first - * enters and never read live again; the ordered set is keyed on that snapshot - * (stable), and an idHex index keeps membership unique and makes removal reliable - * regardless of later created_at changes. + * enters and never read live again; [sorted] is ordered on that snapshot + * (stable) and [byId] is the membership source of truth, keyed by the stable + * idHex. + * + * Observer callbacks fire concurrently from several consume threads (relay + * ingest + UI-side justConsume), and this observer is used everywhere, so it + * stays lock-free: [byId] is a ConcurrentHashMap and every write to [sorted] for + * a given idHex happens INSIDE that key's `compute` critical section. + * ConcurrentHashMap stripes per key, so same-idHex ops serialize while different + * keys run fully in parallel. The invariant that prevents duplicates: an entry + * is added to [sorted] only while its key is absent from [byId], and every path + * that makes a key absent removes its entry from [sorted] first — so [sorted] + * can never hold two entries for one idHex. */ class NoteListMatchingFilter( private val filter: Filter, @@ -87,39 +97,45 @@ class NoteListMatchingFilter( if (!filter.match(event)) return - val entry = entryFor(note) - - // putIfAbsent gates uniqueness atomically: new versions of an already - // listed addressable return here without touching the sorted set. - if (byId.putIfAbsent(note.idHex, entry) != null) return - - sorted.add(entry) + // Add to [sorted] atomically with claiming the idHex slot. New versions + // of an already listed note return the existing entry untouched. + var added = false + byId.compute(note.idHex) { _, existing -> + existing ?: entryFor(note).also { + sorted.add(it) + added = true + } + } + if (!added) return val limit = filter.limit if (limit != null && sorted.size > limit) { + // Drop the oldest (sorts last under [order]). sorted.pollLast()?.let { byId.remove(it.note.idHex, it) } } - update(snapshot()) + update(sorted.map { it.note }) } override fun remove(note: Note) { - val entry = byId.remove(note.idHex) ?: return - sorted.remove(entry) - update(snapshot()) + // Remove from [sorted] atomically with releasing the idHex slot. + var removed = false + byId.compute(note.idHex) { _, existing -> + if (existing != null) { + sorted.remove(existing) + removed = true + } + null + } + if (removed) update(sorted.map { it.note }) } fun init() { sorted.clear() byId.clear() atOnce(filter).forEach { note -> - val entry = entryFor(note) - if (byId.putIfAbsent(note.idHex, entry) == null) { - sorted.add(entry) - } + byId.computeIfAbsent(note.idHex) { entryFor(note).also { sorted.add(it) } } } - update(snapshot()) + update(sorted.map { it.note }) } - - private fun snapshot(): List = sorted.map { it.note } } diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt index 91457f08df..a4711101de 100644 --- a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt @@ -28,8 +28,12 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip89AppHandlers.definition.AppDefinitionEvent import com.vitorpamplona.quartz.utils.EventFactory import java.util.TreeSet +import java.util.concurrent.CountDownLatch +import java.util.concurrent.atomic.AtomicReference +import kotlin.concurrent.thread import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertNull class NoteListMatchingFilterTest { private val author = "d0d0a746b44c9de8422165aef520b1fe041eedf5794f7592505477eeac122c18" @@ -58,12 +62,14 @@ class NoteListMatchingFilterTest { event = appDefinition(dTag(), createdAt) } - private fun newFilter(sink: (List) -> Unit) = - NoteListMatchingFilter( - filter = filter, - atOnce = { TreeSet(CreatedAtIdHexComparator) }, - update = sink, - ) + private fun newFilter( + withFilter: Filter = filter, + sink: (List) -> Unit, + ) = NoteListMatchingFilter( + filter = withFilter, + atOnce = { TreeSet(CreatedAtIdHexComparator) }, + update = sink, + ) @Test fun newerVersionOfAnAddressableDoesNotDuplicateTheKey() { @@ -122,6 +128,64 @@ class NoteListMatchingFilterTest { assertEquals(listOf(b.idHex, a.idHex, c.idHex), last.map { it.idHex }) } + @Test + fun concurrentNewRemoveNeverEmitsDuplicateKeys() { + // No limit: exercises the compute/remove per-key critical sections. + assertNoDuplicateUnderConcurrency(filter) + } + + @Test + fun concurrentNewRemoveWithLimitNeverEmitsDuplicateKeys() { + // With a limit: also exercises the cross-key eviction (pollLast + byId.remove). + assertNoDuplicateUnderConcurrency(Filter(kinds = listOf(AppDefinitionEvent.KIND), limit = 5)) + } + + private fun assertNoDuplicateUnderConcurrency(withFilter: Filter) { + // Observer callbacks fire from several consume threads at once (relay + // ingest + UI-side justConsume). new()/remove() for the same idHex must + // keep the sorted index and the membership map consistent, or a duplicate + // idHex leaks into an emission and crashes the LazyColumn. + val firstViolation = AtomicReference?>(null) + val subject = + newFilter(withFilter) { emitted -> + val ids = emitted.map { it.idHex } + if (ids.size != ids.toSet().size) { + firstViolation.compareAndSet(null, ids) + } + } + subject.init() + + val addresses = (0 until 12).map { "app-$it" } + val notes = addresses.associateWith { noteFor(it) } + val threadCount = 8 + val iterations = 5_000 + val start = CountDownLatch(1) + + val threads = + (0 until threadCount).map { t -> + thread { + start.await() + var seed = t * 31 + 7 + repeat(iterations) { i -> + seed = seed * 1103515245 + 12345 + val note = notes.getValue(addresses[(seed ushr 16) % addresses.size]) + // Move created_at around so the sort key keeps changing under the set. + note.event = appDefinition(note.dTag(), 1_000L + (i % 9)) + if ((seed ushr 8) % 3 == 0) { + subject.remove(note) + } else { + subject.new(note.event!!, note) + } + } + } + } + + start.countDown() + threads.forEach { it.join() } + + assertNull(firstViolation.get(), "an emission carried a duplicate idHex: ${firstViolation.get()}") + } + @Test fun removeDropsTheNoteEvenAfterCreatedAtChanged() { var last: List = emptyList() From 5cb84e16dc589fa378baaae49309a01f7b138b13 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 28 Jul 2026 19:17:57 +0000 Subject: [PATCH 4/5] test: cover the version-note guard path in observeNotes dedup Add coverage for a real consumeBaseReplaceable call the suite was missing: observers are also notified with the "version" note (getOrCreateNote(event.id), a regular Note carrying the AddressableEvent), which the addressable-list guard must drop while still listing the AddressableNote for the same event. Confirmed the concurrency the stress tests exercise is real, not theoretical: relay events are verified+consumed inline on per-relay socket dispatchers, so distinct relays drive new()/remove() on the same note instance concurrently. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ah1aCniyjnzc27x4pwq2Df --- .../observables/NoteListMatchingFilterTest.kt | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt index a4711101de..a77efc8318 100644 --- a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt @@ -186,6 +186,27 @@ class NoteListMatchingFilterTest { assertNull(firstViolation.get(), "an emission carried a duplicate idHex: ${firstViolation.get()}") } + @Test + fun ignoresTheVersionNoteThatHoldsAnAddressableEvent() { + // consumeBaseReplaceable also notifies observers with the "version" note: + // getOrCreateNote(event.id), a regular Note (not AddressableNote) carrying + // the AddressableEvent. It must never enter the addressable list. + var last: List = emptyList() + val subject = newFilter { last = it } + subject.init() + + val event = appDefinition("nostr-dvm-labeler", 1000) + val versionNote = Note(event.id).apply { this.event = event } + subject.new(event, versionNote) + + assertEquals(emptyList(), last) + + // The addressable note for the same event, however, is listed. + val addressable = noteFor("nostr-dvm-labeler").apply { this.event = event } + subject.new(event, addressable) + assertEquals(listOf(addressable.idHex), last.map { it.idHex }) + } + @Test fun removeDropsTheNoteEvenAfterCreatedAtChanged() { var last: List = emptyList() From 58004fa744fa1c986d8d1b1eebbd34655db256c8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 28 Jul 2026 20:42:13 +0000 Subject: [PATCH 5/5] fix: dedup EventListMatchingFilter (observeEvents) and harden emission contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EventListMatchingFilter had the same mutable-sort-key defect as NoteListMatchingFilter: it stored Notes in a ConcurrentSkipListSet ordered by the live created_at, so a newer replaceable version (which mutates the shared AddressableNote in place) stranded its node and let the same instance be inserted twice — the emitted event list then carried the same event twice. It hadn't surfaced as a crash only because its consumers (app recommendations, relay groups, room reactions) happen to dedup downstream. Apply the same capture-key + idHex-dedup + per-key compute design, but preserve EventListMatchingFilter's update-reflecting semantics: an addressable update re-emits (the snapshot reads the refreshed event live off the note) rather than being ignored. It keeps the entry's captured position instead of re-sorting — re-sorting via remove+add let two entries with different captured keys for the same note transiently coexist and both read the same live event, duplicating it. Also harden both filters' emission: a ConcurrentSkipListSet iterator is weakly consistent, so under concurrent add/remove churn a single traversal can momentarily surface a key twice. snapshot() now dedups by idHex so the emitted list — the LazyColumn's source of keys — is always unique, regardless of transient internal states. Corrected the over-claimed "can never hold two" docstrings accordingly. Adds EventListMatchingFilterTest mirroring the note tests: update-reflection, version-note re-emit, sorted order, remove-after-mutation, and two concurrency stress tests (with/without limit) that failed before this fix. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ah1aCniyjnzc27x4pwq2Df --- .../observables/EventListMatchingFilter.kt | 128 +++++++++-- .../observables/NoteListMatchingFilter.kt | 28 ++- .../EventListMatchingFilterTest.kt | 203 ++++++++++++++++++ 3 files changed, 332 insertions(+), 27 deletions(-) create mode 100644 commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilterTest.kt diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilter.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilter.kt index 9117d13cd8..c6373a1267 100644 --- a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilter.kt +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilter.kt @@ -24,22 +24,76 @@ import com.vitorpamplona.amethyst.commons.model.AddressableNote import com.vitorpamplona.amethyst.commons.model.Note import com.vitorpamplona.quartz.nip01Core.core.AddressableEvent import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import java.util.SortedSet +import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentSkipListSet /** - * Creates a list of events (regular and addressable) - * that is updated every time a new event that matches - * the filter is received, including addressables. + * Creates a list of events (regular and addressable), sorted by created_at, that + * is updated every time a new matching event is received — INCLUDING newer + * versions of addressables, whose refreshed content re-emits and re-sorts. + * + * Like [NoteListMatchingFilter], this cannot store mutable [Note]s in a set + * ordered on their live created_at: a newer replaceable event mutates the SAME + * [AddressableNote] instance in place, which strands its node and lets the same + * instance be inserted twice — the emitted list would then carry the same event + * twice. So the sort key is snapshotted into an immutable [Entry], [sorted] is + * ordered on that snapshot, and [byId] is the membership source of truth keyed + * by the stable idHex (the address for addressables, the event id otherwise). + * + * Unlike [NoteListMatchingFilter], an addressable update is NOT ignored: the + * list re-emits so consumers pick up the refreshed [Event] (read live off the + * note). The entry's captured position is kept — re-sorting an updated entry + * would mean a remove+add on [sorted], and two entries with different captured + * keys for the same note would then transiently coexist and both read the same + * live event, duplicating it. Keeping the position matches the original's only + * non-corrupting behavior (it never reliably re-sorted either); consumers that + * care about order re-sort downstream. + * + * Observer callbacks fire concurrently from several consume threads (relay + * ingest + UI-side justConsume), so it stays lock-free: [sorted] is only ever + * written for a key INSIDE that key's `compute` critical section, and only while + * the key is absent. ConcurrentHashMap stripes per key, so same-idHex ops + * serialize while different keys run in parallel. An entry is added only while + * its key is absent from [byId], and every path that frees a key removes its + * entry from [sorted] first. + * + * That keeps [sorted] converged to one entry per idHex, but a + * ConcurrentSkipListSet iterator is only weakly consistent: under concurrent + * add/remove churn a single traversal can momentarily surface a key twice + * (lazy-deleted node not yet unlinked while its replacement is inserted). The + * emitted list must never carry a duplicate id — a LazyColumn keyed on it would + * crash — so [snapshot] deduplicates by the stable idHex as it materializes. */ class EventListMatchingFilter( private val filter: Filter, private val atOnce: (filter: Filter) -> SortedSet, private val update: (List) -> Unit, ) : Observable { - // Keeping this here blocks it from being cleared from memory - var currentResults: ConcurrentSkipListSet = ConcurrentSkipListSet(CreatedAtIdHexComparator) + /** A note plus the sort key captured at insertion time, so ordering never depends on mutable state. */ + private class Entry( + val note: Note, + val createdAt: Long, + val id: HexKey, + ) + + // created_at descending, id ascending as a stable tiebreak. Both fields are + // immutable snapshots, so an Entry never moves once inserted. + private val order = + Comparator { a, b -> + val byCreatedAt = b.createdAt.compareTo(a.createdAt) + if (byCreatedAt != 0) byCreatedAt else a.id.compareTo(b.id) + } + + private val sorted = ConcurrentSkipListSet(order) + private val byId = ConcurrentHashMap() + + private fun entryFor(note: Note): Entry { + val event = note.event + return Entry(note, note.createdAt() ?: Long.MIN_VALUE, event?.id ?: note.idHex) + } @Suppress("UNCHECKED_CAST") override fun new( @@ -47,34 +101,68 @@ class EventListMatchingFilter( note: Note, ) { if (event is AddressableEvent && note !is AddressableNote) { - // event update - if (currentResults.contains(note)) { - update(currentResults.mapNotNull { it.event as? T }) - } + // The "version" note (a regular note holding an addressable event) is + // never stored — the AddressableNote is. Re-emit if that addressable + // is already listed so consumers pick up the refreshed content. + if (byId.containsKey(event.address().toValue())) update(snapshot()) return } - if (filter.match(event)) { - currentResults.add(note) - val limit = filter.limit - if (limit != null && currentResults.size > limit) { - currentResults.remove(currentResults.last()) - } + if (!filter.match(event)) return - update(currentResults.mapNotNull { it.event as? T }) + // Add to [sorted] atomically with claiming the idHex slot, only when the + // key is absent. An update keeps its entry (and position) — the re-emit + // below reflects the refreshed event read live off the note. + var added = false + byId.compute(note.idHex) { _, existing -> + existing ?: entryFor(note).also { + sorted.add(it) + added = true + } } + + if (added) { + val limit = filter.limit + if (limit != null && sorted.size > limit) { + // Drop the oldest (sorts last under [order]). + sorted.pollLast()?.let { byId.remove(it.note.idHex, it) } + } + } + + // Always re-emit on a match: a first insert grows the list, an update + // refreshes the event content the snapshot reads off the note. + update(snapshot()) } @Suppress("UNCHECKED_CAST") override fun remove(note: Note) { - if (currentResults.remove(note)) { - update(currentResults.mapNotNull { it.event as? T }) + var removed = false + byId.compute(note.idHex) { _, existing -> + if (existing != null) { + sorted.remove(existing) + removed = true + } + null } + if (removed) update(snapshot()) } @Suppress("UNCHECKED_CAST") fun init() { - currentResults = ConcurrentSkipListSet(atOnce(filter)) - update(currentResults.mapNotNull { it.event as? T }) + sorted.clear() + byId.clear() + atOnce(filter).forEach { note -> + byId.computeIfAbsent(note.idHex) { entryFor(note).also { sorted.add(it) } } + } + update(snapshot()) + } + + @Suppress("UNCHECKED_CAST") + private fun snapshot(): List { + // Dedup by the stable idHex: the weakly-consistent iterator can transiently + // surface a key twice under concurrent churn. Both would read the same live + // event off the same note, so keeping the first (newest position) is correct. + val seen = HashSet() + return sorted.mapNotNull { e -> if (seen.add(e.note.idHex)) e.note.event as? T else null } } } diff --git a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt index d4de963e52..d2962053fd 100644 --- a/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt +++ b/commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt @@ -55,10 +55,16 @@ import java.util.concurrent.ConcurrentSkipListSet * stays lock-free: [byId] is a ConcurrentHashMap and every write to [sorted] for * a given idHex happens INSIDE that key's `compute` critical section. * ConcurrentHashMap stripes per key, so same-idHex ops serialize while different - * keys run fully in parallel. The invariant that prevents duplicates: an entry - * is added to [sorted] only while its key is absent from [byId], and every path - * that makes a key absent removes its entry from [sorted] first — so [sorted] - * can never hold two entries for one idHex. + * keys run fully in parallel. An entry is added to [sorted] only while its key + * is absent from [byId], and every path that makes a key absent removes its + * entry from [sorted] first, keeping [sorted] converged to one entry per idHex. + * + * That convergence isn't enough on its own: a ConcurrentSkipListSet iterator is + * only weakly consistent, so under concurrent add/remove churn a single + * traversal can momentarily surface a key twice (a lazy-deleted node not yet + * unlinked while its replacement is inserted). The emitted list must never carry + * a duplicate idHex — the LazyColumn keyed on it would crash — so [snapshot] + * deduplicates by idHex as it materializes. */ class NoteListMatchingFilter( private val filter: Filter, @@ -114,7 +120,7 @@ class NoteListMatchingFilter( sorted.pollLast()?.let { byId.remove(it.note.idHex, it) } } - update(sorted.map { it.note }) + update(snapshot()) } override fun remove(note: Note) { @@ -127,7 +133,7 @@ class NoteListMatchingFilter( } null } - if (removed) update(sorted.map { it.note }) + if (removed) update(snapshot()) } fun init() { @@ -136,6 +142,14 @@ class NoteListMatchingFilter( atOnce(filter).forEach { note -> byId.computeIfAbsent(note.idHex) { entryFor(note).also { sorted.add(it) } } } - update(sorted.map { it.note }) + update(snapshot()) + } + + private fun snapshot(): List { + // Dedup by idHex: the weakly-consistent iterator can transiently surface a + // key twice under concurrent churn. Keeping the first (newest position) is + // correct — both nodes point at the same note. + val seen = HashSet() + return sorted.mapNotNull { e -> e.note.takeIf { seen.add(it.idHex) } } } } diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilterTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilterTest.kt new file mode 100644 index 0000000000..b28699e362 --- /dev/null +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilterTest.kt @@ -0,0 +1,203 @@ +/* + * 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.amethyst.commons.model.observables + +import com.vitorpamplona.amethyst.commons.model.AddressableNote +import com.vitorpamplona.amethyst.commons.model.Note +import com.vitorpamplona.quartz.nip01Core.core.Address +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip89AppHandlers.definition.AppDefinitionEvent +import com.vitorpamplona.quartz.utils.EventFactory +import java.util.TreeSet +import java.util.concurrent.CountDownLatch +import java.util.concurrent.atomic.AtomicReference +import kotlin.concurrent.thread +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull + +class EventListMatchingFilterTest { + private val author = "d0d0a746b44c9de8422165aef520b1fe041eedf5794f7592505477eeac122c18" + + private val filter = Filter(kinds = listOf(AppDefinitionEvent.KIND)) + + private fun noteFor(dTag: String) = AddressableNote(Address(AppDefinitionEvent.KIND, author, dTag)) + + private fun appDefinition( + dTag: String, + createdAt: Long, + ): Event = + EventFactory.create( + // Unique per (dTag, createdAt): 8 hex of the dTag hash + 56 hex of createdAt, + // so distinct addresses never collide on an event id. + id = "%08x".format(dTag.hashCode()) + "%056x".format(createdAt), + pubKey = author, + createdAt = createdAt, + kind = AppDefinitionEvent.KIND, + tags = arrayOf(arrayOf("d", dTag)), + content = "{}", + sig = "00".repeat(64), + ) + + private fun AddressableNote.load(createdAt: Long): Event = appDefinition(dTag(), createdAt).also { this.event = it } + + private fun newFilter( + withFilter: Filter = filter, + sink: (List) -> Unit, + ) = EventListMatchingFilter( + filter = withFilter, + atOnce = { TreeSet(CreatedAtIdHexComparator) }, + update = sink, + ) + + @Test + fun newerVersionReflectsUpdatedEventWithoutDuplicate() { + var last: List = emptyList() + val subject = newFilter { last = it } + subject.init() + + // A crowd of app definitions, so a stale skip-set node could be bypassed. + val target = noteFor("nostr-dvm-labeler") + val other = noteFor("other-app") + val top = noteFor("top-app") + + val v1 = target.load(1000) + subject.new(v1, target) + subject.new(other.load(2000), other) + subject.new(top.load(4000), top) + + // Newer version replaces the event on the SAME instance, created_at 1000 -> 3000. + val v2 = target.load(3000) + subject.new(v2, target) + + // Exactly one entry for the target, and it is the NEW version (reflected + re-sorted). + assertEquals(3, last.size, "no duplicate event for the updated addressable") + assertEquals(1, last.count { it.id == v2.id }, "the updated addressable appears exactly once") + assertEquals(0, last.count { it.id == v1.id }, "the old version is gone") + } + + @Test + fun listStaysSortedByCreatedAtDescending() { + var last: List = emptyList() + val subject = newFilter { last = it } + subject.init() + + val a = noteFor("app-a") + val b = noteFor("app-b") + val c = noteFor("app-c") + + val ea = a.load(2000) + subject.new(ea, a) + val eb = b.load(4000) + subject.new(eb, b) + val ec = c.load(1000) + subject.new(ec, c) + + assertEquals(listOf(eb.id, ea.id, ec.id), last.map { it.id }) + } + + @Test + fun versionNoteReEmitsWhenAddressableIsListed() { + val emissions = mutableListOf>() + val subject = newFilter { emissions.add(it) } + subject.init() + + val target = noteFor("nostr-dvm-labeler") + val event = target.load(1000) + subject.new(event, target) + val countAfterInsert = emissions.size + + // The "version" note: a regular Note holding the addressable event. + val versionNote = Note(event.id).apply { this.event = event } + subject.new(event, versionNote) + + // It re-emits (addressable is listed) but never adds a second entry. + assertEquals(countAfterInsert + 1, emissions.size, "version note triggers a re-emit") + assertEquals(listOf(event.id), emissions.last().map { it.id }) + } + + @Test + fun removeDropsTheEventEvenAfterCreatedAtChanged() { + var last: List = emptyList() + val subject = newFilter { last = it } + subject.init() + + val target = noteFor("nostr-dvm-labeler") + subject.new(target.load(1000), target) + assertEquals(1, last.size) + + target.load(2000) // sort key moves before the delete arrives + subject.remove(target) + assertEquals(0, last.size, "remove finds the event despite the created_at change") + } + + @Test + fun concurrentUpdatesNeverEmitDuplicateEvents() { + assertNoDuplicateUnderConcurrency(filter) + } + + @Test + fun concurrentUpdatesWithLimitNeverEmitDuplicateEvents() { + assertNoDuplicateUnderConcurrency(Filter(kinds = listOf(AppDefinitionEvent.KIND), limit = 5)) + } + + private fun assertNoDuplicateUnderConcurrency(withFilter: Filter) { + val firstViolation = AtomicReference?>(null) + val subject = + newFilter(withFilter) { emitted -> + val ids = emitted.map { it.id } + if (ids.size != ids.toSet().size) { + firstViolation.compareAndSet(null, ids) + } + } + subject.init() + + val addresses = (0 until 12).map { "app-$it" } + val notes = addresses.associateWith { noteFor(it) } + val threadCount = 8 + val iterations = 5_000 + val start = CountDownLatch(1) + + val threads = + (0 until threadCount).map { t -> + thread { + start.await() + var seed = t * 31 + 7 + repeat(iterations) { i -> + seed = seed * 1103515245 + 12345 + val note = notes.getValue(addresses[(seed ushr 16) % addresses.size]) + val event = note.load(1_000L + (i % 9)) + if ((seed ushr 8) % 3 == 0) { + subject.remove(note) + } else { + subject.new(event, note) + } + } + } + } + + start.countDown() + threads.forEach { it.join() } + + assertNull(firstViolation.get(), "an emission carried a duplicate event id: ${firstViolation.get()}") + } +}