fix(quartz): serialize PoolRequests state machine to kill shared-sub double-REQ race

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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JmSyzdmKyiz3pPxUZ8Mg8Z
2026-08-09 08:04:45 +00:00 · 2026-07-01 02:30:19 +00:00
parent ed5c25e2b1
commit f357b445c3
2 changed files with 302 additions and 70 deletions
@@ -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<NormalizedRelayUrl> = 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 <R> 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<Filter>? = 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<Filter>? = 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<Filter>? = 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<Pair<String, List<Filter>?>>()
+                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<Filter>?,
-        newFilters: List<Filter>?,
-        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.
        }
    }

@@ -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",
+            )
+        }
+    }
+}