From 4477f4ec7ad8b7d6e78b336110ad02d8a1786d96 Mon Sep 17 00:00:00 2001 From: Pinky <44b8e82baa6e0e254e0208d68f335c283c94e7b78dd1fa10d5a49d3f13dd0435@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 18:32:32 -0600 Subject: [PATCH 1/5] test(desktop): reproduce rate-limited replay flap Co-authored-by: Pinky <44b8e82baa6e0e254e0208d68f335c283c94e7b78dd1fa10d5a49d3f13dd0435@buzz.block.builderlab.xyz> Signed-off-by: Pinky <44b8e82baa6e0e254e0208d68f335c283c94e7b78dd1fa10d5a49d3f13dd0435@buzz.block.builderlab.xyz> --- desktop/src/testing/e2eBridge.ts | 14 +++++ desktop/tests/e2e/relay-reconnect.spec.ts | 68 +++++++++++++++++++++++ 2 files changed, 82 insertions(+) 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"], From e825c9ceef872560184aecf49b89e37ab5affd90 Mon Sep 17 00:00:00 2001 From: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 18:58:11 -0600 Subject: [PATCH 2/5] fix(desktop): keep the authenticated socket when reconnect backfill is rate-limited A CLOSED rate-limited: on a paged history REQ during reconnect replay rejected the history promise, escaped replayLiveSubscriptions, and made the session resetConnection() a healthy, authenticated socket. The client then redialed straight into the same rate-limit window: the 'briefly connected -> can't reach the relay' flap loop users hit on v0.5.5. Contain backfill failures inside the replay: retry each subscription's paged backfill behind the rate-limit gate up to PAGE_REPLAY_MAX_ATTEMPTS, then degrade to live-only for this connection. Socket health no longer depends on backfill success. The missed window is not lost: the replay cursor (lastSeenCreatedAt) only advances on delivered events, so the next reconnect replays it. Red test by Pinky (previous commit) proved the double-dial on main; it passes unchanged with this fix. New unit tests cover containment, gate-aware retry, and abort on subscription replacement. Co-authored-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> Signed-off-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> --- .../shared/api/relayReconnectReplay.test.mjs | 124 ++++++++++++++++++ .../src/shared/api/relayReconnectReplay.ts | 52 +++++++- 2 files changed, 169 insertions(+), 7 deletions(-) diff --git a/desktop/src/shared/api/relayReconnectReplay.test.mjs b/desktop/src/shared/api/relayReconnectReplay.test.mjs index 59253a459f..0466ffc198 100644 --- a/desktop/src/shared/api/relayReconnectReplay.test.mjs +++ b/desktop/src/shared/api/relayReconnectReplay.test.mjs @@ -3,6 +3,7 @@ import test from "node:test"; import { buildReconnectReplayFilter, + PAGE_REPLAY_MAX_ATTEMPTS, replayLiveSubscriptions, REPLAY_BATCH_SIZE, shouldPageReconnectReplay, @@ -600,6 +601,129 @@ 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", + ); +}); + // ── Teardown ────────────────────────────────────────────────────────────────── test("teardown — restore Date.now", () => { diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts index 74b752f666..938d398c91 100644 --- a/desktop/src/shared/api/relayReconnectReplay.ts +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -13,6 +13,21 @@ 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 sub's `lastSeenCreatedAt` is untouched by the failed backfill, so the + * missed window is retried on the next reconnect replay. + */ +export const PAGE_REPLAY_MAX_ATTEMPTS = 3; + /** * Maximum live subscriptions sent per relay REQ burst during reconnect. * @@ -237,13 +252,36 @@ 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. The next + // reconnect replays the same window because `lastSeenCreatedAt` only + // advances on delivered events. + for (let attempt = 1; attempt <= PAGE_REPLAY_MAX_ATTEMPTS; attempt++) { + try { + await replayReconnectHistoryPages({ + subscription, + since: replaySince, + until: now, + isActive: () => subscriptions.get(subId) === subscription, + requestHistory, + }); + 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; + } + } }, ); } From b70a6716d36cc84e55aa433a9087b36a418de8fd Mon Sep 17 00:00:00 2001 From: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 20:54:59 -0600 Subject: [PATCH 3/5] fix(desktop): pin the replay floor until backfill completes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review blocker on #4990: after backfill retries exhaust, events on the restored live REQ still advance lastSeenCreatedAt, so the next reconnect computed replaySince from the newer cursor and permanently skipped the unresolved older window — trading the visible flap for silent message loss. Pin pendingReplaySince on the subscription when paging starts; clear it only when a backfill pass completes. Replay windows start from min(pinned floor, cursor window), so an exhausted backfill followed by live traffic still requests the original missed window on the next reconnect, and a completed backfill returns control to the cursor. Regression test follows the review's exact scenario: cursor=1000, all attempts rate-limited, live event at 2100 advances the cursor via prepareSubscriptionEvent, next replay must request since=995, and a completed pass must clear the floor. Co-authored-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> Signed-off-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> --- desktop/src/shared/api/relayClientShared.ts | 11 +++ .../shared/api/relayReconnectReplay.test.mjs | 77 +++++++++++++++++++ .../src/shared/api/relayReconnectReplay.ts | 27 +++++-- 3 files changed, 109 insertions(+), 6 deletions(-) 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 0466ffc198..4edeeea8e9 100644 --- a/desktop/src/shared/api/relayReconnectReplay.test.mjs +++ b/desktop/src/shared/api/relayReconnectReplay.test.mjs @@ -9,6 +9,7 @@ import { shouldPageReconnectReplay, } from "./relayReconnectReplay.ts"; import { buildChannelFilter } from "./relayChannelFilters.ts"; +import { prepareSubscriptionEvent } from "./relayClosedRecovery.ts"; // ── Fake-timer + Date.now setup for gate tests ──────────────────────────────── @@ -724,6 +725,82 @@ test("backfill retry aborts when the subscription was replaced", async () => { ); }); +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", + ); +}); + // ── Teardown ────────────────────────────────────────────────────────────────── test("teardown — restore Date.now", () => { diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts index 938d398c91..ebecf510c4 100644 --- a/desktop/src/shared/api/relayReconnectReplay.ts +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -23,8 +23,9 @@ export const RECONNECT_REPLAY_PAGE_CONCURRENCY = 4; * 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 sub's `lastSeenCreatedAt` is untouched by the failed backfill, so the - * missed window is retried on the next reconnect replay. + * 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; @@ -182,13 +183,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); @@ -256,9 +265,14 @@ export async function replayLiveSubscriptions({ // 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. The next - // reconnect replays the same window because `lastSeenCreatedAt` only - // advances on delivered events. + // 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 { await replayReconnectHistoryPages({ @@ -268,6 +282,7 @@ export async function replayLiveSubscriptions({ isActive: () => subscriptions.get(subId) === subscription, requestHistory, }); + subscription.pendingReplaySince = undefined; return; } catch (error) { console.warn( From c493d378bcfd8809cb79f04c88bce39faac66616 Mon Sep 17 00:00:00 2001 From: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 21:01:57 -0600 Subject: [PATCH 4/5] fix(desktop): only clear the replay floor on genuine backfill completion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-review blocker: replayReconnectHistoryPages returned the same void for a completed window and for an abort due to a superseded connection. The caller cleared pendingReplaySince after both, so a stale connection A returning mid-flight could erase the floor that superseding connection B still needs — reopening the silent-gap risk on the shared subscription object. Return a completion boolean (false on stale abort, true on a genuinely finished window) and clear the floor only when true. Regression test supersedes the connection while a history REQ is in flight (subscription object kept alive, re-registered under a new subId) and asserts paging stops and the floor stays pinned. Co-authored-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> Signed-off-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> --- .../shared/api/relayReconnectReplay.test.mjs | 41 +++++++++++++++++++ .../src/shared/api/relayReconnectReplay.ts | 29 +++++++++---- 2 files changed, 63 insertions(+), 7 deletions(-) diff --git a/desktop/src/shared/api/relayReconnectReplay.test.mjs b/desktop/src/shared/api/relayReconnectReplay.test.mjs index 4edeeea8e9..e78cc3ba01 100644 --- a/desktop/src/shared/api/relayReconnectReplay.test.mjs +++ b/desktop/src/shared/api/relayReconnectReplay.test.mjs @@ -801,6 +801,47 @@ test("exhausted backfill pins the floor: next replay still requests the original ); }); +test("in-flight stale abort keeps the pinned floor for the superseding connection", async () => { + // Race from re-review of b70a6716d: connection A's paging aborts because it + // was superseded while a history REQ was in flight. That abort must NOT be + // treated as completion — the same subscription object carries the pinned + // floor that connection B's replay still needs. + 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++; + // Connection A is superseded while the REQ is in flight: the sub is + // re-registered by connection B under a new subId. The object stays + // alive; only A's map entry disappears. + subscriptions.delete("live-1"); + subscriptions.set("live-1b", subscription); + // A full page would otherwise continue paging — the post-await + // isActive() check must abort instead. + return eventRange("full", 1001, 500); + }, + }); + + assert.equal(historyCalls, 1, "stale connection must stop paging"); + assert.equal( + subscription.pendingReplaySince, + 995, + "a stale 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 ebecf510c4..350df3f23f 100644 --- a/desktop/src/shared/api/relayReconnectReplay.ts +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -94,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, @@ -106,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( @@ -121,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({ @@ -275,14 +286,18 @@ export async function replayLiveSubscriptions({ subscription.pendingReplaySince = replaySince; for (let attempt = 1; attempt <= PAGE_REPLAY_MAX_ATTEMPTS; attempt++) { try { - await replayReconnectHistoryPages({ + const completed = await replayReconnectHistoryPages({ subscription, since: replaySince, until: now, isActive: () => subscriptions.get(subId) === subscription, requestHistory, }); - subscription.pendingReplaySince = undefined; + // 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( From 6cebdfbd6b2daf99c9537aab426f9cf29c7a9564 Mon Sep 17 00:00:00 2001 From: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> Date: Wed, 5 Aug 2026 21:08:56 -0600 Subject: [PATCH 5/5] fix(desktop): abort stale-generation backfill passes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-review of c493d378b: production connection supersession bumps the generation while the subscription keeps its key AND object in the map, so the paging guard 'subscriptions.get(subId) === subscription' stayed true for a superseded connection. Its pass could run to completion and clear the pendingReplaySince floor the superseding connection needs. Combine the outer generation guard with the identity guard in the paging isActive. The regression now models real supersession: the generation flips false mid-flight while key and object survive (asserted as a precondition); paging must stop after one call and the floor must stay pinned. Against the identity-only guard this test does not merely fail — it loops forever, which is how the gap could silently persist in production. Co-authored-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> Signed-off-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz> --- .../shared/api/relayReconnectReplay.test.mjs | 32 ++++++++++++------- .../src/shared/api/relayReconnectReplay.ts | 9 +++++- 2 files changed, 28 insertions(+), 13 deletions(-) diff --git a/desktop/src/shared/api/relayReconnectReplay.test.mjs b/desktop/src/shared/api/relayReconnectReplay.test.mjs index e78cc3ba01..c175272885 100644 --- a/desktop/src/shared/api/relayReconnectReplay.test.mjs +++ b/desktop/src/shared/api/relayReconnectReplay.test.mjs @@ -802,10 +802,12 @@ test("exhausted backfill pins the floor: next replay still requests the original }); test("in-flight stale abort keeps the pinned floor for the superseding connection", async () => { - // Race from re-review of b70a6716d: connection A's paging aborts because it - // was superseded while a history REQ was in flight. That abort must NOT be - // treated as completion — the same subscription object carries the pinned - // floor that connection B's replay still needs. + // 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 = { @@ -816,29 +818,35 @@ test("in-flight stale abort keeps the pinned floor for the superseding connectio }; 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 sub is - // re-registered by connection B under a new subId. The object stays - // alive; only A's map entry disappears. - subscriptions.delete("live-1"); - subscriptions.set("live-1b", subscription); + // 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 - // isActive() check must abort instead. + // combined guard must abort instead. return eventRange("full", 1001, 500); }, }); - assert.equal(historyCalls, 1, "stale connection must stop paging"); + 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 abort must not clear the floor the new connection needs", + "a stale-generation abort must not clear the floor the new connection needs", ); }); diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts index 350df3f23f..cb962ce846 100644 --- a/desktop/src/shared/api/relayReconnectReplay.ts +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -290,7 +290,14 @@ export async function replayLiveSubscriptions({ subscription, since: replaySince, until: now, - isActive: () => subscriptions.get(subId) === subscription, + // 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