From 0d2e8209a4da66e45f063dfc9647beee207e95a6 Mon Sep 17 00:00:00 2001 From: Riccardo Balbo Date: Fri, 19 Jun 2026 01:18:13 +0200 Subject: [PATCH 1/3] yield event loop when processing multiple invoices --- tasks.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tasks.py b/tasks.py index d7e9015..8dffedb 100644 --- a/tasks.py +++ b/tasks.py @@ -231,6 +231,7 @@ async def _on_multi_pay_invoice( results.append(r) except Exception as e: results.append((None, {"code": "INTERNAL", "message": str(e)}, [])) + await asyncio.sleep(0) # await log_nwc(pubkey, payload) return results @@ -440,6 +441,7 @@ async def _on_list_transactions( "metadata": {}, } ) + await asyncio.sleep(0) # await log_nwc(pubkey, payload) return [({"transactions": transactions}, None, [])] From bb998109f4a468bb9974fe86f337c03bf29c08cd Mon Sep 17 00:00:00 2001 From: Riccardo Balbo Date: Wed, 24 Jun 2026 16:24:40 +0200 Subject: [PATCH 2/3] track seen events until they expire --- nwcp.py | 24 +++++++++++++++++- tests/unit/test_nwcp.py | 55 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 77 insertions(+), 2 deletions(-) diff --git a/nwcp.py b/nwcp.py index efa5279..cafd1ce 100644 --- a/nwcp.py +++ b/nwcp.py @@ -27,6 +27,7 @@ def __init__(self): self.responses_eose = False self.events: dict[str, dict] = {} self.responses: list[str] = [] + self.seen_requests: dict[str, int] = {} def get_stale(self) -> list[dict]: """ @@ -49,6 +50,7 @@ def gc(self, expire: int | None = None): """ Garbage collection, remove all the events that have a response older than expire seconds (defaults to 1 hour if 0 or None) + and all seen requests that are expired """ expire = expire or 1 * 60 * 60 now = int(time.time()) @@ -65,6 +67,11 @@ def gc(self, expire: int | None = None): if len(deleted_ids) > 0: logger.debug("Garbage collected " + str(len(deleted_ids)) + " events") + # Clean seen requests + for event_id, expiry in list(self.seen_requests.items()): + if expiry < now: + del self.seen_requests[event_id] + class Config: arbitrary_types_allowed = True @@ -142,6 +149,8 @@ def __init__( # (handles reboots) self.handle_missed_events = handle_missed_events + self.event_max_age = self.handle_missed_events or 5 * 60 + logger.info( "NWC Service is ready. relay: " + str(self.relay) @@ -270,11 +279,15 @@ async def _ratelimit(self, unit: str, max_sleep_time: int = 120) -> None: await asyncio.sleep(limit.backoff) limit.last_attempt_time = int(time.time()) + def _create_subscription(self) -> MainSubscription: + self.sub = MainSubscription() + return self.sub + async def _subscribe(self): """ [Re]Subscribe to receive nip 47 requests and responses from the relay """ - self.sub = MainSubscription() + self.sub = self._create_subscription() # Create requests subscription req_filter = { "kinds": [23194], @@ -335,6 +348,15 @@ async def _handle_request(self, event: dict) -> list[dict]: """ Handle a nwc request """ + expire = self.sub.seen_requests.get(event["id"]) + if expire or event["created_at"] < int(time.time() - self.event_max_age): + raise Exception("Event is too old or already handled") + + expiration = self._extract_expiration_from_tags(event["tags"]) + if expiration <= 0: + expiration = int(time.time() + self.event_max_age) + self.sub.seen_requests[event["id"]] = expiration + nwc_pubkey = event["pubkey"] content = event["content"] # Decrypt the content diff --git a/tests/unit/test_nwcp.py b/tests/unit/test_nwcp.py index 800e853..6cd4d13 100644 --- a/tests/unit/test_nwcp.py +++ b/tests/unit/test_nwcp.py @@ -2,6 +2,7 @@ import json import random import string +import time import pytest from loguru import logger @@ -75,6 +76,18 @@ def test_signverify(nwc_service_provider, nwc_service_provider2): assert nwc_service_provider2._verify_event(signed) +def test_default_event_max_age(nwc_service_provider): + assert nwc_service_provider.event_max_age == 5 * 60 + assert ( + NWCServiceProvider( + "d7b5232fba0e02e32cfe26f20cdf2c803b27ecd81052c2dd5d17e5e1a333fe58", + "", + handle_missed_events=123, + ).event_max_age + == 123 + ) + + @pytest.mark.asyncio async def test_handle(nwc_service_provider, nwc_service_provider2): content = nwc_service_provider._json_dumps( @@ -87,7 +100,7 @@ async def test_handle(nwc_service_provider, nwc_service_provider2): "kind": 23194, "content": content, "tags": [["p", nwc_service_provider2.public_key_hex]], - "created_at": 1234567890, + "created_at": int(time.time()), } signed = nwc_service_provider._sign_event(event) @@ -101,6 +114,7 @@ async def _send_pass(obj): pass nwc_service_provider2._send = _send_pass + nwc_service_provider2._create_subscription() nwc_service_provider2.add_request_listener("pay_invoice", _handle_pay_invoice) sent_events = await nwc_service_provider2._handle_request(signed) assert len(sent_events) == 1 @@ -128,6 +142,45 @@ async def _send_pass(obj): assert p_tag[0][1] == nwc_service_provider.public_key_hex +@pytest.mark.asyncio +async def test_handle_rejects_same_event_replay( + nwc_service_provider, nwc_service_provider2 +): + content = nwc_service_provider._json_dumps( + {"method": "pay_invoice", "params": {"invoice": "abc"}} + ) + content = nwc_service_provider.private_key.encrypt_message( + content, nwc_service_provider2.public_key_hex + ) + event = { + "kind": 23194, + "content": content, + "tags": [["p", nwc_service_provider2.public_key_hex]], + "created_at": int(time.time()), + } + signed = nwc_service_provider._sign_event(event) + calls = 0 + + async def _handle_pay_invoice(provider, pubkey, content): + nonlocal calls + calls += 1 + return [({"preimage": "00000"}, None, [])] + + async def _send_pass(obj): + pass + + nwc_service_provider2._send = _send_pass + nwc_service_provider2._create_subscription() + nwc_service_provider2.add_request_listener("pay_invoice", _handle_pay_invoice) + + await nwc_service_provider2._handle_request(signed) + + with pytest.raises(Exception, match="already handled"): + await nwc_service_provider2._handle_request(signed) + + assert calls == 1 + + @pytest.mark.asyncio async def test_send_info_event(nwc_service_provider): """_send_info_event should publish a signed kind-13194 event.""" From f3cae9be4f28f48f90503403517c99e6c46c4a7b Mon Sep 17 00:00:00 2001 From: Riccardo Balbo Date: Wed, 24 Jun 2026 17:02:47 +0200 Subject: [PATCH 3/3] linter --- nwcp.py | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/nwcp.py b/nwcp.py index cafd1ce..20c3195 100644 --- a/nwcp.py +++ b/nwcp.py @@ -132,7 +132,7 @@ def __init__( self.info_event_task = None # Subscription - self.sub = None + self.sub: MainSubscription | None = None self.rate_limit: dict[str, RateLimit] = {} # websocket connection @@ -280,14 +280,15 @@ async def _ratelimit(self, unit: str, max_sleep_time: int = 120) -> None: limit.last_attempt_time = int(time.time()) def _create_subscription(self) -> MainSubscription: - self.sub = MainSubscription() - return self.sub + sub = MainSubscription() + self.sub = sub + return sub async def _subscribe(self): """ [Re]Subscribe to receive nip 47 requests and responses from the relay """ - self.sub = self._create_subscription() + sub = self._create_subscription() # Create requests subscription req_filter = { "kinds": [23194], @@ -295,17 +296,17 @@ async def _subscribe(self): # Since the last handle_missed_events seconds (handles reboots) "since": int(time.time()) - self.handle_missed_events, } - self.sub.requests_sub_id = self._get_new_subid() + sub.requests_sub_id = self._get_new_subid() # Create responses subscription (needed to track previosly responded requests) res_filter = { "kinds": [23195], "authors": [self.public_key_hex], "since": int(time.time()) - self.handle_missed_events, } - self.sub.responses_sub_id = self._get_new_subid() + sub.responses_sub_id = self._get_new_subid() # Subscribe - await self._send(["REQ", self.sub.requests_sub_id, req_filter]) - await self._send(["REQ", self.sub.responses_sub_id, res_filter]) + await self._send(["REQ", sub.requests_sub_id, req_filter]) + await self._send(["REQ", sub.responses_sub_id, res_filter]) async def _on_connection(self, _): """ @@ -348,14 +349,18 @@ async def _handle_request(self, event: dict) -> list[dict]: """ Handle a nwc request """ - expire = self.sub.seen_requests.get(event["id"]) + if not self.sub: + raise Exception("Subscription is not established") + sub = self.sub + + expire = sub.seen_requests.get(event["id"]) if expire or event["created_at"] < int(time.time() - self.event_max_age): raise Exception("Event is too old or already handled") expiration = self._extract_expiration_from_tags(event["tags"]) if expiration <= 0: expiration = int(time.time() + self.event_max_age) - self.sub.seen_requests[event["id"]] = expiration + sub.seen_requests[event["id"]] = expiration nwc_pubkey = event["pubkey"] content = event["content"]