fix(plugins): stop the TypeScript telemetry losing events and misattributing accounts (#7358)

integrations/agent-plugin-core/typescript/src/telemetry.ts

@@ -1,3 +1,5 @@
+import { randomUUID } from "node:crypto";
+
 import { redactSecrets } from "./lifecycle.ts";
 
 const POSTHOG_API_KEY = "phc_hgJkUVJFYtmaJqrvf6CYN67TIQ8yhXAkWzUn9AMU4yX";
@@ -74,34 +76,108 @@ export function errorKind(error: unknown): string {
   return error instanceof Error ? error.constructor.name : "other";
 }
 
+// Delivery is retried in memory, not spooled to disk, and that is a decision
+// rather than an omission. The Python core spools because its hooks are separate
+// processes that fire per tool call and exit immediately, so nothing survives
+// without a file. These plugins are loaded into a host that lives for a whole
+// session, so re-queueing covers the same transient failures without the claim
+// and lease machinery a correct cross-process spool needs. What that leaves
+// uncovered is narrow: a session that both starts and ends with no connectivity.
+const RETRY_BACKOFF_CEILING_MS = 60_000;
+// Consecutive failed flushes before the queue is dropped. Deliberately NOT the
+// same thing as Python's budget, which rides in the claim filename and so
+// follows one batch: this counter lives in the closure and counts the outage,
+// not the payload. Events captured between attempts join the same queue and go
+// with it. Per-batch accounting would need an attempt count on every event, and
+// the queue is already bounded, so the simpler rule is the one in force here.
+// Without any bound a payload the server will never accept is retried for the
+// whole session and, now that the backlog is preferred over new events, holds
+// the queue against everything behind it.
+const MAX_DELIVERY_ATTEMPTS = 5;
+
 export function createTelemetry(config: TelemetryConfig) {
   let queue: Record<string, unknown>[] = [];
   let timer: ReturnType<typeof setInterval> | undefined;
+  let consecutiveFailures = 0;
+  let retryNotBefore = 0;
+  let exitFlushAttempted = false;
+  let flushing = false;
   const flushThreshold = config.flushThreshold ?? 10;
   const maxQueueSize = config.maxQueueSize ?? 100;
 
   const deliver = config.delivery ?? (async (batch: Record<string, unknown>[]) => {
-    await fetch(POSTHOG_BATCH_URL, {
+    const response = await fetch(POSTHOG_BATCH_URL, {
       method: "POST",
       headers: { "Content-Type": "application/json" },
       body: JSON.stringify({ api_key: POSTHOG_API_KEY, batch }),
       signal: AbortSignal.timeout(3_000),
     });
+    // fetch only rejects on a network-level failure. Without this check a 500,
+    // a 503 or a 429 resolved normally and the batch was counted as delivered
+    // and dropped, which is the likelier outage than a refused connection.
+    // Any non-2xx is retried, matching the Python core: the backoff and the
+    // queue bound contain a payload that will never be accepted, because the
+    // re-queued batch sits at the front and is the first thing evicted.
+    if (!response.ok) throw new Error(`posthog responded ${response.status}`);
   });
 
-  async function flush(): Promise<void> {
+  async function flush(force = false): Promise<void> {
+    // One at a time. Two overlapping flushes each detach the queue and each
+    // prepend their own batch back on failure, so the later batch lands in front
+    // of the earlier one and the truncation then drops the OLDER events first,
+    // inverting the priority the failure path exists to establish. A second
+    // caller returns immediately; the queue waits for the next flush.
+    if (flushing) return;
     if (!queue.length) return;
+    // `force` skips the cooldown. beforeExit is the last chance this process
+    // gets, and gating it on the same backoff meant that after any failure the
+    // exit flush did nothing and the queue died with the process, which is the
+    // loss this whole mechanism exists to prevent.
+    if (!force && Date.now() < retryNotBefore) return;
     const batch = queue;
     queue = [];
+    flushing = true;
     try {
       await deliver(batch);
+      consecutiveFailures = 0;
+      retryNotBefore = 0;
     } catch {
-      // Telemetry must never affect plugin behavior.
+      // Put it back. Detaching the batch and swallowing the error deleted the
+      // events outright, so any blip silently dropped telemetry with nothing
+      // recording that it had happened. Every event carries a uuid, so a retry
+      // that duplicates one PostHog already accepted is collapsed there.
+      //
+      consecutiveFailures += 1;
+      if (consecutiveFailures >= MAX_DELIVERY_ATTEMPTS) {
+        // Give up on the queue so a failing outage cannot hold it for the
+        // session. This drops whatever is queued now, which includes events
+        // captured during the outage, not only the batch that kept failing.
+        consecutiveFailures = 0;
+        retryNotBefore = 0;
+        return;
+      }
+      // Keep the FRONT on overflow, so the batch being retried survives and a
+      // new event is what gets dropped. Matches the Python core, where record()
+      // refuses new events once the spool is full rather than evicting the
+      // backlog. Keeping the newest would throw away exactly the events this
+      // retry exists to save.
+      queue = [...batch, ...queue].slice(0, maxQueueSize);
+      retryNotBefore = Date.now() + Math.min(2 ** consecutiveFailures * 1_000, RETRY_BACKOFF_CEILING_MS);
+    } finally {
+      flushing = false;
     }
   }
 
   function beforeExit(): void {
-    void flush();
+    // Once, and only once. Node re-emits beforeExit whenever the handler
+    // schedules more async work, so an unconditional forced flush looped until
+    // the attempt budget was spent: five attempts against a 3s delivery timeout
+    // is fifteen seconds added to the shutdown of whatever editor or CLI is
+    // hosting this. The backoff used to end that loop after one attempt, and
+    // removing it for the forced path removed the only thing bounding it.
+    if (exitFlushAttempted) return;
+    exitFlushAttempted = true;
+    void flush(true);
   }
 
   function build(event: string, properties: Record<string, unknown> = {}): Record<string, unknown> | null {
@@ -112,6 +188,15 @@ export function createTelemetry(config: TelemetryConfig) {
       return {
         event: config.eventName?.(event) ?? event,
         distinct_id: distinctId,
+        // Stamped once, at capture. This is what makes retrying safe: a batch
+        // re-sent after a failure carries the same ids, so PostHog collapses
+        // anything it already accepted instead of counting it twice.
+        uuid: randomUUID(),
+        // Capture time, not ingestion time. Events now sit through backoff and
+        // across a whole outage, so without this PostHog records them whenever
+        // delivery happened to succeed. It also matters for the uuid dedupe
+        // above, whose key includes the event date.
+        timestamp: new Date().toISOString(),
         properties: {
           ...safeProperties(properties),
           ...safeProperties(config.commonProperties ?? {}),
@@ -134,8 +219,10 @@ export function createTelemetry(config: TelemetryConfig) {
     try {
       const payload = build(event, properties);
       if (!payload) return;
+      // Full means drop this event, not evict the backlog. Same rule as the
+      // failure path above and as Python's record().
+      if (queue.length >= maxQueueSize) return;
       queue.push(payload);
-      if (queue.length > maxQueueSize) queue = queue.slice(-maxQueueSize);
       if (!timer) {
         timer = setInterval(() => void flush(), config.flushIntervalMs ?? 5_000);
         timer.unref?.();
@@ -149,6 +236,8 @@ export function createTelemetry(config: TelemetryConfig) {
 
   function resetForTesting(): void {
     queue = [];
+    consecutiveFailures = 0;
+    retryNotBefore = 0;
     if (timer) clearInterval(timer);
     timer = undefined;
     process.off("beforeExit", beforeExit);

integrations/agent-plugin-core/typescript/tests/telemetry.test.ts

@@ -122,3 +122,250 @@ test("error classification does not expose messages", () => {
   assert.equal(errorKind(new Error("request timeout")), "timeout");
   assert.equal(errorKind(new Error("fetch failed")), "network");
 });
+
+test("a failed delivery keeps the batch instead of deleting it", async () => {
+  // The defect: the queue was detached before the await and the error swallowed,
+  // so one blip destroyed the events with nothing recording that it happened.
+  const attempts: Record<string, unknown>[][] = [];
+  let failNext = true;
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d",
+    flushThreshold: 1000,
+    delivery: async (batch) => {
+      attempts.push(batch);
+      if (failNext) throw new Error("network down");
+    },
+  });
+
+  telemetry.capture("one");
+  telemetry.capture("two");
+  await telemetry.flush();
+
+  assert.equal(attempts.length, 1);
+  assert.equal(telemetry.queueForTesting().length, 2, "events were dropped on failure");
+
+  failNext = false;
+  // Backoff is in force, so wait it out the way wall time would.
+  await new Promise((resolve) => setTimeout(resolve, 2_100));
+  await telemetry.flush();
+
+  assert.equal(attempts.length, 2, "never retried");
+  assert.equal(telemetry.queueForTesting().length, 0);
+  telemetry.resetForTesting();
+});
+
+test("a retried event carries the same uuid so PostHog can collapse it", async () => {
+  const attempts: Record<string, unknown>[][] = [];
+  let failNext = true;
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d",
+    flushThreshold: 1000,
+    delivery: async (batch) => {
+      attempts.push(batch);
+      if (failNext) throw new Error("network down");
+    },
+  });
+
+  telemetry.capture("once");
+  await telemetry.flush();
+  failNext = false;
+  await new Promise((resolve) => setTimeout(resolve, 2_100));
+  await telemetry.flush();
+
+  assert.equal(attempts.length, 2);
+  const first = attempts[0][0].uuid;
+  assert.ok(first, "events carry no uuid, so a retry would double count");
+  assert.equal(attempts[1][0].uuid, first, "retry minted a new uuid");
+  telemetry.resetForTesting();
+});
+
+test("repeated failures back off instead of retrying every flush", async () => {
+  let calls = 0;
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d",
+    flushThreshold: 1000,
+    delivery: async () => { calls += 1; throw new Error("blocked"); },
+  });
+
+  telemetry.capture("one");
+  await telemetry.flush();
+  await telemetry.flush();
+  await telemetry.flush();
+
+  assert.equal(calls, 1, "a blocked host was hammered on every flush");
+  assert.equal(telemetry.queueForTesting().length, 1, "the event was lost while backing off");
+  telemetry.resetForTesting();
+});
+
+test("a full queue drops the new event and keeps the batch being retried", async () => {
+  // Python's record() refuses new events once the spool is full rather than
+  // evicting the backlog. Keeping the newest here would throw away exactly the
+  // events the retry exists to save.
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d",
+    flushThreshold: 1000, maxQueueSize: 3,
+    delivery: async () => { throw new Error("down"); },
+  });
+
+  // Fill past the cap BEFORE the flush, so the re-queue actually has to truncate.
+  // Capturing only two left the queue empty at re-queue time and the slice on the
+  // failure path never ran, which is the half that decides the direction.
+  telemetry.capture("a");
+  telemetry.capture("b");
+  telemetry.capture("c");
+  await telemetry.flush();
+  telemetry.capture("d");
+  telemetry.capture("e");
+
+  const events = telemetry.queueForTesting().map((e) => (e as any).event);
+  assert.equal(events.length, 3, "queue grew past maxQueueSize");
+  assert.deepEqual(events, ["a", "b", "c"], "the retried batch was evicted instead of the new events");
+  telemetry.resetForTesting();
+});
+
+test("the exit-time flush ignores the backoff", async () => {
+  // beforeExit is the last chance the process gets. Gating it on the same
+  // cooldown meant that after any failure it did nothing and the queue died.
+  let attempts = 0;
+  let failing = true;
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d", flushThreshold: 1000,
+    delivery: async () => { attempts += 1; if (failing) throw new Error("down"); },
+  });
+
+  telemetry.capture("a");
+  await telemetry.flush();
+  assert.equal(attempts, 1);
+
+  failing = false;
+  await telemetry.flush();
+  assert.equal(attempts, 1, "the backoff should still hold for an ordinary flush");
+
+  await telemetry.flush(true);
+  assert.equal(attempts, 2, "the exit flush was suppressed by the backoff");
+  assert.equal(telemetry.queueForTesting().length, 0);
+  telemetry.resetForTesting();
+});
+
+test("every event carries a capture-time timestamp", async () => {
+  const sent: Record<string, unknown>[][] = [];
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d", flushThreshold: 1000,
+    delivery: async (batch) => { sent.push(batch); },
+  });
+
+  telemetry.capture("a");
+  const capturedAt = Date.now();
+  await new Promise((resolve) => setTimeout(resolve, 50));
+  await telemetry.flush();
+
+  const stamped = sent[0][0].timestamp as string;
+  assert.ok(stamped, "no timestamp, so PostHog would record delivery time");
+  assert.ok(Math.abs(Date.parse(stamped) - capturedAt) < 1_000, "not capture time");
+  telemetry.resetForTesting();
+});
+
+test("a batch the server will never accept is eventually given up on", async () => {
+  let attempts = 0;
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d", flushThreshold: 1000,
+    delivery: async () => { attempts += 1; throw new Error("permanently bad"); },
+  });
+
+  telemetry.capture("doomed");
+  for (let i = 0; i < 8; i += 1) await telemetry.flush(true);
+
+  assert.ok(attempts <= 6, `retried ${attempts} times with no cap`);
+  assert.equal(telemetry.queueForTesting().length, 0, "a doomed batch held the queue forever");
+  telemetry.resetForTesting();
+});
+
+test("an HTTP error response is a failure, not a delivery", async () => {
+  // fetch only rejects on a network-level failure, so a 500 used to resolve
+  // normally and the batch was dropped as delivered. Exercises the real default
+  // delivery path rather than an injected one, which is where this hid.
+  const realFetch = globalThis.fetch;
+  let calls = 0;
+  globalThis.fetch = (async () => {
+    calls += 1;
+    return new Response("upstream is unwell", { status: 503 });
+  }) as typeof fetch;
+
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d", flushThreshold: 1000,
+  });
+  try {
+    telemetry.capture("during.outage");
+    await telemetry.flush();
+
+    assert.equal(calls, 1, "never reached the network");
+    assert.equal(telemetry.queueForTesting().length, 1, "a 503 was counted as delivered");
+  } finally {
+    globalThis.fetch = realFetch;
+    telemetry.resetForTesting();
+  }
+});
+
+test("a 2xx is a delivery", async () => {
+  const realFetch = globalThis.fetch;
+  globalThis.fetch = (async () => new Response("ok", { status: 200 })) as typeof fetch;
+
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d", flushThreshold: 1000,
+  });
+  try {
+    telemetry.capture("fine");
+    await telemetry.flush();
+    assert.equal(telemetry.queueForTesting().length, 0, "a good response did not clear the queue");
+  } finally {
+    globalThis.fetch = realFetch;
+    telemetry.resetForTesting();
+  }
+});
+
+test("the exit flush is attempted once, not until the budget is spent", async () => {
+  // Node re-emits beforeExit whenever the handler schedules async work, so an
+  // unconditional forced flush looped until MAX_DELIVERY_ATTEMPTS. Against the
+  // real 3s delivery timeout that is fifteen seconds added to a host's shutdown.
+  let attempts = 0;
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d", flushThreshold: 1000,
+    delivery: async () => { attempts += 1; throw new Error("down"); },
+  });
+
+  telemetry.capture("a");
+  const handlers = process.listeners("beforeExit");
+  const ours = handlers[handlers.length - 1] as () => void;
+  ours();
+  ours();
+  ours();
+  await new Promise((resolve) => setTimeout(resolve, 20));
+
+  assert.equal(attempts, 1, `exit flush ran ${attempts} times`);
+  telemetry.resetForTesting();
+});
+
+test("overlapping flushes do not reorder the backlog behind newer events", async () => {
+  // Each flush detaches the queue and prepends its own batch back on failure, so
+  // two in flight at once put the LATER batch in front of the earlier one. The
+  // truncation then drops the older events first, inverting the priority the
+  // failure path exists to establish.
+  let release: (() => void)[] = [];
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d", flushThreshold: 1000,
+    delivery: () => new Promise((_resolve, reject) => { release.push(() => reject(new Error("down"))); }),
+  });
+
+  telemetry.capture("first");
+  const a = telemetry.flush();
+  telemetry.capture("second");
+  const b = telemetry.flush();
+
+  release.forEach((fn) => fn());
+  await Promise.all([a, b]);
+
+  const events = telemetry.queueForTesting().map((e) => (e as any).event);
+  assert.equal(release.length, 1, "a second delivery started while one was in flight");
+  assert.deepEqual(events, ["first", "second"], `backlog reordered: ${events.join(",")}`);
+  telemetry.resetForTesting();
+});