diff --git a/plugins/apollo.client.ts b/plugins/apollo.client.ts index 981e79109..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,6 +15,7 @@ import type { import { toast } from "@/components/ui/toast"; import { isAuthErrorMessage } from "~/graphql/isAuthError"; import { tournamentInviteErrorKey } from "~/utilities/tournamentInvites"; +import { createWatchedGraphqlWsClient } from "~/utilities/graphqlWsWatchdog"; const mergeObjectFields = ( existing: Record | undefined, @@ -125,7 +125,7 @@ export default defineNuxtPlugin((nuxtApp) => { uri: `https://${config.public.apiDomain}/v1/graphql`, }); - const wsClient = createClient({ + const wsClient = createWatchedGraphqlWsClient({ url: `wss://${config.public.apiDomain}/v1/graphql`, connectionParams: { credentials: "include", diff --git a/tests/utilities/graphqlWsWatchdog.spec.ts b/tests/utilities/graphqlWsWatchdog.spec.ts new file mode 100644 index 000000000..1d449c509 --- /dev/null +++ b/tests/utilities/graphqlWsWatchdog.spec.ts @@ -0,0 +1,336 @@ +// @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"; + +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, 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)).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("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(); + + 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 new file mode 100644 index 000000000..4a32ab923 --- /dev/null +++ b/tests/web-sockets/Socket.spec.ts @@ -0,0 +1,464 @@ +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; +const PONG_GRACE_MS = 10_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 events: MockInstance; +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(); + + 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(() => { + 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(); + + expect(() => replaced.open()).not.toThrow(); + + 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("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(events).toHaveBeenCalledTimes(1); + 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(); + dropped.open(); + vi.advanceTimersByTime(HEARTBEAT_MS - 500); + 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"); + vi.advanceTimersByTime(PONG_GRACE_MS); + + expect(FakeWebSocket.instances).toHaveLength(1); + + 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(); + 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..604e2792a --- /dev/null +++ b/utilities/graphqlWsWatchdog.ts @@ -0,0 +1,100 @@ +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, +) { + 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 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 b9582e115..29ba89efc 100644 --- a/web-sockets/Socket.ts +++ b/web-sockets/Socket.ts @@ -70,11 +70,17 @@ 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 resumeCheck?: 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 +94,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 +107,47 @@ 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; + clearTimeout(this.resumeCheck); + clearInterval(this.heartBeat); + this.connected = false; + // Clean up any existing connection before creating a new one if (this.connection) { try { @@ -119,21 +167,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(() => { @@ -147,16 +202,22 @@ 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); }); 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 +240,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 +251,64 @@ class Socket extends EventEmitter { }; } + private forceReconnect() { + console.warn("[ws] no pong from the server, reconnecting"); + + 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) { + if (this.connection?.readyState === WebSocket.CONNECTING) { + return; + } + + this.retryCount = 0; + this.connect(); + return; + } + + this.heartbeat(); + + if (!this.connected) { + return; + } + + // 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) { const type = data.type ? String(data.type) : ""; const id = data.id ? String(data.id) : ""; @@ -306,6 +426,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(); }