From 5217035f94644582efb60b143597af9015a20b60 Mon Sep 17 00:00:00 2001 From: nrobi144 Date: Tue, 7 Jul 2026 13:18:38 +0300 Subject: [PATCH] fix(coordinator): retryable batched-REQ dedup and race-free EOSE aggregator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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()` 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 --- .../assemblers/FeedMetadataCoordinator.kt | 116 +++++-- .../assemblers/FeedMetadataCoordinatorTest.kt | 297 ++++++++++++++++++ 2 files changed, 395 insertions(+), 18 deletions(-) create mode 100644 commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinator.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinator.kt index ca5f1bb770..d2ba16fca3 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinator.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinator.kt @@ -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() private val queuedKind3Pubkeys = mutableSetOf() + // 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() + private val inFlightBatchedKind3 = mutableSetOf() + /** * 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, 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() - val allEose = CompletableDeferred() + val gate = BatchEoseGate(scope, target = indexRelays.size) val listener = object : SubscriptionListener { @@ -289,16 +307,18 @@ class FeedMetadataCoordinator( relay: NormalizedRelayUrl, forFilters: List?, ) { - 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, 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() - val allEose = CompletableDeferred() + val gate = BatchEoseGate(scope, target = indexRelays.size) val listener = object : SubscriptionListener { @@ -353,16 +383,19 @@ class FeedMetadataCoordinator( relay: NormalizedRelayUrl, forFilters: List?, ) { - 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(Channel.UNLIMITED) + private val done = CompletableDeferred() + + @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() + 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 + } } } diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt new file mode 100644 index 0000000000..c0f8a609d4 --- /dev/null +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt @@ -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() + val subscribeCalls = mutableListOf>>() + var unsubscribeCallCount = 0 + private set + + override fun subscribe( + subId: String, + filters: Map>, + 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) + } +}