Code review:

- private monitor, and correct the commit() KDoc
- replace atomics handshake with a single monitor
- flag userFinder re-entrancy assumption in commit() KDoc

amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssembler.kt

@@ -29,9 +29,6 @@ import com.vitorpamplona.amethyst.service.relayClient.reqCommand.user.UserFinder
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.user.UserFinderQueryState
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.IO
-import kotlin.concurrent.atomics.AtomicBoolean
-import kotlin.concurrent.atomics.AtomicReference
-import kotlin.concurrent.atomics.ExperimentalAtomicApi
 
 /**
  * Bridges missing-addressable-note authors into [UserFinderFilterAssembler].
@@ -43,18 +40,18 @@ import kotlin.concurrent.atomics.ExperimentalAtomicApi
  * kind-0 / kind-10002 and resolve outbox relays via [UserOutboxFinderSubAssembler]. Once the
  * relay list arrives, [EventFinderFilterAssembler] is invalidated and can query the correct relay.
  */
-@OptIn(ExperimentalAtomicApi::class)
 class AddressableAuthorRelayLoaderSubAssembler(
     val cache: LocalCache,
     val allKeys: () -> Set<EventFinderQueryState>,
     val userFinder: UserFinderFilterAssembler,
 ) : IEoseManager {
-    // Immutable snapshots swapped atomically, so a diff can never observe a half-written set.
-    // Mutual exclusion between runs comes from the bundler (one body at a time), not from these
-    // atomics — they exist to hand state over to destroy(), the one caller the bundler cannot
-    // serialize.
-    private val activeSubscriptions = AtomicReference<Set<UserFinderQueryState>>(emptySet())
-    private val destroyed = AtomicBoolean(false)
+    // Private monitor: @Synchronized locks on `this`, which leaves the instance's monitor
+    // reachable to anything holding a reference to this assembler.
+    private val lock = Any()
+
+    // Only ever touched while holding [lock]. See commit() and destroy().
+    private var activeSubscriptions: Set<UserFinderQueryState> = emptySet()
+    private var destroyed = false
 
     // Keeps the scan off the caller's thread. invalidateFilters() is reached synchronously from
     // ComposeSubscriptionManager.subscribe/unsubscribe on every note composable mount/unmount,
@@ -78,26 +75,39 @@ class AddressableAuthorRelayLoaderSubAssembler(
             }
         }
 
-        if (destroyed.load()) return
+        commit(needed)
+    }
 
-        val previous = activeSubscriptions.exchange(needed)
+    /**
+     * Serializes against [destroy] — the one caller the bundler cannot order, because
+     * `bundler.cancel()` cannot stop a body that is already running (it has no suspension points).
+     *
+     * The scan in [forceInvalidate] stays outside [lock], so [destroy] never waits on a
+     * [LocalCache] sweep. It can still wait on the two calls below, which are bounded: a pair of
+     * map updates inside [UserFinderFilterAssembler] plus the coroutine launches its
+     * `invalidateKeys()` fans out to.
+     *
+     * Calling [userFinder] while holding [lock] relies on subscribe/unsubscribe only taking
+     * ComposeSubscriptionManager's own lock and deferring real work to bundled coroutines — they
+     * never call back into this class. Revisit if that changes.
+     */
+    private fun commit(needed: Set<UserFinderQueryState>) {
+        synchronized(lock) {
+            if (destroyed) return
 
-        userFinder.subscribe((needed - previous).toList())
-        userFinder.unsubscribe((previous - needed).toList())
+            userFinder.subscribe((needed - activeSubscriptions).toList())
+            userFinder.unsubscribe((activeSubscriptions - needed).toList())
 
-        // destroy() landed while we were subscribing. bundler.cancel() cannot stop a body that is
-        // already running — it has no suspension points — so the body releases what it just
-        // acquired. destroy() may unsubscribe the same states concurrently; that is a no-op.
-        if (destroyed.load()) {
-            activeSubscriptions.store(emptySet())
-            userFinder.unsubscribe(needed.toList())
+            activeSubscriptions = needed
         }
     }
 
     override fun destroy() {
-        // Flag before cancelling so an in-flight body is guaranteed to see the teardown.
-        destroyed.store(true)
-        bundler.cancel()
-        userFinder.unsubscribe(activeSubscriptions.exchange(emptySet()).toList())
+        synchronized(lock) {
+            destroyed = true
+            bundler.cancel()
+            userFinder.unsubscribe(activeSubscriptions.toList())
+            activeSubscriptions = emptySet()
+        }
     }
 }

amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssemblerTest.kt

@@ -29,9 +29,10 @@ import com.vitorpamplona.quartz.nip01Core.core.Address
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
 import org.junit.Assert.assertTrue
 import org.junit.Test
-import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
@@ -40,7 +41,7 @@ import kotlin.concurrent.thread
 class AddressableAuthorRelayLoaderSubAssemblerTest {
     /**
      * Unique per-test-class identities so the shared [LocalCache] singleton isn't polluted with
-     * notes another test class also claims. The prefix keeps the key a valid 64-char hex pubkey.
+     * notes another test class also claims.
      */
     private fun stubKeys(count: Int): Set<EventFinderQueryState> {
         val account = mockk<Account>()
@@ -51,34 +52,28 @@ class AddressableAuthorRelayLoaderSubAssemblerTest {
     }
 
     /**
-     * Regression test for the `ConcurrentModificationException` in
-     * `SetsKt.minus` reported from `DefaultDispatcher-worker-70`.
+     * Regression test for the `ConcurrentModificationException` in `SetsKt.minus` reported from
+     * `DefaultDispatcher-worker-70`: this manager is genuinely re-entered from several threads at
+     * once, because `ComposeSubscriptionManager.subscribe`/`unsubscribe` call `invalidateKeys()`
+     * *after* releasing their own lock, and `LifecycleAwareSubscription`'s 30s grace-period
+     * unsubscribe fires on a `Dispatchers.Default` worker.
      *
-     * `ComposeSubscriptionManager.subscribe`/`unsubscribe` call `invalidateKeys()` *after*
-     * releasing their own lock, and `LifecycleAwareSubscription`'s 30s grace-period unsubscribe
-     * fires on a `Dispatchers.Default` worker — so this manager is genuinely re-entered from
-     * several threads at once.
-     *
-     * The invariant asserted here is the one that makes the crash impossible: **the body that
-     * reads and swaps the subscription state never runs concurrently with itself.** It's checked
-     * via the injected `allKeys()` lambda (called exactly once per body) rather than by catching
-     * the exception, because once the body is bundled its throwables are swallowed by
-     * `BundledUpdate`'s `CoroutineExceptionHandler` and would never reach the test thread.
+     * Overlap is detected through the injected `allKeys()` lambda rather than by catching — once
+     * the body is bundled its throwables are swallowed by `BundledUpdate`'s
+     * `CoroutineExceptionHandler` and would never reach the test thread.
      */
     @Test
     fun concurrentInvalidateFiltersNeverOverlap() {
-        val errors = CopyOnWriteArrayList<Throwable>()
         val userFinder = mockk<UserFinderFilterAssembler>(relaxed = true)
         val keys = stubKeys(50)
 
         val inFlight = AtomicInteger(0)
+        val overlaps = AtomicInteger(0)
         val assembler =
             AddressableAuthorRelayLoaderSubAssembler(
                 LocalCache,
                 {
-                    if (inFlight.incrementAndGet() > 1) {
-                        errors.add(IllegalStateException("forceInvalidate bodies overlapped"))
-                    }
+                    if (inFlight.incrementAndGet() > 1) overlaps.incrementAndGet()
                     try {
                         keys
                     } finally {
@@ -94,13 +89,7 @@ class AddressableAuthorRelayLoaderSubAssemblerTest {
                 (1..8).map {
                     thread(start = false) {
                         start.await()
-                        repeat(500) {
-                            try {
-                                assembler.invalidateFilters()
-                            } catch (t: Throwable) {
-                                errors.add(t)
-                            }
-                        }
+                        repeat(500) { assembler.invalidateFilters() }
                     }
                 }
             threads.forEach { it.start() }
@@ -110,11 +99,7 @@ class AddressableAuthorRelayLoaderSubAssemblerTest {
             assembler.destroy()
         }
 
-        assertTrue(
-            "invalidateFilters raced under concurrency: " +
-                errors.map { "${it::class.simpleName}: ${it.message}" }.distinct(),
-            errors.isEmpty(),
-        )
+        assertEquals("forceInvalidate bodies overlapped", 0, overlaps.get())
     }
 
     /** The manager still does its job: unresolved stub authors reach the user finder. */
@@ -138,11 +123,9 @@ class AddressableAuthorRelayLoaderSubAssemblerTest {
     }
 
     /**
-     * `bundler.cancel()` cannot stop a body that is already executing — the body has no
-     * suspension points, so it runs to completion after `destroy()` returns. Without the
-     * `destroyed` handshake the body re-subscribes authors that `destroy()` just released,
-     * leaving live kind-0/10002 REQs (and retained `User`/`Account` references) for a dead
-     * account after logout.
+     * `destroy()` must win against a body that is already past its `allKeys()` scan: otherwise the
+     * body re-subscribes authors `destroy()` just released, leaving live kind-0/10002 REQs (and
+     * retained `User`/`Account` references) for a dead account after logout.
      *
      * The body is gated inside the injected `allKeys()` lambda so `destroy()` provably runs
      * underneath an in-flight run rather than racing it by luck.
@@ -150,16 +133,10 @@ class AddressableAuthorRelayLoaderSubAssemblerTest {
     @Test
     fun destroyDuringInFlightInvalidateDoesNotLeakSubscriptions() {
         val userFinder = mockk<UserFinderFilterAssembler>(relaxed = true)
-        val subscribed = CopyOnWriteArrayList<UserFinderQueryState>()
-        val unsubscribed = CopyOnWriteArrayList<UserFinderQueryState>()
         val subscribeHappened = CountDownLatch(1)
         every { userFinder.subscribe(any<List<UserFinderQueryState>>()) } answers {
-            subscribed.addAll(firstArg<List<UserFinderQueryState>>())
             subscribeHappened.countDown()
         }
-        every { userFinder.unsubscribe(any<List<UserFinderQueryState>>()) } answers {
-            unsubscribed.addAll(firstArg<List<UserFinderQueryState>>())
-        }
 
         val keys = stubKeys(3)
         val invalidateEntered = CountDownLatch(1)
@@ -183,14 +160,11 @@ class AddressableAuthorRelayLoaderSubAssemblerTest {
             destroyFinished.countDown()
         }
 
-        // Let the in-flight run finish (it either subscribes — the leak — or
-        // observes the teardown and skips; both settle within the timeout).
-        subscribeHappened.await(2, TimeUnit.SECONDS)
-
-        val leaked = subscribed - unsubscribed.toSet()
-        assertTrue(
-            "destroy() left ${leaked.size} subscriptions alive in userFinder",
-            leaked.isEmpty(),
+        // The gated body resumes the instant destroyFinished counts down, so a leak shows up
+        // immediately; the wait only has to outlast that hand-off.
+        assertFalse(
+            "in-flight body subscribed after destroy()",
+            subscribeHappened.await(500, TimeUnit.MILLISECONDS),
         )
     }
 }

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/eoseManagers/IEoseManager.kt

@@ -21,6 +21,12 @@
 package com.vitorpamplona.amethyst.commons.relayClient.eoseManagers
 
 interface IEoseManager {
+    /**
+     * May be called from any thread, concurrently with itself and with [destroy], and is reached
+     * synchronously from main on every composable mount/unmount. Implementations must return fast
+     * and must serialize their own state — see [BaseEoseManager], which does both by routing the
+     * work through a [com.vitorpamplona.amethyst.commons.service.BundledUpdate].
+     */
     fun invalidateFilters(ignoreIfDoing: Boolean = false)
 
     fun destroy()