diff --git a/desktop/src/shared/api/relayClientResumeCursor.test.mjs b/desktop/src/shared/api/relayClientResumeCursor.test.mjs new file mode 100644 index 0000000000..f9df922b48 --- /dev/null +++ b/desktop/src/shared/api/relayClientResumeCursor.test.mjs @@ -0,0 +1,519 @@ +// Exercise the production session: EOSE -> socket close -> resume cursor -> +// replayed REQ, and the close-code-driven reconnect spreading. +// The IPC transport and clock are fake; no native app or external relay is used. +import assert from "node:assert/strict"; +import { after, beforeEach, afterEach, test } from "node:test"; + +const originalNow = Date.now; +const originalRandom = Math.random; +const originalWindow = globalThis.window; +let now = 0; +let nextTimerId = 1; +const timers = new Map(); +const intervals = new Map(); +const writes = []; +const dials = []; +const repairRequests = []; +const clients = []; +let nextCallbackId = 1; +globalThis.window = { + setTimeout(fn, ms) { + const id = nextTimerId++; + timers.set(id, { fn, at: now + ms }); + return id; + }, + clearTimeout(id) { + timers.delete(id); + }, + // The stall watchdog's idle check; fired explicitly by the stall test. + setInterval(fn) { + const id = nextTimerId++; + intervals.set(id, fn); + return id; + }, + clearInterval(id) { + intervals.delete(id); + }, + __TAURI_INTERNALS__: { + transformCallback: () => nextCallbackId++, + async invoke(command, args) { + if (command === "plugin:websocket|connect") { + dials.push(now); + return 100 + dials.length; + } + if (command === "get_channel_reconnect_repair") { + // The HTTP repair stays in flight until the test settles it. + return new Promise((resolve) => repairRequests.push({ args, resolve })); + } + if (command === "plugin:websocket|send") { + writes.push({ socket: args.id, frame: JSON.parse(args.message.data) }); + return; + } + if (command === "plugin:websocket|disconnect") return; + assert.fail(`unexpected IPC: ${command}`); + }, + }, +}; +Date.now = () => now; +const { RelayClient } = await import("./relayClientSession.ts"); +const { resetRateLimitGate } = await import("./relayRateLimitGate.ts"); +const { buildChannelFilter } = await import("./relayChannelFilters.ts"); + +beforeEach(() => { + resetRateLimitGate(); + now = 0; + timers.clear(); + intervals.clear(); + writes.length = 0; + dials.length = 0; + repairRequests.length = 0; + Math.random = originalRandom; +}); +afterEach(() => { + for (const client of clients.splice(0)) client.disconnect(); + resetRateLimitGate(); + Math.random = originalRandom; + assert.equal(timers.size, 0, "session must release its timers"); +}); +after(() => { + Date.now = originalNow; + Math.random = originalRandom; + globalThis.window = originalWindow; +}); + +async function flush() { + for (let i = 0; i < 20; i++) await Promise.resolve(); +} +function deliver(client, frame) { + return client.handleWsMessage( + { type: "Text", data: JSON.stringify(frame) }, + client.connectionGeneration, + ); +} +function close(client, code) { + return client.handleWsMessage( + { type: "Close", data: { code, reason: "test" } }, + client.connectionGeneration, + ); +} +function reconnectDelayMs(client) { + const timer = timers.get(client.reconnectTimeout); + assert.ok(timer, "a reconnect must be scheduled"); + return timer.at - now; +} + +// A client with one live, non-paged subscription per filter. Only the first +// `eosed` of them receive EOSE on the first connection. +async function liveClient(filters, eosed = filters.length) { + const client = new RelayClient(); + client.wsId = 7; // Already authenticated; subscribe and dispatch stay real. + clients.push(client); + const ready = filters.map((filter) => client.subscribeLive(filter, () => {})); + for (const pending of ready) void pending.catch(() => {}); + await flush(); + for (let i = 0; i < filters.length; i++) { + now += 250; + for (const [id, timer] of [...timers]) { + if (timer.at <= now) { + timers.delete(id); + timer.fn(); + } + } + await flush(); + } + const reqs = writes.filter(({ frame }) => frame[0] === "REQ"); + assert.equal(reqs.length, filters.length); + for (const { frame } of reqs.slice(0, eosed)) { + await deliver(client, ["EOSE", frame[1]]); + } + const ids = reqs.map(({ frame }) => frame[1]); + // Resolve any outstanding readiness without leaving timers behind. + for (const [id, timer] of [...timers]) { + timers.delete(id); + timer.fn(); + } + await flush(); + return { client, ids, ready }; +} + +const quiet = { + kinds: [9, 5, 7], + "#h": ["channel-0"], + since: 123, + limit: 1000, +}; + +test("a close after EOSE stamps liveUntil at the last frame and the replay resumes from it", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([quiet]); + const subscription = client.subscriptions.get(ids[0]); + assert.equal(subscription.liveUntil, undefined); + + // EOSE (the last frame) arrived at 2_000_250; the close is processed later. + now = 2_050_000; + await close(client, 1000); + assert.equal(subscription.liveUntil, 2_000); + window.clearTimeout(client.reconnectTimeout); + client.reconnectTimeout = null; + + // Reconnect late enough that liveUntil, not the silent-drop window, binds. + now = 5_000_000; + client.wsId = 8; + writes.length = 0; + await client.replayLiveSubscriptions(); + assert.deepEqual( + writes.map(({ frame }) => frame), + [["REQ", ids[0], { ...quiet, since: 2_000 - 960, limit: 1000 }]], + ); +}); + +test("a quiet subscription whose fan-out was dropped still replays the drop on a prompt reconnect", async () => { + // The relay can drop a fan-out frame under backpressure and keep the socket + // open, so pings after the drop do not prove the subscription missed nothing. + now = 10_000_000; + const { client, ids } = await liveClient([quiet]); + client.stallWatchdog.start(); + const droppedAt = 10_100; // stored, but its fan-out frame never arrived + for (now = 10_030_000; now <= 12_000_000; now += 30_000) { + await client.handleWsMessage( + { type: "Ping", data: [] }, + client.connectionGeneration, + ); + } + now -= 30_000; + await close(client, 1006); + const subscription = client.subscriptions.get(ids[0]); + assert.equal(subscription.liveUntil, Math.floor(now / 1_000)); + assert.ok( + subscription.liveUntil - 960 > droppedAt, + "precondition: the lookback alone would skip the drop", + ); + window.clearTimeout(client.reconnectTimeout); + client.reconnectTimeout = null; + + now += 2_000; + client.wsId = 8; + writes.length = 0; + await client.replayLiveSubscriptions(); + const { since } = writes[0].frame[2]; + assert.equal(since, Math.floor(now / 1_000) - 3_600); + assert.ok( + droppedAt >= since, + `drop at ${droppedAt} must satisfy since ${since}`, + ); +}); + +test("a subscription that never reached EOSE keeps replaying its original filter", async () => { + now = 2_000_000; + const second = { ...quiet, "#h": ["channel-1"] }; + const { client, ids } = await liveClient([quiet, second], 1); + now = 2_050_000; + await close(client, 1000); + assert.equal(client.subscriptions.get(ids[0]).liveUntil, 2_000); + assert.equal(client.subscriptions.get(ids[1]).liveUntil, undefined); + window.clearTimeout(client.reconnectTimeout); + client.reconnectTimeout = null; + + now = 5_000_000; + client.wsId = 8; + writes.length = 0; + await client.replayLiveSubscriptions(); + const byId = new Map(writes.map(({ frame }) => [frame[1], frame[2]])); + assert.equal(byId.get(ids[0]).since, 2_000 - 960); + assert.deepEqual(byId.get(ids[1]), second); +}); + +test("a relay CLOSED after EOSE means the socket close must not stamp the subscription", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([quiet]); + await deliver(client, ["CLOSED", ids[0], "error: database error"]); + now = 2_050_000; + await close(client, 1000); + assert.equal(client.subscriptions.get(ids[0]).liveUntil, undefined); +}); + +test("a stalled socket stamps liveUntil at its last inbound frame, not at detection", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([quiet]); + client.stallWatchdog.start(); // lastInboundAt = 2_000_000 + now = 2_030_000; + await deliver(client, ["NOTICE", "hello"]); // recordInbound via handleWsMessage + now = 2_100_000; + client.resetConnection(new Error("Relay socket stalled")); + assert.equal(client.subscriptions.get(ids[0]).liveUntil, 2_030); +}); + +test("the real stall path stamps liveUntil at the pre-sleep inbound frame", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([quiet]); + client.stallWatchdog.start(); + now = 2_030_000; + await deliver(client, ["NOTICE", "hello"]); // last inbound frame + // The machine sleeps for an hour; the first overdue idle tick is the wake. + now = 2_030_000 + 3_600_000; + assert.equal(intervals.size, 1, "the production watchdog interval is armed"); + const [tick] = intervals.values(); + tick(); // checkIdle -> stop() -> production onStall -> resetConnection + const subscription = client.subscriptions.get(ids[0]); + assert.equal(client.connectionStateEmitter.get(), "stalled"); + assert.equal( + subscription.liveUntil, + 2_030, + "stamped at the last inbound frame, not at wake time", + ); + window.clearTimeout(client.reconnectTimeout); + client.reconnectTimeout = null; + + // A message stored early in the gap must be inside the reconnect REQ. + const earlyInGap = 2_030 + 120; + client.wsId = 8; + writes.length = 0; + await client.replayLiveSubscriptions(); + const { since } = writes[0].frame[2]; + assert.ok( + earlyInGap >= since, + `message at ${earlyInGap} must satisfy REQ since ${since}`, + ); +}); + +// After a long pause (timers suspended) the native layer's terminal notice can +// be dequeued before the overdue idle tick. It is not proof the socket was +// alive, so it must not refresh the liveness clock that bounds `liveUntil`. +for (const [name, terminal] of [ + ["Close 1006", { type: "Close", data: { code: 1006, reason: "abnormal" } }], + ["Close without a frame", { type: "Close", data: null }], + ["Error", { type: "Error", data: "connection reset" }], +]) { + test(`${name} after a long pause does not count as liveness`, async () => { + now = 2_000_000; + const { client, ids } = await liveClient([quiet]); + client.stallWatchdog.start(); + now = 2_030_000; + await deliver(client, ["NOTICE", "hello"]); // last real relay frame + now = 2_030_000 + 3_600_000; // an hour asleep; the idle tick has not run + await client.handleWsMessage(terminal, client.connectionGeneration); + assert.equal(client.subscriptions.get(ids[0]).liveUntil, 2_030); + window.clearTimeout(client.reconnectTimeout); + client.reconnectTimeout = null; + + const earlyInGap = 2_030 + 120; + client.wsId = 8; + writes.length = 0; + await client.replayLiveSubscriptions(); + const { since } = writes[0].frame[2]; + assert.ok( + earlyInGap >= since, + `message at ${earlyInGap} must satisfy REQ since ${since}`, + ); + }); +} + +test("a peer Ping is liveness; the close after it is not", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([quiet]); + client.stallWatchdog.start(); + now = 2_060_000; + await client.handleWsMessage( + { type: "Ping", data: [] }, + client.connectionGeneration, + ); + now = 2_060_000 + 3_600_000; + await close(client, 1006); + assert.equal(client.subscriptions.get(ids[0]).liveUntil, 2_060); +}); + +test("a frame queued through a sleep does not count as liveness", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([quiet]); + client.stallWatchdog.start(); + now = 2_030_000; + await deliver(client, ["NOTICE", "hello"]); // last frame before the sleep + // A ping and a NOTICE were queued just before the sleep; the renderer + // processes them on wake, ahead of the overdue idle tick and the close. + now = 2_030_000 + 3_600_000; + await client.handleWsMessage( + { type: "Ping", data: [] }, + client.connectionGeneration, + ); + now += 1_000; + await deliver(client, ["NOTICE", "queued"]); + await close(client, 1006); + assert.equal( + client.subscriptions.get(ids[0]).liveUntil, + 2_030, + "stamped at the last frame before the gap, not at wake", + ); +}); + +test("a new connection's liveness credit does not inherit the previous socket's", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([quiet]); + client.relayUrl = "ws://relay.test"; + await close(client, 1006); + // Reconnect well past the gap limit: the AUTH OK starts fresh credit + // instead of reading as a frame after a gap on the old socket's clock. + timers.get(client.reconnectTimeout).at += 300_000; + await redial(client); + const authAt = now; + now += 30_000; + await deliver(client, ["EOSE", ids[0]]); + now += 1_000; + await close(client, 1006); + assert.equal( + client.subscriptions.get(ids[0]).liveUntil, + Math.floor((authAt + 30_000) / 1_000), + ); +}); + +for (const code of [1012, 1001]) { + test(`close ${code} spreads the first reconnect over [0, 10s] and keeps the backoff escalated`, async () => { + const { client } = await liveClient([quiet]); + Math.random = () => 0.37; + // A client bounced repeatedly is already backed off. + for (let i = 0; i < 3; i++) client.reconnectBackoff.next(); + assert.equal(client.reconnectBackoff.currentMs, 8_000); + await close(client, code); + assert.equal(reconnectDelayMs(client), 3_700); + assert.equal(client.reconnectBackoff.currentMs, 16_000); + }); +} + +test("an ordinary close backs off with +/-25% jitter, not the restart spread", async () => { + const { client } = await liveClient([quiet]); + Math.random = () => 0.37; + await close(client, 1000); + assert.ok(Math.abs(reconnectDelayMs(client) - 935) < 1e-6); + assert.equal(client.reconnectBackoff.currentMs, 2_000); +}); + +function fireDueTimers() { + for (const [id, timer] of [...timers]) { + if (timer.at <= now) { + timers.delete(id); + timer.fn(); + } + } +} + +// Drive the production reconnect: the scheduled timer dials through IPC and +// the relay's AUTH OK settles the handshake; replay then runs for real. +async function redial(client) { + now = timers.get(client.reconnectTimeout).at; + fireDueTimers(); + await flush(); + assert.ok(client.authRequest, "the redial is waiting for AUTH"); + await deliver(client, ["OK", client.authRequest.pendingEventId, true, ""]); + await flush(); +} + +const paged = buildChannelFilter("channel-paged", 1000); + +test("a reset before the paged live EOSE releases connect() and the next reconnect dials on time", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([paged]); + client.relayUrl = "ws://relay.test"; + await close(client, 1006); // stamps liveUntil from the EOSE frame + await redial(client); + assert.equal(dials.length, 1); + assert.deepEqual(writes.at(-1).frame, [ + "REQ", + ids[0], + { ...paged, limit: 0 }, + ]); + assert.notEqual(client.connectPromise, null, "precondition: parked on EOSE"); + + // The socket drops before the live-only REQ is acknowledged. + await close(client, 1006); + await flush(); + assert.equal( + client.connectPromise, + null, + "the stale attempt settled at once", + ); + assert.equal(intervals.size, 0, "no watchdog for a dead generation"); + assert.equal(repairRequests.length, 0); + const retryAt = timers.get(client.reconnectTimeout).at; + assert.ok( + retryAt - now < 30_000, + "the retry is a backoff, not the ack timeout", + ); + + await redial(client); + assert.equal(dials.length, 2, "the scheduled reconnect opened a new socket"); + assert.equal(dials[1], retryAt); +}); + +test("a close delivered late during replay stamps the last frame, not the wake", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([quiet, paged]); + client.relayUrl = "ws://relay.test"; + await close(client, 1006); + await redial(client); + // The quiet channel reaches EOSE while the paged repair is still pending. + now += 30_000; + const eoseAt = Math.floor(now / 1_000); + await deliver(client, ["EOSE", ids[0]]); + await deliver(client, ["EOSE", ids[1]]); + await flush(); + assert.equal(repairRequests.length, 1, "precondition: repair in flight"); + assert.equal(intervals.size, 1, "the watchdog runs during replay"); + + // An hour of sleep, then the close is delivered. + now += 3_600_000; + await close(client, 1006); + const subscription = client.subscriptions.get(ids[0]); + assert.equal(subscription.liveUntil, eoseAt, "stamped at the EOSE frame"); + + // The in-flight HTTP page settles; the stale attempt then stands down. + for (const { resolve } of repairRequests.splice(0)) resolve([]); + await flush(); + assert.equal(client.connectPromise, null); + writes.length = 0; + await redial(client); + const quietReq = writes.find(({ frame }) => frame[1] === ids[0]); + assert.equal(quietReq.frame[2].since, eoseAt - 960); +}); + +test("a half-open socket that never acknowledges replay is reset by the stall watchdog", async () => { + now = 2_000_000; + const { client, ids } = await liveClient([paged]); + client.relayUrl = "ws://relay.test"; + await close(client, 1006); + await redial(client); + assert.deepEqual(writes.at(-1).frame, [ + "REQ", + ids[0], + { ...paged, limit: 0 }, + ]); + assert.notEqual(client.connectPromise, null, "precondition: parked on EOSE"); + assert.equal(client.connectionStateEmitter.get(), "connected"); + assert.equal(intervals.size, 1, "the watchdog is armed while parked"); + + // No EOSE, no close: the idle tick is what notices. + now += 70_000; + const [tick] = intervals.values(); + tick(); + await flush(); + assert.equal(client.connectionStateEmitter.get(), "stalled"); + assert.equal(client.connectPromise, null, "connect() released by the reset"); + assert.equal(intervals.size, 0); + assert.equal(repairRequests.length, 0, "no blind repair"); +}); + +test("disposing a subscription parked on its paged live EOSE releases connect()", async () => { + now = 2_000_000; + const { client, ready } = await liveClient([paged]); + const dispose = await ready[0]; + client.relayUrl = "ws://relay.test"; + await close(client, 1006); + await redial(client); + assert.notEqual(client.connectPromise, null, "precondition: parked on EOSE"); + await dispose(); + await flush(); + assert.equal( + client.connectPromise, + null, + "connect() settled without the 25s wait", + ); + assert.equal(client.connectionStateEmitter.get(), "connected"); +}); diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index ce64037570..eac2465a8c 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -36,7 +36,12 @@ import { prepareSubscriptionEvent, } from "@/shared/api/relayClosedRecovery"; import { getChannelReconnectRepairEvents } from "@/shared/api/channelReconnectRepair"; -import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay"; +import { + releaseLiveEoseWaiter, + replayLiveSubscriptions, + retireConnectionSubscriptions, +} from "@/shared/api/relayReconnectReplay"; +import { creditInboundLiveness } from "@/shared/api/relayResumeCursor"; import { RelayLiveReqDrain } from "./relayLiveReqDrain"; import { RelayChannelAccessRevocations } from "./relayChannelAccessRevocations"; import { publishSessionEvent } from "@/shared/api/relayEventPublisher"; @@ -51,6 +56,7 @@ import { isServiceRestartClose, isWebSocketClose, isWebSocketError, + RelayReconnectBackoff, shouldRefuseConnect, shouldScheduleReconnect, shouldWaitForScheduledReconnect, @@ -59,11 +65,8 @@ import { RelayReconnectWaiters } from "@/shared/api/relayReconnectWaiters"; import { RelayStallWatchdog } from "@/shared/api/relayStallWatchdog"; import { AUTH_TIMEOUT_MS, - BACKOFF_RESET_STABLE_MS, EVENT_BATCH_MS, HISTORY_TIMEOUT_MS, - RECONNECT_BASE_DELAY_MS, - RECONNECT_MAX_DELAY_MS, STALL_CHECK_INTERVAL_MS, STALL_IDLE_TIMEOUT_MS, } from "@/shared/api/relayClientTimings"; @@ -82,7 +85,7 @@ export class RelayClient { private connectPromise: Promise | null = null; private reconnectTimeout: number | null = null; private reconnectWaiters = new RelayReconnectWaiters(); - private reconnectDelayMs = RECONNECT_BASE_DELAY_MS; + private reconnectBackoff = new RelayReconnectBackoff(); private keepAliveRequested = false; private authRequest: RelayAuthRequest | null = null; private subscriptions = new Map(); @@ -96,9 +99,9 @@ export class RelayClient { private notifyReconnectListeners = false; private onMessageChannel: Channel | null = null; private connectionGeneration = 0; + private lastInboundAtMs = 0; // Liveness credit on this connection; 0 = none. private sessionEpoch = 0; private liveSessionAbort = new AbortController(); - private stabilityTimer: number | null = null; private visibleChannelId: string | null = null; private authOkTracker = new AuthOkTracker(); private terminal = false; @@ -122,10 +125,6 @@ export class RelayClient { window.clearTimeout(this.reconnectTimeout); this.reconnectTimeout = null; } - if (this.stabilityTimer !== null) { - window.clearTimeout(this.stabilityTimer); - this.stabilityTimer = null; - } this.stallWatchdog.stop(); this.sessionEpoch++; this.liveSessionAbort.abort(); @@ -181,7 +180,7 @@ export class RelayClient { this.channelAccessRevocations.clear(); this.connectionStateEmitter.clear(); this.onMessageChannel = null; - this.reconnectDelayMs = RECONNECT_BASE_DELAY_MS; + this.reconnectBackoff.reset(); } async fetchChannelHistory(channelId: string, limit = 50) { @@ -541,15 +540,13 @@ export class RelayClient { } private async connect() { - if (this.stabilityTimer !== null) { - window.clearTimeout(this.stabilityTimer); - this.stabilityTimer = null; - } + this.reconnectBackoff.cancelStableReset(); this.connectionStateEmitter.set( this.hasConnectedOnce ? "reconnecting" : "connecting", ); const generation = ++this.connectionGeneration; + this.lastInboundAtMs = 0; const inbound = createRelayInboundBuffer( async (delivery) => { for (const message of toRelayFrames(delivery)) @@ -593,14 +590,15 @@ export class RelayClient { await Promise.race([drain, authentication, inbound.overflow]); await drain; await authentication; - this.stabilityTimer = window.setTimeout(() => { - this.stabilityTimer = null; - this.reconnectDelayMs = RECONNECT_BASE_DELAY_MS; - }, BACKOFF_RESET_STABLE_MS); + this.reconnectBackoff.resetWhenStable(); this.connectionStateEmitter.set("connected"); - await this.replayLiveSubscriptions(); + // Watch the socket during replay too: a half-open socket that never + // sends EOSE or a close must not hold recovery behind the EOSE waits. this.stallWatchdog.start(); + await this.replayLiveSubscriptions(); + if (generation !== this.connectionGeneration) + throw new Error("Relay connection attempt was superseded."); this.emitReconnectIfNeeded(); return generation; } catch (error) { @@ -633,6 +631,7 @@ export class RelayClient { signal?.removeEventListener("abort", abort); sessionSignal.removeEventListener("abort", abort); window.clearTimeout(fallbackTimeout); + releaseLiveEoseWaiter(subscription); settleReady(); }; const subscription: Extract = { @@ -866,18 +865,23 @@ export class RelayClient { private async handleWsMessage(message: unknown, generation: number) { if (generation !== this.connectionGeneration) return; - this.stallWatchdog.recordInbound(); if (isWebSocketClose(message)) { - if (isServiceRestartClose(message)) - this.reconnectDelayMs = RECONNECT_BASE_DELAY_MS; - this.resetConnection(new Error("Relay connection closed.")); + this.resetConnection(new Error("Relay connection closed."), { + serviceRestart: isServiceRestartClose(message), + }); return; } if (isWebSocketError(message)) { this.resetConnection(new Error("Relay connection errored.")); return; } + // Only peer frames prove liveness; Close/Error say when the loss was seen. + this.stallWatchdog.recordInbound(); + this.lastInboundAtMs = creditInboundLiveness( + this.lastInboundAtMs, + Date.now(), + ); const payload = getTextPayload(message); if (!payload) { @@ -1066,7 +1070,7 @@ export class RelayClient { } } - private scheduleReconnect() { + private scheduleReconnect(serviceRestart = false) { if ( !shouldScheduleReconnect({ terminal: this.terminal, @@ -1079,14 +1083,7 @@ export class RelayClient { return; } - // ±25% jitter spreads a fleet's AUTH storms across a 50% window instead - // of hitting the relay at the same instant. - const jitter = this.reconnectDelayMs * (0.75 + Math.random() * 0.5); - const delay = Math.min(jitter, RECONNECT_MAX_DELAY_MS); - this.reconnectDelayMs = Math.min( - this.reconnectDelayMs * 2, - RECONNECT_MAX_DELAY_MS, - ); + const delay = this.reconnectBackoff.next(serviceRestart); this.reconnectTimeout = window.setTimeout(() => { this.reconnectTimeout = null; @@ -1125,16 +1122,19 @@ export class RelayClient { error: Error, options?: { reconnect?: boolean; + serviceRestart?: boolean; }, ) { this.onMessageChannel = null; + retireConnectionSubscriptions( + this.subscriptions, + this.connectionGeneration, + this.lastInboundAtMs, + ); this.stallWatchdog.stop(); this.connectionGeneration++; this.liveReqDrain.reset(); - if (this.stabilityTimer !== null) { - window.clearTimeout(this.stabilityTimer); - this.stabilityTimer = null; - } + this.reconnectBackoff.cancelStableReset(); if (this.flushTimeout !== null) window.clearTimeout(this.flushTimeout); this.flushTimeout = null; this.eventBuffer = []; @@ -1193,7 +1193,7 @@ export class RelayClient { this.pendingEvents.delete(eventId); } if (options?.reconnect !== false) { - this.scheduleReconnect(); + this.scheduleReconnect(options?.serviceRestart); } } } diff --git a/desktop/src/shared/api/relayClientShared.ts b/desktop/src/shared/api/relayClientShared.ts index 80f5de206d..351b779f37 100644 --- a/desktop/src/shared/api/relayClientShared.ts +++ b/desktop/src/shared/api/relayClientShared.ts @@ -80,6 +80,14 @@ type LiveSubscription = { /** Release readiness/cancellation listeners when this entry is retired. */ onRemoved?: () => void; lastSeenCreatedAt?: number; + /** Connection generation on which this subscription last reached EOSE. */ + eoseGeneration?: number; + /** + * Unix seconds up to which this subscription was registered on a connection + * that delivered its EOSE: that connection's last inbound frame, stamped on + * close. Not proof of delivery; see `relayResumeCursor.ts`. + */ + liveUntil?: number; /** * Lower bound of a reconnect backfill window that has not yet completed. * @@ -97,6 +105,19 @@ type LiveSubscription = { seenEventIds: Set; liveEose: boolean; repairDone: boolean; + /** When the restored live REQ went out; its EOSE deadline runs from here. */ + liveReqSentAtMs?: number; + /** + * Set when the history fallback follows an unacknowledged live REQ: a late + * original EOSE is then indistinguishable from the fallback's, so no EOSE + * ends dedupe; only the fallback's deadline or teardown does. + */ + ignoreEose?: boolean; + /** + * Wakes the repair worker waiting for the restored live REQ's EOSE: on + * EOSE, or when a newer replay supersedes this one. + */ + liveEoseWaiter?: () => void; }; closedRetryAttempt?: number; closedRetryTimeout?: number; diff --git a/desktop/src/shared/api/relayClosedRecovery.test.mjs b/desktop/src/shared/api/relayClosedRecovery.test.mjs index c09b90e4c5..1f0e39b840 100644 --- a/desktop/src/shared/api/relayClosedRecovery.test.mjs +++ b/desktop/src/shared/api/relayClosedRecovery.test.mjs @@ -784,6 +784,36 @@ test("relayClientSession wires closeSubscription into handleRelayClosed", async // ── Teardown ────────────────────────────────────────────────────────────────── +test("live EOSE records the connection generation; retryable CLOSED clears it", () => { + resetAll(0); + const subscription = { + mode: "live", + filter: { kinds: [9], limit: 10 }, + onEvent: () => {}, + }; + const subscriptions = new Map([["live-1", subscription]]); + handleSubscriptionEose({ + subscriptions, + subId: "live-1", + closeSubscription: async () => {}, + generation: 5, + }); + assert.equal(subscription.eoseGeneration, 5); + + handleRelayClosed({ + subscriptions, + subId: "live-1", + message: "error: database error", + sendReq: async () => {}, + }); + assert.equal( + subscription.eoseGeneration, + undefined, + "a relay-closed subscription is not live, so closing the socket must not stamp it", + ); + window.clearTimeout(subscription.closedRetryTimeout); +}); + test("teardown — restore Date.now", () => { Date.now = origDateNow; assert.ok(true); diff --git a/desktop/src/shared/api/relayClosedRecovery.ts b/desktop/src/shared/api/relayClosedRecovery.ts index 348d55e299..4fd855abb7 100644 --- a/desktop/src/shared/api/relayClosedRecovery.ts +++ b/desktop/src/shared/api/relayClosedRecovery.ts @@ -132,6 +132,9 @@ function recoverLiveSubscriptionFromClosed({ }) { subscription.resolveReady?.("closed"); subscription.resolveReady = undefined; + // The relay dropped this subscription: it is no longer live on this socket, + // so closing the socket must not stamp a `liveUntil` for it. + subscription.eoseGeneration = undefined; // A deadline on a live sub is treated as retryable: resubscribe with the // same filter under the 1-30s backoff, which can keep retrying at the cap. @@ -257,8 +260,10 @@ export function markReconnectLiveEose( generation: number, ) { const replay = subscription.reconnectReplay; - if (!replay || replay.generation !== generation) return; + if (!replay || replay.generation !== generation || replay.ignoreEose) return; replay.liveEose = true; + replay.liveEoseWaiter?.(); + replay.liveEoseWaiter = undefined; if (replay.repairDone) subscription.reconnectReplay = undefined; } @@ -286,8 +291,10 @@ export function handleSubscriptionEose({ const subscription = subscriptions.get(subId); if (!subscription) return; if (subscription.mode === "live") { - if (generation !== undefined) + if (generation !== undefined) { + subscription.eoseGeneration = generation; markReconnectLiveEose(subscription, generation); + } subscription.resolveReady?.("eose"); subscription.resolveReady = undefined; subscription.closedRetryAttempt = 0; diff --git a/desktop/src/shared/api/relayReconnectPolicy.test.mjs b/desktop/src/shared/api/relayReconnectPolicy.test.mjs index e6856ede18..eb904e8618 100644 --- a/desktop/src/shared/api/relayReconnectPolicy.test.mjs +++ b/desktop/src/shared/api/relayReconnectPolicy.test.mjs @@ -3,6 +3,8 @@ import test from "node:test"; import { isServiceRestartClose, + RelayReconnectBackoff, + SERVICE_RESTART_RECONNECT_SPREAD_MS, isWebSocketClose, shouldRefuseConnect, shouldScheduleReconnect, @@ -96,7 +98,11 @@ test("shouldRefuseConnect mirrors terminal", () => { assert.equal(shouldRefuseConnect({ terminal: true }), true); }); -test("only a close frame with code 1012 is a service restart", () => { +test("only a close frame with code 1012 or 1001 is a service restart", () => { + assert.equal( + isServiceRestartClose({ type: "Close", data: { code: 1001 } }), + true, + ); assert.equal(isWebSocketClose({ type: "Close" }), true); assert.equal(isWebSocketClose({ type: "Error" }), false); assert.equal( @@ -117,3 +123,97 @@ test("only a close frame with code 1012 is a service restart", () => { ); assert.equal(isServiceRestartClose(null), false); }); + +test("service-restart close spreads the first reconnect uniformly over [0, 10s]", () => { + assert.equal(SERVICE_RESTART_RECONNECT_SPREAD_MS, 10_000); + for (const [random, expected] of [ + [0, 0], + [0.25, 2_500], + [0.5, 5_000], + [0.999, 9_990], + ]) { + // A freshly reset backoff (1s) must not pin the delay near 1s. + const backoff = new RelayReconnectBackoff(1_000, 30_000, () => random); + assert.equal(backoff.next(true), expected); + } +}); + +test("service-restart spreading does not reset the escalated backoff", () => { + const backoff = new RelayReconnectBackoff(1_000, 30_000, () => 0.5); + backoff.next(); + backoff.next(); + backoff.next(); + assert.equal(backoff.currentMs, 8_000); + backoff.next(true); + assert.equal(backoff.currentMs, 16_000, "bounced client keeps backing off"); + assert.equal(backoff.next(), 16_000, "next attempt uses the escalated base"); +}); + +test("ordinary reconnect jitter is symmetric +/-25% and capped", () => { + const at = (random, start = 0) => { + const backoff = new RelayReconnectBackoff(1_000, 30_000, () => random); + for (let i = 0; i < start; i++) backoff.next(); + return backoff.next(); + }; + assert.equal(at(0), 750); + assert.equal(at(0.5), 1_000); + assert.ok(Math.abs(at(0.999) - 1_249.5) < 1e-6); + // At the cap the jittered delay never exceeds it. + assert.equal(at(0.999, 10), 30_000); + assert.equal(at(0, 10), 22_500); +}); + +test("backoff doubles per attempt up to the cap and only reset() returns it to base", () => { + const backoff = new RelayReconnectBackoff(1_000, 30_000, () => 0.5); + const delays = Array.from({ length: 7 }, () => backoff.next()); + assert.deepEqual( + delays, + [1_000, 2_000, 4_000, 8_000, 16_000, 30_000, 30_000], + ); + backoff.reset(); + assert.equal(backoff.next(), 1_000); +}); + +test("only a connection that stays up resets the backoff; a drop before then cancels it", () => { + const timers = new Map(); + let nextId = 1; + const originalWindow = globalThis.window; + globalThis.window = { + setTimeout(fn, ms) { + timers.set(nextId, { fn, ms }); + return nextId++; + }, + clearTimeout(id) { + timers.delete(id); + }, + }; + try { + const backoff = new RelayReconnectBackoff(1_000, 30_000, () => 0.5); + for (let i = 0; i < 3; i++) backoff.next(); + backoff.resetWhenStable(60_000); + assert.equal( + backoff.currentMs, + 8_000, + "opening a socket alone resets nothing", + ); + backoff.cancelStableReset(); // dropped before it was stable + assert.equal(timers.size, 0); + assert.equal(backoff.currentMs, 8_000); + + backoff.resetWhenStable(60_000); + backoff.resetWhenStable(60_000); // re-arming replaces, never stacks + assert.equal(timers.size, 1); + const [[id, timer]] = [...timers]; + assert.equal(timer.ms, 60_000); + timers.delete(id); + timer.fn(); + assert.equal(backoff.currentMs, 1_000, "stable for 60s: back to base"); + + backoff.next(); + backoff.resetWhenStable(60_000); + backoff.reset(); + assert.equal(timers.size, 0, "reset() also drops a pending stable reset"); + } finally { + globalThis.window = originalWindow; + } +}); diff --git a/desktop/src/shared/api/relayReconnectPolicy.ts b/desktop/src/shared/api/relayReconnectPolicy.ts index 780ad3516b..949d6777ce 100644 --- a/desktop/src/shared/api/relayReconnectPolicy.ts +++ b/desktop/src/shared/api/relayReconnectPolicy.ts @@ -21,6 +21,12 @@ * subscription, no `keepAliveRequested` from `preconnect()` → don't * keep an idle socket. */ +import { + BACKOFF_RESET_STABLE_MS, + RECONNECT_BASE_DELAY_MS, + RECONNECT_MAX_DELAY_MS, +} from "@/shared/api/relayClientTimings"; + export type RelayReconnectInputs = { terminal: boolean; hasPendingReconnect: boolean; @@ -59,6 +65,11 @@ export function isWebSocketClose( ); } +/** + * Whether the relay closed the socket because it is going away: 1012 (Service + * Restart, sent to every client of a pod during a rolling deploy) or 1001 + * (Going Away). Both mean "the whole fleet is about to reconnect". + */ export function isServiceRestartClose(message: unknown): boolean { if (!isWebSocketClose(message)) return false; if (!("data" in message)) return false; @@ -67,10 +78,85 @@ export function isServiceRestartClose(message: unknown): boolean { typeof data === "object" && data !== null && "code" in data && - data.code === 1012 + (data.code === 1012 || data.code === 1001) ); } +/** + * Server-side close drain is spread over ~10s; the first reconnect after a + * restart close spreads arrival across the same window ("full jitter"). + */ +export const SERVICE_RESTART_RECONNECT_SPREAD_MS = 10_000; + +/** + * Exponential reconnect backoff with jitter. + * + * - After a service-restart close the first attempt waits a uniform random + * delay in `[0, SERVICE_RESTART_RECONNECT_SPREAD_MS]` instead of resetting to + * the minimum backoff, so a fleet bounced by one deploy does not stampede. + * - Otherwise the delay gets symmetric +/-25% jitter, hard-capped at `maxMs` + * (a retry never waits longer than the cap). + * - Every attempt, including a spread one, doubles the stored backoff; only + * a connection that stays up (`resetWhenStable`) or `reset()` (community + * switch) returns it to the minimum. Merely opening a socket never does. + * + * `random` is injectable for tests and must return a value in `[0, 1)`. + */ +export class RelayReconnectBackoff { + private backoffMs: number; + private readonly baseMs: number; + private readonly maxMs: number; + private readonly random: () => number; + private stableTimer: number | null = null; + + constructor( + baseMs = RECONNECT_BASE_DELAY_MS, + maxMs = RECONNECT_MAX_DELAY_MS, + random = () => Math.random(), + ) { + this.baseMs = baseMs; + this.backoffMs = baseMs; + this.maxMs = maxMs; + this.random = random; + } + + /** Current un-jittered backoff, for diagnostics and tests. */ + get currentMs(): number { + return this.backoffMs; + } + + /** Back to the minimum now, dropping any pending stable-connection reset. */ + reset(): void { + this.cancelStableReset(); + this.backoffMs = this.baseMs; + } + + /** Reset once the connection that just opened has stayed up `stableMs`. */ + resetWhenStable(stableMs = BACKOFF_RESET_STABLE_MS): void { + this.cancelStableReset(); + this.stableTimer = window.setTimeout(() => { + this.stableTimer = null; + this.backoffMs = this.baseMs; + }, stableMs); + } + + /** The connection dropped (or a new attempt began) before it was stable. */ + cancelStableReset(): void { + if (this.stableTimer === null) return; + window.clearTimeout(this.stableTimer); + this.stableTimer = null; + } + + /** Delay before the next attempt; advances the backoff. */ + next(serviceRestart = false): number { + const delay = serviceRestart + ? this.random() * SERVICE_RESTART_RECONNECT_SPREAD_MS + : Math.min(this.backoffMs * (0.75 + this.random() * 0.5), this.maxMs); + this.backoffMs = Math.min(this.backoffMs * 2, this.maxMs); + return delay; + } +} + /** Whether a WS-layer message is the plugin's `Error` frame. */ export function isWebSocketError(message: unknown): boolean { return ( diff --git a/desktop/src/shared/api/relayReconnectReplay.test.mjs b/desktop/src/shared/api/relayReconnectReplay.test.mjs index 7c7d96f44c..627ada2c2a 100644 --- a/desktop/src/shared/api/relayReconnectReplay.test.mjs +++ b/desktop/src/shared/api/relayReconnectReplay.test.mjs @@ -5,16 +5,20 @@ import test from "node:test"; import { buildReconnectReplayFilter, DB_CREATED_AT_FLOOR_SECS, + LIVE_REQ_ACK_TIMEOUT_MS, FENCE_CLOCK_MARGIN_SECS, RELAY_INGEST_FUTURE_TOLERANCE_SECS, RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS, replayReconnectHistoryPages, PAGE_REPLAY_MAX_ATTEMPTS, replayLiveSubscriptions, + releaseLiveEoseWaiter, REPLAY_BATCH_SIZE, + retireConnectionSubscriptions, shouldPageReconnectReplay, } from "./relayReconnectReplay.ts"; import { buildChannelFilter } from "./relayChannelFilters.ts"; +import { stampLiveUntilOnClose } from "./relayResumeCursor.ts"; import { flushEvents, markReconnectLiveEose, @@ -92,6 +96,20 @@ function eventRange(prefix, start, count) { ).reverse(); } +// Plays the relay for paged channels: acknowledges each restored live REQ with +// EOSE exactly as the session's EOSE handler would, which releases the HTTP +// repair (it must start only after the live REQ is registered). +function ackingSendRaw(subscriptions, generation = 0, sent = []) { + return async (payload) => { + sent.push(payload); + if (payload[0] !== "REQ") return; + queueMicrotask(() => { + const subscription = subscriptions.get(payload[1]); + if (subscription) markReconnectLiveEose(subscription, generation); + }); + }; +} + function replayFilter(filter, since, until) { return buildReconnectReplayFilter(filter, since, until); } @@ -605,15 +623,14 @@ test("channel reconnect replay pages the missed window until a short page", asyn await replayLiveSubscriptions({ subscriptions, - sendRaw: async (payload) => { - sentPayloads.push(payload); - }, + sendRaw: ackingSendRaw(subscriptions, 0, sentPayloads), requestRepair: async (request) => { historyFilters.push(request); return pages.shift() ?? []; }, }); + // The repair pages cover the window, so the live REQ reads no history. assert.deepEqual(sentPayloads, [ [ "REQ", @@ -621,7 +638,7 @@ test("channel reconnect replay pages the missed window until a short page", asyn { kinds: filter.kinds, "#h": ["channel-1"], - limit: 50, + limit: 0, }, ], ]); @@ -673,7 +690,7 @@ test("native repair preserves an explicit subscription since boundary", async () await replayLiveSubscriptions({ subscriptions, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions), requestRepair: async (request) => { repairRequests.push(request); return []; @@ -762,6 +779,9 @@ test("reconnect replay starts live REQs in parallel and preserves per-sub page o for (const resolve of sendResolvers) { resolve(); } + for (const subscription of subscriptions.values()) { + markReconnectLiveEose(subscription, 0); + } await replayPromise; assert.deepEqual(historyFiltersByChannel, { @@ -900,7 +920,7 @@ test("disposed subscription receives no repair events after an in-flight page", await replayLiveSubscriptions({ subscriptions, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions), requestRepair: async () => { assert.equal(subscriptions.delete("live-1"), true); return [event("disposed", 1001)]; @@ -934,7 +954,7 @@ test("stale repair cannot clear the successor generation dedupe state", async () subscriptions, generation: 10, isActive: () => generationActive, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions, 10), requestRepair: async () => { startRepair(); return repairPage; @@ -981,7 +1001,7 @@ test("history backfill rejection never escapes replayLiveSubscriptions", async ( // Must resolve — a rejection here is the socket-killing flap regression. await replayLiveSubscriptions({ subscriptions, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions), requestRepair: async () => { historyCalls++; throw new Error("rate-limited: quota exceeded; retry in 4s"); @@ -1018,7 +1038,7 @@ test("backfill retry waits out the armed gate, then succeeds", async () => { }); const replayPromise = replayLiveSubscriptions({ subscriptions, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions), requestRepair: async () => { attemptAtMs.push(fakeNow); if (attemptAtMs.length === 1) { @@ -1064,7 +1084,7 @@ test("backfill retry aborts when the subscription was replaced", async () => { let historyCalls = 0; await replayLiveSubscriptions({ subscriptions, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions), requestRepair: async () => { historyCalls++; // Simulate the subscription being torn down while the REQ is in flight. @@ -1098,7 +1118,7 @@ test("exhausted backfill pins the floor: next replay still requests the original // Reconnect 1: every backfill attempt is rate-limited. await replayLiveSubscriptions({ subscriptions, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions), requestRepair: async () => { throw new Error("rate-limited: quota exceeded; retry in 4s"); }, @@ -1117,7 +1137,7 @@ test("exhausted backfill pins the floor: next replay still requests the original const historyFilters = []; await replayLiveSubscriptions({ subscriptions, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions), requestRepair: async (request) => { historyFilters.push(request); return []; @@ -1140,7 +1160,7 @@ test("exhausted backfill pins the floor: next replay still requests the original const laterFilters = []; await replayLiveSubscriptions({ subscriptions, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions), requestRepair: async (request) => { laterFilters.push(request); return []; @@ -1174,7 +1194,7 @@ test("in-flight stale abort keeps the pinned floor for the superseding connectio let historyCalls = 0; await replayLiveSubscriptions({ subscriptions, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions), isActive: () => generationActive, requestRepair: async () => { historyCalls++; @@ -1217,10 +1237,11 @@ test("future-dated cursor repairs backdated edits and deletions", async () => { lastSeenCreatedAt: 10_900, }; const requests = []; + const subscriptions = new Map([["live-1", subscription]]); await replayLiveSubscriptions({ - subscriptions: new Map([["live-1", subscription]]), + subscriptions, generation: 11, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions, 11), requestRepair: async (request) => { requests.push(request); return recovered; @@ -1247,10 +1268,11 @@ test("dense-second repair advances by event id without gaps", async () => { const dense = Array.from({ length: 1_205 }, (_, index) => event(index.toString().padStart(64, "0"), 2000), ); + const subscriptions = new Map([["live-1", subscription]]); await replayLiveSubscriptions({ - subscriptions: new Map([["live-1", subscription]]), + subscriptions, generation: 7, - sendRaw: async () => {}, + sendRaw: ackingSendRaw(subscriptions, 7), requestRepair: async (request) => { requests.push(request); const start = request.beforeId @@ -1349,9 +1371,722 @@ test("live-before-repair and repair-before-live dispatch once", async () => { } }); +// ── Resume cursor (liveUntil) and live-only REQ for paged channels ──────────── + +async function drainMicrotasks() { + for (let i = 0; i < 20; i++) await Promise.resolve(); +} + +function liveSub(filter, overrides = {}) { + return { mode: "live", filter, onEvent: () => {}, ...overrides }; +} + +async function replayOne(subscription, { generation = 0, repairs = [] } = {}) { + const subscriptions = new Map([["live-1", subscription]]); + const sent = []; + await replayLiveSubscriptions({ + subscriptions, + generation, + sendRaw: ackingSendRaw(subscriptions, generation, sent), + requestRepair: async (request) => { + repairs.push(request); + return []; + }, + }); + return { sent, repairs }; +} + +// A clock far enough past each test's liveUntil that the silent-drop window +// does not bind; the next test covers the window itself. +const RESUME_NOW_MS = 200_000_000; + +test("a recent liveUntil resumes no later than the silent-drop window", async () => { + resetGate(RESUME_NOW_MS); + const nowSecs = RESUME_NOW_MS / 1_000; + const quiet = { kinds: [9], "#h": ["a", "b"], limit: 1000 }; + const quietReplay = await replayOne(liveSub(quiet, { liveUntil: nowSecs })); + assert.equal(quietReplay.sent[0][2].since, nowSecs - 3_600); + const paged = buildChannelFilter("channel-1", 1000); + const { repairs } = await replayOne(liveSub(paged, { liveUntil: nowSecs })); + assert.equal( + repairs[0].since, + nowSecs - 3_600, + "a fan-out the relay dropped within the window is repaired", + ); +}); + +test("quiet non-paged subscription resumes from liveUntil - 960s with its full limit", async () => { + resetGate(RESUME_NOW_MS); // liveUntil is older than the silent-drop window + const filter = { kinds: [9], "#h": ["a", "b"], limit: 1000, since: 100 }; + const { sent, repairs } = await replayOne( + liveSub(filter, { liveUntil: 10_000 }), + ); + assert.deepEqual(sent, [ + ["REQ", "live-1", { ...filter, since: 9_040, limit: 1000 }], + ]); + assert.deepEqual( + repairs, + [], + "non-paged subscriptions never use HTTP repair", + ); +}); + +test("a backdated event stored after the close is inside the resume window", async () => { + resetGate(); + const filter = { kinds: [9], "#h": ["a", "b"], limit: 1000 }; + const liveUntil = 100_000; + const { sent } = await replayOne(liveSub(filter, { liveUntil })); + const { since } = sent[0][2]; + // The relay accepts created_at up to 900s behind its clock, so a row stored + // just after the close can be backdated by the full drift window. Add the + // client clock margin: a client clock 60s ahead must not lose it either. + const stored = [ + event("backdated-max", liveUntil - RELAY_INGEST_FUTURE_TOLERANCE_SECS), + event( + "backdated-skewed-clock", + liveUntil - RELAY_INGEST_FUTURE_TOLERANCE_SECS - 59, + ), + event("backdated-1s", liveUntil - 1), + ]; + for (const row of stored) { + assert.ok( + row.created_at >= since, + `${row.id} (created_at ${row.created_at}) must satisfy REQ since ${since}`, + ); + } +}); + +test("resume since never drops below the original filter since", async () => { + resetGate(); + const filter = { kinds: [9], "#h": ["a", "b"], limit: 50, since: 9_990 }; + const { sent } = await replayOne(liveSub(filter, { liveUntil: 10_000 })); + assert.equal(sent[0][2].since, 9_990); +}); + +test("resume uses the newer of lastSeen - 5s and liveUntil - 960s and caps limit as before", async () => { + resetGate(RESUME_NOW_MS); // liveUntil is older than the silent-drop window + const filter = { kinds: [9], "#h": ["a", "b"], limit: 1000 }; + for (const [lastSeen, liveUntil, since] of [ + [10_000, 9_000, 9_995], + [9_000, 10_000, 9_040], + ]) { + const { sent } = await replayOne( + liveSub(filter, { lastSeenCreatedAt: lastSeen, liveUntil }), + ); + assert.deepEqual(sent[0][2], { ...filter, since, limit: 500 }); + } +}); + +test("a subscription with no cursor still replays its original filter", async () => { + resetGate(); + const filter = { kinds: [9], "#h": ["a", "b"], limit: 1000, since: 100 }; + const { sent } = await replayOne(liveSub(filter)); + assert.equal(sent[0][2], filter); + const paged = buildChannelFilter("channel-1", 1000); + const second = await replayOne(liveSub(paged)); + assert.equal( + second.sent[0][2], + paged, + "no cursor means no repair to lean on", + ); + assert.deepEqual(second.repairs, []); +}); + +test("paged channel resumes from liveUntil with a live-only REQ and a bounded HTTP repair", async () => { + resetGate(RESUME_NOW_MS); // liveUntil is older than the silent-drop window + const filter = { ...buildChannelFilter("channel-1", 1000), since: 100 }; + const { sent, repairs } = await replayOne( + liveSub(filter, { liveUntil: 100_000 }), + ); + // The repair covers the window, so the relay does no history read. + assert.deepEqual(sent, [["REQ", "live-1", { ...filter, limit: 0 }]]); + assert.equal( + repairs[0].since, + 100_000 - RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS, + "channel repair keeps the wide event-time lookback", + ); + assert.equal(repairs.length, 1); +}); + +test("a pinned failed-backfill floor beats a newer liveUntil", async () => { + resetGate(RESUME_NOW_MS); // liveUntil is older than the silent-drop window + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + liveUntil: 100_000, + pendingReplaySince: 4_000, + }); + const { repairs } = await replayOne(subscription); + assert.equal(repairs[0].since, 4_000, "never advance past a recorded gap"); +}); + +test("paged repair starts only after the live-only REQ is acknowledged", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + const subscriptions = new Map([["live-1", subscription]]); + const repairs = []; + const replay = replayLiveSubscriptions({ + subscriptions, + sendRaw: async () => {}, + requestRepair: async (request) => { + repairs.push(request); + return []; + }, + }); + await drainMicrotasks(); + assert.deepEqual( + repairs, + [], + "no repair before EOSE: the gap would be unowned", + ); + markReconnectLiveEose(subscription, 0); + await replay; + assert.equal(repairs.length, 1); +}); + +test("an unacknowledged live REQ falls back after one deadline instead of repairing blind", async () => { + resetGate(); + const filter = buildChannelFilter("channel-1", 1000); + const subscription = liveSub(filter, { lastSeenCreatedAt: 5_000 }); + const repairs = []; + const sent = []; + const replay = replayLiveSubscriptions({ + subscriptions: new Map([["live-1", subscription]]), + sendRaw: async (payload) => { + sent.push(payload); + }, + requestRepair: async (request) => { + repairs.push(request); + return []; + }, + }); + await drainMicrotasks(); + tickTo(fakeNow + LIVE_REQ_ACK_TIMEOUT_MS); + await replay; + // Waiting again on the same unchanged REQ cannot make its EOSE arrive. + assert.deepEqual(sent, [ + ["REQ", "live-1", { ...filter, limit: 0 }], + [ + "REQ", + "live-1", + buildReconnectReplayFilter( + filter, + 5_000 - RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS, + ), + ], + ]); + assert.deepEqual(repairs, [], "bounded: never repairs without the overlap"); + assert.equal( + subscription.pendingReplaySince, + 5_000 - RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS, + "the window stays pinned for the next reconnect", + ); + // Only the history fallback's dedupe deadline remains; it releases the set. + assert.equal(pendingTimers.size, 1, "no ack timer is left armed"); + tickTo(fakeNow + LIVE_REQ_ACK_TIMEOUT_MS); + assert.equal(subscription.reconnectReplay, undefined); + assert.equal(pendingTimers.size, 0); +}); + +test("every paged repair floor is pinned before any live-only REQ, so a queued channel survives a second disconnect", async () => { + resetGate(); + const lastSeen = 5_000; + const delivered = []; + const originalFloor = lastSeen - RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS; + const subscriptions = new Map( + Array.from({ length: 5 }, (_, i) => [ + `live-${i + 1}`, + liveSub(buildChannelFilter(`channel-${i + 1}`, 1000), { + lastSeenCreatedAt: lastSeen, + onEvent: (value) => delivered.push(value.id), + }), + ]), + ); + // Only the first PAGE_CONCURRENCY (4) repairs start; the 5th stays queued. + const blocked = []; + let firstConnectionActive = true; + const first = replayLiveSubscriptions({ + subscriptions, + generation: 0, + isActive: () => firstConnectionActive, + sendRaw: ackingSendRaw(subscriptions, 0), + requestRepair: (request) => + new Promise((resolve) => blocked.push({ request, resolve })), + }); + await drainMicrotasks(); + assert.equal(blocked.length, 4, "precondition: the 5th repair is queued"); + const queued = subscriptions.get("live-5"); + assert.equal(queued.reconnectReplay.liveEose, true, "5th live REQ was acked"); + assert.equal( + queued.pendingReplaySince, + originalFloor, + "the queued channel's floor is pinned although its worker never ran", + ); + + // The session's EOSE handler records the generation, and the socket drops + // again before the queued repair ever ran. + for (const subscription of subscriptions.values()) { + subscription.eoseGeneration = 0; + } + stampLiveUntilOnClose(subscriptions, 0, 200_000_000, 200_000_000); + assert.equal(queued.liveUntil, 200_000); + firstConnectionActive = false; + + // A message older than the (advanced) lookback that was never fetched. + const missed = event("missed-old", 4_000); + const repairs = []; + await replayLiveSubscriptions({ + subscriptions, + generation: 1, + sendRaw: ackingSendRaw(subscriptions, 1), + requestRepair: async (request) => { + repairs.push(request); + return request.channelId === "channel-5" && + missed.created_at >= request.since + ? [missed] + : []; + }, + }); + const fifth = repairs.find((request) => request.channelId === "channel-5"); + assert.equal( + fifth.since, + originalFloor, + "never advance past an unrepaired floor", + ); + assert.ok( + missed.created_at < 200_000 - RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS, + ); + assert.deepEqual(delivered, ["missed-old"], "the old message is recovered"); + assert.equal( + queued.pendingReplaySince, + undefined, + "cleared by the completed repair", + ); + + for (const { resolve } of blocked) resolve([]); + await first; +}); + +test("a superseding replay releases the stale worker without a repair", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + const subscriptions = new Map([["live-1", subscription]]); + let active = true; + const repairs = []; + const stale = replayLiveSubscriptions({ + subscriptions, + generation: 1, + isActive: () => active, + sendRaw: async () => {}, + requestRepair: async (request) => { + repairs.push(request); + return []; + }, + }); + await drainMicrotasks(); + active = false; + // The successor installs its own replay state on the shared subscription. + await replayLiveSubscriptions({ + subscriptions, + generation: 2, + sendRaw: ackingSendRaw(subscriptions, 2), + requestRepair: async (request) => { + repairs.push(request); + return []; + }, + }); + await stale; + assert.equal(repairs.length, 1, "only the live generation repairs"); + assert.equal(pendingTimers.size, 0); +}); + // ── Teardown ────────────────────────────────────────────────────────────────── test("teardown — restore Date.now", () => { Date.now = origDateNow; assert.ok(true); }); + +test("a connection reset releases repair workers parked on EOSE at once", async () => { + resetGate(); + const subscriptions = new Map( + ["channel-1", "channel-2"].map((channelId, i) => [ + `live-${i + 1}`, + liveSub(buildChannelFilter(channelId, 1000), { + lastSeenCreatedAt: 5_000, + eoseGeneration: 0, + }), + ]), + ); + let active = true; + let settled = false; + const repairs = []; + const replay = replayLiveSubscriptions({ + subscriptions, + isActive: () => active, + // One worker: the second channel is still queued behind the first. + pageReplayConcurrency: 1, + sendRaw: async () => {}, + requestRepair: async (request) => { + repairs.push(request); + return []; + }, + }).then(() => { + settled = true; + }); + await drainMicrotasks(); + assert.equal(settled, false, "precondition: parked on the live EOSE"); + + // The session's resetConnection: generation moves on, then retirement. + active = false; + retireConnectionSubscriptions(subscriptions, 0, 4_000_000); + await drainMicrotasks(); + assert.equal( + settled, + true, + "connect() must not wait LIVE_REQ_ACK_TIMEOUT_MS on a dead socket", + ); + await replay; + assert.deepEqual( + repairs, + [], + "neither the parked nor the queued worker repaired", + ); + assert.equal(pendingTimers.size, 0, "no ack timer outlives the connection"); + for (const subscription of subscriptions.values()) { + assert.equal( + subscription.pendingReplaySince, + 5_000 - RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS, + "the unrepaired window stays pinned", + ); + } +}); + +test("disposing a subscription releases its parked repair worker", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + const subscriptions = new Map([["live-1", subscription]]); + let settled = false; + const replay = replayLiveSubscriptions({ + subscriptions, + sendRaw: async () => {}, + requestRepair: async () => assert.fail("a disposed sub must not repair"), + }).then(() => { + settled = true; + }); + await drainMicrotasks(); + subscriptions.delete("live-1"); + releaseLiveEoseWaiter(subscription); + await drainMicrotasks(); + assert.equal(settled, true); + await replay; + assert.equal(pendingTimers.size, 0); +}); + +test("the EOSE wait cancels its timer through the injected clear function", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + const injected = new Map(); + const cleared = []; + let nextHandle = 9_000; + const replay = replayLiveSubscriptions({ + subscriptions: new Map([["live-1", subscription]]), + sendRaw: async () => {}, + requestRepair: async () => [], + setTimeoutFn: (fn, ms) => { + injected.set(++nextHandle, { fn, ms }); + return nextHandle; + }, + clearTimeoutFn: (handle) => { + cleared.push(handle); + injected.delete(handle); + }, + }); + await drainMicrotasks(); + const [[handle, timer]] = [...injected]; + assert.equal(timer.ms, LIVE_REQ_ACK_TIMEOUT_MS); + markReconnectLiveEose(subscription, 0); + await replay; + assert.deepEqual(cleared, [handle], "cancelled by the matching clear"); + assert.equal(injected.size, 0); +}); + +test("an exhausted repair falls back to the window's bounded history on the same connection", async () => { + resetGate(); + const filter = buildChannelFilter("channel-1", 1000); + const subscription = liveSub(filter, { lastSeenCreatedAt: 5_000 }); + const subscriptions = new Map([["live-1", subscription]]); + const sent = []; + let attempts = 0; + await replayLiveSubscriptions({ + subscriptions, + sendRaw: ackingSendRaw(subscriptions, 0, sent), + requestRepair: async () => { + attempts++; + throw new Error("error: database error"); + }, + }); + const floor = 5_000 - RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS; + assert.equal(attempts, PAGE_REPLAY_MAX_ATTEMPTS); + assert.deepEqual(sent, [ + ["REQ", "live-1", { ...filter, limit: 0 }], + ["REQ", "live-1", buildReconnectReplayFilter(filter, floor)], + ]); + assert.equal(sent[1][2].since, floor, "history from the missed window"); + assert.ok(sent[1][2].limit > 0, "the fallback reads history"); + assert.equal( + subscription.pendingReplaySince, + floor, + "the floor stays pinned for anything the cap left out", + ); +}); + +test("a superseded replay sends no history fallback", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + const subscriptions = new Map([["live-1", subscription]]); + const sent = []; + let active = true; + let attempts = 0; + await replayLiveSubscriptions({ + subscriptions, + isActive: () => active, + sendRaw: ackingSendRaw(subscriptions, 0, sent), + requestRepair: async () => { + if (++attempts === PAGE_REPLAY_MAX_ATTEMPTS) active = false; + throw new Error("error: database error"); + }, + }); + assert.equal(attempts, PAGE_REPLAY_MAX_ATTEMPTS); + assert.equal(sent.length, 1, "only the original live-only REQ went out"); +}); + +test("the exhausted-repair fallback keeps dedupe until its own EOSE", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + const subscriptions = new Map([["live-1", subscription]]); + const event = { id: "seen-live", created_at: 4_900 }; + let dispatched = 0; + let dedupeDuringFallback; + await replayLiveSubscriptions({ + subscriptions, + sendRaw: async (payload) => { + if (payload[0] !== "REQ") return; + const fallback = payload[2].limit !== 0; + queueMicrotask(() => { + if (fallback) dedupeDuringFallback = !!subscription.reconnectReplay; + // Live delivers it first; the fallback's history returns it again. + if (shouldDispatchSubscriptionEvent(subscription, event)) dispatched++; + markReconnectLiveEose(subscription, 0); + }); + }, + requestRepair: async () => { + throw new Error("error: database error"); + }, + }); + await drainMicrotasks(); + assert.equal(dedupeDuringFallback, true); + assert.equal(dispatched, 1, "the fallback must not redeliver a live event"); + assert.equal( + subscription.reconnectReplay, + undefined, + "the fallback's own EOSE ends dedupe", + ); + assert.equal(pendingTimers.size, 0, "and cancels the dedupe deadline"); +}); + +test("a failed fallback REQ ends dedupe", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + const subscriptions = new Map([["live-1", subscription]]); + const acking = ackingSendRaw(subscriptions); + await replayLiveSubscriptions({ + subscriptions, + sendRaw: async (payload) => { + if (payload[0] === "REQ" && payload[2].limit !== 0) { + throw new Error("socket closed"); + } + await acking(payload); + }, + requestRepair: async () => { + throw new Error("error: database error"); + }, + }); + assert.equal(subscription.reconnectReplay, undefined); +}); + +for (const replaced of [false, true]) { + test(`a ${replaced ? "replacement" : "disposal"} racing the fallback REQ ${replaced ? "is left open" : "is closed again"}`, async () => { + resetGate(); + const filter = buildChannelFilter("channel-1", 1000); + const subscription = liveSub(filter, { lastSeenCreatedAt: 5_000 }); + const subscriptions = new Map([["live-1", subscription]]); + const sent = []; + let markStarted; + const started = new Promise((resolve) => { + markStarted = resolve; + }); + let release; + const held = new Promise((resolve) => { + release = resolve; + }); + const run = replayLiveSubscriptions({ + subscriptions, + sendRaw: async (payload) => { + sent.push(payload); + if (payload[0] !== "REQ") return; + if (payload[2].limit === 0) { + queueMicrotask(() => markReconnectLiveEose(subscription, 0)); + return; + } + markStarted(); + await held; + }, + requestRepair: async () => { + throw new Error("error: database error"); + }, + }); + await started; + subscriptions.delete("live-1"); + if (replaced) subscriptions.set("live-1", liveSub(filter)); + release(); + await run; + assert.deepEqual( + sent.map((frame) => frame[0]), + replaced ? ["REQ", "REQ"] : ["REQ", "REQ", "CLOSE"], + ); + if (!replaced) assert.deepEqual(sent[2], ["CLOSE", "live-1"]); + }); +} + +async function exhaustWithLostFallbackEose(subscription) { + const subscriptions = new Map([["live-1", subscription]]); + await replayLiveSubscriptions({ + subscriptions, + sendRaw: async (payload) => { + // The relay acknowledges the restored live REQ but drops the + // fallback's EOSE while the socket stays up. + if (payload[0] !== "REQ" || payload[2].limit !== 0) return; + queueMicrotask(() => markReconnectLiveEose(subscription, 0)); + }, + requestRepair: async () => { + throw new Error("error: database error"); + }, + }); + await drainMicrotasks(); +} + +test("a lost fallback EOSE releases dedupe at the deadline", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + await exhaustWithLostFallbackEose(subscription); + const retained = subscription.reconnectReplay; + assert.ok(retained, "precondition: dedupe waits for the fallback's EOSE"); + for (let i = 0; i < 50; i++) { + shouldDispatchSubscriptionEvent(subscription, event(`live-${i}`, 5_100)); + } + assert.equal(retained.seenEventIds.size, 50, "live traffic grows the set"); + tickTo(fakeNow + LIVE_REQ_ACK_TIMEOUT_MS); + assert.equal(subscription.reconnectReplay, undefined, "released"); + assert.equal(pendingTimers.size, 0); + assert.equal( + subscription.pendingReplaySince, + 5_000 - RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS, + "the unrepaired window stays pinned", + ); +}); + +test("a stale fallback release cannot clear a successor's dedupe", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + await exhaustWithLostFallbackEose(subscription); + const staleRelease = subscription.reconnectReplay.liveEoseWaiter; + const successorWaiter = () => {}; + const successor = { + generation: 1, + seenEventIds: new Set(["next"]), + liveEose: false, + repairDone: false, + liveEoseWaiter: successorWaiter, + }; + subscription.reconnectReplay = successor; + staleRelease(); + tickTo(fakeNow + LIVE_REQ_ACK_TIMEOUT_MS); + assert.equal(subscription.reconnectReplay, successor); + assert.equal(successor.liveEoseWaiter, successorWaiter); +}); + +test("each EOSE deadline runs from its REQ's send, not from when a queued worker starts", async () => { + resetGate(); + const subscriptions = new Map( + [1, 2].map((i) => [ + `live-${i}`, + liveSub(buildChannelFilter(`channel-${i}`, 1000), { + lastSeenCreatedAt: 5_000, + }), + ]), + ); + const fallbacks = []; + const replay = replayLiveSubscriptions({ + subscriptions, + pageReplayConcurrency: 1, + sendRaw: async (payload) => { + if (payload[0] === "REQ" && payload[2].limit !== 0) { + fallbacks.push(payload[1]); + } + }, + requestRepair: async () => assert.fail("never acknowledged"), + }); + await drainMicrotasks(); + // One deadline covers both: the queued second channel gets no fresh wait. + tickTo(fakeNow + LIVE_REQ_ACK_TIMEOUT_MS); + await drainMicrotasks(); + tickTo(fakeNow); + await drainMicrotasks(); + assert.deepEqual(fallbacks, ["live-1", "live-2"]); + await replay; +}); + +test("a late original EOSE after the ack timeout does not end the fallback's dedupe", async () => { + resetGate(); + const subscription = liveSub(buildChannelFilter("channel-1", 1000), { + lastSeenCreatedAt: 5_000, + }); + const event = { id: "seen-live", created_at: 4_900 }; + let dispatched = 0; + const deliver = () => { + if (shouldDispatchSubscriptionEvent(subscription, event)) dispatched++; + }; + const replay = replayLiveSubscriptions({ + subscriptions: new Map([["live-1", subscription]]), + sendRaw: async () => {}, + requestRepair: async () => assert.fail("never acknowledged"), + }); + await drainMicrotasks(); + deliver(); // live delivers it while the REQ is unacknowledged + tickTo(fakeNow + LIVE_REQ_ACK_TIMEOUT_MS); + await replay; // the fallback REQ went out + markReconnectLiveEose(subscription, 0); // the original's EOSE, late + deliver(); // the fallback's history returns it again + assert.equal(dispatched, 1, "dispatched once"); + assert.ok(subscription.reconnectReplay, "dedupe waits for the deadline"); + tickTo(fakeNow + LIVE_REQ_ACK_TIMEOUT_MS); + assert.equal(subscription.reconnectReplay, undefined); + assert.equal(pendingTimers.size, 0); +}); diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts index 18b8f9fd57..5c6d778ae4 100644 --- a/desktop/src/shared/api/relayReconnectReplay.ts +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -13,13 +13,18 @@ import { isRateLimited, waitForRateLimit, } from "@/shared/api/relayRateLimitGate"; +import { + computeResumeSince, + RELAY_INGEST_FUTURE_TOLERANCE_SECS, + stampLiveUntilOnClose, +} from "@/shared/api/relayResumeCursor"; + +export { RELAY_INGEST_FUTURE_TOLERANCE_SECS }; -const RECONNECT_REPLAY_SKEW_SECS = 5; /** * Legacy cursors are event-time maxima, so cover relay future tolerance (900s) * plus the channel-row storage fence and margin (960s + 5s). */ -export const RELAY_INGEST_FUTURE_TOLERANCE_SECS = 900; export const DB_CREATED_AT_FLOOR_SECS = 960; export const FENCE_CLOCK_MARGIN_SECS = 5; export const RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS = @@ -61,6 +66,16 @@ export const REPLAY_BATCH_SIZE = 8; */ export const REPLAY_INTER_BATCH_DELAY_MS = 50; +/** + * Longest a paged channel's HTTP repair waits for its live-only REQ to be + * acknowledged (EOSE) before treating the attempt as failed. + * + * The repair must start after the relay registered the live subscription: + * the live REQ no longer reads history, so an event stored between the + * repair's upper bound and registration would be delivered by neither. + */ +export const LIVE_REQ_ACK_TIMEOUT_MS = 25_000; + async function runWithConcurrency( items: T[], concurrency: number, @@ -79,6 +94,65 @@ async function runWithConcurrency( ); } +/** + * Resolves `true` once the restored live REQ reached EOSE on this replay's + * connection, `false` on timeout. Also resolves early (`true`) when this + * replay is superseded or its connection or subscription is retired; callers + * re-check `isActive()` and subscription identity after waking. A waiter that + * outlived its connection would hold `connect()` (and with it the session's + * `connectPromise`) for the whole timeout, stalling the scheduled reconnect. + */ +function awaitLiveEose( + replay: NonNullable< + Extract["reconnectReplay"] + >, + isActive: () => boolean, + setTimeoutFn: (fn: () => void, ms: number) => number, + clearTimeoutFn: (handle: number) => void, +): Promise { + if (replay.liveEose || !isActive()) return Promise.resolve(true); + // The deadline runs from the REQ's send, not from when a (concurrency + // limited) worker got here: a queued channel must not get a fresh wait. + const sentAtMs = replay.liveReqSentAtMs ?? Date.now(); + const remainingMs = Math.max( + 0, + sentAtMs + LIVE_REQ_ACK_TIMEOUT_MS - Date.now(), + ); + return new Promise((resolve) => { + const timer = setTimeoutFn(() => { + replay.liveEoseWaiter = undefined; + resolve(false); + }, remainingMs); + replay.liveEoseWaiter = () => { + replay.liveEoseWaiter = undefined; + clearTimeoutFn(timer); + resolve(true); + }; + }); +} + +/** Wake a repair worker parked on this subscription's live EOSE, if any. */ +export function releaseLiveEoseWaiter(subscription: RelaySubscription) { + if (subscription.mode !== "live") return; + subscription.reconnectReplay?.liveEoseWaiter?.(); +} + +/** + * Per-subscription bookkeeping for a connection that is being torn down: + * stamp the resume cursor, then release every repair worker parked on that + * connection's EOSE so the stale `connect()` settles at once. + */ +export function retireConnectionSubscriptions( + subscriptions: Map, + generation: number, + lastInboundMs: number, +) { + stampLiveUntilOnClose(subscriptions, generation, lastInboundMs); + for (const subscription of subscriptions.values()) { + releaseLiveEoseWaiter(subscription); + } +} + export function buildReconnectReplayFilter( filter: RelaySubscriptionFilter, since?: number, @@ -189,6 +263,7 @@ export async function replayLiveSubscriptions({ interBatchDelayMs = REPLAY_INTER_BATCH_DELAY_MS, setTimeoutFn = (fn: () => void, ms: number) => window.setTimeout(fn, ms) as unknown as number, + clearTimeoutFn = (handle: number) => window.clearTimeout(handle), isActive = () => true, }: { subscriptions: Map; @@ -206,6 +281,8 @@ export async function replayLiveSubscriptions({ interBatchDelayMs?: number; /** setTimeout implementation (injectable for tests). */ setTimeoutFn?: (fn: () => void, ms: number) => number; + /** clearTimeout matching `setTimeoutFn` (injectable for tests). */ + clearTimeoutFn?: (handle: number) => void; /** * Returns false when the connection that initiated this replay has been * superseded by a newer one. After the gate await resumes, a stale replay @@ -233,16 +310,17 @@ export async function replayLiveSubscriptions({ const shouldPageReplay = channelId !== undefined && shouldPageReconnectReplay(subscription.filter); - const cursorSince = - subscription.lastSeenCreatedAt === undefined - ? undefined - : Math.max( - 0, - subscription.lastSeenCreatedAt - - (shouldPageReplay - ? RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS - : RECONNECT_REPLAY_SKEW_SECS), - ); + // Paged channels keep the wide event-time lookback for both terms: the + // relay accepts `created_at` up to 900s either side of storage time, so + // a narrower window could skip a backdated row stored after the close. + const cursorSince = computeResumeSince({ + lastSeenCreatedAt: subscription.lastSeenCreatedAt, + liveUntil: subscription.liveUntil, + ...(shouldPageReplay && { + lastSeenSkewSecs: RECONNECT_REPLAY_CHANNEL_LOOKBACK_SECS, + liveUntilSkewSecs: RECONNECT_REPLAY_CHANNEL_LOOKBACK_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 @@ -259,9 +337,21 @@ export async function replayLiveSubscriptions({ ? undefined : Math.max(replayFloor, subscription.filter.since ?? 0); const willRepair = shouldPageReplay && replaySince !== undefined; + let replay: typeof subscription.reconnectReplay; if (willRepair) { + // Pin the window's lower bound NOW, before any live-only REQ is sent + // or any await: every live REQ goes out before the (concurrency + // limited) repair workers start, so a queued channel can reach EOSE + // and have its socket close stamp `liveUntil` while its history was + // never fetched. The pin keeps the next replay's floor at or below + // this one until a repair pass actually completes. Events on the + // live REQ advance `lastSeenCreatedAt` independently of backfill + // success, so the cursor alone could skip the window permanently. + subscription.pendingReplaySince = replaySince; + // A superseded replay's worker is parked on its own state object. + subscription.reconnectReplay?.liveEoseWaiter?.(); // Install before the restored live REQ: a live frame may beat page one. - subscription.reconnectReplay = { + replay = subscription.reconnectReplay = { generation, seenEventIds: new Set(), liveEose: false, @@ -274,7 +364,14 @@ export async function replayLiveSubscriptions({ subscription, channelId, replaySince, + replay, shouldPageReplay: willRepair, + // Quiet subscriptions resume from `liveUntil` alone; keep their full + // original limit rather than the steady-state cap. + resumeLimit: + subscription.lastSeenCreatedAt === undefined + ? subscription.filter.limit + : undefined, }; }); @@ -305,17 +402,33 @@ export async function replayLiveSubscriptions({ const batch = replayRequests.slice(i, i + replayBatchSize); await Promise.all( batch.map( - async ({ subId, subscription, replaySince, shouldPageReplay }) => { + async ({ + subId, + subscription, + replaySince, + shouldPageReplay, + resumeLimit, + replay, + }) => { // The subscription map may change while replay is paused behind the // gate or an inter-batch delay. Never resurrect a disposed entry (or // overwrite its replacement) with a REQ from this stale snapshot. if (!isActive() || subscriptions.get(subId) !== subscription) return; + if (replay) replay.liveReqSentAtMs = Date.now(); await sendRaw([ "REQ", subId, + // The HTTP repair below already covers the missed window, so the + // live REQ reads no history (`limit: 0`); the repair starts only + // after this REQ's EOSE, which keeps backfill and live overlapping. shouldPageReplay - ? subscription.filter - : buildReconnectReplayFilter(subscription.filter, replaySince), + ? { ...subscription.filter, limit: 0 } + : buildReconnectReplayFilter( + subscription.filter, + replaySince, + undefined, + resumeLimit, + ), ]); // Disposal can race the asynchronous websocket invoke after the // identity check. If this exact entry was deleted while REQ was in @@ -342,6 +455,7 @@ export async function replayLiveSubscriptions({ ): request is typeof request & { channelId: string; replaySince: number; + replay: NonNullable; shouldPageReplay: true; } => request.shouldPageReplay && @@ -349,19 +463,91 @@ export async function replayLiveSubscriptions({ request.replaySince !== undefined, ), pageReplayConcurrency, - async ({ subId, subscription, channelId, replaySince }) => { + async ({ subId, subscription, channelId, replaySince, replay }) => { // 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. + // of times, then fall back to a bounded history REQ. // - // 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; + // The window's lower bound was pinned when the replay state was built; + // it is cleared only on a completed pass below. + const isCurrent = () => + isActive() && subscriptions.get(subId) === subscription; + + // The live REQ reads no history (`limit: 0`) and the pinned floor only + // helps the next reconnect, so without this the missed window stays + // empty for the rest of a stable connection. Fall back to the + // pre-repair behavior: re-issue the REQ (same ID, so the relay replaces + // the live subscription) with the window's bounded history. The floor + // stays pinned for anything the cap left out. + const sendHistoryFallback = async () => { + // That history overlaps what live (and any completed page) already + // dispatched, so keep the seen IDs until the fallback's own EOSE. If + // the original EOSE was consumed, the next one for this generation is + // the fallback's; if it never came, a late one could pass for the + // fallback's, so only the deadline below ends dedupe. + replay.repairDone = true; + if (replay.liveEose) replay.liveEose = false; + else replay.ignoreEose = true; + if (isRateLimited()) await waitForRateLimit(); + // Dedupe ends on that EOSE, on teardown (both call the waiter), or at + // the deadline: the relay may drop an EOSE without closing the + // socket, and the set must not grow with live traffic for the rest of + // the connection. Identity-fenced so a stale release cannot clear a + // successor's state. + let deadline: number | undefined; + const endDedupe = () => { + if (deadline !== undefined) clearTimeoutFn(deadline); + replay.liveEoseWaiter = undefined; + if (subscription.reconnectReplay === replay) { + subscription.reconnectReplay = undefined; + } + }; + if (!isCurrent()) { + endDedupe(); + return; + } + replay.liveEoseWaiter = endDedupe; + deadline = setTimeoutFn(endDedupe, LIVE_REQ_ACK_TIMEOUT_MS); + try { + await sendRaw([ + "REQ", + subId, + buildReconnectReplayFilter(subscription.filter, replaySince), + ]); + } catch (sendError: unknown) { + endDedupe(); + console.warn( + `[reconnect replay] history fallback REQ failed for ${subId}:`, + sendError, + ); + return; + } + // Same convergence as the restored live REQ: a disposal that raced + // this send must still end closed on the wire. + if (isActive() && !subscriptions.has(subId)) { + await sendRaw(["CLOSE", subId]).catch(() => {}); + } + }; + + // Take the repair's upper bound only after the relay acknowledged the + // live-only REQ (registered it), never before. One deadline per REQ: + // waiting again on the same unchanged REQ cannot make it arrive. + const acked = await awaitLiveEose( + replay, + isCurrent, + setTimeoutFn, + clearTimeoutFn, + ); + if (!isCurrent()) return; + if (!acked) { + console.warn( + `[reconnect replay] live REQ for ${subId} was not acknowledged; falling back to bounded history`, + ); + await sendHistoryFallback(); + return; + } for (let attempt = 1; attempt <= PAGE_REPLAY_MAX_ATTEMPTS; attempt++) { try { const completed = await replayReconnectHistoryPages({ @@ -377,8 +563,7 @@ export async function replayLiveSubscriptions({ // 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, + isActive: isCurrent, requestRepair, }); // A stale-connection abort is NOT completion: the superseding @@ -396,14 +581,14 @@ export async function replayLiveSubscriptions({ error, ); if (attempt === PAGE_REPLAY_MAX_ATTEMPTS) { - markReconnectRepairDone(subscription, generation); + await sendHistoryFallback(); 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; + if (!isCurrent()) return; } } }, diff --git a/desktop/src/shared/api/relayResumeCursor.test.mjs b/desktop/src/shared/api/relayResumeCursor.test.mjs new file mode 100644 index 0000000000..64aade8a9f --- /dev/null +++ b/desktop/src/shared/api/relayResumeCursor.test.mjs @@ -0,0 +1,166 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + STALL_CHECK_INTERVAL_MS, + STALL_IDLE_TIMEOUT_MS, +} from "./relayClientTimings.ts"; +import { + computeResumeSince, + creditInboundLiveness, + LIVENESS_MAX_GAP_MS, + RESUME_LAST_SEEN_SKEW_SECS, + RESUME_LIVE_UNTIL_SKEW_SECS, + RESUME_SILENT_DROP_WINDOW_SECS, + stampLiveUntilOnClose, +} from "./relayResumeCursor.ts"; + +test("resume skews match the shared reconnect contract", () => { + assert.equal(RESUME_LAST_SEEN_SKEW_SECS, 5); + // Relay created_at drift window (900s, pinned to ingest.rs in the replay + // test) plus a 60s client clock margin. + assert.equal(RESUME_LIVE_UNTIL_SKEW_SECS, 960); + // Matches the relay-side silent-drop repair window (one hour). + assert.equal(RESUME_SILENT_DROP_WINDOW_SECS, 3_600); +}); + +// since = max(lastSeen - 5, liveUntil - 960); the original filter's own `since` +// is applied by the replay path and covered in relayReconnectReplay.test.mjs. +for (const [name, input, expected] of [ + ["no cursor replays the original filter", {}, undefined], + ["lastSeen only", { lastSeenCreatedAt: 1_000 }, 995], + ["liveUntil only", { liveUntil: 1_000 }, 40], + ["newer lastSeen wins", { lastSeenCreatedAt: 1_000, liveUntil: 900 }, 995], + ["newer liveUntil wins", { lastSeenCreatedAt: 900, liveUntil: 2_000 }, 1_040], + [ + "a future-dated lastSeen is still the lower bound", + { lastSeenCreatedAt: 2_000, liveUntil: 1_000 }, + 1_995, + ], + ["never negative", { lastSeenCreatedAt: 3, liveUntil: 10 }, 0], + [ + "the liveUntil term never resumes later than the silent-drop window", + { liveUntil: 10_000, nowSecs: 10_000 }, + 10_000 - RESUME_SILENT_DROP_WINDOW_SECS, + ], + [ + "the window caps liveUntil only, never lastSeen", + { lastSeenCreatedAt: 10_000, liveUntil: 10_000, nowSecs: 10_000 }, + 9_995, + ], + [ + "an old liveUntil is below the window and is used as is", + { liveUntil: 5_000, nowSecs: 10_000 }, + 4_040, + ], + [ + "custom skews apply to each term", + { + lastSeenCreatedAt: 1_000, + liveUntil: 1_000, + lastSeenSkewSecs: 100, + liveUntilSkewSecs: 200, + }, + 900, + ], +]) { + test(`computeResumeSince: ${name}`, () => { + assert.equal(computeResumeSince(input), expected); + }); +} + +test("liveness credit survives only gaps the stall watchdog would have allowed", () => { + assert.equal( + LIVENESS_MAX_GAP_MS, + STALL_IDLE_TIMEOUT_MS + STALL_CHECK_INTERVAL_MS, + ); + assert.equal(creditInboundLiveness(0, 5_000), 5_000, "first frame credits"); + assert.equal( + creditInboundLiveness(5_000, 5_000 + LIVENESS_MAX_GAP_MS), + 5_000 + LIVENESS_MAX_GAP_MS, + "a gap at the limit still credits", + ); + assert.equal( + creditInboundLiveness(5_000, 5_001 + LIVENESS_MAX_GAP_MS), + 5_000, + "a frame processed after a longer gap may be queued through a sleep", + ); + // Credit stays at the pre-gap frame: later frames never re-arm it. + let credit = 5_000; + for (const nowMs of [3_600_000, 3_601_000, 3_602_000]) { + credit = creditInboundLiveness(credit, nowMs); + } + assert.equal(credit, 5_000); +}); + +function live(overrides = {}) { + return { mode: "live", filter: { kinds: [9], limit: 1 }, ...overrides }; +} + +test("stampLiveUntilOnClose stamps only subscriptions that reached EOSE on the closing generation", () => { + const eosed = live({ eoseGeneration: 7 }); + const otherGeneration = live({ eoseGeneration: 6, liveUntil: 500 }); + const never = live(); + const history = { mode: "history", eoseGeneration: 7 }; + const subscriptions = new Map([ + ["eosed", eosed], + ["older", otherGeneration], + ["never", never], + ["history", history], + ]); + + stampLiveUntilOnClose(subscriptions, 7, 1_234_999, 1_300_000); + + assert.equal(eosed.liveUntil, 1_234); + assert.equal( + otherGeneration.liveUntil, + 500, + "a subscription that missed EOSE on this connection keeps its old stamp", + ); + assert.equal(never.liveUntil, undefined); + assert.equal(history.liveUntil, undefined); + for (const subscription of [eosed, otherGeneration, never]) { + assert.equal(subscription.eoseGeneration, undefined); + } +}); + +test("stampLiveUntilOnClose never moves liveUntil backwards", () => { + const subscription = live({ eoseGeneration: 3, liveUntil: 9_999 }); + stampLiveUntilOnClose( + new Map([["a", subscription]]), + 3, + 5_000_000, + 5_000_000, + ); + assert.equal(subscription.liveUntil, 9_999); +}); + +test("stampLiveUntilOnClose caps the stamp at the last inbound frame of a stalled socket", () => { + const subscription = live({ eoseGeneration: 3 }); + // Stall detected 70s after the last frame: the close time overstates liveness. + stampLiveUntilOnClose( + new Map([["a", subscription]]), + 3, + 1_000_000, + 1_070_000, + ); + assert.equal(subscription.liveUntil, 1_000); +}); + +test("stampLiveUntilOnClose keeps the previous stamp when no inbound frame is known", () => { + const stamped = live({ eoseGeneration: 3, liveUntil: 700 }); + const fresh = live({ eoseGeneration: 3 }); + stampLiveUntilOnClose( + new Map([ + ["stamped", stamped], + ["fresh", fresh], + ]), + 3, + 0, + 9_000_000, + ); + assert.equal(stamped.liveUntil, 700, "never guessed from the close clock"); + assert.equal(fresh.liveUntil, undefined); + assert.equal(stamped.eoseGeneration, undefined); + assert.equal(fresh.eoseGeneration, undefined); +}); diff --git a/desktop/src/shared/api/relayResumeCursor.ts b/desktop/src/shared/api/relayResumeCursor.ts new file mode 100644 index 0000000000..964653c5b2 --- /dev/null +++ b/desktop/src/shared/api/relayResumeCursor.ts @@ -0,0 +1,141 @@ +import type { RelaySubscription } from "@/shared/api/relayClientShared"; +import { + STALL_CHECK_INTERVAL_MS, + STALL_IDLE_TIMEOUT_MS, +} from "@/shared/api/relayClientTimings"; + +/** + * Reconnect resume cursor. + * + * A subscription that reached EOSE on a connection was complete up to that + * EOSE and subscribed afterwards until the socket's last inbound frame, so a + * quiet subscription needs no history older than that on its next reconnect. + * Without this, a subscription that never received an event replays its + * ORIGINAL filter (startup `since`, full `limit`) on every reconnect, which + * multiplies reader load across the fleet during a rolling relay deploy. + * + * Subscribed is not delivered: the relay drops fan-out frames for a client + * whose send buffer is full while the drop count is under its grace limit + * (`ConnectionState::send` in `crates/buzz-relay/src/connection.rs`), so a + * quiet subscription can miss its only event and still count as live. The + * liveness term therefore never resumes later than + * `RESUME_SILENT_DROP_WINDOW_SECS` ago: a drop within that window is replayed + * on the next reconnect; an older one is left to the timeline's own refetch. + */ + +/** + * Every liveness-derived resume re-reads at least this much recent history, + * bounding how long a silently dropped fan-out frame can stay unrepaired. + */ +export const RESUME_SILENT_DROP_WINDOW_SECS = 3_600; + +/** Skew below the newest event seen (matches the pre-existing replay skew). */ +export const RESUME_LAST_SEEN_SKEW_SECS = 5; + +/** + * The relay rejects events whose `created_at` is more than this far from its + * clock in either direction (`MAX_TIMESTAMP_DRIFT_SECS` in + * `crates/buzz-relay/src/handlers/ingest.rs`; the replay test pins the + * coupling). A row stored after a close can therefore carry a `created_at` + * this far in the past. + */ +export const RELAY_INGEST_FUTURE_TOLERANCE_SECS = 900; + +/** Client-vs-relay clock skew margin on top of the relay's drift window. */ +export const RESUME_CLIENT_CLOCK_MARGIN_SECS = 60; + +/** + * Skew below the recorded live-until time: the relay's `created_at` drift + * window plus a client clock margin. Anything stored after the close has + * `created_at >= closeTime - 900s` on the relay clock, so a narrower window + * would skip a backdated event that was stored after the subscription went + * quiet. Paged channels use the wider channel lookback instead. + */ +export const RESUME_LIVE_UNTIL_SKEW_SECS = + RELAY_INGEST_FUTURE_TOLERANCE_SECS + RESUME_CLIENT_CLOCK_MARGIN_SECS; + +/** + * Lower bound for a reconnect replay, or `undefined` when the subscription has + * no cursor at all (never saw an event and never reached EOSE on a connection + * that has since closed) and must replay its original filter. + * + * `max(lastSeen - lastSeenSkew, min(liveUntil - liveUntilSkew, now - + * silentDropWindow))`: the callers apply the original filter's own `since` as + * a further floor and lower it with any pinned failed-backfill floor, so a + * recorded gap always wins. + */ +export function computeResumeSince({ + lastSeenCreatedAt, + liveUntil, + lastSeenSkewSecs = RESUME_LAST_SEEN_SKEW_SECS, + liveUntilSkewSecs = RESUME_LIVE_UNTIL_SKEW_SECS, + nowSecs = Math.floor(Date.now() / 1_000), +}: { + lastSeenCreatedAt?: number; + liveUntil?: number; + lastSeenSkewSecs?: number; + liveUntilSkewSecs?: number; + nowSecs?: number; +}): number | undefined { + const bounds: number[] = []; + if (lastSeenCreatedAt !== undefined) { + bounds.push(lastSeenCreatedAt - lastSeenSkewSecs); + } + if (liveUntil !== undefined) { + bounds.push( + Math.min( + liveUntil - liveUntilSkewSecs, + nowSecs - RESUME_SILENT_DROP_WINDOW_SECS, + ), + ); + } + return bounds.length === 0 ? undefined : Math.max(0, ...bounds); +} + +/** + * Longest silence a running renderer can see before the stall watchdog resets + * the socket. A frame processed after a longer silence may have been queued + * through a sleep, since frames are stamped when processed, not received. + */ +export const LIVENESS_MAX_GAP_MS = + STALL_IDLE_TIMEOUT_MS + STALL_CHECK_INTERVAL_MS; + +/** + * Next liveness credit for a connection after an inbound frame at `nowMs`. + * Once a gap longer than `LIVENESS_MAX_GAP_MS` is seen, credit stays at the + * last frame before it for the rest of the connection: whatever the relay + * sent (or dropped) during the gap was never confirmed. + */ +export function creditInboundLiveness(previousMs: number, nowMs: number) { + return previousMs === 0 || nowMs - previousMs <= LIVENESS_MAX_GAP_MS + ? nowMs + : previousMs; +} + +/** + * Record `liveUntil` for every live subscription that reached EOSE on the + * connection that is closing. Subscriptions that did not reach EOSE on it keep + * their previous `liveUntil` (if any): their history was never confirmed + * delivered on this connection. + * + * The stamp is the time of the connection's last inbound peer frame, never + * the close: after a sleep or a stalled socket, the close is processed long + * after the subscription last heard from the relay. When that time is unknown + * (`0`) the previous `liveUntil` is kept rather than guessed from the clock. + */ +export function stampLiveUntilOnClose( + subscriptions: Map, + generation: number, + /** Time of the connection's last inbound peer frame, or 0 when unknown. */ + lastInboundMs: number, + nowMs = Date.now(), +) { + const liveUntil = Math.floor(Math.min(nowMs, lastInboundMs) / 1_000); + for (const subscription of subscriptions.values()) { + if (subscription.mode !== "live") continue; + if (subscription.eoseGeneration === generation && lastInboundMs > 0) { + subscription.liveUntil = Math.max(subscription.liveUntil ?? 0, liveUntil); + } + subscription.eoseGeneration = undefined; + } +} diff --git a/desktop/tests/e2e/relay-reconnect.spec.ts b/desktop/tests/e2e/relay-reconnect.spec.ts index 7868e577da..89cf737ac3 100644 --- a/desktop/tests/e2e/relay-reconnect.spec.ts +++ b/desktop/tests/e2e/relay-reconnect.spec.ts @@ -35,20 +35,6 @@ async function disconnectMockWebsockets(page: import("@playwright/test").Page) { expect(disconnected).toBeGreaterThan(0); } -async function restartMockWebsockets(page: import("@playwright/test").Page) { - const restarted = await page.evaluate(() => { - const restart = ( - window as Window & { - __BUZZ_E2E_RESTART_MOCK_WEBSOCKETS__?: () => number; - } - ).__BUZZ_E2E_RESTART_MOCK_WEBSOCKETS__; - if (!restart) - throw new Error("E2E websocket restart seam is not installed."); - return restart(); - }); - expect(restarted).toBeGreaterThan(0); -} - async function setMockWebsocketUnavailable( page: import("@playwright/test").Page, unavailable: boolean, @@ -363,14 +349,16 @@ test("rate-limited reconnect backfill does not tear down the authenticated socke ).toBe("connected"); }); -test("service restart close resets accumulated backoff", async ({ page }) => { +test("service restart close redials after its spread draw, not the backoff", async ({ + page, +}) => { await installMockBridge(page, { websocketConnectErrors: ["down 1", "down 2", "down 3"], }); await page.goto("/"); - // Three rejected dials put the session deep in its backoff loop before it - // connects; wait for that connect rather than for the Tauri-backed channel - // list, which paints long before the websocket is up. + // Three rejected dials escalate the backoff before the session connects; + // wait for that connect rather than for the Tauri-backed channel list, + // which paints long before the websocket is up. await expect .poll( () => @@ -379,16 +367,30 @@ test("service restart close resets accumulated backoff", async ({ page }) => { ) .toBe("connected"); - const startedAt = Date.now(); - await restartMockWebsockets(page); + // Pin the restart spread draw: 0.6 of the 10s window is a 6s redial. The + // escalated backoff with the same draw is 1.05x a power of two (1.05s, + // 2.1s, 4.2s, 8.4s, ...), so only the spread lands in [5.8s, 7s). + const restart = await page.evaluate(() => { + Math.random = () => 0.6; + window.__BUZZ_E2E_RESET_WEBSOCKET_CONNECT_ATTEMPTS__?.(); + const at = Date.now(); + return { at, sockets: window.__BUZZ_E2E_RESTART_MOCK_WEBSOCKETS__?.() }; + }); + expect(restart.sockets).toBeGreaterThan(0); await expect .poll( () => page.evaluate(() => window.__BUZZ_E2E_GET_RELAY_CONNECTION_STATE__?.()), - { timeout: 2_500 }, + { timeout: 10_000 }, ) .toBe("connected"); - expect(Date.now() - startedAt).toBeLessThan(2_500); + const attempts = await page.evaluate( + () => window.__BUZZ_E2E_GET_WEBSOCKET_CONNECT_ATTEMPTS__?.() ?? [], + ); + expect(attempts.length).toBeGreaterThan(0); + const redialDelayMs = attempts[0] - restart.at; + expect(redialDelayMs).toBeGreaterThanOrEqual(5_800); + expect(redialDelayMs).toBeLessThan(7_000); }); test("passive relay watchdog does not write while the websocket is half-open", async ({