diff --git a/desktop/src/features/channels/unreadReadMarker.test.mjs b/desktop/src/features/channels/unreadReadMarker.test.mjs index e05e2ba0b2..6ea0916413 100644 --- a/desktop/src/features/channels/unreadReadMarker.test.mjs +++ b/desktop/src/features/channels/unreadReadMarker.test.mjs @@ -18,6 +18,7 @@ import { } from "./useUnreadChannels.ts"; import { isChannelUnreadTriggerKind, + trackSeenEvent, withChannelTagFallback, } from "./useLiveChannelUpdates.ts"; import { @@ -90,6 +91,16 @@ test("live event with h tag is preserved", () => { assert.equal(withChannelTagFallback(message, "other-channel"), message); }); +test("notification event guard suppresses reconnect replay and stays bounded", () => { + const seen = new Set(); + + assert.equal(trackSeenEvent(seen, "event-a", 2), true); + assert.equal(trackSeenEvent(seen, "event-a", 2), false); + assert.equal(trackSeenEvent(seen, "event-b", 2), true); + assert.equal(trackSeenEvent(seen, "event-c", 2), true); + assert.deepEqual([...seen], ["event-b", "event-c"]); +}); + test("dmHuddleStart_isDmOnlyUnreadTrigger", () => { assert.equal( isChannelUnreadTriggerKind(KIND_HUDDLE_STARTED, true), diff --git a/desktop/src/features/channels/useLiveChannelUpdates.ts b/desktop/src/features/channels/useLiveChannelUpdates.ts index aeb3abb905..800467b6ea 100644 --- a/desktop/src/features/channels/useLiveChannelUpdates.ts +++ b/desktop/src/features/channels/useLiveChannelUpdates.ts @@ -110,13 +110,19 @@ function isExternalMentionEvent(event: RelayEvent, currentPubkey: string) { ); } -function trackSeenEvent(seenEventIds: Set, eventId: string): boolean { +const SEEN_NOTIFICATION_EVENT_LIMIT = 5_000; + +export function trackSeenEvent( + seenEventIds: Set, + eventId: string, + limit = 200, +): boolean { if (seenEventIds.has(eventId)) { return false; } seenEventIds.add(eventId); - if (seenEventIds.size > 200) { + if (seenEventIds.size > limit) { const oldestEventId = seenEventIds.values().next().value; if (oldestEventId) { seenEventIds.delete(oldestEventId); @@ -135,6 +141,11 @@ export function useLiveChannelUpdates( const normalizedCurrentPubkey = options.currentPubkey?.trim().toLowerCase() ?? ""; const seenMentionEventIdsRef = React.useRef(new Set()); + // Reconnect replay overlaps each live filter by five seconds so no message is + // lost at the boundary. Keep one shared guard for every notification side + // effect: the same event can be replayed repeatedly while a relay flaps, and + // mention events also arrive through both the channel and mention filters. + const seenNotificationEventIdsRef = React.useRef(new Set()); const channelsInvalidateRef = React.useRef(null); if (channelsInvalidateRef.current === null) { channelsInvalidateRef.current = createTrailingDebounce(() => { @@ -164,7 +175,6 @@ export function useLiveChannelUpdates( ), [channels], ); - const seenDmEventIdsRef = React.useRef(new Set()); const dmSubscriptionStartedAtRef = React.useRef(0); // Reset subscription timestamp when identity changes. @@ -181,44 +191,42 @@ export function useLiveChannelUpdates( [channels], ); - const handleDmEvent = React.useEffectEvent((event: RelayEvent) => { - // Only human-visible message kinds should fire DM notifications. - if (!isDmNotifiableKind(event.kind)) { - return; - } - - // Suppress backlog events that predate our subscription — these are - // historical replays, not live messages. - if (event.created_at < dmSubscriptionStartedAtRef.current) { - return; - } + const handleDmEvent = React.useEffectEvent( + (event: RelayEvent, isFirstNotificationDelivery: boolean) => { + // Only human-visible message kinds should fire DM notifications. + if (!isDmNotifiableKind(event.kind) || !isFirstNotificationDelivery) { + return; + } - const channelId = getChannelIdFromTags(event.tags); - if (!channelId) { - return; - } + // Suppress backlog events that predate our subscription — these are + // historical replays, not live messages. + if (event.created_at < dmSubscriptionStartedAtRef.current) { + return; + } - if (!isExternalMentionEvent(event, normalizedCurrentPubkey)) { - return; - } + const channelId = getChannelIdFromTags(event.tags); + if (!channelId) { + return; + } - const dmChannel = dmChannelMap.get(channelId); - if (!dmChannel) { - return; - } + if (!isExternalMentionEvent(event, normalizedCurrentPubkey)) { + return; + } - if (!trackSeenEvent(seenDmEventIdsRef.current, event.id)) { - return; - } + const dmChannel = dmChannelMap.get(channelId); + if (!dmChannel) { + return; + } - // Don't fire a notification for the channel the user is already viewing, - // unless the notify-while-viewing setting opts in. - if (channelId === activeChannelId && !options.notifyForActiveChannel) { - return; - } + // Don't fire a notification for the channel the user is already viewing, + // unless the notify-while-viewing setting opts in. + if (channelId === activeChannelId && !options.notifyForActiveChannel) { + return; + } - options.onDmMessage?.(event, dmChannel); - }); + options.onDmMessage?.(event, dmChannel); + }, + ); const handleIncomingMessage = React.useEffectEvent((event: RelayEvent) => { const channelId = getChannelIdFromTags(event.tags); @@ -226,12 +234,6 @@ export function useLiveChannelUpdates( return; } - // Track DM events even for the active channel so the dedup set stays - // current. The handler itself skips firing the notification callback - // when the user is already viewing the DM (unless opted in via - // notifyForActiveChannel). - handleDmEvent(event); - if (!liveChannelIds.has(channelId)) { if (channelId !== activeChannelId) { invalidateChannelsDebounced(); @@ -263,9 +265,21 @@ export function useLiveChannelUpdates( isUnreadTriggerKind && (normalizedCurrentPubkey.length === 0 || event.pubkey.toLowerCase() !== normalizedCurrentPubkey); + const isFirstNotificationDelivery = + !isExternalTriggerEvent || + trackSeenEvent( + seenNotificationEventIdsRef.current, + event.id, + SEEN_NOTIFICATION_EVENT_LIMIT, + ); const isThreadedReply = isThreadReply(event.tags); - if (isExternalTriggerEvent) { + // DM alerts and every other notification side effect share this delivery + // decision, preventing a replayed event from escaping through a second + // callback path. + handleDmEvent(event, isFirstNotificationDelivery); + + if (isExternalTriggerEvent && isFirstNotificationDelivery) { const shouldNotify = shouldNotifyForEvent( event, normalizedCurrentPubkey, diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 84ee10b68d..8274034ed5 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -49,7 +49,9 @@ import { isWebSocketClose, shouldRefuseConnect, shouldScheduleReconnect, + shouldWaitForScheduledReconnect, } from "@/shared/api/relayReconnectPolicy"; +import { RelayReconnectWaiters } from "@/shared/api/relayReconnectWaiters"; import { RelayStallWatchdog } from "@/shared/api/relayStallWatchdog"; import { closeWebSocket } from "@/shared/api/relayWebSocketClose"; import { buildThreadReferenceTags } from "@/features/messages/lib/threading"; @@ -57,26 +59,12 @@ const RECONNECT_BASE_DELAY_MS = 1_000, RECONNECT_MAX_DELAY_MS = 30_000, EVENT_BATCH_MS = 16; -/** - * Op-level timeout constants. Raised from 8 s to 25 s to survive degraded - * networks where TLS handshakes and DNS resolution can take 3–10 s. - */ export const AUTH_TIMEOUT_MS = 25_000; export const HISTORY_TIMEOUT_MS = 25_000; export const PUBLISH_TIMEOUT_MS = 25_000; -/** - * The connection must remain stable for this long after a successful AUTH - * before the reconnect backoff delay resets to its base value. Stability- - * gated reset prevents repeated fast reconnects (flapping) from erasing the - * backoff that throttles them. - */ export const BACKOFF_RESET_STABLE_MS = 60_000; -/** - * Passive liveness check. The relay sends heartbeat pings every 30s; if no - * inbound frame arrives for two heartbeat windows, treat the socket as stalled. - */ const STALL_CHECK_INTERVAL_MS = 10_000; const STALL_IDLE_TIMEOUT_MS = 60_000; @@ -85,6 +73,7 @@ export class RelayClient { private relayUrl: string | null = null; private connectPromise: Promise | null = null; private reconnectTimeout: number | null = null; + private reconnectWaiters = new RelayReconnectWaiters(); private reconnectDelayMs = RECONNECT_BASE_DELAY_MS; private keepAliveRequested = false; private authRequest: { @@ -105,16 +94,6 @@ export class RelayClient { private stabilityTimer: number | null = null; private visibleChannelId: string | null = null; - /** - * Sticky terminal flag. Set when `resetConnection` is called with - * `reconnect: false` (today: auth rejection). Acts as a hard guard against - * the reconnect-timer / retry-wrapper paths racing back to "reconnecting" - * after we've already declared the session dead. - * - * Cleared only on explicit user re-engagement: `disconnect()` (community - * switch — the singleton is being reused for a different community) and - * `preconnect()` (caller is asking us to come back up). - */ private terminal = false; private connectionStateEmitter = new RelayConnectionStateEmitter("idle"); @@ -127,21 +106,10 @@ export class RelayClient { }, }); - /** - * Track which channel the user is currently viewing so its subscriptions - * are sent first during reconnect replay — reducing visible latency on - * degraded networks where the relay REQ storm would otherwise delay all - * channels equally. - */ setVisibleChannelId(id: string | null) { this.visibleChannelId = id; } - /** - * Cleanly tear down the connection without scheduling a reconnect. - * Used during community switches to reset the singleton before the - * new community applies. - */ disconnect() { const error = new Error("Relay disconnected for community switch."); @@ -169,6 +137,7 @@ export class RelayClient { } this.connectPromise = null; + this.reconnectWaiters.settle(error); if (this.authRequest) { window.clearTimeout(this.authRequest.timeout); @@ -247,10 +216,6 @@ export class RelayClient { return this.fetchHistory(filter); } - /** - * Return the first event matching `filter` as soon as it arrives, without - * waiting for EOSE. Resolves to `null` when EOSE arrives before any event. - */ async fetchFirstEvent( filter: RelaySubscriptionFilter, ): Promise { @@ -462,10 +427,24 @@ export class RelayClient { async preconnect() { // Explicit re-engagement. If the session went terminal (auth rejection) - // the caller is asking us to try again, so clear the latch. + // the caller is asking us to try again, so clear the latch. A manual + // reconnect also bypasses the current delay once; ordinary operations do + // not, so background traffic cannot continuously defeat backoff. this.terminal = false; this.keepAliveRequested = true; - await this.ensureConnected(); + if (this.reconnectTimeout !== null) { + window.clearTimeout(this.reconnectTimeout); + this.reconnectTimeout = null; + } + try { + await this.ensureConnected(); + this.reconnectWaiters.settle(); + } catch (error) { + this.reconnectWaiters.settle( + this.normalizeRelayError(error, "Relay reconnect failed."), + ); + throw error; + } } subscribeToReconnects(listener: () => void) { @@ -508,9 +487,15 @@ export class RelayClient { return; } - if (this.reconnectTimeout) { - window.clearTimeout(this.reconnectTimeout); - this.reconnectTimeout = null; + if ( + shouldWaitForScheduledReconnect({ + hasPendingReconnect: this.reconnectTimeout !== null, + }) + ) { + // The reconnect coordinator owns outage pacing. Query, publish, and + // subscription callers must wait for its scheduled attempt instead of + // clearing the timer and creating an immediate reconnect storm. + return this.waitForScheduledReconnect(); } const connectPromise = this.connect(); @@ -964,6 +949,13 @@ export class RelayClient { } } + private waitForScheduledReconnect(): Promise { + if (this.reconnectTimeout === null) { + return this.ensureConnected(); + } + return this.reconnectWaiters.wait(); + } + private scheduleReconnect() { if ( !shouldScheduleReconnect({ @@ -989,9 +981,14 @@ export class RelayClient { this.reconnectTimeout = window.setTimeout(() => { this.reconnectTimeout = null; - void this.ensureConnected().catch(() => { - this.scheduleReconnect(); - }); + void this.ensureConnected() + .then(() => this.reconnectWaiters.settle()) + .catch((error) => { + this.reconnectWaiters.settle( + this.normalizeRelayError(error, "Relay reconnect failed."), + ); + this.scheduleReconnect(); + }); }, delay); } @@ -1049,6 +1046,9 @@ export class RelayClient { window.clearTimeout(this.reconnectTimeout); this.reconnectTimeout = null; } + if (options?.reconnect === false) { + this.reconnectWaiters.settle(error); + } if (this.wsId !== null) { void closeWebSocket(this.wsId, "connection reset"); diff --git a/desktop/src/shared/api/relayReconnectPolicy.test.mjs b/desktop/src/shared/api/relayReconnectPolicy.test.mjs index 6f375fb436..e6856ede18 100644 --- a/desktop/src/shared/api/relayReconnectPolicy.test.mjs +++ b/desktop/src/shared/api/relayReconnectPolicy.test.mjs @@ -6,6 +6,7 @@ import { isWebSocketClose, shouldRefuseConnect, shouldScheduleReconnect, + shouldWaitForScheduledReconnect, } from "./relayReconnectPolicy.ts"; // The "happy" baseline that *should* schedule a reconnect: not terminal, @@ -79,6 +80,17 @@ test("keep-alive alone is enough to schedule", () => { ); }); +test("ordinary operations wait for a scheduled reconnect instead of bypassing backoff", () => { + assert.equal( + shouldWaitForScheduledReconnect({ hasPendingReconnect: true }), + true, + ); + assert.equal( + shouldWaitForScheduledReconnect({ hasPendingReconnect: false }), + false, + ); +}); + test("shouldRefuseConnect mirrors terminal", () => { assert.equal(shouldRefuseConnect({ terminal: false }), false); assert.equal(shouldRefuseConnect({ terminal: true }), true); diff --git a/desktop/src/shared/api/relayReconnectPolicy.ts b/desktop/src/shared/api/relayReconnectPolicy.ts index a2cf358216..00d8e412bd 100644 --- a/desktop/src/shared/api/relayReconnectPolicy.ts +++ b/desktop/src/shared/api/relayReconnectPolicy.ts @@ -37,6 +37,12 @@ export function shouldScheduleReconnect(inputs: RelayReconnectInputs): boolean { return true; } +export function shouldWaitForScheduledReconnect(inputs: { + hasPendingReconnect: boolean; +}): boolean { + return inputs.hasPendingReconnect; +} + /** Whether `ensureConnected()` should refuse with a terminal error. */ export function shouldRefuseConnect(inputs: { terminal: boolean }): boolean { return inputs.terminal; diff --git a/desktop/src/shared/api/relayReconnectWaiters.test.mjs b/desktop/src/shared/api/relayReconnectWaiters.test.mjs new file mode 100644 index 0000000000..ddf3244526 --- /dev/null +++ b/desktop/src/shared/api/relayReconnectWaiters.test.mjs @@ -0,0 +1,26 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { RelayReconnectWaiters } from "./relayReconnectWaiters.ts"; + +test("settle releases every operation waiting on a successful reconnect", async () => { + const waiters = new RelayReconnectWaiters(); + const first = waiters.wait(); + const second = waiters.wait(); + + waiters.settle(); + + await Promise.all([first, second]); +}); + +test("settle rejects every operation after a failed reconnect", async () => { + const waiters = new RelayReconnectWaiters(); + const first = waiters.wait(); + const second = waiters.wait(); + const error = new Error("relay unavailable"); + + waiters.settle(error); + + await assert.rejects(first, error); + await assert.rejects(second, error); +}); diff --git a/desktop/src/shared/api/relayReconnectWaiters.ts b/desktop/src/shared/api/relayReconnectWaiters.ts new file mode 100644 index 0000000000..06ad1daa3f --- /dev/null +++ b/desktop/src/shared/api/relayReconnectWaiters.ts @@ -0,0 +1,21 @@ +export class RelayReconnectWaiters { + private waiters = new Set<{ + resolve: () => void; + reject: (error: Error) => void; + }>(); + + wait(): Promise { + return new Promise((resolve, reject) => { + this.waiters.add({ resolve, reject }); + }); + } + + settle(error?: Error) { + const waiters = [...this.waiters]; + this.waiters.clear(); + for (const waiter of waiters) { + if (error) waiter.reject(error); + else waiter.resolve(); + } + } +} diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index d15f2269d3..0074069acf 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -1100,6 +1100,11 @@ declare global { __BUZZ_E2E_SET_STALL_WEBSOCKET_SENDS__?: (stall: boolean) => void; __BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__?: () => number; __BUZZ_E2E_RESTART_MOCK_WEBSOCKETS__?: () => number; + __BUZZ_E2E_SET_MOCK_WEBSOCKET_UNAVAILABLE__?: ( + unavailable: boolean, + ) => void; + __BUZZ_E2E_GET_WEBSOCKET_CONNECT_ATTEMPTS__?: () => number[]; + __BUZZ_E2E_RESET_WEBSOCKET_CONNECT_ATTEMPTS__?: () => void; __BUZZ_E2E_SET_MESH__?: (mesh: { admitted?: boolean; models?: Array<{ id: string; name: string | null }>; @@ -2812,6 +2817,8 @@ const mockReminderEvents: RelayEvent[] = []; const mockPersonaEvents: RelayEvent[] = []; let mockRelayMembers: RawRelayMember[] = []; const mockSockets = new Map(); +let mockWebsocketUnavailable = false; +const relayWebsocketConnectAttemptStarts: number[] = []; let mockWebsocketSendMutexWedged = false; let mockClosedChannelLiveSubscription = false; const realSockets = new Map(); @@ -8922,6 +8929,7 @@ async function resolveGetEvent( } async function connectRealSocket(args: { url?: string; onMessage: unknown }) { + relayWebsocketConnectAttemptStarts.push(Date.now()); const wsId = nextSocketId++; const ws = new WebSocket(args.url ?? DEFAULT_RELAY_WS_URL); const handler = resolveHandler(args.onMessage); @@ -8950,6 +8958,10 @@ async function connectRealSocket(args: { url?: string; onMessage: unknown }) { } async function connectMockSocket(args: { onMessage: unknown }) { + relayWebsocketConnectAttemptStarts.push(Date.now()); + if (mockWebsocketUnavailable) { + throw new Error("mock relay unavailable"); + } const connectError = getConfig()?.mock?.websocketConnectErrors?.shift(); if (connectError) { throw new Error(connectError); @@ -9395,6 +9407,8 @@ export function maybeInstallE2eTauriMocks() { } mockClosedChannelLiveSubscription = false; + mockWebsocketUnavailable = false; + relayWebsocketConnectAttemptStarts.length = 0; mockGlobalAgentConfig = config.mock?.globalAgentConfig ? { ...config.mock.globalAgentConfig } : null; @@ -9618,6 +9632,16 @@ export function maybeInstallE2eTauriMocks() { } return sockets.length; }; + window.__BUZZ_E2E_SET_MOCK_WEBSOCKET_UNAVAILABLE__ = (unavailable) => { + mockWebsocketUnavailable = unavailable; + if (unavailable) relayWebsocketConnectAttemptStarts.length = 0; + }; + window.__BUZZ_E2E_GET_WEBSOCKET_CONNECT_ATTEMPTS__ = () => [ + ...relayWebsocketConnectAttemptStarts, + ]; + window.__BUZZ_E2E_RESET_WEBSOCKET_CONNECT_ATTEMPTS__ = () => { + relayWebsocketConnectAttemptStarts.length = 0; + }; // Tests vary mesh admission and models to exercise provider discovery and // the managed-agent start preflight. window.__BUZZ_E2E_SET_MESH__ = (mesh) => { diff --git a/desktop/tests/e2e/helpers/twoRelayHarness.ts b/desktop/tests/e2e/helpers/twoRelayHarness.ts index 98acd86398..43eb458417 100644 --- a/desktop/tests/e2e/helpers/twoRelayHarness.ts +++ b/desktop/tests/e2e/helpers/twoRelayHarness.ts @@ -162,6 +162,7 @@ export class TwoRelayHarness { BUZZ_METRICS_PORT: String(relay.ports.metrics), BUZZ_REQUIRE_AUTH_TOKEN: "false", BUZZ_RECONCILE_CHANNELS: "true", + BUZZ_AUTO_MIGRATE: "true", }); await this.waitForHealth(relay, child); } diff --git a/desktop/tests/e2e/relay-reconnect.spec.ts b/desktop/tests/e2e/relay-reconnect.spec.ts index 6240cca0ba..67ce725da8 100644 --- a/desktop/tests/e2e/relay-reconnect.spec.ts +++ b/desktop/tests/e2e/relay-reconnect.spec.ts @@ -49,6 +49,31 @@ async function restartMockWebsockets(page: import("@playwright/test").Page) { expect(restarted).toBeGreaterThan(0); } +async function setMockWebsocketUnavailable( + page: import("@playwright/test").Page, + unavailable: boolean, +) { + await page.evaluate((value) => { + const setUnavailable = window.__BUZZ_E2E_SET_MOCK_WEBSOCKET_UNAVAILABLE__; + if (!setUnavailable) { + throw new Error("E2E websocket availability seam is not installed."); + } + setUnavailable(value); + }, unavailable); +} + +async function getMockWebsocketConnectAttempts( + page: import("@playwright/test").Page, +) { + return page.evaluate(() => { + const getAttempts = window.__BUZZ_E2E_GET_WEBSOCKET_CONNECT_ATTEMPTS__; + if (!getAttempts) { + throw new Error("E2E websocket attempt seam is not installed."); + } + return getAttempts(); + }); +} + async function emitMockMessages( page: import("@playwright/test").Page, messages: Array<{ content: string; createdAt: number }>, @@ -121,6 +146,55 @@ test("failed initial relay dial retries automatically", async ({ page }) => { await expect(page.getByTestId("channel-general")).toBeVisible(); }); +test("routine traffic cannot bypass outage backoff and recovery stays automatic", async ({ + page, +}) => { + await page.goto("/"); + await expect(page.getByTestId("channel-general")).toBeVisible(); + + await setMockWebsocketUnavailable(page, true); + await disconnectMockWebsockets(page); + + // Exercise the production query path throughout the outage. Before the + // coordinator fix, each rejected query called ensureConnected(), cancelled + // the scheduled timer, and dialed immediately. The fixed session keeps these + // callers behind its single jittered exponential-backoff attempt. + await page.evaluate(async () => { + const deadline = Date.now() + 4_200; + while (Date.now() < deadline) { + await window.__BUZZ_E2E_QUERY_CLIENT__?.invalidateQueries({ + queryKey: ["channels"], + }); + await new Promise((resolve) => window.setTimeout(resolve, 100)); + } + }); + + const attempts = await getMockWebsocketConnectAttempts(page); + expect(attempts.length).toBeGreaterThanOrEqual(2); + expect(attempts.length).toBeLessThanOrEqual(3); + for (let index = 1; index < attempts.length; index += 1) { + expect(attempts[index] - attempts[index - 1]).toBeGreaterThanOrEqual(700); + } + + await setMockWebsocketUnavailable(page, false); + await expect + .poll( + () => + page.evaluate(() => window.__BUZZ_E2E_GET_RELAY_CONNECTION_STATE__?.()), + { timeout: 10_000 }, + ) + .toBe("connected"); + + const afterRecovery = `automatic outage recovery ${Date.now()}`; + await emitMockMessages(page, [ + { content: afterRecovery, createdAt: Math.floor(Date.now() / 1_000) }, + ]); + await page.getByTestId("channel-general").click(); + await expect(page.getByTestId("message-timeline")).toContainText( + afterRecovery, + ); +}); + test("service restart close resets accumulated backoff", async ({ page }) => { await installMockBridge(page, { websocketConnectErrors: ["down 1", "down 2", "down 3"], diff --git a/desktop/tests/e2e/relay-restart.live.spec.ts b/desktop/tests/e2e/relay-restart.live.spec.ts index 057ef46f52..f3c80e69a1 100644 --- a/desktop/tests/e2e/relay-restart.live.spec.ts +++ b/desktop/tests/e2e/relay-restart.live.spec.ts @@ -1,8 +1,13 @@ +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; + import { expect, test, type Page } from "@playwright/test"; -import { installBridge } from "../helpers/bridge"; +import { installBridge, TEST_IDENTITIES } from "../helpers/bridge"; import { TwoRelayHarness, type RelaySpec } from "./helpers/twoRelayHarness"; +const exec = promisify(execFile); + // Live gate: boots a REAL buzz-relay process, points the app at it, SIGTERMs // the relay mid-session, restarts it on the same port, and asserts the client // converges back to "connected". This proves the full restart story end to @@ -20,6 +25,55 @@ function required(name: string, value: string | undefined): string { return value; } +async function runCli(args: string[], relayUrl: string, privateKey: string) { + const binary = required("BUZZ_E2E_CLI_BIN", process.env.BUZZ_E2E_CLI_BIN); + const { stdout } = await exec(binary, args, { + cwd: "..", + env: { + ...process.env, + BUZZ_AUTH_TAG: "", + BUZZ_PRIVATE_KEY: privateKey, + BUZZ_RELAY_URL: relayUrl, + }, + }); + return stdout; +} + +async function seedLiveChannel(relayUrl: string) { + const name = `reconnect-live-${process.pid}`; + const created = JSON.parse( + await runCli( + [ + "channels", + "create", + "--name", + name, + "--type", + "stream", + "--visibility", + "open", + ], + relayUrl, + TEST_IDENTITIES.alice.privateKey, + ), + ) as { channel_id: string }; + await runCli( + [ + "channels", + "add-member", + "--channel", + created.channel_id, + "--pubkey", + TEST_IDENTITIES.tyler.pubkey, + "--role", + "member", + ], + relayUrl, + TEST_IDENTITIES.alice.privateKey, + ); + return { id: created.channel_id, name }; +} + async function connectionState(page: Page): Promise { return page.evaluate(() => { const win = window as Window & { @@ -29,13 +83,61 @@ async function connectionState(page: Page): Promise { }); } +async function exerciseBackgroundTraffic(page: Page, durationMs: number) { + await page.evaluate(async (duration) => { + const deadline = Date.now() + duration; + while (Date.now() < deadline) { + void window.__BUZZ_E2E_QUERY_CLIENT__?.invalidateQueries({ + queryKey: ["channels"], + }); + await new Promise((resolve) => window.setTimeout(resolve, 100)); + } + }, durationMs); +} + +async function resetConnectAttempts(page: Page) { + await page.evaluate(() => { + window.__BUZZ_E2E_RESET_WEBSOCKET_CONNECT_ATTEMPTS__?.(); + }); +} + +async function assertConnectAttemptsArePaced(page: Page) { + const attempts = await page.evaluate( + () => window.__BUZZ_E2E_GET_WEBSOCKET_CONNECT_ATTEMPTS__?.() ?? [], + ); + expect(attempts.length).toBeGreaterThanOrEqual(2); + expect(attempts.length).toBeLessThanOrEqual(4); + for (let index = 1; index < attempts.length; index += 1) { + expect(attempts[index] - attempts[index - 1]).toBeGreaterThanOrEqual(700); + } +} + +async function proveLiveDelivery( + page: Page, + relayUrl: string, + channel: { id: string; name: string }, + label: string, +) { + await page.getByTestId(`channel-${channel.name}`).click(); + await expect(page.getByTestId("chat-title")).toHaveText(channel.name); + const message = `${label} ${Date.now()}`; + await runCli( + ["messages", "send", "--channel", channel.id, "--content", message], + relayUrl, + TEST_IDENTITIES.alice.privateKey, + ); + await expect(page.getByTestId("message-timeline")).toContainText(message, { + timeout: 30_000, + }); +} + test.describe("relay restart live gate", () => { test.skip(!enabled, "set BUZZ_E2E_RELAY_RESTART=1 to run live gate"); test("client reconnects after the relay is SIGTERMed and restarted", async ({ page, }) => { - test.setTimeout(180_000); + test.setTimeout(240_000); const portBase = 26_000 + (process.pid % 3_000); const spec: RelaySpec = { name: "relay-restart", @@ -56,6 +158,8 @@ test.describe("relay restart live gate", () => { await harness.startRelays(); const relayHttpUrl = `http://127.0.0.1:${spec.ports.main}`; + const channel = await test.step("seed live channel and membership", () => + seedLiveChannel(relayHttpUrl)); await installBridge(page, { mode: "relay", user: "tyler", @@ -64,28 +168,68 @@ test.describe("relay restart live gate", () => { }); await page.goto("/"); - // Baseline: the app converges to a live authenticated session. - await expect - .poll(() => connectionState(page), { timeout: 60_000 }) - .toBe("connected"); + // Baseline: the app converges to a live authenticated session and sees + // the channel created for this fresh database. + await test.step("wait for initial authenticated connection", async () => { + await expect + .poll(() => connectionState(page), { timeout: 60_000 }) + .toBe("connected"); + await expect(page.getByTestId(`channel-${channel.name}`)).toBeVisible({ + timeout: 30_000, + }); + }); // Roll the pod. Graceful drain: readiness 503 → 5s grace → 1012 close // broadcast → process exit. The client must observe the close (not a // silent stall) and start retrying. + await resetConnectAttempts(page); await harness.terminateRelayGracefully(spec.name); await expect .poll(() => connectionState(page), { timeout: 30_000 }) .not.toBe("connected"); + // Keep the real relay unavailable across several reconnect windows while + // ordinary app traffic continues. This is the production-shaped race: + // background queries must not bypass the session coordinator's backoff. + await exerciseBackgroundTraffic(page, 8_000); + await expect.poll(() => connectionState(page)).not.toBe("connected"); + await assertConnectAttemptsArePaced(page); + // Bring the "new pod" up on the same address, exactly like a k8s // restart behind a stable service endpoint. await harness.restartRelay(spec.name); // The client's retry loop must find the fresh relay and converge back - // to connected without any user interaction. + // to connected without any user interaction, then prove that AUTH and + // live-subscription replay finished by receiving an event published by + // a second identity through the real CLI/relay boundary. await expect .poll(() => connectionState(page), { timeout: 60_000 }) .toBe("connected"); + await proveLiveDelivery( + page, + relayHttpUrl, + channel, + "first automatic recovery", + ); + + // Flap the fresh pod once more. A second recovery catches stale timer, + // waiter, generation, and subscription state that a single cycle cannot. + await harness.terminateRelayGracefully(spec.name); + await expect + .poll(() => connectionState(page), { timeout: 30_000 }) + .not.toBe("connected"); + await exerciseBackgroundTraffic(page, 4_000); + await harness.restartRelay(spec.name); + await expect + .poll(() => connectionState(page), { timeout: 60_000 }) + .toBe("connected"); + await proveLiveDelivery( + page, + relayHttpUrl, + channel, + "second automatic recovery", + ); } catch (error) { console.error(await harness.logs()); throw error;