fix(desktop): drain kind:10002 back-fill fully and close on EOSE

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KSc3LhGFSF5VZKr9h3qfn3

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
             }
         }
 

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<Filter>?,
-                            ) {
-                                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()
             }
     }