ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(scripts): prevent gateway benchmarks from stalling on probes (#119063)

Browse Source 
* fix(scripts): bound gateway benchmark probes

* fix(scripts): bound concurrency response deadlines

* fix(perf): preserve benchmark socket reuse
This commit is contained in:
Vincent Koc
2026-08-04 16:23:45 +08:00
committed by GitHub
parent b0c5d48f74
commit 98feafe785
4 changed files with 268 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/bench-gateway-concurrency.ts
+36 -15
View File
@@ -239,11 +239,23 @@ async function requestHttp(params: {
port: number;
}): Promise<{ body: string; latencyMs: number; status: number }> {
const startedAt = performance.now();
const timeoutMs = Math.max(
1,
Math.min(HTTP_TIMEOUT_MS, requireRemainingMs(params.deadlineAt, `requesting ${params.path}`)),
);
const requestDeadlineAt = Math.min(params.deadlineAt, startedAt + HTTP_TIMEOUT_MS);
requireRemainingMs(requestDeadlineAt, `requesting ${params.path}`);
return await new Promise((resolve, reject) => {
let settled = false;
const settle = (run: () => void) => {
if (settled) {
return;
}
settled = true;
clearTimeout(timer);
run();
};
const fail = (error: Error) =>
settle(() => {
req.destroy();
reject(error);
});
const req = request(
{
headers: { accept: params.accept },
@@ -251,7 +263,6 @@ async function requestHttp(params: {
method: "GET",
path: params.path,
port: params.port,
timeout: timeoutMs,
},
(res) => {
const chunks: Buffer[] = [];
@@ -259,22 +270,31 @@ async function requestHttp(params: {
res.on("data", (chunk: Buffer) => {
bytes += chunk.length;
if (bytes > MAX_HTTP_BODY_BYTES) {
req.destroy(new Error(`${params.path} response exceeded ${MAX_HTTP_BODY_BYTES} bytes`));
fail(new Error(`${params.path} response exceeded ${MAX_HTTP_BODY_BYTES} bytes`));
return;
}
chunks.push(chunk);
});
res.on("end", () => {
resolve({
body: Buffer.concat(chunks).toString("utf8"),
latencyMs: performance.now() - startedAt,
status: res.statusCode ?? 0,
});
});
res.once("aborted", () => fail(new Error(`${params.path} response aborted`)));
res.once("error", fail);
res.once("end", () =>
settle(() =>
resolve({
body: Buffer.concat(chunks).toString("utf8"),
latencyMs: performance.now() - startedAt,
status: res.statusCode ?? 0,
}),
),
);
},
);
req.once("error", reject);
req.once("timeout", () => req.destroy(new Error(`${params.path} request timed out`)));
req.once("error", fail);
// Request/socket timeouts measure inactivity; this timer owns the wall-clock deadline.
const timer = setTimeout(
() => fail(new Error(`${params.path} request timed out`)),
Math.max(1, Math.ceil(remainingMs(requestDeadlineAt))),
);
timer.unref?.();
req.end();
});
}
@@ -841,6 +861,7 @@ export const testing = {
parseOptions,
formatProbeFailure,
formatRunFailure,
requestHttp,
runTurn,
sampleGateway,
summarizeNumbers,
scripts/lib/gateway-bench-probes.ts
+23 -10
View File
@@ -4,6 +4,8 @@ import { request } from "node:http";
import { createServer } from "node:net";
import { expectDefined } from "../../packages/normalization-core/src/expect.ts";
const PROBE_REQUEST_TIMEOUT_MS = 100;
export async function getFreePort(): Promise<number> {
return new Promise((resolve, reject) => {
const server = createServer();
@@ -133,17 +135,28 @@ export function readProcessTreeCpuMs(rootPid: number | undefined): number | null
function requestStatus(port: number, pathname: string): Promise<number> {
return new Promise((resolve, reject) => {
const req = request(
{ host: "127.0.0.1", method: "GET", path: pathname, port, timeout: 100 },
(res) => {
res.resume();
res.on("end", () => resolve(res.statusCode ?? 0));
},
);
req.on("error", reject);
req.on("timeout", () => {
req.destroy(new Error("probe timeout"));
let settled = false;
const settle = (run: () => void) => {
if (settled) {
return;
}
settled = true;
clearTimeout(timer);
run();
};
const req = request({ host: "127.0.0.1", method: "HEAD", path: pathname, port }, (res) => {
const status = res.statusCode ?? 0;
// Gateway probe HEAD responses carry the same status without a body to drain.
settle(() => resolve(status));
});
req.on("error", (error) => settle(() => reject(error)));
// Socket timeouts reset on activity, so enforce the attempt budget as wall-clock time.
const timer = setTimeout(() => {
const error = new Error("probe timeout");
settle(() => reject(error));
req.destroy(error);
}, PROBE_REQUEST_TIMEOUT_MS);
timer.unref?.();
req.end();
});
}
test/scripts/bench-gateway-concurrency.test.ts
+110 -3
View File
@@ -1,6 +1,7 @@
// Gateway concurrency benchmark tests cover CLI parsing and bounded percentile summaries.
import { spawnSync } from "node:child_process";
import { createServer } from "node:http";
import { createServer as createHttpServer } from "node:http";
import { createServer as createRawServer, type Socket } from "node:net";
import { performance } from "node:perf_hooks";
import { describe, expect, it } from "vitest";
import { testing } from "../../scripts/bench-gateway-concurrency.ts";
@@ -77,12 +78,14 @@ describe("gateway concurrency benchmark script", () => {
it("preserves HTTP and RPC failures in baseline probe diagnostics", async () => {
const probeOrder: string[] = [];
const server = createServer((req, res) => {
const server = createHttpServer((req, res) => {
probeOrder.push(req.url ?? "missing-url");
res.statusCode = req.url === "/readyz" ? 503 : 200;
res.end(req.url === "/readyz" ? '{"status":"starting"}' : "not html");
});
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
await new Promise<void>((resolve) => {
server.listen(0, "127.0.0.1", resolve);
});
const address = server.address();
if (!address || typeof address === "string") {
server.close();
@@ -148,6 +151,110 @@ describe("gateway concurrency benchmark script", () => {
}
});
it("bounds trickled response bodies by the benchmark deadline", async () => {
const sockets = new Set<Socket>();
let bodyChunksSent = 0;
let serverEndedResponse = false;
const server = createRawServer((socket) => {
sockets.add(socket);
socket.setNoDelay(true);
socket.on("error", () => {});
socket.once("close", () => sockets.delete(socket));
socket.once("data", () => {
socket.write(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n ",
);
bodyChunksSent += 1;
const interval = setInterval(() => {
socket.write(" ");
bodyChunksSent += 1;
}, 10);
const endTimer = setTimeout(() => {
serverEndedResponse = true;
socket.end();
}, 500);
socket.once("close", () => {
clearInterval(interval);
clearTimeout(endTimer);
});
});
});
await new Promise<void>((resolve) => {
server.listen(0, "127.0.0.1", () => resolve());
});
const address = server.address();
if (!address || typeof address === "string") {
server.close();
throw new Error("expected raw HTTP test server address");
}
const startedAt = performance.now();
try {
await expect(
testing.requestHttp({
accept: "application/json",
deadlineAt: startedAt + 150,
path: "/readyz",
port: address.port,
}),
).rejects.toThrow("/readyz request timed out");
expect(bodyChunksSent).toBeGreaterThan(1);
expect(serverEndedResponse).toBe(false);
} finally {
for (const socket of sockets) {
socket.destroy();
}
await new Promise<void>((resolve) => {
server.close(() => resolve());
});
}
});
it("reuses one connection for sequential successful HTTP samples", async () => {
let connectionCount = 0;
const server = createHttpServer((request, response) => {
response.setHeader(
"content-type",
request.url === "/readyz" ? "application/json" : "text/html",
);
response.end(request.url === "/readyz" ? '{"status":"ok"}' : "<html></html>");
});
server.on("connection", () => {
connectionCount += 1;
});
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", resolve);
});
try {
const address = server.address();
if (!address || typeof address === "string") {
throw new Error("expected HTTP test server address");
}
const deadlineAt = performance.now() + 5_000;
await testing.requestHttp({
accept: "application/json",
deadlineAt,
path: "/readyz",
port: address.port,
});
await testing.requestHttp({
accept: "text/html",
deadlineAt,
path: "/",
port: address.port,
});
expect(connectionCount).toBe(1);
} finally {
server.closeAllConnections();
await new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
});
}
});
it("loads through native Node TypeScript stripping", () => {
const result = spawnSync(process.execPath, ["scripts/bench-gateway-concurrency.ts", "--help"], {
cwd: process.cwd(),
test/scripts/bench-gateway-restart.test.ts
+99
View File
@@ -2,11 +2,13 @@
import { spawn } from "node:child_process";
import fs from "node:fs";
import { createServer } from "node:http";
import { createServer as createNetServer, type Socket } from "node:net";
import os from "node:os";
import path from "node:path";
import { performance } from "node:perf_hooks";
import { beforeAll, describe, expect, it } from "vitest";
import { testing } from "../../scripts/bench-gateway-restart.ts";
import { requestProbeStatus } from "../../scripts/lib/gateway-bench-probes.ts";
import {
executeSqliteQueryTakeFirstSync,
getNodeSqliteKysely,
@@ -26,6 +28,25 @@ type BenchCliResult = {
stdout: string;
};
async function withWallClockDeadline<T>(
promise: Promise<T>,
timeoutMs: number,
label: string,
): Promise<T> {
let timer: NodeJS.Timeout | undefined;
try {
return await Promise.race([
promise,
new Promise<never>((_resolve, reject) => {
timer = setTimeout(() => reject(new Error(`${label} exceeded ${timeoutMs}ms`)), timeoutMs);
timer.unref?.();
}),
]);
} finally {
clearTimeout(timer);
}
}
function runBenchCli(args: string[]): Promise<BenchCliResult> {
return new Promise((resolve, reject) => {
const child = spawn(
@@ -244,6 +265,84 @@ node 1234 user 12u IPv4 0t0 TCP localhost:1234
expect(testing.parseProcessRssKb("")).toBeNull();
});
it("accepts healthy probe headers without waiting for the response body", async () => {
let requestMethod: string | undefined;
const server = createServer((request, response) => {
requestMethod = request.method;
response.writeHead(200, { "content-type": "application/json" });
response.flushHeaders();
});
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", resolve);
});
try {
const address = server.address();
if (!address || typeof address === "string") {
throw new Error("test server did not bind to a TCP port");
}
await expect(
withWallClockDeadline(
requestProbeStatus(address.port, "/healthz"),
750,
"healthy-header probe",
),
).resolves.toEqual({ errorKind: null, status: 200 });
expect(requestMethod).toBe("HEAD");
} finally {
server.closeAllConnections();
await new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
});
}
});
it("bounds probes by a wall-clock deadline before response headers arrive", async () => {
const sockets = new Set<Socket>();
const intervals = new Set<NodeJS.Timeout>();
const server = createNetServer((socket) => {
sockets.add(socket);
socket.on("error", () => undefined);
socket.on("close", () => sockets.delete(socket));
socket.once("data", () => {
socket.write("HTTP/1.1 200 OK\r\nX-Drip: ");
// Keep the socket active without completing headers so an idle timeout cannot end the probe.
const interval = setInterval(() => socket.write("x"), 20);
intervals.add(interval);
interval.unref?.();
socket.on("close", () => {
clearInterval(interval);
intervals.delete(interval);
});
});
});
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", resolve);
});
try {
const address = server.address();
if (!address || typeof address === "string") {
throw new Error("test server did not bind to a TCP port");
}
await expect(
withWallClockDeadline(requestProbeStatus(address.port, "/readyz"), 750, "headerless probe"),
).resolves.toEqual({ errorKind: "timeout", status: null });
} finally {
for (const interval of intervals) {
clearInterval(interval);
}
for (const socket of sockets) {
socket.destroy();
}
await new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
});
}
});
it("enables both startup and restart trace in the child gateway environment", () => {
const env = testing.sanitizedEnv("/tmp/openclaw-bench", "/tmp/openclaw-bench/config.json", {
config: {},