diff --git a/qa/scenarios/runtime/gateway-client-transport-defaults.yaml b/qa/scenarios/runtime/gateway-client-transport-defaults.yaml new file mode 100644 index 000000000000..3ec66ad17a4a --- /dev/null +++ b/qa/scenarios/runtime/gateway-client-transport-defaults.yaml @@ -0,0 +1,29 @@ +title: Gateway client transport defaults + +scenario: + id: gateway-client-transport-defaults + surface: runtime + category: gateway.protocol-compatibility + coverage: + primary: + - gateway.client-transport-defaults + objective: Verify the public Gateway client applies its documented request, handshake, reconnect, and tick-watch defaults over a real WebSocket transport. + successCriteria: + - Requests without an override time out after 30 seconds. + - A connected socket that receives no challenge closes after the 15 second challenge watchdog. + - Consecutive transport failures reconnect after 1, 2, 4, 8, and 16 seconds before capping at 30 seconds. + - The hello-ok tick interval replaces the pre-handshake watchdog basis. + - Missing ticks close the socket with code 4000 and the tick-timeout reason. + - Inbound ticks keep an unbounded request alive beyond the default request deadline. + docsRefs: + - docs/gateway/protocol.md + - docs/reference/test.md + codeRefs: + - packages/gateway-client/src/client.ts + - packages/gateway-client/src/protocol-client.ts + - packages/gateway-client/src/timeouts.ts + - test/e2e/qa-lab/runtime/gateway-client-transport-defaults.e2e.test.ts + execution: + kind: vitest + path: test/e2e/qa-lab/runtime/gateway-client-transport-defaults.e2e.test.ts + summary: Run the public GatewayClient against a real local WebSocketServer with deterministic post-connect timers for request, challenge, reconnect, and tick-watch defaults. diff --git a/test/e2e/qa-lab/runtime/gateway-client-transport-defaults.e2e.test.ts b/test/e2e/qa-lab/runtime/gateway-client-transport-defaults.e2e.test.ts new file mode 100644 index 000000000000..83f2749ce384 --- /dev/null +++ b/test/e2e/qa-lab/runtime/gateway-client-transport-defaults.e2e.test.ts @@ -0,0 +1,368 @@ +import { setImmediate as waitForImmediate } from "node:timers/promises"; +import { PROTOCOL_VERSION } from "@openclaw/gateway-protocol/version"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { type RawData, type WebSocket, WebSocketServer } from "ws"; +import { + GatewayClient, + GatewayClientRequestTimeoutError, + type GatewayClientOptions, +} from "../../../../packages/gateway-client/src/index.js"; + +type RequestFrame = { + id: string; + method: string; + params?: unknown; + type: "req"; +}; + +const clients: GatewayClient[] = []; +let server: WebSocketServer | undefined; + +function rawDataToString(data: RawData): string { + if (Array.isArray(data)) { + return Buffer.concat(data).toString("utf8"); + } + return Buffer.isBuffer(data) + ? data.toString("utf8") + : Buffer.from(new Uint8Array(data)).toString("utf8"); +} + +function parseRequest(data: RawData): RequestFrame { + return JSON.parse(rawDataToString(data)) as RequestFrame; +} + +function createDeferred(): { + promise: Promise; + reject: (reason?: unknown) => void; + resolve: (value: T) => void; +} { + let resolve!: (value: T) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((onResolve, onReject) => { + resolve = onResolve; + reject = onReject; + }); + return { promise, reject, resolve }; +} + +function enableFakeTimeAfterSocketEstablishment(): void { + vi.useFakeTimers({ + toFake: ["Date", "clearInterval", "clearTimeout", "setInterval", "setTimeout"], + }); +} + +async function flushSocketIo(turns = 2): Promise { + for (let index = 0; index < turns; index += 1) { + await waitForImmediate(); + } +} + +async function waitForCondition(predicate: () => boolean, description: string): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (predicate()) { + return; + } + await flushSocketIo(1); + } + throw new Error(`timed out waiting for ${description}`); +} + +async function listen( + onConnection: (socket: WebSocket, request: RequestFrame) => void, +): Promise { + server = new WebSocketServer({ host: "127.0.0.1", port: 0 }); + server.on("connection", (socket) => { + socket.on("message", (data) => onConnection(socket, parseRequest(data))); + }); + await new Promise((resolve) => { + server?.once("listening", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("gateway client defaults server did not get a TCP address"); + } + return `ws://127.0.0.1:${address.port}`; +} + +function sendChallenge(socket: WebSocket, sequence = 1): void { + socket.send( + JSON.stringify({ + type: "event", + event: "connect.challenge", + seq: sequence, + payload: { nonce: `defaults-${sequence}`, ts: 1_777_777_777_000 }, + }), + ); +} + +function sendHello(socket: WebSocket, requestId: string, tickIntervalMs: number): void { + socket.send( + JSON.stringify({ + type: "res", + id: requestId, + ok: true, + payload: { + type: "hello-ok", + protocol: PROTOCOL_VERSION, + server: { version: "gateway-client-defaults", connId: "defaults-connection" }, + features: { methods: ["chat.send", "status"], events: ["tick"] }, + snapshot: { + presence: [], + health: {}, + stateVersion: { presence: 1, health: 1 }, + uptimeMs: 1, + }, + auth: { role: "operator", scopes: ["operator.admin"] }, + policy: { + maxPayload: 512 * 1024, + maxBufferedBytes: 1024 * 1024, + tickIntervalMs, + }, + }, + }), + ); +} + +async function connectWithFakeTime(params: { + clientOptions?: Omit; + onRequest?: (socket: WebSocket, request: RequestFrame) => void; + tickIntervalMs: number; +}): Promise<{ client: GatewayClient; socket: WebSocket }> { + const socketReady = createDeferred(); + const helloReady = createDeferred(); + const url = await listen((socket, request) => { + if (request.method === "connect") { + sendHello(socket, request.id, params.tickIntervalMs); + return; + } + params.onRequest?.(socket, request); + }); + server?.once("connection", (socket) => { + // The real transport is established before timers become deterministic. + enableFakeTimeAfterSocketEstablishment(); + socketReady.resolve(socket); + }); + const client = new GatewayClient({ + ...params.clientOptions, + url, + onHelloOk: () => helloReady.resolve(undefined), + onConnectError: (error) => helloReady.reject(error), + }); + clients.push(client); + client.start(); + + const socket = await socketReady.promise; + await flushSocketIo(); + sendChallenge(socket); + await helloReady.promise; + return { client, socket }; +} + +afterEach(async () => { + for (const client of clients.splice(0)) { + client.stop(); + } + vi.useRealTimers(); + if (server) { + for (const socket of server.clients) { + socket.terminate(); + } + await new Promise((resolve) => { + server?.close(() => resolve()); + }); + server = undefined; + } +}); + +describe("GatewayClient transport defaults", () => { + it("uses a 30 second default request timeout", async () => { + const requestReady = createDeferred(); + const { client } = await connectWithFakeTime({ + tickIntervalMs: 60_000, + onRequest: (_socket, request) => { + if (request.method === "status") { + requestReady.resolve(undefined); + } + }, + }); + + let settled = false; + const outcome = client.request("status").then( + () => { + settled = true; + return undefined; + }, + (error: unknown) => { + settled = true; + return error; + }, + ); + await requestReady.promise; + + await vi.advanceTimersByTimeAsync(29_999); + expect(settled).toBe(false); + + await vi.advanceTimersByTimeAsync(1); + const error = await outcome; + expect(error).toBeInstanceOf(GatewayClientRequestTimeoutError); + expect(error).toMatchObject({ + message: "gateway request timeout for status", + method: "status", + requestSent: true, + timeoutMs: 30_000, + }); + }); + + it("enforces the 15 second connect-challenge watchdog", async () => { + const socketReady = createDeferred(); + const errors: Error[] = []; + const url = await listen(() => {}); + server?.once("connection", (socket) => { + enableFakeTimeAfterSocketEstablishment(); + socketReady.resolve(socket); + }); + const client = new GatewayClient({ + url, + onConnectError: (error) => errors.push(error), + }); + clients.push(client); + client.start(); + + const socket = await socketReady.promise; + const closed = new Promise<{ code: number; reason: string }>((resolve) => { + socket.once("close", (code: number, reason: Buffer) => { + resolve({ code, reason: reason.toString() }); + }); + }); + await flushSocketIo(); + + await vi.advanceTimersByTimeAsync(14_999); + expect(errors).toEqual([]); + + await vi.advanceTimersByTimeAsync(1); + expect(errors).toHaveLength(1); + expect(errors[0]?.message).toContain( + "gateway connect challenge timeout (waited 15000ms, limit 15000ms)", + ); + await expect(closed).resolves.toEqual({ + code: 1008, + reason: "connect challenge timeout", + }); + }); + + it("reconnects after 1/2/4 second delays capped at 30 seconds", async () => { + const sockets: WebSocket[] = []; + const closeEvents: Array<{ code: number; reason: string }> = []; + const firstSocket = createDeferred(); + const url = await listen(() => {}); + server?.on("connection", (socket) => { + sockets.push(socket); + if (sockets.length === 1) { + enableFakeTimeAfterSocketEstablishment(); + firstSocket.resolve(socket); + return; + } + waitForImmediate().then(() => socket.close(1012, "retry")); + }); + const client = new GatewayClient({ + url, + onClose: (code, reason) => closeEvents.push({ code, reason }), + }); + clients.push(client); + client.start(); + + const initialSocket = await firstSocket.promise; + await flushSocketIo(); + initialSocket.close(1012, "retry"); + + const expectedDelays = [1_000, 2_000, 4_000, 8_000, 16_000, 30_000, 30_000]; + for (const [index, delayMs] of expectedDelays.entries()) { + await waitForCondition(() => closeEvents.length >= index + 1, `close event ${index + 1}`); + const connectionCount = sockets.length; + + await vi.advanceTimersByTimeAsync(delayMs - 1); + await flushSocketIo(); + expect(sockets).toHaveLength(connectionCount); + + await vi.advanceTimersByTimeAsync(1); + await waitForCondition( + () => sockets.length === connectionCount + 1, + `reconnect ${index + 1}`, + ); + } + + expect(closeEvents).toHaveLength(expectedDelays.length); + expect(closeEvents).toEqual(expectedDelays.map(() => ({ code: 1012, reason: "retry" }))); + }); + + it("uses the server tick interval and closes missing ticks with code 4000", async () => { + const closeEvents: Array<{ code: number; reason: string }> = []; + const { socket } = await connectWithFakeTime({ + tickIntervalMs: 50, + clientOptions: { + tickWatchMinIntervalMs: 1, + onClose: (code, reason) => closeEvents.push({ code, reason }), + }, + }); + const serverClose = new Promise<{ code: number; reason: string }>((resolve) => { + socket.once("close", (code: number, reason: Buffer) => { + resolve({ code, reason: reason.toString() }); + }); + }); + + await vi.advanceTimersByTimeAsync(100); + expect(closeEvents).toEqual([]); + + await vi.advanceTimersByTimeAsync(50); + await expect(serverClose).resolves.toEqual({ code: 4000, reason: "tick timeout" }); + await waitForCondition(() => closeEvents.length === 1, "client tick-timeout close callback"); + expect(closeEvents[0]).toEqual({ code: 4000, reason: "tick timeout" }); + }); + + it("keeps an unbounded request alive while inbound ticks continue", async () => { + const requestReady = createDeferred(); + const closeEvents: Array<{ code: number; reason: string }> = []; + const { client, socket } = await connectWithFakeTime({ + tickIntervalMs: 5_000, + clientOptions: { + tickWatchMinIntervalMs: 1, + onClose: (code, reason) => closeEvents.push({ code, reason }), + }, + onRequest: (_socket, request) => { + if (request.method === "chat.send") { + requestReady.resolve(request); + } + }, + }); + const request = client.request<{ status: string }>( + "chat.send", + { message: "defaults" }, + { timeoutMs: null }, + ); + const requestFrame = await requestReady.promise; + + for (let sequence = 1; sequence <= 8; sequence += 1) { + await vi.advanceTimersByTimeAsync(4_000); + socket.send( + JSON.stringify({ + type: "event", + event: "tick", + seq: sequence, + payload: { ts: Date.now() }, + }), + ); + await flushSocketIo(); + } + + expect(closeEvents).toEqual([]); + socket.send( + JSON.stringify({ + type: "res", + id: requestFrame.id, + ok: true, + payload: { status: "ok" }, + }), + ); + await expect(request).resolves.toEqual({ status: "ok" }); + }); +});