Netcode: publishAndWait ack tracking + offline-queue dedupe + CI

.github/workflows/ci.yml

@@ -0,0 +1,30 @@
+name: CI
+
+on:
+  push:
+    branches: ['**']
+  pull_request:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: '20'
+
+      - name: Parse-check src
+        run: |
+          for f in src/*.js; do node --check "$f" || exit 1; done
+          echo "src parse OK"
+
+      - name: Install test deps
+        run: npm install ws nostr-tools --no-save
+
+      - name: Integration witness (test.js)
+        run: node test.js

AGENTS.md

@@ -43,6 +43,33 @@ caps at `PENDING_MAX` (500, drop-oldest), and drops entries older than
 `PENDING_TTL_MS` (120s) on drain — never let it grow unbounded or replay stale
 events. Covered by `testRelayReconnectCancel` + `testRelayPendingCapTtl` (fake-WS).
 
+## RelayPool publish-ack and pending dedupe
+
+`publish(event)` is still fire-and-forget (returns a `sent` boolean). For
+delivery confidence use `publishAndWait(event, { timeoutMs = 8000 })`: it sends
+then resolves `true` on the first relay `OK <id> true`, `false` on a relay
+reject (`OK <id> false`), and `false` on timeout. It keys pending ack records
+by `event.id` in `this._acks` (a `{resolve, timer}` Map); `_settleAck` resolves
+and `disconnect()` flushes every outstanding ack to `false` so no promise hangs.
+The `_handle` OK branch now emits both an `ok` and a `reject` event (previously
+only `reject`) and settles the ack. An event with no `id` cannot be acked, so
+`publishAndWait` falls back to resolving the `sent` boolean.
+
+The offline queue is deduped by `event.id` via `this._pendingIds` (a Set kept in
+lockstep with `this.pending`): `_queuePending` skips an id already queued, the
+cap-drop path removes the dropped id from the Set, and `_drainPending` clears the
+Set before re-publishing. This prevents a reconnect drain from double-publishing
+the same event. Covered by `testRelayPendingDedupe` + `testRelayPublishAck`
+(fake-WS, offline).
+
+## CI
+
+`.github/workflows/ci.yml` runs `node --check src/*.js` then installs `ws` +
+`nostr-tools` (`--no-save`) and runs `node test.js` on every push/PR. The
+real-relay phases tolerate single-relay flake via the multi-relay `RELAYS`
+array; `compose`/`data` tests skip when `xstate` is absent (not installed in
+CI) — that is expected, not a failure.
+
 ## test.js size cap
 
 The single integration witness (`test.js`) grows as coverage expands. The previous <=200 line cap is superseded: the file may grow freely as long as it remains a single file at repo root, mock-free for network tests, and real-services only for the relay round-trip. Current size: ~620 lines (26 tests). Do not split into a `test/` directory.

src/relay-pool.js

@@ -38,8 +38,10 @@ export class RelayPool extends EventTarget {
     this.relays = new Map();
     this.subs = new Map();
     this.pending = [];
+    this._pendingIds = new Set();
     this.seen = new Map();
     this._reconnectTimers = new Map();
+    this._acks = new Map();
     this._closed = false;
     this.verifyEvent = verifyEvent;
     this.WS = WebSocketImpl || (typeof WebSocket !== 'undefined' ? WebSocket : null);
@@ -55,6 +57,8 @@ export class RelayPool extends EventTarget {
     this._closed = true;
     for (const [, t] of this._reconnectTimers) clearTimeout(t);
     this._reconnectTimers.clear();
+    for (const [, rec] of this._acks) { clearTimeout(rec.timer); rec.resolve(false); }
+    this._acks.clear();
     for (const [, r] of this.relays) {
       if (r.ws) {
         r.ws.onclose = null; r.ws.onerror = null; r.ws.onopen = null; r.ws.onmessage = null;
@@ -132,8 +136,12 @@ export class RelayPool extends EventTarget {
       this._emit('eose', { subId });
     } else if (type === 'NOTICE') {
       this._emit('notice', { url, message: msg[1] });
-    } else if (type === 'OK' && !msg[2]) {
-      this._emit('reject', { id: msg[1], reason: msg[3] || '' });
+    } else if (type === 'OK') {
+      const accepted = msg[2] === true;
+      const id = msg[1], reason = msg[3] || '';
+      if (accepted) this._emit('ok', { url, id });
+      else this._emit('reject', { url, id, reason });
+      this._settleAck(id, accepted, reason);
     }
   }
 
@@ -169,16 +177,51 @@ export class RelayPool extends EventTarget {
         sent = true;
       }
     }
-    if (!sent) {
-      this.pending.push({ event, ts: Date.now() });
-      if (this.pending.length > PENDING_MAX) this.pending.splice(0, this.pending.length - PENDING_MAX);
+    if (sent) {
+      if (event?.id) this._pendingIds.delete(event.id);
+    } else {
+      this._queuePending(event);
     }
     return sent;
   }
 
+  _queuePending(event) {
+    if (event?.id && this._pendingIds.has(event.id)) return;
+    if (event?.id) this._pendingIds.add(event.id);
+    this.pending.push({ event, ts: Date.now() });
+    while (this.pending.length > PENDING_MAX) {
+      const dropped = this.pending.shift();
+      if (dropped.event?.id) this._pendingIds.delete(dropped.event.id);
+    }
+  }
+
+  // Resolves true once any relay sends OK accepted, false on relay reject,
+  // or false on timeout. Gives callers delivery confidence beyond fire-and-forget.
+  publishAndWait(event, { timeoutMs = 8000 } = {}) {
+    const sent = this.publish(event);
+    if (!event?.id) return Promise.resolve(sent);
+    return new Promise((resolve) => {
+      const prior = this._acks.get(event.id);
+      if (prior) clearTimeout(prior.timer);
+      const settle = (ok) => {
+        const rec = this._acks.get(event.id);
+        if (rec) { clearTimeout(rec.timer); this._acks.delete(event.id); }
+        resolve(ok);
+      };
+      const timer = setTimeout(() => settle(false), timeoutMs);
+      this._acks.set(event.id, { resolve: settle, timer });
+    });
+  }
+
+  _settleAck(id, accepted) {
+    const rec = this._acks.get(id);
+    if (rec) rec.resolve(accepted);
+  }
+
   _drainPending() {
     const cutoff = Date.now() - PENDING_TTL_MS;
     const pending = this.pending.splice(0);
+    this._pendingIds.clear();
     for (const p of pending) { if (p.ts >= cutoff) this.publish(p.event); }
   }
 

test.js

@@ -588,6 +588,39 @@ async function testRelayPendingCapTtl() {
   console.log('  relay pending cap/TTL: pass');
 }
 
+async function testRelayPendingDedupe() {
+  const WS = fakeWSImpl();
+  const pool = new RelayPool({ relays: ['wss://a'], WebSocketImpl: WS });
+  pool.connect();                          // CONNECTING -> publishes queue
+  pool.publish({ id: 'dup' });
+  pool.publish({ id: 'dup' });             // same id must not double-queue
+  pool.publish({ id: 'other' });
+  assert.strictEqual(pool.pending.length, 2, 'pending deduped by event.id');
+  assert.strictEqual(pool._pendingIds.size, 2, 'pendingIds tracks unique ids');
+  pool.disconnect();
+  console.log('  relay pending dedupe: pass');
+}
+
+async function testRelayPublishAck() {
+  const WS = fakeWSImpl();
+  const pool = new RelayPool({ relays: ['wss://a'], WebSocketImpl: WS });
+  pool.connect();
+  WS.created[0].open();                     // readyState 1 -> publish sends
+  const okP = pool.publishAndWait({ id: 'acc' }, { timeoutMs: 1000 });
+  WS.created[0].onmessage({ data: JSON.stringify(['OK', 'acc', true, '']) });
+  assert.strictEqual(await okP, true, 'publishAndWait resolves true on accepted OK');
+
+  const rejP = pool.publishAndWait({ id: 'rej' }, { timeoutMs: 1000 });
+  WS.created[0].onmessage({ data: JSON.stringify(['OK', 'rej', false, 'blocked']) });
+  assert.strictEqual(await rejP, false, 'publishAndWait resolves false on relay reject');
+
+  const toP = pool.publishAndWait({ id: 'tmo' }, { timeoutMs: 30 });
+  assert.strictEqual(await toP, false, 'publishAndWait resolves false on timeout');
+  assert.strictEqual(pool._acks.size, 0, 'ack records cleaned up after settle');
+  pool.disconnect();
+  console.log('  relay publish ack: pass');
+}
+
 async function main() {
   console.log('magicwand test.js');
   await testAuth();
@@ -615,6 +648,8 @@ async function main() {
   await testRelayDisconnectConnecting();
   await testRelayReconnectCancel();
   await testRelayPendingCapTtl();
+  await testRelayPendingDedupe();
+  await testRelayPublishAck();
   await testRelay();
   console.log('all pass');
 }