From 9e07d18b4253591dc223b2f5eb774157fdac7264 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Mon, 20 Jul 2026 15:07:42 -0700 Subject: [PATCH] fix(desktop): dedupe replayed thread reply notifications Persist notified reply IDs per channel and retain replay-boundary entries until reconnect catch-up completes. Co-authored-by: Ian Oberst Signed-off-by: Ian Oberst Co-authored-by: Codex Ai-assisted: true --- .../threadReplyNotificationDedupe.test.mjs | 597 ++++++++++++++++++ .../channels/threadReplyNotificationDedupe.ts | 283 +++++++++ .../channels/useLiveChannelUpdates.ts | 169 ++++- desktop/src/shared/api/relayClientSession.ts | 36 +- desktop/src/shared/api/relayClientShared.ts | 5 + .../shared/api/relayClosedRecovery.test.mjs | 338 +++++++++- desktop/src/shared/api/relayClosedRecovery.ts | 78 ++- .../src/shared/api/relayEventBuffer.test.mjs | 40 ++ desktop/src/shared/api/relayEventBuffer.ts | 15 + .../shared/api/relayReconnectReplay.test.mjs | 109 +++- .../src/shared/api/relayReconnectReplay.ts | 144 +++-- 11 files changed, 1731 insertions(+), 83 deletions(-) create mode 100644 desktop/src/features/channels/threadReplyNotificationDedupe.test.mjs create mode 100644 desktop/src/features/channels/threadReplyNotificationDedupe.ts create mode 100644 desktop/src/shared/api/relayEventBuffer.test.mjs create mode 100644 desktop/src/shared/api/relayEventBuffer.ts diff --git a/desktop/src/features/channels/threadReplyNotificationDedupe.test.mjs b/desktop/src/features/channels/threadReplyNotificationDedupe.test.mjs new file mode 100644 index 0000000000..2e07bedae4 --- /dev/null +++ b/desktop/src/features/channels/threadReplyNotificationDedupe.test.mjs @@ -0,0 +1,597 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + replayLiveSubscriptions, + replayReconnectHistoryPages, +} from "@/shared/api/relayReconnectReplay"; +import { buildChannelFilter } from "@/shared/api/relayChannelFilters"; +import { + handleRelayClosed, + handleSubscriptionEose, +} from "@/shared/api/relayClosedRecovery"; +import { + shouldDeliverThreadReplyDesktopNotification, + ThreadReplyNotificationDedupe, + THREAD_REPLY_SEEN_MAX_ITEMS, +} from "./threadReplyNotificationDedupe.ts"; +import { + disposeStaleLiveChannelSubscriptions, + reconcileAfterLiveSubscriptionAdditions, +} from "./useLiveChannelUpdates.ts"; + +function memoryStorage(initialRecords = []) { + const values = new Map(); + let seeded = false; + let writeCount = 0; + return { + getItem: (key) => { + if (!seeded && initialRecords.length > 0) { + values.set(key, JSON.stringify(initialRecords)); + seeded = true; + } + return values.get(key) ?? null; + }, + setItem: (key, value) => { + writeCount += 1; + values.set(key, value); + }, + get writeCount() { + return writeCount; + }, + }; +} + +function deferred() { + let resolve; + const promise = new Promise((next) => { + resolve = next; + }); + return { promise, resolve }; +} + +test("reconnect prunes before newest-first replay and retains the boundary ID", async () => { + const dedupe = new ThreadReplyNotificationDedupe("user", memoryStorage()); + const notified = []; + + dedupe.setChannelReplayFloor("channel", 900); + assert.equal(dedupe.record("channel", "boundary", 996), true); + + // The original live filter starts at 900, so the 996 boundary event remains + // inside the history window that the relay can restore. + dedupe.handleConnectionState("reconnecting"); + + const newestPage = Array.from({ length: 500 }, (_, index) => ({ + id: `offline-${index}`, + created_at: 1_501 + index, + })); + const pages = [newestPage, [{ id: "boundary", created_at: 996 }]]; + await replayReconnectHistoryPages({ + subscription: { + mode: "live", + filter: { kinds: [], limit: 500 }, + onEvent: (event) => { + if (dedupe.record("channel", event.id, event.created_at)) { + notified.push(event.id); + } + }, + }, + since: 995, + until: 2_000, + isActive: () => true, + requestHistory: async () => pages.shift() ?? [], + }); + + assert.equal(notified.length, 500); + assert.equal(notified.includes("offline-0"), true); + assert.equal(notified.includes("boundary"), false); + assert.equal( + dedupe.record("channel", "boundary", 996), + false, + "newest replay delivery must not evict the older boundary event", + ); +}); + +test("offline arrival notifies once and is persisted across remount", () => { + const storage = memoryStorage(); + const first = new ThreadReplyNotificationDedupe("user", storage); + + first.setChannelReplayFloor("channel", 100); + first.handleConnectionState("reconnecting"); + + assert.equal(first.record("channel", "offline", 200), true); + assert.equal(first.record("channel", "offline", 200), false); + first.flush(); + + const remounted = new ThreadReplyNotificationDedupe("user", storage); + remounted.setChannelReplayFloor("channel", 100); + remounted.handleConnectionState("reconnecting"); + assert.equal(remounted.record("channel", "offline", 200), false); +}); + +test("late restored-live duplicate stays suppressed until EOSE completes", async () => { + const dedupe = new ThreadReplyNotificationDedupe("user", memoryStorage()); + const notified = []; + const onEvent = (event) => { + if (dedupe.record("channel", event.id, event.created_at)) { + notified.push(event.id); + } + }; + const subscription = { + mode: "live", + filter: { + ...buildChannelFilter("channel", 1_000), + since: 900, + }, + onEvent, + lastSeenCreatedAt: 1_000, + }; + const subscriptions = new Map([["live", subscription]]); + + dedupe.setChannelReplayFloor("channel", 900); + assert.equal(dedupe.record("channel", "boundary", 996), true); + dedupe.handleConnectionState("reconnecting"); + + const replayPromise = replayLiveSubscriptions({ + subscriptions, + now: 2_000, + sendRaw: async () => {}, + requestHistory: async () => [{ id: "offline", created_at: 2_000 }], + }); + await new Promise((resolve) => setImmediate(resolve)); + + // The restored live REQ can deliver its duplicate after HTTP paging. EOSE + // is the barrier that must keep the connection in replay mode until this + // callback has been handed to the client event buffer. + onEvent({ id: "boundary", created_at: 996 }); + subscription.resolveReconnectEose(); + await replayPromise; + dedupe.handleConnectionState("connected"); + + assert.deepEqual(notified, ["offline"]); +}); + +test("active-channel reply is recorded before desktop suppression", () => { + const dedupe = new ThreadReplyNotificationDedupe("user", memoryStorage()); + + dedupe.setChannelReplayFloor("channel", 100); + const firstSeen = dedupe.record("channel", "active-reply", 100); + assert.equal( + shouldDeliverThreadReplyDesktopNotification({ + isFirstSeen: firstSeen, + isEligible: true, + isActiveChannel: true, + notifyForActiveChannel: false, + }), + false, + ); + + dedupe.handleConnectionState("reconnecting"); + const replayFirstSeen = dedupe.record("channel", "active-reply", 100); + assert.equal( + shouldDeliverThreadReplyDesktopNotification({ + isFirstSeen: replayFirstSeen, + isEligible: true, + isActiveChannel: false, + notifyForActiveChannel: false, + }), + false, + "navigating away before reconnect must not surface a stale notification", + ); +}); + +test("reply observed while ineligible cannot notify after eligibility changes", () => { + const dedupe = new ThreadReplyNotificationDedupe("user", memoryStorage()); + dedupe.reconcileActiveChannels(["channel"]); + dedupe.setChannelReplayFloor("channel", 100); + + const firstSeen = dedupe.record("channel", "muted-reply", 100); + assert.equal( + shouldDeliverThreadReplyDesktopNotification({ + isFirstSeen: firstSeen, + isEligible: false, + isActiveChannel: false, + notifyForActiveChannel: false, + }), + false, + ); + + dedupe.handleConnectionState("reconnecting"); + const replayFirstSeen = dedupe.record("channel", "muted-reply", 100); + assert.equal( + shouldDeliverThreadReplyDesktopNotification({ + isFirstSeen: replayFirstSeen, + isEligible: true, + isActiveChannel: false, + notifyForActiveChannel: false, + }), + false, + "unmute/follow changes must not turn a replay into a stale notification", + ); +}); + +test("startup reconciliation removes departed records before cap eviction", () => { + const activeRecord = { + eventId: "active-boundary", + channelId: "active", + createdAt: 1, + }; + const departedRecords = Array.from( + { length: THREAD_REPLY_SEEN_MAX_ITEMS }, + (_, index) => ({ + eventId: `departed-${index}`, + channelId: "departed", + createdAt: index + 2, + }), + ); + const dedupe = new ThreadReplyNotificationDedupe( + "user", + memoryStorage([activeRecord, ...departedRecords]), + ); + + dedupe.reconcileActiveChannels(["active"]); + dedupe.setChannelReplayFloor("active", 0); + + assert.equal( + dedupe.record("active", "active-boundary", 1), + false, + "departed records must be removed before they can evict an active ID", + ); + assert.equal(dedupe.record("departed", "departed-0", 2), false); + dedupe.reconcileActiveChannels(["active", "departed"]); + dedupe.setChannelReplayFloor("departed", 0); + assert.equal(dedupe.record("departed", "departed-0", 2), true); +}); + +test("startup provisional target retains its persisted same-second boundary", () => { + const boundary = { + eventId: "persisted-boundary", + channelId: "target", + createdAt: 100, + }; + const dedupe = new ThreadReplyNotificationDedupe( + "user", + memoryStorage([boundary]), + ); + + dedupe.reconcileActiveChannels(["target"]); + dedupe.setChannelReplayFloor("target", 100); + + assert.equal( + dedupe.record("target", boundary.eventId, boundary.createdAt), + false, + "a target channel must stay persisted while its subscription settles", + ); +}); + +test("channel removal deletes its replay floor and persisted records", () => { + const storage = memoryStorage(); + const dedupe = new ThreadReplyNotificationDedupe("user", storage); + dedupe.reconcileActiveChannels(["active", "departed"]); + dedupe.setChannelReplayFloor("active", 0); + dedupe.setChannelReplayFloor("departed", 0); + assert.equal(dedupe.record("active", "active-reply", 1), true); + assert.equal(dedupe.record("departed", "departed-reply", 2), true); + + dedupe.reconcileActiveChannels(["active"]); + + const remounted = new ThreadReplyNotificationDedupe("user", storage); + remounted.reconcileActiveChannels(["active"]); + remounted.setChannelReplayFloor("active", 0); + assert.equal(remounted.record("active", "active-reply", 1), false); + remounted.reconcileActiveChannels(["active", "departed"]); + remounted.setChannelReplayFloor("departed", 0); + assert.equal(remounted.record("departed", "departed-reply", 2), true); +}); + +test("canceled sync cannot overwrite replacement reconciliation", async () => { + const dedupe = new ThreadReplyNotificationDedupe("user", memoryStorage()); + const additionA = deferred(); + const additionB = deferred(); + let currentGeneration = 1; + let syncACancelled = false; + + const syncA = reconcileAfterLiveSubscriptionAdditions({ + additions: [additionA.promise], + isCurrentSync: () => !syncACancelled && currentGeneration === 1, + reconcile: () => dedupe.reconcileActiveChannels([]), + }); + + syncACancelled = true; + currentGeneration = 2; + dedupe.reconcileActiveChannels([]); + dedupe.setChannelReplayFloor("replacement", 100); + const syncB = reconcileAfterLiveSubscriptionAdditions({ + additions: [additionB.promise], + isCurrentSync: () => currentGeneration === 2, + reconcile: () => dedupe.reconcileActiveChannels(["replacement"]), + }); + + additionA.resolve(); + assert.equal(await syncA, false); + assert.equal( + dedupe.record("replacement", "live-reply", 100), + true, + "a stale sync must not reject an event delivered before sync B resolves", + ); + + additionB.resolve(); + assert.equal(await syncB, true); + dedupe.handleConnectionState("reconnecting"); + assert.equal( + dedupe.record("replacement", "live-reply", 100), + false, + "the live sighting must stay recorded when the same event replays", + ); +}); + +test("superseding pending sync provisionally retains the target boundary", async () => { + const boundary = { + eventId: "pending-boundary", + channelId: "replacement", + createdAt: 100, + }; + const dedupe = new ThreadReplyNotificationDedupe( + "user", + memoryStorage([boundary]), + ); + const addition = deferred(); + + dedupe.reconcileActiveChannels(["replacement"]); + dedupe.setChannelReplayFloor("replacement", 100); + const sync = reconcileAfterLiveSubscriptionAdditions({ + additions: [addition.promise], + isCurrentSync: () => true, + reconcile: () => dedupe.reconcileActiveChannels(["replacement"]), + }); + + assert.equal( + dedupe.record("replacement", boundary.eventId, boundary.createdAt), + false, + ); + addition.resolve(); + assert.equal(await sync, true); +}); + +test("identity handoff disposes old recovery ownership before recreating", () => { + const oldStorage = memoryStorage(); + const nextStorage = memoryStorage(); + const oldDedupe = new ThreadReplyNotificationDedupe("old", oldStorage); + const nextDedupe = new ThreadReplyNotificationDedupe("next", nextStorage); + oldDedupe.reconcileActiveChannels(["channel"]); + oldDedupe.handleSubscriptionRecoveryState(true); + const oldWritesDuringRecovery = oldStorage.writeCount; + let disposeCalls = 0; + const activeSubs = new Map([ + [ + "channel", + { + replayFloor: 100, + dedupe: oldDedupe, + dispose: async () => { + disposeCalls += 1; + oldDedupe.handleSubscriptionRecoveryState(false); + }, + }, + ], + ]); + + disposeStaleLiveChannelSubscriptions({ + activeSubs, + targetIds: new Set(["channel"]), + dedupe: nextDedupe, + }); + assert.equal(disposeCalls, 1); + assert.equal(activeSubs.size, 0); + assert.equal(oldStorage.writeCount, oldWritesDuringRecovery + 1); + + nextDedupe.reconcileActiveChannels(["channel"]); + nextDedupe.handleSubscriptionRecoveryState(true); + const nextWritesDuringRecovery = nextStorage.writeCount; + assert.equal(nextDedupe.record("channel", "next-replay", 100), true); + assert.equal(nextStorage.writeCount, nextWritesDuringRecovery); + nextDedupe.handleSubscriptionRecoveryState(false); + assert.equal(nextStorage.writeCount, nextWritesDuringRecovery + 1); +}); + +test("full-cap newest-first CLOSED recovery retains the older boundary ID", () => { + const originalWindow = globalThis.window; + const scheduled = []; + globalThis.window = { + clearTimeout: () => {}, + setTimeout: (callback, delay) => { + scheduled.push({ callback, delay }); + return scheduled.length; + }, + }; + try { + const dedupe = new ThreadReplyNotificationDedupe("user", undefined); + dedupe.reconcileActiveChannels(["channel"]); + dedupe.setChannelReplayFloor("channel", 0); + assert.equal(dedupe.record("channel", "boundary", 1), true); + for (let index = 1; index < THREAD_REPLY_SEEN_MAX_ITEMS; index += 1) { + assert.equal(dedupe.record("channel", `seen-${index}`, index + 1), true); + } + + const subscription = { + mode: "live", + filter: { kinds: [9], "#h": ["channel"], limit: 1_000, since: 0 }, + onEvent: () => {}, + onClosedRecoveryStateChange: (recovering) => + dedupe.handleSubscriptionRecoveryState(recovering), + }; + const subscriptions = new Map([["live-1", subscription]]); + handleRelayClosed({ + subscriptions, + subId: "live-1", + message: "rate-limited: retry later", + sendReq: async () => {}, + }); + + assert.deepEqual( + scheduled.map(({ delay }) => delay).sort((left, right) => left - right), + [1_000, 8_000], + ); + assert.equal(dedupe.record("channel", "newest-replay", 2_001), true); + assert.equal( + dedupe.record("channel", "boundary", 1), + false, + "cap eviction must remain suspended until the older replay page arrives", + ); + + handleSubscriptionEose({ + subscriptions, + subId: "live-1", + closeSubscription: async () => {}, + }); + } finally { + globalThis.window = originalWindow; + } +}); + +test("overlapping socket and CLOSED recovery flush only after both complete", () => { + const storage = memoryStorage(); + const dedupe = new ThreadReplyNotificationDedupe("user", storage); + dedupe.reconcileActiveChannels(["channel"]); + const writesBeforeReplay = storage.writeCount; + + dedupe.handleSubscriptionRecoveryState(true); + dedupe.handleConnectionState("reconnecting"); + assert.equal(dedupe.record("channel", "offline", 100), true); + dedupe.handleSubscriptionRecoveryState(false); + assert.equal(storage.writeCount, writesBeforeReplay); + + dedupe.handleConnectionState("connected"); + assert.equal(storage.writeCount, writesBeforeReplay + 1); +}); + +test("stalled socket acquires replay ownership before CLOSED recovery ends", () => { + const dedupe = new ThreadReplyNotificationDedupe("user", undefined); + dedupe.reconcileActiveChannels(["channel"]); + dedupe.setChannelReplayFloor("channel", 0); + assert.equal(dedupe.record("channel", "boundary", 1), true); + for (let index = 1; index < THREAD_REPLY_SEEN_MAX_ITEMS; index += 1) { + assert.equal(dedupe.record("channel", `seen-${index}`, index + 1), true); + } + + dedupe.handleSubscriptionRecoveryState(true); + assert.equal(dedupe.record("channel", "newest-replay", 2_001), true); + dedupe.handleConnectionState("stalled"); + dedupe.handleSubscriptionRecoveryState(false); + assert.equal( + dedupe.record("channel", "boundary", 1), + false, + "stalled reset must retain the boundary until socket replay completes", + ); + dedupe.handleConnectionState("reconnecting"); + dedupe.handleConnectionState("connected"); +}); + +test("pruning is per-channel and follows each live filter's actual since", () => { + const dedupe = new ThreadReplyNotificationDedupe("user", memoryStorage()); + + dedupe.reconcileActiveChannels(["busy", "quiet"]); + dedupe.setChannelReplayFloor("busy", 900); + dedupe.setChannelReplayFloor("quiet", 0); + assert.equal(dedupe.record("busy", "busy-old", 994), true); + assert.equal(dedupe.record("busy", "busy-floor", 995), true); + assert.equal(dedupe.record("quiet", "quiet-boundary", 96), true); + + dedupe.setChannelReplayFloor("busy", 995); + dedupe.setChannelReplayFloor("quiet", 95); + + assert.equal(dedupe.record("busy", "busy-old", 994), true); + assert.equal(dedupe.record("busy", "busy-floor", 995), false); + assert.equal( + dedupe.record("quiet", "quiet-boundary", 96), + false, + "busy-channel replay floor must not evict quiet-channel records", + ); +}); + +test("size cap evicts oldest records only after replay completes", () => { + const dedupe = new ThreadReplyNotificationDedupe("user", memoryStorage()); + + dedupe.reconcileActiveChannels(["channel"]); + dedupe.setChannelReplayFloor("channel", 0); + dedupe.handleConnectionState("reconnecting"); + for (let index = 0; index <= THREAD_REPLY_SEEN_MAX_ITEMS; index += 1) { + assert.equal(dedupe.record("channel", `event-${index}`, index + 1), true); + } + + assert.equal( + dedupe.record("channel", "event-0", 1), + false, + "cap must remain suspended while older replay pages can still arrive", + ); + + dedupe.handleConnectionState("connected"); + assert.equal(dedupe.record("channel", "event-0", 1), true); + assert.equal( + dedupe.record( + "channel", + `event-${THREAD_REPLY_SEEN_MAX_ITEMS}`, + THREAD_REPLY_SEEN_MAX_ITEMS + 1, + ), + false, + "newest records must survive the cap", + ); +}); + +test("replay coalesces persistence and stores the final capped set", () => { + const storage = memoryStorage(); + const dedupe = new ThreadReplyNotificationDedupe("user", storage); + + dedupe.reconcileActiveChannels(["channel"]); + dedupe.setChannelReplayFloor("channel", 0); + dedupe.handleConnectionState("reconnecting"); + const writesBeforeReplay = storage.writeCount; + + for (let index = 0; index <= THREAD_REPLY_SEEN_MAX_ITEMS; index += 1) { + assert.equal(dedupe.record("channel", `replay-${index}`, index + 1), true); + } + + assert.equal( + storage.writeCount, + writesBeforeReplay, + "replayed events must not synchronously rewrite the growing cache", + ); + + dedupe.handleConnectionState("connected"); + assert.equal(storage.writeCount, writesBeforeReplay + 1); + + const remounted = new ThreadReplyNotificationDedupe("user", storage); + remounted.reconcileActiveChannels(["channel"]); + remounted.setChannelReplayFloor("channel", 0); + assert.equal(remounted.record("channel", "replay-0", 1), true); + assert.equal( + remounted.record( + "channel", + `replay-${THREAD_REPLY_SEEN_MAX_ITEMS}`, + THREAD_REPLY_SEEN_MAX_ITEMS + 1, + ), + false, + "the single replay flush must persist the capped newest records", + ); +}); + +test("storage failure disables repeated serialization attempts", () => { + let writeCount = 0; + const storage = { + getItem: () => null, + setItem: () => { + writeCount += 1; + throw new Error("quota exceeded"); + }, + }; + const dedupe = new ThreadReplyNotificationDedupe("user", storage); + + dedupe.reconcileActiveChannels(["channel"]); + dedupe.setChannelReplayFloor("channel", 0); + for (let index = 0; index < 100; index += 1) { + assert.equal(dedupe.record("channel", `event-${index}`, index), true); + } + + assert.equal(writeCount, 1); + assert.equal(dedupe.record("channel", "event-99", 99), false); +}); diff --git a/desktop/src/features/channels/threadReplyNotificationDedupe.ts b/desktop/src/features/channels/threadReplyNotificationDedupe.ts new file mode 100644 index 0000000000..2e9471e6a1 --- /dev/null +++ b/desktop/src/features/channels/threadReplyNotificationDedupe.ts @@ -0,0 +1,283 @@ +import type { ConnectionState } from "@/shared/api/relayClientShared"; + +const STORAGE_KEY_PREFIX = "buzz-thread-reply-desktop-seen.v1"; +export const THREAD_REPLY_SEEN_MAX_ITEMS = 2_000; + +type SeenThreadReply = { + eventId: string; + channelId: string; + createdAt: number; +}; + +type SeenStorage = Pick; + +function storageKey(pubkey: string) { + return `${STORAGE_KEY_PREFIX}:${pubkey}`; +} + +function defaultStorage(): SeenStorage | undefined { + if (typeof window === "undefined") { + return undefined; + } + + try { + return window.localStorage; + } catch { + return undefined; + } +} + +function isSeenThreadReply(value: unknown): value is SeenThreadReply { + if (typeof value !== "object" || value === null) { + return false; + } + + const candidate = value as Partial; + return ( + typeof candidate.eventId === "string" && + typeof candidate.channelId === "string" && + typeof candidate.createdAt === "number" && + Number.isFinite(candidate.createdAt) + ); +} + +function readSeenThreadReplies( + pubkey: string, + storage: SeenStorage | undefined, +): SeenThreadReply[] { + if (!storage || pubkey.length === 0) { + return []; + } + + try { + const rawValue = storage.getItem(storageKey(pubkey)); + if (!rawValue) { + return []; + } + + const parsed: unknown = JSON.parse(rawValue); + return Array.isArray(parsed) ? parsed.filter(isSeenThreadReply) : []; + } catch { + return []; + } +} + +/** + * Remembers thread replies already offered to the desktop notification path. + * + * Reconnect history is delivered newest-page first, so cap eviction is + * suspended during socket reconnect and retryable CLOSED recovery; otherwise + * a burst could discard an older boundary ID before its replay page arrives. + * Age eviction follows each broad subscription's original `since`, because + * that is the actual history window restored by both recovery paths. + */ +export class ThreadReplyNotificationDedupe { + private readonly pubkey: string; + private readonly storage: SeenStorage | undefined; + private readonly seenById = new Map(); + private readonly channelReplayFloors = new Map(); + private activeChannelIds: Set | undefined; + private connectionReplayInProgress = false; + private subscriptionReplayCount = 0; + private persistenceDirty = false; + private persistenceDisabled = false; + + constructor( + pubkey: string, + storage: SeenStorage | undefined = defaultStorage(), + ) { + this.pubkey = pubkey; + this.storage = storage; + for (const record of readSeenThreadReplies(pubkey, storage)) { + this.seenById.set(record.eventId, record); + } + // Do not cap persisted data until active channels are known. Departed + // records must be removed before they can displace an active-channel ID. + } + + /** Removes replay state for channels without an active broad subscription. */ + reconcileActiveChannels(channelIds: Iterable) { + const activeChannelIds = new Set(channelIds); + for (const channelId of this.channelReplayFloors.keys()) { + if (!activeChannelIds.has(channelId)) { + this.channelReplayFloors.delete(channelId); + } + } + for (const [eventId, record] of this.seenById) { + if (!activeChannelIds.has(record.channelId)) { + this.seenById.delete(eventId); + } + } + + this.activeChannelIds = activeChannelIds; + this.pruneAllChannels(); + if (!this.isReplayInProgress()) { + this.enforceSizeCap(); + } + this.persist(); + } + + setChannelReplayFloor(channelId: string, replayFloor: number) { + this.activeChannelIds?.add(channelId); + this.channelReplayFloors.set(channelId, replayFloor); + if (this.activeChannelIds && !this.isReplayInProgress()) { + this.pruneChannel(channelId, replayFloor); + this.enforceSizeCap(); + this.persist(); + } + } + + /** Records an observed reply and returns true only for its first sighting. */ + record(channelId: string, eventId: string, createdAt: number) { + if (this.activeChannelIds && !this.activeChannelIds.has(channelId)) { + return false; + } + if (this.seenById.has(eventId)) { + return false; + } + + this.seenById.set(eventId, { eventId, channelId, createdAt }); + if (this.activeChannelIds && !this.isReplayInProgress()) { + this.enforceSizeCap(); + } + this.persist(); + return true; + } + + handleConnectionState(state: ConnectionState) { + if (state === "reconnecting" || state === "stalled") { + if (!this.connectionReplayInProgress) { + const replayWasInProgress = this.isReplayInProgress(); + this.connectionReplayInProgress = true; + if (replayWasInProgress) return; + // The original long-lived live REQ is restored on reconnect, so its + // immutable `since` is the actual replay floor for persisted IDs. + this.beginReplay(); + } + return; + } + + if (state === "connected" && this.connectionReplayInProgress) { + this.connectionReplayInProgress = false; + this.finishReplayIfIdle(); + } + } + + handleSubscriptionRecoveryState(recovering: boolean) { + if (recovering) { + const replayWasInProgress = this.isReplayInProgress(); + this.subscriptionReplayCount += 1; + if (!replayWasInProgress) { + this.beginReplay(); + } + return; + } + + if (this.subscriptionReplayCount === 0) return; + this.subscriptionReplayCount -= 1; + this.finishReplayIfIdle(); + } + + /** Flushes pending replay state when this dedupe instance is handed off. */ + flush() { + this.pruneAllChannels(); + if (this.activeChannelIds) { + this.enforceSizeCap(); + } + this.flushPersistence(); + } + + private pruneAllChannels() { + for (const [channelId, replayFloor] of this.channelReplayFloors) { + this.pruneChannel(channelId, replayFloor); + } + } + + private pruneChannel(channelId: string, replayFloor: number) { + for (const [eventId, record] of this.seenById) { + if (record.channelId === channelId && record.createdAt < replayFloor) { + this.seenById.delete(eventId); + } + } + } + + private enforceSizeCap() { + const excess = this.seenById.size - THREAD_REPLY_SEEN_MAX_ITEMS; + if (excess <= 0) { + return; + } + + const oldest = [...this.seenById.values()] + .sort((left, right) => left.createdAt - right.createdAt) + .slice(0, excess); + for (const record of oldest) { + this.seenById.delete(record.eventId); + } + } + + private persist() { + this.persistenceDirty = true; + if (this.isReplayInProgress()) { + return; + } + this.flushPersistence(); + } + + private isReplayInProgress() { + return this.connectionReplayInProgress || this.subscriptionReplayCount > 0; + } + + private beginReplay() { + this.pruneAllChannels(); + this.persist(); + } + + private finishReplayIfIdle() { + if (this.isReplayInProgress()) return; + this.pruneAllChannels(); + if (this.activeChannelIds) { + this.enforceSizeCap(); + } + this.persist(); + } + + private flushPersistence() { + if (!this.persistenceDirty) { + return; + } + if (!this.storage || this.pubkey.length === 0 || this.persistenceDisabled) { + this.persistenceDirty = false; + return; + } + + try { + this.storage.setItem( + storageKey(this.pubkey), + JSON.stringify([...this.seenById.values()]), + ); + this.persistenceDirty = false; + } catch { + // In-memory dedupe still works when localStorage is unavailable/full. + // Disable more serialization attempts for this instance so a full store + // cannot add repeated main-thread work after the first failed flush. + this.persistenceDisabled = true; + this.persistenceDirty = false; + } + } +} + +export function shouldDeliverThreadReplyDesktopNotification({ + isFirstSeen, + isEligible, + isActiveChannel, + notifyForActiveChannel, +}: { + isFirstSeen: boolean; + isEligible: boolean; + isActiveChannel: boolean; + notifyForActiveChannel: boolean; +}) { + return ( + isFirstSeen && isEligible && (!isActiveChannel || notifyForActiveChannel) + ); +} diff --git a/desktop/src/features/channels/useLiveChannelUpdates.ts b/desktop/src/features/channels/useLiveChannelUpdates.ts index f8db0af5e2..a30255b31f 100644 --- a/desktop/src/features/channels/useLiveChannelUpdates.ts +++ b/desktop/src/features/channels/useLiveChannelUpdates.ts @@ -22,6 +22,10 @@ import { import { isDmNotifiableKind } from "./isDmNotifiableKind"; import { refreshChannelsWhenIdle } from "./refreshChannelsWhenIdle"; +import { + shouldDeliverThreadReplyDesktopNotification, + ThreadReplyNotificationDedupe, +} from "./threadReplyNotificationDedupe"; export type UseLiveChannelUpdatesOptions = { currentPubkey?: string; @@ -71,6 +75,45 @@ export type UseLiveChannelUpdatesOptions = { const LIVE_SUBSCRIPTION_RETRY_BASE_MS = 1_000; const LIVE_SUBSCRIPTION_RETRY_MAX_MS = 30_000; +type LiveChannelSubscription = { + dispose: () => Promise; + replayFloor: number; + dedupe: ThreadReplyNotificationDedupe; +}; + +export function disposeStaleLiveChannelSubscriptions({ + activeSubs, + targetIds, + dedupe, +}: { + activeSubs: Map; + targetIds: ReadonlySet; + dedupe: ThreadReplyNotificationDedupe; +}) { + for (const [channelId, active] of activeSubs) { + if (targetIds.has(channelId) && active.dedupe === dedupe) continue; + activeSubs.delete(channelId); + void active.dispose().catch(() => {}); + } +} + +export async function reconcileAfterLiveSubscriptionAdditions({ + additions, + isCurrentSync, + reconcile, +}: { + additions: Promise[]; + isCurrentSync: () => boolean; + reconcile: () => void; +}) { + await Promise.allSettled(additions); + if (!isCurrentSync()) { + return false; + } + reconcile(); + return true; +} + // get_channels is an expensive O(channels) relay fan-out. Incoming traffic for // non-active channels arrives in bursts, so coalesce the refetch into a single // trailing invalidation instead of one per event. @@ -159,6 +202,21 @@ export function useLiveChannelUpdates( ); const seenDmEventIdsRef = React.useRef(new Set()); const dmSubscriptionStartedAtRef = React.useRef(0); + const threadReplyDedupeRef = React.useRef<{ + pubkey: string; + value: ThreadReplyNotificationDedupe; + } | null>(null); + if (threadReplyDedupeRef.current?.pubkey !== normalizedCurrentPubkey) { + threadReplyDedupeRef.current = { + pubkey: normalizedCurrentPubkey, + value: new ThreadReplyNotificationDedupe(normalizedCurrentPubkey), + }; + } + const threadReplyDedupe = threadReplyDedupeRef.current.value; + const pendingThreadReplyDedupeFlushRef = React.useRef<{ + dedupe: ThreadReplyNotificationDedupe; + timeout: number; + } | null>(null); // Reset subscription timestamp when identity changes. React.useEffect(() => { @@ -259,6 +317,12 @@ export function useLiveChannelUpdates( const isThreadedReply = isThreadReply(event.tags); if (isExternalTriggerEvent) { + // Eligibility can change after delivery (mute, follow, participation), + // so record first sight before consulting that mutable state. + const isFirstSeenThreadReply = + isThreadedReply && !isDmChannel + ? threadReplyDedupe.record(channelId, event.id, event.created_at) + : false; const shouldNotify = shouldNotifyForEvent( event, normalizedCurrentPubkey, @@ -283,11 +347,15 @@ export function useLiveChannelUpdates( } } - if (shouldNotify && isThreadedReply) { - if ( - !dmChannelMap.has(channelId) && - (channelId !== activeChannelId || options.notifyForActiveChannel) - ) { + if (isThreadedReply) { + const shouldDeliverDesktopNotification = + shouldDeliverThreadReplyDesktopNotification({ + isFirstSeen: isFirstSeenThreadReply, + isEligible: shouldNotify, + isActiveChannel: channelId === activeChannelId, + notifyForActiveChannel: options.notifyForActiveChannel ?? false, + }); + if (shouldDeliverDesktopNotification) { options.onThreadReplyDesktopNotification?.(channelId, event); } } @@ -309,6 +377,12 @@ export function useLiveChannelUpdates( ); }); + const handleLiveChannelEvent = React.useEffectEvent( + (channelId: string, event: RelayEvent) => { + handleIncomingMessage(withChannelTagFallback(event, channelId)); + }, + ); + const handleMentionEvent = React.useEffectEvent((event: RelayEvent) => { if (!isExternalMentionEvent(event, normalizedCurrentPubkey)) { return; @@ -322,6 +396,32 @@ export function useLiveChannelUpdates( options.onLiveMention?.(); }); + React.useEffect(() => { + const pendingFlush = pendingThreadReplyDedupeFlushRef.current; + if (pendingFlush?.dedupe === threadReplyDedupe) { + window.clearTimeout(pendingFlush.timeout); + pendingThreadReplyDedupeFlushRef.current = null; + } + const unsubscribe = relayClient.subscribeToConnectionState((state) => { + threadReplyDedupe.handleConnectionState(state); + }); + return () => { + unsubscribe(); + // Delay handoff flushing by one task so React StrictMode's immediate + // effect cleanup/setup cycle can cancel it for the same live instance. + const timeout = window.setTimeout(() => { + threadReplyDedupe.flush(); + if (pendingThreadReplyDedupeFlushRef.current?.timeout === timeout) { + pendingThreadReplyDedupeFlushRef.current = null; + } + }, 0); + pendingThreadReplyDedupeFlushRef.current = { + dedupe: threadReplyDedupe, + timeout, + }; + }; + }, [threadReplyDedupe]); + React.useEffect(() => { return relayClient.subscribeToReconnects(() => { void queryClient.invalidateQueries({ queryKey: channelsQueryKey }); @@ -332,23 +432,36 @@ export function useLiveChannelUpdates( }); }, [queryClient]); - const liveSubsRef = React.useRef(new Map Promise>()); + const liveSubsRef = React.useRef(new Map()); + const liveSyncGenerationRef = React.useRef(0); React.useEffect(() => { let isCancelled = false; let retryTimeout: number | undefined; let retryAttempt = 0; + const syncGeneration = liveSyncGenerationRef.current + 1; + liveSyncGenerationRef.current = syncGeneration; + const isCurrentSync = () => + !isCancelled && liveSyncGenerationRef.current === syncGeneration; const syncSubs = async (): Promise => { + if (!isCurrentSync()) { + return true; + } const activeSubs = liveSubsRef.current; const targetIds = new Set(channelIdsKey ? channelIdsKey.split(",") : []); - for (const [channelId, dispose] of activeSubs) { - if (!targetIds.has(channelId)) { - activeSubs.delete(channelId); - void dispose().catch(() => {}); - } + disposeStaleLiveChannelSubscriptions({ + activeSubs, + targetIds, + dedupe: threadReplyDedupe, + }); + for (const [channelId, active] of activeSubs) { + threadReplyDedupe.setChannelReplayFloor(channelId, active.replayFloor); } + // Keep target channels provisionally active while their subscriptions + // settle so same-second replay cannot lose persisted boundary IDs. + threadReplyDedupe.reconcileActiveChannels(targetIds); if (targetIds.size > 0) { // Record the subscription start time so handleDmEvent can distinguish @@ -361,21 +474,31 @@ export function useLiveChannelUpdates( .filter((channelId) => !activeSubs.has(channelId)) .map(async (channelId) => { try { + const subscriptionSince = Math.floor(Date.now() / 1_000); + threadReplyDedupe.setChannelReplayFloor( + channelId, + subscriptionSince, + ); const dispose = await relayClient.subscribeLive( { kinds: [...CHANNEL_EVENT_KINDS], "#h": [channelId], limit: 1000, - since: Math.floor(Date.now() / 1_000), + since: subscriptionSince, }, - (event) => - handleIncomingMessage(withChannelTagFallback(event, channelId)), + (event) => handleLiveChannelEvent(channelId, event), + (recovering) => + threadReplyDedupe.handleSubscriptionRecoveryState(recovering), ); - if (isCancelled) { + if (!isCurrentSync()) { void dispose().catch(() => {}); return; } - activeSubs.set(channelId, dispose); + activeSubs.set(channelId, { + dispose, + replayFloor: subscriptionSince, + dedupe: threadReplyDedupe, + }); } catch (err) { anyFailed = true; console.error( @@ -385,7 +508,15 @@ export function useLiveChannelUpdates( ); } }); - await Promise.allSettled(additions); + const didReconcile = await reconcileAfterLiveSubscriptionAdditions({ + additions, + isCurrentSync, + reconcile: () => + threadReplyDedupe.reconcileActiveChannels(activeSubs.keys()), + }); + if (!didReconcile) { + return true; + } return !anyFailed; }; @@ -415,7 +546,7 @@ export function useLiveChannelUpdates( window.clearTimeout(retryTimeout); } }; - }, [channelIdsKey]); + }, [channelIdsKey, threadReplyDedupe]); // Subscribe to mention events per channel with a diff-based manager: only // subscribe newly-added channels and unsubscribe removed ones on each sync. @@ -520,7 +651,7 @@ export function useLiveChannelUpdates( return () => { channelsInvalidateRef.current?.cancel(); - for (const dispose of liveSubsRef.current.values()) { + for (const { dispose } of liveSubsRef.current.values()) { void dispose().catch(() => {}); } liveSubsRef.current.clear(); diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 7a151e5d28..8addff0264 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -29,13 +29,14 @@ import { } from "@/shared/api/relayChannelFilters"; import { collectWithConcurrency } from "@/shared/api/concurrency"; import { - clearClosedRetry, handleRelayClosed, handleSubscriptionEose, prepareSubscriptionEvent, + releaseLiveSubscription, } from "@/shared/api/relayClosedRecovery"; import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay"; import { RelayConnectionStateEmitter } from "@/shared/api/relayConnectionStateEmitter"; +import { deliverBufferedSubscriptionEvents } from "@/shared/api/relayEventBuffer"; import { shouldRefuseConnect, shouldScheduleReconnect, @@ -139,7 +140,7 @@ export class RelayClient { window.clearTimeout(sub.timeout); sub.reject(error); } else { - clearClosedRetry(sub); + releaseLiveSubscription(sub); } this.subscriptions.delete(subId); } @@ -427,12 +428,12 @@ export class RelayClient { async subscribeToAllStreamMessages(onEvent: (event: RelayEvent) => void) { return this.subscribe(buildGlobalStreamFilter(50), onEvent); } - async subscribeLive( filter: RelaySubscriptionFilter, onEvent: (event: RelayEvent) => void, + onClosedRecoveryStateChange?: (recovering: boolean) => void, ) { - return this.subscribe(filter, onEvent); + return this.subscribe(filter, onEvent, onClosedRecoveryStateChange); } async subscribeToChannelMentionEvents( @@ -447,8 +448,7 @@ export class RelayClient { } async preconnect() { - // Explicit re-engagement. If the session went terminal (auth rejection) - // the caller is asking us to try again, so clear the latch. + // Explicit re-engagement after a terminal session clears the latch. this.terminal = false; this.keepAliveRequested = true; await this.ensureConnected(); @@ -550,6 +550,9 @@ export class RelayClient { this.reconnectDelayMs = RECONNECT_BASE_DELAY_MS; await this.replayLiveSubscriptions(); + if (generation !== this.connectionGeneration) { + throw new Error("Relay changed during subscription replay."); + } this.connectionStateEmitter.set("connected"); this.stallWatchdog.start(); this.emitReconnectIfNeeded(); @@ -558,6 +561,7 @@ export class RelayClient { private async subscribe( filter: RelaySubscriptionFilter, onEvent: (event: RelayEvent) => void, + onClosedRecoveryStateChange?: (recovering: boolean) => void, ) { await this.ensureConnected(); @@ -579,6 +583,8 @@ export class RelayClient { mode: "live", filter, onEvent, + onClosedRecoveryStateChange, + onClosedRecoveryTimeout: (error) => this.resetConnection(error), resolveReady, }); @@ -601,7 +607,7 @@ export class RelayClient { } this.subscriptions.delete(subId); - clearClosedRetry(active); + releaseLiveSubscription(active); await this.closeSubscription(subId); }; } @@ -825,17 +831,11 @@ export class RelayClient { } private flushEventBuffer() { + window.clearTimeout(this.flushTimeout ?? undefined); this.flushTimeout = null; const buffer = this.eventBuffer; this.eventBuffer = []; - - // Re-lookup: subscriptions removed during batch window are intentionally skipped. - for (const { subId, event } of buffer) { - const subscription = this.subscriptions.get(subId); - if (subscription?.mode === "live") { - subscription.onEvent(event); - } - } + deliverBufferedSubscriptionEvents(buffer, this.subscriptions); } private handleEose(subId: string) { @@ -843,6 +843,7 @@ export class RelayClient { subscriptions: this.subscriptions, subId, closeSubscription: (id) => this.closeSubscription(id), + beforeLiveRecoveryComplete: () => this.flushEventBuffer(), }); } @@ -895,6 +896,7 @@ export class RelayClient { sendRaw: (payload) => this.sendRaw(payload), requestHistory: (filter) => this.requestHistory(filter), }); + this.flushEventBuffer(); } catch (error) { const reconnectError = error instanceof Error @@ -1003,9 +1005,7 @@ export class RelayClient { continue; } - subscription.resolveReady?.(); - subscription.resolveReady = undefined; - clearClosedRetry(subscription); + releaseLiveSubscription(subscription); } for (const [eventId, pendingEvent] of this.pendingEvents) { diff --git a/desktop/src/shared/api/relayClientShared.ts b/desktop/src/shared/api/relayClientShared.ts index 76b7448c06..c919a63f5a 100644 --- a/desktop/src/shared/api/relayClientShared.ts +++ b/desktop/src/shared/api/relayClientShared.ts @@ -50,8 +50,13 @@ type LiveSubscription = { mode: "live"; filter: RelaySubscriptionFilter; onEvent: (event: RelayEvent) => void; + onClosedRecoveryStateChange?: (recovering: boolean) => void; + onClosedRecoveryTimeout?: (error: Error) => void; resolveReady?: () => void; + resolveReconnectEose?: () => void; lastSeenCreatedAt?: number; + closedRecoveryInProgress?: boolean; + closedRecoveryTimeout?: number; closedRetryAttempt?: number; closedRetryTimeout?: number; }; diff --git a/desktop/src/shared/api/relayClosedRecovery.test.mjs b/desktop/src/shared/api/relayClosedRecovery.test.mjs index 9c9b59440f..7963d7e4b0 100644 --- a/desktop/src/shared/api/relayClosedRecovery.test.mjs +++ b/desktop/src/shared/api/relayClosedRecovery.test.mjs @@ -1,7 +1,12 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { handleRelayClosed } from "./relayClosedRecovery.ts"; +import { + handleRelayClosed, + handleSubscriptionEose, + releaseLiveSubscription, +} from "./relayClosedRecovery.ts"; +import { replayLiveSubscriptions } from "./relayReconnectReplay.ts"; test("production CLOSED handler rejects history once and clears its timeout", () => { const originalWindow = globalThis.window; @@ -47,6 +52,7 @@ test("production CLOSED handler rejects history once and clears its timeout", () test("production CLOSED handler removes terminal live subscriptions", () => { let readyCalls = 0; + let reconnectEoseCalls = 0; const subscriptions = new Map([ [ "live-1", @@ -57,6 +63,9 @@ test("production CLOSED handler removes terminal live subscriptions", () => { resolveReady: () => { readyCalls += 1; }, + resolveReconnectEose: () => { + reconnectEoseCalls += 1; + }, }, ], ]); @@ -68,4 +77,331 @@ test("production CLOSED handler removes terminal live subscriptions", () => { }); assert.equal(subscriptions.has("live-1"), false); assert.equal(readyCalls, 1); + assert.equal(reconnectEoseCalls, 1); +}); + +test("retryable CLOSED preserves the original long-lived filter", async () => { + const originalWindow = globalThis.window; + const scheduled = []; + globalThis.window = { + clearTimeout: () => {}, + setTimeout: (callback, delay) => { + scheduled.push({ callback, delay }); + return 42; + }, + }; + try { + const sentFilters = []; + const subscriptions = new Map([ + [ + "live-1", + { + mode: "live", + filter: { + kinds: [9], + "#h": ["channel-1"], + limit: 1_000, + since: 900, + }, + onEvent: () => {}, + lastSeenCreatedAt: 1_000, + }, + ], + ]); + + handleRelayClosed({ + subscriptions, + subId: "live-1", + message: "rate-limited: retry later", + sendReq: async (_subId, filter) => { + sentFilters.push(filter); + }, + }); + + assert.deepEqual( + scheduled.map(({ delay }) => delay), + [8_000, 1_000], + ); + scheduled.find(({ delay }) => delay === 1_000).callback(); + await Promise.resolve(); + assert.deepEqual(sentFilters, [ + { + kinds: [9], + "#h": ["channel-1"], + limit: 1_000, + since: 900, + }, + ]); + } finally { + globalThis.window = originalWindow; + } +}); + +test("retryable CLOSED recovery drains buffered events before completing", () => { + const originalWindow = globalThis.window; + const scheduled = []; + globalThis.window = { + clearTimeout: () => {}, + setTimeout: (callback, delay) => { + scheduled.push({ callback, delay }); + return scheduled.length; + }, + }; + try { + const lifecycle = []; + const subscription = { + mode: "live", + filter: { kinds: [9], limit: 50 }, + onEvent: () => {}, + onClosedRecoveryStateChange: (recovering) => + lifecycle.push(recovering ? "recovering" : "live"), + }; + const subscriptions = new Map([["live-1", subscription]]); + const closedInput = { + subscriptions, + subId: "live-1", + message: "rate-limited: retry later", + sendReq: async () => {}, + }; + + handleRelayClosed(closedInput); + handleRelayClosed(closedInput); + assert.deepEqual(lifecycle, ["recovering"]); + assert.deepEqual( + scheduled.map(({ delay }) => delay), + [8_000, 1_000], + ); + + handleSubscriptionEose({ + subscriptions, + subId: "live-1", + closeSubscription: async () => {}, + beforeLiveRecoveryComplete: () => lifecycle.push("drained"), + }); + assert.deepEqual(lifecycle, ["recovering", "drained", "live"]); + } finally { + globalThis.window = originalWindow; + } +}); + +test("repeated retryable CLOSED cannot extend the reconnect EOSE deadline", async () => { + const originalWindow = globalThis.window; + const scheduled = []; + globalThis.window = { + clearTimeout: () => {}, + setTimeout: (callback, delay) => { + scheduled.push({ callback, delay }); + return scheduled.length; + }, + }; + try { + const subscription = { + mode: "live", + filter: { + kinds: [9], + "#h": ["channel-1"], + limit: 1_000, + since: 900, + }, + onEvent: () => {}, + lastSeenCreatedAt: 1_000, + }; + const subscriptions = new Map([["live-1", subscription]]); + const sentFilters = []; + const replayOutcome = replayLiveSubscriptions({ + subscriptions, + now: 2_000, + eoseTimeoutMs: 10, + sendRaw: async () => {}, + requestHistory: async () => [], + }).then( + () => "resolved", + (error) => error, + ); + + await new Promise((resolve) => setImmediate(resolve)); + for (let attempt = 0; attempt < 3; attempt += 1) { + handleRelayClosed({ + subscriptions, + subId: "live-1", + message: "rate-limited: retry later", + sendReq: async (_subId, filter) => { + sentFilters.push(filter); + }, + }); + assert.equal(scheduled.filter(({ delay }) => delay === 8_000).length, 1); + scheduled.find(({ delay }) => delay === 1_000 * 2 ** attempt).callback(); + await new Promise((resolve) => setImmediate(resolve)); + } + + const outcome = await replayOutcome; + assert.equal(outcome instanceof Error, true); + assert.match(outcome.message, /EOSE/); + assert.equal(subscription.resolveReconnectEose, undefined); + assert.deepEqual( + sentFilters.map((filter) => filter.since), + [900, 900, 900], + ); + releaseLiveSubscription(subscription); + } finally { + globalThis.window = originalWindow; + } +}); + +test("standalone CLOSED recovery has a bounded EOSE deadline", async () => { + const originalWindow = globalThis.window; + const scheduled = []; + const clearedTimeouts = []; + globalThis.window = { + clearTimeout: (timeout) => clearedTimeouts.push(timeout), + setTimeout: (callback, delay) => { + scheduled.push({ callback, delay, id: scheduled.length + 1 }); + return scheduled.length; + }, + }; + try { + const lifecycle = []; + let timeoutError; + const subscription = { + mode: "live", + filter: { kinds: [9], limit: 50 }, + onEvent: () => {}, + onClosedRecoveryStateChange: (recovering) => lifecycle.push(recovering), + onClosedRecoveryTimeout: (error) => { + assert.equal(subscription.closedRecoveryInProgress, true); + timeoutError = error; + releaseLiveSubscription(subscription); + }, + }; + const subscriptions = new Map([["live-1", subscription]]); + + handleRelayClosed({ + subscriptions, + subId: "live-1", + message: "rate-limited: retry later", + sendReq: async () => {}, + }); + scheduled.find(({ delay }) => delay === 1_000).callback(); + await Promise.resolve(); + assert.deepEqual(lifecycle, [true]); + + scheduled.find(({ delay }) => delay === 8_000).callback(); + assert.match(timeoutError.message, /EOSE/); + assert.deepEqual(lifecycle, [true, false]); + assert.equal(subscription.closedRecoveryTimeout, undefined); + assert.equal(clearedTimeouts.includes(1), false); + } finally { + globalThis.window = originalWindow; + } +}); + +test("socket reset starts CLOSED retry backoff from the base delay", () => { + const originalWindow = globalThis.window; + const scheduled = []; + globalThis.window = { + clearTimeout: () => {}, + setTimeout: (callback, delay) => { + scheduled.push({ callback, delay }); + return scheduled.length; + }, + }; + try { + const subscription = { + mode: "live", + filter: { kinds: [9], limit: 50 }, + onEvent: () => {}, + closedRetryAttempt: 4, + onClosedRecoveryTimeout: () => releaseLiveSubscription(subscription), + }; + const subscriptions = new Map([["live-1", subscription]]); + const closedInput = { + subscriptions, + subId: "live-1", + message: "rate-limited: retry later", + sendReq: async () => {}, + }; + + handleRelayClosed(closedInput); + assert.deepEqual( + scheduled.map(({ delay }) => delay), + [8_000, 16_000], + ); + scheduled.find(({ delay }) => delay === 8_000).callback(); + assert.equal(subscription.closedRetryAttempt, 0); + + handleRelayClosed(closedInput); + assert.deepEqual( + scheduled.slice(2).map(({ delay }) => delay), + [8_000, 1_000], + ); + assert.equal(subscription.closedRetryAttempt, 1); + } finally { + globalThis.window = originalWindow; + } +}); + +test("live EOSE releases the reconnect replay barrier", () => { + let reconnectEoseCalls = 0; + const subscriptions = new Map([ + [ + "live-1", + { + mode: "live", + filter: { kinds: [9], limit: 50 }, + onEvent: () => {}, + resolveReconnectEose: () => { + reconnectEoseCalls += 1; + }, + }, + ], + ]); + + handleSubscriptionEose({ + subscriptions, + subId: "live-1", + closeSubscription: () => Promise.resolve(), + }); + + assert.equal(reconnectEoseCalls, 1); + assert.equal(subscriptions.get("live-1").resolveReconnectEose, undefined); +}); + +test("live subscription cleanup releases every reconnect waiter", () => { + const originalWindow = globalThis.window; + const clearedTimeouts = []; + globalThis.window = { + clearTimeout: (timeout) => clearedTimeouts.push(timeout), + }; + try { + let readyCalls = 0; + let reconnectEoseCalls = 0; + const subscription = { + mode: "live", + filter: { kinds: [9], limit: 50 }, + onEvent: () => {}, + resolveReady: () => { + readyCalls += 1; + }, + resolveReconnectEose: () => { + reconnectEoseCalls += 1; + }, + closedRetryTimeout: 42, + closedRetryAttempt: 4, + closedRecoveryInProgress: true, + closedRecoveryTimeout: 43, + }; + + releaseLiveSubscription(subscription); + + assert.equal(readyCalls, 1); + assert.equal(reconnectEoseCalls, 1); + assert.equal(subscription.resolveReady, undefined); + assert.equal(subscription.resolveReconnectEose, undefined); + assert.equal(subscription.closedRetryTimeout, undefined); + assert.equal(subscription.closedRetryAttempt, 0); + assert.equal(subscription.closedRecoveryTimeout, undefined); + assert.deepEqual(clearedTimeouts, [42, 43]); + } finally { + globalThis.window = originalWindow; + } }); diff --git a/desktop/src/shared/api/relayClosedRecovery.ts b/desktop/src/shared/api/relayClosedRecovery.ts index e8038bb71c..09b39e078a 100644 --- a/desktop/src/shared/api/relayClosedRecovery.ts +++ b/desktop/src/shared/api/relayClosedRecovery.ts @@ -8,15 +8,74 @@ import type { RelayEvent } from "@/shared/api/types"; const RETRY_BASE_DELAY_MS = 1_000; const RETRY_MAX_DELAY_MS = 30_000; +export const CLOSED_RECOVERY_EOSE_TIMEOUT_MS = 8_000; type LiveSubscription = Extract; +function setClosedRecoveryState( + subscription: LiveSubscription, + recovering: boolean, +) { + if (Boolean(subscription.closedRecoveryInProgress) === recovering) return; + if (!recovering && subscription.closedRecoveryTimeout !== undefined) { + window.clearTimeout(subscription.closedRecoveryTimeout); + subscription.closedRecoveryTimeout = undefined; + } + subscription.closedRecoveryInProgress = recovering || undefined; + subscription.onClosedRecoveryStateChange?.(recovering); +} + +function armClosedRecoveryTimeout({ + subscriptions, + subId, + subscription, +}: { + subscriptions: Map; + subId: string; + subscription: LiveSubscription; +}) { + if (subscription.closedRecoveryTimeout !== undefined) return; + subscription.closedRecoveryTimeout = window.setTimeout(() => { + subscription.closedRecoveryTimeout = undefined; + if ( + subscriptions.get(subId) !== subscription || + !subscription.closedRecoveryInProgress + ) { + return; + } + const error = new Error( + `Timed out waiting for relay EOSE while recovering subscription ${subId}.`, + ); + if (subscription.onClosedRecoveryTimeout) { + subscription.onClosedRecoveryTimeout(error); + return; + } + subscriptions.delete(subId); + releaseLiveSubscription(subscription); + }, CLOSED_RECOVERY_EOSE_TIMEOUT_MS); +} + +export function resolveReconnectEose(subscription: LiveSubscription) { + const resolve = subscription.resolveReconnectEose; + subscription.resolveReconnectEose = undefined; + resolve?.(); +} + export function clearClosedRetry(subscription: LiveSubscription) { if (subscription.closedRetryTimeout === undefined) return; window.clearTimeout(subscription.closedRetryTimeout); subscription.closedRetryTimeout = undefined; } +export function releaseLiveSubscription(subscription: LiveSubscription) { + subscription.resolveReady?.(); + subscription.resolveReady = undefined; + resolveReconnectEose(subscription); + clearClosedRetry(subscription); + subscription.closedRetryAttempt = 0; + setClosedRecoveryState(subscription, false); +} + export function handleRelayClosed({ subscriptions, subId, @@ -63,9 +122,14 @@ function recoverLiveSubscriptionFromClosed({ subscription.resolveReady?.(); subscription.resolveReady = undefined; if (!isRetryableRelayClosed(message)) { + resolveReconnectEose(subscription); + clearClosedRetry(subscription); + setClosedRecoveryState(subscription, false); subscriptions.delete(subId); return; } + setClosedRecoveryState(subscription, true); + armClosedRecoveryTimeout({ subscriptions, subId, subscription }); if (subscription.closedRetryTimeout !== undefined) return; const attempt = subscription.closedRetryAttempt ?? 0; @@ -78,7 +142,12 @@ function recoverLiveSubscriptionFromClosed({ subscription.closedRetryTimeout = undefined; if (subscriptions.get(subId) !== subscription) return; void sendReq(subId, subscription.filter).catch((error) => { - if (subscriptions.get(subId) !== subscription) return; + if ( + subscriptions.get(subId) !== subscription || + !subscription.closedRecoveryInProgress + ) { + return; + } console.error("Failed to restore closed relay subscription", error); recoverLiveSubscriptionFromClosed({ subscriptions, @@ -112,18 +181,25 @@ export function handleSubscriptionEose({ subscriptions, subId, closeSubscription, + beforeLiveRecoveryComplete, }: { subscriptions: Map; subId: string; closeSubscription: (subId: string) => Promise; + beforeLiveRecoveryComplete?: () => void; }) { const subscription = subscriptions.get(subId); if (!subscription) return; if (subscription.mode === "live") { + if (subscription.closedRecoveryInProgress) { + beforeLiveRecoveryComplete?.(); + } + resolveReconnectEose(subscription); subscription.resolveReady?.(); subscription.resolveReady = undefined; subscription.closedRetryAttempt = 0; clearClosedRetry(subscription); + setClosedRecoveryState(subscription, false); return; } window.clearTimeout(subscription.timeout); diff --git a/desktop/src/shared/api/relayEventBuffer.test.mjs b/desktop/src/shared/api/relayEventBuffer.test.mjs new file mode 100644 index 0000000000..0604a45f6a --- /dev/null +++ b/desktop/src/shared/api/relayEventBuffer.test.mjs @@ -0,0 +1,40 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { deliverBufferedSubscriptionEvents } from "./relayEventBuffer.ts"; + +test("event-buffer drain delivers only still-live subscriptions in order", () => { + const delivered = []; + const subscriptions = new Map([ + [ + "live", + { + mode: "live", + filter: { kinds: [9], limit: 50 }, + onEvent: (event) => delivered.push(event.id), + }, + ], + [ + "history", + { + mode: "history", + events: [], + resolve: () => {}, + reject: () => {}, + timeout: 1, + }, + ], + ]); + + deliverBufferedSubscriptionEvents( + [ + { subId: "removed", event: { id: "removed" } }, + { subId: "live", event: { id: "first" } }, + { subId: "history", event: { id: "history" } }, + { subId: "live", event: { id: "second" } }, + ], + subscriptions, + ); + + assert.deepEqual(delivered, ["first", "second"]); +}); diff --git a/desktop/src/shared/api/relayEventBuffer.ts b/desktop/src/shared/api/relayEventBuffer.ts new file mode 100644 index 0000000000..491a4eb24e --- /dev/null +++ b/desktop/src/shared/api/relayEventBuffer.ts @@ -0,0 +1,15 @@ +import type { RelaySubscription } from "@/shared/api/relayClientShared"; +import type { RelayEvent } from "@/shared/api/types"; + +export function deliverBufferedSubscriptionEvents( + buffer: Array<{ subId: string; event: RelayEvent }>, + subscriptions: Map, +) { + // Re-lookup: subscriptions removed during the batch window are skipped. + for (const { subId, event } of buffer) { + const subscription = subscriptions.get(subId); + if (subscription?.mode === "live") { + subscription.onEvent(event); + } + } +} diff --git a/desktop/src/shared/api/relayReconnectReplay.test.mjs b/desktop/src/shared/api/relayReconnectReplay.test.mjs index 63f6402cda..5e23b87fd7 100644 --- a/desktop/src/shared/api/relayReconnectReplay.test.mjs +++ b/desktop/src/shared/api/relayReconnectReplay.test.mjs @@ -111,7 +111,10 @@ test("channel reconnect replay pages the missed window until a short page", asyn eventRange("middle", 1002, 500), eventRange("oldest", 995, 8), ]; - const filter = buildChannelFilter("channel-1", 50); + const filter = { + ...buildChannelFilter("channel-1", 1_000), + since: 900, + }; const subscriptions = new Map([ [ "live-1", @@ -129,6 +132,7 @@ test("channel reconnect replay pages the missed window until a short page", asyn now: 2000, sendRaw: async (payload) => { sentPayloads.push(payload); + subscriptions.get(payload[1])?.resolveReconnectEose?.(); }, requestHistory: async (filter) => { historyFilters.push(filter); @@ -143,7 +147,8 @@ test("channel reconnect replay pages the missed window until a short page", asyn { kinds: filter.kinds, "#h": ["channel-1"], - limit: 50, + limit: 1_000, + since: 900, }, ], ]); @@ -173,6 +178,101 @@ test("channel reconnect replay pages the missed window until a short page", asyn assert.equal(delivered.length, 1008); }); +test("reconnect replay waits for restored live EOSE before completing", async () => { + const subscription = { + mode: "live", + filter: { + ...buildChannelFilter("channel-1", 1_000), + since: 900, + }, + onEvent: () => {}, + lastSeenCreatedAt: 1_000, + }; + const subscriptions = new Map([["live-1", subscription]]); + let replayCompleted = false; + + const replayPromise = replayLiveSubscriptions({ + subscriptions, + now: 2_000, + sendRaw: async () => {}, + requestHistory: async () => [], + }).then(() => { + replayCompleted = true; + }); + + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(replayCompleted, false); + assert.equal(typeof subscription.resolveReconnectEose, "function"); + + subscription.resolveReconnectEose(); + await replayPromise; + assert.equal(replayCompleted, true); +}); + +test("restored live filter delivers post-EOSE events below the catch-up floor", async () => { + const delivered = []; + const subscription = { + mode: "live", + filter: { + ...buildChannelFilter("channel-1", 1_000), + since: 900, + }, + onEvent: (received) => delivered.push(received.id), + lastSeenCreatedAt: 1_000, + }; + const subscriptions = new Map([["live-1", subscription]]); + let restoredFilter; + + await replayLiveSubscriptions({ + subscriptions, + now: 2_000, + sendRaw: async (payload) => { + restoredFilter = payload[2]; + subscription.resolveReconnectEose?.(); + }, + requestHistory: async () => [], + }); + + const delayedEvent = event("delayed", 950); + if ( + restoredFilter.since === undefined || + delayedEvent.created_at >= restoredFilter.since + ) { + subscription.onEvent(delayedEvent); + } + + assert.equal(restoredFilter.since, 900); + assert.deepEqual(delivered, ["delayed"]); +}); + +test("reconnect replay rejects instead of hanging when EOSE is missing", async () => { + const subscription = { + mode: "live", + filter: buildChannelFilter("channel-1", 50), + onEvent: () => {}, + lastSeenCreatedAt: 1_000, + }; + const subscriptions = new Map([["live-1", subscription]]); + + const outcome = await Promise.race([ + replayLiveSubscriptions({ + subscriptions, + now: 2_000, + eoseTimeoutMs: 5, + sendRaw: async () => {}, + requestHistory: () => new Promise(() => {}), + }).then( + () => "resolved", + (error) => error, + ), + new Promise((resolve) => setTimeout(() => resolve("still-pending"), 50)), + ]); + + assert.equal(outcome instanceof Error, true); + assert.match(outcome.message, /EOSE/); + assert.equal(subscription.resolveReconnectEose, undefined); +}); + test("reconnect replay starts live REQs in parallel and preserves per-sub page order", async () => { const sentPayloads = []; const sendResolvers = []; @@ -218,7 +318,10 @@ test("reconnect replay starts live REQs in parallel and preserves per-sub page o sendRaw: (payload) => { sentPayloads.push(payload); return new Promise((resolve) => { - sendResolvers.push(resolve); + sendResolvers.push(() => { + resolve(); + subscriptions.get(payload[1])?.resolveReconnectEose?.(); + }); }); }, requestHistory: async (filter) => { diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts index 65709d351a..f651882f82 100644 --- a/desktop/src/shared/api/relayReconnectReplay.ts +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -5,9 +5,10 @@ import type { } from "@/shared/api/relayClientShared"; import type { RelayEvent } from "@/shared/api/types"; -const RECONNECT_REPLAY_SKEW_SECS = 5; +export const RECONNECT_REPLAY_SKEW_SECS = 5; export const RECONNECT_REPLAY_PAGE_LIMIT = 500; export const RECONNECT_REPLAY_PAGE_CONCURRENCY = 4; +export const RECONNECT_EOSE_TIMEOUT_MS = 8_000; async function runWithConcurrency( items: T[], @@ -58,6 +59,14 @@ export function shouldPageReconnectReplay(filter: RelaySubscriptionFilter) { ); } +export function reconnectReplaySince( + subscription: Extract, +) { + return subscription.lastSeenCreatedAt === undefined + ? undefined + : Math.max(0, subscription.lastSeenCreatedAt - RECONNECT_REPLAY_SKEW_SECS); +} + export async function replayReconnectHistoryPages({ subscription, since, @@ -104,12 +113,14 @@ export async function replayLiveSubscriptions({ requestHistory, now = Math.floor(Date.now() / 1_000), pageReplayConcurrency = RECONNECT_REPLAY_PAGE_CONCURRENCY, + eoseTimeoutMs = RECONNECT_EOSE_TIMEOUT_MS, }: { subscriptions: Map; sendRaw: (payload: unknown[]) => Promise; requestHistory: (filter: RelaySubscriptionFilter) => Promise; now?: number; pageReplayConcurrency?: number; + eoseTimeoutMs?: number; }) { const replayRequests = Array.from(subscriptions.entries()) .filter( @@ -119,51 +130,102 @@ export async function replayLiveSubscriptions({ entry[1].mode === "live", ) .map(([subId, subscription]) => { - const replaySince = - subscription.lastSeenCreatedAt === undefined - ? undefined - : Math.max( - 0, - subscription.lastSeenCreatedAt - RECONNECT_REPLAY_SKEW_SECS, - ); + const replaySince = reconnectReplaySince(subscription); const shouldPageReplay = replaySince !== undefined && shouldPageReconnectReplay(subscription.filter); - return { subId, subscription, replaySince, shouldPageReplay }; + let resolvePromise = () => {}; + let rejectPromise = (_error: Error) => {}; + const reconnectEose = new Promise((resolve, reject) => { + resolvePromise = resolve; + rejectPromise = reject; + }); + let settled = false; + let timeout: ReturnType | undefined; + const clearSubscriptionWaiter = () => { + if (subscription.resolveReconnectEose === resolveReconnectEose) { + subscription.resolveReconnectEose = undefined; + } + }; + const settle = (error?: Error) => { + if (settled) return; + settled = true; + if (timeout !== undefined) { + globalThis.clearTimeout(timeout); + timeout = undefined; + } + clearSubscriptionWaiter(); + if (error) { + rejectPromise(error); + } else { + resolvePromise(); + } + }; + const resolveReconnectEose = () => settle(); + const rejectReconnectEose = (error: Error) => settle(error); + const armReconnectEoseTimeout = () => { + if (settled || timeout !== undefined) return; + timeout = globalThis.setTimeout(() => { + rejectReconnectEose( + new Error( + `Timed out waiting for relay EOSE while restoring subscription ${subId}.`, + ), + ); + }, eoseTimeoutMs); + }; + subscription.resolveReconnectEose = resolveReconnectEose; + + return { + subId, + subscription, + replaySince, + shouldPageReplay, + reconnectEose, + resolveReconnectEose, + armReconnectEoseTimeout, + }; }); - await Promise.all( - replayRequests.map( - ({ subId, subscription, replaySince, shouldPageReplay }) => - sendRaw([ - "REQ", - subId, - shouldPageReplay - ? subscription.filter - : buildReconnectReplayFilter(subscription.filter, replaySince), - ]), - ), - ); + try { + await Promise.all( + replayRequests.map(({ subId, subscription }) => + sendRaw(["REQ", subId, subscription.filter]), + ), + ); - await runWithConcurrency( - replayRequests.filter( - ( - request, - ): request is typeof request & { - replaySince: number; - shouldPageReplay: true; - } => request.shouldPageReplay && request.replaySince !== undefined, - ), - pageReplayConcurrency, - async ({ subId, subscription, replaySince }) => { - await replayReconnectHistoryPages({ - subscription, - since: replaySince, - until: now, - isActive: () => subscriptions.get(subId) === subscription, - requestHistory, - }); - }, - ); + for (const { armReconnectEoseTimeout } of replayRequests) { + armReconnectEoseTimeout(); + } + await Promise.all([ + runWithConcurrency( + replayRequests.filter( + ( + request, + ): request is typeof request & { + replaySince: number; + shouldPageReplay: true; + } => request.shouldPageReplay && request.replaySince !== undefined, + ), + pageReplayConcurrency, + async ({ subId, subscription, replaySince }) => { + await replayReconnectHistoryPages({ + subscription, + since: replaySince, + until: now, + isActive: () => subscriptions.get(subId) === subscription, + requestHistory, + }); + }, + ), + Promise.all(replayRequests.map(({ reconnectEose }) => reconnectEose)), + ]); + } finally { + for (const { subscription, resolveReconnectEose } of replayRequests) { + if (subscription.resolveReconnectEose === resolveReconnectEose) { + subscription.resolveReconnectEose = undefined; + } + resolveReconnectEose(); + } + } }