From 98ad1b711b91d3e4780a3712b479efefe6f99311 Mon Sep 17 00:00:00 2001 From: Luke Policinski Date: Mon, 28 Sep 2026 14:43:08 -0400 Subject: [PATCH 1/3] bug: detect zombie websockets and recover realtime connections The /ws/web socket now expects the api's pong: once a connection has answered a ping, an unanswered ping past the timeout replaces the socket instead of pinging a dead TCP connection forever. Coming back to the tab or the network reconnects straight away, including after the 50-retry give-up, and late events from a replaced socket are ignored. The graphql-ws client pings every 15s, closes (4408) and terminates a socket that does not answer within 10s, and retries forever with a capped backoff so subscriptions come back after a long sleep. --- plugins/apollo.client.ts | 12 + tests/utilities/graphqlWsWatchdog.spec.ts | 146 ++++++++ tests/web-sockets/Socket.spec.ts | 392 ++++++++++++++++++++++ utilities/graphqlWsWatchdog.ts | 72 ++++ web-sockets/Socket.ts | 128 ++++++- 5 files changed, 744 insertions(+), 6 deletions(-) create mode 100644 tests/utilities/graphqlWsWatchdog.spec.ts create mode 100644 tests/web-sockets/Socket.spec.ts create mode 100644 utilities/graphqlWsWatchdog.ts diff --git a/plugins/apollo.client.ts b/plugins/apollo.client.ts index 981e79109..f3441ec68 100644 --- a/plugins/apollo.client.ts +++ b/plugins/apollo.client.ts @@ -16,6 +16,10 @@ import type { import { toast } from "@/components/ui/toast"; import { isAuthErrorMessage } from "~/graphql/isAuthError"; import { tournamentInviteErrorKey } from "~/utilities/tournamentInvites"; +import { + createGraphqlWsWatchdog, + graphqlWsRetryWait, +} from "~/utilities/graphqlWsWatchdog"; const mergeObjectFields = ( existing: Record | undefined, @@ -125,13 +129,21 @@ export default defineNuxtPlugin((nuxtApp) => { uri: `https://${config.public.apiDomain}/v1/graphql`, }); + const wsWatchdog = createGraphqlWsWatchdog(); + const wsClient = createClient({ url: `wss://${config.public.apiDomain}/v1/graphql`, connectionParams: { credentials: "include", }, + keepAlive: 15_000, + retryAttempts: Infinity, + retryWait: graphqlWsRetryWait, + on: wsWatchdog.on, }); + wsWatchdog.watch(wsClient); + nuxtApp.provide("wsClient", wsClient); const wsLink = new GraphQLWsLink(wsClient); diff --git a/tests/utilities/graphqlWsWatchdog.spec.ts b/tests/utilities/graphqlWsWatchdog.spec.ts new file mode 100644 index 000000000..e5d8abb46 --- /dev/null +++ b/tests/utilities/graphqlWsWatchdog.spec.ts @@ -0,0 +1,146 @@ +// @vitest-environment node +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + GRAPHQL_WS_MAX_RETRY_WAIT_MS, + GRAPHQL_WS_PONG_TIMEOUT_MS, + GRAPHQL_WS_TIMEOUT_CLOSE_CODE, + createGraphqlWsWatchdog, + graphqlWsRetryWait, +} from "~/utilities/graphqlWsWatchdog"; + +function watchedConnection() { + const watchdog = createGraphqlWsWatchdog(); + const socket = { close: vi.fn() }; + const client = { terminate: vi.fn() }; + + watchdog.watch(client); + watchdog.on.connected(socket); + + return { watchdog, socket, client }; +} + +describe("graphql-ws watchdog", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(console, "warn").mockImplementation(() => {}); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("closes and terminates a socket that never answers its ping", () => { + const { watchdog, socket, client } = watchedConnection(); + + watchdog.on.ping(false); + vi.advanceTimersByTime(GRAPHQL_WS_PONG_TIMEOUT_MS - 1); + + expect(socket.close).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1); + + expect(socket.close).toHaveBeenCalledWith( + GRAPHQL_WS_TIMEOUT_CLOSE_CODE, + "Request Timeout", + ); + expect(client.terminate).toHaveBeenCalledTimes(1); + }); + + it("stands down once the pong arrives", () => { + const { watchdog, socket, client } = watchedConnection(); + + watchdog.on.ping(false); + vi.advanceTimersByTime(GRAPHQL_WS_PONG_TIMEOUT_MS / 2); + watchdog.on.pong(true); + vi.advanceTimersByTime(GRAPHQL_WS_PONG_TIMEOUT_MS * 2); + + expect(socket.close).not.toHaveBeenCalled(); + expect(client.terminate).not.toHaveBeenCalled(); + }); + + it("stands down when the socket closes on its own", () => { + const { watchdog, socket, client } = watchedConnection(); + + watchdog.on.ping(false); + watchdog.on.closed({ code: 1006 }); + vi.advanceTimersByTime(GRAPHQL_WS_PONG_TIMEOUT_MS * 2); + + expect(socket.close).not.toHaveBeenCalled(); + expect(client.terminate).not.toHaveBeenCalled(); + }); + + it("ignores the server's pings and our replies to them", () => { + const { watchdog, socket, client } = watchedConnection(); + + watchdog.on.ping(true); + vi.advanceTimersByTime(GRAPHQL_WS_PONG_TIMEOUT_MS * 2); + + expect(socket.close).not.toHaveBeenCalled(); + + watchdog.on.ping(false); + watchdog.on.pong(false); + vi.advanceTimersByTime(GRAPHQL_WS_PONG_TIMEOUT_MS); + + expect(socket.close).toHaveBeenCalledTimes(1); + expect(client.terminate).toHaveBeenCalledTimes(1); + }); + + it("times each ping from when it was sent", () => { + const { watchdog, socket } = watchedConnection(); + + watchdog.on.ping(false); + vi.advanceTimersByTime(GRAPHQL_WS_PONG_TIMEOUT_MS - 1); + watchdog.on.pong(true); + watchdog.on.ping(false); + vi.advanceTimersByTime(GRAPHQL_WS_PONG_TIMEOUT_MS - 1); + + expect(socket.close).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1); + + expect(socket.close).toHaveBeenCalledTimes(1); + }); +}); + +describe("graphqlWsRetryWait", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + async function waitedMs(retries: number, random: number) { + vi.spyOn(Math, "random").mockReturnValue(random); + + let resolved = false; + const started = Date.now(); + void graphqlWsRetryWait(retries).then(() => { + resolved = true; + }); + + while (!resolved) { + await vi.advanceTimersByTimeAsync(10); + } + + return Date.now() - started; + } + + it("backs off exponentially", async () => { + expect(await waitedMs(0, 1)).toBeLessThanOrEqual(1_000); + expect(await waitedMs(3, 1)).toBeLessThanOrEqual(8_000); + expect(await waitedMs(3, 0)).toBeGreaterThanOrEqual(4_000); + }); + + it("never waits longer than the cap, however many retries", async () => { + expect(await waitedMs(1_000, 1)).toBeLessThanOrEqual( + GRAPHQL_WS_MAX_RETRY_WAIT_MS, + ); + expect(await waitedMs(1_000, 0)).toBeGreaterThanOrEqual( + GRAPHQL_WS_MAX_RETRY_WAIT_MS / 2, + ); + }); +}); diff --git a/tests/web-sockets/Socket.spec.ts b/tests/web-sockets/Socket.spec.ts new file mode 100644 index 000000000..5fad14f3c --- /dev/null +++ b/tests/web-sockets/Socket.spec.ts @@ -0,0 +1,392 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { MockInstance } from "vitest"; +import { Socket } from "~/web-sockets/Socket"; + +const HEARTBEAT_MS = 15_000; +const PONG_TIMEOUT_MS = 45_000; + +type Listener = (event: any) => void; + +class FakeWebSocket { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSING = 2; + static readonly CLOSED = 3; + + static instances: FakeWebSocket[] = []; + + readyState = FakeWebSocket.CONNECTING; + sent: Array<{ event: string; data?: unknown }> = []; + closeCalls = 0; + onclose: Listener | null = null; + onerror: Listener | null = null; + + private listeners = new Map(); + + constructor(public url: string) { + FakeWebSocket.instances.push(this); + } + + addEventListener(type: string, listener: Listener) { + this.listeners.set(type, [...(this.listeners.get(type) ?? []), listener]); + } + + send(payload: string) { + if (this.readyState === FakeWebSocket.CONNECTING) { + throw new Error("InvalidStateError: still CONNECTING"); + } + + this.sent.push(JSON.parse(payload)); + } + + // A dead TCP connection never gets far enough to report a close. + close() { + this.closeCalls++; + this.readyState = FakeWebSocket.CLOSING; + } + + open() { + this.readyState = FakeWebSocket.OPEN; + this.dispatch("open", {}); + } + + receive(event: string, data?: unknown) { + this.dispatch("message", { data: JSON.stringify({ event, data }) }); + } + + fail() { + this.readyState = FakeWebSocket.CLOSED; + this.onclose?.({ code: 1006 }); + } + + pings() { + return this.sent.filter(({ event }) => event === "ping").length; + } + + private dispatch(type: string, event: unknown) { + for (const listener of this.listeners.get(type) ?? []) { + listener(event); + } + } +} + +function latest() { + return FakeWebSocket.instances[FakeWebSocket.instances.length - 1]; +} + +function setVisibility(state: DocumentVisibilityState) { + Object.defineProperty(document, "visibilityState", { + configurable: true, + get: () => state, + }); + document.dispatchEvent(new Event("visibilitychange")); +} + +let socket: Socket; +let lifecycleSpies: Array<[EventTarget, MockInstance]> = []; + +beforeEach(() => { + vi.useFakeTimers(); + vi.stubGlobal("WebSocket", FakeWebSocket); + vi.spyOn(Math, "random").mockReturnValue(0); + vi.spyOn(console, "info").mockImplementation(() => {}); + vi.spyOn(console, "warn").mockImplementation(() => {}); + + lifecycleSpies = [ + [document, vi.spyOn(document, "addEventListener")], + [window, vi.spyOn(window, "addEventListener")], + ]; + + FakeWebSocket.instances = []; + socket = new Socket(); +}); + +afterEach(() => { + for (const [target, spy] of lifecycleSpies) { + for (const [type, listener] of spy.mock.calls) { + target.removeEventListener(type, listener); + } + } + + Reflect.deleteProperty(document, "visibilityState"); + vi.useRealTimers(); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); +}); + +describe("Socket pong watchdog", () => { + it("never reconnects a connection whose api has never answered a ping", () => { + socket.connect(); + latest().open(); + + vi.advanceTimersByTime(10 * 60_000); + + expect(FakeWebSocket.instances).toHaveLength(1); + expect(latest().pings()).toBeGreaterThan(30); + }); + + it("keeps a connection that keeps answering", () => { + socket.connect(); + const connection = latest(); + connection.open(); + + for (let tick = 0; tick < 40; tick++) { + connection.receive("pong"); + vi.advanceTimersByTime(HEARTBEAT_MS); + } + + expect(FakeWebSocket.instances).toHaveLength(1); + }); + + it("reconnects once an answering connection goes silent", () => { + const offline = vi.fn(); + const online = vi.fn(); + socket.on("offline", offline); + socket.on("online", online); + + socket.connect(); + const zombie = latest(); + zombie.open(); + zombie.receive("pong"); + + vi.advanceTimersByTime(PONG_TIMEOUT_MS); + + expect(FakeWebSocket.instances).toHaveLength(1); + + vi.advanceTimersByTime(HEARTBEAT_MS); + + expect(FakeWebSocket.instances).toHaveLength(2); + expect(zombie.closeCalls).toBe(1); + expect(offline).toHaveBeenCalledTimes(1); + + latest().open(); + + expect(online).toHaveBeenCalledTimes(2); + }); + + it("rejoins its rooms on the connection that replaces a zombie", () => { + socket.join("lobby", { type: "match", id: "match-1" }); + socket.connect(); + latest().open(); + latest().receive("pong"); + + vi.advanceTimersByTime(PONG_TIMEOUT_MS + HEARTBEAT_MS); + latest().open(); + + expect(latest().sent).toContainEqual({ + event: "lobby:join", + data: { type: "match", id: "match-1" }, + }); + }); + + it("does not arm on a new connection until that connection answers", () => { + socket.connect(); + latest().open(); + latest().receive("pong"); + + vi.advanceTimersByTime(PONG_TIMEOUT_MS + HEARTBEAT_MS); + + expect(FakeWebSocket.instances).toHaveLength(2); + + latest().open(); + vi.advanceTimersByTime(10 * 60_000); + + expect(FakeWebSocket.instances).toHaveLength(2); + }); + + it("ignores everything a replaced connection says after it is replaced", () => { + const event = vi.fn(); + const online = vi.fn(); + const offline = vi.fn(); + socket.on("match:updated", event); + socket.on("online", online); + socket.on("offline", offline); + + socket.connect(); + const replaced = latest(); + + socket.connect(); + const current = latest(); + + replaced.open(); + + expect(online).not.toHaveBeenCalled(); + expect(replaced.sent).toEqual([]); + + current.open(); + + expect(online).toHaveBeenCalledTimes(1); + + replaced.receive("match:updated", { id: "stale" }); + replaced.receive("pong"); + replaced.fail(); + + expect(event).not.toHaveBeenCalled(); + expect(offline).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(10 * 60_000); + + expect(FakeWebSocket.instances).toHaveLength(2); + + current.receive("match:updated", { id: "live" }); + + expect(event).toHaveBeenCalledWith({ id: "live" }); + }); + + it("stops the heartbeat while the connection is down", () => { + socket.connect(); + const dropped = latest(); + dropped.open(); + const pings = dropped.pings(); + + dropped.fail(); + vi.advanceTimersByTime(HEARTBEAT_MS * 2); + + expect(dropped.pings()).toBe(pings); + expect(latest()).not.toBe(dropped); + expect(latest().sent).toEqual([]); + }); +}); + +describe("Socket lifecycle recovery", () => { + function exhaustRetries() { + socket.connect(); + + for (let attempt = 0; attempt <= 50; attempt++) { + latest().fail(); + vi.runOnlyPendingTimers(); + } + } + + it("gives up after 50 failed reconnects", () => { + exhaustRetries(); + const attempts = FakeWebSocket.instances.length; + + vi.advanceTimersByTime(60 * 60_000); + + expect(attempts).toBe(51); + expect(FakeWebSocket.instances).toHaveLength(attempts); + }); + + it("reconnects when the tab comes back after giving up, from a fresh backoff", () => { + exhaustRetries(); + const attempts = FakeWebSocket.instances.length; + + setVisibility("hidden"); + + expect(FakeWebSocket.instances).toHaveLength(attempts); + + setVisibility("visible"); + + expect(FakeWebSocket.instances).toHaveLength(attempts + 1); + + latest().fail(); + vi.advanceTimersByTime(1_000); + + expect(FakeWebSocket.instances).toHaveLength(attempts + 2); + }); + + it("reconnects immediately when the network comes back", () => { + socket.connect(); + + for (let attempt = 0; attempt < 3; attempt++) { + latest().fail(); + vi.runOnlyPendingTimers(); + } + + latest().fail(); + const attempts = FakeWebSocket.instances.length; + + window.dispatchEvent(new Event("online")); + + expect(FakeWebSocket.instances).toHaveLength(attempts + 1); + + latest().open(); + vi.advanceTimersByTime(60_000); + + expect(FakeWebSocket.instances).toHaveLength(attempts + 1); + }); + + it("leaves a connection that is still being opened alone", () => { + socket.connect(); + latest().fail(); + vi.runOnlyPendingTimers(); + + expect(FakeWebSocket.instances).toHaveLength(2); + + window.dispatchEvent(new Event("online")); + setVisibility("visible"); + + expect(FakeWebSocket.instances).toHaveLength(2); + }); + + it("pings straight away when the tab comes back to a live connection", () => { + socket.connect(); + const connection = latest(); + connection.open(); + connection.receive("pong"); + const pings = connection.pings(); + + vi.advanceTimersByTime(5_000); + setVisibility("visible"); + + expect(connection.pings()).toBe(pings + 1); + expect(FakeWebSocket.instances).toHaveLength(1); + }); + + it("replaces a stale connection when the tab comes back", () => { + socket.connect(); + const zombie = latest(); + zombie.open(); + zombie.receive("pong"); + + vi.setSystemTime(Date.now() + 2 * 60 * 60_000); + setVisibility("visible"); + + expect(FakeWebSocket.instances).toHaveLength(1); + + vi.advanceTimersByTime(HEARTBEAT_MS); + + expect(FakeWebSocket.instances).toHaveLength(2); + expect(zombie.closeCalls).toBe(1); + }); + + it("binds the page listeners only once", () => { + socket.connect(); + socket.connect(); + socket.connect(); + + const [, documentSpy] = lifecycleSpies[0]; + const [, windowSpy] = lifecycleSpies[1]; + + expect( + documentSpy.mock.calls.filter(([type]) => type === "visibilitychange"), + ).toHaveLength(1); + expect( + windowSpy.mock.calls.filter(([type]) => type === "online"), + ).toHaveLength(1); + }); +}); + +describe("Socket.isStale", () => { + const now = 1_000_000; + + it("is never stale before the first pong", () => { + expect(Socket.isStale(false, 0, now - 10 * 60_000, now)).toBe(false); + }); + + it("is not stale while every ping has been answered", () => { + expect(Socket.isStale(true, now - 2 * 60_000, undefined, now)).toBe(false); + }); + + it("gives a fresh ping time to be answered", () => { + expect(Socket.isStale(true, now - 2 * 60_000, now - 5_000, now)).toBe( + false, + ); + }); + + it("is stale once a ping goes unanswered past the pong timeout", () => { + expect(Socket.isStale(true, now - 46_000, now - 31_000, now)).toBe(true); + expect(Socket.isStale(true, now - 44_000, now - 29_000, now)).toBe(false); + }); +}); diff --git a/utilities/graphqlWsWatchdog.ts b/utilities/graphqlWsWatchdog.ts new file mode 100644 index 000000000..16f3bdb8d --- /dev/null +++ b/utilities/graphqlWsWatchdog.ts @@ -0,0 +1,72 @@ +import type { Client, ClientOptions } from "graphql-ws"; + +type WatchedSocket = { + close: (code?: number, reason?: string) => void; +}; + +export const GRAPHQL_WS_PONG_TIMEOUT_MS = 10_000; +export const GRAPHQL_WS_TIMEOUT_CLOSE_CODE = 4408; +export const GRAPHQL_WS_MAX_RETRY_WAIT_MS = 30_000; + +export function createGraphqlWsWatchdog( + timeoutMs = GRAPHQL_WS_PONG_TIMEOUT_MS, +) { + let socket: WatchedSocket | undefined; + let client: Pick | undefined; + let pongTimer: ReturnType | undefined; + + function disarm() { + clearTimeout(pongTimer); + pongTimer = undefined; + } + + function timedOut() { + pongTimer = undefined; + console.warn(`[graphql-ws] no pong within ${timeoutMs}ms, reconnecting`); + + socket?.close(GRAPHQL_WS_TIMEOUT_CLOSE_CODE, "Request Timeout"); + + // A socket whose connection died never reports its close, and graphql-ws + // only retries once it hears one. + client?.terminate(); + } + + const on = { + connected: (connectedSocket: unknown) => { + socket = connectedSocket as WatchedSocket; + }, + ping: (received: boolean) => { + if (received) { + return; + } + + disarm(); + pongTimer = setTimeout(timedOut, timeoutMs); + }, + pong: (received: boolean) => { + if (received) { + disarm(); + } + }, + closed: () => { + disarm(); + socket = undefined; + }, + } satisfies ClientOptions["on"]; + + return { + on, + watch(watchedClient: Pick) { + client = watchedClient; + }, + }; +} + +export function graphqlWsRetryWait(retries: number) { + const ceiling = Math.min(1000 * 2 ** retries, GRAPHQL_WS_MAX_RETRY_WAIT_MS); + const wait = ceiling / 2 + Math.random() * (ceiling / 2); + + return new Promise((resolve) => { + setTimeout(resolve, wait); + }); +} diff --git a/web-sockets/Socket.ts b/web-sockets/Socket.ts index b9582e115..f48a7ad6a 100644 --- a/web-sockets/Socket.ts +++ b/web-sockets/Socket.ts @@ -70,11 +70,16 @@ export function chatMessageTime(message: LobbyMessage) { return new Date(message?.timestamp).getTime() || 0; } -class Socket extends EventEmitter { +export class Socket extends EventEmitter { private listening = new Set(); private connection?: WebSocket; private connected = false; private heartBeat?: NodeJS.Timeout; + private reconnectTimer?: ReturnType; + private lifecycleBound = false; + private pongSeen = false; + private lastPongAt = 0; + private unansweredPingAt?: number; private rejoinTimers: Map = new Map(); private offlineQueue: Array<{ event: string; @@ -88,6 +93,8 @@ class Socket extends EventEmitter { private static readonly MAX_RETRIES = 50; private static readonly BASE_DELAY_MS = 1000; private static readonly MAX_DELAY_MS = 30000; + private static readonly PONG_TIMEOUT_MS = 45_000; + private static readonly PONG_GRACE_MS = 10_000; private lobbies: Map = new Map(); private instanceCounter = 0; @@ -99,7 +106,44 @@ class Socket extends EventEmitter { } > = new Map(); + constructor() { + super(); + + this.on("pong", () => { + this.pongSeen = true; + this.lastPongAt = Date.now(); + this.unansweredPingAt = undefined; + }); + } + + // Only armed once this connection has answered a ping: an api that predates + // the pong reply would otherwise be reconnected every 45 seconds. + // + // Measured from the oldest unanswered ping and not only from the last pong, + // because a tab hidden for a few minutes runs the heartbeat once a minute and + // every pong would look late. + public static isStale( + pongSeen: boolean, + lastPongAt: number, + unansweredPingAt: number | undefined, + now: number, + ) { + if (!pongSeen || unansweredPingAt === undefined) { + return false; + } + + return ( + now - unansweredPingAt > Socket.PONG_GRACE_MS && + now - lastPongAt > Socket.PONG_TIMEOUT_MS + ); + } + public connect() { + this.bindLifecycle(); + + clearTimeout(this.reconnectTimer); + this.reconnectTimer = undefined; + // Clean up any existing connection before creating a new one if (this.connection) { try { @@ -119,21 +163,28 @@ class Socket extends EventEmitter { this.connection = webSocket; webSocket.addEventListener("message", (message) => { + if (this.connection !== webSocket) { + return; + } + const { event, data } = JSON.parse(message.data); this.emit(event, data); }); webSocket.addEventListener("open", () => { + if (this.connection !== webSocket) { + return; + } + this.emit("online"); this.connected = true; this.retryCount = 0; + this.pongSeen = false; + this.lastPongAt = 0; + this.unansweredPingAt = undefined; clearInterval(this.heartBeat); - if (!this.connection) { - return; - } - this.heartbeat(); this.heartBeat = setInterval(() => { @@ -157,6 +208,11 @@ class Socket extends EventEmitter { }); webSocket.onclose = (closeEvent) => { + if (this.connection !== webSocket) { + return; + } + + clearInterval(this.heartBeat); this.emit("offline"); this.connected = false; console.warn("[ws] lost connection to websocket server", closeEvent); @@ -179,7 +235,8 @@ class Socket extends EventEmitter { `[ws] reconnecting in ${Math.round(delay + jitter)}ms (attempt ${this.retryCount}/${Socket.MAX_RETRIES})`, ); - setTimeout(() => { + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = undefined; this.connect(); }, delay + jitter); }; @@ -189,6 +246,52 @@ class Socket extends EventEmitter { }; } + private forceReconnect() { + console.warn("[ws] no pong from the server, reconnecting"); + + clearInterval(this.heartBeat); + this.connected = false; + this.emit("offline"); + this.retryCount = 0; + this.connect(); + } + + private bindLifecycle() { + if ( + this.lifecycleBound || + typeof window === "undefined" || + typeof document === "undefined" + ) { + return; + } + + this.lifecycleBound = true; + + document.addEventListener("visibilitychange", () => { + if (document.visibilityState === "visible") { + this.resume(); + } + }); + + window.addEventListener("online", () => { + this.resume(); + }); + } + + private resume() { + if (this.connected) { + this.heartbeat(); + return; + } + + if (this.connection?.readyState === WebSocket.CONNECTING) { + return; + } + + this.retryCount = 0; + this.connect(); + } + private getRoomKey(room: string, data: Record) { const type = data.type ? String(data.type) : ""; const id = data.id ? String(data.id) : ""; @@ -306,6 +409,19 @@ class Socket extends EventEmitter { // the server expires a focus after a couple of heartbeats and a tab left open // on a conversation has to keep saying so. private heartbeat() { + const now = Date.now(); + + if ( + Socket.isStale(this.pongSeen, this.lastPongAt, this.unansweredPingAt, now) + ) { + this.forceReconnect(); + return; + } + + if (this.unansweredPingAt === undefined) { + this.unansweredPingAt = now; + } + this.connection?.send(JSON.stringify({ event: "ping" })); this.sendPresence(); } From 042861a3cacd7050e422aa1d9ff196f9b021bb87 Mon Sep 17 00:00:00 2001 From: Luke Policinski Date: Mon, 28 Sep 2026 15:00:52 -0400 Subject: [PATCH 2/3] bug: retry failed graphql-ws handshakes and recheck a resumed socket graphql-ws only retries close events by default, and a reconnect that fails its handshake reports an error first, which left subscriptions dead after waking offline. The retry wait keeps the library's spread (1s + up to 3s) under the 30s cap. A tab that comes back to a zombie is replaced ~10s after the resume ping instead of at a later heartbeat. connect() now owns the teardown so events queue while a replacement opens, and the offline queue flush no longer spins forever when the socket closes before it runs. --- plugins/apollo.client.ts | 16 +- tests/utilities/graphqlWsWatchdog.spec.ts | 188 ++++++++++++++++++++-- tests/web-sockets/Socket.spec.ts | 63 +++++++- utilities/graphqlWsWatchdog.ts | 32 +++- web-sockets/Socket.ts | 37 +++-- 5 files changed, 300 insertions(+), 36 deletions(-) diff --git a/plugins/apollo.client.ts b/plugins/apollo.client.ts index f3441ec68..54f216975 100644 --- a/plugins/apollo.client.ts +++ b/plugins/apollo.client.ts @@ -2,7 +2,6 @@ import { onError } from "@apollo/client/link/error"; import { RetryLink } from "@apollo/client/link/retry"; import { getMainDefinition } from "@apollo/client/utilities"; import { GraphQLWsLink } from "@apollo/client/link/subscriptions"; -import { createClient } from "graphql-ws"; import { DefaultApolloClient, provideApolloClient, @@ -16,10 +15,7 @@ import type { import { toast } from "@/components/ui/toast"; import { isAuthErrorMessage } from "~/graphql/isAuthError"; import { tournamentInviteErrorKey } from "~/utilities/tournamentInvites"; -import { - createGraphqlWsWatchdog, - graphqlWsRetryWait, -} from "~/utilities/graphqlWsWatchdog"; +import { createWatchedGraphqlWsClient } from "~/utilities/graphqlWsWatchdog"; const mergeObjectFields = ( existing: Record | undefined, @@ -129,21 +125,13 @@ export default defineNuxtPlugin((nuxtApp) => { uri: `https://${config.public.apiDomain}/v1/graphql`, }); - const wsWatchdog = createGraphqlWsWatchdog(); - - const wsClient = createClient({ + const wsClient = createWatchedGraphqlWsClient({ url: `wss://${config.public.apiDomain}/v1/graphql`, connectionParams: { credentials: "include", }, - keepAlive: 15_000, - retryAttempts: Infinity, - retryWait: graphqlWsRetryWait, - on: wsWatchdog.on, }); - wsWatchdog.watch(wsClient); - nuxtApp.provide("wsClient", wsClient); const wsLink = new GraphQLWsLink(wsClient); diff --git a/tests/utilities/graphqlWsWatchdog.spec.ts b/tests/utilities/graphqlWsWatchdog.spec.ts index e5d8abb46..e09e5bd96 100644 --- a/tests/utilities/graphqlWsWatchdog.spec.ts +++ b/tests/utilities/graphqlWsWatchdog.spec.ts @@ -1,10 +1,13 @@ // @vitest-environment node import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { + GRAPHQL_WS_KEEP_ALIVE_MS, GRAPHQL_WS_MAX_RETRY_WAIT_MS, GRAPHQL_WS_PONG_TIMEOUT_MS, + GRAPHQL_WS_RETRY_JITTER_MS, GRAPHQL_WS_TIMEOUT_CLOSE_CODE, createGraphqlWsWatchdog, + createWatchedGraphqlWsClient, graphqlWsRetryWait, } from "~/utilities/graphqlWsWatchdog"; @@ -129,18 +132,185 @@ describe("graphqlWsRetryWait", () => { return Date.now() - started; } - it("backs off exponentially", async () => { - expect(await waitedMs(0, 1)).toBeLessThanOrEqual(1_000); - expect(await waitedMs(3, 1)).toBeLessThanOrEqual(8_000); - expect(await waitedMs(3, 0)).toBeGreaterThanOrEqual(4_000); + it("backs off exponentially, spread by the jitter", async () => { + expect(await waitedMs(0, 0)).toBe(1_000); + expect(await waitedMs(0, 1)).toBe(1_000 + GRAPHQL_WS_RETRY_JITTER_MS); + expect(await waitedMs(3, 0)).toBe(8_000); }); it("never waits longer than the cap, however many retries", async () => { - expect(await waitedMs(1_000, 1)).toBeLessThanOrEqual( - GRAPHQL_WS_MAX_RETRY_WAIT_MS, - ); - expect(await waitedMs(1_000, 0)).toBeGreaterThanOrEqual( - GRAPHQL_WS_MAX_RETRY_WAIT_MS / 2, + expect(await waitedMs(1_000, 1)).toBe(GRAPHQL_WS_MAX_RETRY_WAIT_MS); + expect(await waitedMs(1_000, 0)).toBe( + GRAPHQL_WS_MAX_RETRY_WAIT_MS - GRAPHQL_WS_RETRY_JITTER_MS, ); }); }); + +class FakeGraphqlSocket { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSING = 2; + static readonly CLOSED = 3; + + static instances: FakeGraphqlSocket[] = []; + + readyState = FakeGraphqlSocket.CONNECTING; + sent: Array<{ type: string }> = []; + closes: Array<{ code?: number; reason?: string }> = []; + onopen: (() => void) | null = null; + onmessage: ((event: { data: string }) => void) | null = null; + onerror: ((event: unknown) => void) | null = null; + onclose: ((event: unknown) => void) | null = null; + + constructor(public url: string) { + FakeGraphqlSocket.instances.push(this); + } + + send(payload: string) { + this.sent.push(JSON.parse(payload)); + } + + // A dead TCP connection never gets far enough to report a close. + close(code?: number, reason?: string) { + this.closes.push({ code, reason }); + this.readyState = FakeGraphqlSocket.CLOSING; + } + + open() { + this.readyState = FakeGraphqlSocket.OPEN; + this.onopen?.(); + } + + receive(message: { type: string }) { + this.onmessage?.({ data: JSON.stringify(message) }); + } + + serverClose(code: number, reason = "") { + this.readyState = FakeGraphqlSocket.CLOSED; + this.onclose?.({ code, reason, wasClean: code !== 1006 }); + } + + failHandshake() { + this.readyState = FakeGraphqlSocket.CLOSED; + this.onerror?.({ type: "error" }); + this.onclose?.({ code: 1006, reason: "", wasClean: false }); + } + + types() { + return this.sent.map(({ type }) => type); + } +} + +describe("createWatchedGraphqlWsClient", () => { + let unsubscribe: (() => void) | undefined; + + beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0); + vi.spyOn(console, "warn").mockImplementation(() => {}); + FakeGraphqlSocket.instances = []; + }); + + afterEach(() => { + unsubscribe?.(); + unsubscribe = undefined; + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + function latest() { + return FakeGraphqlSocket.instances[FakeGraphqlSocket.instances.length - 1]; + } + + async function acknowledge(socket: FakeGraphqlSocket) { + socket.open(); + await vi.advanceTimersByTimeAsync(0); + socket.receive({ type: "connection_ack" }); + await vi.advanceTimersByTimeAsync(0); + } + + async function subscribed() { + const client = createWatchedGraphqlWsClient({ + url: "wss://api.test/v1/graphql", + webSocketImpl: FakeGraphqlSocket, + }); + const sink = { next: vi.fn(), error: vi.fn(), complete: vi.fn() }; + + unsubscribe = client.subscribe({ query: "subscription { ok }" }, sink); + await acknowledge(latest()); + + expect(latest().types()).toEqual(["connection_init", "subscribe"]); + + return sink; + } + + it("replaces a socket that stops answering and resubscribes", async () => { + const sink = await subscribed(); + const zombie = latest(); + + await vi.advanceTimersByTimeAsync(GRAPHQL_WS_KEEP_ALIVE_MS); + + expect(zombie.types()).toContain("ping"); + + await vi.advanceTimersByTimeAsync(GRAPHQL_WS_PONG_TIMEOUT_MS); + + expect(zombie.closes[0]).toEqual({ + code: GRAPHQL_WS_TIMEOUT_CLOSE_CODE, + reason: "Request Timeout", + }); + + await vi.advanceTimersByTimeAsync(1_000); + + expect(FakeGraphqlSocket.instances).toHaveLength(2); + + await acknowledge(latest()); + + expect(latest().types()).toEqual(["connection_init", "subscribe"]); + expect(sink.error).not.toHaveBeenCalled(); + }); + + it("keeps a socket that answers every ping", async () => { + await subscribed(); + const socket = latest(); + + for (let ping = 0; ping < 4; ping++) { + await vi.advanceTimersByTimeAsync(GRAPHQL_WS_KEEP_ALIVE_MS); + socket.receive({ type: "pong" }); + } + + await vi.advanceTimersByTimeAsync(GRAPHQL_WS_PONG_TIMEOUT_MS); + + expect(socket.types().filter((type) => type === "ping")).toHaveLength(4); + expect(socket.closes).toEqual([]); + expect(FakeGraphqlSocket.instances).toHaveLength(1); + }); + + it("keeps retrying through reconnects that fail their handshake", async () => { + const sink = await subscribed(); + + latest().serverClose(1006); + await vi.advanceTimersByTimeAsync(1_000); + + expect(FakeGraphqlSocket.instances).toHaveLength(2); + + latest().failHandshake(); + await vi.advanceTimersByTimeAsync(2_000); + + expect(FakeGraphqlSocket.instances).toHaveLength(3); + + await acknowledge(latest()); + + expect(latest().types()).toEqual(["connection_init", "subscribe"]); + expect(sink.error).not.toHaveBeenCalled(); + }); + + it("still gives up on a close graphql-ws treats as fatal", async () => { + const sink = await subscribed(); + + latest().serverClose(4400, "Bad request"); + await vi.advanceTimersByTimeAsync(GRAPHQL_WS_MAX_RETRY_WAIT_MS); + + expect(sink.error).toHaveBeenCalledTimes(1); + expect(FakeGraphqlSocket.instances).toHaveLength(1); + }); +}); diff --git a/tests/web-sockets/Socket.spec.ts b/tests/web-sockets/Socket.spec.ts index 5fad14f3c..dd6036af9 100644 --- a/tests/web-sockets/Socket.spec.ts +++ b/tests/web-sockets/Socket.spec.ts @@ -4,6 +4,7 @@ import { Socket } from "~/web-sockets/Socket"; const HEARTBEAT_MS = 15_000; const PONG_TIMEOUT_MS = 45_000; +const PONG_GRACE_MS = 10_000; type Listener = (event: any) => void; @@ -233,6 +234,51 @@ describe("Socket pong watchdog", () => { expect(event).toHaveBeenCalledWith({ id: "live" }); }); + it("queues events while a replacement connection is still opening", () => { + socket.connect(); + latest().open(); + + socket.connect(); + const replacement = latest(); + + expect(() => { + socket.event("match:ready", { id: "match-1" }); + }).not.toThrow(); + + vi.advanceTimersByTime(HEARTBEAT_MS * 2); + replacement.open(); + vi.advanceTimersByTime(100); + + expect(replacement.sent).toContainEqual({ + event: "match:ready", + data: { id: "match-1" }, + }); + }); + + it("keeps queued events for the next connection when one closes before flushing", () => { + socket.event("match:ready", { id: "match-1" }); + socket.connect(); + const shortLived = latest(); + shortLived.open(); + shortLived.fail(); + + vi.advanceTimersByTime(100); + + expect(shortLived.sent.map(({ event }) => event)).not.toContain( + "match:ready", + ); + + vi.advanceTimersByTime(1_000); + latest().open(); + vi.advanceTimersByTime(100); + + expect(latest()).not.toBe(shortLived); + expect(latest().sent).toContainEqual({ + event: "match:ready", + data: { id: "match-1" }, + }); + }); + it("stops the heartbeat while the connection is down", () => { socket.connect(); const dropped = latest(); @@ -342,15 +388,30 @@ describe("Socket lifecycle recovery", () => { vi.setSystemTime(Date.now() + 2 * 60 * 60_000); setVisibility("visible"); + vi.advanceTimersByTime(PONG_GRACE_MS); expect(FakeWebSocket.instances).toHaveLength(1); - vi.advanceTimersByTime(HEARTBEAT_MS); + vi.advanceTimersByTime(1); expect(FakeWebSocket.instances).toHaveLength(2); expect(zombie.closeCalls).toBe(1); }); + it("keeps a connection that answers when the tab comes back", () => { + socket.connect(); + const connection = latest(); + connection.open(); + connection.receive("pong"); + + vi.setSystemTime(Date.now() + 2 * 60 * 60_000); + setVisibility("visible"); + connection.receive("pong"); + vi.advanceTimersByTime(PONG_GRACE_MS + 1); + + expect(FakeWebSocket.instances).toHaveLength(1); + }); + it("binds the page listeners only once", () => { socket.connect(); socket.connect(); diff --git a/utilities/graphqlWsWatchdog.ts b/utilities/graphqlWsWatchdog.ts index 16f3bdb8d..604e2792a 100644 --- a/utilities/graphqlWsWatchdog.ts +++ b/utilities/graphqlWsWatchdog.ts @@ -1,12 +1,15 @@ +import { createClient } from "graphql-ws"; import type { Client, ClientOptions } from "graphql-ws"; type WatchedSocket = { close: (code?: number, reason?: string) => void; }; +export const GRAPHQL_WS_KEEP_ALIVE_MS = 15_000; export const GRAPHQL_WS_PONG_TIMEOUT_MS = 10_000; export const GRAPHQL_WS_TIMEOUT_CLOSE_CODE = 4408; export const GRAPHQL_WS_MAX_RETRY_WAIT_MS = 30_000; +export const GRAPHQL_WS_RETRY_JITTER_MS = 3_000; export function createGraphqlWsWatchdog( timeoutMs = GRAPHQL_WS_PONG_TIMEOUT_MS, @@ -63,10 +66,35 @@ export function createGraphqlWsWatchdog( } export function graphqlWsRetryWait(retries: number) { - const ceiling = Math.min(1000 * 2 ** retries, GRAPHQL_WS_MAX_RETRY_WAIT_MS); - const wait = ceiling / 2 + Math.random() * (ceiling / 2); + const backoff = Math.min( + 1000 * 2 ** retries, + GRAPHQL_WS_MAX_RETRY_WAIT_MS - GRAPHQL_WS_RETRY_JITTER_MS, + ); + const wait = backoff + Math.random() * GRAPHQL_WS_RETRY_JITTER_MS; return new Promise((resolve) => { setTimeout(resolve, wait); }); } + +export function createWatchedGraphqlWsClient( + options: Pick, +) { + const watchdog = createGraphqlWsWatchdog(); + + const client = createClient({ + ...options, + keepAlive: GRAPHQL_WS_KEEP_ALIVE_MS, + retryAttempts: Infinity, + retryWait: graphqlWsRetryWait, + // The default only retries close events, and a reconnect that fails its + // handshake reports an error first. graphql-ws still gives up on the + // close codes it treats as fatal (auth and protocol failures). + shouldRetry: () => true, + on: watchdog.on, + }); + + watchdog.watch(client); + + return client; +} diff --git a/web-sockets/Socket.ts b/web-sockets/Socket.ts index f48a7ad6a..29ba89efc 100644 --- a/web-sockets/Socket.ts +++ b/web-sockets/Socket.ts @@ -76,6 +76,7 @@ export class Socket extends EventEmitter { private connected = false; private heartBeat?: NodeJS.Timeout; private reconnectTimer?: ReturnType; + private resumeCheck?: ReturnType; private lifecycleBound = false; private pongSeen = false; private lastPongAt = 0; @@ -143,6 +144,9 @@ export class Socket extends EventEmitter { clearTimeout(this.reconnectTimer); this.reconnectTimer = undefined; + clearTimeout(this.resumeCheck); + clearInterval(this.heartBeat); + this.connected = false; // Clean up any existing connection before creating a new one if (this.connection) { @@ -198,11 +202,12 @@ export class Socket extends EventEmitter { } setTimeout(() => { - for (let i = 0; i < this.offlineQueue.length; i++) { - const { event, data } = this.offlineQueue[i]; + if (this.connection !== webSocket || !this.connected) { + return; + } + + for (const { event, data } of this.offlineQueue.splice(0)) { this.event(event, data); - this.offlineQueue.shift(); - i--; } }, 100); }); @@ -249,7 +254,6 @@ export class Socket extends EventEmitter { private forceReconnect() { console.warn("[ws] no pong from the server, reconnecting"); - clearInterval(this.heartBeat); this.connected = false; this.emit("offline"); this.retryCount = 0; @@ -279,17 +283,30 @@ export class Socket extends EventEmitter { } private resume() { - if (this.connected) { - this.heartbeat(); + if (!this.connected) { + if (this.connection?.readyState === WebSocket.CONNECTING) { + return; + } + + this.retryCount = 0; + this.connect(); return; } - if (this.connection?.readyState === WebSocket.CONNECTING) { + this.heartbeat(); + + if (!this.connected) { return; } - this.retryCount = 0; - this.connect(); + // A zombie only gives itself away by not answering, so look again as soon + // as this ping is overdue instead of waiting for the next heartbeat. + clearTimeout(this.resumeCheck); + this.resumeCheck = setTimeout(() => { + if (this.connected) { + this.heartbeat(); + } + }, Socket.PONG_GRACE_MS + 1); } private getRoomKey(room: string, data: Record) { From f905ec32df63894a1f489114cd2e674ee70ed534 Mon Sep 17 00:00:00 2001 From: Luke Policinski Date: Mon, 28 Sep 2026 20:37:26 -0400 Subject: [PATCH 3/3] test: prove the offline-queue hang, heartbeat stop and graphql-ws retry limit are fixed --- tests/utilities/graphqlWsWatchdog.spec.ts | 20 ++++++++++++++++++++ tests/web-sockets/Socket.spec.ts | 13 ++++++++++++- 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/tests/utilities/graphqlWsWatchdog.spec.ts b/tests/utilities/graphqlWsWatchdog.spec.ts index e09e5bd96..1d449c509 100644 --- a/tests/utilities/graphqlWsWatchdog.spec.ts +++ b/tests/utilities/graphqlWsWatchdog.spec.ts @@ -304,6 +304,26 @@ describe("createWatchedGraphqlWsClient", () => { expect(sink.error).not.toHaveBeenCalled(); }); + it("keeps retrying past graphql-ws's default of five attempts", async () => { + const sink = await subscribed(); + + latest().serverClose(1006); + + for (let attempt = 1; attempt <= 8; attempt++) { + await vi.advanceTimersByTimeAsync(GRAPHQL_WS_MAX_RETRY_WAIT_MS); + + expect(FakeGraphqlSocket.instances).toHaveLength(attempt + 1); + + latest().failHandshake(); + } + + await vi.advanceTimersByTimeAsync(GRAPHQL_WS_MAX_RETRY_WAIT_MS); + await acknowledge(latest()); + + expect(latest().types()).toEqual(["connection_init", "subscribe"]); + expect(sink.error).not.toHaveBeenCalled(); + }); + it("still gives up on a close graphql-ws treats as fatal", async () => { const sink = await subscribed(); diff --git a/tests/web-sockets/Socket.spec.ts b/tests/web-sockets/Socket.spec.ts index dd6036af9..4a32ab923 100644 --- a/tests/web-sockets/Socket.spec.ts +++ b/tests/web-sockets/Socket.spec.ts @@ -84,6 +84,7 @@ function setVisibility(state: DocumentVisibilityState) { } let socket: Socket; +let events: MockInstance; let lifecycleSpies: Array<[EventTarget, MockInstance]> = []; beforeEach(() => { @@ -100,6 +101,14 @@ beforeEach(() => { FakeWebSocket.instances = []; socket = new Socket(); + + const event = socket.event.bind(socket); + events = vi.spyOn(socket, "event").mockImplementation((name, data) => { + // Capped so a flush that re-queues forever fails a test instead of hanging. + if (events.mock.calls.length <= 1000) { + event(name, data); + } + }); }); afterEach(() => { @@ -209,7 +218,7 @@ describe("Socket pong watchdog", () => { socket.connect(); const current = latest(); - replaced.open(); + expect(() => replaced.open()).not.toThrow(); expect(online).not.toHaveBeenCalled(); expect(replaced.sent).toEqual([]); @@ -264,6 +273,7 @@ describe("Socket pong watchdog", () => { vi.advanceTimersByTime(100); + expect(events).toHaveBeenCalledTimes(1); expect(shortLived.sent.map(({ event }) => event)).not.toContain( "match:ready", ); @@ -283,6 +293,7 @@ describe("Socket pong watchdog", () => { socket.connect(); const dropped = latest(); dropped.open(); + vi.advanceTimersByTime(HEARTBEAT_MS - 500); const pings = dropped.pings(); dropped.fail();