From 07d4d36665f0c208b6a77b55cff1dc3be52ff0bf Mon Sep 17 00:00:00 2001 From: ContextVM Date: Mon, 12 Oct 2026 16:15:08 +0100 Subject: [PATCH 1/3] fix(relay): bound reconnect replay with resumed since (#105) Restored subscriptions re-request only the newest sane created_at minus a 30-minute guard (RESUBSCRIBE_SINCE_GUARD_SECONDS) instead of the whole session, so a relay reconnect can no longer re-dispatch stored requests whose ids have aged out of the dedup cache and re-execute completed tool calls. Future-dated events cannot move the resume point; filters without since are restored unchanged. The transports' seenEventIds cache grows to 40,000 entries (DEFAULT_DEDUP_CACHE_SIZE, ~20,000 encrypted requests) to cover the replay window. --- .changeset/bounded-reconnect-replay.md | 5 + docs/contextvm-docs | 2 +- src/core/constants.ts | 8 + src/relay/applesauce-relay-pool.test.ts | 240 +++++++++++++++++++++++- src/relay/applesauce-relay-pool.ts | 81 ++++++-- src/transport/nostr-client-transport.ts | 3 +- src/transport/nostr-server-transport.ts | 4 +- 7 files changed, 311 insertions(+), 32 deletions(-) create mode 100644 .changeset/bounded-reconnect-replay.md diff --git a/.changeset/bounded-reconnect-replay.md b/.changeset/bounded-reconnect-replay.md new file mode 100644 index 0000000..67798a9 --- /dev/null +++ b/.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 sane `created_at` seen 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 from senders whose clocks trail the newest event by more than the guard may be dropped on reconnect, and same-event transparent retries (CEP-8) older than the guard may not be re-observed after a reconnect. diff --git a/docs/contextvm-docs b/docs/contextvm-docs index a268677..9e74196 160000 --- a/docs/contextvm-docs +++ b/docs/contextvm-docs @@ -1 +1 @@ -Subproject commit a2686779dec6db135edb3280a88f65163a2d3bb4 +Subproject commit 9e7419677cbdff0e1555153d7074c4ff8813f16c diff --git a/src/core/constants.ts b/src/core/constants.ts index f0633cc..b02fa67 100644 --- a/src/core/constants.ts +++ b/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). */ diff --git a/src/relay/applesauce-relay-pool.test.ts b/src/relay/applesauce-relay-pool.test.ts index 4e3dfca..2618fe7 100644 --- a/src/relay/applesauce-relay-pool.test.ts +++ b/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 { @@ -30,6 +34,7 @@ type TestableApplesauceRelayPool = Omit & { onEvent: (event: NostrEvent) => void; onEose?: () => void; unsubscribe?: () => void; + resumeWatermark?: number; } >; rebuildInFlight?: Promise; @@ -712,31 +717,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 +1287,218 @@ 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 { + 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 { + 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( + 'future-dated events do not advance the restore point', + 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()}`; + const oldEvent = await signTaggedEvent( + signer, + pubkey, + tag, + now - RESUBSCRIBE_SINCE_GUARD_SECONDS * 2, + 'old', + ); + // Junk with a far-future created_at must not push `since` past real + // traffic: if it did, the restore would drop everything and the old + // event would not be replayed below. + const futureEvent = await signTaggedEvent( + signer, + pubkey, + tag, + now + RESUBSCRIBE_SINCE_GUARD_SECONDS * 48, + 'future', + ); + + const pool = new ApplesauceRelayPool([relayUrl]); + await pool.connect(); + await pool.publish(oldEvent); + await pool.publish(futureEvent); + + 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); + + const testPool = pool as unknown as TestableApplesauceRelayPool; + testPool.rebuild('test-resume-clamp'); + await testPool.rebuildInFlight; + await waitFor(() => eoseCount >= 2, 5_000); + + const replayed = received.slice(2).map((e) => e.id); + expect(replayed).toContain(oldEvent.id); + expect(replayed).toContain(futureEvent.id); + + 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, + ); +}); diff --git a/src/relay/applesauce-relay-pool.ts b/src/relay/applesauce-relay-pool.ts index dad8ea1..dbc8449 100644 --- a/src/relay/applesauce-relay-pool.ts +++ b/src/relay/applesauce-relay-pool.ts @@ -30,6 +30,42 @@ export const PING_FILTER: Filter = { limit: 0, }; +/** + * How far back a restored subscription re-requests stored events, measured + * from the newest `created_at` observed on it. Sits well above the payment + * TTLs documented in CEP-8 (600s examples) so same-event retries of the + * transparent lifecycle (CEP-8 idempotency) stay deliverable across + * reconnects. + */ +export const RESUBSCRIBE_SINCE_GUARD_SECONDS = 1800; + +/** + * Advances `since` on filters that already carry one so a restored + * subscription re-requests only the guard window instead of the whole + * session (whole-session relay replay after reconnect re-dispatched every + * stored request past the transport dedup horizon — issue #105). Filters + * without `since` are history-seeking and pass through unchanged. + */ +function buildResumeFilters( + filters: Filter[], + resumeWatermark: number | undefined, +): Filter[] { + if (resumeWatermark === undefined) { + return filters; + } + return filters.map((filter) => + filter.since === undefined + ? filter + : { + ...filter, + since: Math.max( + filter.since, + resumeWatermark - RESUBSCRIBE_SINCE_GUARD_SECONDS, + ), + }, + ); +} + /** Subscription intent stored for replay after rebuild */ type SubscriptionDescriptor = { id: string; @@ -41,6 +77,8 @@ type SubscriptionDescriptor = { type SubscriptionState = SubscriptionDescriptor & { /** Runtime unsubscribe handle for the currently active RxJS subscription (if started). */ unsubscribe?: () => void; + /** Newest sane `created_at` observed; drives the `since` used on restore. */ + resumeWatermark?: number; }; /** How one relay's per-attempt publish ladder settled. */ @@ -462,17 +500,12 @@ export class ApplesauceRelayPool implements RelayHandler { /** * Creates a subscription wrapper around the RelayGroup's subscription method. - * @param filters - Array of filters to subscribe to. - * @param onEvent - Callback function for received events. - * @param onEose - Optional callback function for end-of-stream events. + * @param state - Subscription descriptor; its filters are re-evaluated on + * every REQ send so restores pick up the advanced `since`. * @returns Object with unsubscribe method. */ - private createSubscription( - filters: Filter[], - onEvent: (event: NostrEvent) => void, - onEose?: () => void, - ): () => void { - logger.debug('Creating subscription', { filters }); + private createSubscription(state: SubscriptionState): () => void { + logger.debug('Creating subscription', { filters: state.filters }); // NOTE: applesauce-relay 6.0.3 changed `RelayGroup.subscription()` to emit // only deduplicated NostrEvents and no longer surfaces EOSE markers. @@ -488,14 +521,30 @@ export class ApplesauceRelayPool implements RelayHandler { // and rely on the server re-observing it. Relay-layer dedup by event id // (whether applesauce 6.0.3's `distinct()` or a local `Set`) silently // swallows that retry and deadlocks the flow. + const trackResumeWatermark = (event: NostrEvent): void => { + // Future-dated junk must not move the restore point past real traffic; + // its created_at says nothing about how far the relay's stream advanced. + const now = Math.floor(Date.now() / 1000); + if (event.created_at <= now + RESUBSCRIBE_SINCE_GUARD_SECONDS) { + state.resumeWatermark = Math.max( + state.resumeWatermark ?? 0, + event.created_at, + ); + } + state.onEvent(event); + }; + const sub = this.relayGroup - .req(filters, { reconnect: Infinity, resubscribe: Infinity }) + .req( + () => buildResumeFilters(state.filters, state.resumeWatermark), + { reconnect: Infinity, resubscribe: Infinity }, + ) .subscribe({ next: (message) => { if (message.type === 'EOSE') { - onEose?.(); + state.onEose?.(); } else if (message.type === 'EVENT') { - onEvent(message.event); + trackResumeWatermark(message.event); } // OPEN / CLOSED / ERROR are intentionally ignored; the old // `subscription()` filtered them out as well. @@ -528,7 +577,7 @@ export class ApplesauceRelayPool implements RelayHandler { const id = `sub:${Date.now()}:${Math.random().toString(16).slice(2)}`; const state: SubscriptionState = { id, filters, onEvent, onEose }; - state.unsubscribe = this.createSubscription(filters, onEvent, onEose); + state.unsubscribe = this.createSubscription(state); this.subscriptions.set(id, state); // Start ping monitor lazily on first subscription @@ -836,11 +885,7 @@ export class ApplesauceRelayPool implements RelayHandler { // Replay all stored subscription descriptors // Note: New subscriptions added during rebuild will be replayed after this completes for (const sub of this.subscriptions.values()) { - sub.unsubscribe = this.createSubscription( - sub.filters, - sub.onEvent, - sub.onEose, - ); + sub.unsubscribe = this.createSubscription(sub); } logger.info('Relay pool rebuilt successfully'); diff --git a/src/transport/nostr-client-transport.ts b/src/transport/nostr-client-transport.ts index 159429b..b724638 100644 --- a/src/transport/nostr-client-transport.ts +++ b/src/transport/nostr-client-transport.ts @@ -10,6 +10,7 @@ import type { Transport } from '@contextvm/mcp-sdk/shared/transport'; import { DEFAULT_BOOTSTRAP_RELAY_URLS, DEFAULT_TIMEOUT_MS, + DEFAULT_DEDUP_CACHE_SIZE, DEFAULT_LRU_SIZE, INITIALIZE_METHOD, } from '../core/index.js'; @@ -191,7 +192,7 @@ export class NostrClientTransport * * Used for gift-wrap envelopes (outer event ids). Kept as a bounded LRU. */ - private readonly seenEventIds = new LruCache(DEFAULT_LRU_SIZE); + private readonly seenEventIds = new LruCache(DEFAULT_DEDUP_CACHE_SIZE); /** * Creates a new NostrClientTransport instance. diff --git a/src/transport/nostr-server-transport.ts b/src/transport/nostr-server-transport.ts index 4e53950..5941f5e 100644 --- a/src/transport/nostr-server-transport.ts +++ b/src/transport/nostr-server-transport.ts @@ -16,7 +16,7 @@ import { CTXVM_MESSAGES_KIND, DEFAULT_TIMEOUT_MS, NOSTR_TAGS, - DEFAULT_LRU_SIZE, + DEFAULT_DEDUP_CACHE_SIZE, } from '../core/index.js'; import { NostrEvent } from 'nostr-tools'; import { LogLevel } from '../core/utils/logger.js'; @@ -220,7 +220,7 @@ export class NostrServerTransport * * Used for gift-wrap envelopes (outer event ids) and decrypted inner events. */ - private readonly seenEventIds = new LruCache(DEFAULT_LRU_SIZE); + private readonly seenEventIds = new LruCache(DEFAULT_DEDUP_CACHE_SIZE); /** Receives inbound oversized-transfer frames from clients (client→server requests). */ private readonly oversizedReceiver: OversizedTransferReceiver; From 027659603ccb3dddb4d4f17d9e855ed31b4bf64c Mon Sep 17 00:00:00 2001 From: ContextVM Date: Tue, 13 Oct 2026 11:28:57 +0100 Subject: [PATCH 2/3] fix(relay): re-evaluate resume filters per REQ and ignore future timestamps Review findings on the reconnect-resume change: - applesauce-relay@6.2.1 evaluates a filter function once at req() construction and connection retries reuse that result, so the resume point only applied on pool rebuilds. The filter input is now a deferred stream, recomputed on every REQ (re)send including automatic recovery after a connection drop. - The watermark accepted created_at up to 1800s ahead of the local clock and advanced before the transport validated the envelope: a rejected near-future envelope could push the restore point past unseen requests from slower clocks. Future timestamps are now excluded outright. Regression tests cover automatic recovery after a real connection drop (no rebuild) and a near-future envelope followed by a slower-clock request. --- .changeset/bounded-reconnect-replay.md | 2 +- docs/contextvm-docs | 2 +- src/relay/applesauce-relay-pool.test.ts | 106 ++++++++++++++++++++---- src/relay/applesauce-relay-pool.ts | 28 +++++-- 4 files changed, 115 insertions(+), 23 deletions(-) diff --git a/.changeset/bounded-reconnect-replay.md b/.changeset/bounded-reconnect-replay.md index 67798a9..58f2df8 100644 --- a/.changeset/bounded-reconnect-replay.md +++ b/.changeset/bounded-reconnect-replay.md @@ -2,4 +2,4 @@ '@contextvm/sdk': patch --- -Bound relay-reconnect replay (#105). Restored subscriptions now advance their `since` to the newest sane `created_at` seen 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 from senders whose clocks trail the newest event by more than the guard may be dropped on reconnect, and same-event transparent retries (CEP-8) older than the guard may not be re-observed after a reconnect. +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 from senders whose clocks trail the newest event by more than the guard may be dropped on reconnect, and same-event transparent retries (CEP-8) older than the guard may not be re-observed after a reconnect. diff --git a/docs/contextvm-docs b/docs/contextvm-docs index 9e74196..12bacfb 160000 --- a/docs/contextvm-docs +++ b/docs/contextvm-docs @@ -1 +1 @@ -Subproject commit 9e7419677cbdff0e1555153d7074c4ff8813f16c +Subproject commit 12bacfb54b75c4ab1b677f4b8a74ce37a273566b diff --git a/src/relay/applesauce-relay-pool.test.ts b/src/relay/applesauce-relay-pool.test.ts index 2618fe7..e5e26d1 100644 --- a/src/relay/applesauce-relay-pool.test.ts +++ b/src/relay/applesauce-relay-pool.test.ts @@ -1393,13 +1393,13 @@ describe('ApplesauceRelayPool reconnect resume', () => { ); test.serial( - 'future-dated events do not advance the restore point', + '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-clamp-${Date.now()}`; + const tag = `resume-auto-${Date.now()}`; const oldEvent = await signTaggedEvent( signer, pubkey, @@ -1407,21 +1407,87 @@ describe('ApplesauceRelayPool reconnect resume', () => { now - RESUBSCRIBE_SINCE_GUARD_SECONDS * 2, 'old', ); - // Junk with a far-future created_at must not push `since` past real - // traffic: if it did, the restore would drop everything and the old - // event would not be replayed below. - const futureEvent = await signTaggedEvent( + 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 * 48, - 'future', + now + RESUBSCRIBE_SINCE_GUARD_SECONDS - 10, + 'junk', ); const pool = new ApplesauceRelayPool([relayUrl]); await pool.connect(); - await pool.publish(oldEvent); - await pool.publish(futureEvent); + await pool.publish(junkEvent); const received: NostrEvent[] = []; let eoseCount = 0; @@ -1440,16 +1506,28 @@ describe('ApplesauceRelayPool reconnect resume', () => { eoseCount += 1; }, ); - await waitFor(() => received.length >= 2, 5_000); + 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); - const replayed = received.slice(2).map((e) => e.id); - expect(replayed).toContain(oldEvent.id); - expect(replayed).toContain(futureEvent.id); + // 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(); }, diff --git a/src/relay/applesauce-relay-pool.ts b/src/relay/applesauce-relay-pool.ts index dbc8449..4c71ec9 100644 --- a/src/relay/applesauce-relay-pool.ts +++ b/src/relay/applesauce-relay-pool.ts @@ -10,7 +10,11 @@ import { createLogger } from '../core/utils/logger.js'; import { sleep } from '../core/utils/utils.js'; import { ensureWebSocket } from '../core/utils/websocket.js'; import { + defer, lastValueFrom, + merge, + NEVER, + of, Subscription, takeUntil, timer, @@ -522,10 +526,11 @@ export class ApplesauceRelayPool implements RelayHandler { // (whether applesauce 6.0.3's `distinct()` or a local `Set`) silently // swallows that retry and deadlocks the flow. const trackResumeWatermark = (event: NostrEvent): void => { - // Future-dated junk must not move the restore point past real traffic; - // its created_at says nothing about how far the relay's stream advanced. + // Only timestamps at or below the local clock advance the watermark: + // a rejected envelope with a near-future created_at must not push the + // restore point past unseen requests from slower clocks. const now = Math.floor(Date.now() / 1000); - if (event.created_at <= now + RESUBSCRIBE_SINCE_GUARD_SECONDS) { + if (event.created_at <= now) { state.resumeWatermark = Math.max( state.resumeWatermark ?? 0, event.created_at, @@ -534,11 +539,20 @@ export class ApplesauceRelayPool implements RelayHandler { state.onEvent(event); }; + // Deferred filter stream: applesauce-relay evaluates a filter function + // once at req() construction and connection retries reuse that result, so + // each REQ (re)send — initial, retry after a connection drop, resubscribe + // after CLOSED, rebuild — re-subscribes this input and picks up the + // current resume point. + const filtersInput = merge( + defer(() => + of(buildResumeFilters(state.filters, state.resumeWatermark)), + ), + NEVER, + ); + const sub = this.relayGroup - .req( - () => buildResumeFilters(state.filters, state.resumeWatermark), - { reconnect: Infinity, resubscribe: Infinity }, - ) + .req(filtersInput, { reconnect: Infinity, resubscribe: Infinity }) .subscribe({ next: (message) => { if (message.type === 'EOSE') { From 3e1177f781045848dbef42c91794205c57204aa2 Mon Sep 17 00:00:00 2001 From: ContextVM Date: Tue, 13 Oct 2026 11:45:39 +0100 Subject: [PATCH 3/3] chore(relay): align resume wording and drop unused test type Review follow-ups: the reconnect loss window wording now names the real condition (created_at trailing the newest event seen, e.g. a lagging relay rather than only a slow sender clock), the resumeWatermark comment matches the implemented rule, and the unused test-helper type entry is gone. --- .changeset/bounded-reconnect-replay.md | 2 +- docs/contextvm-docs | 2 +- src/relay/applesauce-relay-pool.test.ts | 1 - src/relay/applesauce-relay-pool.ts | 2 +- 4 files changed, 3 insertions(+), 4 deletions(-) diff --git a/.changeset/bounded-reconnect-replay.md b/.changeset/bounded-reconnect-replay.md index 58f2df8..4157d3d 100644 --- a/.changeset/bounded-reconnect-replay.md +++ b/.changeset/bounded-reconnect-replay.md @@ -2,4 +2,4 @@ '@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 from senders whose clocks trail the newest event by more than the guard may be dropped on reconnect, and same-event transparent retries (CEP-8) older than the guard may not be re-observed after a reconnect. +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. diff --git a/docs/contextvm-docs b/docs/contextvm-docs index 12bacfb..a41063a 160000 --- a/docs/contextvm-docs +++ b/docs/contextvm-docs @@ -1 +1 @@ -Subproject commit 12bacfb54b75c4ab1b677f4b8a74ce37a273566b +Subproject commit a41063a62cfaee07cd2895582be7037371ebea72 diff --git a/src/relay/applesauce-relay-pool.test.ts b/src/relay/applesauce-relay-pool.test.ts index e5e26d1..9ab297e 100644 --- a/src/relay/applesauce-relay-pool.test.ts +++ b/src/relay/applesauce-relay-pool.test.ts @@ -34,7 +34,6 @@ type TestableApplesauceRelayPool = Omit & { onEvent: (event: NostrEvent) => void; onEose?: () => void; unsubscribe?: () => void; - resumeWatermark?: number; } >; rebuildInFlight?: Promise; diff --git a/src/relay/applesauce-relay-pool.ts b/src/relay/applesauce-relay-pool.ts index 4c71ec9..1fc5dbe 100644 --- a/src/relay/applesauce-relay-pool.ts +++ b/src/relay/applesauce-relay-pool.ts @@ -81,7 +81,7 @@ type SubscriptionDescriptor = { type SubscriptionState = SubscriptionDescriptor & { /** Runtime unsubscribe handle for the currently active RxJS subscription (if started). */ unsubscribe?: () => void; - /** Newest sane `created_at` observed; drives the `since` used on restore. */ + /** Newest `created_at` at or below the local clock; drives the `since` used on restore. */ resumeWatermark?: number; };