Code review:

rethrow CancellationException; offload NWC verify off WS thread align verification API with Amethyst Android
2026-04-25 18:02:39 +02:00
parent 68fb6fb703
commit 3dc79ac6a5
6 changed files with 102 additions and 126 deletions
@@ -29,6 +29,7 @@ import com.vitorpamplona.amethyst.commons.model.cache.ICacheProvider
 import com.vitorpamplona.amethyst.commons.model.cache.LargeSoftCache
 import com.vitorpamplona.amethyst.commons.services.nwc.NwcPaymentTracker
 import com.vitorpamplona.quartz.nip01Core.core.Address
+import com.vitorpamplona.quartz.nip01Core.core.AddressableEvent
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
 import com.vitorpamplona.quartz.nip01Core.crypto.checkSignature
@@ -52,6 +53,7 @@ import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent
 import com.vitorpamplona.quartz.nip57Zaps.LnZapRequestEvent
 import com.vitorpamplona.quartz.utils.DualCase
 import com.vitorpamplona.quartz.utils.Log
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.channels.BufferOverflow
@@ -165,19 +167,12 @@ class DesktopLocalCache : ICacheProvider {
        }
    }

-    // ----- Event verification -----
-
-    /**
-     * Verifies an event's id-hash and Schnorr signature. On failure, logs
-     * the offending kind/createdAt and returns false so callers can drop it.
-     * Mirrors Amethyst Android's [com.vitorpamplona.amethyst.model.LocalCache.justVerify]
-     * so unverified relay traffic never reaches the desktop cache.
-     */
    fun justVerify(event: Event): Boolean =
        if (!event.verify()) {
            try {
                event.checkSignature()
            } catch (e: Exception) {
+                if (e is CancellationException) throw e
                Log.w("Event Verification Failed") {
                    "Kind: ${event.kind} createdAt=${event.createdAt} id=${event.id} pubkey=${event.pubKey} reason=${e.message}"
                }
@@ -189,25 +184,16 @@ class DesktopLocalCache : ICacheProvider {

    // ----- Event consumption -----

-    /**
-     * Routes an event to the appropriate consume method.
-     * Returns true if the event was consumed (new), false if already seen
-     * or if signature verification failed.
-     */
    fun consume(
        event: Event,
        relay: NormalizedRelayUrl?,
+        wasVerified: Boolean = false,
    ): Boolean {
-        if (!justVerify(event)) return false
-        return consumeAssumingVerified(event, relay)
+        if (!wasVerified && !justVerify(event)) return false
+        return route(event, relay)
    }

-    /**
-     * Routes an already-verified event to its consume method. Internal seam
-     * exposed for tests that exercise routing/cache logic with synthetic
-     * events whose signatures are not real.
-     */
-    internal fun consumeAssumingVerified(
+    private fun route(
        event: Event,
        relay: NormalizedRelayUrl?,
    ): Boolean =
@@ -493,8 +479,9 @@ class DesktopLocalCache : ICacheProvider {
        zappedNote: Note?,
        relay: NormalizedRelayUrl?,
        onResponse: suspend (LnZapPaymentResponseEvent) -> Unit,
+        wasVerified: Boolean = false,
    ): Boolean {
-        if (!justVerify(event)) return false
+        if (!wasVerified && !justVerify(event)) return false
        val note = getOrCreateNote(event.id)
        val author = getOrCreateUser(event.pubKey)

@@ -521,8 +508,9 @@ class DesktopLocalCache : ICacheProvider {
    fun consume(
        event: LnZapPaymentResponseEvent,
        relay: NormalizedRelayUrl?,
+        wasVerified: Boolean = false,
    ): Boolean {
-        if (!justVerify(event)) return false
+        if (!wasVerified && !justVerify(event)) return false
        val requestId = event.requestId()
        val pending = paymentTracker.onResponseReceived(requestId) ?: return false

@@ -589,7 +577,7 @@ class DesktopLocalCache : ICacheProvider {
        if (!justVerify(event)) return false
        // For addressable/replaceable events, store in the addressable note cache
        // so state holders (Nip65RelayListState, etc.) pick it up via their flows
-        if (event is com.vitorpamplona.quartz.nip01Core.core.AddressableEvent) {
+        if (event is AddressableEvent) {
            val address = event.address()
            val note = getOrCreateAddressableNote(address)
            val author = getOrCreateUser(event.pubKey) ?: return false
@@ -33,6 +33,7 @@ import com.vitorpamplona.quartz.nip47WalletConnect.events.LnZapPaymentResponseEv
 import com.vitorpamplona.quartz.nip47WalletConnect.rpc.PayInvoiceErrorResponse
 import com.vitorpamplona.quartz.nip47WalletConnect.rpc.PayInvoiceSuccessResponse
 import com.vitorpamplona.quartz.nip47WalletConnect.rpc.Response
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlinx.coroutines.withTimeoutOrNull
 import kotlin.coroutines.resume
@@ -134,27 +135,31 @@ class NwcPaymentHandler(
                filters = listOf(filter),
                onEvent = { event, relay ->
                    if (event is LnZapPaymentResponseEvent && event.requestId() == requestId) {
-                        // Unsubscribe
-                        relayManager.closeSubscription(nwcConnection.relayUri, subId)
+                        // Move verify + cache mutation + decrypt off the relay's
+                        // WebSocket reader thread; Schnorr verify is non-trivial
+                        // CPU work and shouldn't block frame parsing.
+                        @OptIn(kotlinx.coroutines.DelicateCoroutinesApi::class)
+                        kotlinx.coroutines.GlobalScope.launch(kotlinx.coroutines.Dispatchers.IO) {
+                            if (!localCache.justVerify(event)) return@launch

-                        // Store response note and link to zapped note
-                        val responseNote = localCache.getOrCreateNote(event.id)
-                        responseNote.loadEvent(event, localCache.getOrCreateUser(event.pubKey), emptyList())
-                        responseNote.addRelay(relay)
-                        zappedNote?.addZapPayment(requestNote, responseNote)
+                            relayManager.closeSubscription(nwcConnection.relayUri, subId)

-                        // Decrypt and process response
-                        try {
-                            kotlinx.coroutines.runBlocking {
+                            val responseNote = localCache.getOrCreateNote(event.id)
+                            responseNote.loadEvent(event, localCache.getOrCreateUser(event.pubKey), emptyList())
+                            responseNote.addRelay(relay)
+                            zappedNote?.addZapPayment(requestNote, responseNote)
+
+                            try {
                                val response = event.decrypt(nwcSigner)
                                val result = processResponse(response)
                                if (continuation.isActive) {
                                    continuation.resume(result)
                                }
-                            }
-                        } catch (e: Exception) {
-                            if (continuation.isActive) {
-                                continuation.resume(PaymentResult.Error("Failed to decrypt response: ${e.message}"))
+                            } catch (e: Exception) {
+                                if (e is kotlinx.coroutines.CancellationException) throw e
+                                if (continuation.isActive) {
+                                    continuation.resume(PaymentResult.Error("Failed to decrypt response: ${e.message}"))
+                                }
                            }
                        }
                    }
@@ -33,6 +33,8 @@ 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 com.vitorpamplona.quartz.utils.Log
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -120,46 +122,25 @@ class DesktopRelaySubscriptionsCoordinator(
    fun consumeEvent(
        event: Event,
        relay: NormalizedRelayUrl?,
+        wasVerified: Boolean = false,
    ) {
        scope.launch(Dispatchers.IO) {
            try {
-                runConsume(localCache.consume(event, relay), event)
+                if (!localCache.consume(event, relay, wasVerified)) return@launch
+                _lastEventAt.value = System.currentTimeMillis()
+                val note = localCache.getNoteIfExists(event.id) ?: return@launch
+                eventBundler.invalidateList(note) { batch ->
+                    localCache.eventStream.emitNewNotes(batch)
+                }
            } catch (e: Exception) {
-                println("Coordinator: failed to consume kind=${event.kind} id=${event.id} relay=$relay: ${e.message}")
+                if (e is CancellationException) throw e
+                Log.w("DesktopRelaySubscriptionsCoordinator") {
+                    "Failed to consume kind=${event.kind} id=${event.id} relay=$relay: ${e.message}"
+                }
            }
        }
    }

-    /**
-     * Test-only seam — bypasses signature verification in the cache so unit
-     * tests can pump synthetic events through the full coordinator pipeline.
-     * Production code MUST use [consumeEvent].
-     */
-    internal fun consumeEventAssumingVerified(
-        event: Event,
-        relay: NormalizedRelayUrl?,
-    ) {
-        scope.launch(Dispatchers.IO) {
-            try {
-                runConsume(localCache.consumeAssumingVerified(event, relay), event)
-            } catch (e: Exception) {
-                println("Coordinator: failed to consume kind=${event.kind} id=${event.id} relay=$relay: ${e.message}")
-            }
-        }
-    }
-
-    private suspend fun runConsume(
-        consumed: Boolean,
-        event: Event,
-    ) {
-        if (!consumed) return
-        _lastEventAt.value = System.currentTimeMillis()
-        val note = localCache.getNoteIfExists(event.id) ?: return
-        eventBundler.invalidateList(note) { batch ->
-            localCache.eventStream.emitNewNotes(batch)
-        }
-    }
-
    /**
     * Request a consolidated interaction subscription for the given note IDs.
     * Subscribes to kinds 7 (reactions), 9735 (zaps), 6 (reposts), and 1 (replies)