Compare commits

..

1 commit

Author SHA1 Message Date
115e869225 fix: queue outgoing events when relay connection is down
Some checks failed
ci.yml / fix: queue outgoing events when relay connection is down (pull_request) Failing after 0s
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) <noreply@anthropic.com>
2026-04-27 20:09:26 +02:00
10 changed files with 135 additions and 322 deletions

59
CLAUDE.md Normal file
View file

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

View file

@ -50,7 +50,6 @@ flowchart LR
3. **Fan-Out** - Subscription requests are sent to all configured relays 3. **Fan-Out** - Subscription requests are sent to all configured relays
4. **Aggregation** - Events from all relays are collected and deduplicated 4. **Aggregation** - Events from all relays are collected and deduplicated
5. **Response** - Events are sent back to the client with the original subscription ID 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", <id>, <accepted>, <message>]` 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 ## Configuration

View file

@ -1,7 +1,7 @@
{ {
"name": "Nostr Client", "name": "Nostr Client",
"short_description": "Nostr relay multiplexer", "short_description": "Nostr relay multiplexer",
"version": "1.2.0-aio.3", "version": "1.1.0",
"tile": "/nostrclient/static/images/nostr-bitcoin.png", "tile": "/nostrclient/static/images/nostr-bitcoin.png",
"contributors": ["calle", "motorina0", "dni"], "contributors": ["calle", "motorina0", "dni"],
"min_lnbits_version": "1.4.0", "min_lnbits_version": "1.4.0",

View file

@ -39,13 +39,11 @@ class NostrClient:
callback_events_func=None, callback_events_func=None,
callback_notices_func=None, callback_notices_func=None,
callback_eosenotices_func=None, callback_eosenotices_func=None,
callback_command_results_func=None,
): ):
while self.running: while self.running:
self._check_events(callback_events_func) self._check_events(callback_events_func)
self._check_notices(callback_notices_func) self._check_notices(callback_notices_func)
self._check_eos_notices(callback_eosenotices_func) self._check_eos_notices(callback_eosenotices_func)
self._check_command_results(callback_command_results_func)
await asyncio.sleep(0.2) await asyncio.sleep(0.2)
@ -75,12 +73,3 @@ class NostrClient:
callback_eosenotices_func(event_msg) callback_eosenotices_func(event_msg)
except Exception as e: except Exception as e:
logger.debug(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)

View file

@ -27,22 +27,11 @@ class EndOfStoredEventsMessage:
self.url = url self.url = url
class CommandResultMessage:
"""An `["OK", <event_id>, <accepted>, <message>]` 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: class MessagePool:
def __init__(self) -> None: def __init__(self) -> None:
self.events: Queue[EventMessage] = Queue() self.events: Queue[EventMessage] = Queue()
self.notices: Queue[NoticeMessage] = Queue() self.notices: Queue[NoticeMessage] = Queue()
self.eose_notices: Queue[EndOfStoredEventsMessage] = Queue() self.eose_notices: Queue[EndOfStoredEventsMessage] = Queue()
self.command_results: Queue[CommandResultMessage] = Queue()
self._unique_events: set = set() self._unique_events: set = set()
self.lock: Lock = Lock() self.lock: Lock = Lock()
@ -58,9 +47,6 @@ class MessagePool:
def get_eose_notice(self): def get_eose_notice(self):
return self.eose_notices.get() return self.eose_notices.get()
def get_command_result(self):
return self.command_results.get()
def has_events(self): def has_events(self):
return self.events.qsize() > 0 return self.events.qsize() > 0
@ -70,9 +56,6 @@ class MessagePool:
def has_eose_notices(self): def has_eose_notices(self):
return self.eose_notices.qsize() > 0 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): def _process_message(self, message: str, url: str):
message_json = json.loads(message) message_json = json.loads(message)
message_type = message_json[0] message_type = message_json[0]
@ -92,13 +75,6 @@ class MessagePool:
self.notices.put(NoticeMessage(message_json[1], url)) self.notices.put(NoticeMessage(message_json[1], url))
elif message_type == RelayMessageType.END_OF_STORED_EVENTS: elif message_type == RelayMessageType.END_OF_STORED_EVENTS:
self.eose_notices.put(EndOfStoredEventsMessage(message_json[1], url)) 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): def _accept_event(self, event_message: EventMessage):
""" """

View file

@ -1,6 +1,9 @@
from __future__ import annotations
import asyncio import asyncio
import json import json
import time import time
from collections.abc import Callable
from queue import Queue from queue import Queue
from loguru import logger from loguru import logger
@ -28,6 +31,7 @@ class Relay:
self.num_subscriptions: int = 0 self.num_subscriptions: int = 0
self.queue: Queue = Queue() self.queue: Queue = Queue()
self.on_connect: Callable[[Relay], None] | None = None
def connect(self): def connect(self):
self.ws = WebSocketApp( self.ws = WebSocketApp(
@ -97,6 +101,11 @@ class Relay:
logger.info(f"[Relay: {self.url}] Connected.") logger.info(f"[Relay: {self.url}] Connected.")
self.connected = True self.connected = True
self.shutdown = False 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): def _on_close(self, _, status_code, message):
logger.warning( logger.warning(

View file

@ -1,6 +1,8 @@
import asyncio import asyncio
import json
import threading import threading
import time import time
from collections import deque
from typing import List from typing import List
from loguru import logger from loguru import logger
@ -9,6 +11,8 @@ from .message_pool import MessagePool, NoticeMessage
from .relay import Relay from .relay import Relay
from .subscription import Subscription from .subscription import Subscription
PENDING_EVENTS_MAX = 100
class RelayManager: class RelayManager:
def __init__(self) -> None: def __init__(self) -> None:
@ -18,6 +22,8 @@ class RelayManager:
self.message_pool = MessagePool() self.message_pool = MessagePool()
self._cached_subscriptions: dict[str, Subscription] = {} self._cached_subscriptions: dict[str, Subscription] = {}
self._subscriptions_lock = threading.Lock() 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: def add_relay(self, url: str) -> Relay:
if url in list(self.relays.keys()): if url in list(self.relays.keys()):
@ -25,6 +31,7 @@ class RelayManager:
return self.relays[url] return self.relays[url]
relay = Relay(url, self.message_pool) relay = Relay(url, self.message_pool)
relay.on_connect = self._on_relay_connect
self.relays[url] = relay self.relays[url] = relay
self._open_connection(relay) self._open_connection(relay)
@ -100,8 +107,19 @@ class RelayManager:
relay.close() relay.close()
def publish_message(self, message: str): def publish_message(self, message: str):
for relay in self.relays.values(): connected_relays = [
relay.publish(message) 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): def handle_notice(self, notice: NoticeMessage):
relay = next((r for r in self.relays.values() if r.url == notice.url)) 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}'") logger.info(f"Restarting connection to relay '{relay.url}'")
self._drain_relay_queue(relay)
self.remove_relay(relay.url) self.remove_relay(relay.url)
new_relay = self.add_relay(relay.url) new_relay = self.add_relay(relay.url)
new_relay.error_counter = relay.error_counter new_relay.error_counter = relay.error_counter
new_relay.error_list = relay.error_list 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)."
)

101
router.py
View file

@ -1,6 +1,5 @@
import asyncio import asyncio
import json import json
import time
from typing import ClassVar from typing import ClassVar
from fastapi import WebSocket, WebSocketDisconnect from fastapi import WebSocket, WebSocketDisconnect
@ -10,30 +9,11 @@ from loguru import logger
from .nostr.client.client import NostrClient from .nostr.client.client import NostrClient
# from . import nostr_client # from . import nostr_client
from .nostr.message_pool import ( from .nostr.message_pool import EndOfStoredEventsMessage, EventMessage, NoticeMessage
CommandResultMessage,
EndOfStoredEventsMessage,
EventMessage,
NoticeMessage,
)
nostr_client: NostrClient = NostrClient() nostr_client: NostrClient = NostrClient()
all_routers: list["NostrRouter"] = [] 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: class NostrRouter:
received_subscription_events: ClassVar[dict[str, list[EventMessage]]] = {} received_subscription_events: ClassVar[dict[str, list[EventMessage]]] = {}
@ -41,14 +21,12 @@ class NostrRouter:
received_subscription_eosenotices: ClassVar[dict[str, EndOfStoredEventsMessage]] = ( received_subscription_eosenotices: ClassVar[dict[str, EndOfStoredEventsMessage]] = (
{} {}
) )
received_command_results: ClassVar[dict[str, list[CommandResultMessage]]] = {}
def __init__(self, websocket: WebSocket): def __init__(self, websocket: WebSocket):
self.connected: bool = True self.connected: bool = True
self.websocket: WebSocket = websocket self.websocket: WebSocket = websocket
self.tasks: list[asyncio.Task] = [] self.tasks: list[asyncio.Task] = []
self.original_subscription_ids: dict[str, str] = {} self.original_subscription_ids: dict[str, str] = {}
self.pending_publishes: dict[str, PendingPublish] = {}
@property @property
def subscriptions(self) -> list[str]: def subscriptions(self) -> list[str]:
@ -62,9 +40,6 @@ class NostrRouter:
async def stop(self): async def stop(self):
nostr_client.relay_manager.close_subscriptions(self.subscriptions) nostr_client.relay_manager.close_subscriptions(self.subscriptions)
self.connected = False 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: for t in self.tasks:
try: try:
@ -99,7 +74,6 @@ class NostrRouter:
while self.connected: while self.connected:
try: try:
await self._handle_subscriptions() await self._handle_subscriptions()
await self._handle_command_results()
self._handle_notices() self._handle_notices()
except Exception as e: except Exception as e:
logger.debug(f"Failed to handle response for client: '{e!s}'.") logger.debug(f"Failed to handle response for client: '{e!s}'.")
@ -146,42 +120,6 @@ class NostrRouter:
f"[NOSTRCLIENT] Error in _handle_received_subscription_events: {e}" f"[NOSTRCLIENT] Error in _handle_received_subscription_events: {e}"
) )
async def _handle_command_results(self):
"""
Reply exactly one `["OK", <id>, <accepted>, <message>]` 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): def _handle_notices(self):
while len(NostrRouter.received_subscription_notices): while len(NostrRouter.received_subscription_notices):
my_event = NostrRouter.received_subscription_notices.pop(0) my_event = NostrRouter.received_subscription_notices.pop(0)
@ -195,7 +133,7 @@ class NostrRouter:
assert len(json_data), "Bad JSON array" assert len(json_data), "Bad JSON array"
if json_data[0] == "REQ": if json_data[0] == "REQ":
await self._handle_client_req(json_data) self._handle_client_req(json_data)
return return
if json_data[0] == "CLOSE": if json_data[0] == "CLOSE":
@ -203,49 +141,18 @@ class NostrRouter:
return return
if json_data[0] == "EVENT": if json_data[0] == "EVENT":
await self._handle_client_event(json_data, json_str) nostr_client.relay_manager.publish_message(json_str)
return return
async def _handle_client_event(self, json_data, json_str): def _handle_client_req(self, json_data):
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)
async def _handle_client_req(self, json_data):
subscription_id = json_data[1] subscription_id = json_data[1]
logger.info(f"New subscription: '{subscription_id}'") 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() subscription_id_rewritten = urlsafe_short_hash()
self.original_subscription_ids[subscription_id_rewritten] = subscription_id self.original_subscription_ids[subscription_id_rewritten] = subscription_id
filters = json_data[2:] filters = json_data[2:]
nostr_client.relay_manager.add_subscription(subscription_id_rewritten, filters) 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): def _handle_client_close(self, subscription_id):
subscription_id_rewritten = next( subscription_id_rewritten = next(
( (

View file

@ -4,13 +4,8 @@ import threading
from loguru import logger from loguru import logger
from .crud import get_relays from .crud import get_relays
from .nostr.message_pool import ( from .nostr.message_pool import EndOfStoredEventsMessage, EventMessage, NoticeMessage
CommandResultMessage, from .router import NostrRouter, nostr_client
EndOfStoredEventsMessage,
EventMessage,
NoticeMessage,
)
from .router import NostrRouter, all_routers, nostr_client
async def init_relays(): async def init_relays():
@ -60,24 +55,12 @@ async def subscribe_events():
NostrRouter.received_subscription_eosenotices[sub_id] = event_message 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(): def wrap_async_subscribe():
asyncio.run( asyncio.run(
nostr_client.subscribe( nostr_client.subscribe(
callback_events, callback_events,
callback_notices, callback_notices,
callback_eose_notices, callback_eose_notices,
callback_command_results,
) )
) )

View file

@ -1,167 +0,0 @@
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] = []
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
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 == {}
@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