From ccf5f20b3beea2b245c398a89cb686077b546de2 Mon Sep 17 00:00:00 2001 From: kiritowoo Date: Sun, 12 Jul 2026 21:05:20 +0800 Subject: [PATCH] fix(cli): stop losing render telemetry to the exit race (#2105) --- packages/cli/src/telemetry/client.test.ts | 134 ++++++++++++++++++++++ packages/cli/src/telemetry/client.ts | 58 +++++++--- packages/cli/src/telemetry/events.test.ts | 10 ++ packages/cli/src/telemetry/events.ts | 13 ++- 4 files changed, 200 insertions(+), 15 deletions(-) create mode 100644 packages/cli/src/telemetry/client.test.ts diff --git a/packages/cli/src/telemetry/client.test.ts b/packages/cli/src/telemetry/client.test.ts new file mode 100644 index 000000000..63d26b6fb --- /dev/null +++ b/packages/cli/src/telemetry/client.test.ts @@ -0,0 +1,134 @@ +import { describe, expect, it, vi, beforeEach } from "vitest"; + +// CI exports HYPERFRAMES_NO_TELEMETRY=1 (and users may set DO_NOT_TRACK), which +// makes shouldTrack() short-circuit — and it caches that decision for the module's +// lifetime. Clear both BEFORE the module under test is imported / first tracks. +vi.stubEnv("HYPERFRAMES_NO_TELEMETRY", ""); +vi.stubEnv("DO_NOT_TRACK", ""); + +// Pin config so the queue never touches disk and telemetry is enabled. +vi.mock("./config.js", () => ({ + readConfig: () => ({ anonymousId: "anon-test-123", telemetryEnabled: true }), + writeConfig: () => {}, +})); + +// shouldTrack() short-circuits in dev mode — force production behavior. +vi.mock("../utils/env.js", () => ({ + isDevMode: () => false, +})); + +// Intercept the exit-time child process so flushSync delivery is assertable. +const spawnMock = vi.fn(() => ({ unref: vi.fn() })); +vi.mock("node:child_process", () => ({ + spawn: (...args: unknown[]) => spawnMock(...(args as [])), +})); + +const { trackEvent, flush, flushSync } = await import("./client.js"); + +type Batch = { uuid: string; event: string }[]; + +function sentBatch(fetchMock: ReturnType, call = 0): Batch { + const init = fetchMock.mock.calls[call]?.[1] as { body: string } | undefined; + if (!init) throw new Error(`expected fetch call #${call} to have been made`); + return JSON.parse(init.body).batch; +} + +describe("telemetry queue delivery", () => { + beforeEach(async () => { + vi.unstubAllGlobals(); + // Drain anything a previous test left behind. + vi.stubGlobal( + "fetch", + vi.fn(() => Promise.resolve(new Response(""))), + ); + await flush(); + vi.unstubAllGlobals(); + }); + + it("forgets events only after the request completes, and stamps each with a uuid", async () => { + const fetchMock = vi.fn(() => Promise.resolve(new Response(""))); + vi.stubGlobal("fetch", fetchMock); + + trackEvent("render_complete", { quality: "draft" }); + await flush(); + + expect(fetchMock).toHaveBeenCalledTimes(1); + const batch = sentBatch(fetchMock); + expect(batch).toHaveLength(1); + expect(batch[0]?.event).toBe("render_complete"); + expect(batch[0]?.uuid).toMatch(/^[0-9a-f-]{36}$/); + + // Delivered — a second flush has nothing to send. + await flush(); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("keeps events queued when the request fails, and re-sends them with the SAME uuid", async () => { + const failing = vi.fn(() => Promise.reject(new Error("network down"))); + vi.stubGlobal("fetch", failing); + + trackEvent("render_complete", { quality: "draft" }); + await flush(); + expect(failing).toHaveBeenCalledTimes(1); + + // Queue survived the failed send — the retry carries the same event uuid, + // so PostHog would dedupe even if the first request had actually landed. + const succeeding = vi.fn(() => Promise.resolve(new Response(""))); + vi.stubGlobal("fetch", succeeding); + await flush(); + + expect(succeeding).toHaveBeenCalledTimes(1); + const first = sentBatch(failing); + const retry = sentBatch(succeeding); + expect(retry).toHaveLength(1); + expect(retry[0]?.uuid).toBe(first[0]?.uuid); + + await flush(); + expect(succeeding).toHaveBeenCalledTimes(1); + }); + + it("does not drop events queued while a flush is in flight", async () => { + let resolveFetch: (r: Response) => void = () => {}; + const gated = vi.fn(() => new Promise((res) => (resolveFetch = res))); + vi.stubGlobal("fetch", gated); + + trackEvent("render_complete", { quality: "draft" }); + const inFlight = flush(); + trackEvent("cli_command_result", { command: "render" }); + resolveFetch(new Response("")); + await inFlight; + + // Only the snapshot was forgotten; the late event is still queued. + const succeeding = vi.fn(() => Promise.resolve(new Response(""))); + vi.stubGlobal("fetch", succeeding); + await flush(); + const batch = sentBatch(succeeding); + expect(batch).toHaveLength(1); + expect(batch[0]?.event).toBe("cli_command_result"); + }); + + it("flushSync hands the queue to a detached child that carries the payload", async () => { + spawnMock.mockClear(); + trackEvent("render_complete", { quality: "draft" }); + flushSync(); + + // The child was spawned detached with the batch inlined into its -e script. + expect(spawnMock).toHaveBeenCalledTimes(1); + const [execPath, args, opts] = spawnMock.mock.calls[0] as unknown as [ + string, + string[], + { detached: boolean }, + ]; + expect(execPath).toBe(process.execPath); + expect(args[0]).toBe("-e"); + expect(args[1]).toContain("render_complete"); + expect(args[1]).toMatch(/[0-9a-f-]{36}/); // event uuid rides along + expect(opts.detached).toBe(true); + + // Queue handed to the child — nothing left for a regular flush. + const fetchMock = vi.fn(() => Promise.resolve(new Response(""))); + vi.stubGlobal("fetch", fetchMock); + await flush(); + expect(fetchMock).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/cli/src/telemetry/client.ts b/packages/cli/src/telemetry/client.ts index 3ca96a437..afd059ffc 100644 --- a/packages/cli/src/telemetry/client.ts +++ b/packages/cli/src/telemetry/client.ts @@ -1,3 +1,5 @@ +import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; import { readConfig, writeConfig } from "./config.js"; import { VERSION } from "../version.js"; import { c } from "../ui/colors.js"; @@ -20,7 +22,11 @@ interface EventProperties { [key: string]: string | number | boolean | null | undefined; } -let eventQueue: Array<{ +interface QueuedEvent { + // Client-generated event id. PostHog dedupes on it, so an event that gets + // sent by an interrupted flush() AND re-sent by the exit-time flushSync() + // fallback still counts once. + uuid: string; event: string; properties: EventProperties; timestamp: string; @@ -28,7 +34,9 @@ let eventQueue: Array<{ // Used to attribute server-side studio renders to the browser user who // triggered them, so the render funnel is joinable across processes. distinctId?: string; -}> = []; +} + +let eventQueue: QueuedEvent[] = []; let telemetryEnabled: boolean | null = null; @@ -72,6 +80,7 @@ export function trackEvent( const sys = getSystemMeta(); eventQueue.push({ + uuid: randomUUID(), event, distinctId, properties: { @@ -102,32 +111,48 @@ export function trackEvent( } /** - * Drain the in-memory queue into a PostHog `/batch/` payload string. - * Returns null when there's nothing to send. Resets the queue as a side effect - * so callers can fire-and-forget the resulting payload. + * Serialize events into a PostHog `/batch/` payload string. Pure — the queue + * is untouched, so callers decide when events count as delivered. + * + * Each event carries its client-generated `uuid`, which PostHog treats as the + * event id — re-sending the same event is idempotent, not a duplicate. * * $ip:null tells PostHog not to record the request IP for any of these events. * Server-side "Discard client IP data" is also enabled in project settings. */ -function drainQueueToPayload(): string | null { - if (eventQueue.length === 0) return null; +function buildPayload(events: readonly QueuedEvent[]): string | null { + if (events.length === 0) return null; const config = readConfig(); - const batch = eventQueue.map((e) => ({ + const batch = events.map((e) => ({ + uuid: e.uuid, event: e.event, properties: { ...e.properties, $ip: null }, distinct_id: e.distinctId ?? config.anonymousId, timestamp: e.timestamp, })); - eventQueue = []; return JSON.stringify({ api_key: POSTHOG_API_KEY, batch }); } /** * Flush all queued events to PostHog via async HTTP POST. - * Called before normal process exit via `beforeExit`. + * Call sites: the `beforeExit` hook in cli.ts (normal exit), eager sends right + * after high-value events (trackRenderComplete / trackRenderError), and the + * `events` beacon command, which awaits delivery before its process exits. + * + * Events are only removed from the queue once the request has completed. + * The old drain-first version silently lost the whole batch whenever the + * process died with the fetch in flight — which is the NORMAL exit path for + * `render`: an agent pipe closing triggers the EPIPE `process.exit(0)`, and + * error paths call `process.exit(1)` directly, both killing the in-flight + * request that `beforeExit` had just started. Keeping the queue intact until + * delivery lets the exit-time flushSync() child (which survives the parent) + * re-send anything unconfirmed; event uuids make that re-send idempotent. */ export async function flush(): Promise { - const payload = drainQueueToPayload(); + // Copy, not alias — events queued while the request is in flight must not + // be swept into the "delivered" set below. + const snapshot = eventQueue.slice(); + const payload = buildPayload(snapshot); if (payload == null) return; const controller = new AbortController(); @@ -140,8 +165,13 @@ export async function flush(): Promise { body: payload, signal: controller.signal, }); + // Delivered — forget exactly what was sent (events queued while the + // request was in flight stay for the next flush). + const sent = new Set(snapshot); + eventQueue = eventQueue.filter((e) => !sent.has(e)); } catch { - // Silently ignore — telemetry must never break the CLI + // Silently ignore — telemetry must never break the CLI. The events stay + // queued so the exit-time flushSync() fallback can still deliver them. } finally { clearTimeout(timeout); } @@ -153,11 +183,11 @@ export async function flush(): Promise { * so the parent process exits immediately without waiting. */ export function flushSync(): void { - const payload = drainQueueToPayload(); + const payload = buildPayload(eventQueue); if (payload == null) return; + eventQueue = []; try { - const { spawn } = require("node:child_process") as typeof import("node:child_process"); const child = spawn( process.execPath, [ diff --git a/packages/cli/src/telemetry/events.test.ts b/packages/cli/src/telemetry/events.test.ts index 9df3498cf..a9cbd1b5a 100644 --- a/packages/cli/src/telemetry/events.test.ts +++ b/packages/cli/src/telemetry/events.test.ts @@ -1,8 +1,10 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; const trackEvent = vi.fn(); +const flush = vi.fn(() => Promise.resolve()); vi.mock("./client.js", () => ({ trackEvent: (...args: unknown[]) => trackEvent(...args), + flush: () => flush(), })); // identifyUser reads the install anonymousId; pin it so the $identify alias is @@ -174,6 +176,14 @@ describe("trackCheckReport", () => { describe("render telemetry events", () => { beforeEach(() => { trackEvent.mockClear(); + flush.mockClear(); + }); + + it("flushes immediately after render_complete and render_error (exit races the lazy flush)", () => { + trackRenderComplete({ durationMs: 1000, fps: 30, quality: "draft", docker: false, gpu: false }); + expect(flush).toHaveBeenCalledTimes(1); + trackRenderError({ fps: 30, quality: "draft", docker: false }); + expect(flush).toHaveBeenCalledTimes(2); }); it("redacts paths and URL query strings from render error messages", () => { diff --git a/packages/cli/src/telemetry/events.ts b/packages/cli/src/telemetry/events.ts index 88196e247..23ea37327 100644 --- a/packages/cli/src/telemetry/events.ts +++ b/packages/cli/src/telemetry/events.ts @@ -1,6 +1,6 @@ import { redactTelemetryString, type OutputResolutionIssueKind } from "@hyperframes/core"; import type { SubTimelineWaitOutcome } from "@hyperframes/engine"; -import { trackEvent } from "./client.js"; +import { flush, trackEvent } from "./client.js"; import { readConfig } from "./config.js"; // run_id is attached only when the orchestrator set HYPERFRAMES_RUN_ID — an @@ -298,6 +298,14 @@ export function trackRenderComplete( }, props.distinctId, ); + // Send immediately instead of waiting for the exit-time flush. The render + // command's normal teardown (agent-pipe EPIPE → process.exit(0), or an + // explicit process.exit) kills the lazy beforeExit flush mid-flight, which + // is why only ~10-15% of successful renders ever produced a render_complete + // — and the survivors skewed toward users with low RTT to PostHog. The + // process is alive and idle here; if it still dies mid-request, the queue + // keeps the event for the exit-time flushSync() fallback. + void flush(); } export function trackRenderError( @@ -339,6 +347,9 @@ export function trackRenderError( }, props.distinctId, ); + // Same rationale as trackRenderComplete: error paths process.exit(1) before + // the lazy flush can win its race — send now, exit-time fallback covers the rest. + void flush(); } export function trackRenderObservation(props: {