From 275df2ae9c3e587cbcea4a751aca87c02657c115 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 16 Jul 2026 00:31:18 +0000 Subject: [PATCH] fix(desktop): drain kind:10002 back-fill fully and close on EOSE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit follow-ups on the DM relay-list work: - Bug: scheduleOutboxBackfill re-scheduled itself from inside the still-active drain job, so the isActive guard made the tail call a no-op — any authors past the first 100-author batch in a burst were stranded until another kind:0 happened to arrive. Drain in a while-loop inside one job instead, and mark it @Synchronized so concurrent consume-path callers can't spawn duplicate jobs. - Perf: the back-fill held each REQ open for a fixed 8s. Use fetchAll, which returns on EOSE (bounded by the timeout), and reuses existing infra. - DmInboxRelayResolver: drop the redundant `+ cachedOutbox` in the phase-2 seed (cached outbox is already in phase1Seed, so it was always subtracted back out), and bound the phase-2 fallback fetch to 5s so a cold send can't stack two full fan-out timeouts. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01KSc3LhGFSF5VZKr9h3qfn3 --- .../nip17Dm/DmInboxRelayResolver.kt | 14 ++++-- .../DesktopRelaySubscriptionsCoordinator.kt | 47 +++++++++---------- 2 files changed, 31 insertions(+), 30 deletions(-) diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/nip17Dm/DmInboxRelayResolver.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/nip17Dm/DmInboxRelayResolver.kt index 977782eda7..6c75108430 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/nip17Dm/DmInboxRelayResolver.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/nip17Dm/DmInboxRelayResolver.kt @@ -138,13 +138,17 @@ class DmInboxRelayResolver( // Phase 2: the recipient's own write relays are where their kind:10050 // lives per NIP-65. If phase 1 (mostly the curated indexers) didn't - // surface it, read straight from the write relays we just learned about - // — the kind:10002 the indexers returned, unioned with anything cached — - // minus what we already queried. + // surface it, read straight from the write relays the indexers' + // kind:10002 just pointed us at — minus what phase 1 already queried + // (which already includes any cached outbox relays). Bounded by a + // shorter timeout so a cold send doesn't stack two full fan-out waits. if (relays.isEmpty()) { - val writeRelays = (lists.nip65Write().toSet() + cachedOutbox) - phase1Seed + val writeRelays = lists.nip65Write().toSet() - phase1Seed if (writeRelays.isNotEmpty()) { - relays = RecipientRelayFetcher.fetchRelayLists(unauthenticatedClient, pubkey, writeRelays).dmInbox + relays = + RecipientRelayFetcher + .fetchRelayLists(unauthenticatedClient, pubkey, writeRelays, timeoutMs = 5_000L) + .dmInbox } } diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt index 371b54f1b9..9aff62678a 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt @@ -32,6 +32,7 @@ import com.vitorpamplona.amethyst.desktop.model.DesktopDmRelayState import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll 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 @@ -188,46 +189,42 @@ class DesktopRelaySubscriptionsCoordinator( * Drain [pendingOutbox] in batches of ≤100 authors, one REQ at a time, * coalescing bursts with a short delay. kind:10002 lives on the index / * discovery relays (they mirror it widely), so we query those; results go - * through [consumeEvent] → cache. Reschedules itself while work remains. + * through [consumeEvent] → cache. + * + * `@Synchronized` + the `isActive` guard make this launch-once: concurrent + * callers from the consume path don't spawn duplicate drain jobs. The job + * loops until [pendingOutbox] is empty, so authors that arrive mid-fetch + * (or a burst larger than one 100-author batch) are picked up in the same + * run instead of stranded until the next profile happens to arrive. */ + @Synchronized private fun scheduleOutboxBackfill() { if (outboxBackfillJob?.isActive == true) return outboxBackfillJob = scope.launch(Dispatchers.IO) { - // Coalesce a burst of profile arrivals into one batched REQ. + // Coalesce a burst of profile arrivals into the first batch. delay(500) - val batch = pendingOutbox.take(100).toList() - if (batch.isEmpty()) return@launch - pendingOutbox.removeAll(batch.toSet()) + while (pendingOutbox.isNotEmpty()) { + val batch = pendingOutbox.take(100).toList() + pendingOutbox.removeAll(batch.toSet()) + if (batch.isEmpty() || indexRelays.isEmpty()) continue - if (indexRelays.isNotEmpty()) { - val subId = generateSubId("outbox-backfill") val filter = Filter( kinds = listOf(AdvertisedRelayListEvent.KIND), authors = batch, limit = batch.size, ) - val listener = - object : SubscriptionListener { - override fun onEvent( - event: Event, - isLive: Boolean, - relay: NormalizedRelayUrl, - forFilters: List?, - ) { - consumeEvent(event, relay) - } - } - client.subscribe(subId, indexRelays.associateWith { listOf(filter) }, listener) - // Give the index relays time to return before closing. - delay(8.seconds) - client.unsubscribe(subId) + // fetchAll closes on EOSE (bounded by the timeout) rather + // than holding the sub open for a fixed window. + val events = + client.fetchAll( + filters = indexRelays.associateWith { listOf(filter) }, + timeoutMs = 8.seconds.inWholeMilliseconds, + ) + events.forEach { consumeEvent(it, null) } } - - // More arrived while we were fetching — go again. - if (pendingOutbox.isNotEmpty()) scheduleOutboxBackfill() } }