fix(telemetry): ensure queued events are preserved on dispatcher initialization failure

Author: hopelynd

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

@@ -1117,7 +1117,6 @@ module.exports = (function (e) {
         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;
@@ -1138,6 +1137,7 @@ module.exports = (function (e) {
                 })
               : new Agent({ connect });
         }
+        const payloads = [...this._offlineQueue.splice(0), payload];
         try {
           await Promise.all(
             payloads.map(async (eventPayload) => {

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

@@ -52,6 +52,38 @@ test("an aborted send neither opens a request nor replays queued events", async
   expect(queued._offlineQueue).toEqual(["queued"]);
 });
 
+test.each(["HTTP_PROXY", "HTTPS_PROXY"])(
+  "dispatcher initialization failure preserves queued events (%s)",
+  async (proxyVariable) => {
+    vi.stubEnv("HTTP_PROXY", "");
+    vi.stubEnv("HTTPS_PROXY", "");
+    vi.stubEnv(proxyVariable, "not-a-url");
+    const tracker = new Tracker({ pid: "fixture-pid" });
+    const queued = tracker as unknown as { _offlineQueue: string[] };
+    queued._offlineQueue.push("old event");
+    const controller = new AbortController();
+    try {
+      await expect(tracker.send("new event", { signal: controller.signal })).rejects.toMatchObject({
+        code: "ERR_INVALID_URL",
+      });
+      expect(fetchSpy).not.toHaveBeenCalled();
+      expect(queued._offlineQueue).toEqual(["old event"]);
+
+      vi.stubEnv(proxyVariable, "");
+      await tracker.send("new event", { 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" },
+        ],
+      );
+      expect(queued._offlineQueue).toEqual([]);
+    } finally {
+      vi.unstubAllEnvs();
+    }
+  },
+);
+
 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");

packages/runtime/tests/harness/telemetry-exit-main.ts

@@ -1,5 +1,6 @@
 import dns from "node:dns";
 import { createRequire } from "node:module";
+import { performance } from "node:perf_hooks";
 import { httpTransport } from "../../../core/src/client/transport.ts";
 
 const [mode, origin, dnsServer, tlsOrigin] = process.argv.slice(2);
@@ -36,13 +37,15 @@ if (process.versions.bun) {
 }
 
 const { createCli } = await import("../../src/create-cli.ts");
+let outputCompletedAt = 0;
 await createCli(
   {
     probe: {
       description: "Telemetry exit fixture",
       auth: "none",
       async run(context) {
         context.output.result({ ok: true });
+        outputCompletedAt = performance.now();
       },
     },
   },
@@ -53,7 +56,8 @@ await createCli(
     npmPackage: "telemetry-fixture",
   },
 ).run(["probe", "--output", "json"]);
+const outputToRunMs = performance.now() - outputCompletedAt;
 
 // Inspect closure while this process is still alive: exit must not hide a leak.
 const status = await directFetch(new URL("/status", origin));
-process.stderr.write(JSON.stringify(await status.json()) + "\n");
+process.stderr.write(JSON.stringify({ ...(await status.json()), outputToRunMs }) + "\n");

packages/runtime/tests/telemetry-exit.e2e.test.ts

@@ -6,6 +6,7 @@ import { createServer as createTcpServer, type Socket } from "node:net";
 import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { dirname, join } from "node:path";
+import { performance } from "node:perf_hooks";
 import { fileURLToPath } from "node:url";
 import { promisify } from "node:util";
 import { buildIsolatedCliEnv } from "e2e/isolation";
@@ -128,7 +129,7 @@ describe.each(["node", ...(hasBun ? ["bun"] : [])])(
       });
       try {
         // runNodeMain forces DO_NOT_TRACK=1; this test intentionally exercises telemetry.
-        const { stdout, stderr } = await execFileAsync(
+        const execution = execFileAsync(
           runtime === "node" ? process.execPath : bun,
           [
             ...(runtime === "node" ? ["--import", import.meta.resolve("tsx")] : []),
@@ -140,8 +141,27 @@ describe.each(["node", ...(hasBun ? ["bun"] : [])])(
           ],
           { cwd: runtimeRoot, env: isolated.env, timeout: 8_000, maxBuffer: 1024 * 1024 },
         );
+        let outputBuffer = "";
+        let outputCompletedAt: number | undefined;
+        execution.child.stdout?.on("data", (chunk: string) => {
+          outputBuffer += chunk;
+          if (outputCompletedAt === undefined && outputBuffer.trimEnd().endsWith("}")) {
+            outputCompletedAt = performance.now();
+          }
+        });
+        const { stdout, stderr } = await execution;
+        const exitedAt = performance.now();
         expect(JSON.parse(stdout)).toEqual({ ok: true });
-        expect(JSON.parse(stderr)).toEqual({ requests: disabled ? 0 : 1, closed: !disabled });
+        const { outputToRunMs, ...connectionStatus } = JSON.parse(stderr);
+        expect(connectionStatus).toEqual({ requests: disabled ? 0 : 1, closed: !disabled });
+        if (["fast", "headers", "body"].includes(mode)) {
+          expect(outputToRunMs).toBeGreaterThanOrEqual(0);
+          expect(outputToRunMs, "output completion to run() return").toBeLessThan(3_000);
+          expect(outputCompletedAt).toBeDefined();
+          expect(exitedAt - outputCompletedAt!, "output completion to process exit").toBeLessThan(
+            3_000,
+          );
+        }
         expect(dnsQueries).toBe(0);
         if (!disabled && !mode.startsWith("tls")) {
           expect(requests[0]?.method).toBe("POST");