fix(transport): harden RelayPool — connect timeout, stop-race, cross-relay dedup
Some checks failed
Docker image / build-and-push-image (push) Has been cancelled
Some checks failed
Docker image / build-and-push-image (push) Has been cancelled
Folds three medium findings from the transport review into the RelayPool:
- RP-4: connectOnce used the static Relay.connect(), which silently DROPS a
{timeout} option in nostr-tools 2.20.0, so a black-holed TCP connect (SYN
accepted, never upgraded) stalled that relay's loop for the OS socket timeout
(minutes) with no retry. Now constructs the Relay and calls the instance
.connect({ timeout: 5000 }), which honours the timeout → prompt reject + backoff.
- RP-2: connectOnce didn't re-check `stopped` after the await. If stop() ran
while a connect was in flight, the resolved socket re-armed subscriptions on a
relay we meant to abandon, leaked the socket, and hung connectLoop (its promise
never resolved because onclose never fired). Now drops the socket cleanly and
resolves if stopped mid-connect.
- RP-1: no cross-relay event dedup — a kind:24133 request published to N relays
(the normal NIP-46 pattern), or re-delivered after a reconnect, drove the
daemon handler + recordSigning N times, making rate caps bind ~N× tighter
(fails closed, not open). Added a bounded (4000-id, ≈LRU) pool-wide seen-set;
onevent fires at most once per event id. Closes the CS-4 replay vector too.
Test: tests/relay-pool.test.ts asserts a duplicate event id is delivered once.
relay 3 / nip46 1 / admin 2 green; daemon bundles; tsc at baseline.
Refs: transport review RP-1/RP-2/RP-4, CS-4; #42
This commit is contained in:
parent
1754027249
commit
1c16a2a4b7
2 changed files with 88 additions and 5 deletions
|
|
@ -56,6 +56,8 @@ interface ActiveSub {
|
||||||
|
|
||||||
const RECONNECT_BASE_MS = 1_000;
|
const RECONNECT_BASE_MS = 1_000;
|
||||||
const RECONNECT_CAP_MS = 10_000; // match #20's cap — cheap to retry a LAN relay
|
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
|
* A single relay connection that owns its (re)connect loop. On every successful
|
||||||
|
|
@ -78,6 +80,9 @@ class ManagedRelay {
|
||||||
public readonly url: string,
|
public readonly url: string,
|
||||||
private readonly registry: Map<string, PoolSubscription>,
|
private readonly registry: Map<string, PoolSubscription>,
|
||||||
private readonly log: (...args: any[]) => void,
|
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 {
|
start(): void {
|
||||||
|
|
@ -122,7 +127,12 @@ class ManagedRelay {
|
||||||
this.active.get(id)?.close();
|
this.active.get(id)?.close();
|
||||||
try {
|
try {
|
||||||
const sub = this.relay.subscribe(s.filters, {
|
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?.(),
|
oneose: () => s.oneose?.(),
|
||||||
});
|
});
|
||||||
this.active.set(id, sub);
|
this.active.set(id, sub);
|
||||||
|
|
@ -179,16 +189,39 @@ class ManagedRelay {
|
||||||
private connectOnce(): Promise<boolean> {
|
private connectOnce(): Promise<boolean> {
|
||||||
return new Promise<boolean>((resolve) => {
|
return new Promise<boolean>((resolve) => {
|
||||||
void (async () => {
|
void (async () => {
|
||||||
|
// enableReconnect:false — WE own reconnect, not nostr-tools.
|
||||||
let relay: Relay;
|
let relay: Relay;
|
||||||
try {
|
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) {
|
} catch (e: any) {
|
||||||
this.log("connect failed:", e?.message ?? e);
|
this.log("connect failed:", e?.message ?? e);
|
||||||
|
try { relay.close(); } catch { /* ignore */ }
|
||||||
resolve(false);
|
resolve(false);
|
||||||
return;
|
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.relay = relay;
|
||||||
this.connected = true;
|
this.connected = true;
|
||||||
this.lastConnectedAt = Date.now();
|
this.lastConnectedAt = Date.now();
|
||||||
|
|
@ -265,6 +298,9 @@ export class RelayPool {
|
||||||
private counter = 0;
|
private counter = 0;
|
||||||
private heartbeatTimer: ReturnType<typeof setInterval> | undefined;
|
private heartbeatTimer: ReturnType<typeof setInterval> | undefined;
|
||||||
private lastHeartbeat = 0;
|
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(
|
constructor(
|
||||||
public readonly relayUrls: string[],
|
public readonly relayUrls: string[],
|
||||||
|
|
@ -273,12 +309,27 @@ export class RelayPool {
|
||||||
this.log = opts.log ?? (() => {});
|
this.log = opts.log ?? (() => {});
|
||||||
this.relays = relayUrls.map(
|
this.relays = relayUrls.map(
|
||||||
(url) =>
|
(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 every relay's connect loop + (optionally) the sleep/wake heartbeat. */
|
||||||
start(): void {
|
start(): void {
|
||||||
for (const r of this.relays) r.start();
|
for (const r of this.relays) r.start();
|
||||||
|
|
|
||||||
|
|
@ -102,3 +102,35 @@ test("RelayPool.healthy() is false until the registry is subscribed on the wire
|
||||||
pool.stop();
|
pool.stop();
|
||||||
await relay.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();
|
||||||
|
});
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue