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()}") + } +}