From 2e4afbf75ee285fba86b7d423b734af1365c96a8 Mon Sep 17 00:00:00 2001 From: davotoula Date: Wed, 22 Jul 2026 13:41:55 +0100 Subject: [PATCH 1/3] fix: stop ConcurrentModificationException in AddressableAuthorRelayLoaderSubAssembler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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>` 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. --- ...ddressableAuthorRelayLoaderSubAssembler.kt | 47 ++++- ...ssableAuthorRelayLoaderSubAssemblerTest.kt | 196 ++++++++++++++++++ 2 files changed, 234 insertions(+), 9 deletions(-) create mode 100644 amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/event/loaders/AddressableAuthorRelayLoaderSubAssemblerTest.kt 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(), + ) + } +} From 53a823e1b2a7c603e1ee10cd9f83274be5be4fcd Mon Sep 17 00:00:00 2001 From: davotoula Date: Wed, 22 Jul 2026 14:00:57 +0100 Subject: [PATCH 2/3] Code review: - private monitor, and correct the commit() KDoc - replace atomics handshake with a single monitor - flag userFinder re-entrancy assumption in commit() KDoc --- ...ddressableAuthorRelayLoaderSubAssembler.kt | 58 ++++++++------- ...ssableAuthorRelayLoaderSubAssemblerTest.kt | 72 ++++++------------- .../relayClient/eoseManagers/IEoseManager.kt | 6 ++ 3 files changed, 63 insertions(+), 73 deletions(-) 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() From 71620380fb7ec7985ab236ec2bd160ce9f5f7d16 Mon Sep 17 00:00:00 2001 From: davotoula Date: Wed, 22 Jul 2026 14:24:02 +0100 Subject: [PATCH 3/3] update KotlinJpsPluginSettings version --- .idea/kotlinc.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.idea/kotlinc.xml b/.idea/kotlinc.xml index 5877ff2fad..ceeca8c125 100644 --- a/.idea/kotlinc.xml +++ b/.idea/kotlinc.xml @@ -8,6 +8,6 @@ \ No newline at end of file