fix(telemetry): drop permanently-rejected outbox batches instead of retrying forever

.changeset/drop-poisoned-telemetry-batches.md

@@ -0,0 +1,5 @@
+---
+"@evlog/telemetry": patch
+---
+
+Fix the outbox getting permanently stuck when the ingest endpoint rejects a batch with a 4xx response (oversized batch, unknown tool, schema drift on older buffered events, etc.). Previously, any non-2xx response kept the batch buffered forever, and since the whole outbox is resent together on every run, one bad batch silently blocked all future telemetry for the tool. Batches rejected with 400/401/403/404 (anything but 429, which is treated as transient) are now dropped instead of retried indefinitely.

packages/telemetry/src/drain.ts

@@ -2,7 +2,10 @@ import type { RunEvent } from './types'
 
 /**
  * Drain function that attempts delivery for buffered run events.
- * @returns `true` when at least one downstream sink reports successful delivery.
+ * @returns `true` when the batch should be removed from the outbox — either
+ * because it was delivered, or because the server permanently rejected it
+ * and retrying would never succeed. `false` keeps the batch buffered for a
+ * later, transient-failure retry (network error, timeout, 5xx, 429).
  */
 export type TelemetryDrain = (events: RunEvent[]) => Promise<boolean>
 
@@ -27,7 +30,7 @@ export function createDebugDrain(): TelemetryDrain {
   }
 }
 
-/** HTTP ingestion drain with timeout. Returns true when delivery succeeds. */
+/** HTTP ingestion drain with timeout. Returns true when the outbox should drop the batch. */
 export function createHttpDrain(endpoint: string, timeoutMs = 400): TelemetryDrain {
   return async (events) => {
     if (events.length === 0) return true
@@ -40,7 +43,16 @@ export function createHttpDrain(endpoint: string, timeoutMs = 400): TelemetryDra
         body: JSON.stringify({ events }),
         signal: controller.signal,
       })
-      return res.ok
+      if (res.ok) return true
+      // A 4xx (excluding 429, which is transient rate limiting) means the
+      // server permanently rejected this exact payload — an oversized batch,
+      // a schema it doesn't recognise, etc. Retrying identical bytes will
+      // never succeed, and leaving them in the outbox would silently poison
+      // it forever: every future run re-sends the same broken batch first,
+      // blocking all telemetry for this tool indefinitely. Drop it instead —
+      // losing one batch beats losing all of them.
+      if (res.status >= 400 && res.status < 500 && res.status !== 429) return true
+      return false
     } catch {
       return false
     } finally {

packages/telemetry/test/drain.test.ts

@@ -0,0 +1,54 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { createHttpDrain } from '../src/drain'
+import { exampleRunEvent } from '../src/disclosure'
+
+function mockFetch(status: number): void {
+  vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(null, { status })))
+}
+
+describe('createHttpDrain', () => {
+  afterEach(() => {
+    vi.unstubAllGlobals()
+  })
+
+  it('reports delivered (drop from outbox) on a 2xx response', async () => {
+    mockFetch(204)
+    const drain = createHttpDrain('http://example.test/ingest')
+    await expect(drain([exampleRunEvent()])).resolves.toBe(true)
+  })
+
+  it('drops permanently-rejected batches on 400 instead of retrying forever', async () => {
+    // The exact failure mode this guards: a stale/oversized outbox batch
+    // that will never become valid by resending the same bytes — without
+    // this, it would block all future telemetry for the tool indefinitely.
+    mockFetch(400)
+    const drain = createHttpDrain('http://example.test/ingest')
+    await expect(drain([exampleRunEvent()])).resolves.toBe(true)
+  })
+
+  it('keeps retrying on 429 (rate limited)', async () => {
+    mockFetch(429)
+    const drain = createHttpDrain('http://example.test/ingest')
+    await expect(drain([exampleRunEvent()])).resolves.toBe(false)
+  })
+
+  it('keeps retrying on a 5xx server error', async () => {
+    mockFetch(500)
+    const drain = createHttpDrain('http://example.test/ingest')
+    await expect(drain([exampleRunEvent()])).resolves.toBe(false)
+  })
+
+  it('keeps retrying on a network failure', async () => {
+    vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new Error('network down')))
+    const drain = createHttpDrain('http://example.test/ingest')
+    await expect(drain([exampleRunEvent()])).resolves.toBe(false)
+  })
+
+  it('short-circuits on an empty batch without calling fetch', async () => {
+    const fetchSpy = vi.fn()
+    vi.stubGlobal('fetch', fetchSpy)
+    const drain = createHttpDrain('http://example.test/ingest')
+    await expect(drain([])).resolves.toBe(true)
+    expect(fetchSpy).not.toHaveBeenCalled()
+  })
+})