From 115e8692258e26fb1285a585afe1a2b044a40717 Mon Sep 17 00:00:00 2001 From: Padreug Date: Mon, 27 Apr 2026 20:09:26 +0200 Subject: [PATCH 1/5] fix: queue outgoing events when relay connection is down When all relay connections are temporarily lost, EVENT messages published by extensions (nostrmarket, events) are now queued in a bounded deque (max 100) instead of being silently dropped. On reconnection, queued events are flushed to all connected relays. Dead relay queues are also drained before restart to preserve in-flight events. Closes aiolabs/nostrclient#1 Co-Authored-By: Claude Opus 4.6 (1M context) --- CLAUDE.md | 59 ++++++++++++++++++++++++++++++++++++++++ nostr/relay.py | 9 ++++++ nostr/relay_manager.py | 62 ++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 128 insertions(+), 2 deletions(-) create mode 100644 CLAUDE.md diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..981d3f0 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,59 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +Nostrclient is an LNbits extension that acts as an always-on Nostr relay multiplexer. Multiple Nostr clients connect to a single WebSocket endpoint, which fans out requests to multiple configured Nostr relays and aggregates/deduplicates responses. It rewrites subscription IDs per-client to prevent conflicts. + +## Build & Development Commands + +All commands use `uv` as the Python package manager: + +```bash +make format # Format: prettier + black + ruff --fix +make check # All checks: mypy, pyright, black --check, ruff check, prettier --check +make test # Run pytest (DEBUG=true PYTHONUNBUFFERED=1 uv run pytest) +make mypy # Type check (excludes nostr/ directory) +make pre-commit # Run pre-commit hooks on all files +``` + +Individual checks: `make checkblack`, `make checkruff`, `make checkprettier`. + +CI runs lint then pytest with `LNBITS_BACKEND_WALLET_CLASS=FakeWallet`. + +## Architecture + +**Request flow:** Nostr Clients → WebSocket → NostrRouter → RelayManager → Nostr Relays + +Key components: + +- **NostrRouter** (`router.py`) — One per client WebSocket connection. Rewrites subscription IDs (original → hashed → original) to isolate clients. Two async tasks: `_client_to_nostr` (forward requests) and `_nostr_to_client` (deliver aggregated responses). + +- **NostrClient** (`nostr/client/client.py`) — Singleton orchestrator. Owns the RelayManager. Polls MessagePool and dispatches events via callbacks to routers. + +- **RelayManager** (`nostr/relay_manager.py`) — Manages connections to multiple relays. Caches subscriptions so new relays receive existing subscriptions. Runs health checks via `check_and_restart_relays()`. + +- **Relay** (`nostr/relay.py`) — Individual relay WebSocket connection with retry/backoff, ping latency tracking, and error counting. + +- **MessagePool** (`nostr/message_pool.py`) — Thread-safe event aggregation with deduplication by event ID across all relays. + +**Hybrid threading model:** Relay connections use threads (via `RelayManager.open_connections()`); client communication uses asyncio. The bridge is in `tasks.py` where `subscribe_events()` runs in a thread executor. + +**Lifecycle** (`__init__.py`): `nostrclient_start()` spawns three background tasks (init relays, subscribe events, check relays). `nostrclient_stop()` cancels tasks, stops routers, closes the client. + +## API Endpoints (views_api.py) + +- REST endpoints under `/api/v1/` for relay CRUD and config (admin-authenticated) +- WebSocket endpoints: `/api/v1/{ws_id}` (private, encrypted ID) and `/api/v1/relay` (public, if enabled) + +## Database + +Three migrations in `migrations.py`: relays table, config table (JSON `extra` field), config owner scoping. CRUD in `crud.py` uses LNbits database abstraction. + +## Code Quality Notes + +- **mypy excludes `nostr/*`** — this is a custom Nostr protocol implementation, not a third-party package +- **Ruff rules:** F, E, W, I, A, C, N, UP, RUF, B +- **Frontend:** Vue.js + Quasar via LNbits base templates (`templates/nostrclient/index.html`) +- Pub key helpers in `helpers.py` normalize between hex and bech32 (npub1) formats diff --git a/nostr/relay.py b/nostr/relay.py index d762963..bac1be7 100644 --- a/nostr/relay.py +++ b/nostr/relay.py @@ -1,6 +1,9 @@ +from __future__ import annotations + import asyncio import json import time +from collections.abc import Callable from queue import Queue from loguru import logger @@ -28,6 +31,7 @@ class Relay: self.num_subscriptions: int = 0 self.queue: Queue = Queue() + self.on_connect: Callable[[Relay], None] | None = None def connect(self): self.ws = WebSocketApp( @@ -97,6 +101,11 @@ class Relay: logger.info(f"[Relay: {self.url}] Connected.") self.connected = True self.shutdown = False + if self.on_connect: + try: + self.on_connect(self) + except Exception as e: + logger.warning(f"[Relay: {self.url}] on_connect callback error: {e}") def _on_close(self, _, status_code, message): logger.warning( diff --git a/nostr/relay_manager.py b/nostr/relay_manager.py index 2aa27c5..2b90216 100644 --- a/nostr/relay_manager.py +++ b/nostr/relay_manager.py @@ -1,6 +1,8 @@ import asyncio +import json import threading import time +from collections import deque from typing import List from loguru import logger @@ -9,6 +11,8 @@ from .message_pool import MessagePool, NoticeMessage from .relay import Relay from .subscription import Subscription +PENDING_EVENTS_MAX = 100 + class RelayManager: def __init__(self) -> None: @@ -18,6 +22,8 @@ class RelayManager: self.message_pool = MessagePool() self._cached_subscriptions: dict[str, Subscription] = {} self._subscriptions_lock = threading.Lock() + self._pending_events: deque[str] = deque(maxlen=PENDING_EVENTS_MAX) + self._pending_events_lock = threading.Lock() def add_relay(self, url: str) -> Relay: if url in list(self.relays.keys()): @@ -25,6 +31,7 @@ class RelayManager: return self.relays[url] relay = Relay(url, self.message_pool) + relay.on_connect = self._on_relay_connect self.relays[url] = relay self._open_connection(relay) @@ -100,8 +107,19 @@ class RelayManager: relay.close() def publish_message(self, message: str): - for relay in self.relays.values(): - relay.publish(message) + connected_relays = [ + r for r in self.relays.values() if r.connected and not r.shutdown + ] + if connected_relays: + for relay in self.relays.values(): + relay.publish(message) + else: + with self._pending_events_lock: + self._pending_events.append(message) + logger.warning( + f"No connected relays. Queued outgoing event " + f"({len(self._pending_events)}/{PENDING_EVENTS_MAX})." + ) def handle_notice(self, notice: NoticeMessage): relay = next((r for r in self.relays.values() if r.url == notice.url)) @@ -137,7 +155,47 @@ class RelayManager: logger.info(f"Restarting connection to relay '{relay.url}'") + self._drain_relay_queue(relay) self.remove_relay(relay.url) new_relay = self.add_relay(relay.url) new_relay.error_counter = relay.error_counter new_relay.error_list = relay.error_list + + def _drain_relay_queue(self, relay: Relay): + """Move pending EVENT messages from a dead relay's queue to the + manager's pending queue so they can be resent on reconnection.""" + drained = 0 + while not relay.queue.empty(): + try: + message = relay.queue.get_nowait() + data = json.loads(message) + if data[0] == "EVENT": + with self._pending_events_lock: + self._pending_events.append(message) + drained += 1 + except Exception: + break + if drained: + logger.info(f"Drained {drained} pending event(s) from relay '{relay.url}'.") + + def _on_relay_connect(self, _relay: Relay): + self._flush_pending_events() + + def _flush_pending_events(self): + with self._pending_events_lock: + if not self._pending_events: + return + connected_relays = [ + r for r in self.relays.values() if r.connected and not r.shutdown + ] + if not connected_relays: + return + count = len(self._pending_events) + while self._pending_events: + message = self._pending_events.popleft() + for relay in connected_relays: + relay.publish(message) + logger.info( + f"Flushed {count} pending event(s) to " + f"{len(connected_relays)} relay(s)." + ) From 7aa0ce14dbe1146de9bcc2926bb923c4fcdd8ba9 Mon Sep 17 00:00:00 2001 From: Padreug Date: Sat, 12 Sep 2026 13:31:00 +0200 Subject: [PATCH 2/5] feat: reply OK to client-published events nostrclient never answered a client's EVENT with the NIP-01 `["OK", , , ]` command result. Clients built on nostr-tools and similar libraries wait for that reply before treating a publish as successful, so NWC wallet apps paired through the public endpoint reported "publish failed" even though the request had been fanned out and answered. Relay OKs now flow through the message pool like events and notices. The router tracks each EVENT a client publishes and replies exactly once: `true` as soon as any relay accepts, `false` once every relay connected at publish time has rejected it, after a 10 s timeout, or immediately when no relay is connected. OKs nobody is waiting on are dropped at the pump so the shared result map cannot grow unbounded. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013Tbyw6FwjhEJg3gHfPHxWt --- README.md | 1 + nostr/client/client.py | 11 ++++ nostr/message_pool.py | 24 +++++++ router.py | 83 ++++++++++++++++++++++- tasks.py | 21 +++++- tests/test_router_ok.py | 141 ++++++++++++++++++++++++++++++++++++++++ 6 files changed, 277 insertions(+), 4 deletions(-) create mode 100644 tests/test_router_ok.py diff --git a/README.md b/README.md index 70593a8..dee86a1 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,7 @@ flowchart LR 3. **Fan-Out** - Subscription requests are sent to all configured relays 4. **Aggregation** - Events from all relays are collected and deduplicated 5. **Response** - Events are sent back to the client with the original subscription ID +6. **Publish Acknowledgement** - Every `EVENT` a client publishes gets exactly one `["OK", , , ]` reply (NIP-01): `true` as soon as any relay accepts it, `false` once every relay has rejected it, the wait times out (10 s), or no relay is connected ## Configuration diff --git a/nostr/client/client.py b/nostr/client/client.py index d6fb5c8..0412c1b 100644 --- a/nostr/client/client.py +++ b/nostr/client/client.py @@ -39,11 +39,13 @@ class NostrClient: callback_events_func=None, callback_notices_func=None, callback_eosenotices_func=None, + callback_command_results_func=None, ): while self.running: self._check_events(callback_events_func) self._check_notices(callback_notices_func) self._check_eos_notices(callback_eosenotices_func) + self._check_command_results(callback_command_results_func) await asyncio.sleep(0.2) @@ -73,3 +75,12 @@ class NostrClient: callback_eosenotices_func(event_msg) except Exception as e: logger.debug(e) + + def _check_command_results(self, callback_command_results_func=None): + try: + while self.relay_manager.message_pool.has_command_results(): + result_msg = self.relay_manager.message_pool.get_command_result() + if callback_command_results_func: + callback_command_results_func(result_msg) + except Exception as e: + logger.debug(e) diff --git a/nostr/message_pool.py b/nostr/message_pool.py index a3e6c5f..30c1469 100644 --- a/nostr/message_pool.py +++ b/nostr/message_pool.py @@ -27,11 +27,22 @@ class EndOfStoredEventsMessage: self.url = url +class CommandResultMessage: + """An `["OK", , , ]` reply from one relay.""" + + def __init__(self, event_id: str, accepted: bool, message: str, url: str) -> None: + self.event_id = event_id + self.accepted = accepted + self.message = message + self.url = url + + class MessagePool: def __init__(self) -> None: self.events: Queue[EventMessage] = Queue() self.notices: Queue[NoticeMessage] = Queue() self.eose_notices: Queue[EndOfStoredEventsMessage] = Queue() + self.command_results: Queue[CommandResultMessage] = Queue() self._unique_events: set = set() self.lock: Lock = Lock() @@ -47,6 +58,9 @@ class MessagePool: def get_eose_notice(self): return self.eose_notices.get() + def get_command_result(self): + return self.command_results.get() + def has_events(self): return self.events.qsize() > 0 @@ -56,6 +70,9 @@ class MessagePool: def has_eose_notices(self): return self.eose_notices.qsize() > 0 + def has_command_results(self): + return self.command_results.qsize() > 0 + def _process_message(self, message: str, url: str): message_json = json.loads(message) message_type = message_json[0] @@ -75,6 +92,13 @@ class MessagePool: self.notices.put(NoticeMessage(message_json[1], url)) elif message_type == RelayMessageType.END_OF_STORED_EVENTS: self.eose_notices.put(EndOfStoredEventsMessage(message_json[1], url)) + elif message_type == RelayMessageType.COMMAND_RESULT: + message = message_json[3] if len(message_json) > 3 else "" + self.command_results.put( + CommandResultMessage( + message_json[1], bool(message_json[2]), str(message), url + ) + ) def _accept_event(self, event_message: EventMessage): """ diff --git a/router.py b/router.py index a7054e9..53ce387 100644 --- a/router.py +++ b/router.py @@ -1,5 +1,6 @@ import asyncio import json +import time from typing import ClassVar from fastapi import WebSocket, WebSocketDisconnect @@ -9,11 +10,30 @@ from loguru import logger from .nostr.client.client import NostrClient # from . import nostr_client -from .nostr.message_pool import EndOfStoredEventsMessage, EventMessage, NoticeMessage +from .nostr.message_pool import ( + CommandResultMessage, + EndOfStoredEventsMessage, + EventMessage, + NoticeMessage, +) nostr_client: NostrClient = NostrClient() all_routers: list["NostrRouter"] = [] +# How long to wait for relays to answer an EVENT before replying `OK false`. +PUBLISH_TIMEOUT_SECONDS = 10 + + +class PendingPublish: + """An EVENT a client sent that still awaits its `OK` reply.""" + + def __init__(self, event_id: str, relay_count: int) -> None: + self.event_id = event_id + # relays connected at publish time; once this many answered we stop + # waiting for an acceptance + self.relay_count = relay_count + self.sent_at = time.time() + class NostrRouter: received_subscription_events: ClassVar[dict[str, list[EventMessage]]] = {} @@ -21,12 +41,14 @@ class NostrRouter: received_subscription_eosenotices: ClassVar[dict[str, EndOfStoredEventsMessage]] = ( {} ) + received_command_results: ClassVar[dict[str, list[CommandResultMessage]]] = {} def __init__(self, websocket: WebSocket): self.connected: bool = True self.websocket: WebSocket = websocket self.tasks: list[asyncio.Task] = [] self.original_subscription_ids: dict[str, str] = {} + self.pending_publishes: dict[str, PendingPublish] = {} @property def subscriptions(self) -> list[str]: @@ -40,6 +62,9 @@ class NostrRouter: async def stop(self): nostr_client.relay_manager.close_subscriptions(self.subscriptions) self.connected = False + for event_id in self.pending_publishes: + NostrRouter.received_command_results.pop(event_id, None) + self.pending_publishes.clear() for t in self.tasks: try: @@ -74,6 +99,7 @@ class NostrRouter: while self.connected: try: await self._handle_subscriptions() + await self._handle_command_results() self._handle_notices() except Exception as e: logger.debug(f"Failed to handle response for client: '{e!s}'.") @@ -120,6 +146,42 @@ class NostrRouter: f"[NOSTRCLIENT] Error in _handle_received_subscription_events: {e}" ) + async def _handle_command_results(self): + """ + Reply exactly one `["OK", , , ]` per EVENT a + client published (NIP-01). The EVENT was fanned out to every relay, so + several OKs can come back for one id: the client gets `true` as soon + as any relay accepts, and `false` once every relay has rejected it or + the wait times out. + """ + for event_id in list(self.pending_publishes.keys()): + pending = self.pending_publishes[event_id] + results = NostrRouter.received_command_results.get(event_id, []) + accepted = next((r for r in results if r.accepted), None) + if accepted: + await self._send_ok(event_id, True, accepted.message) + elif len(results) >= pending.relay_count: + await self._send_ok(event_id, False, results[-1].message) + elif time.time() - pending.sent_at > PUBLISH_TIMEOUT_SECONDS: + message = ( + results[-1].message + if results + else "error: timed out waiting for relays" + ) + await self._send_ok(event_id, False, message) + else: + continue + self.pending_publishes.pop(event_id, None) + NostrRouter.received_command_results.pop(event_id, None) + + async def _send_ok(self, event_id: str, accepted: bool, message: str): + try: + await self.websocket.send_text( + json.dumps(["OK", event_id, accepted, message]) + ) + except Exception as e: + logger.debug(f"Failed to send OK for '{event_id}': {e}") + def _handle_notices(self): while len(NostrRouter.received_subscription_notices): my_event = NostrRouter.received_subscription_notices.pop(0) @@ -141,9 +203,26 @@ class NostrRouter: return if json_data[0] == "EVENT": - nostr_client.relay_manager.publish_message(json_str) + await self._handle_client_event(json_data, json_str) return + async def _handle_client_event(self, json_data, json_str): + event = json_data[1] if len(json_data) > 1 else None + event_id = event.get("id") if isinstance(event, dict) else None + if not event_id: + logger.debug("Ignoring EVENT without an id.") + return + + connected = [ + r for r in nostr_client.relay_manager.relays.values() if r.connected + ] + if not connected: + await self._send_ok(event_id, False, "error: no relays connected") + return + + self.pending_publishes[event_id] = PendingPublish(event_id, len(connected)) + nostr_client.relay_manager.publish_message(json_str) + def _handle_client_req(self, json_data): subscription_id = json_data[1] logger.info(f"New subscription: '{subscription_id}'") diff --git a/tasks.py b/tasks.py index 2a76765..e2abdcb 100644 --- a/tasks.py +++ b/tasks.py @@ -4,8 +4,13 @@ import threading from loguru import logger from .crud import get_relays -from .nostr.message_pool import EndOfStoredEventsMessage, EventMessage, NoticeMessage -from .router import NostrRouter, nostr_client +from .nostr.message_pool import ( + CommandResultMessage, + EndOfStoredEventsMessage, + EventMessage, + NoticeMessage, +) +from .router import NostrRouter, all_routers, nostr_client async def init_relays(): @@ -55,12 +60,24 @@ async def subscribe_events(): NostrRouter.received_subscription_eosenotices[sub_id] = event_message + def callback_command_results(result_message: CommandResultMessage): + event_id = result_message.event_id + # Only keep OKs some client is still waiting on; the rest would + # accumulate forever (events published by other relay users, or + # results arriving after the client already got its reply). + if not any(event_id in r.pending_publishes for r in all_routers): + return + NostrRouter.received_command_results.setdefault(event_id, []).append( + result_message + ) + def wrap_async_subscribe(): asyncio.run( nostr_client.subscribe( callback_events, callback_notices, callback_eose_notices, + callback_command_results, ) ) diff --git a/tests/test_router_ok.py b/tests/test_router_ok.py new file mode 100644 index 0000000..44863a1 --- /dev/null +++ b/tests/test_router_ok.py @@ -0,0 +1,141 @@ +import json +import time + +import pytest + +from .. import router as router_module +from ..nostr.message_pool import CommandResultMessage +from ..router import PUBLISH_TIMEOUT_SECONDS, NostrRouter + + +class FakeWebSocket: + def __init__(self): + self.sent: list[list] = [] + + async def send_text(self, text: str): + self.sent.append(json.loads(text)) + + +class FakeRelay: + def __init__(self, connected: bool): + self.connected = connected + + +class FakeRelayManager: + def __init__(self, relays: dict[str, FakeRelay]): + self.relays = relays + self.published: list[str] = [] + + def publish_message(self, message: str): + self.published.append(message) + + def close_subscriptions(self, subscriptions): + pass + + +EVENT_ID = "ab" * 32 +EVENT_MSG = json.dumps(["EVENT", {"id": EVENT_ID, "kind": 1, "content": "hi"}]) + + +def _router(monkeypatch, relays: dict[str, FakeRelay]): + manager = FakeRelayManager(relays) + monkeypatch.setattr(router_module.nostr_client, "relay_manager", manager) + NostrRouter.received_command_results.clear() + ws = FakeWebSocket() + return NostrRouter(ws), ws, manager # type: ignore[arg-type] + + +def _ok_from(url: str, accepted: bool, message: str = ""): + NostrRouter.received_command_results.setdefault(EVENT_ID, []).append( + CommandResultMessage(EVENT_ID, accepted, message, url) + ) + + +@pytest.mark.asyncio +async def test_no_connected_relays_replies_ok_false_immediately(monkeypatch): + router, ws, manager = _router(monkeypatch, {"wss://a": FakeRelay(False)}) + + await router._handle_client_to_nostr(EVENT_MSG) + + assert manager.published == [] + assert ws.sent == [["OK", EVENT_ID, False, "error: no relays connected"]] + assert router.pending_publishes == {} + + +@pytest.mark.asyncio +async def test_event_without_id_is_ignored(monkeypatch): + router, ws, manager = _router(monkeypatch, {"wss://a": FakeRelay(True)}) + + await router._handle_client_to_nostr(json.dumps(["EVENT", {"kind": 1}])) + + assert manager.published == [] + assert ws.sent == [] + + +@pytest.mark.asyncio +async def test_any_accepting_relay_yields_ok_true(monkeypatch): + relays = {"wss://a": FakeRelay(True), "wss://b": FakeRelay(True)} + router, ws, manager = _router(monkeypatch, relays) + + await router._handle_client_to_nostr(EVENT_MSG) + assert manager.published == [EVENT_MSG] + assert EVENT_ID in router.pending_publishes + + # nothing answered yet: no OK + await router._handle_command_results() + assert ws.sent == [] + + _ok_from("wss://a", False, "blocked: kind not allowed") + await router._handle_command_results() + assert ws.sent == [] # one rejection out of two relays: keep waiting + + _ok_from("wss://b", True, "") + await router._handle_command_results() + assert ws.sent == [["OK", EVENT_ID, True, ""]] + assert router.pending_publishes == {} + assert EVENT_ID not in NostrRouter.received_command_results + + # a late OK must not produce a second reply + _ok_from("wss://a", True, "") + await router._handle_command_results() + assert len(ws.sent) == 1 + + +@pytest.mark.asyncio +async def test_all_relays_rejecting_yields_ok_false(monkeypatch): + relays = {"wss://a": FakeRelay(True), "wss://b": FakeRelay(True)} + router, ws, _ = _router(monkeypatch, relays) + + await router._handle_client_to_nostr(EVENT_MSG) + _ok_from("wss://a", False, "invalid: bad sig") + _ok_from("wss://b", False, "blocked: kind not allowed") + await router._handle_command_results() + + assert ws.sent == [["OK", EVENT_ID, False, "blocked: kind not allowed"]] + assert router.pending_publishes == {} + + +@pytest.mark.asyncio +async def test_timeout_yields_ok_false(monkeypatch): + router, ws, _ = _router(monkeypatch, {"wss://a": FakeRelay(True)}) + + await router._handle_client_to_nostr(EVENT_MSG) + router.pending_publishes[EVENT_ID].sent_at = ( + time.time() - PUBLISH_TIMEOUT_SECONDS - 1 + ) + await router._handle_command_results() + + assert ws.sent == [["OK", EVENT_ID, False, "error: timed out waiting for relays"]] + assert router.pending_publishes == {} + + +@pytest.mark.asyncio +async def test_stop_drops_pending_publishes(monkeypatch): + router, _, _ = _router(monkeypatch, {"wss://a": FakeRelay(True)}) + + await router._handle_client_to_nostr(EVENT_MSG) + _ok_from("wss://a", True) + await router.stop() + + assert router.pending_publishes == {} + assert NostrRouter.received_command_results == {} From 98e5accdd0d150efd36ddb72c9f4949c76a0ddc9 Mon Sep 17 00:00:00 2001 From: Padreug Date: Sat, 12 Sep 2026 13:31:01 +0200 Subject: [PATCH 3/5] chore: bump version to 1.2.0-aio.2 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013Tbyw6FwjhEJg3gHfPHxWt --- config.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/config.json b/config.json index 1f58e7b..b9e4956 100644 --- a/config.json +++ b/config.json @@ -1,7 +1,7 @@ { "name": "Nostr Client", "short_description": "Nostr relay multiplexer", - "version": "1.1.0", + "version": "1.2.0-aio.2", "tile": "/nostrclient/static/images/nostr-bitcoin.png", "contributors": ["calle", "motorina0", "dni"], "min_lnbits_version": "1.4.0", From 7e7120fa8298fb758080abc3dab5da2f3b6ec4db Mon Sep 17 00:00:00 2001 From: Padreug Date: Sun, 13 Sep 2026 16:07:37 +0200 Subject: [PATCH 4/5] feat: answer REQ with CLOSED when no relay is configured Ported from the upstream PR (lnbits/nostrclient#74, idea from #68). Nothing will ever serve a subscription when nostrclient has no relay at all, so refuse it per NIP-01 instead of registering it silently. Keyed on "no relay configured" rather than "none connected": a configured relay that is merely reconnecting gets the cached subscription replayed, so refusing the REQ there would be a regression. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013Tbyw6FwjhEJg3gHfPHxWt --- README.md | 2 +- router.py | 18 ++++++++++++++++-- tests/test_router_ok.py | 26 ++++++++++++++++++++++++++ 3 files changed, 43 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index dee86a1..2006af6 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,7 @@ flowchart LR 3. **Fan-Out** - Subscription requests are sent to all configured relays 4. **Aggregation** - Events from all relays are collected and deduplicated 5. **Response** - Events are sent back to the client with the original subscription ID -6. **Publish Acknowledgement** - Every `EVENT` a client publishes gets exactly one `["OK", , , ]` reply (NIP-01): `true` as soon as any relay accepts it, `false` once every relay has rejected it, the wait times out (10 s), or no relay is connected +6. **Publish Acknowledgement** - Every `EVENT` a client publishes gets exactly one `["OK", , , ]` reply (NIP-01): `true` as soon as any relay accepts it, `false` once every relay has rejected it, the wait times out (10 s), or no relay is connected. A `REQ` while no relay is configured at all is answered with `CLOSED` ## Configuration diff --git a/router.py b/router.py index 53ce387..8446512 100644 --- a/router.py +++ b/router.py @@ -195,7 +195,7 @@ class NostrRouter: assert len(json_data), "Bad JSON array" if json_data[0] == "REQ": - self._handle_client_req(json_data) + await self._handle_client_req(json_data) return if json_data[0] == "CLOSE": @@ -223,15 +223,29 @@ class NostrRouter: self.pending_publishes[event_id] = PendingPublish(event_id, len(connected)) nostr_client.relay_manager.publish_message(json_str) - def _handle_client_req(self, json_data): + async def _handle_client_req(self, json_data): subscription_id = json_data[1] logger.info(f"New subscription: '{subscription_id}'") + if not nostr_client.relay_manager.relays: + # NIP-01: nothing will ever serve this subscription, so refuse it. + # Only when no relay is configured at all: a configured relay that + # is merely reconnecting gets the cached subscription replayed. + await self._send_closed(subscription_id, "error: no relay connections") + return subscription_id_rewritten = urlsafe_short_hash() self.original_subscription_ids[subscription_id_rewritten] = subscription_id filters = json_data[2:] nostr_client.relay_manager.add_subscription(subscription_id_rewritten, filters) + async def _send_closed(self, subscription_id: str, message: str): + try: + await self.websocket.send_text( + json.dumps(["CLOSED", subscription_id, message]) + ) + except Exception as e: + logger.debug(f"Failed to send CLOSED for '{subscription_id}': {e}") + def _handle_client_close(self, subscription_id): subscription_id_rewritten = next( ( diff --git a/tests/test_router_ok.py b/tests/test_router_ok.py index 44863a1..e13301a 100644 --- a/tests/test_router_ok.py +++ b/tests/test_router_ok.py @@ -25,10 +25,14 @@ class FakeRelayManager: def __init__(self, relays: dict[str, FakeRelay]): self.relays = relays self.published: list[str] = [] + self.subscriptions: list[str] = [] def publish_message(self, message: str): self.published.append(message) + def add_subscription(self, subscription_id: str, filters): + self.subscriptions.append(subscription_id) + def close_subscriptions(self, subscriptions): pass @@ -139,3 +143,25 @@ async def test_stop_drops_pending_publishes(monkeypatch): assert router.pending_publishes == {} assert NostrRouter.received_command_results == {} + + +@pytest.mark.asyncio +async def test_req_without_configured_relays_is_closed(monkeypatch): + router, ws, manager = _router(monkeypatch, {}) + + await router._handle_client_to_nostr(json.dumps(["REQ", "sub1", {"kinds": [1]}])) + + assert ws.sent == [["CLOSED", "sub1", "error: no relay connections"]] + assert router.subscriptions == [] + assert manager.subscriptions == [] + + +@pytest.mark.asyncio +async def test_req_with_configured_but_disconnected_relay_is_kept(monkeypatch): + router, ws, manager = _router(monkeypatch, {"wss://a": FakeRelay(False)}) + + await router._handle_client_to_nostr(json.dumps(["REQ", "sub1", {"kinds": [1]}])) + + assert ws.sent == [] + assert len(router.subscriptions) == 1 + assert len(manager.subscriptions) == 1 From c4c1a1c342ad560664759273cd60d5f65c603fd7 Mon Sep 17 00:00:00 2001 From: Padreug Date: Sun, 13 Sep 2026 16:07:37 +0200 Subject: [PATCH 5/5] chore: bump version to 1.2.0-aio.3 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013Tbyw6FwjhEJg3gHfPHxWt --- config.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/config.json b/config.json index b9e4956..048a5a7 100644 --- a/config.json +++ b/config.json @@ -1,7 +1,7 @@ { "name": "Nostr Client", "short_description": "Nostr relay multiplexer", - "version": "1.2.0-aio.2", + "version": "1.2.0-aio.3", "tile": "/nostrclient/static/images/nostr-bitcoin.png", "contributors": ["calle", "motorina0", "dni"], "min_lnbits_version": "1.4.0",