From 3a55867ea1ecc0ffb0e4b2bc66e310c046bc50ce Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 11 Aug 2026 10:10:20 -0700 Subject: [PATCH] fix(gateway): bound stalled session subscriptions (#121164) --- .../gateway-runtime.json | 2 +- packages/gateway-client/README.md | 3 + .../src/client.request-timeout.test.ts | 7 +- packages/gateway-client/src/client.ts | 20 +- .../src/client.watchdog.test.ts | 18 +- .../gateway-client/src/pending-request.ts | 227 +++++++++++- .../src/protocol-client.request.test.ts | 350 ++++++++++++++++++ .../gateway-client/src/protocol-client.ts | 167 ++------- .../gateway-client/src/protocol-request.ts | 19 + .../lib/sessions/connection-hydration.test.ts | 90 +++++ .../sessions/session-event-subscription.ts | 19 +- 11 files changed, 754 insertions(+), 168 deletions(-) create mode 100644 packages/gateway-client/src/protocol-client.request.test.ts diff --git a/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json b/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json index 12c5ad7f3fda..7bc27301cb45 100644 --- a/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json +++ b/docs/.generated/plugin-sdk-api-baseline/gateway-runtime.json @@ -1 +1 @@ -{"contentHash":"1456c41077c880274a40b67c96c974ec8b22736b77abe8cec9966d0725fe18d3","entrypoint":"gateway-runtime","importSpecifier":"openclaw/plugin-sdk/gateway-runtime"} +{"contentHash":"737ad952cf6bb0f9d65d97bd17498af686dae166716274017fd26e5f602981d2","entrypoint":"gateway-runtime","importSpecifier":"openclaw/plugin-sdk/gateway-runtime"} diff --git a/packages/gateway-client/README.md b/packages/gateway-client/README.md index 8a4440033928..3994729cae16 100644 --- a/packages/gateway-client/README.md +++ b/packages/gateway-client/README.md @@ -138,6 +138,9 @@ from `@openclaw/gateway-protocol`, not from bundled implementation paths. the socket; `stop()` closes it and rejects pending requests. - A request uses `request(method, params)` after `hello-ok`. Passing `timeoutMs: null` creates an intentionally unbounded request. +- Finite request deadlines reject with `GatewayProtocolRequestTimeoutError`, + whose `CLIENT_TIMEOUT` code, method, deadline, and send-boundary flag remain + distinct from authoritative Gateway response errors. - Device identity persistence, signing, proxy routing, TLS formatting, and logging stay host-owned through `GatewayClientHostDeps`. - Protocol changes are additive first. Incompatible changes require an explicit diff --git a/packages/gateway-client/src/client.request-timeout.test.ts b/packages/gateway-client/src/client.request-timeout.test.ts index e9d4dbcfd0b3..e832af8a61d7 100644 --- a/packages/gateway-client/src/client.request-timeout.test.ts +++ b/packages/gateway-client/src/client.request-timeout.test.ts @@ -1,6 +1,9 @@ import { afterEach, expect, test, vi } from "vitest"; import { GatewayClient, GatewayClientRequestTimeoutError } from "./client.js"; -import type { GatewayProtocolSocket } from "./protocol-client.js"; +import { + GatewayProtocolRequestTimeoutError, + type GatewayProtocolSocket, +} from "./protocol-client.js"; afterEach(() => { vi.useRealTimers(); @@ -28,7 +31,9 @@ test("reports that a timed-out request crossed the transport send boundary", asy const error = await outcome; expect(error).toBeInstanceOf(GatewayClientRequestTimeoutError); + expect(error).toBeInstanceOf(GatewayProtocolRequestTimeoutError); expect(error).toMatchObject({ + code: "CLIENT_TIMEOUT", method: "node.invoke", timeoutMs: 100, requestSent: true, diff --git a/packages/gateway-client/src/client.ts b/packages/gateway-client/src/client.ts index 9d02e9c311c7..d1799a4d763c 100644 --- a/packages/gateway-client/src/client.ts +++ b/packages/gateway-client/src/client.ts @@ -42,7 +42,10 @@ import { type GatewayProtocolSocket, type GatewayProtocolSocketHandlers, } from "./protocol-client.js"; -import { GatewayProtocolRequestError } from "./protocol-request.js"; +import { + GatewayProtocolRequestError, + GatewayProtocolRequestTimeoutError, +} from "./protocol-request.js"; import { shouldPauseGatewayReconnect } from "./reconnect-policy.js"; import { GatewayClientRequestError } from "./request-error.js"; import { @@ -242,17 +245,10 @@ export type GatewayClientCloseInfo = { export { GatewayClientRequestError } from "./request-error.js"; -export class GatewayClientRequestTimeoutError extends Error { - readonly method: string; - readonly timeoutMs: number; - readonly requestSent: boolean; - +export class GatewayClientRequestTimeoutError extends GatewayProtocolRequestTimeoutError { constructor(params: { method: string; timeoutMs: number; requestSent: boolean }) { - super(`gateway request timeout for ${params.method}`); + super(params, `gateway request timeout for ${params.method}`); this.name = "GatewayClientRequestTimeoutError"; - this.method = params.method; - this.timeoutMs = params.timeoutMs; - this.requestSent = params.requestSent; } } @@ -404,7 +400,7 @@ export class GatewayClient { }; this.requestTimeoutMs = typeof opts.requestTimeoutMs === "number" && Number.isFinite(opts.requestTimeoutMs) - ? resolveSafeTimeoutDelayMs(opts.requestTimeoutMs, { minMs: 0 }) + ? opts.requestTimeoutMs : DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS; const connectChallengeTimeoutMs = resolveConnectChallengeTimeoutMs( this.opts.connectChallengeTimeoutMs, @@ -1359,7 +1355,7 @@ export class GatewayClient { opts?.timeoutMs === null ? null : typeof opts?.timeoutMs === "number" && Number.isFinite(opts.timeoutMs) - ? resolveSafeTimeoutDelayMs(opts.timeoutMs, { minMs: 0 }) + ? opts.timeoutMs : expectFinal ? null : this.requestTimeoutMs; diff --git a/packages/gateway-client/src/client.watchdog.test.ts b/packages/gateway-client/src/client.watchdog.test.ts index 5a47c0fbf434..1390ca521ced 100644 --- a/packages/gateway-client/src/client.watchdog.test.ts +++ b/packages/gateway-client/src/client.watchdog.test.ts @@ -35,8 +35,8 @@ function createOpenGatewayClient(requestTimeoutMs: number): { return { client, send }; } -function getPendingCount(client: GatewayClient): number { - return protocolHarness(client).pending.size; +function hasPendingRequests(client: GatewayClient): boolean { + return protocolHarness(client).hasPendingRequests; } test("decodes every ws raw-data shape", () => { @@ -50,8 +50,8 @@ type ProtocolHarness = { socket: GatewayProtocolSocket | null; stopped: boolean; generation: number; + hasPendingRequests: boolean; reconnectSupervisor: { reset(initialMs?: number): void }; - pending: Map; handleMessage: (socket: GatewayProtocolSocket, generation: number, raw: string) => void; }; @@ -915,7 +915,7 @@ describe("GatewayClient", () => { "synthetic send failure", ); expect(onSent).not.toHaveBeenCalled(); - expect(getPendingCount(client)).toBe(0); + expect(hasPendingRequests(client)).toBe(false); }); test("notifies accepted expectFinal requests while continuing to wait for final", async () => { @@ -939,7 +939,7 @@ describe("GatewayClient", () => { expect(onSent).toHaveBeenCalledOnce(); expect(onAccepted).toHaveBeenCalledWith({ status: "accepted", runId: "run-1" }); - expect(getPendingCount(client)).toBe(1); + expect(hasPendingRequests(client)).toBe(true); handleGatewayMessage(client, { type: "res", @@ -949,7 +949,7 @@ describe("GatewayClient", () => { }); await expect(requestPromise).resolves.toEqual({ status: "ok" }); - expect(getPendingCount(client)).toBe(0); + expect(hasPendingRequests(client)).toBe(false); }); test("aborts in-flight requests from caller AbortSignal", async () => { @@ -961,12 +961,12 @@ describe("GatewayClient", () => { timeoutMs: null, }); expect(send).toHaveBeenCalledTimes(1); - expect(getPendingCount(client)).toBe(1); + expect(hasPendingRequests(client)).toBe(true); controller.abort(); await expect(requestPromise).rejects.toThrow("gateway request aborted for status"); - expect(getPendingCount(client)).toBe(0); + expect(hasPendingRequests(client)).toBe(false); }); test.each([ @@ -984,7 +984,7 @@ describe("GatewayClient", () => { await vi.advanceTimersByTimeAsync(1); expect(isSettled()).toBe(false); - expect(getPendingCount(client)).toBe(1); + expect(hasPendingRequests(client)).toBe(true); client.stop(); await expect(requestPromise).rejects.toThrow("gateway client stopped"); diff --git a/packages/gateway-client/src/pending-request.ts b/packages/gateway-client/src/pending-request.ts index 5b5f11ed86b0..83558ff79a6e 100644 --- a/packages/gateway-client/src/pending-request.ts +++ b/packages/gateway-client/src/pending-request.ts @@ -1,5 +1,26 @@ -/** Owned settlement, cleanup, and timing state for one Gateway wire request. */ -export type GatewayPendingRequest = { +import type { ErrorShape, ResponseFrame } from "@openclaw/gateway-protocol"; +import { + GatewayProtocolRequestError, + GatewayProtocolRequestTimeoutError, + type GatewayProtocolRequestOptions, +} from "./protocol-request.js"; +import { resolveSafeTimeoutDelayMs } from "./timeouts.js"; + +export type GatewayProtocolRequestTiming = { + id: string; + method: string; + ok: boolean; + durationMs: number; + startedAtMs: number; + endedAtMs: number; + errorCode?: string; +}; + +type GatewayRequestSender = { + send: (data: string) => void; +}; + +type GatewayPendingRequest = { resolve: (value: unknown) => void; reject: (error: Error) => void; expectFinal: boolean; @@ -10,3 +31,205 @@ export type GatewayPendingRequest = { method: string; startedAtMs: number; }; + +type GatewayPendingRequestsOptions = { + createRequestId: () => string; + createRequestError?: (error: Partial) => GatewayProtocolRequestError; + createRequestTimeoutError?: (method: string, timeoutMs: number, requestSent: boolean) => Error; + createRequestAbortError?: (method: string) => Error; + requestTimeoutMs?: number; + nowMs: () => number; + onTiming?: (timing: GatewayProtocolRequestTiming) => void; + onCallbackError?: (label: string, error: unknown) => void; +}; + +/** Owns request deadlines, correlation, settlement, and generation-scoped IDs. */ +export class GatewayPendingRequests { + private readonly pending = new Map(); + private readonly retiredIds = new Set(); + private collisionSuffix = 0; + + constructor(private readonly opts: GatewayPendingRequestsOptions) {} + + get hasPending(): boolean { + return this.pending.size > 0; + } + + get hasUnboundedPending(): boolean { + return [...this.pending.values()].some((pending) => pending.unbounded); + } + + request( + sender: GatewayRequestSender, + method: string, + params?: unknown, + options?: GatewayProtocolRequestOptions, + ): Promise { + let id: string; + try { + id = this.allocateRequestId(); + } catch (error) { + return Promise.reject(error instanceof Error ? error : new Error(String(error))); + } + const requestedTimeoutMs = + options?.timeoutMs === null ? undefined : (options?.timeoutMs ?? this.opts.requestTimeoutMs); + const timeoutMs = + typeof requestedTimeoutMs === "number" && Number.isFinite(requestedTimeoutMs) + ? resolveSafeTimeoutDelayMs(requestedTimeoutMs, { minMs: 0 }) + : undefined; + return new Promise((resolve, reject) => { + let timeout: ReturnType | undefined; + let requestSent = false; + const pending: GatewayPendingRequest = { + resolve: (value) => resolve(value as T), + reject, + expectFinal: options?.expectFinal === true, + acceptedNotified: false, + onAccepted: options?.onAccepted, + unbounded: timeoutMs === undefined, + method, + startedAtMs: this.opts.nowMs(), + }; + const cleanup = () => { + if (timeout !== undefined) { + clearTimeout(timeout); + } + options?.signal?.removeEventListener("abort", onAbort); + }; + const retire = (errorCode: string): boolean => { + if (this.pending.get(id) !== pending) { + return false; + } + this.pending.delete(id); + this.retiredIds.add(id); + cleanup(); + this.finishTiming(id, pending, false, errorCode); + return true; + }; + const onAbort = () => { + if (!retire("CLIENT_ABORTED")) { + return; + } + reject( + this.opts.createRequestAbortError?.(method) ?? + new Error(`gateway request aborted for ${method}`), + ); + }; + if (options?.signal?.aborted) { + reject( + this.opts.createRequestAbortError?.(method) ?? + new Error(`gateway request aborted for ${method}`), + ); + return; + } + pending.cleanup = cleanup; + if (timeoutMs !== undefined) { + timeout = setTimeout(() => { + if (!retire("CLIENT_TIMEOUT")) { + return; + } + reject( + this.opts.createRequestTimeoutError?.(method, timeoutMs, requestSent) ?? + new GatewayProtocolRequestTimeoutError({ method, timeoutMs, requestSent }), + ); + }, timeoutMs); + timeout.unref?.(); + } + options?.signal?.addEventListener("abort", onAbort, { once: true }); + this.pending.set(id, pending); + try { + sender.send(JSON.stringify({ type: "req", id, method, params })); + if (this.pending.get(id) !== pending) { + return; + } + requestSent = true; + this.invoke("sent", () => options?.onSent?.()); + } catch (error) { + if (retire("CLIENT_SEND_ERROR")) { + reject(error instanceof Error ? error : new Error(String(error))); + } + } + }); + } + + handleResponse(frame: ResponseFrame): void { + const pending = this.pending.get(frame.id); + if (!pending) { + return; + } + const status = (frame.payload as { status?: unknown } | undefined)?.status; + if (pending.expectFinal && status === "accepted") { + if (!pending.acceptedNotified) { + pending.acceptedNotified = true; + this.invoke("accepted", () => pending.onAccepted?.(frame.payload)); + } + return; + } + this.pending.delete(frame.id); + pending.cleanup?.(); + if (frame.ok) { + this.finishTiming(frame.id, pending, true); + pending.resolve(frame.payload); + return; + } + this.finishTiming(frame.id, pending, false, frame.error?.code); + pending.reject( + this.opts.createRequestError?.(frame.error ?? {}) ?? + new GatewayProtocolRequestError(frame.error ?? {}), + ); + } + + flush(error: Error): void { + for (const [id, pending] of this.pending) { + this.finishTiming(id, pending, false, "CLIENT_CLOSED"); + pending.cleanup?.(); + pending.reject(error); + } + this.pending.clear(); + // IDs are tombstoned only for one socket generation. Retired socket frames + // are fenced by GatewayProtocolClient before a replacement generation runs. + this.retiredIds.clear(); + this.collisionSuffix = 0; + } + + private allocateRequestId(): string { + const id = this.opts.createRequestId(); + if (!this.pending.has(id) && !this.retiredIds.has(id)) { + return id; + } + let uniqueId: string; + do { + this.collisionSuffix += 1; + uniqueId = `${id}:${this.collisionSuffix}`; + } while (this.pending.has(uniqueId) || this.retiredIds.has(uniqueId)); + return uniqueId; + } + + private finishTiming( + id: string, + pending: GatewayPendingRequest, + ok: boolean, + errorCode?: string, + ): void { + const endedAtMs = this.opts.nowMs(); + this.invoke("request timing", () => + this.opts.onTiming?.({ + id, + method: pending.method, + ok, + durationMs: Math.max(0, endedAtMs - pending.startedAtMs), + startedAtMs: pending.startedAtMs, + endedAtMs, + errorCode, + }), + ); + } + + private invoke(label: string, callback: () => void): void { + try { + callback(); + } catch (error) { + this.opts.onCallbackError?.(label, error); + } + } +} diff --git a/packages/gateway-client/src/protocol-client.request.test.ts b/packages/gateway-client/src/protocol-client.request.test.ts new file mode 100644 index 000000000000..60b101bc8612 --- /dev/null +++ b/packages/gateway-client/src/protocol-client.request.test.ts @@ -0,0 +1,350 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + GatewayProtocolClient, + GatewayProtocolRequestError, + GatewayProtocolRequestTimeoutError, + type GatewayProtocolRequestOptions, + type GatewayProtocolRequestTiming, + type GatewayProtocolSocketHandlers, +} from "./protocol-client.js"; +import { MAX_SAFE_TIMEOUT_DELAY_MS } from "./timeouts.js"; + +type RequestFrame = { + id: string; + method: string; +}; + +type RequestConnection = { + handlers: GatewayProtocolSocketHandlers; + frames: RequestFrame[]; + close: (code?: number, reason?: string) => void; +}; + +function createRequestHarness(options?: { + createRequestId?: () => string; + requestTimeoutMs?: number; + onRequestTiming?: (timing: GatewayProtocolRequestTiming) => void; + onCallbackError?: (label: string, error: unknown) => void; + send?: (frame: RequestFrame) => void; + nowMs?: () => number; +}) { + const connections: RequestConnection[] = []; + let nextRequestId = 0; + const client = new GatewayProtocolClient>({ + createSocket: (handlers) => { + let open = true; + const frames: RequestFrame[] = []; + const close = (code = 1000, reason = "") => { + open = false; + handlers.close(code, reason); + }; + connections.push({ handlers, frames, close }); + return { + isOpen: () => open, + send: (data) => { + const frame = JSON.parse(data) as RequestFrame; + frames.push(frame); + options?.send?.(frame); + }, + close, + }; + }, + createRequestId: options?.createRequestId ?? (() => `request-${++nextRequestId}`), + buildConnectPlan: () => ({}), + buildConnectParams: (plan) => plan, + resolveClose: () => ({ retry: false, notify: false }), + handshake: { mode: "require-challenge", timeoutMs: 100 }, + reconnect: { initialMs: 10, multiplier: 2, maxMs: 100 }, + requestTimeoutMs: options?.requestTimeoutMs, + onRequestTiming: options?.onRequestTiming, + onCallbackError: options?.onCallbackError, + nowMs: options?.nowMs, + }); + client.start(); + return { client, connections }; +} + +function latestFrame(connection: RequestConnection): RequestFrame { + const frame = connection.frames.at(-1); + if (!frame) { + throw new Error("expected request frame"); + } + return frame; +} + +function respond(connection: RequestConnection, id: string, payload: unknown, ok = true): void { + connection.handlers.message( + JSON.stringify({ + type: "res", + id, + ok, + ...(ok ? { payload } : { error: payload }), + }), + ); +} + +afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +describe("GatewayProtocolClient requests", () => { + it.each([ + { + label: "an explicit finite deadline", + requestTimeoutMs: undefined, + requestOptions: { timeoutMs: 25 } satisfies GatewayProtocolRequestOptions, + expectedTimerMs: 25, + unbounded: false, + }, + { + label: "the client default deadline", + requestTimeoutMs: 30, + requestOptions: undefined, + expectedTimerMs: 30, + unbounded: false, + }, + { + label: "an oversized finite deadline", + requestTimeoutMs: undefined, + requestOptions: { + timeoutMs: Number.MAX_SAFE_INTEGER, + } satisfies GatewayProtocolRequestOptions, + expectedTimerMs: MAX_SAFE_TIMEOUT_DELAY_MS, + unbounded: false, + }, + { + label: "an explicit null deadline", + requestTimeoutMs: 30, + requestOptions: { timeoutMs: null } satisfies GatewayProtocolRequestOptions, + expectedTimerMs: null, + unbounded: true, + }, + { + label: "the browser default", + requestTimeoutMs: undefined, + requestOptions: undefined, + expectedTimerMs: null, + unbounded: true, + }, + ])( + "normalizes $label only in the scheduling owner", + ({ requestTimeoutMs, requestOptions, expectedTimerMs, unbounded }) => { + vi.useFakeTimers(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const { client } = createRequestHarness({ requestTimeoutMs }); + const request = client.request("status", {}, requestOptions); + void request.catch(() => {}); + + expect(client.hasUnboundedPendingRequests).toBe(unbounded); + if (expectedTimerMs === null) { + expect(setTimeoutSpy).not.toHaveBeenCalled(); + } else { + expect(setTimeoutSpy).toHaveBeenLastCalledWith(expect.any(Function), expectedTimerMs); + } + client.stop(); + }, + ); + + it("reports typed deadlines before and after the send boundary", async () => { + vi.useFakeTimers(); + const sentHarness = createRequestHarness(); + const sentRequest = sentHarness.client.request("sent.request", {}, { timeoutMs: 5 }); + const sentOutcome = sentRequest.catch((error: unknown) => error); + + await vi.advanceTimersByTimeAsync(5); + + await expect(sentOutcome).resolves.toMatchObject({ + code: "CLIENT_TIMEOUT", + method: "sent.request", + timeoutMs: 5, + requestSent: true, + }); + + let deadline: (() => void) | undefined; + vi.spyOn(globalThis, "setTimeout").mockImplementation((( + callback: Parameters[0], + ) => { + deadline = callback as () => void; + return { unref: () => undefined } as unknown as ReturnType; + }) as unknown as typeof setTimeout); + const onSent = vi.fn(); + const unsentHarness = createRequestHarness({ send: () => deadline?.() }); + const unsentRequest = unsentHarness.client.request( + "unsent.request", + {}, + { timeoutMs: 5, onSent }, + ); + + await expect(unsentRequest).rejects.toMatchObject({ + code: "CLIENT_TIMEOUT", + method: "unsent.request", + timeoutMs: 5, + requestSent: false, + }); + expect(onSent).not.toHaveBeenCalled(); + expect(unsentHarness.client.hasPendingRequests).toBe(false); + expect(sentHarness.client.hasPendingRequests).toBe(false); + sentHarness.client.stop(); + unsentHarness.client.stop(); + }); + + it("retires aborted and send-failed IDs before a replacement request", async () => { + const controller = new AbortController(); + let sendCalls = 0; + const { client, connections } = createRequestHarness({ + createRequestId: () => "same-id", + send: () => { + sendCalls += 1; + if (sendCalls === 3) { + throw new Error("synthetic send failure"); + } + }, + }); + const connection = connections[0]; + if (!connection) { + throw new Error("expected request connection"); + } + const aborted = client.request("aborted", {}, { timeoutMs: null, signal: controller.signal }); + controller.abort(); + await expect(aborted).rejects.toThrow("gateway request aborted for aborted"); + + const replacement = client.request("replacement", {}, { timeoutMs: null }); + expect(latestFrame(connection)).toMatchObject({ id: "same-id:1", method: "replacement" }); + respond(connection, "same-id", { stale: true }); + expect(client.hasPendingRequests).toBe(true); + respond(connection, "same-id:1", { current: true }); + await expect(replacement).resolves.toEqual({ current: true }); + + await expect(client.request("send.failure", {}, { timeoutMs: null })).rejects.toThrow( + "synthetic send failure", + ); + expect(latestFrame(connection)).toMatchObject({ id: "same-id:2", method: "send.failure" }); + expect(client.hasPendingRequests).toBe(false); + client.stop(); + }); + + it("ignores late accepted and final replies after a timeout collision", async () => { + vi.useFakeTimers(); + const onAccepted = vi.fn(); + const { client, connections } = createRequestHarness({ createRequestId: () => "same-id" }); + const connection = connections[0]; + if (!connection) { + throw new Error("expected request connection"); + } + const retired = client.request("agent", {}, { timeoutMs: 5, expectFinal: true, onAccepted }); + const retiredOutcome = retired.catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(5); + await expect(retiredOutcome).resolves.toBeInstanceOf(GatewayProtocolRequestTimeoutError); + + const replacement = client.request( + "agent", + {}, + { timeoutMs: null, expectFinal: true, onAccepted }, + ); + expect(latestFrame(connection)).toMatchObject({ id: "same-id:1", method: "agent" }); + respond(connection, "same-id", { status: "accepted", runId: "old" }); + respond(connection, "same-id", { status: "ok", runId: "old" }); + expect(onAccepted).not.toHaveBeenCalled(); + expect(client.hasPendingRequests).toBe(true); + + respond(connection, "same-id:1", { status: "accepted", runId: "new" }); + respond(connection, "same-id:1", { status: "ok", runId: "new" }); + await expect(replacement).resolves.toEqual({ status: "ok", runId: "new" }); + expect(onAccepted).toHaveBeenCalledExactlyOnceWith({ status: "accepted", runId: "new" }); + client.stop(); + }); + + it("keeps authoritative Gateway errors distinct from local deadlines", async () => { + const { client, connections } = createRequestHarness(); + const connection = connections[0]; + if (!connection) { + throw new Error("expected request connection"); + } + const request = client.request("sessions.subscribe", {}, { timeoutMs: 25 }); + const frame = latestFrame(connection); + respond( + connection, + frame.id, + { code: "FORBIDDEN", message: "subscription rejected", retryable: false }, + false, + ); + + const error = await request.catch((value: unknown) => value); + expect(error).toBeInstanceOf(GatewayProtocolRequestError); + expect(error).not.toBeInstanceOf(GatewayProtocolRequestTimeoutError); + expect(error).toMatchObject({ code: "FORBIDDEN", retryable: false }); + client.stop(); + }); + + it("isolates callbacks while preserving accepted/final settlement and timing", async () => { + let nowMs = 10; + const onRequestTiming = vi.fn<(timing: GatewayProtocolRequestTiming) => void>(); + const onCallbackError = vi.fn<(label: string, error: unknown) => void>(); + const { client, connections } = createRequestHarness({ + nowMs: () => nowMs, + onRequestTiming, + onCallbackError, + }); + const connection = connections[0]; + if (!connection) { + throw new Error("expected request connection"); + } + const request = client.request( + "agent", + {}, + { + timeoutMs: null, + expectFinal: true, + onSent: () => { + throw new Error("sent callback failed"); + }, + onAccepted: () => { + throw new Error("accepted callback failed"); + }, + }, + ); + const frame = latestFrame(connection); + respond(connection, frame.id, { status: "accepted" }); + expect(client.hasPendingRequests).toBe(true); + nowMs = 25; + respond(connection, frame.id, { status: "ok" }); + + await expect(request).resolves.toEqual({ status: "ok" }); + expect(onCallbackError.mock.calls.map(([label]) => label)).toEqual(["sent", "accepted"]); + expect(onRequestTiming).toHaveBeenCalledExactlyOnceWith({ + id: frame.id, + method: "agent", + ok: true, + durationMs: 15, + startedAtMs: 10, + endedAtMs: 25, + }); + client.stop(); + }); + + it("clears generation tombstones when the socket flushes", async () => { + vi.useFakeTimers(); + const { client, connections } = createRequestHarness({ createRequestId: () => "same-id" }); + const firstConnection = connections[0]; + if (!firstConnection) { + throw new Error("expected first request connection"); + } + const retired = client.request("first", {}, { timeoutMs: 5 }); + const retiredOutcome = retired.catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(5); + await expect(retiredOutcome).resolves.toBeInstanceOf(GatewayProtocolRequestTimeoutError); + + firstConnection.close(1000, "socket generation complete"); + client.start(); + const secondConnection = connections[1]; + if (!secondConnection) { + throw new Error("expected replacement request connection"); + } + const replacement = client.request("second", {}, { timeoutMs: null }); + expect(latestFrame(secondConnection)).toMatchObject({ id: "same-id", method: "second" }); + respond(secondConnection, "same-id", { ok: true }); + await expect(replacement).resolves.toEqual({ ok: true }); + client.stop(); + }); +}); diff --git a/packages/gateway-client/src/protocol-client.ts b/packages/gateway-client/src/protocol-client.ts index 6cdc9e6ee88b..8ce4ca530fe8 100644 --- a/packages/gateway-client/src/protocol-client.ts +++ b/packages/gateway-client/src/protocol-client.ts @@ -1,18 +1,24 @@ -import type { ErrorShape, EventFrame, HelloOk, ResponseFrame } from "@openclaw/gateway-protocol"; +import type { ErrorShape, EventFrame, HelloOk } from "@openclaw/gateway-protocol"; import { isGatewayEventFrame, isGatewayResponseFrame, } from "@openclaw/gateway-protocol/frame-guards"; import { RetrySupervisor, sleepWithAbort } from "@openclaw/retry"; import { GatewayEventListeners } from "./event-listeners.js"; -import type { GatewayPendingRequest } from "./pending-request.js"; +import { GatewayPendingRequests, type GatewayProtocolRequestTiming } from "./pending-request.js"; import { GatewayProtocolRequestError, + GatewayProtocolRequestTimeoutError, type GatewayProtocolRequestOptions, } from "./protocol-request.js"; import { clearGatewayConnectTimeout, startGatewayConnectTimeout } from "./timeouts.js"; -export { GatewayProtocolRequestError, type GatewayProtocolRequestOptions }; +export { + GatewayProtocolRequestError, + GatewayProtocolRequestTimeoutError, + type GatewayProtocolRequestOptions, + type GatewayProtocolRequestTiming, +}; export type GatewayProtocolSocket = { isOpen: () => boolean; @@ -71,15 +77,6 @@ export type GatewayProtocolTiming = { plan?: TPlan; detail?: unknown; }; -export type GatewayProtocolRequestTiming = { - id: string; - method: string; - ok: boolean; - durationMs: number; - startedAtMs: number; - endedAtMs: number; - errorCode?: string; -}; type GatewayProtocolClientOptions = { createSocket: (handlers: GatewayProtocolSocketHandlers) => GatewayProtocolSocket; createRequestId: () => string; @@ -139,7 +136,7 @@ type CloseSnapshot = Omit; */ export class GatewayProtocolClient { private socket: GatewayProtocolSocket | null = null; - private readonly pending = new Map(); + private readonly requests: GatewayPendingRequests; private readonly listeners = new GatewayEventListeners(); private stopped = true; private generation = 0; @@ -164,6 +161,16 @@ export class GatewayProtocolClient { factor: opts.reconnect.multiplier, jitter: 0, }); + this.requests = new GatewayPendingRequests({ + createRequestId: opts.createRequestId, + createRequestError: opts.createRequestError, + createRequestTimeoutError: opts.createRequestTimeoutError, + createRequestAbortError: opts.createRequestAbortError, + requestTimeoutMs: opts.requestTimeoutMs, + nowMs: () => this.nowMs(), + onTiming: opts.onRequestTiming, + onCallbackError: opts.onCallbackError, + }); } get connected(): boolean { @@ -171,7 +178,7 @@ export class GatewayProtocolClient { } get hasPendingRequests(): boolean { - return this.pending.size > 0; + return this.requests.hasPending; } get connecting(): boolean { @@ -179,7 +186,7 @@ export class GatewayProtocolClient { } get hasUnboundedPendingRequests(): boolean { - return [...this.pending.values()].some((pending) => pending.unbounded); + return this.requests.hasUnboundedPending; } start(): void { @@ -205,7 +212,7 @@ export class GatewayProtocolClient { this.socket = null; this.connectFailure = undefined; this.connectTiming = null; - this.flushRequests(new Error("gateway client stopped")); + this.requests.flush(new Error("gateway client stopped")); socket?.close(); } @@ -221,73 +228,7 @@ export class GatewayProtocolClient { if (typeof method !== "string" || method.length === 0) { return Promise.reject(new Error("invalid request frame: method must be a non-empty string")); } - const id = this.opts.createRequestId(); - const timeoutMs = - options?.timeoutMs === null ? undefined : (options?.timeoutMs ?? this.opts.requestTimeoutMs); - return new Promise((resolve, reject) => { - let timeout: ReturnType | undefined; - let requestSent = false; - const pending: GatewayPendingRequest = { - resolve: (value) => resolve(value as T), - reject, - expectFinal: options?.expectFinal === true, - acceptedNotified: false, - onAccepted: options?.onAccepted, - unbounded: timeoutMs === undefined, - method, - startedAtMs: this.nowMs(), - }; - const onAbort = () => { - this.pending.delete(id); - pending.cleanup?.(); - this.finishRequestTiming(id, pending, false, "CLIENT_ABORTED"); - reject( - this.opts.createRequestAbortError?.(method) ?? - new Error(`gateway request aborted for ${method}`), - ); - }; - const cleanup = () => { - if (timeout) { - clearTimeout(timeout); - } - options?.signal?.removeEventListener("abort", onAbort); - }; - if (options?.signal?.aborted) { - reject( - this.opts.createRequestAbortError?.(method) ?? - new Error(`gateway request aborted for ${method}`), - ); - return; - } - pending.cleanup = cleanup; - if (timeoutMs !== undefined && timeoutMs >= 0) { - timeout = setTimeout(() => { - if (this.pending.get(id) !== pending) { - return; - } - this.pending.delete(id); - options?.signal?.removeEventListener("abort", onAbort); - this.finishRequestTiming(id, pending, false, "CLIENT_TIMEOUT"); - reject( - this.opts.createRequestTimeoutError?.(method, timeoutMs, requestSent) ?? - new Error(`gateway request timed out after ${timeoutMs}ms: ${method}`), - ); - }, timeoutMs); - timeout.unref?.(); - } - options?.signal?.addEventListener("abort", onAbort, { once: true }); - this.pending.set(id, pending); - try { - socket.send(JSON.stringify({ type: "req", id, method, params })); - requestSent = true; - this.invoke("sent", () => options?.onSent?.()); - } catch (error) { - this.pending.delete(id); - cleanup(); - this.finishRequestTiming(id, pending, false, "CLIENT_SEND_ERROR"); - reject(error instanceof Error ? error : new Error(String(error))); - } - }); + return this.requests.request(socket, method, params, options); } addEventListener(listener: (event: EventFrame) => void): () => void { @@ -588,34 +529,7 @@ export class GatewayProtocolClient { return; } this.opts.onActivity?.(); - this.handleResponse(parsed); - } - - private handleResponse(frame: ResponseFrame): void { - const pending = this.pending.get(frame.id); - if (!pending) { - return; - } - const status = (frame.payload as { status?: unknown } | undefined)?.status; - if (pending.expectFinal && status === "accepted") { - if (!pending.acceptedNotified) { - pending.acceptedNotified = true; - this.invoke("accepted", () => pending.onAccepted?.(frame.payload)); - } - return; - } - this.pending.delete(frame.id); - pending.cleanup?.(); - if (frame.ok) { - this.finishRequestTiming(frame.id, pending, true); - pending.resolve(frame.payload); - return; - } - this.finishRequestTiming(frame.id, pending, false, frame.error?.code); - pending.reject( - this.opts.createRequestError?.(frame.error ?? {}) ?? - new GatewayProtocolRequestError(frame.error ?? {}), - ); + this.requests.handleResponse(parsed); } private handleClose( @@ -642,7 +556,7 @@ export class GatewayProtocolClient { }; this.connectFailure = undefined; const decision = this.opts.resolveClose(context); - this.flushRequests( + this.requests.flush( decision.pendingError ?? context.connectFailure?.error ?? new Error(`gateway closed (${code}): ${reason}`), @@ -660,35 +574,6 @@ export class GatewayProtocolClient { this.opts.onConnectError?.(error); } - private flushRequests(error: Error): void { - for (const [id, pending] of this.pending) { - this.finishRequestTiming(id, pending, false, "CLIENT_CLOSED"); - pending.cleanup?.(); - pending.reject(error); - } - this.pending.clear(); - } - - private finishRequestTiming( - id: string, - pending: GatewayPendingRequest, - ok: boolean, - errorCode?: string, - ): void { - const endedAtMs = this.nowMs(); - this.invoke("request timing", () => - this.opts.onRequestTiming?.({ - id, - method: pending.method, - ok, - durationMs: Math.max(0, endedAtMs - pending.startedAtMs), - startedAtMs: pending.startedAtMs, - endedAtMs, - errorCode, - }), - ); - } - private scheduleReconnect(overrideMs?: number): void { if (overrideMs !== undefined) { // Retry-After is a floor for this wait, not a failed attempt. Preserve diff --git a/packages/gateway-client/src/protocol-request.ts b/packages/gateway-client/src/protocol-request.ts index b11a7391600c..2b1038b45b0e 100644 --- a/packages/gateway-client/src/protocol-request.ts +++ b/packages/gateway-client/src/protocol-request.ts @@ -25,3 +25,22 @@ export class GatewayProtocolRequestError extends Error { this.retryAfterMs = error.retryAfterMs; } } + +/** A local transport deadline, distinct from a Gateway's authoritative rejection. */ +export class GatewayProtocolRequestTimeoutError extends Error { + readonly code = "CLIENT_TIMEOUT"; + readonly method: string; + readonly timeoutMs: number; + readonly requestSent: boolean; + + constructor( + params: { method: string; timeoutMs: number; requestSent: boolean }, + message = `gateway request timed out after ${params.timeoutMs}ms: ${params.method}`, + ) { + super(message); + this.name = "GatewayProtocolRequestTimeoutError"; + this.method = params.method; + this.timeoutMs = params.timeoutMs; + this.requestSent = params.requestSent; + } +} diff --git a/ui/src/lib/sessions/connection-hydration.test.ts b/ui/src/lib/sessions/connection-hydration.test.ts index fe5aee332a55..c2eba83672f1 100644 --- a/ui/src/lib/sessions/connection-hydration.test.ts +++ b/ui/src/lib/sessions/connection-hydration.test.ts @@ -1,4 +1,9 @@ // @vitest-environment node +import { + DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS, + GatewayProtocolClient, + type GatewayProtocolSocketHandlers, +} from "@openclaw/gateway-client/browser"; import { describe, expect, it, vi } from "vitest"; import { GatewayRequestError, @@ -336,6 +341,91 @@ describe("session connection hydration", () => { } }); + it("recovers the roster after a subscription deadline without admitting late replies", async () => { + vi.useFakeTimers(); + const initialResult = emptySessionsResult(); + const recoveredResult: SessionsListResult = { + ...initialResult, + count: 1, + sessions: [{ key: "agent:main:recovered", kind: "direct", updatedAt: 2 }], + }; + const sent: Array<{ id: string; method: string }> = []; + let handlers: GatewayProtocolSocketHandlers | undefined; + const protocol = new GatewayProtocolClient>({ + createSocket: (nextHandlers) => { + handlers = nextHandlers; + return { + isOpen: () => true, + send: (data) => { + const frame = JSON.parse(data) as { id: string; method: string }; + sent.push({ id: frame.id, method: frame.method }); + }, + close: () => undefined, + }; + }, + createRequestId: () => "request", + buildConnectPlan: () => ({}), + buildConnectParams: (plan) => plan, + resolveClose: () => ({ retry: false, notify: false }), + handshake: { mode: "require-challenge", timeoutMs: 100 }, + reconnect: { initialMs: 10, multiplier: 2, maxMs: 100 }, + }); + protocol.start(); + const request = protocol.request.bind(protocol) as GatewayBrowserClient["request"]; + const { sessions, connect } = createSubscriptionHydrationHarness(request); + const recoveredStates: SessionsListResult[] = []; + const unsubscribe = sessions.subscribe((next) => { + if (next.result?.sessions.some((row) => row.key === "agent:main:recovered")) { + recoveredStates.push(next.result); + } + }); + const respond = (id: string, payload: unknown) => { + handlers?.message(JSON.stringify({ type: "res", id, ok: true, payload })); + }; + + try { + connect(); + expect(sent).toEqual([{ id: "request", method: "sessions.subscribe" }]); + + await vi.advanceTimersByTimeAsync(DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS); + expect(sent).toContainEqual({ id: "request:1", method: "sessions.list" }); + respond("request:1", initialResult); + await vi.advanceTimersByTimeAsync(0); + expect(sessions.state.result).toEqual(initialResult); + + await vi.advanceTimersByTimeAsync(500); + expect(sent).toContainEqual({ id: "request:2", method: "sessions.subscribe" }); + + respond("request", { status: "accepted" }); + respond("request", { subscribed: true }); + await vi.advanceTimersByTimeAsync(0); + expect(sessions.state.error).not.toBeNull(); + expect(sent.filter(({ method }) => method === "sessions.list")).toHaveLength(1); + + respond("request:2", { subscribed: true }); + await vi.advanceTimersByTimeAsync(0); + expect(sent).toContainEqual({ id: "request:3", method: "sessions.list" }); + respond("request:3", recoveredResult); + await vi.advanceTimersByTimeAsync(0); + + respond("request", { status: "accepted" }); + respond("request", { subscribed: true }); + await vi.advanceTimersByTimeAsync(0); + + expect(sessions.state.error).toBeNull(); + expect(sessions.state.result).toEqual(recoveredResult); + expect(recoveredStates).toEqual([recoveredResult]); + expect(sent.filter(({ method }) => method === "sessions.subscribe")).toHaveLength(2); + expect(sent.filter(({ method }) => method === "sessions.list")).toHaveLength(2); + expect(vi.getTimerCount()).toBe(0); + } finally { + unsubscribe(); + sessions.dispose(); + protocol.stop(); + vi.useRealTimers(); + } + }); + it.each([ { description: "an explicitly declined", response: { subscribed: false } }, { description: "an unacknowledged", response: {} }, diff --git a/ui/src/lib/sessions/session-event-subscription.ts b/ui/src/lib/sessions/session-event-subscription.ts index b4caecd1b30c..7982975d1e00 100644 --- a/ui/src/lib/sessions/session-event-subscription.ts +++ b/ui/src/lib/sessions/session-event-subscription.ts @@ -1,3 +1,7 @@ +import { + DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS, + GatewayProtocolRequestTimeoutError, +} from "@openclaw/gateway-client/browser"; import { GatewayRequestError, type GatewayBrowserClient } from "../../api/gateway.ts"; type SessionEventSubscriptionScope = { @@ -46,6 +50,7 @@ export function createSessionEventSubscriptionOwner(params: { const response = await scope.client.request<{ subscribed?: boolean }>( "sessions.subscribe", {}, + { timeoutMs: DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS }, ); if (!isCurrent(scope, expectedGeneration)) { return; @@ -63,8 +68,18 @@ export function createSessionEventSubscriptionOwner(params: { if (!isCurrent(scope, expectedGeneration)) { return; } - params.onError(scope, String(error)); - const delayMs = params.retryDelayMs(error); + // A connected transport can outlive an application acknowledgement. + // Only this idempotent observer turns its typed deadline into a retry. + const failure = + error instanceof GatewayProtocolRequestTimeoutError + ? new GatewayRequestError({ + code: error.code, + message: error.message, + retryable: true, + }) + : error; + params.onError(scope, String(failure)); + const delayMs = params.retryDelayMs(failure); if (delayMs === null || !isCurrent(scope, expectedGeneration)) { return; }