fix(relay): bound reconnect replay with resumed since (#105)

.changeset/bounded-reconnect-replay.md

@@ -0,0 +1,5 @@
+---
+'@contextvm/sdk': patch
+---
+
+Bound relay-reconnect replay (#105). Restored subscriptions now advance their `since` to the newest `created_at` seen (timestamps ahead of the local clock are ignored) minus a 30-minute guard (`RESUBSCRIBE_SINCE_GUARD_SECONDS`), so a relay reconnect re-requests only that window instead of the whole session; filters without `since` are unchanged. Previously, whole-session replay re-dispatched every stored request whose event id had aged out of the dedup cache, re-executing completed tool calls without any server restart. The transports' `seenEventIds` cache now holds 40,000 entries (`DEFAULT_DEDUP_CACHE_SIZE`, ~20,000 encrypted requests) to cover the replay window. This also corrects the earlier claim that transport dedup fully protects against re-delivery on relay reconnects: suppression stays best-effort and bounded, requests whose `created_at` trails the newest event seen by more than the guard may be dropped on reconnect (slow sender clock or lagging relay), and same-event transparent retries (CEP-8) older than the guard may not be re-observed after a reconnect.

src/core/constants.ts

@@ -128,6 +128,14 @@ export const NOSTR_TAGS = {
 
 export const DEFAULT_LRU_SIZE = 5000;
 
+/**
+ * Capacity of the transports' `seenEventIds` dedup caches. Sized to cover the
+ * 30-minute reconnect replay window (see `RESUBSCRIBE_SINCE_GUARD_SECONDS`):
+ * each encrypted request consumes two entries (gift-wrap envelope + decrypted
+ * inner event), so 40,000 entries hold ~20,000 encrypted requests.
+ */
+export const DEFAULT_DEDUP_CACHE_SIZE = 40_000;
+
 /**
  * Default timeout for network/relay operations (30 seconds).
  */

src/relay/applesauce-relay-pool.test.ts

@@ -10,7 +10,11 @@ import {
   type UnsignedEvent,
 } from 'nostr-tools';
 import { PrivateKeySigner } from '../signer/private-key-signer.js';
-import { ApplesauceRelayPool, PING_FILTER } from './applesauce-relay-pool.js';
+import {
+  ApplesauceRelayPool,
+  PING_FILTER,
+  RESUBSCRIBE_SINCE_GUARD_SECONDS,
+} from './applesauce-relay-pool.js';
 import { Relay, RelayGroup } from 'applesauce-relay';
 import { DEFAULT_TIMEOUT_MS } from '../core/constants.js';
 import {
@@ -712,31 +716,31 @@ describe('ApplesauceRelayPool Integration', () => {
 
       // Temporarily disable applesauce's auto-recovery for this test
       const testPool = relayPool as unknown as {
-        createSubscription: (
-          filters: Filter[],
-          onEvent: (event: NostrEvent) => void,
-          onEose?: () => void,
-        ) => () => void;
+        createSubscription: (state: {
+          filters: Filter[];
+          onEvent: (event: NostrEvent) => void;
+          onEose?: () => void;
+        }) => () => void;
         relayGroup: RelayGroup;
       };
-      testPool.createSubscription = function (filters, onEvent, onEose) {
+      testPool.createSubscription = function (state) {
         // Mirror production shape: subscribe to the raw req() message stream so
         // the test override exercises the same dispatch path as the pool. No
         // dedup (see production createSubscription for rationale).
         const sub = testPool.relayGroup
-          .req(filters, {
+          .req(state.filters, {
             reconnect: false, // Disable applesauce recovery
             resubscribe: false, // Disable applesauce recovery
           })
           .subscribe({
             next: (message) => {
               if (message.type === 'EOSE') {
-                onEose?.();
+                state.onEose?.();
                 return;
               }
 
               if (message.type === 'EVENT') {
-                onEvent(message.event);
+                state.onEvent(message.event);
               }
             },
             error: () => {},
@@ -1282,3 +1286,296 @@ describe('ApplesauceRelayPool rebuild failure containment', () => {
     }
   }, 10_000);
 });
+
+describe('ApplesauceRelayPool reconnect resume', () => {
+  let stopRelay: (() => void) | undefined;
+  let relayUrl: string;
+
+  beforeAll(async () => {
+    const relay = await spawnMockRelayWithEnv({ DISABLE_MOCK_RESPONSES: 'true' });
+    stopRelay = relay.stop;
+    relayUrl = relay.relayUrl;
+  });
+
+  afterAll(() => {
+    stopRelay?.();
+  });
+
+  /** Signs a kind-1 event with a pinned created_at and a unique tag. */
+  async function signTaggedEvent(
+    signer: PrivateKeySigner,
+    pubkey: string,
+    tag: string,
+    createdAt: number,
+    content: string,
+  ): Promise<NostrEvent> {
+    return signer.signEvent({
+      kind: 1,
+      pubkey,
+      created_at: createdAt,
+      tags: [['t', tag]],
+      content,
+    });
+  }
+
+  /** Polls until the condition holds or the deadline passes. */
+  async function waitFor(
+    condition: () => boolean,
+    timeoutMs: number,
+  ): Promise<void> {
+    const deadline = Date.now() + timeoutMs;
+    while (!condition() && Date.now() < deadline) {
+      await sleep(10);
+    }
+  }
+
+  test.serial(
+    'restored subscriptions re-request only the guard window instead of the whole session',
+    async () => {
+      const now = Math.floor(Date.now() / 1000);
+      const privateKey = generateSecretKey();
+      const signer = new PrivateKeySigner(bytesToHex(privateKey));
+      const pubkey = getPublicKey(privateKey);
+      const tag = `resume-window-${Date.now()}`;
+      const oldEvent = await signTaggedEvent(
+        signer,
+        pubkey,
+        tag,
+        now - RESUBSCRIBE_SINCE_GUARD_SECONDS * 2,
+        'old',
+      );
+      const newEvent = await signTaggedEvent(signer, pubkey, tag, now, 'new');
+
+      const pool = new ApplesauceRelayPool([relayUrl]);
+      await pool.connect();
+      await pool.publish(oldEvent);
+      await pool.publish(newEvent);
+
+      const received: NostrEvent[] = [];
+      let eoseCount = 0;
+      await pool.subscribe(
+        [
+          {
+            kinds: [1],
+            '#t': [tag],
+            since: now - RESUBSCRIBE_SINCE_GUARD_SECONDS * 4,
+          },
+        ],
+        (event) => {
+          received.push(event);
+        },
+        () => {
+          eoseCount += 1;
+        },
+      );
+      await waitFor(() => received.length >= 2, 5_000);
+      expect(received.map((e) => e.id).sort()).toEqual(
+        [oldEvent.id, newEvent.id].sort(),
+      );
+
+      const testPool = pool as unknown as TestableApplesauceRelayPool;
+      testPool.rebuild('test-resume-window');
+      await testPool.rebuildInFlight;
+
+      // The restored subscription ends its replay with an EOSE of its own.
+      await waitFor(() => eoseCount >= 2, 5_000);
+      expect(eoseCount).toBeGreaterThanOrEqual(2);
+
+      // Only events inside the guard window are re-requested after the
+      // restore; the older event must not be replayed (and thus not
+      // re-dispatched).
+      expect(received.slice(2).map((e) => e.id)).toEqual([newEvent.id]);
+
+      await pool.disconnect();
+    },
+    15_000,
+  );
+
+  test.serial(
+    'automatic recovery after a connection drop re-requests only the guard window',
+    async () => {
+      const now = Math.floor(Date.now() / 1000);
+      const privateKey = generateSecretKey();
+      const signer = new PrivateKeySigner(bytesToHex(privateKey));
+      const pubkey = getPublicKey(privateKey);
+      const tag = `resume-auto-${Date.now()}`;
+      const oldEvent = await signTaggedEvent(
+        signer,
+        pubkey,
+        tag,
+        now - RESUBSCRIBE_SINCE_GUARD_SECONDS * 2,
+        'old',
+      );
+      const newEvent = await signTaggedEvent(signer, pubkey, tag, now, 'new');
+
+      const relay = await spawnMockRelayWithEnv({ DISABLE_MOCK_RESPONSES: 'true' });
+      try {
+        // Fast reconnect backoff; liveness pings disabled so only applesauce's
+        // automatic recovery can restore the subscription — no rebuild().
+        const pool = new ApplesauceRelayPool([relay.relayUrl], {
+          reconnectBaseDelayMs: 200,
+          reconnectMaxDelayMs: 500,
+          pingFrequencyMs: 300_000,
+        });
+        await pool.connect();
+        await pool.publish(oldEvent);
+        await pool.publish(newEvent);
+
+        const received: NostrEvent[] = [];
+        let eoseCount = 0;
+        await pool.subscribe(
+          [
+            {
+              kinds: [1],
+              '#t': [tag],
+              since: now - RESUBSCRIBE_SINCE_GUARD_SECONDS * 4,
+            },
+          ],
+          (event) => {
+            received.push(event);
+          },
+          () => {
+            eoseCount += 1;
+          },
+        );
+        await waitFor(() => received.length >= 2, 5_000);
+
+        // Unclean drop: applesauce's reconnect retry re-sends the REQ
+        // without any pool rebuild.
+        relay.relay.pause();
+        await sleep(300);
+        relay.relay.resume();
+
+        // The re-sent REQ replays stored events and ends with its own EOSE.
+        await waitFor(() => eoseCount >= 2, 10_000);
+        expect(eoseCount).toBeGreaterThanOrEqual(2);
+
+        // The recovered subscription must not re-request the old event...
+        expect(received.filter((e) => e.id === oldEvent.id).length).toBe(1);
+        // ...while the recent one is replayed (proves recovery actually ran).
+        expect(
+          received.filter((e) => e.id === newEvent.id).length,
+        ).toBeGreaterThanOrEqual(2);
+
+        await pool.disconnect();
+      } finally {
+        relay.stop();
+      }
+    },
+    20_000,
+  );
+
+  test.serial(
+    'a near-future envelope cannot push the restore point past unseen requests',
+    async () => {
+      const now = Math.floor(Date.now() / 1000);
+      const privateKey = generateSecretKey();
+      const signer = new PrivateKeySigner(bytesToHex(privateKey));
+      const pubkey = getPublicKey(privateKey);
+      const tag = `resume-clamp-${Date.now()}`;
+      // Just inside the future-time allowance the watermark used to accept:
+      // the transport rejects such envelopes, yet they must not move the
+      // restore point past requests from slower clocks.
+      const junkEvent = await signTaggedEvent(
+        signer,
+        pubkey,
+        tag,
+        now + RESUBSCRIBE_SINCE_GUARD_SECONDS - 10,
+        'junk',
+      );
+
+      const pool = new ApplesauceRelayPool([relayUrl]);
+      await pool.connect();
+      await pool.publish(junkEvent);
+
+      const received: NostrEvent[] = [];
+      let eoseCount = 0;
+      await pool.subscribe(
+        [
+          {
+            kinds: [1],
+            '#t': [tag],
+            since: now - RESUBSCRIBE_SINCE_GUARD_SECONDS * 4,
+          },
+        ],
+        (event) => {
+          received.push(event);
+        },
+        () => {
+          eoseCount += 1;
+        },
+      );
+      await waitFor(() => received.length >= 1, 5_000);
+
+      const testPool = pool as unknown as TestableApplesauceRelayPool;
+      testPool.rebuild('test-resume-clamp');
+      await testPool.rebuildInFlight;
+      await waitFor(() => eoseCount >= 2, 5_000);
+
+      // A request from a slower clock, published after the restore, must
+      // still match the restored filter (live delivery applies it too).
+      const slowEvent = await signTaggedEvent(
+        signer,
+        pubkey,
+        tag,
+        now - 60,
+        'slow',
+      );
+      await pool.publish(slowEvent);
+      await waitFor(
+        () => received.some((e) => e.id === slowEvent.id),
+        5_000,
+      );
+      expect(received.some((e) => e.id === slowEvent.id)).toBe(true);
+
+      await pool.disconnect();
+    },
+    15_000,
+  );
+
+  test.serial(
+    'history-seeking filters without since are restored unchanged',
+    async () => {
+      const now = Math.floor(Date.now() / 1000);
+      const privateKey = generateSecretKey();
+      const signer = new PrivateKeySigner(bytesToHex(privateKey));
+      const pubkey = getPublicKey(privateKey);
+      const tag = `resume-nosince-${Date.now()}`;
+      const oldEvent = await signTaggedEvent(
+        signer,
+        pubkey,
+        tag,
+        now - RESUBSCRIBE_SINCE_GUARD_SECONDS * 2,
+        'old',
+      );
+
+      const pool = new ApplesauceRelayPool([relayUrl]);
+      await pool.connect();
+      await pool.publish(oldEvent);
+
+      const received: NostrEvent[] = [];
+      let eoseCount = 0;
+      await pool.subscribe(
+        [{ kinds: [1], '#t': [tag] }],
+        (event) => {
+          received.push(event);
+        },
+        () => {
+          eoseCount += 1;
+        },
+      );
+      await waitFor(() => received.length >= 1, 5_000);
+
+      const testPool = pool as unknown as TestableApplesauceRelayPool;
+      testPool.rebuild('test-resume-nosince');
+      await testPool.rebuildInFlight;
+      await waitFor(() => eoseCount >= 2, 5_000);
+
+      // A since-less filter seeks history: the whole store must replay again.
+      expect(received.map((e) => e.id)).toEqual([oldEvent.id, oldEvent.id]);
+
+      await pool.disconnect();
+    },
+    15_000,
+  );
+});