From 98feafe78506fabb7c928585a79bb803305ad061 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Tue, 4 Aug 2026 16:23:45 +0800 Subject: [PATCH] fix(scripts): prevent gateway benchmarks from stalling on probes (#119063) * fix(scripts): bound gateway benchmark probes * fix(scripts): bound concurrency response deadlines * fix(perf): preserve benchmark socket reuse --- scripts/bench-gateway-concurrency.ts | 51 +++++--- scripts/lib/gateway-bench-probes.ts | 33 +++-- .../scripts/bench-gateway-concurrency.test.ts | 113 +++++++++++++++++- test/scripts/bench-gateway-restart.test.ts | 99 +++++++++++++++ 4 files changed, 268 insertions(+), 28 deletions(-) diff --git a/scripts/bench-gateway-concurrency.ts b/scripts/bench-gateway-concurrency.ts index e55f977e3ccb..e04f7c7a7741 100644 --- a/scripts/bench-gateway-concurrency.ts +++ b/scripts/bench-gateway-concurrency.ts @@ -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, diff --git a/scripts/lib/gateway-bench-probes.ts b/scripts/lib/gateway-bench-probes.ts index ecb9a260719b..c1422cb7af0e 100644 --- a/scripts/lib/gateway-bench-probes.ts +++ b/scripts/lib/gateway-bench-probes.ts @@ -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 { 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 { 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(); }); } diff --git a/test/scripts/bench-gateway-concurrency.test.ts b/test/scripts/bench-gateway-concurrency.test.ts index 8b2bb9bdca39..f3fb0fbeee84 100644 --- a/test/scripts/bench-gateway-concurrency.test.ts +++ b/test/scripts/bench-gateway-concurrency.test.ts @@ -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((resolve) => server.listen(0, "127.0.0.1", resolve)); + await new Promise((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(); + 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((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((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"}' : ""); + }); + server.on("connection", () => { + connectionCount += 1; + }); + await new Promise((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((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(), diff --git a/test/scripts/bench-gateway-restart.test.ts b/test/scripts/bench-gateway-restart.test.ts index 12120d8a9d79..8ae261f6edce 100644 --- a/test/scripts/bench-gateway-restart.test.ts +++ b/test/scripts/bench-gateway-restart.test.ts @@ -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( + promise: Promise, + timeoutMs: number, + label: string, +): Promise { + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + promise, + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(`${label} exceeded ${timeoutMs}ms`)), timeoutMs); + timer.unref?.(); + }), + ]); + } finally { + clearTimeout(timer); + } +} + function runBenchCli(args: string[]): Promise { 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((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((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(); + const intervals = new Set(); + 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((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((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: {},