diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/auth/RelayAuthenticator.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/auth/RelayAuthenticator.kt index cce594855a..846469de0b 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/auth/RelayAuthenticator.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/auth/RelayAuthenticator.kt @@ -36,6 +36,8 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.launch +import kotlin.concurrent.atomics.AtomicReference +import kotlin.concurrent.atomics.ExperimentalAtomicApi interface IAuthStatus { fun hasFinishedAuthentication(relay: NormalizedRelayUrl): Boolean @@ -45,12 +47,36 @@ object EmptyIAuthStatus : IAuthStatus { override fun hasFinishedAuthentication(relay: NormalizedRelayUrl) = true } +@OptIn(ExperimentalAtomicApi::class) class RelayAuthenticator( val client: INostrClient, val scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()), val signWithAllLoggedInUsers: suspend (EventTemplate) -> List, ) : IAuthStatus { - private val authStatus = mutableMapOf() + // Connection callbacks fire on the per-relay OkHttp dispatcher thread, so + // this state is mutated concurrently — copy-on-write under AtomicReference. + private val authStatus: AtomicReference> = + AtomicReference(emptyMap()) + + private fun putAuthStatus( + relay: NormalizedRelayUrl, + status: RelayAuthStatus, + ) { + while (true) { + val current = authStatus.load() + val next = current + (relay to status) + if (authStatus.compareAndSet(current, next)) return + } + } + + private fun removeAuthStatus(relay: NormalizedRelayUrl) { + while (true) { + val current = authStatus.load() + if (relay !in current) return + val next = current - relay + if (authStatus.compareAndSet(current, next)) return + } + } private val clientListener = object : RelayConnectionListener { @@ -66,11 +92,11 @@ class RelayAuthenticator( } override fun onConnecting(relay: IRelayClient) { - authStatus[relay.url] = RelayAuthStatus() + putAuthStatus(relay.url, RelayAuthStatus()) } override fun onDisconnected(relay: IRelayClient) { - authStatus.remove(relay.url) + removeAuthStatus(relay.url) } } @@ -82,7 +108,7 @@ class RelayAuthenticator( val ev = RelayAuthEvent.build(relay.url, msg.challenge) signWithAllLoggedInUsers(ev).forEach { authEvent -> // only send replies to new challenges to avoid infinite loop: - if (authStatus[relay.url]?.saveAuthSubmission(authEvent) == true) { + if (authStatus.load()[relay.url]?.saveAuthSubmission(authEvent) == true) { relay.sendIfConnected(AuthCmd(authEvent)) } } @@ -94,12 +120,12 @@ class RelayAuthenticator( msg: OkMessage, ) { // if this is the OK of an auth event, renew all subscriptions and resend all outgoing events. - if (authStatus[relay.url]?.checkAuthResults(msg.eventId, msg.success) == true) { + if (authStatus.load()[relay.url]?.checkAuthResults(msg.eventId, msg.success) == true) { client.syncFilters(relay) } } - override fun hasFinishedAuthentication(relay: NormalizedRelayUrl) = authStatus[relay]?.hasFinishedAllAuths() != false + override fun hasFinishedAuthentication(relay: NormalizedRelayUrl) = authStatus.load()[relay]?.hasFinishedAllAuths() != false init { Log.d("RelayAuthenticator", "Init, Subscribe") diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/auth/RelayAuthenticatorConcurrencyTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/auth/RelayAuthenticatorConcurrencyTest.kt new file mode 100644 index 0000000000..67fa00c2b7 --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/auth/RelayAuthenticatorConcurrencyTest.kt @@ -0,0 +1,126 @@ +/* + * 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.quartz.nip01Core.relay.client.auth + +import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import kotlin.test.Test + +/** + * Reproduces issue #2946 — `ClassCastException: LinkedHashMap$Entry cannot be + * cast to HashMap$TreeNode` thrown from + * `RelayAuthenticator$clientListener.onDisconnected`. + * + * OkHttp dispatches WebSocket callbacks on one thread per relay, so when many + * relays connect/disconnect simultaneously the listener's internal map is + * mutated concurrently. Once a bucket exceeds the HashMap TREEIFY_THRESHOLD (8) + * the concurrent treeification corrupts internal state. + * + * On the buggy code this test fails non-deterministically with a + * `ClassCastException` (or `ConcurrentModificationException` / + * `NullPointerException`). After the fix it must pass cleanly every run. + */ +class RelayAuthenticatorConcurrencyTest { + private class CapturingClient( + private val delegate: INostrClient = EmptyNostrClient(), + ) : INostrClient by delegate { + @Volatile var captured: RelayConnectionListener? = null + + override fun addConnectionListener(listener: RelayConnectionListener) { + captured = listener + } + } + + private class FakeRelayClient( + override val url: NormalizedRelayUrl, + ) : IRelayClient { + override fun connect() = Unit + + override fun needsToReconnect() = false + + override fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean) = Unit + + override fun isConnected() = false + + override fun sendOrConnectAndSync(cmd: Command) = Unit + + override fun sendIfConnected(cmd: Command) = Unit + + override fun disconnect() = Unit + } + + @Test + fun concurrentConnectingAndDisconnecting_doesNotCorruptInternalState() { + runBlocking { + // The race only fires while the underlying HashMap is structurally + // growing — rehashing and bucket treeification. Once the map reaches + // its steady-state size, put/remove on existing keys touch a single + // node and won't reproduce. So drive many short "burst" cycles, each + // starting from an empty map and growing it past + // MIN_TREEIFY_CAPACITY (64) under concurrent load. + repeat(50) { burst -> + val client = CapturingClient() + val authenticator = + RelayAuthenticator( + client = client, + signWithAllLoggedInUsers = { emptyList() }, + ) + val listener = + client.captured + ?: error("RelayAuthenticator did not register a listener") + + val relays = + (0 until 256).map { + FakeRelayClient(NormalizedRelayUrl("wss://relay-$burst-$it.example/")) + } + + withContext(Dispatchers.IO) { + (0 until 64) + .map { workerId -> + async { + // Each worker walks the relay set, connecting and + // disconnecting. Connects grow the map (rehash / + // treeify); disconnects shrink it; concurrent reads + // run alongside. + relays.forEachIndexed { idx, relay -> + if ((workerId + idx) and 1 == 0) { + listener.onConnecting(relay) + } else { + listener.onDisconnected(relay) + } + authenticator.hasFinishedAuthentication(relay.url) + } + } + }.awaitAll() + } + } + } + } +}