diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssembler.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssembler.kt index fc72c295f9..9dba6ed2bd 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssembler.kt +++ b/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, 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>(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 = 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) { + 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() + } } } diff --git a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssemblerTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssemblerTest.kt index db81e6c618..842471e6a4 100644 --- a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssemblerTest.kt +++ b/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 { val account = mockk() @@ -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() val userFinder = mockk(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(relaxed = true) - val subscribed = CopyOnWriteArrayList() - val unsubscribed = CopyOnWriteArrayList() val subscribeHappened = CountDownLatch(1) every { userFinder.subscribe(any>()) } answers { - subscribed.addAll(firstArg>()) subscribeHappened.countDown() } - every { userFinder.unsubscribe(any>()) } answers { - unsubscribed.addAll(firstArg>()) - } 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), ) } } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/eoseManagers/IEoseManager.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/eoseManagers/IEoseManager.kt index 9d4e940c7d..6373a64cf8 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/eoseManagers/IEoseManager.kt +++ b/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()