Merge pull request 'fix(transport): harden RelayPool — connect timeout, stop-race, cross-relay dedup (review RP-1/2/4)' (#50) from fix/relay-pool-hardening into dev

Reviewed-on: #50

src/daemon/lib/relay-pool.ts

@@ -56,6 +56,8 @@ interface ActiveSub {
 
 const RECONNECT_BASE_MS = 1_000;
 const RECONNECT_CAP_MS = 10_000; // match #20's cap — cheap to retry a LAN relay
+const CONNECT_TIMEOUT_MS = 5_000; // bound a black-holed TCP connect (review RP-4)
+const SEEN_EVENT_CAP = 4_000; // bounded cross-relay event-id dedup set (review RP-1)
 
 /**
  * A single relay connection that owns its (re)connect loop. On every successful
@@ -78,6 +80,9 @@ class ManagedRelay {
         public readonly url: string,
         private readonly registry: Map<string, PoolSubscription>,
         private readonly log: (...args: any[]) => void,
+        /** Pool-wide first-seen gate: true the first time an event id is seen
+         *  across ALL relays, false on a duplicate. (review RP-1) */
+        private readonly markSeen: (id: string) => boolean,
     ) {}
 
     start(): void {
@@ -122,7 +127,12 @@ class ManagedRelay {
         this.active.get(id)?.close();
         try {
             const sub = this.relay.subscribe(s.filters, {
-                onevent: (e: Event) => s.onevent(e),
+                // Dedup across relays (and across reconnect re-deliveries): a
+                // kind:24133 request published to N relays must drive the daemon
+                // handler / recordSigning ONCE, not N times (review RP-1 / CS-4).
+                onevent: (e: Event) => {
+                    if (this.markSeen(e.id)) s.onevent(e);
+                },
                 oneose: () => s.oneose?.(),
             });
             this.active.set(id, sub);
@@ -179,16 +189,39 @@ class ManagedRelay {
     private connectOnce(): Promise<boolean> {
         return new Promise<boolean>((resolve) => {
             void (async () => {
+                // enableReconnect:false — WE own reconnect, not nostr-tools.
                 let relay: Relay;
                 try {
-                    // enableReconnect:false — WE own reconnect, not nostr-tools.
+                    relay = new Relay(this.url, { enableReconnect: false });
-                    relay = await Relay.connect(this.url, { enableReconnect: false });
+                } catch (e: any) {
+                    this.log("relay construct failed:", e?.message ?? e);
+                    resolve(false);
+                    return;
+                }
+                try {
+                    // Use the instance .connect({timeout}) rather than the static
+                    // Relay.connect(), which silently DROPS a timeout option in
+                    // nostr-tools 2.20.0. Without it a black-holed TCP connect
+                    // (SYN accepted, never upgraded) stalls this loop for the OS
+                    // socket timeout (minutes) with no retry (review RP-4).
+                    await relay.connect({ timeout: CONNECT_TIMEOUT_MS });
                 } catch (e: any) {
                     this.log("connect failed:", e?.message ?? e);
+                    try { relay.close(); } catch { /* ignore */ }
                     resolve(false);
                     return;
                 }
 
+                // If stop() ran while the connect was in flight, drop this socket
+                // cleanly — otherwise we'd re-arm subscriptions on a relay we mean
+                // to abandon, leak the socket, and hang connectLoop (its promise
+                // never resolves because onclose never fires). (review RP-2)
+                if (this.stopped) {
+                    try { relay.close(); } catch { /* ignore */ }
+                    resolve(true);
+                    return;
+                }
+
                 this.relay = relay;
                 this.connected = true;
                 this.lastConnectedAt = Date.now();
@@ -265,6 +298,9 @@ export class RelayPool {
     private counter = 0;
     private heartbeatTimer: ReturnType<typeof setInterval> | undefined;
     private lastHeartbeat = 0;
+    /** Insertion-ordered (≈LRU) bounded set of event ids already delivered to a
+     *  subscription callback — the cross-relay/replay dedup gate (review RP-1). */
+    private readonly seen: Set<string> = new Set();
 
     constructor(
         public readonly relayUrls: string[],
@@ -273,12 +309,27 @@ export class RelayPool {
         this.log = opts.log ?? (() => {});
         this.relays = relayUrls.map(
             (url) =>
-                new ManagedRelay(url, this.registry, (...a: any[]) =>
+                new ManagedRelay(
-                    this.log(`[relay:${url}]`, ...a),
+                    url,
+                    this.registry,
+                    (...a: any[]) => this.log(`[relay:${url}]`, ...a),
+                    (id) => this.markSeen(id),
                 ),
         );
     }
 
+    /** Returns true the first time `id` is seen pool-wide, false on a duplicate;
+     *  evicts the oldest id past the cap so this never grows unbounded. */
+    private markSeen(id: string): boolean {
+        if (this.seen.has(id)) return false;
+        this.seen.add(id);
+        if (this.seen.size > SEEN_EVENT_CAP) {
+            const oldest = this.seen.values().next().value;
+            if (oldest !== undefined) this.seen.delete(oldest);
+        }
+        return true;
+    }
+
     /** Start every relay's connect loop + (optionally) the sleep/wake heartbeat. */
     start(): void {
         for (const r of this.relays) r.start();

tests/relay-pool.test.ts

@@ -102,3 +102,35 @@ test("RelayPool.healthy() is false until the registry is subscribed on the wire
     pool.stop();
     await relay.stop();
 });
+
+/**
+ * RP-1: a NIP-46 request published to multiple relays (or re-delivered after a
+ * reconnect) must drive the subscription callback ONCE — otherwise the daemon
+ * signs N times and over-counts rate caps. The pool dedups by event id.
+ */
+test("RelayPool delivers each event id at most once (#RP-1)", async () => {
+    const relay = new MockRelay();
+    await relay.start();
+
+    const received: string[] = [];
+    const pool = new RelayPool([relay.url], { log: () => {} });
+    pool.start();
+    await pool.subscribeAwaitingEose([{ kinds: [24133], "#p": [PUBKEY] }], (e) =>
+        received.push(e.id),
+    );
+
+    const ev = makeEvent();
+    relay.inject(ev);
+    relay.inject(ev); // same id again (simulates a second relay / a replay)
+    await waitFor(() => received.includes(ev.id));
+    await new Promise((r) => setTimeout(r, 200)); // give a 2nd delivery a chance
+
+    assert.equal(
+        received.filter((id) => id === ev.id).length,
+        1,
+        "a duplicate event id is delivered to the callback only once",
+    );
+
+    pool.stop();
+    await relay.stop();
+});