fix: stop ConcurrentModificationException in AddressableAuthorRelayLoaderSubAssembler

`activeSubscriptions` was a plain LinkedHashSet iterated (`SetsKt.minus`) and mutated (`clear`/`addAll`) with no synchronization, while `invalidateFilters()` ran synchronously on whatever thread called subscribe/unsubscribe. Those callers are genuinely concurrent: `ComposeSubscriptionManager` invokes `invalidateKeys()` after releasing its own lock, and `LifecycleAwareSubscription`'s 30s grace-period unsubscribe fires on a `Dispatchers.Default` worker while composition subscribes from elsewhere. Hence the reported `ConcurrentModificationException` on `DefaultDispatcher-worker-70`. This is the only member of `EventFinderFilterAssembler.group` that implements `IEoseManager` directly; its two siblings extend `BaseEoseManager`, whose `invalidateFilters` hands off to `BundledUpdate` and is therefore never run on the caller's thread nor concurrently with itself. Fix, matching that existing pattern and avoiding locks: - Route through `BundledUpdate`. `BasicBundledUpdate` holds `isProcessing` under a Mutex, so only one body runs at a time — the concurrent iterate-vs-mutate window is gone by construction. It also moves the `allKeys()` scan and per-stub `getOrCreateUser` off the caller thread, which `ComposeSubscriptionManager` documents as "called by main. Keep it really fast." - Hold the state in an `AtomicReference<Set<...>>` of immutable snapshots swapped with `exchange()`, plus an `AtomicBoolean` teardown flag, mirroring the `AtomicReference` + CAS idiom in `FilterIndex`/`BanStore`. - `bundler.cancel()` cannot stop a body already executing (no suspension points), so `destroy()` flags first and an in-flight body compensates by releasing what it just acquired. Double-unsubscribe is a no-op. No locks are introduced; the hot path is strictly cheaper than before. Tested: the new concurrency test reproduces the exact production failure against the pre-fix code (`ConcurrentModificationException` alongside the overlap detector) and passes after. Full :amethyst suite green (941 tests). Note the pre-existing `UserFinderQueryState` identity-equality churn is deliberately NOT addressed here: the set-diff never converges because each run allocates fresh wrappers. It widens this race window but is an independent defect needing its own design decision.
2026-08-10 08:27:04 +00:00 · 2026-07-22 14:40:14 +01:00
parent c9e8cd4e0c
commit 2e4afbf75e
2 changed files with 234 additions and 9 deletions
@@ -21,11 +21,17 @@
 package com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.loaders

 import com.vitorpamplona.amethyst.commons.relayClient.eoseManagers.IEoseManager
+import com.vitorpamplona.amethyst.commons.service.BundledUpdate
 import com.vitorpamplona.amethyst.model.AddressableNote
 import com.vitorpamplona.amethyst.model.LocalCache
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.EventFinderQueryState
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.user.UserFinderFilterAssembler
 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].
@@ -37,14 +43,29 @@ import com.vitorpamplona.amethyst.service.relayClient.reqCommand.user.UserFinder
 * 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 {
-    private val activeSubscriptions = mutableSetOf<UserFinderQueryState>()
+    // 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)
+
+    // Keeps the scan off the caller's thread. invalidateFilters() is reached synchronously from
+    // ComposeSubscriptionManager.subscribe/unsubscribe on every note composable mount/unmount,
+    // and those are documented "called by main. Keep it really fast."
+    private val bundler = BundledUpdate(500, Dispatchers.IO)

    override fun invalidateFilters(ignoreIfDoing: Boolean) {
+        bundler.invalidate(ignoreIfDoing, ::forceInvalidate)
+    }
+
+    private fun forceInvalidate() {
        val needed = mutableSetOf<UserFinderQueryState>()

        allKeys().forEach { key ->
@@ -57,18 +78,26 @@ class AddressableAuthorRelayLoaderSubAssembler(
            }
        }

-        val toAdd = needed - activeSubscriptions
-        val toRemove = activeSubscriptions - needed
+        if (destroyed.load()) return

-        userFinder.subscribe(toAdd.toList())
-        userFinder.unsubscribe(toRemove.toList())
+        val previous = activeSubscriptions.exchange(needed)

-        activeSubscriptions.clear()
-        activeSubscriptions.addAll(needed)
+        userFinder.subscribe((needed - previous).toList())
+        userFinder.unsubscribe((previous - 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())
+        }
    }

    override fun destroy() {
-        userFinder.unsubscribe(activeSubscriptions.toList())
-        activeSubscriptions.clear()
+        // 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())
    }
 }
@@ -0,0 +1,196 @@
+/*
+ * 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.service.relayClient.reqCommand.event.loaders
+
+import com.vitorpamplona.amethyst.model.Account
+import com.vitorpamplona.amethyst.model.LocalCache
+import com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.EventFinderQueryState
+import com.vitorpamplona.amethyst.service.relayClient.reqCommand.user.UserFinderFilterAssembler
+import com.vitorpamplona.amethyst.service.relayClient.reqCommand.user.UserFinderQueryState
+import com.vitorpamplona.quartz.nip01Core.core.Address
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+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
+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.
+     */
+    private fun stubKeys(count: Int): Set<EventFinderQueryState> {
+        val account = mockk<Account>()
+        return (1..count).mapTo(mutableSetOf()) { i ->
+            val address = Address(30023, "ad04%060x".format(i), "d$i")
+            EventFinderQueryState(LocalCache.getOrCreateAddressableNoteInternal(address), account)
+        }
+    }
+
+    /**
+     * Regression test for the `ConcurrentModificationException` in
+     * `SetsKt.minus` reported from `DefaultDispatcher-worker-70`.
+     *
+     * `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.
+     */
+    @Test
+    fun concurrentInvalidateFiltersNeverOverlap() {
+        val errors = CopyOnWriteArrayList<Throwable>()
+        val userFinder = mockk<UserFinderFilterAssembler>(relaxed = true)
+        val keys = stubKeys(50)
+
+        val inFlight = AtomicInteger(0)
+        val assembler =
+            AddressableAuthorRelayLoaderSubAssembler(
+                LocalCache,
+                {
+                    if (inFlight.incrementAndGet() > 1) {
+                        errors.add(IllegalStateException("forceInvalidate bodies overlapped"))
+                    }
+                    try {
+                        keys
+                    } finally {
+                        inFlight.decrementAndGet()
+                    }
+                },
+                userFinder,
+            )
+
+        val start = CountDownLatch(1)
+        try {
+            val threads =
+                (1..8).map {
+                    thread(start = false) {
+                        start.await()
+                        repeat(500) {
+                            try {
+                                assembler.invalidateFilters()
+                            } catch (t: Throwable) {
+                                errors.add(t)
+                            }
+                        }
+                    }
+                }
+            threads.forEach { it.start() }
+            start.countDown()
+            threads.forEach { it.join() }
+        } finally {
+            assembler.destroy()
+        }
+
+        assertTrue(
+            "invalidateFilters raced under concurrency: " +
+                errors.map { "${it::class.simpleName}: ${it.message}" }.distinct(),
+            errors.isEmpty(),
+        )
+    }
+
+    /** The manager still does its job: unresolved stub authors reach the user finder. */
+    @Test
+    fun bridgesMissingAuthorsIntoUserFinder() {
+        val userFinder = mockk<UserFinderFilterAssembler>(relaxed = true)
+        val keys = stubKeys(3)
+        val assembler = AddressableAuthorRelayLoaderSubAssembler(LocalCache, { keys }, userFinder)
+
+        try {
+            assembler.invalidateFilters()
+
+            verify(timeout = 3000) {
+                userFinder.subscribe(
+                    match<List<UserFinderQueryState>> { it.size == 3 },
+                )
+            }
+        } finally {
+            assembler.destroy()
+        }
+    }
+
+    /**
+     * `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.
+     *
+     * The body is gated inside the injected `allKeys()` lambda so `destroy()` provably runs
+     * underneath an in-flight run rather than racing it by luck.
+     */
+    @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)
+        val destroyFinished = CountDownLatch(1)
+        val assembler =
+            AddressableAuthorRelayLoaderSubAssembler(
+                LocalCache,
+                {
+                    invalidateEntered.countDown()
+                    destroyFinished.await(5, TimeUnit.SECONDS)
+                    keys
+                },
+                userFinder,
+            )
+
+        try {
+            assembler.invalidateFilters()
+            assertTrue("bundled run never started", invalidateEntered.await(5, TimeUnit.SECONDS))
+        } finally {
+            assembler.destroy()
+            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(),
+        )
+    }
+}