From 233a5b25b62a5fb1a6eceb2c480d024aeab63ef1 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Wed, 27 May 2026 17:14:44 -0400 Subject: [PATCH] fix(nwc): trim subscription filter on response to avoid relay replays Some NWC relays (notably relay-nwc.rizful.com) replay cached kind-23195 events every time a REQ filter mutates. Because we previously left each reqId in the filter for a full 60s after sending its request, every new NWC call added a fresh reqId to a filter that still listed several stale ones, and the relay would re-deliver every matching cached response. That arrived here as a flurry of NO_MATCH events, wasted work on each new send, and made the transactions screen feel stuck. Cleanup is now driven by the response itself: - NwcSignerState cancels the 60s safety-net job on response and asks the assembler to drop the filter through unsubscribeSoon, which is debounced 1.5s so a burst of responses produces one relay-side filter update instead of one per response. - subscribeAndFlush drains any pending unsubscribes synchronously before adding the new query, so the relay sees a direct {old}->{new} transition instead of the {old}->{old,new}->{new} that triggered Rizful's replay path. - The 60s timeout remains as a safety net for wallets that never reply. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../nip47WalletConnect/NwcSignerState.kt | 28 +++++-- .../nwc/NWCPaymentFilterAssembler.kt | 82 ++++++++++++++++++- 2 files changed, 101 insertions(+), 9 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nip47WalletConnect/NwcSignerState.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nip47WalletConnect/NwcSignerState.kt index 2647288d1a..d7abdecf83 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nip47WalletConnect/NwcSignerState.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nip47WalletConnect/NwcSignerState.kt @@ -157,14 +157,20 @@ class NwcSignerState( // be missed. assembler.subscribeAndFlush(filter) - scope.launch(Dispatchers.IO) { - delay(60000) - assembler.unsubscribe(filter) - } + // Safety net: drop the filter after 60s if the wallet never replies. + // The happy path (response arrives) cancels this job and unsubscribes + // through assembler.unsubscribeSoon, which debounces. + val timeoutJob = + scope.launch(Dispatchers.IO) { + delay(60000) + assembler.unsubscribe(filter) + } val responseCache = NostrWalletConnectResponseCache(walletSigner) cache.consume(event, null, true, walletService.relayUri) { + timeoutJob.cancel() onResponse(responseCache.decryptResponse(it)) + assembler.unsubscribeSoon(filter) } return Pair(event, walletService.relayUri) @@ -195,13 +201,19 @@ class NwcSignerState( // See sendNwcRequestToWallet above for the rationale. assembler.subscribeAndFlush(filter) - scope.launch(Dispatchers.IO) { - delay(60000) // waits 1 minute to complete payment. - assembler.unsubscribe(filter) - } + // Safety net: drop the filter after 60s if the wallet never replies. + // The happy path (response arrives) cancels this job and instead + // hands off to assembler.unsubscribeSoon, which debounces. + val timeoutJob = + scope.launch(Dispatchers.IO) { + delay(60000) // waits 1 minute to complete payment. + assembler.unsubscribe(filter) + } cache.consume(event, zappedNote, true, walletService.relayUri) { + timeoutJob.cancel() onResponse(decryptResponse(it)) + assembler.unsubscribeSoon(filter) } return Pair(event, walletService.relayUri) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/nwc/NWCPaymentFilterAssembler.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/nwc/NWCPaymentFilterAssembler.kt index ef171dafde..cd02ea9577 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/nwc/NWCPaymentFilterAssembler.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/nwc/NWCPaymentFilterAssembler.kt @@ -25,6 +25,14 @@ import com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManager import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch // This allows multiple screen to be listening to tags, even the same tag. // The subscription filter carries `#e: [request id]` plus `#p: [client pubkey]`. @@ -48,6 +56,20 @@ class NWCPaymentFilterAssembler( NWCPaymentWatcherSubAssembler(client, ::allKeys), ) + // Trailing-edge debounce: after a response arrives we wait this long + // before actually removing the request id from the subscription filter. + // If more responses (or a fresh subscribe of any query) arrive within + // the window they coalesce into ONE relay-side filter update. This + // matters because some NWC relays (e.g. relay-nwc.rizful.com) replay + // cached kind-23195 events every time the REQ filter changes — so each + // unbatched filter mutation triggers a fresh replay storm. + private val unsubDelayMs = 1500L + private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + private val pendingLock = Any() + private val pendingUnsubs = mutableSetOf() + + @Volatile private var pendingJob: Job? = null + override fun invalidateFilters() = group.forEach { it.invalidateFilters() } override fun invalidateKeys() = invalidateFilters() @@ -58,11 +80,69 @@ class NWCPaymentFilterAssembler( * ephemeral event (kind 23195) and the subscription must be active on the * relay before we publish the request event — otherwise the relay drops * the response with no replay. + * + * Also drains any pending unsubscribes synchronously so the relay sees + * a direct {old} → {new} transition instead of {old} → {old, new} → {new}. + * The intermediate `{old, new}` state is the trigger that makes Rizful's + * relay replay cached responses for `old`. */ fun subscribeAndFlush(query: NWCPaymentQueryState) { + drainPendingUnsubs() subscribe(query) group.forEach { it.forceInvalidate() } } - override fun destroy() = group.forEach { it.destroy() } + /** + * Removes [query] from the active subscription after a short debounce. + * Repeated calls within the window coalesce into a single batch + * unsubscribe. Idempotent: calling with a query that's already been + * unsubscribed is a no-op. + */ + fun unsubscribeSoon(query: NWCPaymentQueryState) { + val oldJob: Job? + synchronized(pendingLock) { + pendingUnsubs.add(query) + oldJob = pendingJob + pendingJob = + scope.launch { + try { + delay(unsubDelayMs) + val batch = + synchronized(pendingLock) { + val copy = pendingUnsubs.toList() + pendingUnsubs.clear() + pendingJob = null + copy + } + if (batch.isNotEmpty()) { + unsubscribe(batch) + } + } catch (_: CancellationException) { + // Superseded by a newer schedule; the replacement + // job owns the pending set from here. + } + } + } + oldJob?.cancel() + } + + private fun drainPendingUnsubs() { + val batch: List + val cancelled: Job? + synchronized(pendingLock) { + batch = pendingUnsubs.toList() + pendingUnsubs.clear() + cancelled = pendingJob + pendingJob = null + } + cancelled?.cancel() + if (batch.isNotEmpty()) { + unsubscribe(batch) + } + } + + override fun destroy() { + scope.cancel() + group.forEach { it.destroy() } + } }