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 5f81cb8d22..fc72c295f9 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 @@ -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, val userFinder: UserFinderFilterAssembler, ) : IEoseManager { - private val activeSubscriptions = mutableSetOf() + // 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) + + // 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() 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()) } } 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 new file mode 100644 index 0000000000..db81e6c618 --- /dev/null +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssemblerTest.kt @@ -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 { + val account = mockk() + 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() + val userFinder = mockk(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(relaxed = true) + val keys = stubKeys(3) + val assembler = AddressableAuthorRelayLoaderSubAssembler(LocalCache, { keys }, userFinder) + + try { + assembler.invalidateFilters() + + verify(timeout = 3000) { + userFinder.subscribe( + match> { 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(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) + 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(), + ) + } +}