Merge pull request #259 from modelstudioai/fix/http-headers-timeout

fix(core): honor configured request timeout for response headers

Author: XXPermanentXX

docs/agents/command-flag-change.md

@@ -53,6 +53,7 @@ pnpm -F bailian-cli exec tsx src/main.ts <command> --new-flag x   # 实测一遍
 
 ## 常见漏点
 
+- 修改请求超时时，检查底层 HTTP 限制：core 的公共 HTTP 请求层仅为已设置超时计时器的请求关闭 `headersTimeout`，由请求 `AbortSignal` 控制响应头等待；保留 `bodyTimeout` 空闲保护，不全局关闭其他请求的超时。仅设置 Agent 默认值会被 Node `fetch` 的逐请求默认值覆盖。流式读取的总时长另由调用方管理，不能据此宣称 `--timeout` 覆盖整个流。
 - ✗ 加了 array 型 flag 但没考虑用户可能传多次
 - ✗ 改默认值忘记更新 description 里的 "(default: xxx)" 文案
 - ✗ 在 `usageArgs` / `exampleArgs` 里写死 `bl <path>`,导致其它产品入口复用时 help 错

packages/core/package.json

@@ -41,6 +41,7 @@
   },
   "dependencies": {
     "tar-stream": "catalog:",
+    "undici": "catalog:",
     "yaml": "^2.8.3",
     "yauzl": "catalog:"
   },

packages/core/src/client/http.ts

@@ -5,6 +5,14 @@ import { ExitCode } from "../errors/codes.ts";
 import { mapApiError } from "../errors/api.ts";
 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 =
+  (dispatch) => (options, handler) =>
+    dispatch({ ...options, headersTimeout: 0 }, handler);
 
 /** 传输层依赖:UA 用 identity,timeout/verbose 用 settings。凭证由调用方(Client)注头。 */
 export interface HttpDeps {
@@ -77,7 +85,7 @@ async function performRequest(
     console.error(`> x-dashscope-source-config: ${sourceConfig(deps.identity)}`);
   }
 
-  const res = await fetch(opts.url, {
+  const requestInit = {
     method: opts.method ?? "GET",
     headers,
     body: opts.body
@@ -86,7 +94,14 @@ 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"],
+  };
+  const res = await fetch(opts.url, requestInit);
 
   if (deps.settings.verbose) {
     console.error(`< ${res.status} ${res.statusText}`);

packages/runtime/tests/http-timeout.e2e.test.ts

@@ -0,0 +1,117 @@
+import { execFile } from "node:child_process";
+import { createServer, type Server } from "node:http";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import type { AddressInfo } from "node:net";
+import { tmpdir } from "node:os";
+import { dirname, join } from "node:path";
+import { fileURLToPath } from "node:url";
+import { promisify } from "node:util";
+import { afterAll, beforeAll, expect, test } from "vite-plus/test";
+import { resolveTsxBin } from "e2e/runner";
+
+const execFileAsync = promisify(execFile);
+const runtimeRoot = join(dirname(fileURLToPath(import.meta.url)), "..");
+let server: Server;
+let serverUrl: string;
+let scriptDir: string;
+let scriptPath: string;
+
+beforeAll(async () => {
+  server = createServer((request, response) => {
+    if (request.url === "/body") {
+      response.write("first");
+    }
+    const timer = setTimeout(() => response.end("done"), 2500);
+    response.on("close", () => clearTimeout(timer));
+  });
+  await new Promise<void>((resolve, reject) => {
+    server.once("error", reject);
+    server.listen(0, "127.0.0.1", resolve);
+  });
+  serverUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
+  scriptDir = mkdtempSync(join(tmpdir(), "bl-http-timeout-"));
+  scriptPath = join(scriptDir, "probe.mts");
+  writeFileSync(
+    scriptPath,
+    `
+import { Agent, setGlobalDispatcher } from ${JSON.stringify(join(runtimeRoot, "node_modules/undici/index.js"))};
+import { setupProxyFromEnv } from ${JSON.stringify(join(runtimeRoot, "src/proxy.ts"))};
+import { request } from ${JSON.stringify(join(runtimeRoot, "../core/src/client/http.ts"))};
+// Scale the hidden transport limit down while keeping real sockets and native fetch.
+const transport = new Agent().compose((dispatch) => (options, handler) =>
+  dispatch({ ...options, headersTimeout: options.headersTimeout === 0 ? 0 : 100,
+    bodyTimeout: options.bodyTimeout === 0 ? 0 : 100 }, handler));
+setGlobalDispatcher(process.argv[4] === "custom" ? {
+  dispatch: transport.dispatch.bind(transport),
+  close: transport.close.bind(transport),
+  destroy: transport.destroy.bind(transport),
+} : transport);
+setupProxyFromEnv();
+try {
+  const response = process.argv[5] === "raw"
+    ? await fetch(process.argv[2])
+    : await request({
+      identity: { clientName: "timeout-test", version: "0.0.0" },
+      settings: { timeout: Number(process.argv[3]) / 1000, verbose: false },
+    }, { url: process.argv[2] });
+  console.log(await response.text());
+} catch (error) {
+  console.log(error.cause?.code ?? error.name);
+} finally {
+  await transport.destroy();
+}
+`,
+  );
+});
+
+afterAll(async () => {
+  server.closeAllConnections();
+  await new Promise<void>((resolve) => server.close(() => resolve()));
+  if (scriptDir) rmSync(scriptDir, { recursive: true, force: true });
+});
+
+async function runProbe(
+  path: string,
+  timeoutMs: number,
+  dispatcher = "default",
+  mode = "request",
+): Promise<string> {
+  const { stdout } = await execFileAsync(
+    resolveTsxBin(),
+    [scriptPath, `${serverUrl}${path}`, String(timeoutMs), dispatcher, mode],
+    {
+      cwd: runtimeRoot,
+      timeout: 10_000,
+      env: {
+        ...process.env,
+        HTTP_PROXY: "",
+        HTTPS_PROXY: "",
+        NO_PROXY: "",
+        http_proxy: "",
+        https_proxy: "",
+        no_proxy: "",
+      },
+    },
+  );
+  return stdout.trim();
+}
+
+test("delayed headers can outlive the transport limit when the request deadline allows it", async () => {
+  expect(await runProbe("/headers", 6000)).toBe("done");
+});
+
+test("the request deadline still aborts delayed headers", async () => {
+  expect(await runProbe("/headers", 100)).toBe("AbortError");
+});
+
+test("unrelated fetch without a deadline keeps its headers timeout", async () => {
+  expect(await runProbe("/headers", 6000, "default", "raw")).toBe("UND_ERR_HEADERS_TIMEOUT");
+});
+
+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");
+});