laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-09 08:04:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: dedup EventListMatchingFilter (observeEvents) and harden emission contract

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ah1aCniyjnzc27x4pwq2Df
This commit is contained in:
Claude
2026-07-28 20:42:13 +00:00
parent 5cb84e16dc
commit 58004fa744
3 changed files with 332 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilter.kt
+108 -20
View File
@@ -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<T : Event>(
private val filter: Filter,
private val atOnce: (filter: Filter) -> SortedSet<Note>,
private val update: (List<T>) -> Unit,
) : Observable {
// Keeping this here blocks it from being cleared from memory
var currentResults: ConcurrentSkipListSet<Note> = 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<Entry> { 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<HexKey, Entry>()
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<T : Event>(
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<T> {
// 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<HexKey>()
return sorted.mapNotNull { e -> if (seen.add(e.note.idHex)) e.note.event as? T else null }
}
}
commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilter.kt
+21 -7
View File
@@ -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<Note> {
// 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<HexKey>()
return sorted.mapNotNull { e -> e.note.takeIf { seen.add(it.idHex) } }
}
}
commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/EventListMatchingFilterTest.kt
+203
View File
@@ -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<Event>) -> Unit,
) = EventListMatchingFilter<Event>(
filter = withFilter,
atOnce = { TreeSet(CreatedAtIdHexComparator) },
update = sink,
)
@Test
fun newerVersionReflectsUpdatedEventWithoutDuplicate() {
var last: List<Event> = 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<Event> = 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<List<Event>>()
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<Event> = 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<List<String>?>(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()}")
}
}