mirror of
https://github.com/openclaw/openclaw.git
synced 2026-04-18 10:37:27 +00:00
perf(test): avoid gateway boot for late invoke results
This commit is contained in:
Peter Steinberger
71
src/gateway/server-methods/nodes.handlers.invoke-result.ts
Normal file
71
src/gateway/server-methods/nodes.handlers.invoke-result.ts
Normal file
@@ -0,0 +1,71 @@
|
||||
import type { GatewayRequestHandler } from "./types.js";
|
||||
import { ErrorCodes, errorShape, validateNodeInvokeResultParams } from "../protocol/index.js";
|
||||
import { respondInvalidParams } from "./nodes.helpers.js";
|
||||
|
||||
function normalizeNodeInvokeResultParams(params: unknown): unknown {
|
||||
if (!params || typeof params !== "object") {
|
||||
return params;
|
||||
}
|
||||
const raw = params as Record<string, unknown>;
|
||||
const normalized: Record<string, unknown> = { ...raw };
|
||||
if (normalized.payloadJSON === null) {
|
||||
delete normalized.payloadJSON;
|
||||
} else if (normalized.payloadJSON !== undefined && typeof normalized.payloadJSON !== "string") {
|
||||
if (normalized.payload === undefined) {
|
||||
normalized.payload = normalized.payloadJSON;
|
||||
}
|
||||
delete normalized.payloadJSON;
|
||||
}
|
||||
if (normalized.error === null) {
|
||||
delete normalized.error;
|
||||
}
|
||||
return normalized;
|
||||
}
|
||||
|
||||
export const handleNodeInvokeResult: GatewayRequestHandler = async ({
|
||||
params,
|
||||
respond,
|
||||
context,
|
||||
client,
|
||||
}) => {
|
||||
const normalizedParams = normalizeNodeInvokeResultParams(params);
|
||||
if (!validateNodeInvokeResultParams(normalizedParams)) {
|
||||
respondInvalidParams({
|
||||
respond,
|
||||
method: "node.invoke.result",
|
||||
validator: validateNodeInvokeResultParams,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const p = normalizedParams as {
|
||||
id: string;
|
||||
nodeId: string;
|
||||
ok: boolean;
|
||||
payload?: unknown;
|
||||
payloadJSON?: string | null;
|
||||
error?: { code?: string; message?: string } | null;
|
||||
};
|
||||
const callerNodeId = client?.connect?.device?.id ?? client?.connect?.client?.id;
|
||||
if (callerNodeId && callerNodeId !== p.nodeId) {
|
||||
respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, "nodeId mismatch"));
|
||||
return;
|
||||
}
|
||||
|
||||
const ok = context.nodeRegistry.handleInvokeResult({
|
||||
id: p.id,
|
||||
nodeId: p.nodeId,
|
||||
ok: p.ok,
|
||||
payload: p.payload,
|
||||
payloadJSON: p.payloadJSON ?? null,
|
||||
error: p.error ?? null,
|
||||
});
|
||||
if (!ok) {
|
||||
// Late-arriving results (after invoke timeout) are expected and harmless.
|
||||
// Return success instead of error to reduce log noise; client can discard.
|
||||
context.logGateway.debug(`late invoke result ignored: id=${p.id} node=${p.nodeId}`);
|
||||
respond(true, { ok: true, ignored: true }, undefined);
|
||||
return;
|
||||
}
|
||||
|
||||
respond(true, { ok: true }, undefined);
|
||||
};
|
||||
@@ -26,6 +26,7 @@ import {
|
||||
validateNodePairVerifyParams,
|
||||
validateNodeRenameParams,
|
||||
} from "../protocol/index.js";
|
||||
import { handleNodeInvokeResult } from "./nodes.handlers.invoke-result.js";
|
||||
import {
|
||||
respondInvalidParams,
|
||||
respondUnavailableOnThrow,
|
||||
@@ -43,26 +44,6 @@ function isNodeEntry(entry: { role?: string; roles?: string[] }) {
|
||||
return false;
|
||||
}
|
||||
|
||||
function normalizeNodeInvokeResultParams(params: unknown): unknown {
|
||||
if (!params || typeof params !== "object") {
|
||||
return params;
|
||||
}
|
||||
const raw = params as Record<string, unknown>;
|
||||
const normalized: Record<string, unknown> = { ...raw };
|
||||
if (normalized.payloadJSON === null) {
|
||||
delete normalized.payloadJSON;
|
||||
} else if (normalized.payloadJSON !== undefined && typeof normalized.payloadJSON !== "string") {
|
||||
if (normalized.payload === undefined) {
|
||||
normalized.payload = normalized.payloadJSON;
|
||||
}
|
||||
delete normalized.payloadJSON;
|
||||
}
|
||||
if (normalized.error === null) {
|
||||
delete normalized.error;
|
||||
}
|
||||
return normalized;
|
||||
}
|
||||
|
||||
export const nodeHandlers: GatewayRequestHandlers = {
|
||||
"node.pair.request": async ({ params, respond, context }) => {
|
||||
if (!validateNodePairRequestParams(params)) {
|
||||
@@ -477,46 +458,7 @@ export const nodeHandlers: GatewayRequestHandlers = {
|
||||
);
|
||||
});
|
||||
},
|
||||
"node.invoke.result": async ({ params, respond, context, client }) => {
|
||||
const normalizedParams = normalizeNodeInvokeResultParams(params);
|
||||
if (!validateNodeInvokeResultParams(normalizedParams)) {
|
||||
respondInvalidParams({
|
||||
respond,
|
||||
method: "node.invoke.result",
|
||||
validator: validateNodeInvokeResultParams,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const p = normalizedParams as {
|
||||
id: string;
|
||||
nodeId: string;
|
||||
ok: boolean;
|
||||
payload?: unknown;
|
||||
payloadJSON?: string | null;
|
||||
error?: { code?: string; message?: string } | null;
|
||||
};
|
||||
const callerNodeId = client?.connect?.device?.id ?? client?.connect?.client?.id;
|
||||
if (callerNodeId && callerNodeId !== p.nodeId) {
|
||||
respond(false, undefined, errorShape(ErrorCodes.INVALID_REQUEST, "nodeId mismatch"));
|
||||
return;
|
||||
}
|
||||
const ok = context.nodeRegistry.handleInvokeResult({
|
||||
id: p.id,
|
||||
nodeId: p.nodeId,
|
||||
ok: p.ok,
|
||||
payload: p.payload,
|
||||
payloadJSON: p.payloadJSON ?? null,
|
||||
error: p.error ?? null,
|
||||
});
|
||||
if (!ok) {
|
||||
// Late-arriving results (after invoke timeout) are expected and harmless.
|
||||
// Return success instead of error to reduce log noise; client can discard.
|
||||
context.logGateway.debug(`late invoke result ignored: id=${p.id} node=${p.nodeId}`);
|
||||
respond(true, { ok: true, ignored: true }, undefined);
|
||||
return;
|
||||
}
|
||||
respond(true, { ok: true }, undefined);
|
||||
},
|
||||
"node.invoke.result": handleNodeInvokeResult,
|
||||
"node.event": async ({ params, respond, context, client }) => {
|
||||
if (!validateNodeEventParams(params)) {
|
||||
respondInvalidParams({
|
||||
|
||||
@@ -1,65 +1,9 @@
|
||||
import { afterAll, beforeAll, describe, expect, test, vi } from "vitest";
|
||||
import { WebSocket } from "ws";
|
||||
import { loadOrCreateDeviceIdentity } from "../infra/device-identity.js";
|
||||
import { GATEWAY_CLIENT_MODES, GATEWAY_CLIENT_NAMES } from "../utils/message-channel.js";
|
||||
|
||||
vi.mock("../infra/update-runner.js", () => ({
|
||||
runGatewayUpdate: vi.fn(async () => ({
|
||||
status: "ok",
|
||||
mode: "git",
|
||||
root: "/repo",
|
||||
steps: [],
|
||||
durationMs: 12,
|
||||
})),
|
||||
}));
|
||||
|
||||
import {
|
||||
connectOk,
|
||||
getFreePort,
|
||||
installGatewayTestHooks,
|
||||
rpcReq,
|
||||
startGatewayServer,
|
||||
} from "./test-helpers.js";
|
||||
import { testState } from "./test-helpers.mocks.js";
|
||||
|
||||
installGatewayTestHooks({ scope: "suite" });
|
||||
|
||||
let server: Awaited<ReturnType<typeof startGatewayServer>>;
|
||||
let port: number;
|
||||
let nodeWs: WebSocket;
|
||||
let nodeId: string;
|
||||
|
||||
beforeAll(async () => {
|
||||
const token = "test-gateway-token-1234567890";
|
||||
testState.gatewayAuth = { mode: "token", token };
|
||||
port = await getFreePort();
|
||||
server = await startGatewayServer(port, { bind: "loopback" });
|
||||
|
||||
nodeWs = new WebSocket(`ws://127.0.0.1:${port}`);
|
||||
await new Promise<void>((resolve) => nodeWs.once("open", resolve));
|
||||
|
||||
const identity = loadOrCreateDeviceIdentity();
|
||||
nodeId = identity.deviceId;
|
||||
await connectOk(nodeWs, {
|
||||
role: "node",
|
||||
client: {
|
||||
id: GATEWAY_CLIENT_NAMES.NODE_HOST,
|
||||
version: "1.0.0",
|
||||
platform: "darwin",
|
||||
mode: GATEWAY_CLIENT_MODES.NODE,
|
||||
},
|
||||
commands: ["canvas.snapshot"],
|
||||
token,
|
||||
});
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
nodeWs.terminate();
|
||||
await server.close();
|
||||
});
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
import { handleNodeInvokeResult } from "./server-methods/nodes.handlers.invoke-result.js";
|
||||
|
||||
describe("late-arriving invoke results", () => {
|
||||
test("returns success for unknown invoke ids for both success and error payloads", async () => {
|
||||
const nodeId = "node-123";
|
||||
const cases = [
|
||||
{
|
||||
id: "unknown-invoke-id-12345",
|
||||
@@ -74,19 +18,31 @@ describe("late-arriving invoke results", () => {
|
||||
] as const;
|
||||
|
||||
for (const params of cases) {
|
||||
const result = await rpcReq<{ ok?: boolean; ignored?: boolean }>(
|
||||
nodeWs,
|
||||
"node.invoke.result",
|
||||
{
|
||||
...params,
|
||||
nodeId,
|
||||
},
|
||||
);
|
||||
const respond = vi.fn();
|
||||
const context = {
|
||||
nodeRegistry: { handleInvokeResult: () => false },
|
||||
logGateway: { debug: vi.fn() },
|
||||
} as any;
|
||||
const client = {
|
||||
connect: { device: { id: nodeId } },
|
||||
} as any;
|
||||
|
||||
await handleNodeInvokeResult({
|
||||
req: { method: "node.invoke.result" } as any,
|
||||
params: { ...params, nodeId } as any,
|
||||
client,
|
||||
isWebchatConnect: () => false,
|
||||
respond,
|
||||
context,
|
||||
});
|
||||
|
||||
const [ok, payload, error] = respond.mock.lastCall ?? [];
|
||||
|
||||
// Late-arriving results return success instead of error to reduce log noise.
|
||||
expect(result.ok).toBe(true);
|
||||
expect(result.payload?.ok).toBe(true);
|
||||
expect(result.payload?.ignored).toBe(true);
|
||||
expect(ok).toBe(true);
|
||||
expect(error).toBeUndefined();
|
||||
expect(payload?.ok).toBe(true);
|
||||
expect(payload?.ignored).toBe(true);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user