fix(desktop): stop rate-limited reconnect backfill from tearing down the authenticated socket (#4990)

## Problem

Users on v0.5.5 report "Can't reach the relay" toggling with brief
"connected" flashes (field reports; also the macOS confirmation in
#4908). #4737 closed the stuck-reconnect gaps; this is the opposite
failure: the client redials fine, but then kills its own healthy socket.

Mechanism (all on `main`):

1. AUTH succeeds → session emits `connected`
(`relayClientSession.ts:583`), then awaits `replayLiveSubscriptions()`.
2. Paged channel backfill issues history REQs
(`relayReconnectReplay.ts`, page limit 500).
3. A `CLOSED rate-limited:` on a **history** REQ arms the rate-limit
gate but still rejects the history promise
(`relayClosedRecovery.ts:38-51`).
4. The rejection escapes `replayLiveSubscriptions()` →
`resetConnection()` tears down the authenticated socket.
5. Reconnect → AUTH OK → replay rate-limited again → loop. Each
iteration re-spends the rate-limit budget, so the loop is
self-sustaining.

## Fix

Contain backfill failures inside the replay. Each subscription's paged
backfill now retries behind the rate-limit gate up to
`PAGE_REPLAY_MAX_ATTEMPTS` (3), then degrades to live-only **for this
connection**. Socket health no longer depends on backfill success.
Nothing is lost: the replay cursor (`lastSeenCreatedAt`) only advances
on delivered events, so the next reconnect replays the same missed
window.

## Red/green proof

- Commit 1 (Pinky): e2e injecting `CLOSED rate-limited:` into the
mid-replay history REQ — **red on main** (expected 1 reconnect dial,
observed 2; connected-flash then teardown).
- Commit 2 (this fix): same test **green unchanged** — one dial, state
stays `connected` through the rate-limit hint plus the next backoff
window.

Why existing coverage missed it: the prior rate-limit e2e pre-armed the
gate *before* replay (replay politely waits), and the CLOSED-injection
test targeted a *live* subscription (which has its own retry path).
Nobody injected back-pressure from the history REQ itself.

## Verification

- `pnpm test`: 4374/4374 pass.
- `playwright test tests/e2e/relay-reconnect.spec.ts`: 14/14 pass,
including the new spec.
- `tsc --noEmit` clean; Biome clean on touched files (pre-existing
warnings on main in `personaCatalogRelay.test.mjs` / `terminal.css`
untouched).

## Not addressed here (follow-ups from the same field reports)

- AUTH terminal latch is too aggressive for relay-internal `error:`
rejections (3 strikes during a relay bad window → stuck until
click/relaunch; #4908).
- Server-side: `relay.drainJitterMs` (#4542) defaults to 0 — enabling it
on the hosted relay removes the deploy thundering herd that triggers
these rate-limit storms.

---------

Signed-off-by: Pinky <44b8e82baa6e0e254e0208d68f335c283c94e7b78dd1fa10d5a49d3f13dd0435@buzz.block.builderlab.xyz>
Signed-off-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz>
Co-authored-by: Pinky <44b8e82baa6e0e254e0208d68f335c283c94e7b78dd1fa10d5a49d3f13dd0435@buzz.block.builderlab.xyz>
Co-authored-by: Brain <21994759fc7a6fa6b965551d35cfd7897d262f2495467f2d78694ddcfa6a5c7e@buzz.block.builderlab.xyz>
This commit is contained in:
Wes
2026-08-06 08:39:17 -06:00
committed by GitHub
parent 96ae141763
commit 19b41e9c8e
5 changed files with 431 additions and 13 deletions

View File

@@ -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;
};

View File

@@ -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", () => {

View File

@@ -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<RelayEvent[]>;
}) {
}): Promise<boolean> {
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;
}
}
},
);
}

View File

@@ -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<number, MockSocket>();
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()) {

View File

@@ -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"],