From f357b445c3906b22f7cb054fa6b6fa34c8bfd976 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 1 Jul 2026 02:30:19 +0000 Subject: [PATCH] fix(quartz): serialize PoolRequests state machine to kill shared-sub double-REQ race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A subscription id is driven from two threads at once: the app thread (the subscribe/unsubscribe path) and every relay's socket-reader thread (an EOSE that triggers an auto-resend). Both read the subscription state and both can decide "the filters changed, send a REQ", but the decision (read state) and the send (mark state SENT in onSent) were not atomic. So the reader could observe the pre-send state — filters still on the previous value — while the app had already moved the desired filters forward, and both would send a REQ for the same sub id. Two REQs on one id race on the wire: the relay answers with duplicate EOSEs and events, or — if a CLOSE interleaves — an empty result that silently truncates a paged download. This is what intermittently broke fetchAllPages on large sets (fixed at the call site in ed5c25e2 by using a fresh sub id per page); this commit fixes the underlying race in the relay-client layer, which could equally corrupt any subscription that spans multiple relays (several reader threads mutate the same RequestSubscriptionState maps concurrently). The fix: - Add a tiny non-reentrant spin lock (withStateLock, same AtomicBoolean primitive BasicRelayClient uses) guarding every access to the subscription state machine. Listener callbacks and socket sends stay OUTSIDE the lock — they re-enter this class via onSent, so holding it across them would deadlock. - Fold the send decision into decideCommandLocked, which runs under the lock and pre-marks the state SENT (+ filters) the moment it decides to send a REQ. A concurrent decider then sees SENT/updated filters and declines, so exactly one REQ is ever produced. Verified with a deterministic A/B repro that pins the exact interleaving open: pre-fix 300/300 episodes produced a duplicate REQ; post-fix 0/300 (max one REQ per episode). Kept as PoolRequestsConcurrencyTest. Full relay test suite passes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z --- .../relay/client/pool/PoolRequests.kt | 226 ++++++++++++------ .../relay/PoolRequestsConcurrencyTest.kt | 146 +++++++++++ 2 files changed, 302 insertions(+), 70 deletions(-) create mode 100644 quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/PoolRequestsConcurrencyTest.kt diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt index 5bf4dff727..8ed3822081 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolRequests.kt @@ -35,6 +35,8 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.flow.MutableStateFlow +import kotlin.concurrent.atomics.AtomicBoolean +import kotlin.concurrent.atomics.ExperimentalAtomicApi /** * Manages relay subscriptions for the entire pool in a way that only @@ -43,6 +45,7 @@ import kotlinx.coroutines.flow.MutableStateFlow * This code also awaits a subscription to come to EOSE since many relays * have through switching subs while they are processing the past. */ +@OptIn(ExperimentalAtomicApi::class) class PoolRequests { /** * Desired subs and listeners @@ -64,6 +67,42 @@ class PoolRequests { fun subState(subId: String): RequestSubscriptionState = relayState.getOrCreate(subId) { RequestSubscriptionState() } + /** + * Serializes every access to the subscription state machine + * ([RequestSubscriptionState]) and the "should I send a REQ?" decision. + * + * A single subscription can span many relays, and each relay's + * socket-reader thread delivers messages into this class concurrently while + * the app thread adds/removes subscriptions — so the plain maps inside + * [RequestSubscriptionState] are written from several threads at once. That + * is both a memory hazard (concurrent map mutation) and, more importantly, + * a logic hazard: the check-then-send in [decideCommandLocked] must be + * atomic, otherwise two threads can both observe "no REQ in flight" and both + * send a REQ for the same sub id. + * + * This is a tiny non-reentrant spin lock (the same [AtomicBoolean] primitive + * used by BasicRelayClient's connecting mutex): the critical sections are a + * handful of map operations, never any I/O. Listener callbacks and the + * actual socket sends are ALWAYS performed outside the lock — they re-enter + * this class through [onSent], so holding the lock across them would + * self-deadlock. + */ + private val stateLock = AtomicBoolean(false) + + private inline fun withStateLock(block: () -> R): R { + while (stateLock.exchange(true)) { + // Another thread holds the lock. Spin-read until it looks free + // (test-and-test-and-set: cheaper on the cache line than hammering + // exchange) then retry the acquisition above. + while (stateLock.load()) { } + } + try { + return block() + } finally { + stateLock.store(false) + } + } + /** * This is called when a sub is added or removed from this class and * should update the desired relay list to get the pool to connect @@ -150,8 +189,10 @@ class PoolRequests { */ fun onConnecting(url: NormalizedRelayUrl) { // Change states to connecting. - relayState.forEach { subId, state -> - state.connecting(url) + withStateLock { + relayState.forEach { subId, state -> + state.connecting(url) + } } } @@ -164,7 +205,9 @@ class PoolRequests { ) { when (cmd) { is ReqCmd -> { - subState(cmd.subId).onOpenReq(relay, cmd.filters) + withStateLock { + subState(cmd.subId).onOpenReq(relay, cmd.filters) + } desiredSubListeners.get(cmd.subId)?.onSubscriptionStarted( relay = relay.url, forFilters = cmd.filters, @@ -172,7 +215,9 @@ class PoolRequests { } is CloseCmd -> { - subState(cmd.subId).onSubscriptionClosed(relay) + withStateLock { + subState(cmd.subId).onSubscriptionClosed(relay) + } desiredSubListeners.get(cmd.subId)?.onSubscriptionClosed( relay = relay.url, ) @@ -189,46 +234,63 @@ class PoolRequests { ) { when (msg) { is EventMessage -> { - val state = relayState.get(msg.subId) - state?.onNewEvent(relay.url) + var isLive = false + var forFilters: List? = null + withStateLock { + val state = relayState.get(msg.subId) + state?.onNewEvent(relay.url) + isLive = state?.currentState(relay.url) == ReqSubStatus.LIVE + forFilters = state?.lastKnownFilterStates(relay.url) + } desiredSubListeners.get(msg.subId)?.onEvent( event = msg.event, - isLive = state?.currentState(relay.url) == ReqSubStatus.LIVE, + isLive = isLive, relay = relay.url, - forFilters = state?.lastKnownFilterStates(relay.url), + forFilters = forFilters, ) } is EoseMessage -> { - val state = relayState.get(msg.subId) - state?.onEose(relay.url) + var forFilters: List? = null + val cmd = + withStateLock { + val state = relayState.get(msg.subId) + state?.onEose(relay.url) + forFilters = state?.lastKnownFilterStates(relay.url) + // Decide (and pre-mark) the resend while still holding the + // lock, so a concurrent subscribe/unsubscribe on the app + // thread can't also decide to send a REQ for this sub. + decideCommandLocked(msg.subId, relay.url) + } desiredSubListeners.get(msg.subId)?.onEose( relay = relay.url, - forFilters = state?.lastKnownFilterStates(relay.url), + forFilters = forFilters, ) // send a newer version when done - sendToRelayIfChanged(msg.subId, relay.url) { cmd -> + if (cmd != null) { relay.sendOrConnectAndSync(cmd) } } is ClosedMessage -> { - val state = relayState.get(msg.subId) - state?.onClosed(relay.url) - + var forFilters: List? = null + val cmd = + withStateLock { + val state = relayState.get(msg.subId) + state?.onClosed(relay.url) + forFilters = state?.lastKnownFilterStates(relay.url) + decideCommandLocked(msg.subId, relay.url) + } desiredSubListeners.get(msg.subId)?.onClosed( message = msg.message, relay = relay.url, - forFilters = state?.lastKnownFilterStates(relay.url), + forFilters = forFilters, ) - // send a newer version when done - sendToRelayIfChanged(msg.subId, relay.url) { cmd -> - // don't send a close if just closed - if (cmd !is CloseCmd) { - relay.sendOrConnectAndSync(cmd) - } + // send a newer version when done, but don't send a close if just closed + if (cmd != null && cmd !is CloseCmd) { + relay.sendOrConnectAndSync(cmd) } } } @@ -238,8 +300,10 @@ class PoolRequests { * When the relay disconnects */ fun onDisconnected(url: NormalizedRelayUrl) { - relayState.forEach { subId, state -> - state.disconnected(url) + withStateLock { + relayState.forEach { subId, state -> + state.disconnected(url) + } } } @@ -262,16 +326,27 @@ class PoolRequests { url: NormalizedRelayUrl, errorMessage: String, ) { - relayState.forEach { subId, state -> - // These are all my subs.. need to figure out which relays have them - val subs = desiredSubs.get(subId) - if (subs != null && url in subs.keys) { - desiredSubListeners.get(subId)?.onCannotConnect( - relay = url, - message = errorMessage, - forFilters = state.lastKnownFilterStates(url), - ) + // Snapshot the affected subs (and their last-known filters) under the + // lock, then notify listeners outside it. + val toNotify = + withStateLock { + val list = mutableListOf?>>() + relayState.forEach { subId, state -> + // These are all my subs.. need to figure out which relays have them + val subs = desiredSubs.get(subId) + if (subs != null && url in subs.keys) { + list.add(subId to state.lastKnownFilterStates(url)) + } + } + list } + + toNotify.forEach { (subId, forFilters) -> + desiredSubListeners.get(subId)?.onCannotConnect( + relay = url, + message = errorMessage, + forFilters = forFilters, + ) } } @@ -281,54 +356,65 @@ class PoolRequests { sync: (NormalizedRelayUrl, Command) -> Unit, ) { relaysToUpdate.forEach { relay -> - sendToRelayIfChanged(subId, relay) { cmd -> - if (cmd is ReqCmd) { - val currentState = relayState.get(subId)?.currentState(relay) - - if (currentState == ReqSubStatus.SENT || currentState == ReqSubStatus.QUERYING_PAST) { - // sending multiple REQs triggers multiple EOSEs back and we then don't know which - // one is which. - } else { - sync(relay, cmd) - } - } else { - sync(relay, cmd) - } + // Decide + pre-mark atomically under the lock, then send outside it. + val cmd = withStateLock { decideCommandLocked(subId, relay) } + if (cmd != null) { + sync(relay, cmd) } } } - fun sendToRelayIfChanged( + /** + * Decides which command (if any) must be sent to [relay] to bring it in line + * with the desired filters for [subId], and — for a REQ — pre-marks the + * subscription state as SENT before returning. + * + * Pre-marking is what makes the check-then-send atomic: a second thread that + * runs this method for the same sub sees the SENT state (or the + * already-updated filters) and declines to send a duplicate REQ. Two REQs on + * one sub id race on the wire and produce duplicate EOSEs/events (or, if a + * CLOSE interleaves, an empty result that silently truncates a paged + * download) — that is the bug this guards against. + * + * MUST be called while holding [withStateLock]. + */ + private fun decideCommandLocked( subId: String, relay: NormalizedRelayUrl, - sync: (Command) -> Unit, - ) { + ): Command? { val state = relayState.get(subId) val oldFilters = state?.currentFilters(relay) val newFilters = desiredSubs.get(subId)?.get(relay) - sendToRelayIfChanged(subId, oldFilters, newFilters, sync) - } - fun sendToRelayIfChanged( - subId: String, - oldFilters: List?, - newFilters: List?, - sync: (Command) -> Unit, - ) { - if (newFilters.isNullOrEmpty()) { - // some relays are not in this sub anymore. Stop their subscriptions - if (!oldFilters.isNullOrEmpty()) { - // only update if the old filters are not already closed. - sync(CloseCmd(subId)) + return when { + newFilters.isNullOrEmpty() -> { + // some relays are not in this sub anymore. Stop their subscriptions + // only if the old filters are not already closed. + if (!oldFilters.isNullOrEmpty()) CloseCmd(subId) else null + } + + oldFilters.isNullOrEmpty() || FiltersChanged.needsToResendRequest(oldFilters, newFilters) -> { + // A REQ is warranted: a brand new sub, or the filters changed + // enough (not just a `since` bump) to need a resend. But if a REQ + // is already in flight, don't send another — multiple REQs on one + // sub id trigger multiple EOSEs and we can no longer tell which + // reply belongs to which REQ. The pending change is picked up + // later by the EOSE handler, which runs this method again once the + // sub reaches LIVE. + val current = state?.currentState(relay) + if (current == ReqSubStatus.SENT || current == ReqSubStatus.QUERYING_PAST) { + null + } else { + // Pre-mark SENT + filters so a concurrent decider skips. + subState(subId).onOpenReq(relay, newFilters) + ReqCmd(subId, newFilters) + } + } + + else -> { + // Filters are effectively the same; nothing to do. + null } - } else if (oldFilters.isNullOrEmpty()) { - // new relays were added. Start a new sub in them - sync(ReqCmd(subId, newFilters)) - } else if (FiltersChanged.needsToResendRequest(oldFilters, newFilters)) { - // filters were changed enough (not only an update in since) to warn a new update - sync(ReqCmd(subId, newFilters)) - } else { - // They are the same don't do anything. } } diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/PoolRequestsConcurrencyTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/PoolRequestsConcurrencyTest.kt new file mode 100644 index 0000000000..6d21d526e3 --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/PoolRequestsConcurrencyTest.kt @@ -0,0 +1,146 @@ +/* + * 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 + +import com.vitorpamplona.quartz.nip01Core.relay.client.pool.PoolRequests +import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.EoseMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import java.util.concurrent.CountDownLatch +import java.util.concurrent.atomic.AtomicInteger +import kotlin.concurrent.thread +import kotlin.test.Test +import kotlin.test.assertEquals + +/** + * Regression guard for the shared-sub-id double-REQ race in [PoolRequests]. + * + * A single subscription id is driven from two threads at once: the app thread + * (the subscribe path, [PoolRequests.sendToRelayIfChanged]) and the relay reader + * thread (an EOSE that triggers an auto-resend, [PoolRequests.onIncomingMessage]). + * The subscription is already LIVE and its desired filters have just changed, so + * both threads independently conclude "the filters changed, send a REQ". + * + * The bug: the decision (read state) and the send (mark state SENT via onSent) + * were not atomic, so the reader could read the pre-send state (filters still on + * the previous value) while the app had already moved the desired filters + * forward — and both would send a REQ for the same sub id. Two REQs on one id + * race on the wire: the relay answers with two EOSEs and duplicate events, or — + * if a CLOSE interleaves — an empty result that silently truncates a paged + * download (this is what broke `fetchAllPages` on large sets). + * + * The fix makes the "should I send a REQ?" decision pre-mark the state + * atomically, so exactly one REQ is ever produced. This test pins the exact + * interleaving the bug needs (app has produced its REQ but not yet run onSent) + * open and asserts only one REQ comes out. + */ +class PoolRequestsConcurrencyTest { + private class FakeRelay( + override val url: NormalizedRelayUrl, + val onCmd: (Command) -> Unit, + ) : IRelayClient { + override fun connect() {} + + override fun needsToReconnect() = false + + override fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean) {} + + override fun isConnected() = true + + override fun sendOrConnectAndSync(cmd: Command) = onCmd(cmd) + + override fun sendIfConnected(cmd: Command) = onCmd(cmd) + + override fun disconnect() {} + } + + @Test + fun concurrentEoseResendAndSubscribeSendExactlyOneReq() { + val url = RelayUrlNormalizer.normalize("ws://race/") + val subId = "shared-sub" + val filtersA = listOf(Filter(kinds = listOf(1))) + val filtersB = listOf(Filter(kinds = listOf(2))) + val listener = object : SubscriptionListener {} + + // Many episodes so a regression that only sometimes doubles still trips. + repeat(300) { episode -> + val pool = PoolRequests() + val reqBCount = AtomicInteger(0) + + fun countReqB(cmd: Command) { + if (cmd is ReqCmd && cmd.filters == filtersB) reqBCount.incrementAndGet() + } + + val fakeRelay = + FakeRelay(url) { cmd -> + // relay-reader auto-resend send path + countReqB(cmd) + pool.onSent(url, cmd) + } + + // Bring the sub to LIVE with filters A. + val setupRelays = pool.addOrUpdate(subId, mapOf(url to filtersA), listener) + pool.sendToRelayIfChanged(subId, setupRelays) { _, cmd -> pool.onSent(url, cmd) } + pool.onIncomingMessage(fakeRelay, EoseMessage(subId)) + + // The desired filters change to B (e.g. the next page of a paged download). + pool.addOrUpdate(subId, mapOf(url to filtersB), listener) + + val appProducedReq = CountDownLatch(1) + val readerDone = CountDownLatch(1) + + val appThread = + thread { + pool.sendToRelayIfChanged(subId, setOf(url)) { _, cmd -> + countReqB(cmd) + // App has produced its REQ(B); park before onSent so the + // subscription state is not yet advanced — the exact window + // the race needs. + appProducedReq.countDown() + readerDone.await() + pool.onSent(url, cmd) + } + } + + val readerThread = + thread { + appProducedReq.await() + pool.onIncomingMessage(fakeRelay, EoseMessage(subId)) + readerDone.countDown() + } + + appThread.join() + readerThread.join() + + assertEquals( + 1, + reqBCount.get(), + "episode $episode: exactly one REQ must be sent for the changed filters, " + + "never a duplicate from the app + reader race", + ) + } + } +}