fix(coordinator): retryable batched-REQ dedup and race-free EOSE aggregator

Reviewer davotoula (PR #3483) flagged two commons/relayClient issues on FeedMetadataCoordinator that both bite the Android app once WoT is wired there: 5. loadKind3Batched / loadMetadataBatched marked pubkeys as sent BEFORE any relay EOSE'd. On flaky-network cold-starts where every index relay timed out, the pubkeys stayed permanently marked and WoT was silently empty for the whole session — the next call short-circuited. Fix: pubkeys enter `queuedKind3Pubkeys` / `queuedPubkeys` only after ≥1 EOSE; on zero-EOSE timeout they roll out of the new `inFlightBatched*` sets so a subsequent call retries. 6. `val eoseReceived = mutableSetOf<NormalizedRelayUrl>()` was mutated from per-relay `onEose` callbacks the client dispatches on `Dispatchers.IO`. Concurrent `add()`/`size` on an unsynchronised HashSet could drop entries or throw CME, forcing the batch to wait the full timeout instead of firing early. Fix: `BatchEoseGate` funnels EOSE notifications through a `Channel` so a single consumer coroutine is the sole reader/writer of the `seen` set — KMP-safe, no `synchronized {}` or JVM-only atomics. Tests exercise: - zero-EOSE timeout → retry re-fires - ≥1 EOSE → next call short-circuits - full-EOSE from 20 relays hammered from Dispatchers.IO in parallel - clear() releases in-flight dedup - same semantics on loadMetadataBatched Plan: commons/plans/2026-07-06-fix-wot-outbox-model-and-review-fixes-plan.md
2026-07-30 19:46:17 +00:00 · 2026-07-07 13:18:38 +03:00
parent 5166216e2e
commit 5217035f94
2 changed files with 395 additions and 18 deletions
@@ -39,6 +39,7 @@ import com.vitorpamplona.quartz.nip18Reposts.RepostEvent
 import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeoutOrNull

@@ -76,6 +77,14 @@ class FeedMetadataCoordinator(
    private val queuedBoostedIds = mutableSetOf<HexKey>()
    private val queuedKind3Pubkeys = mutableSetOf<HexKey>()

+    // Batched paths only — pubkeys currently in-flight in a batched REQ.
+    // Prevents rapid re-fire of the same batch. Distinct from queuedPubkeys
+    // and queuedKind3Pubkeys (which record "asked and at least one relay
+    // returned EOSE") so a batch that times out with zero events can be
+    // retried on the next call — see PR #3483 review finding 5.
+    private val inFlightBatchedMetadata = mutableSetOf<HexKey>()
+    private val inFlightBatchedKind3 = mutableSetOf<HexKey>()
+
    /**
     * Start processing the subscription queue.
     * Call once when coordinator is created.
@@ -253,14 +262,24 @@ class FeedMetadataCoordinator(
    /**
     * Fast-path: batched metadata subscription for visible-viewport authors.
     * Bypasses rate limiter. Single filter with all authors. Closes after EOSE.
+     *
+     * Pubkeys are moved into [queuedPubkeys] (dedup) only after at least one
+     * relay EOSE'd. On timeout with zero EOSE (index relays all unreachable)
+     * they roll out of [inFlightBatchedMetadata] so a subsequent call can
+     * retry — see PR #3483 review finding 5.
     */
    fun loadMetadataBatched(
        pubkeys: List<HexKey>,
        timeoutMs: Long = 5_000L,
    ) {
-        val newPubkeys = pubkeys.filter { it !in queuedPubkeys }.distinct()
+        val newPubkeys =
+            pubkeys
+                .asSequence()
+                .filter { it !in queuedPubkeys && it !in inFlightBatchedMetadata }
+                .distinct()
+                .toList()
        if (newPubkeys.isEmpty()) return
-        queuedPubkeys.addAll(newPubkeys)
+        inFlightBatchedMetadata.addAll(newPubkeys)

        scope.launch {
            val filter =
@@ -271,8 +290,7 @@ class FeedMetadataCoordinator(
                )
            val filterMap = indexRelays.associateWith { listOf(filter) }
            val subId = newSubId()
-            val eoseReceived = mutableSetOf<NormalizedRelayUrl>()
-            val allEose = CompletableDeferred<Unit>()
+            val gate = BatchEoseGate(scope, target = indexRelays.size)

            val listener =
                object : SubscriptionListener {
@@ -289,16 +307,18 @@ class FeedMetadataCoordinator(
                        relay: NormalizedRelayUrl,
                        forFilters: List<Filter>?,
                    ) {
-                        eoseReceived.add(relay)
-                        if (eoseReceived.size >= indexRelays.size) {
-                            allEose.complete(Unit)
-                        }
+                        gate.notifyEose(relay)
                    }
                }

            client.subscribe(subId, filterMap, listener)
-            withTimeoutOrNull(timeoutMs) { allEose.await() }
+            val eosedRelays = gate.awaitAll(timeoutMs)
            client.unsubscribe(subId)
+
+            if (eosedRelays > 0) {
+                queuedPubkeys.addAll(newPubkeys)
+            }
+            inFlightBatchedMetadata.removeAll(newPubkeys.toSet())
        }
    }

@@ -311,18 +331,29 @@ class FeedMetadataCoordinator(
     * so relays with per-filter author caps (nostr-rs-relay defaults to
     * ~100) don't silently truncate the batch. Aggregates EOSE across
     * chunks and calls [onEose] once (or after [timeoutMs]).
+     *
+     * Pubkeys are moved into [queuedKind3Pubkeys] (dedup) only after at
+     * least one relay EOSE'd. On timeout with zero EOSE (index relays all
+     * unreachable — common on flaky mobile networks) they roll out of
+     * [inFlightBatchedKind3] so the next `loadKind3Batched` call retries
+     * — see PR #3483 review finding 5.
     */
    fun loadKind3Batched(
        pubkeys: Collection<HexKey>,
        timeoutMs: Long = 5_000L,
        onEose: () -> Unit = {},
    ) {
-        val newPubkeys = pubkeys.filter { it !in queuedKind3Pubkeys }.distinct()
+        val newPubkeys =
+            pubkeys
+                .asSequence()
+                .filter { it !in queuedKind3Pubkeys && it !in inFlightBatchedKind3 }
+                .distinct()
+                .toList()
        if (newPubkeys.isEmpty()) {
            onEose()
            return
        }
-        queuedKind3Pubkeys.addAll(newPubkeys)
+        inFlightBatchedKind3.addAll(newPubkeys)

        scope.launch {
            val filters =
@@ -335,8 +366,7 @@ class FeedMetadataCoordinator(
                }
            val filterMap = indexRelays.associateWith { filters }
            val subId = newSubId()
-            val eoseReceived = mutableSetOf<NormalizedRelayUrl>()
-            val allEose = CompletableDeferred<Unit>()
+            val gate = BatchEoseGate(scope, target = indexRelays.size)

            val listener =
                object : SubscriptionListener {
@@ -353,16 +383,19 @@ class FeedMetadataCoordinator(
                        relay: NormalizedRelayUrl,
                        forFilters: List<Filter>?,
                    ) {
-                        eoseReceived.add(relay)
-                        if (eoseReceived.size >= indexRelays.size) {
-                            allEose.complete(Unit)
-                        }
+                        gate.notifyEose(relay)
                    }
                }

            client.subscribe(subId, filterMap, listener)
-            withTimeoutOrNull(timeoutMs) { allEose.await() }
+            val eosedRelays = gate.awaitAll(timeoutMs)
            client.unsubscribe(subId)
+
+            if (eosedRelays > 0) {
+                queuedKind3Pubkeys.addAll(newPubkeys)
+            }
+            inFlightBatchedKind3.removeAll(newPubkeys.toSet())
+
            onEose()
        }
    }
@@ -375,5 +408,52 @@ class FeedMetadataCoordinator(
        queuedPubkeys.clear()
        queuedNoteIds.clear()
        queuedKind3Pubkeys.clear()
+        inFlightBatchedMetadata.clear()
+        inFlightBatchedKind3.clear()
+    }
+
+    /**
+     * Aggregates EOSE notifications from per-relay `onEose` callbacks
+     * (which the client may dispatch on `Dispatchers.IO`) via a
+     * [Channel]. The consumer coroutine is the sole reader/writer of the
+     * `seen` set, eliminating the race the previous `mutableSetOf` +
+     * shared-state check had — see PR #3483 review finding 6.
+     *
+     * [awaitAll] blocks up to [timeoutMs] and returns the number of
+     * relays that EOSE'd (may be less than [target] on timeout). The
+     * count feeds the retry decision in the batched loaders.
+     */
+    private class BatchEoseGate(
+        private val scope: CoroutineScope,
+        private val target: Int,
+    ) {
+        private val incoming = Channel<NormalizedRelayUrl>(Channel.UNLIMITED)
+        private val done = CompletableDeferred<Unit>()
+
+        @Volatile private var lastCount = 0
+
+        fun notifyEose(relay: NormalizedRelayUrl) {
+            incoming.trySend(relay)
+        }
+
+        suspend fun awaitAll(timeoutMs: Long): Int {
+            if (target <= 0) return 0
+            val consumer =
+                scope.launch {
+                    val seen = mutableSetOf<NormalizedRelayUrl>()
+                    for (relay in incoming) {
+                        if (seen.add(relay)) {
+                            lastCount = seen.size
+                            if (seen.size >= target && !done.isCompleted) {
+                                done.complete(Unit)
+                            }
+                        }
+                    }
+                }
+            withTimeoutOrNull(timeoutMs) { done.await() }
+            incoming.close()
+            consumer.join()
+            return lastCount
+        }
    }
 }
@@ -0,0 +1,297 @@
+/*
+ * 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.commons.relayClient.assemblers
+
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+
+/**
+ * Regression tests for PR #3483 review findings on FeedMetadataCoordinator:
+ *
+ *   - Finding 5: `queuedKind3Pubkeys` was marked-on-send, so if every index
+ *     relay timed out the pubkeys were permanently marked and subsequent
+ *     calls short-circuited — WoT stayed empty for the whole session.
+ *     Fix: pubkeys land in `queuedKind3Pubkeys` only after ≥1 EOSE; on
+ *     zero-EOSE timeout they roll out of `inFlightBatchedKind3` for retry.
+ *
+ *   - Finding 6: `eoseReceived: MutableSet` was mutated from per-relay
+ *     `onEose` callbacks running on `Dispatchers.IO` with no sync. Fix:
+ *     `BatchEoseGate` funnels EOSE notifications through a `Channel` so a
+ *     single consumer coroutine is the sole reader/writer of the `seen`
+ *     set.
+ */
+class FeedMetadataCoordinatorTest {
+    private lateinit var scope: CoroutineScope
+    private val relay1 = NormalizedRelayUrl("wss://relay1.test/")
+    private val relay2 = NormalizedRelayUrl("wss://relay2.test/")
+    private val relay3 = NormalizedRelayUrl("wss://relay3.test/")
+    private val indexRelays = setOf(relay1, relay2, relay3)
+
+    @Before
+    fun setup() {
+        scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+    }
+
+    @After
+    fun teardown() {
+        scope.cancel()
+    }
+
+    private fun pubkey(seed: Int): HexKey = seed.toString(16).padStart(64, '0')
+
+    /**
+     * Fake client that captures subscribe/unsubscribe and lets the test
+     * drive EOSE notifications on any dispatcher we choose.
+     */
+    private class ControllableClient(
+        private val delegate: INostrClient = EmptyNostrClient(),
+    ) : INostrClient by delegate {
+        val subscriptions = mutableMapOf<String, SubscriptionListener?>()
+        val subscribeCalls = mutableListOf<Map<NormalizedRelayUrl, List<Filter>>>()
+        var unsubscribeCallCount = 0
+            private set
+
+        override fun subscribe(
+            subId: String,
+            filters: Map<NormalizedRelayUrl, List<Filter>>,
+            listener: SubscriptionListener?,
+        ) {
+            subscriptions[subId] = listener
+            subscribeCalls.add(filters)
+        }
+
+        override fun unsubscribe(subId: String) {
+            subscriptions.remove(subId)
+            unsubscribeCallCount++
+        }
+
+        fun fireEose(relay: NormalizedRelayUrl) {
+            subscriptions.values.filterNotNull().forEach { it.onEose(relay, forFilters = null) }
+        }
+    }
+
+    @Test
+    fun `loadKind3Batched retries after zero-EOSE timeout`() =
+        runBlocking {
+            val client = ControllableClient()
+            val coordinator =
+                FeedMetadataCoordinator(
+                    client = client,
+                    scope = scope,
+                    indexRelays = indexRelays,
+                )
+
+            val pubkeys = listOf(pubkey(1), pubkey(2), pubkey(3))
+
+            // Call 1 — no relay EOSEs; must time out.
+            coordinator.loadKind3Batched(pubkeys, timeoutMs = 200)
+            delay(350) // exceed the timeout
+
+            // Call 2 — the same pubkeys must be re-subscribed since call 1
+            // never got a successful EOSE. The old code would silently
+            // short-circuit here.
+            coordinator.loadKind3Batched(pubkeys, timeoutMs = 200)
+            delay(50) // let the launcher run
+
+            assertEquals(
+                "Zero-EOSE timeout must not permanently dedup pubkeys",
+                2,
+                client.subscribeCalls.size,
+            )
+            assertEquals(
+                "Second call must re-request the same author set",
+                pubkeys.size,
+                client.subscribeCalls[1]
+                    .values
+                    .first()
+                    .first()
+                    .authors!!
+                    .size,
+            )
+        }
+
+    @Test
+    fun `loadKind3Batched short-circuits after successful EOSE`() =
+        runBlocking {
+            val client = ControllableClient()
+            val coordinator =
+                FeedMetadataCoordinator(
+                    client = client,
+                    scope = scope,
+                    indexRelays = indexRelays,
+                )
+
+            val pubkeys = listOf(pubkey(1), pubkey(2))
+
+            coordinator.loadKind3Batched(pubkeys, timeoutMs = 1_000)
+            // Give the launcher time to register the listener before we fire.
+            delay(50)
+            indexRelays.forEach(client::fireEose)
+            delay(200) // let the coordinator finish + promote to queued
+
+            coordinator.loadKind3Batched(pubkeys, timeoutMs = 200)
+            delay(50)
+
+            assertEquals(
+                "Successful call must dedup subsequent identical calls",
+                1,
+                client.subscribeCalls.size,
+            )
+        }
+
+    @Test
+    fun `loadKind3Batched promotes even when only some relays EOSE`() =
+        runBlocking {
+            val client = ControllableClient()
+            val coordinator =
+                FeedMetadataCoordinator(
+                    client = client,
+                    scope = scope,
+                    indexRelays = indexRelays,
+                )
+
+            val pubkeys = listOf(pubkey(1))
+
+            coordinator.loadKind3Batched(pubkeys, timeoutMs = 300)
+            delay(30)
+            // Only 1 of 3 EOSEs — timeout still fires but we made progress.
+            client.fireEose(relay1)
+            delay(400)
+
+            coordinator.loadKind3Batched(pubkeys, timeoutMs = 200)
+            delay(50)
+
+            assertEquals(
+                "≥1 EOSE = progress = promote to queued (avoid re-asking)",
+                1,
+                client.subscribeCalls.size,
+            )
+        }
+
+    /**
+     * Regression for finding 6 — pumps EOSE from many dispatchers in
+     * parallel. The old MutableSet-based code could drop entries or throw
+     * ConcurrentModificationException on the internal HashSet iterator.
+     * BatchEoseGate must aggregate every distinct relay exactly once.
+     */
+    @Test
+    fun `EOSE aggregator is safe under concurrent per-relay callbacks`() =
+        runBlocking {
+            val bigIndexSet =
+                (0..19).map { NormalizedRelayUrl("wss://relay$it.test/") }.toSet()
+            val client = ControllableClient()
+            val coordinator =
+                FeedMetadataCoordinator(
+                    client = client,
+                    scope = scope,
+                    indexRelays = bigIndexSet,
+                )
+
+            coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 2_000)
+            delay(50) // wait for subscription
+
+            // Fire EOSEs concurrently from many dispatchers.
+            val jobs =
+                bigIndexSet.map { relay ->
+                    scope.launch(Dispatchers.IO) {
+                        client.fireEose(relay)
+                    }
+                }
+            jobs.forEach { it.join() }
+
+            // The 2nd call must short-circuit — every relay EOSE'd, so
+            // pubkey(1) is now in queuedKind3Pubkeys.
+            delay(100)
+            coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 200)
+            delay(50)
+
+            assertEquals(
+                "Under concurrent EOSE from all relays, aggregator must reach target",
+                1,
+                client.subscribeCalls.size,
+            )
+        }
+
+    @Test
+    fun `loadMetadataBatched follows the same retry semantics`() =
+        runBlocking {
+            val client = ControllableClient()
+            val coordinator =
+                FeedMetadataCoordinator(
+                    client = client,
+                    scope = scope,
+                    indexRelays = indexRelays,
+                )
+
+            val pubkeys = listOf(pubkey(1), pubkey(2))
+
+            // Call 1 — zero EOSE, timeout.
+            coordinator.loadMetadataBatched(pubkeys, timeoutMs = 200)
+            delay(350)
+            // Call 2 — must re-subscribe.
+            coordinator.loadMetadataBatched(pubkeys, timeoutMs = 200)
+            delay(50)
+
+            assertTrue(
+                "Metadata batch also retries on zero-EOSE timeout",
+                client.subscribeCalls.size >= 2,
+            )
+        }
+
+    @Test
+    fun `clear releases in-flight dedup so a fresh call always fires`() =
+        runBlocking {
+            val client = ControllableClient()
+            val coordinator =
+                FeedMetadataCoordinator(
+                    client = client,
+                    scope = scope,
+                    indexRelays = indexRelays,
+                )
+
+            coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 200)
+            delay(50)
+            // clear() must drop the in-flight tracker even mid-request.
+            coordinator.clear()
+            delay(300) // let call 1 finish + roll back
+
+            coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 200)
+            delay(50)
+
+            assertTrue(client.subscribeCalls.size >= 2)
+        }
+}