Merge pull request #3777 from vitorpamplona/claude/lazycolumn-duplicate-key-ahgh60

Fix NoteListMatchingFilter to prevent duplicate entries under concurrent updates

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<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

@@ -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 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
+ * New versions of addressables do not update the list.
+ *
+ * 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; [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. 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,
     private val atOnce: (filter: Filter) -> SortedSet<Note>,
     private val update: (List<Note>) -> Unit,
 ) : Observable {
-    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 {
+        // 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,
@@ -47,26 +101,55 @@ 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())
-                }
+        if (!filter.match(event)) return
 
-                update(currentResults.toList())
+        // 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())
     }
 
     override fun remove(note: Note) {
-        if (currentResults.remove(note)) {
-            update(currentResults.toList())
+        // 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(snapshot())
     }
 
     fun init() {
-        currentResults = ConcurrentSkipListSet(atOnce(filter))
-        update(currentResults.toList())
+        sorted.clear()
+        byId.clear()
+        atOnce(filter).forEach { note ->
+            byId.computeIfAbsent(note.idHex) { entryFor(note).also { sorted.add(it) } }
+        }
+        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

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

commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/model/observables/NoteListMatchingFilterTest.kt

@@ -0,0 +1,227 @@
+/*
+ * 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 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(
+        withFilter: Filter = filter,
+        sink: (List<Note>) -> Unit,
+    ) = NoteListMatchingFilter(
+        filter = withFilter,
+        atOnce = { TreeSet(CreatedAtIdHexComparator) },
+        update = sink,
+    )
+
+    @Test
+    fun newerVersionOfAnAddressableDoesNotDuplicateTheKey() {
+        var last: List<Note> = 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 listStaysSortedByCreatedAtDescendingAsNotesArriveOutOfOrder() {
+        var last: List<Note> = 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 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<List<String>?>(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 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<Note> = 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<Note> = 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")
+    }
+}