diff --git a/desktop/src/shared/api/relayClientShared.ts b/desktop/src/shared/api/relayClientShared.ts index 9108f7b6d7..3952b6af6f 100644 --- a/desktop/src/shared/api/relayClientShared.ts +++ b/desktop/src/shared/api/relayClientShared.ts @@ -60,6 +60,17 @@ type LiveSubscription = { onEvent: (event: RelayEvent) => void; resolveReady?: () => void; lastSeenCreatedAt?: number; + /** + * Lower bound of a reconnect backfill window that has not yet completed. + * + * Events on the restored live REQ advance `lastSeenCreatedAt` regardless of + * backfill success, so after an exhausted backfill the cursor alone would + * make the next reconnect skip the unresolved older window — silent message + * loss. This floor is pinned when paging starts and cleared only when a + * backfill pass completes; the next replay starts from + * `min(pendingReplaySince, cursor window)`. + */ + pendingReplaySince?: number; closedRetryAttempt?: number; closedRetryTimeout?: number; }; diff --git a/desktop/src/shared/api/relayReconnectReplay.test.mjs b/desktop/src/shared/api/relayReconnectReplay.test.mjs index 59253a459f..c175272885 100644 --- a/desktop/src/shared/api/relayReconnectReplay.test.mjs +++ b/desktop/src/shared/api/relayReconnectReplay.test.mjs @@ -3,11 +3,13 @@ import test from "node:test"; import { buildReconnectReplayFilter, + PAGE_REPLAY_MAX_ATTEMPTS, replayLiveSubscriptions, REPLAY_BATCH_SIZE, shouldPageReconnectReplay, } from "./relayReconnectReplay.ts"; import { buildChannelFilter } from "./relayChannelFilters.ts"; +import { prepareSubscriptionEvent } from "./relayClosedRecovery.ts"; // ── Fake-timer + Date.now setup for gate tests ──────────────────────────────── @@ -600,6 +602,254 @@ test("batch-1 arms gate mid-replay: batch-2 is withheld until gate expires", asy ); }); +// ── Backfill failure containment ───────────────────────────────────────────── + +test("history backfill rejection never escapes replayLiveSubscriptions", async () => { + resetGate(0); + const filter = buildChannelFilter("channel-1", 50); + const subscriptions = new Map([ + [ + "live-1", + { + mode: "live", + filter, + onEvent: () => {}, + lastSeenCreatedAt: 1000, + }, + ], + ]); + + let historyCalls = 0; + // Must resolve — a rejection here is the socket-killing flap regression. + await replayLiveSubscriptions({ + subscriptions, + now: 2000, + sendRaw: async () => {}, + requestHistory: async () => { + historyCalls++; + throw new Error("rate-limited: quota exceeded; retry in 4s"); + }, + }); + + assert.equal( + historyCalls, + PAGE_REPLAY_MAX_ATTEMPTS, + "backfill must retry a bounded number of times, then degrade", + ); +}); + +test("backfill retry waits out the armed gate, then succeeds", async () => { + resetGate(0); + const delivered = []; + const filter = buildChannelFilter("channel-1", 50); + const subscriptions = new Map([ + [ + "live-1", + { + mode: "live", + filter, + onEvent: (event) => delivered.push(event), + lastSeenCreatedAt: 1000, + }, + ], + ]); + + const attemptAtMs = []; + let armGate; + const gateArmed = new Promise((resolve) => { + armGate = resolve; + }); + const replayPromise = replayLiveSubscriptions({ + subscriptions, + now: 2000, + sendRaw: async () => {}, + requestHistory: async () => { + attemptAtMs.push(fakeNow); + if (attemptAtMs.length === 1) { + // Mirror relayClosedRecovery: the CLOSED handler arms the gate + // before rejecting the history promise. + activateRateLimit(4); + armGate(); + throw new Error("rate-limited: quota exceeded; retry in 4s"); + } + return [event("recovered", 1500)]; + }, + }); + + // Wait until the gate is actually armed, then expire it. The retry loop is + // (or will be) suspended in waitForRateLimit; expiring the gate releases it. + await gateArmed; + tickTo(4_001); + await replayPromise; + + assert.equal(attemptAtMs.length, 2, "one failure, one retry"); + assert.ok( + attemptAtMs[1] >= 4_001, + "retry must not fire before the rate-limit gate expires", + ); + assert.deepEqual( + delivered.map((e) => e.id), + ["recovered"], + "the retried backfill must deliver its events", + ); +}); + +test("backfill retry aborts when the subscription was replaced", async () => { + resetGate(0); + const filter = buildChannelFilter("channel-1", 50); + const subscription = { + mode: "live", + filter, + onEvent: () => {}, + lastSeenCreatedAt: 1000, + }; + const subscriptions = new Map([["live-1", subscription]]); + + let historyCalls = 0; + await replayLiveSubscriptions({ + subscriptions, + now: 2000, + sendRaw: async () => {}, + requestHistory: async () => { + historyCalls++; + // Simulate the subscription being torn down while the REQ is in flight. + subscriptions.delete("live-1"); + throw new Error("rate-limited: quota exceeded; retry in 4s"); + }, + }); + + assert.equal( + historyCalls, + 1, + "no retry may target a subscription that no longer exists", + ); +}); + +test("exhausted backfill pins the floor: next replay still requests the original window after live events advance the cursor", async () => { + // The blocking review scenario on PR #4990: cursor=1000, all backfill + // attempts fail, a live event at 2100 then advances lastSeenCreatedAt via + // prepareSubscriptionEvent. Without the pinned floor, the next reconnect + // would start near 2095 and silently skip 1001..1999. + resetGate(0); + const filter = buildChannelFilter("channel-1", 50); + const subscription = { + mode: "live", + filter, + onEvent: () => {}, + lastSeenCreatedAt: 1000, + }; + const subscriptions = new Map([["live-1", subscription]]); + + // Reconnect 1: every backfill attempt is rate-limited. + await replayLiveSubscriptions({ + subscriptions, + now: 2000, + sendRaw: async () => {}, + requestHistory: async () => { + throw new Error("rate-limited: quota exceeded; retry in 4s"); + }, + }); + assert.equal( + subscription.pendingReplaySince, + 995, + "exhausted backfill must pin the unresolved window's lower bound", + ); + + // A live event arrives through the normal cursor path. + prepareSubscriptionEvent(subscription, event("live-newer", 2100)); + assert.equal(subscription.lastSeenCreatedAt, 2100); + + // Reconnect 2: backfill now succeeds. It must request the ORIGINAL window. + const historyFilters = []; + await replayLiveSubscriptions({ + subscriptions, + now: 2200, + sendRaw: async () => {}, + requestHistory: async (filter) => { + historyFilters.push(filter); + return []; + }, + }); + + assert.equal(historyFilters.length, 1); + assert.equal( + historyFilters[0].since, + 995, + "replay must start from the pinned floor, not the advanced cursor", + ); + assert.equal( + subscription.pendingReplaySince, + undefined, + "a completed backfill must clear the pinned floor", + ); + + // Reconnect 3: with the floor cleared, replay returns to the cursor. + const laterFilters = []; + await replayLiveSubscriptions({ + subscriptions, + now: 2300, + sendRaw: async () => {}, + requestHistory: async (filter) => { + laterFilters.push(filter); + return []; + }, + }); + assert.equal( + laterFilters[0].since, + 2095, + "after recovery the cursor governs again", + ); +}); + +test("in-flight stale abort keeps the pinned floor for the superseding connection", async () => { + // Race from re-review of b70a6716d/c493d378b: production supersession bumps + // the connection GENERATION while the same subscription key and object + // survive in the map. The identity guard alone stays true, so only the + // combined guard (outer isActive && identity) aborts the stale pass. That + // abort must NOT count as completion — the pinned floor belongs to the + // superseding connection's replay. + resetGate(0); + const filter = buildChannelFilter("channel-1", 50); + const subscription = { + mode: "live", + filter, + onEvent: () => {}, + lastSeenCreatedAt: 1000, + }; + const subscriptions = new Map([["live-1", subscription]]); + + let generationActive = true; + let historyCalls = 0; + await replayLiveSubscriptions({ + subscriptions, + now: 2000, + sendRaw: async () => {}, + isActive: () => generationActive, + requestHistory: async () => { + historyCalls++; + // Connection A is superseded while the REQ is in flight: the generation + // advances, but the subscription keeps its key AND object identity — + // exactly what production supersession does. + generationActive = false; + // A full page would otherwise continue paging — the post-await + // combined guard must abort instead. + return eventRange("full", 1001, 500); + }, + }); + + assert.equal(historyCalls, 1, "stale generation must stop paging"); + assert.equal( + subscriptions.get("live-1"), + subscription, + "precondition: key and object survive supersession untouched", + ); + assert.equal( + subscription.pendingReplaySince, + 995, + "a stale-generation abort must not clear the floor the new connection needs", + ); +}); + // ── Teardown ────────────────────────────────────────────────────────────────── test("teardown — restore Date.now", () => { diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts index 74b752f666..cb962ce846 100644 --- a/desktop/src/shared/api/relayReconnectReplay.ts +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -13,6 +13,22 @@ const RECONNECT_REPLAY_SKEW_SECS = 5; export const RECONNECT_REPLAY_PAGE_LIMIT = 500; export const RECONNECT_REPLAY_PAGE_CONCURRENCY = 4; +/** + * Maximum attempts for one subscription's paged history backfill. + * + * Backfill failures must never escape `replayLiveSubscriptions`: by the time + * paging starts, every live REQ has already been re-established on a healthy, + * authenticated socket. Letting a history rejection propagate makes the + * session tear that socket down (`resetConnection`) and reconnect straight + * into the same rate-limit window — the "briefly connected → can't reach the + * relay" flap loop. Instead each sub retries behind the rate-limit gate a + * bounded number of times, then degrades to live-only for this connection. + * The window's lower bound is pinned in `pendingReplaySince` while unresolved + * (live events advance `lastSeenCreatedAt` regardless of backfill success), + * so the next reconnect still requests the missed window. + */ +export const PAGE_REPLAY_MAX_ATTEMPTS = 3; + /** * Maximum live subscriptions sent per relay REQ burst during reconnect. * @@ -78,6 +94,16 @@ export function shouldPageReconnectReplay(filter: RelaySubscriptionFilter) { ); } +/** + * Page one subscription's missed-window history. + * + * Returns `true` only when the window was genuinely completed (short page or + * boundary reached). Returns `false` when the pass aborted because the + * connection went stale (`isActive()` false) — callers must NOT treat that as + * completion: the same subscription object is shared with the superseding + * connection, and clearing its pinned `pendingReplaySince` on a stale abort + * would erase the floor the new connection still needs. + */ export async function replayReconnectHistoryPages({ subscription, since, @@ -90,11 +116,11 @@ export async function replayReconnectHistoryPages({ until: number; isActive: () => boolean; requestHistory: (filter: RelaySubscriptionFilter) => Promise; -}) { +}): Promise { let pageUntil = until; while (pageUntil >= since) { - if (!isActive()) return; + if (!isActive()) return false; const events = await requestHistory( buildReconnectReplayFilter( @@ -105,17 +131,18 @@ export async function replayReconnectHistoryPages({ ), ); - if (!isActive()) return; + if (!isActive()) return false; for (const event of events) subscription.onEvent(event); - if (events.length < RECONNECT_REPLAY_PAGE_LIMIT) return; + if (events.length < RECONNECT_REPLAY_PAGE_LIMIT) return true; const oldestCreatedAt = events[0]?.created_at; - if (oldestCreatedAt === undefined || oldestCreatedAt <= since) return; + if (oldestCreatedAt === undefined || oldestCreatedAt <= since) return true; pageUntil = oldestCreatedAt < pageUntil ? oldestCreatedAt : oldestCreatedAt - 1; } + return true; } export async function replayLiveSubscriptions({ @@ -167,13 +194,21 @@ export async function replayLiveSubscriptions({ entry[1].mode === "live", ) .map(([subId, subscription]) => { - const replaySince = + const cursorSince = subscription.lastSeenCreatedAt === undefined ? undefined : Math.max( 0, subscription.lastSeenCreatedAt - RECONNECT_REPLAY_SKEW_SECS, ); + // A pinned floor from a previously failed backfill takes precedence + // over the cursor: live events kept advancing `lastSeenCreatedAt` + // while the older window stayed unresolved, and starting from the + // cursor would skip it permanently. + const replaySince = + cursorSince === undefined + ? subscription.pendingReplaySince + : Math.min(cursorSince, subscription.pendingReplaySince ?? Infinity); const shouldPageReplay = replaySince !== undefined && shouldPageReconnectReplay(subscription.filter); @@ -237,13 +272,53 @@ export async function replayLiveSubscriptions({ ), pageReplayConcurrency, async ({ subId, subscription, replaySince }) => { - await replayReconnectHistoryPages({ - subscription, - since: replaySince, - until: now, - isActive: () => subscriptions.get(subId) === subscription, - requestHistory, - }); + // Backfill is best-effort: a failure here (typically a `rate-limited:` + // CLOSED on a history REQ) must never escape to the session and tear + // down the healthy, authenticated socket carrying the live REQs — that + // is the connect→drop flap loop. Retry behind the gate a bounded number + // of times, then degrade to live-only for this connection. + // + // Pin the window's lower bound before the first attempt: events on the + // already-restored live REQ advance `lastSeenCreatedAt` independently + // of backfill success, so without the pin an exhausted backfill + // followed by one live event would make the next reconnect skip the + // unresolved window permanently. Cleared only on a completed pass. + subscription.pendingReplaySince = replaySince; + for (let attempt = 1; attempt <= PAGE_REPLAY_MAX_ATTEMPTS; attempt++) { + try { + const completed = await replayReconnectHistoryPages({ + subscription, + since: replaySince, + until: now, + // Both guards are required. The identity check catches the sub + // being torn down/replaced; the outer isActive() catches + // connection supersession, which bumps the generation while the + // SAME subscription key and object survive in the map — identity + // alone stays true and a stale pass could complete and clear the + // floor the superseding connection needs. + isActive: () => + isActive() && subscriptions.get(subId) === subscription, + requestHistory, + }); + // A stale-connection abort is NOT completion: the superseding + // connection shares this subscription object and still needs the + // pinned floor for its own replay. Only a genuinely completed + // window may release it. + if (completed) subscription.pendingReplaySince = undefined; + return; + } catch (error) { + console.warn( + `[reconnect replay] history backfill attempt ${attempt}/${PAGE_REPLAY_MAX_ATTEMPTS} failed for ${subId}:`, + error, + ); + if (attempt === PAGE_REPLAY_MAX_ATTEMPTS) return; + // The failed REQ's CLOSED handler arms the rate-limit gate before + // rejecting; wait for it (no-op when the failure wasn't back-pressure) + // and re-check that this replay's connection is still current. + if (isRateLimited()) await waitForRateLimit(); + if (subscriptions.get(subId) !== subscription || !isActive()) return; + } + } }, ); } diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 54238323ae..1987e00ff1 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -1211,6 +1211,8 @@ declare global { ) => void; /** Inject CLOSED into every active mock live subscription. */ __BUZZ_E2E_CLOSE_LIVE_SUBSCRIPTIONS__?: (reason: string) => number; + /** Queue CLOSED responses for channel history REQs. */ + __BUZZ_E2E_QUEUE_CHANNEL_HISTORY_CLOSES__?: (reasons: string[]) => void; __BUZZ_E2E_SET_STALL_WEBSOCKET_SENDS__?: (stall: boolean) => void; __BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__?: () => number; __BUZZ_E2E_RESTART_MOCK_WEBSOCKETS__?: () => number; @@ -2967,6 +2969,7 @@ const mockPersonaEvents: RelayEvent[] = []; let mockRelayMembers: RawRelayMember[] = []; const mockSockets = new Map(); const mockAuthResponses: Array<{ success: boolean; message: string }> = []; +const mockChannelHistoryCloses: string[] = []; let mockWebsocketUnavailable = false; const relayWebsocketConnectAttemptStarts: number[] = []; let mockWebsocketSendMutexWedged = false; @@ -9691,6 +9694,13 @@ function sendToMockSocket(args: { } const channelId = filter["#h"]?.[0]; + if (channelId && subId.startsWith("history-")) { + const closeReason = mockChannelHistoryCloses.shift(); + if (closeReason) { + sendWsText(socket.handler, ["CLOSED", subId, closeReason]); + return; + } + } if (!channelId) { // Aux-backfill filters (reactions/deletions) are `#e`-keyed with no // channel tag — serve them across all channel stores like the relay. @@ -9945,6 +9955,7 @@ export function maybeInstallE2eTauriMocks() { mockClosedChannelLiveSubscription = false; mockWebsocketUnavailable = false; mockAuthResponses.length = 0; + mockChannelHistoryCloses.length = 0; relayWebsocketConnectAttemptStarts.length = 0; deferredSendMessageLiveEchoes.length = 0; mockGlobalAgentConfig = config.mock?.globalAgentConfig @@ -10188,6 +10199,9 @@ export function maybeInstallE2eTauriMocks() { window.__BUZZ_E2E_QUEUE_AUTH_RESPONSES__ = (responses) => { mockAuthResponses.push(...responses); }; + window.__BUZZ_E2E_QUEUE_CHANNEL_HISTORY_CLOSES__ = (reasons) => { + mockChannelHistoryCloses.push(...reasons); + }; window.__BUZZ_E2E_CLOSE_LIVE_SUBSCRIPTIONS__ = (reason) => { let closed = 0; for (const socket of mockSockets.values()) { diff --git a/desktop/tests/e2e/relay-reconnect.spec.ts b/desktop/tests/e2e/relay-reconnect.spec.ts index ec3e87171d..289d8ef318 100644 --- a/desktop/tests/e2e/relay-reconnect.spec.ts +++ b/desktop/tests/e2e/relay-reconnect.spec.ts @@ -134,6 +134,19 @@ async function closeLiveSubscriptions( expect(closed).toBeGreaterThan(0); } +async function queueChannelHistoryCloses( + page: import("@playwright/test").Page, + reasons: string[], +) { + await page.evaluate((queued) => { + const queue = window.__BUZZ_E2E_QUEUE_CHANNEL_HISTORY_CLOSES__; + if (!queue) { + throw new Error("E2E channel history CLOSED seam is not installed."); + } + queue(queued); + }, reasons); +} + async function driveConnectionDegraded( page: import("@playwright/test").Page, state: "connected" | "reconnecting" | "stalled" | "disconnected", @@ -253,6 +266,61 @@ test("authenticated reconnect reports connected while replay is rate-limited", a await expect(page.getByTestId("sidebar-relay-unreachable")).toHaveCount(0); }); +test("rate-limited reconnect backfill does not tear down the authenticated socket", async ({ + page, +}) => { + await page.goto("/"); + await page.getByTestId("channel-general").click(); + await expect(page.getByTestId("chat-title")).toHaveText("general"); + + // Give the channel live subscription a replay cursor. On reconnect this + // causes a paged channel-history REQ in addition to restoring the live REQ. + await emitMockMessages(page, [ + { + content: `replay cursor ${Date.now()}`, + createdAt: Math.floor(Date.now() / 1_000), + }, + ]); + const attemptsBeforeReconnect = (await getMockWebsocketConnectAttempts(page)) + .length; + + // Inject back-pressure from the history REQ itself. This differs from a + // pre-armed gate and from CLOSED on the live subscription: AUTH has already + // succeeded and the socket is healthy when replay backfill is rejected. + await queueChannelHistoryCloses(page, [ + "rate-limited: quota exceeded; retry in 1s", + ]); + await disconnectMockWebsockets(page); + + await expect + .poll( + async () => + (await getMockWebsocketConnectAttempts(page)).length - + attemptsBeforeReconnect, + { timeout: 3_000 }, + ) + .toBe(1); + await expect + .poll(() => + page.evaluate(() => window.__BUZZ_E2E_GET_RELAY_CONNECTION_STATE__?.()), + ) + .toBe("connected"); + + // Hold through the rate-limit hint plus the next base-backoff window. The + // authenticated socket must remain the only reconnect attempt, rather than + // flashing connected and redialing after replay rejects. + await page.waitForTimeout(2_500); + expect( + (await getMockWebsocketConnectAttempts(page)).length - + attemptsBeforeReconnect, + ).toBe(1); + expect( + await page.evaluate(() => + window.__BUZZ_E2E_GET_RELAY_CONNECTION_STATE__?.(), + ), + ).toBe("connected"); +}); + test("service restart close resets accumulated backoff", async ({ page }) => { await installMockBridge(page, { websocketConnectErrors: ["down 1", "down 2", "down 3"],