fix(telemetry): cancel pending sends before CLI exit

Author: XXPermanentXX

docs/agents/telemetry-change.md

@@ -163,6 +163,8 @@ AEM 映射：
 - [ ] 成功与失败均覆盖；遥测异常必须静默且不改变业务退出码
 - [ ] runtime 本地语义错误应记录 `exitCode`；新增字段默认随 AEM `ext` 上报，无需占用新的 `c1`—`c5`
 - [ ] 检查 `DO_NOT_TRACK=1` 与 `telemetry: false` 两个关闭入口
+- [ ] 遥测发送的 DNS/HTTP/响应体必须受可取消预算控制；flush 超时必须取消在途操作，不能只停止等待，也不能修改全局 DNS/dispatcher 或强制退出宿主
+- [ ] 退出生命周期变更保留真实子进程回归：挂起响应头/响应体、DNS 预查询和关闭遥测；检查实际退出及进程仍存活时的连接释放，不以 flush Promise 返回代替验收
 - [ ] 错误字段不得额外拼接 token、请求体、prompt 或本地路径
 
 ## 完成后自查

packages/core/lib/remote-telemetry/tracker.cjs

@@ -887,9 +887,6 @@ module.exports = (function (e) {
       },
     };
     var _ = y,
-      B = n(3),
-      S = n.n(B),
-      T = n(4),
       j = n(5),
       N = n.n(j);
     function w(e) {
@@ -1106,8 +1103,7 @@ module.exports = (function (e) {
           app_version: x,
           device_id: N.a.createHash("md5").update(E()).digest("hex"),
           platform: "node",
-        }),
-      I = Object(T.promisify)(S.a.resolve);
+        });
     function H(e) {
       ((this._offlineQueue = []),
         (e.endpoint = e.endpoint || "gm.mmstat.com"),
@@ -1116,27 +1112,58 @@ module.exports = (function (e) {
     }
     ((H.prototype = (((M = function () {}).prototype = _.prototype), new M())),
       (H.prototype.constructor = H),
-      (H.prototype.send = function (e) {
-        var t,
-          n = this;
-        return ((t = this._config.endpoint), I(t))
-          .then(function (t) {
-            return (
-              n._offlineQueue.forEach(function (e) {
-                n.send(e);
-              }),
-              (n._offlineQueue = []),
-              n._print("send", e),
-              a()(n._config.endpoint_url, {
-                method: "POST",
-                keepalive: true,
-                body: JSON.stringify({ gokey: encodeURIComponent(e), gmkey: "EXP" }),
-              }).catch(function () {})
-            );
-          })
-          .catch(function (t) {
-            (n._offlineQueue.length > 500 && n._offlineQueue.shift(), n._offlineQueue.push(e));
-          });
+      (H.prototype.send = async function (payload, options = {}) {
+        const signal = options.signal;
+        if (signal?.aborted) return;
+        // DNS follows the cancellable fetch; a separate preflight could outlive
+        // the CLI's flush budget and also bypass an explicitly configured proxy.
+        const payloads = [...this._offlineQueue.splice(0), payload];
+        // Undici request cancellation alone does not own an unfinished connector.
+        // Only telemetry gets this private dispatcher, including its proxy hops.
+        let dispatcher;
+        if (!process.versions.bun) {
+          const { Agent, EnvHttpProxyAgent } = require("undici");
+          const connect = { signal, timeout: 1000 };
+          const httpProxy = process.env.HTTP_PROXY?.trim() || undefined;
+          const httpsProxy = process.env.HTTPS_PROXY?.trim() || undefined;
+          dispatcher =
+            httpProxy || httpsProxy
+              ? new EnvHttpProxyAgent({
+                  httpProxy,
+                  httpsProxy,
+                  noProxy: process.env.NO_PROXY?.trim() || undefined,
+                  connect,
+                  requestTls: connect,
+                  proxyTls: connect,
+                })
+              : new Agent({ connect });
+        }
+        try {
+          await Promise.all(
+            payloads.map(async (eventPayload) => {
+              if (signal?.aborted) return;
+              try {
+                this._print("send", eventPayload);
+                const response = await a()(this._config.endpoint_url, {
+                  method: "POST",
+                  keepalive: true,
+                  signal,
+                  ...(dispatcher ? { dispatcher } : {}),
+                  body: JSON.stringify({ gokey: encodeURIComponent(eventPayload), gmkey: "EXP" }),
+                });
+                // Telemetry never uses response data. Release streaming bodies too.
+                await response.body?.cancel().catch(() => {});
+              } catch {
+                if (!signal?.aborted) {
+                  if (this._offlineQueue.length > 500) this._offlineQueue.shift();
+                  this._offlineQueue.push(eventPayload);
+                }
+              }
+            }),
+          );
+        } finally {
+          await dispatcher?.destroy().catch(() => {});
+        }
       }));
     t.default = H;
   },

packages/core/lib/remote-telemetry/tracker.d.cts

@@ -1,7 +1,7 @@
 declare class RemoteTracker {
   constructor(config: { pid: string; debug?: boolean; [key: string]: unknown });
   use(plugin: unknown): (...args: any[]) => void;
-  send(gokey: string): Promise<void>;
+  send(gokey: string, options?: { signal?: AbortSignal }): Promise<void>;
 }
 
 export = RemoteTracker;

packages/core/src/telemetry/sink.ts

@@ -17,7 +17,12 @@ let remoteSendEvent: ((name: string, opts: Record<string, unknown>) => void) | u
 // 如果不追踪，CLI 在短命令或 SIGINT 场景下会在请求真正落网络前就退出。
 // 配合 tracker.js 中的 `keepalive: true`，await 这些 promise 可以在不阻塞
 // 主流程的前提下，尽最大努力把埋点送达。
-const inflightSends = new Set<Promise<unknown>>();
+const SEND_TIMEOUT_MS = 1000;
+interface PendingSend {
+  controller: AbortController;
+  promise: Promise<void>;
+}
+const inflightSends = new Set<PendingSend>();
 let remoteClient: any = undefined;
 
 try {
@@ -28,13 +33,24 @@ try {
   // 不修改 prototype，避免污染其他可能存在的 Tracker 使用方。
   const originalSend = client.send.bind(client);
   client.send = function (payload: string) {
-    const result = originalSend(payload);
-    if (result && typeof (result as Promise<unknown>).then === "function") {
-      const p = result as Promise<unknown>;
-      inflightSends.add(p);
-      void p.finally(() => inflightSends.delete(p));
-    }
-    return result;
+    const controller = new AbortController();
+    const deadline = setTimeout(() => controller.abort(), SEND_TIMEOUT_MS);
+    deadline.unref?.();
+    const pending: PendingSend = {
+      controller,
+      promise: Promise.resolve()
+        .then(() => originalSend(payload, { signal: controller.signal }))
+        .catch(() => {})
+        .finally(() => {
+          // Bun may resolve body.cancel() before stopping its underlying fetch.
+          // Finish this send's private scope before removing its deadline.
+          controller.abort();
+          clearTimeout(deadline);
+          inflightSends.delete(pending);
+        }),
+    };
+    inflightSends.add(pending);
+    return pending.promise;
   };
   remoteClient = client;
   remoteSendEvent = client.use(EventPlugin);
@@ -47,12 +63,13 @@ try {
  *
  * 1. 先调用 `_sendAll` 排空 tracker 内部的去抖队列，把还卡在 500ms 合并窗口里
  *    的事件立刻推上网络。
- * 2. 然后用硬超时 race 所有已追踪的 fetch promise。
+ * 2. 在预算内等待发送，预算耗尽时真正取消 DNS/HTTP/body 工作。
  *
  * 埋点永远不应阻塞 CLI：调用方应传入较短的超时（例如 1000ms），并始终与超时
  * race。错误与超时一律静默吞掉。
  */
 export async function flushTelemetry(timeoutMs = 1000): Promise<void> {
+  let deadline: ReturnType<typeof setTimeout> | undefined;
   try {
     if (remoteClient) {
       try {
@@ -66,13 +83,20 @@ export async function flushTelemetry(timeoutMs = 1000): Promise<void> {
       }
     }
     if (inflightSends.size === 0) return;
-    const pending = [...inflightSends].map((p) => p.catch(() => undefined));
     await Promise.race([
-      Promise.allSettled(pending),
-      new Promise<void>((resolve) => setTimeout(resolve, timeoutMs).unref?.()),
+      Promise.allSettled([...inflightSends].map((pending) => pending.promise)),
+      new Promise<void>((resolve) => {
+        deadline = setTimeout(() => {
+          for (const pending of inflightSends) pending.controller.abort();
+          resolve();
+        }, timeoutMs);
+        deadline.unref?.();
+      }),
     ]);
   } catch {
     // 埋点逻辑任何异常都不能影响 CLI 主流程
+  } finally {
+    clearTimeout(deadline);
   }
 }
 

packages/core/tests/telemetry-send.test.ts

@@ -0,0 +1,119 @@
+import { createRequire } from "node:module";
+import dns from "node:dns";
+import { afterEach, beforeEach, expect, test, vi, type MockInstance } from "vite-plus/test";
+import Tracker from "../lib/remote-telemetry/tracker.cjs";
+
+const require = createRequire(import.meta.url);
+const transport = require("undici") as typeof import("undici");
+let fetchSpy: MockInstance<typeof transport.fetch>;
+let resolveSpy: MockInstance<typeof dns.resolve>;
+
+beforeEach(() => {
+  fetchSpy = vi
+    .spyOn(transport, "fetch")
+    .mockResolvedValue(new Response(null, { status: 204 }) as never);
+  resolveSpy = vi.spyOn(dns, "resolve").mockImplementation((...args: unknown[]) => {
+    (args.at(-1) as (error: null, addresses: string[]) => void)(null, ["127.0.0.1"]);
+  });
+});
+afterEach(() => vi.restoreAllMocks());
+
+test("telemetry uses the bounded transport without a separate DNS preflight", async () => {
+  const controller = new AbortController();
+  const tracker = new Tracker({ pid: "fixture-pid", env: "dev" });
+  await tracker.send("fixture event", { signal: controller.signal });
+  expect(resolveSpy).not.toHaveBeenCalled();
+  expect(fetchSpy).toHaveBeenCalledOnce();
+  expect(fetchSpy.mock.calls[0]).toEqual([
+    "https://gm.mmstat.com/aes.1.1",
+    expect.objectContaining({
+      method: "POST",
+      signal: controller.signal,
+      body: JSON.stringify({ gokey: "fixture%20event", gmkey: "EXP" }),
+    }),
+  ]);
+});
+
+test("telemetry releases the unused response body before completing", async () => {
+  const cancel = vi.fn();
+  fetchSpy.mockResolvedValue(new Response(new ReadableStream({ cancel })) as never);
+  await new Tracker({ pid: "fixture-pid" }).send("event", { signal: new AbortController().signal });
+  expect(cancel).toHaveBeenCalledOnce();
+});
+
+test("an aborted send neither opens a request nor replays queued events", async () => {
+  const tracker = new Tracker({ pid: "fixture-pid" });
+  const queued = tracker as unknown as { _offlineQueue: string[] };
+  queued._offlineQueue.push("queued");
+  const controller = new AbortController();
+  controller.abort();
+  await tracker.send("event", { signal: controller.signal });
+  expect(fetchSpy).not.toHaveBeenCalled();
+  expect(queued._offlineQueue).toEqual(["queued"]);
+});
+
+test("queued retries share the current cancellation signal and retain event encoding", async () => {
+  const tracker = new Tracker({ pid: "fixture-pid" });
+  (tracker as unknown as { _offlineQueue: string[] })._offlineQueue.push("old event");
+  const controller = new AbortController();
+  await tracker.send("new event", { signal: controller.signal });
+  expect(fetchSpy).toHaveBeenCalledTimes(2);
+  expect(fetchSpy.mock.calls.map(([, options]) => options?.signal)).toEqual([
+    controller.signal,
+    controller.signal,
+  ]);
+  expect(fetchSpy.mock.calls.map(([, options]) => JSON.parse(options?.body as string))).toEqual([
+    { gokey: "old%20event", gmkey: "EXP" },
+    { gokey: "new%20event", gmkey: "EXP" },
+  ]);
+});
+
+test("cancellation failures are silent and are not queued for retry", async () => {
+  const tracker = new Tracker({ pid: "fixture-pid" });
+  const controller = new AbortController();
+  fetchSpy.mockImplementation(async () => {
+    controller.abort();
+    throw new DOMException("cancelled", "AbortError");
+  });
+  await expect(tracker.send("event", { signal: controller.signal })).resolves.toBeUndefined();
+  expect((tracker as unknown as { _offlineQueue: string[] })._offlineQueue).toEqual([]);
+});
+
+test("response cleanup failures do not replay an already acknowledged event", async () => {
+  const tracker = new Tracker({ pid: "fixture-pid" });
+  fetchSpy.mockResolvedValue(
+    new Response(
+      new ReadableStream({
+        cancel() {
+          return Promise.reject(new Error("cleanup failed"));
+        },
+      }),
+    ) as never,
+  );
+  await expect(
+    tracker.send("event", { signal: new AbortController().signal }),
+  ).resolves.toBeUndefined();
+  expect((tracker as unknown as { _offlineQueue: string[] })._offlineQueue).toEqual([]);
+});
+
+test.each([false, true])(
+  "Node telemetry owns its dispatcher without replacing the global one (proxy=%s)",
+  async (proxy) => {
+    vi.stubEnv("HTTP_PROXY", proxy ? "http://127.0.0.1:9" : "");
+    vi.stubEnv("HTTPS_PROXY", proxy ? "http://127.0.0.1:9" : "");
+    const globalDispatcher = transport.getGlobalDispatcher();
+    const destroy = vi.spyOn(globalDispatcher, "destroy");
+    try {
+      await new Tracker({ pid: "fixture-pid" }).send("event", {
+        signal: new AbortController().signal,
+      });
+      const dispatcher = fetchSpy.mock.calls[0]?.[1]?.dispatcher;
+      expect(dispatcher).toBeInstanceOf(proxy ? transport.EnvHttpProxyAgent : transport.Agent);
+      expect(dispatcher).not.toBe(globalDispatcher);
+      expect(transport.getGlobalDispatcher()).toBe(globalDispatcher);
+      expect(destroy).not.toHaveBeenCalled();
+    } finally {
+      vi.unstubAllEnvs();
+    }
+  },
+);

packages/core/tests/telemetry-sink.test.ts

@@ -0,0 +1,68 @@
+import { afterEach, expect, test, vi } from "vite-plus/test";
+import type { TrackingEvent } from "../src/telemetry/event.ts";
+
+const fixture = vi.hoisted(() => ({
+  signals: [] as Array<AbortSignal | undefined>,
+  complete: false,
+}));
+vi.mock("../lib/remote-telemetry/tracker.cjs", () => ({
+  default: class {
+    send(_payload: string, options: { signal?: AbortSignal } = {}) {
+      fixture.signals.push(options.signal);
+      if (fixture.complete) return Promise.resolve();
+      return new Promise<void>((resolve) =>
+        options.signal?.addEventListener("abort", () => resolve(), { once: true }),
+      );
+    }
+    use() {
+      return () => {
+        void this.send("event");
+      };
+    }
+    _sendAll() {}
+  },
+}));
+
+afterEach(() => {
+  vi.clearAllTimers();
+  vi.useRealTimers();
+  vi.resetModules();
+  fixture.signals = [];
+  fixture.complete = false;
+});
+
+test("flush cancels in-flight work when its shorter shutdown budget expires", async () => {
+  vi.useFakeTimers();
+  const { remoteSink, flushTelemetry } = await import("../src/telemetry/sink.ts");
+  await remoteSink({ command: "fixture" } as TrackingEvent);
+  const pending = flushTelemetry(50);
+  await vi.advanceTimersByTimeAsync(49);
+  expect(fixture.signals).toHaveLength(1);
+  expect(fixture.signals[0]?.aborted).toBe(false);
+  await vi.advanceTimersByTimeAsync(1);
+  await pending;
+  expect(fixture.signals[0]?.aborted).toBe(true);
+  expect(vi.getTimerCount()).toBe(0);
+});
+
+test("send has a bounded lifetime even when a host does not flush", async () => {
+  vi.useFakeTimers();
+  const { remoteSink } = await import("../src/telemetry/sink.ts");
+  await remoteSink({ command: "fixture" } as TrackingEvent);
+  await vi.advanceTimersByTimeAsync(999);
+  expect(fixture.signals[0]?.aborted).toBe(false);
+  await vi.advanceTimersByTimeAsync(1);
+  expect(fixture.signals[0]?.aborted).toBe(true);
+  expect(vi.getTimerCount()).toBe(0);
+});
+
+test("successful sending and repeated flushes leave no deadlines behind", async () => {
+  vi.useFakeTimers();
+  fixture.complete = true;
+  const { remoteSink, flushTelemetry } = await import("../src/telemetry/sink.ts");
+  await remoteSink({ command: "fixture" } as TrackingEvent);
+  await flushTelemetry(50);
+  await flushTelemetry(50);
+  expect(fixture.signals[0]?.aborted).toBe(true);
+  expect(vi.getTimerCount()).toBe(0);
+});