From c05ec96ad6c6fb9ee9f566981fcf5e1ceacfee7a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=A5=E9=BA=92?= Date: Mon, 19 Oct 2026 16:48:57 +0800 Subject: [PATCH] fix(http): align Node and Bun timeouts and response cleanup - Preserve native fetch in both runtimes without upgrading Bun - Apply --timeout to request deadlines and stream idle protection - Abort underlying requests on cancellation and release unused Sandbox response bodies - Add regression tests for Node, Bun, and compiled binaries --- .github/workflows/ci.yml | 5 + .../commands/src/commands/sandbox/instance.ts | 6 +- .../commands/src/commands/sandbox/template.ts | 3 +- packages/commands/tests/sandbox.test.ts | 50 ++++ packages/core/src/client/http.ts | 153 ++++++++-- packages/core/src/client/stream.ts | 3 + packages/core/tests/http.test.ts | 157 +++++++++- packages/runtime/tests/http-bun.e2e.test.ts | 270 ++++++++++++++++++ .../runtime/tests/http-timeout.e2e.test.ts | 11 +- 9 files changed, 627 insertions(+), 31 deletions(-) create mode 100644 packages/runtime/tests/http-bun.e2e.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ef8ff66ca..743b073ff 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,6 +23,11 @@ jobs: node-version: "22" cache: pnpm + # Keep native-fetch and compiled Binary tests on the release Bun version. + - uses: oven-sh/setup-bun@v2 + with: + bun-version: "1.3.14" + - run: pnpm install --frozen-lockfile - run: pnpm run sync:skill-assets diff --git a/packages/commands/src/commands/sandbox/instance.ts b/packages/commands/src/commands/sandbox/instance.ts index 1893083b7..141de52e3 100644 --- a/packages/commands/src/commands/sandbox/instance.ts +++ b/packages/commands/src/commands/sandbox/instance.ts @@ -383,7 +383,8 @@ export const sandboxPause = defineCommand({ emitResult({ method: "POST", endpoint, request: null }, format); return; } - await ctx.client.request({ path: endpoint, method: "POST" }); + const response = await ctx.client.request({ path: endpoint, method: "POST" }); + await response.body?.cancel(); if (ctx.settings.quiet) emitBare(ctx.flags.sandboxId); else emitResult({ sandboxID: ctx.flags.sandboxId, paused: true }, format); }, @@ -410,7 +411,8 @@ export const sandboxDelete = defineCommand({ emitResult({ method: "DELETE", endpoint, request: null }, format); return; } - await ctx.client.request({ path: endpoint, method: "DELETE" }); + const response = await ctx.client.request({ path: endpoint, method: "DELETE" }); + await response.body?.cancel(); if (ctx.settings.quiet) emitBare(ctx.flags.sandboxId); else emitResult({ sandboxID: ctx.flags.sandboxId, deleted: true }, format); }, diff --git a/packages/commands/src/commands/sandbox/template.ts b/packages/commands/src/commands/sandbox/template.ts index cee17fb62..4b40a4b54 100644 --- a/packages/commands/src/commands/sandbox/template.ts +++ b/packages/commands/src/commands/sandbox/template.ts @@ -644,7 +644,8 @@ export const sandboxTemplateDelete = defineCommand({ emitResult({ method: "DELETE", endpoint, request: null }, format); return; } - await ctx.client.request({ path: endpoint, method: "DELETE" }); + const response = await ctx.client.request({ path: endpoint, method: "DELETE" }); + await response.body?.cancel(); if (ctx.settings.quiet) emitBare(ctx.flags.templateId); else emitResult({ templateID: ctx.flags.templateId, deleted: true }, format); }, diff --git a/packages/commands/tests/sandbox.test.ts b/packages/commands/tests/sandbox.test.ts index 891e72f08..cbd7cc585 100644 --- a/packages/commands/tests/sandbox.test.ts +++ b/packages/commands/tests/sandbox.test.ts @@ -1,3 +1,4 @@ +import { getEventListeners } from "node:events"; import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -640,3 +641,52 @@ test("Sandbox validation errors use CLI usage exit codes", async () => { BailianError, ); }); + +describe("Sandbox unused response cleanup", () => { + test.each([ + { name: "pause instance", command: sandboxPause, flags: { sandboxId: "sandbox-test" } }, + { name: "delete instance", command: sandboxDelete, flags: { sandboxId: "sandbox-test" } }, + { + name: "delete template", + command: sandboxTemplateDelete, + flags: { templateId: "template-test" }, + }, + ])("$name releases complete and stalled response bodies", async ({ command, flags }) => { + vi.spyOn(process.stdout, "write").mockImplementation(() => true); + const client = new Client({ + identity: { + binName: "bl", + version: "test", + npmPackage: "bailian-cli", + clientName: "bailian-cli", + }, + settings: SETTINGS, + baseUrl: "https://dashscope.aliyuncs.com", + apiCred: { + token: "local-test-only", + baseUrl: "https://dashscope.aliyuncs.com", + source: "flag", + }, + }); + for (const complete of [true, false]) { + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("unused response")); + if (complete) controller.close(); + }, + }); + const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(new Response(body)); + await command.run({ + identity: { binName: "bl" }, + settings: SETTINGS, + flags: { workspaceId: "ws-test", ...flags }, + client, + } as never); + const signal = fetchMock.mock.calls[0][1]!.signal!; + expect(signal.aborted).toBe(true); + expect(getEventListeners(signal, "abort")).toHaveLength(0); + expect(body.locked).toBe(false); + fetchMock.mockRestore(); + } + }); +}); diff --git a/packages/core/src/client/http.ts b/packages/core/src/client/http.ts index e144ab8cc..f945fd571 100644 --- a/packages/core/src/client/http.ts +++ b/packages/core/src/client/http.ts @@ -7,12 +7,11 @@ import { maskToken } from "../utils/token.ts"; import { sourceConfig, trackingHeaders } from "./headers.ts"; import { Dispatcher, getGlobalDispatcher } from "undici"; -// 请求已由 --timeout 对应的 signal 控制超时。这里在实际发送前关闭底层默认的 -// 300 秒响应头超时，避免用户设置更长等待时间后，请求仍在 300 秒时被提前中断。 -// 只覆盖 headersTimeout，响应体的空闲超时保持原值。 -const withoutHeadersTimeout: Dispatcher.DispatcherComposeInterceptor = +// Core owns the request deadline and body idle window. Override transport +// limits only for this request, preserving the current proxy/connection pool. +const withoutTransportTimeouts: Dispatcher.DispatcherComposeInterceptor = (dispatch) => (options, handler) => - dispatch({ ...options, headersTimeout: 0 }, handler); + dispatch({ ...options, headersTimeout: 0, bodyTimeout: 0 }, handler); /** 传输层依赖:UA 用 identity,timeout/verbose 用 settings。凭证由调用方(Client)注头。 */ export interface HttpDeps { @@ -42,14 +41,19 @@ function bodyReferencesOssUrl(body: unknown): boolean { return JSON.stringify(body).includes("oss://"); } +/** Successful raw response bodies must be consumed or explicitly cancelled by the caller. */ export async function request(deps: HttpDeps, opts: RequestOpts): Promise { const timeoutMs = (opts.timeout ?? deps.settings.timeout) * 1000; const requestSignal = createRequestSignal(timeoutMs, opts.signal); try { - // Raw/streaming callers retain ownership of successful response bodies. - return await performRequest(deps, opts, requestSignal.signal); - } finally { + const response = await performRequest(deps, opts, requestSignal.signal); + // After headers, raw/streaming responses use a rolling idle window rather + // than a total deadline. The body owns cleanup until EOF/cancel/error. + return protectResponseBody(response, requestSignal); + } catch (error) { requestSignal.cleanup(); + if (requestSignal.signal.aborted) throw requestSignal.signal.reason; + throw error; } } @@ -94,12 +98,16 @@ async function performRequest( : JSON.stringify(opts.body) : undefined, signal, - // 复用当前客户端，仅关闭本次请求的响应头超时。 - // 从原型调用 compose，兼容没有 compose 方法的客户端；类型断言适配 Node 的 fetch 类型。 - dispatcher: Dispatcher.prototype.compose.call( - getGlobalDispatcher(), - withoutHeadersTimeout, - ) as unknown as RequestInit["dispatcher"], + // Bun's native fetch ignores dispatchers and its undici shim has no compose. + // Keep native fetch in both runtimes, adapting only transport controls. + ...(process.versions.bun + ? { timeout: false as const } + : { + dispatcher: Dispatcher.prototype.compose.call( + getGlobalDispatcher(), + withoutTransportTimeouts, + ) as unknown as RequestInit["dispatcher"], + }), }; const res = await fetch(opts.url, requestInit); @@ -125,14 +133,26 @@ async function performRequest( return res; } -function createRequestSignal( - timeoutMs: number, - parentSignal?: AbortSignal, -): { signal: AbortSignal; cleanup: () => void } { +interface RequestSignal { + signal: AbortSignal; + cleanup: () => void; + abort: (reason?: unknown) => void; + resetIdleTimeout: () => void; + pauseTimeout: () => void; +} + +function createRequestSignal(timeoutMs: number, parentSignal?: AbortSignal): RequestSignal { const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), timeoutMs); + let timeout: ReturnType | undefined; + let finished = false; + const armTimer = (reason?: unknown) => { + if (finished) return; + clearTimeout(timeout); + timeout = setTimeout(() => controller.abort(reason), timeoutMs); + }; const abortFromParent = () => controller.abort(parentSignal?.reason); const cleanup = () => { + finished = true; clearTimeout(timeout); parentSignal?.removeEventListener("abort", abortFromParent); controller.signal.removeEventListener("abort", cleanup); @@ -140,9 +160,100 @@ function createRequestSignal( controller.signal.addEventListener("abort", cleanup, { once: true }); if (parentSignal?.aborted) abortFromParent(); - else parentSignal?.addEventListener("abort", abortFromParent, { once: true }); + else { + parentSignal?.addEventListener("abort", abortFromParent, { once: true }); + armTimer(); + } + + return { + signal: controller.signal, + cleanup, + abort: (reason) => controller.abort(reason), + pauseTimeout: () => clearTimeout(timeout), + resetIdleTimeout: () => + armTimer( + new BailianError( + `Response body idle timeout: no data received for ${timeoutMs / 1000}s.`, + ExitCode.TIMEOUT, + "Increase --timeout to allow a longer interval between response chunks.", + ), + ), + }; +} - return { signal: controller.signal, cleanup }; +function protectResponseBody(response: Response, requestSignal: RequestSignal): Response { + if (!response.body) { + requestSignal.cleanup(); + return response; + } + // Only count time spent waiting for network data, not consumer processing. + // An unused raw response must not keep a command alive through a timer. + requestSignal.pauseTimeout(); + const reader = response.body.getReader(); + const { signal } = requestSignal; + let finished = false; + let onAbort: () => void; + const finish = () => { + finished = true; + signal.removeEventListener("abort", onAbort); + requestSignal.cleanup(); + }; + const cancelSource = (reason?: unknown) => + reader.cancel(reason).finally(() => reader.releaseLock()); + const body = new ReadableStream( + { + start(controller) { + onAbort = () => { + if (finished) return; + finish(); + controller.error(signal.reason); + void cancelSource(signal.reason).catch(() => {}); + }; + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) onAbort(); + }, + async pull(controller) { + try { + requestSignal.resetIdleTimeout(); + const part = await reader.read(); + requestSignal.pauseTimeout(); + if (finished) return; + if (part.done) { + finish(); + reader.releaseLock(); + controller.close(); + } else { + controller.enqueue(part.value); + } + } catch (error) { + if (finished) return; + finish(); + reader.releaseLock(); + controller.error(signal.aborted ? signal.reason : error); + } + }, + cancel(reason) { + finish(); + // Own transport teardown as well as reader cleanup: cancelling the + // reader alone can leave native Bun's HTTP connection open. + requestSignal.abort(reason); + return cancelSource(reason).catch(() => {}); + }, + }, + { highWaterMark: 0 }, + ); + const protectedResponse = new Response(body, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); + // A constructed Response otherwise loses fetch's URL/redirect metadata. + Object.defineProperties(protectedResponse, { + url: { value: response.url }, + redirected: { value: response.redirected }, + type: { value: response.type }, + }); + return protectedResponse; } export async function requestJson(deps: HttpDeps, opts: RequestOpts): Promise { diff --git a/packages/core/src/client/stream.ts b/packages/core/src/client/stream.ts index 05e70ca3e..ba2ad81e1 100644 --- a/packages/core/src/client/stream.ts +++ b/packages/core/src/client/stream.ts @@ -130,6 +130,9 @@ export async function* parseSSE(response: Response): AsyncGenerator {}); reader.releaseLock(); } } diff --git a/packages/core/tests/http.test.ts b/packages/core/tests/http.test.ts index c80681613..e0634fcdc 100644 --- a/packages/core/tests/http.test.ts +++ b/packages/core/tests/http.test.ts @@ -6,6 +6,7 @@ import { Client } from "../src/client/client.ts"; import { request, requestJson, type HttpDeps } from "../src/client/http.ts"; import { BailianError } from "../src/errors/base.ts"; import { ExitCode } from "../src/errors/codes.ts"; +import { parseSSE } from "../src/client/stream.ts"; function testDeps(timeout = 5): HttpDeps { return { @@ -300,7 +301,7 @@ describe("request signal cleanup", () => { }, ); - test("raw success returns the identical unconsumed Response and detaches at headers", async () => { + test("raw success keeps cancellation without an active timer before body consumption", async () => { vi.useFakeTimers(); const parent = new AbortController(); const response = new Response('{"stream":"still readable"}'); @@ -311,14 +312,162 @@ describe("request signal cleanup", () => { stream: true, }); const signal = fetchMock.mock.calls[0][1]!.signal!; - expect(result).toBe(response); expect(result.bodyUsed).toBe(false); expect(vi.getTimerCount()).toBe(0); + expect(getEventListeners(parent.signal, "abort")).toHaveLength(1); + await expect(result.json()).resolves.toEqual({ stream: "still readable" }); + expect(vi.getTimerCount()).toBe(0); expect(getEventListeners(parent.signal, "abort")).toHaveLength(0); expect(getEventListeners(signal, "abort")).toHaveLength(0); parent.abort(); - await vi.advanceTimersByTimeAsync(10_000); expect(signal.aborted).toBe(false); - await expect(result.json()).resolves.toEqual({ stream: "still readable" }); }); }); + +describe("raw response idle protection", () => { + test("excludes consumer processing time from the network idle window", async () => { + await withServer( + (_incoming, response) => { + response.write("first"); + const timer = setInterval(() => response.write("next"), 20); + response.on("close", () => clearInterval(timer)); + }, + async (url) => { + const response = await request(testDeps(0.1), { url, stream: true }); + const reader = response.body!.getReader(); + try { + expect((await reader.read()).done).toBe(false); + await new Promise((resolve) => setTimeout(resolve, 180)); + expect((await reader.read()).done).toBe(false); + } finally { + await reader.cancel(); + reader.releaseLock(); + } + }, + ); + }); + + test("aborts a stalled body using the per-request timeout", async () => { + await withServer( + (_incoming, response) => response.write("first"), + async (url) => { + const response = await request(testDeps(5), { url, timeout: 0.15, stream: true }); + await expect(response.text()).rejects.toMatchObject({ + name: "BailianError", + exitCode: ExitCode.TIMEOUT, + message: expect.stringContaining("idle"), + }); + }, + ); + }); + + test("keeps a progressing stream alive beyond its timeout window", async () => { + await withServer( + (_incoming, response) => { + let chunks = 0; + const timer = setInterval(() => { + response.write("x"); + if (++chunks === 8) { + clearInterval(timer); + response.end(); + } + }, 50); + response.on("close", () => clearInterval(timer)); + }, + async (url) => { + const started = performance.now(); + const response = await request(testDeps(0.2), { url, stream: true }); + await expect(response.text()).resolves.toBe("xxxxxxxx"); + expect(performance.now() - started).toBeGreaterThan(300); + }, + ); + }); + + test("preserves caller cancellation after successful response headers", async () => { + const parent = new AbortController(); + const reason = new Error("Caller cancelled stream"); + await withServer( + (_incoming, response) => response.write("first"), + async (url) => { + const response = await request(testDeps(), { url, signal: parent.signal, stream: true }); + const reading = response.text(); + const rejected = expect(reading).rejects.toBe(reason); + parent.abort(reason); + await rejected; + expect(getEventListeners(parent.signal, "abort")).toHaveLength(0); + }, + ); + }); + + test("cancels the connection when an SSE consumer exits before EOF", async () => { + const parent = new AbortController(); + let responseClosed = false; + await withServer( + (_incoming, response) => { + response.on("close", () => { + responseClosed = true; + }); + response.write("data: [DONE]\n\n"); + }, + async (url) => { + const response = await request(testDeps(), { url, signal: parent.signal, stream: true }); + for await (const event of parseSSE(response)) { + expect(event.data).toBe("[DONE]"); + break; + } + expect(getEventListeners(parent.signal, "abort")).toHaveLength(0); + expect(response.body?.locked).toBe(false); + await expect.poll(() => responseClosed, { timeout: 1000 }).toBe(true); + }, + ); + }); + + test("preserves response metadata when protecting a redirected body", async () => { + await withServer( + (incoming, response) => { + if (incoming.url === "/") { + response.writeHead(302, { location: "/result" }); + response.end(); + } else { + response.writeHead(201, "Created", { "x-request-id": "local-request" }); + response.end("done"); + } + }, + async (url) => { + const response = await request(testDeps(), { url }); + expect(response.url).toBe(`${url}/result`); + expect(response.redirected).toBe(true); + expect(response.status).toBe(201); + expect(response.statusText).toBe("Created"); + expect(response.headers.get("x-request-id")).toBe("local-request"); + await expect(response.text()).resolves.toBe("done"); + }, + ); + }); +}); + +describe("raw response explicit cancellation", () => { + test.each([undefined, new Error("Consumer finished")])( + "aborts the owned request and releases a pending read", + async (reason) => { + const parent = new AbortController(); + const source = new ReadableStream(); + const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(new Response(source)); + const response = await request(testDeps(), { + url: "http://127.0.0.1/unused", + signal: parent.signal, + }); + const reader = response.body!.getReader(); + const pending = reader.read(); + await reader.cancel(reason); + await expect(pending).resolves.toEqual({ value: undefined, done: true }); + reader.releaseLock(); + const signal = fetchMock.mock.calls[0][1]!.signal!; + expect(signal.aborted).toBe(true); + if (reason !== undefined) expect(signal.reason).toBe(reason); + expect(parent.signal.aborted).toBe(false); + expect(getEventListeners(parent.signal, "abort")).toHaveLength(0); + expect(source.locked).toBe(false); + }, + ); +}); diff --git a/packages/runtime/tests/http-bun.e2e.test.ts b/packages/runtime/tests/http-bun.e2e.test.ts new file mode 100644 index 000000000..c0473363e --- /dev/null +++ b/packages/runtime/tests/http-bun.e2e.test.ts @@ -0,0 +1,270 @@ +import { execFile, spawnSync } from "node:child_process"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { tmpdir } from "node:os"; +import { delimiter, dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; +import { afterAll, beforeAll, describe, expect, test } from "vite-plus/test"; + +const execFileAsync = promisify(execFile); +const runtimeRoot = join(dirname(fileURLToPath(import.meta.url)), ".."); +const repositoryRoot = join(runtimeRoot, "../.."); +// Local runs can select the release-pinned Bun without replacing the user's Bun. +const bun = process.env.BAILIAN_TEST_BUN || "bun"; +const hasBun = spawnSync(bun, ["--version"], { stdio: "ignore" }).status === 0; + +describe.skipIf(!hasBun)("core HTTP with native Bun and compiled binaries", () => { + let server: Server; + let proxy: Server; + let serverUrl: string; + let proxyUrl: string; + let probeDirectory: string; + let scriptPath: string; + let binaryPath: string; + const connects: string[] = []; + const closedResponses = new Set(); + const childEnv = { + ...process.env, + PATH: bun.includes("/") ? `${dirname(bun)}${delimiter}${process.env.PATH}` : process.env.PATH, + HTTP_PROXY: "", + HTTPS_PROXY: "", + NO_PROXY: "", + ALL_PROXY: "", + http_proxy: "", + https_proxy: "", + no_proxy: "", + all_proxy: "", + }; + + beforeAll(async () => { + server = createServer((incoming, response) => { + if (incoming.url?.startsWith("/closed?")) { + const path = new URL(incoming.url, "http://localhost").searchParams.get("path")!; + response.end(JSON.stringify(closedResponses.has(path))); + return; + } + if (incoming.url === "/cancel-body" || incoming.url === "/sse") { + const path = incoming.url; + closedResponses.delete(path); + response.on("close", () => closedResponses.add(path)); + response.write(path === "/sse" ? "data: [DONE]\n\n" : "first"); + return; + } + if (incoming.url === "/error") { + response.writeHead(401, { "content-type": "application/json" }); + response.end(JSON.stringify({ code: "InvalidApiKey", message: "local key rejected" })); + return; + } + if (incoming.url === "/multipart") { + let body = ""; + incoming.setEncoding("utf8"); + incoming.on("data", (chunk: string) => { + body += chunk; + }); + incoming.on("end", () => + response.end( + JSON.stringify({ + contentType: incoming.headers["content-type"], + body, + }), + ), + ); + return; + } + if (incoming.url === "/progress" || incoming.url === "/json") { + let chunks = 0; + response.write(incoming.url === "/json" ? '["' : "x"); + const interval = setInterval(() => { + response.write("x"); + if (++chunks === 8) { + clearInterval(interval); + response.end(incoming.url === "/json" ? '"]' : ""); + } + }, 100); + response.on("close", () => clearInterval(interval)); + return; + } + if (incoming.url === "/body") response.write("first"); + const timer = setTimeout(() => response.end("done"), incoming.url === "/body" ? 1_500 : 600); + response.on("close", () => clearTimeout(timer)); + }); + proxy = createServer(); + proxy.on("connect", (incoming, socket) => { + connects.push(incoming.url ?? ""); + // The client can reset a CONNECT socket after the intentional refusal. + socket.on("error", () => {}); + socket.end("HTTP/1.1 502 Bad Gateway\r\n\r\n"); + }); + for (const listener of [server, proxy]) { + await new Promise((resolve, reject) => { + listener.once("error", reject); + listener.listen(0, "127.0.0.1", resolve); + }); + } + serverUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + proxyUrl = `http://127.0.0.1:${(proxy.address() as AddressInfo).port}`; + probeDirectory = mkdtempSync(join(tmpdir(), "bl-native-http-")); + scriptPath = join(probeDirectory, "probe.mts"); + binaryPath = join(probeDirectory, process.platform === "win32" ? "probe.exe" : "probe"); + writeFileSync( + scriptPath, + ` +import { request, requestJson } from ${JSON.stringify(join(runtimeRoot, "../core/src/client/http.ts"))}; +import { parseSSE } from ${JSON.stringify(join(runtimeRoot, "../core/src/client/stream.ts"))}; +import { setupProxyFromEnv } from ${JSON.stringify(join(runtimeRoot, "src/proxy.ts"))}; +setupProxyFromEnv(); +const [url, timeout, mode] = process.argv.slice(2); +const parent = new AbortController(); +async function verifyRemoteClosed() { + // Query while this process is still alive: process exit must not hide a leak. + const checkUrl = new URL("/closed", url); + checkUrl.searchParams.set("path", new URL(url).pathname); + const deadline = performance.now() + 1500; + do { + const check = await fetch(checkUrl, { signal: AbortSignal.timeout(1000) }); + if (await check.json()) { + if (parent.signal.aborted) throw new Error("Body cancellation aborted its caller"); + console.log("closed"); + return; + } + await new Promise(resolve => setTimeout(resolve, 25)); + } while (performance.now() < deadline); + throw new Error("Server still has an active response"); +} +const deps = { identity: { clientName: "native-bun-probe", version: "0" }, + settings: { timeout: Number(timeout), verbose: false } }; +try { + if (mode === "json") { + console.log(JSON.stringify(await requestJson(deps, { url }))); + } else { + const form = new FormData(); + form.set("file", new Blob(["file-content"]), "probe.txt"); + const response = await request(deps, { url, signal: parent.signal, + ...(mode === "multipart" ? { method: "POST", body: form } : {}) }); + if (mode === "cancel") { + const reading = response.text(); + parent.abort(new Error("caller cancelled")); + await reading; + } else if (mode === "cancel-body") { + await response.body.cancel(new Error("Consumer finished")); + await verifyRemoteClosed(); + } else if (mode === "sse") { + for await (const event of parseSSE(response)) { console.log(event.data); break; } + await verifyRemoteClosed(); + } else console.log(await response.text()); + } +} catch (error) { + console.log(error.api?.apiCode ?? (error.exitCode ? "exit:" + error.exitCode : error.name + ":" + error.message)); +} +`, + ); + }); + + afterAll(async () => { + for (const listener of [server, proxy]) { + if (!listener?.listening) continue; + listener.closeAllConnections(); + await new Promise((resolve) => listener.close(() => resolve())); + } + if (probeDirectory) rmSync(probeDirectory, { recursive: true, force: true }); + }); + + describe.each(["bun", "binary"])("%s", (runtime) => { + beforeAll(async () => { + if (runtime !== "binary") return; + const target = `bun-${process.platform === "win32" ? "windows" : process.platform}-${process.arch}`; + await execFileAsync( + bun, + [ + join(repositoryRoot, "tools/release/lib/binary-compile.mjs"), + "--entry", + scriptPath, + "--outfile", + binaryPath, + "--target", + target, + ], + { cwd: repositoryRoot, env: childEnv, timeout: 60_000 }, + ); + if (process.platform === "darwin") { + await execFileAsync( + process.execPath, + [ + "--input-type=module", + "-e", + `import { signDarwinAdhoc } from ${JSON.stringify(join(repositoryRoot, "tools/release/lib/binary-codesign.mjs"))}; signDarwinAdhoc(process.argv[1]);`, + binaryPath, + ], + { timeout: 30_000 }, + ); + } + }); + + async function run(path: string, timeout: number, mode = "body", env = {}) { + const args = [ + path.startsWith("https:") ? path : `${serverUrl}${path}`, + String(timeout), + mode, + ]; + const { stdout } = await execFileAsync( + runtime === "bun" ? bun : binaryPath, + runtime === "bun" ? [scriptPath, ...args] : args, + { cwd: runtimeRoot, env: { ...childEnv, ...env }, timeout: 30_000 }, + ).catch((error) => { + throw new Error( + JSON.stringify({ + stdout: error.stdout, + stderr: error.stderr, + signal: error.signal, + killed: error.killed, + code: error.code, + }), + ); + }); + return stdout.trim(); + } + + test("allows delayed headers within the configured deadline", async () => { + expect(await run("/headers", 2)).toBe("done"); + }); + test("aborts delayed headers at the configured deadline", async () => { + expect(await run("/headers", 0.15)).toMatch(/^AbortError:/); + }); + test("body idle window follows timeout", async () => { + // Leave enough time for headers under parallel suite load; the body stalls longer. + expect(await run("/body", 0.5)).toBe("exit:5"); + expect(await run("/body", 2)).toBe("firstdone"); + }); + test("progressing responses can outlive one idle window", async () => { + expect(await run("/progress", 0.4)).toBe("xxxxxxxxx"); + }); + test("JSON keeps its total deadline while the body progresses", async () => { + expect(await run("/json", 0.4, "json")).toMatch(/^AbortError:/); + }); + test("caller cancellation still works after headers", async () => { + expect(await run("/body", 2, "cancel")).toBe("Error:caller cancelled"); + }); + test("an early SSE exit releases the request", async () => { + expect(await run("/sse", 2, "sse")).toBe("[DONE]\nclosed"); + }); + test("body cancellation closes the request before the process exits", async () => { + expect(await run("/cancel-body", 2, "cancel-body")).toBe("closed"); + }); + test("preserves the service authentication error", async () => { + expect(await run("/error", 2)).toBe("InvalidApiKey"); + }); + test("preserves native multipart uploads", async () => { + const result = JSON.parse(await run("/multipart", 2, "multipart")); + expect(result.contentType).toMatch(/^multipart\/form-data; boundary=/); + expect(result.body).toContain('filename="probe.txt"'); + expect(result.body).toContain("file-content"); + }); + test("native fetch still uses the configured proxy", async () => { + connects.length = 0; + await run("https://bl-native-proxy.invalid/probe", 1, "body", { HTTPS_PROXY: proxyUrl }); + expect(connects).toContain("bl-native-proxy.invalid:443"); + }); + }); +}); diff --git a/packages/runtime/tests/http-timeout.e2e.test.ts b/packages/runtime/tests/http-timeout.e2e.test.ts index dbd3f33f5..8fd6712bd 100644 --- a/packages/runtime/tests/http-timeout.e2e.test.ts +++ b/packages/runtime/tests/http-timeout.e2e.test.ts @@ -81,7 +81,8 @@ async function runProbe( [scriptPath, `${serverUrl}${path}`, String(timeoutMs), dispatcher, mode], { cwd: runtimeRoot, - timeout: 10_000, + // Full-suite process startup can exceed the scaled request deadline. + timeout: 30_000, env: { ...process.env, HTTP_PROXY: "", @@ -112,6 +113,10 @@ test("existing custom dispatchers do not need a compose method", async () => { expect(await runProbe("/headers", 6000, "custom")).toBe("done"); }); -test("body idle timeout remains active after headers arrive", async () => { - expect(await runProbe("/body", 6000)).toBe("UND_ERR_BODY_TIMEOUT"); +test("body idle timeout follows the request setting beyond the transport limit", async () => { + expect(await runProbe("/body", 6000)).toBe("firstdone"); +}); + +test("the upper body idle timeout aborts stalled responses", async () => { + expect(await runProbe("/body", 150)).toBe("BailianError"); });