Merge pull request #269 from modelstudioai/fix/http-native-fetch-timeouts

fix(http): align Node and Bun timeouts and response cleanup

Author: XXPermanentXX

.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

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);
   },

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);
   },

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<Uint8Array>({
+        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();
+    }
+  });
+});

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<Response> {
   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,24 +133,127 @@ 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<typeof setTimeout> | 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);
   };
 
   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<Uint8Array>(
+    {
+      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<T>(deps: HttpDeps, opts: RequestOpts): Promise<T> {

packages/core/src/client/stream.ts

@@ -130,6 +130,9 @@ export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEv
       yield { data: event.data, event: event.event, id: event.id };
     }
   } finally {
+    // A consumer may stop at [DONE] before network EOF. Cancel the body so
+    // core HTTP releases its idle timer, abort listener, and connection.
+    await reader.cancel().catch(() => {});
     reader.releaseLock();
   }
 }