diff --git a/packages/shared/src/telemetry/client-contract.test.ts b/packages/shared/src/telemetry/client-contract.test.ts new file mode 100644 index 0000000000..523bb64018 --- /dev/null +++ b/packages/shared/src/telemetry/client-contract.test.ts @@ -0,0 +1,205 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { TelemetryClient } from "./client.js"; +import type { TelemetryConfig, TelemetryState } from "./types.js"; + +/** + * Wire-contract suite for the telemetry client. + * + * These tests assert the client honors the backend wire contract using ONLY + * relative invariants derived from injected config — never the server's literal + * threshold values. There are intentionally no literal `50` / `512` / `524288` + * / `2` assertion targets in this file: caps are injected small and every bound + * is checked against the injected config value. This keeps the client + * "compatible, not a mirror" — server numbers live in config, not test logic. + */ + +// The server envelope allow-set (`validators.mjs` ENVELOPE_FIELDS). Any extra +// top-level key would be rejected, so every emitted envelope must be a subset. +const CONTRACT_KEYS = new Set(["app", "schemaVersion", "installId", "version", "events", "batchId"]); + +const TEST_STATE: TelemetryState = { + installId: "contract-install", + salt: "contract-salt", + createdAt: "2026-01-01T00:00:00Z", + firstSeenVersion: "0.0.0", +}; + +function makeClient( + config?: Partial, + stateFactory: () => TelemetryState = () => TEST_STATE, + // 0.5 => zero jitter; deterministic backoff. + random: () => number = () => 0.5, +) { + return new TelemetryClient( + { enabled: true, endpoint: "http://localhost:9999/ingest", ...config }, + stateFactory, + "0.0.0-test", + random, + ); +} + +function sentBodies(): Array> { + return vi.mocked(fetch).mock.calls.map((call) => { + const requestInit = call[1] as RequestInit | undefined; + return JSON.parse(String(requestInit?.body ?? "{}")); + }); +} + +function sentRawBodies(): string[] { + return vi.mocked(fetch).mock.calls.map((call) => String((call[1] as RequestInit).body)); +} + +describe("telemetry client wire contract", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("emits only allow-set keys with schemaVersion '1' on every POST", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const maxEventsPerBatch = 3; + const client = makeClient({ maxEventsPerBatch }); + + for (let i = 0; i < 7; i++) client.trackDynamic("plugin.telemetry.evt", { i }); + await client.flush(); + + for (const body of sentBodies()) { + for (const key of Object.keys(body)) { + expect(CONTRACT_KEYS.has(key)).toBe(true); + } + expect(body.schemaVersion).toBe("1"); + } + }); + + it("never exceeds the injected maxEventsPerBatch on any POST", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const maxEventsPerBatch = 4; + const client = makeClient({ maxEventsPerBatch }); + + for (let i = 0; i < 10; i++) client.trackDynamic("plugin.telemetry.evt", { i }); + await client.flush(); + + const bodies = sentBodies(); + expect(bodies.length).toBeGreaterThan(1); // proves it actually chunked + for (const body of bodies) { + expect((body.events as unknown[]).length).toBeLessThanOrEqual(maxEventsPerBatch); + } + }); + + it("never exceeds the injected maxBodyBytes on any POST", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const maxBodyBytes = 1500; + const client = makeClient({ maxEventsPerBatch: 1000, maxBodyBytes }); + + const blob = "y".repeat(300); + for (let i = 0; i < 12; i++) client.trackDynamic("plugin.telemetry.evt", { i, blob }); + await client.flush(); + + const raw = sentRawBodies(); + expect(raw.length).toBeGreaterThan(1); // proves byte-splitting happened + for (const body of raw) { + expect(Buffer.byteLength(body)).toBeLessThanOrEqual(maxBodyBytes); + } + }); + + it("a large flush splits into N envelopes that are each count- and byte-compliant", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const maxEventsPerBatch = 5; + const maxBodyBytes = 4096; + const client = makeClient({ maxEventsPerBatch, maxBodyBytes }); + + for (let i = 0; i < 23; i++) client.trackDynamic("plugin.telemetry.evt", { i }); + await client.flush(); + + const bodies = sentBodies(); + const raw = sentRawBodies(); + const total = bodies.reduce((sum, b) => sum + (b.events as unknown[]).length, 0); + expect(total).toBe(23); // no events lost + bodies.forEach((body, idx) => { + expect((body.events as unknown[]).length).toBeLessThanOrEqual(maxEventsPerBatch); + expect(Buffer.byteLength(raw[idx] as string)).toBeLessThanOrEqual(maxBodyBytes); + }); + }); +}); + +describe("telemetry client wire contract — retry semantics", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("retries a 429 and re-sends the identical batchId (server idempotency)", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce({ ok: false, status: 429 }) + .mockResolvedValue({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + const client = makeClient({ + backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 }, + }); + + client.trackDynamic("plugin.telemetry.evt", { a: 1 }); + await client.flush(); + await vi.advanceTimersByTimeAsync(1_000); + + const bodies = sentBodies(); + expect(bodies).toHaveLength(2); + expect(bodies[0]?.batchId).toBe(bodies[1]?.batchId); + expect(bodies[1]?.events).toEqual(bodies[0]?.events); // no re-mix + client.stop(); + }); + + it("does not retry a terminal 400 or 413", async () => { + for (const status of [400, 413]) { + const fetchMock = vi.fn().mockResolvedValue({ ok: false, status }); + vi.stubGlobal("fetch", fetchMock); + const client = makeClient(); + + client.trackDynamic("plugin.telemetry.evt", { a: 1 }); + await client.flush(); + await vi.advanceTimersByTimeAsync(120_000); + + expect(fetchMock).toHaveBeenCalledTimes(1); + client.stop(); + vi.restoreAllMocks(); + } + }); +}); + +describe("telemetry client wire contract — batchId identity", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("derives a different batchId for the same events under a different installId", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + + const a = makeClient(undefined, () => ({ ...TEST_STATE, installId: "install-a" })); + a.trackDynamic("plugin.telemetry.evt", { a: 1 }); + await a.flush(); + const idA = sentBodies().at(-1)?.batchId; + + vi.mocked(fetch).mockClear(); + + const b = makeClient(undefined, () => ({ ...TEST_STATE, installId: "install-b" })); + b.trackDynamic("plugin.telemetry.evt", { a: 1 }); + await b.flush(); + const idB = sentBodies().at(-1)?.batchId; + + expect(idA).not.toBe(idB); + }); + + it("emits a batchId of at least 32 hex chars — 128-bit collision floor", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const client = makeClient(); + + client.trackDynamic("plugin.telemetry.evt", { a: 1 }); + await client.flush(); + + const id = sentBodies().at(-1)?.batchId as string; + expect(id.length).toBeGreaterThanOrEqual(32); + expect(id).toMatch(/^[0-9a-f]+$/); + }); +}); diff --git a/packages/shared/src/telemetry/client.test.ts b/packages/shared/src/telemetry/client.test.ts index af9ec68d0c..9bcd1e8576 100644 --- a/packages/shared/src/telemetry/client.test.ts +++ b/packages/shared/src/telemetry/client.test.ts @@ -1,7 +1,7 @@ -import { afterEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { TelemetryClient } from "./client.js"; import { resolveTelemetryConfig } from "./config.js"; -import type { TelemetryConfig, TelemetryState } from "./types.js"; +import type { TelemetryConfig, TelemetryDimensions, TelemetryState } from "./types.js"; const TEST_STATE: TelemetryState = { installId: "test-install", @@ -10,12 +10,19 @@ const TEST_STATE: TelemetryState = { firstSeenVersion: "0.0.0", }; -function makeClient(stateFactory = vi.fn(() => TEST_STATE), config?: Partial) { +function makeClient( + stateFactory = vi.fn(() => TEST_STATE), + config?: Partial, + // Seeded RNG for deterministic backoff jitter. 0.5 => zero jitter (the + // symmetric midpoint), so retry delays equal the un-jittered exponential base. + random: () => number = () => 0.5, +) { return { client: new TelemetryClient( { enabled: true, endpoint: "http://localhost:9999/ingest", ...config }, stateFactory, "0.0.0-test", + random, ), stateFactory, }; @@ -26,6 +33,14 @@ function sentBody() { return JSON.parse(String(requestInit?.body ?? "{}")); } +// Parsed request bodies for every POST the client made this test, in call order. +function sentBodies(): Array> { + return vi.mocked(fetch).mock.calls.map((call) => { + const requestInit = call[1] as RequestInit | undefined; + return JSON.parse(String(requestInit?.body ?? "{}")); + }); +} + describe("TelemetryClient runtime event gate", () => { afterEach(() => { vi.restoreAllMocks(); @@ -103,89 +118,9 @@ describe("TelemetryClient runtime event gate", () => { }); }); -// Stubs `fetch` to reject the batch with a given non-OK HTTP status for every -// endpoint the client may try. Returns the mock so call counts can be asserted. -function stubFetchStatus(status: number) { - const fetchMock = vi.fn().mockResolvedValue({ ok: false, status }); - vi.stubGlobal("fetch", fetchMock); - return fetchMock; -} - -// Phase 1 (PAP-2862): characterization pins for today's best-effort, silent-drop -// flush. On ANY non-OK response or network error the drained batch is dropped -// with no re-queue and no second attempt, and no `batchId` is emitted. These pins -// lock the current baseline; Impl-2 (PAP-2853) replaces them when retry lands. -describe("TelemetryClient silent-drop baseline (characterization)", () => { - afterEach(() => { - vi.restoreAllMocks(); - }); - - it("drops the batch on a 429 with no re-queue", async () => { - const fetchMock = stubFetchStatus(429); - const { client } = makeClient(); - - client.track("install.started", {}); - await client.flush(); - expect(fetchMock).toHaveBeenCalledTimes(1); - - // Queue was drained despite the failure: a second flush sends nothing. - await client.flush(); - expect(fetchMock).toHaveBeenCalledTimes(1); - }); - - it("drops the batch on a 413 with no re-queue", async () => { - const fetchMock = stubFetchStatus(413); - const { client } = makeClient(); - - client.track("install.started", {}); - await client.flush(); - expect(fetchMock).toHaveBeenCalledTimes(1); - - await client.flush(); - expect(fetchMock).toHaveBeenCalledTimes(1); - }); - - it("drops the batch on a 400 with no re-queue", async () => { - const fetchMock = stubFetchStatus(400); - const { client } = makeClient(); - - client.track("install.started", {}); - await client.flush(); - expect(fetchMock).toHaveBeenCalledTimes(1); - - await client.flush(); - expect(fetchMock).toHaveBeenCalledTimes(1); - }); - - it("drops the batch on network error with no re-queue", async () => { - const fetchMock = vi.fn().mockRejectedValue(new Error("network down")); - vi.stubGlobal("fetch", fetchMock); - const { client } = makeClient(); - - client.track("install.started", {}); - await client.flush(); - expect(fetchMock).toHaveBeenCalledTimes(1); - - await client.flush(); - expect(fetchMock).toHaveBeenCalledTimes(1); - }); - - it("emits no batchId today", async () => { - vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); - const { client } = makeClient(); - - client.track("install.started", {}); - await client.flush(); - - expect(fetch).toHaveBeenCalledTimes(1); - expect(sentBody()).not.toHaveProperty("batchId"); - }); -}); - -// Phase 2 (PAP-2862): config surface for soft caps + backoff. Fields are optional -// and additive; `resolveTelemetryConfig` fills documented defaults centrally so no -// existing caller changes behavior. Nothing reads these yet — Impl-2 is the first -// consumer. +// Config surface for soft caps + backoff. Fields are optional and additive; +// `resolveTelemetryConfig` fills documented defaults centrally so no existing +// caller changes behavior. describe("resolveTelemetryConfig caps + backoff surface", () => { it("resolveTelemetryConfig returns default caps and backoff", () => { const config = resolveTelemetryConfig(); @@ -225,3 +160,456 @@ describe("resolveTelemetryConfig caps + backoff surface", () => { }); }); }); + +// flush() must never emit an oversized batch. The drained +// queue is sub-divided into envelopes of <= config.maxEventsPerBatch events AND +// <= config.maxBodyBytes serialized bytes, one POST per chunk. Caps are injected +// (small) so assertions are RELATIVE invariants, never server literals. +describe("TelemetryClient chunking (count + bytes)", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("splits more than maxEventsPerBatch events into multiple compliant POSTs", async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + const cap = 2; + const { client } = makeClient(undefined, { maxEventsPerBatch: cap }); + + for (let i = 0; i < 5; i++) client.track("install.started", {}); + await client.flush(); + + // 5 events / cap 2 => 3 POSTs (2 + 2 + 1) + expect(fetchMock).toHaveBeenCalledTimes(3); + const bodies = sentBodies(); + expect(bodies.map((b) => (b.events as unknown[]).length)).toEqual([2, 2, 1]); + for (const body of bodies) { + expect((body.events as unknown[]).length).toBeLessThanOrEqual(cap); + } + }); + + it("splits a chunk whose serialized bytes exceed maxBodyBytes", async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + // Big enough to hold a single fat event but force >1 event to split. + const maxBodyBytes = 900; + const { client } = makeClient(undefined, { maxEventsPerBatch: 100, maxBodyBytes }); + + const blob = "x".repeat(300); + for (let i = 0; i < 6; i++) client.trackDynamic("plugin.telemetry.blob", { blob }); + await client.flush(); + + expect(fetchMock.mock.calls.length).toBeGreaterThan(1); + for (const call of fetchMock.mock.calls) { + const body = String((call[1] as RequestInit).body); + expect(Buffer.byteLength(body)).toBeLessThanOrEqual(maxBodyBytes); + } + }); + + it("drops a single event larger than maxBodyBytes and logs, sending nothing", async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const { client } = makeClient(undefined, { maxBodyBytes: 200 }); + + client.trackDynamic("plugin.telemetry.blob", { blob: "x".repeat(5000) }); + await client.flush(); + + expect(fetchMock).not.toHaveBeenCalled(); + expect(warn).toHaveBeenCalled(); + }); +}); + +// Every emitted chunk carries a deterministic, salt-free content-hash `batchId` +// so server-side retries de-dupe (202) instead of double-counting. Two +// invariants hold: the hash input includes `installId` (so two installs sending +// identical events get distinct ids), and the id is >= 32 hex chars (128-bit +// collision floor). +describe("TelemetryClient deterministic batchId", () => { + // `deriveBatchId` intentionally hashes the full event objects, INCLUDING each + // event's `occurredAt` wall-clock stamp, so the id stays a faithful content + // hash: two genuinely distinct sends get distinct ids and the server ledger + // (keyed on `batchId`) counts both. Dropping `occurredAt` from the hash would + // collapse same-shape sends at different times onto one id and make the server + // silently drop the later batch as a replay — a silent-loss failure mode. + // These cross-instance determinism checks therefore freeze the clock so + // both clients stamp an identical `occurredAt`, isolating the hash's structural + // determinism from wall-clock skew. + beforeEach(() => { + vi.useFakeTimers({ now: new Date("2026-01-01T00:00:00.000Z") }); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("emits a batchId on every envelope", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const { client } = makeClient(); + + client.track("install.started", {}); + await client.flush(); + + expect(typeof sentBody().batchId).toBe("string"); + }); + + it("derives a stable batchId for identical events (idempotent retry key)", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + + const first = makeClient(); + first.client.trackDynamic("plugin.telemetry.evt", { a: 1, b: "two" }); + await first.client.flush(); + const idA = sentBody().batchId; + + vi.mocked(fetch).mockClear(); + + const second = makeClient(); + second.client.trackDynamic("plugin.telemetry.evt", { a: 1, b: "two" }); + await second.client.flush(); + const idB = sentBody().batchId; + + // Same installId + same event content => identical id. + expect(idA).toBe(idB); + }); + + it("derives a different batchId for different events", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + + const first = makeClient(); + first.client.trackDynamic("plugin.telemetry.evt", { a: 1 }); + await first.client.flush(); + const idA = sentBody().batchId; + + vi.mocked(fetch).mockClear(); + + const second = makeClient(); + second.client.trackDynamic("plugin.telemetry.evt", { a: 2 }); + await second.client.flush(); + const idB = sentBody().batchId; + + expect(idA).not.toBe(idB); + }); + + it("derives a different batchId for the same events under a different installId", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + + const stateA = vi.fn(() => ({ ...TEST_STATE, installId: "install-a" })); + const first = makeClient(stateA); + first.client.trackDynamic("plugin.telemetry.evt", { a: 1 }); + await first.client.flush(); + const idA = sentBody().batchId; + + vi.mocked(fetch).mockClear(); + + const stateB = vi.fn(() => ({ ...TEST_STATE, installId: "install-b" })); + const second = makeClient(stateB); + second.client.trackDynamic("plugin.telemetry.evt", { a: 1 }); + await second.client.flush(); + const idB = sentBody().batchId; + + expect(idA).not.toBe(idB); + }); + + it("emits a batchId of at least 32 hex chars (128-bit collision floor)", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const { client } = makeClient(); + + client.track("install.started", {}); + await client.flush(); + + const id = sentBody().batchId as string; + expect(id.length).toBeGreaterThanOrEqual(32); + expect(id).toMatch(/^[0-9a-f]+$/); + }); + + it("does not crash flush on a circular dimension; drops-and-logs it and still sends valid events", async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const { client } = makeClient(); + + // A plugin passing a circular `dimensions` object. `enqueue` shallow-copies + // the top level, so the cycle survives one level down and would drive both + // `stableStringify` (deriveBatchId) and `JSON.stringify` (serializedBytes / + // wire body) into an unhandled throw without the guards. + const circular: Record = { a: 1 }; + circular.self = circular; + // Cast models an untyped third-party plugin passing a malformed object at + // runtime — TypeScript would reject the cycle, but the wire path cannot. + client.trackDynamic("plugin.telemetry.circular", circular as unknown as TelemetryDimensions); + // A well-formed event queued alongside it must still be delivered. + client.trackDynamic("plugin.telemetry.ok", { a: 1 }); + + // flush() resolves instead of rejecting with a RangeError/TypeError. + await expect(client.flush()).resolves.toBeUndefined(); + + // The circular event was dropped-and-logged (fail loudly), not sent. + expect(warn).toHaveBeenCalledWith( + expect.stringContaining("non-serializable dimension (circular reference?)"), + ); + + // The valid event still went out on the wire with a well-formed batchId. + expect(fetchMock).toHaveBeenCalledTimes(1); + const body = sentBody(); + expect(body.events).toHaveLength(1); + expect((body.events as Array<{ name: string }>)[0]?.name).toBe("plugin.telemetry.ok"); + const id = body.batchId as string; + expect(id.length).toBeGreaterThanOrEqual(32); + expect(id).toMatch(/^[0-9a-f]+$/); + }); +}); + +// Batch-grouped retry with capped, jittered exponential +// backoff. Retryable statuses (429/502/503/504 + network) re-send the EXACT +// same events + batchId (no re-mix — preserves server idempotency); terminal +// statuses (400/405/409/413) never retry. Backoff uses the injected seeded RNG +// and retries are driven off fake timers so tests are deterministic. +describe("TelemetryClient batched retry + backoff", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("retries a 429 with the same batchId and eventually succeeds", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce({ ok: false, status: 429 }) + .mockResolvedValue({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + const { client } = makeClient(undefined, { + backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 }, + }); + + client.track("install.started", {}); + await client.flush(); + expect(fetchMock).toHaveBeenCalledTimes(1); // attempt 1 -> 429, queued for retry + + // Un-jittered delay for attempt 1 == baseDelayMs. + await vi.advanceTimersByTimeAsync(1_000); + expect(fetchMock).toHaveBeenCalledTimes(2); // attempt 2 -> 200 + + const bodies = sentBodies(); + expect(bodies[0]?.batchId).toBe(bodies[1]?.batchId); // identical id on retry + client.stop(); + }); + + it("does not retry a terminal 400", async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 400 }); + vi.stubGlobal("fetch", fetchMock); + const { client } = makeClient(); + + client.track("install.started", {}); + await client.flush(); + await vi.advanceTimersByTimeAsync(120_000); + + expect(fetchMock).toHaveBeenCalledTimes(1); + client.stop(); + }); + + it("does not retry a terminal 413", async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 413 }); + vi.stubGlobal("fetch", fetchMock); + const { client } = makeClient(); + + client.track("install.started", {}); + await client.flush(); + await vi.advanceTimersByTimeAsync(120_000); + + expect(fetchMock).toHaveBeenCalledTimes(1); + client.stop(); + }); + + it("stops after maxAttempts on a persistent 429 and drops-and-logs", async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 429 }); + vi.stubGlobal("fetch", fetchMock); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const maxAttempts = 3; + const { client } = makeClient(undefined, { + backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts, jitterRatio: 0.25 }, + }); + + client.track("install.started", {}); + await client.flush(); + // Drive all remaining scheduled retries. + await vi.advanceTimersByTimeAsync(600_000); + + expect(fetchMock).toHaveBeenCalledTimes(maxAttempts); + expect(warn).toHaveBeenCalled(); + client.stop(); + }); + + // Issue 1 (PR #9946): a transient upstream 5xx on the primary endpoint must + // fall through to the healthy secondary endpoint instead of returning early. + it("falls through to the secondary endpoint on a transient 5xx", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce({ ok: false, status: 503 }) // primary endpoint: transient + .mockResolvedValueOnce({ ok: true }); // secondary endpoint: healthy + vi.stubGlobal("fetch", fetchMock); + // Empty endpoint => the two built-in DEFAULT_ENDPOINTS are used. + const { client } = makeClient(undefined, { endpoint: "" }); + + client.track("install.started", {}); + await client.flush(); + + // Both endpoints tried within a single attempt; delivered on the secondary, + // so no retry is queued. + expect(fetchMock).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(120_000); + expect(fetchMock).toHaveBeenCalledTimes(2); + client.stop(); + }); + + // Issue 1 (PR #9946): when every endpoint returns a transient 5xx the status + // is still surfaced as retryable (not swallowed). + it("surfaces the transient status for retry when all endpoints 5xx", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce({ ok: false, status: 502 }) // primary + .mockResolvedValueOnce({ ok: false, status: 502 }) // secondary + .mockResolvedValue({ ok: true }); // retry succeeds + vi.stubGlobal("fetch", fetchMock); + const { client } = makeClient(undefined, { + endpoint: "", + backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 }, + }); + + client.track("install.started", {}); + await client.flush(); + expect(fetchMock).toHaveBeenCalledTimes(2); // both endpoints 502 -> queued for retry + + await vi.advanceTimersByTimeAsync(1_000); // attempt 2 -> ok on primary + expect(fetchMock).toHaveBeenCalledTimes(3); + client.stop(); + }); + + // Issue 3 (PR #9946): an out-of-range `Retry-After` hint is clamped to + // maxDelayMs rather than overflowing the timer range into a near-immediate + // (Node-clamped ~1ms) retry. + it("caps a large Retry-After hint at maxDelayMs", async () => { + const retryAfterResponse = { + ok: false, + status: 429, + headers: { get: (name: string) => (name === "retry-after" ? "999999999" : null) }, + }; + const fetchMock = vi.fn().mockResolvedValueOnce(retryAfterResponse).mockResolvedValue({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + const { client } = makeClient(undefined, { + backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 }, + }); + + client.track("install.started", {}); + await client.flush(); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // Well before the cap: without clamping the huge hint overflows the timer + // range and Node fires it near-immediately, so this would already be 2. + await vi.advanceTimersByTimeAsync(5_000); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // At the cap the retry fires. + await vi.advanceTimersByTimeAsync(25_000); + expect(fetchMock).toHaveBeenCalledTimes(2); + client.stop(); + }); +}); + +// The pending-retry store is bounded at +// config.maxPendingRetryBatches. On overflow the OLDEST batch is evicted +// (newest prioritized) and each eviction is logged (no silent loss). In-memory +// only. Caps are injected; assertions are relative. +describe("TelemetryClient bounded pending-retry store", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("evicts the oldest batch and retries only the newest within the bound", async () => { + // First 3 POSTs (one per single-event chunk) fail 429; retries then succeed. + const fetchMock = vi + .fn() + .mockResolvedValueOnce({ ok: false, status: 429 }) + .mockResolvedValueOnce({ ok: false, status: 429 }) + .mockResolvedValueOnce({ ok: false, status: 429 }) + .mockResolvedValue({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const { client } = makeClient(undefined, { maxEventsPerBatch: 1, maxPendingRetryBatches: 2 }); + + client.trackDynamic("plugin.telemetry.evt", { n: 1 }); + client.trackDynamic("plugin.telemetry.evt", { n: 2 }); + client.trackDynamic("plugin.telemetry.evt", { n: 3 }); + await client.flush(); + + // 3 initial attempts (one per chunk), each 429 -> enqueued; the 3rd enqueue + // overflows the bound (2) and evicts the oldest. + expect(fetchMock).toHaveBeenCalledTimes(3); + const initial = sentBodies(); + const [oldestId, midId, newestId] = initial.map((b) => b.batchId as string); + expect(warn).toHaveBeenCalled(); // eviction logged + + await vi.advanceTimersByTimeAsync(120_000); + + // Only the 2 newest were retained and retried; the oldest was dropped. + const retriedIds = sentBodies() + .slice(3) + .map((b) => b.batchId as string); + expect(retriedIds.sort()).toEqual([midId, newestId].sort()); + expect(retriedIds).not.toContain(oldestId); + client.stop(); + }); + + // Issue 2 (PR #9946): a batch that is immediately evicted by the bound must + // NOT leave a retry timer behind. With maxPendingRetryBatches: 0 every failed + // batch is discarded on enqueue, so no timer should ever fire. + it("schedules no retry timer for a batch evicted on enqueue (bound 0)", async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 429 }); + vi.stubGlobal("fetch", fetchMock); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const { client } = makeClient(undefined, { maxEventsPerBatch: 1, maxPendingRetryBatches: 0 }); + + client.trackDynamic("plugin.telemetry.evt", { n: 1 }); + client.trackDynamic("plugin.telemetry.evt", { n: 2 }); + await client.flush(); + + // 2 initial attempts (one per chunk); each 429 is enqueued then immediately + // evicted by the 0 bound, so nothing is retried. + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(warn).toHaveBeenCalled(); // eviction logged + + await vi.advanceTimersByTimeAsync(120_000); + expect(fetchMock).toHaveBeenCalledTimes(2); // no timer fired + client.stop(); + }); + + // A batch evicted to keep `pending` within the bound must also have its + // already-scheduled retry timer cancelled — otherwise each overflow strands a + // live timer for a batch that no longer exists, and the timer set grows + // unbounded even though `pending` stays bounded. + it("cancels the retry timer of a batch evicted from the bounded store", async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 429 }); + vi.stubGlobal("fetch", fetchMock); + vi.spyOn(console, "warn").mockImplementation(() => {}); + const { client } = makeClient(undefined, { maxEventsPerBatch: 1, maxPendingRetryBatches: 1 }); + + client.trackDynamic("plugin.telemetry.evt", { n: 1 }); + client.trackDynamic("plugin.telemetry.evt", { n: 2 }); + client.trackDynamic("plugin.telemetry.evt", { n: 3 }); + await client.flush(); + + // 3 chunks each 429 -> enqueued; every enqueue past the first overflows the + // bound (1) and evicts the previous batch. Only the newest batch survives, so + // exactly one retry timer should remain live (the two evicted timers were + // cancelled), not one per failed chunk. + expect(fetchMock).toHaveBeenCalledTimes(3); + expect(vi.getTimerCount()).toBe(1); + client.stop(); + }); +}); diff --git a/packages/shared/src/telemetry/client.ts b/packages/shared/src/telemetry/client.ts index 6af4e7f375..4a131d2029 100644 --- a/packages/shared/src/telemetry/client.ts +++ b/packages/shared/src/telemetry/client.ts @@ -4,35 +4,111 @@ import type { TelemetryDimensions, TelemetryEvent, TelemetryEventDimensions, + TelemetryEventEnvelope, TelemetryEventName, TelemetryState, } from "./types.js"; +import { type ResolvedTelemetryCaps, resolveCaps } from "./config.js"; import { PAPERCLIP_EVENTS } from "./generated/paperclip-telemetry.js"; const DEFAULT_ENDPOINTS = [ "https://telemetry.paperclip.ing/ingest", "https://rusqrrg391.execute-api.us-east-1.amazonaws.com/ingest", ] as const; +// Queue-pressure valve: auto-flush once this many events are buffered. This is +// an in-memory backpressure trigger, independent of the wire caps that +// `chunkForSend` enforces on each POST. const BATCH_SIZE = 50; const SEND_TIMEOUT_MS = 5_000; +/** + * Deterministic, key-stable JSON serialization used to derive a content-stable + * `batchId`. Object keys are sorted so two structurally-equal event sets always + * produce the same string regardless of insertion order. Mirrors the + * `stableStringify` exemplar in `external-objects-server.ts`. + */ +function stableStringify(value: unknown): string { + if (Array.isArray(value)) { + return `[${value.map((entry) => stableStringify(entry)).join(",")}]`; + } + if (value && typeof value === "object") { + const record = value as Record; + return `{${Object.keys(record) + .sort() + .map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`) + .join(",")}}`; + } + return JSON.stringify(value); +} + +/** + * Parses a `Retry-After` header into milliseconds — defensively, only when + * present and a non-negative number of seconds. The telemetry backend emits no + * `Retry-After` today, so this never fires in practice; it exists so a future + * server hint is honored rather than ignored. The HTTP-date form is + * intentionally not parsed. + */ +function parseRetryAfterMs(response: { headers?: { get?(name: string): string | null } }): number | undefined { + const raw = response.headers?.get?.("retry-after"); + if (!raw) return undefined; + const seconds = Number(raw); + if (!Number.isFinite(seconds) || seconds < 0) return undefined; + return Math.round(seconds * 1000); +} + type TrackArgs = keyof TelemetryEventDimensions extends never ? [dimensions?: TelemetryEventDimensions] : [dimensions: TelemetryEventDimensions]; +// Length of the truncated hex `batchId`. 32 hex chars = 128 bits — the +// collision-safe floor. Do not lower below 32. +const BATCH_ID_HEX_LENGTH = 32; + +/** + * A chunk awaiting retry. `events` + `batchId` are frozen at first send and + * re-sent verbatim on every attempt — re-mixing events would change the content + * the server hashed under this id and trigger a 409. `attempt` is 1-based (the + * value of the attempt about to be made); `nextAttemptAt` is an epoch-ms gate. + * `timerId` is this batch's scheduled retry wake-up (if any) — held on the record + * so it can be cancelled when the batch is evicted from the bounded store, + * otherwise the timer would keep firing for a batch that no longer exists. + */ +interface PendingBatch { + events: TelemetryEvent[]; + batchId: string; + attempt: number; + nextAttemptAt: number; + timerId?: ReturnType; +} + export class TelemetryClient { private queue: TelemetryEvent[] = []; private readonly config: TelemetryConfig; + private readonly caps: ResolvedTelemetryCaps; private readonly stateFactory: () => TelemetryState; private readonly version: string; + private readonly random: () => number; private state: TelemetryState | null = null; private flushInterval: ReturnType | null = null; + // In-memory pending-retry store (best-effort; never persisted). Bounded by + // `maxPendingRetryBatches`. Insertion order == age (oldest at the front). + private pending: PendingBatch[] = []; + private readonly retryTimers = new Set>(); - constructor(config: TelemetryConfig, stateFactory: () => TelemetryState, version: string) { + constructor( + config: TelemetryConfig, + stateFactory: () => TelemetryState, + version: string, + // Injectable RNG for backoff jitter — defaults to `Math.random`; tests pass + // a seeded function for deterministic backoff. Callers keep the 3-arg form. + random: () => number = Math.random, + ) { this.config = config; + this.caps = resolveCaps(config); this.stateFactory = stateFactory; this.version = version; + this.random = random; } /** @@ -70,21 +146,263 @@ export class TelemetryClient { } async flush(): Promise { - if (!this.config.enabled || this.queue.length === 0) return; + if (!this.config.enabled) return; + + // Re-send any due retries first, then send freshly-queued events. + await this.drainPending(); + if (this.queue.length === 0) return; const events = this.queue.splice(0); + for (const chunk of this.chunkForSend(events)) { + await this.attemptSend({ + events: chunk, + batchId: this.deriveBatchId(this.getState().installId, chunk), + attempt: 1, + nextAttemptAt: 0, + }); + } + } + + /** + * Partitions a drained event list into wire-compliant chunks: first by + * `maxEventsPerBatch` (count), then recursively byte-splits any chunk whose + * serialized envelope still exceeds `maxBodyBytes` (halving). A single event + * that alone exceeds `maxBodyBytes` is dropped-and-logged (fail loudly) rather + * than sent over-limit. + */ + private chunkForSend(events: TelemetryEvent[]): TelemetryEvent[][] { + const maxCount = Math.max(1, this.caps.maxEventsPerBatch); + const out: TelemetryEvent[][] = []; + for (let i = 0; i < events.length; i += maxCount) { + this.splitByBytes(events.slice(i, i + maxCount), out); + } + return out; + } + + private splitByBytes(chunk: TelemetryEvent[], out: TelemetryEvent[][]): void { + if (chunk.length === 0) return; + const bytes = this.serializedBytes(this.buildEnvelope(chunk)); + if (bytes <= this.caps.maxBodyBytes) { + out.push(chunk); + return; + } + if (chunk.length === 1) { + this.warn( + Number.isFinite(bytes) + ? `dropping 1 event whose serialized envelope exceeds maxBodyBytes (${this.caps.maxBodyBytes} bytes); event="${chunk[0]?.name}"` + : `dropping 1 event with a non-serializable dimension (circular reference?); event="${chunk[0]?.name}"`, + ); + return; + } + const mid = Math.ceil(chunk.length / 2); + this.splitByBytes(chunk.slice(0, mid), out); + this.splitByBytes(chunk.slice(mid), out); + } + + private buildEnvelope(events: TelemetryEvent[], batchId?: string): TelemetryEventEnvelope { const state = this.getState(); - const endpoints = this.resolveEndpoints(); - const app = this.config.app ?? "paperclip"; - const schemaVersion = this.config.schemaVersion ?? "1"; - const body = JSON.stringify({ - app, - schemaVersion, + return { + app: this.config.app ?? "paperclip", + schemaVersion: this.config.schemaVersion ?? "1", installId: state.installId, version: this.version, events, - }); + batchId: batchId ?? this.deriveBatchId(state.installId, events), + }; + } + /** + * Deterministic, salt-free content-hash idempotency key. Hashes + * `{installId, events}` (scopes the key per install, matching the + * server's `contentSha256` scope so cross-install batches never collide on the + * server ledger key) and truncates to 128 bits. Identical input always + * yields the same id, so a retried batch replays idempotently (202) instead of + * double-counting; different events/install yield a different id. + */ + private deriveBatchId(installId: string, events: TelemetryEvent[]): string { + try { + return createHash("sha256") + .update(stableStringify({ installId, events })) + .digest("hex") + .slice(0, BATCH_ID_HEX_LENGTH); + } catch { + // A plugin can pass a circular (or otherwise non-serializable) `dimensions` + // object via `trackDynamic`; `stableStringify` would then recurse until it + // throws a `RangeError`. Fall back to a count-based hash so this call never + // crashes the flush. Such an event can't be JSON-serialized for the wire + // either, so it is dropped-and-logged in `splitByBytes` — this fallback id + // is only ever attached to a batch that is about to be dropped, so its + // weakened idempotency is moot. The single operator-facing signal is the + // drop-and-log warning, not a (recursively-repeated) warning here. + return createHash("sha256") + .update(JSON.stringify({ installId, count: events.length })) + .digest("hex") + .slice(0, BATCH_ID_HEX_LENGTH); + } + } + + private serializedBytes(envelope: TelemetryEventEnvelope): number { + try { + return Buffer.byteLength(JSON.stringify(envelope)); + } catch { + // A non-serializable event (e.g. a circular `dimensions` object from a + // plugin) can neither be byte-measured nor sent over the wire. Report it + // as effectively unbounded so `splitByBytes` routes it to the existing + // over-limit drop-and-log path instead of throwing out of the flush. + return Number.POSITIVE_INFINITY; + } + } + + /** + * Sends one batch (its current `attempt`). On a retryable failure the EXACT + * same events + `batchId` are re-queued with capped, jittered backoff; on a + * terminal failure or after `maxAttempts` the batch is dropped-and-logged. + */ + private async attemptSend(batch: PendingBatch): Promise { + const body = JSON.stringify(this.buildEnvelope(batch.events, batch.batchId)); + const outcome = this.classifyOutcome(await this.postEnvelope(body)); + + if (outcome.kind === "ok") return; + + if (outcome.kind === "terminal") { + this.warn( + `dropping batch ${batch.batchId} on terminal response (HTTP ${outcome.status}); ${batch.events.length} event(s) lost`, + ); + return; + } + + // Retryable (429/502/503/504 or network/timeout). + if (batch.attempt >= this.caps.backoff.maxAttempts) { + this.warn( + `dropping batch ${batch.batchId} after ${batch.attempt} attempt(s); ${batch.events.length} event(s) lost`, + ); + return; + } + // Cap the delay at maxDelayMs. `computeBackoffMs` is already capped, but a + // server `Retry-After` hint is not — an out-of-range value could otherwise + // overflow the runtime timer range and be clamped by Node to a near-immediate + // timeout, causing rapid retries. The cap keeps every retry within the + // configured backoff ceiling. + const delayMs = Math.min( + outcome.retryAfterMs ?? this.computeBackoffMs(batch.attempt), + this.caps.backoff.maxDelayMs, + ); + this.enqueuePending({ + events: batch.events, + batchId: batch.batchId, + attempt: batch.attempt + 1, + nextAttemptAt: Date.now() + delayMs, + }); + } + + private classifyOutcome( + result: { kind: "ok" } | { kind: "status"; status: number; retryAfterMs?: number } | { kind: "network" }, + ): { kind: "ok" } | { kind: "retry"; retryAfterMs?: number } | { kind: "terminal"; status: number } { + if (result.kind === "ok") return { kind: "ok" }; + if (result.kind === "network") return { kind: "retry" }; + if (this.isRetryableStatus(result.status)) { + return { kind: "retry", retryAfterMs: result.retryAfterMs }; + } + return { kind: "terminal", status: result.status }; + } + + private isRetryableStatus(status: number): boolean { + return status === 429 || status === 502 || status === 503 || status === 504; + } + + /** + * Capped exponential backoff with symmetric jitter: + * `min(maxDelayMs, baseDelayMs * 2^(attempt-1)) * (1 ± jitterRatio)`, using the + * injected RNG. `attempt` is the failed attempt (1-based). + */ + private computeBackoffMs(attempt: number): number { + const { baseDelayMs, maxDelayMs, jitterRatio } = this.caps.backoff; + const base = Math.min(maxDelayMs, baseDelayMs * 2 ** (attempt - 1)); + const jitter = base * jitterRatio * (this.random() * 2 - 1); + return Math.max(0, Math.min(maxDelayMs, Math.round(base + jitter))); + } + + /** + * Pushes a batch onto the pending store (bounded), then schedules its retry + * wake-up. On overflow the OLDEST batches (front) are evicted first — newest + * prioritized — and each eviction is logged so no batch is lost silently. + */ + private enqueuePending(batch: PendingBatch): void { + this.pending.push(batch); + const bound = Math.max(0, this.caps.maxPendingRetryBatches); + while (this.pending.length > bound) { + const evicted = this.pending.shift(); + // Cancel the evicted batch's scheduled retry so its timer doesn't keep the + // wake-up alive for a batch that is no longer in the store. Without this a + // flush that overflows the bound would strand one live timer per evicted + // batch even though `pending` itself stays bounded. + if (evicted) this.cancelRetryTimer(evicted); + this.warn( + `pending-retry store full (bound=${bound}); evicted oldest batch ${evicted?.batchId}; ${evicted?.events.length ?? 0} event(s) lost`, + ); + } + // Only schedule a wake-up if this batch actually survived eviction. When the + // batch is immediately evicted by the bound (e.g. a large failed flush, or + // `maxPendingRetryBatches: 0`) it has no pending work, so scheduling a timer + // for it would strand thousands of no-op timers behind a small bound. + if (this.pending.includes(batch)) { + this.scheduleDrain(batch); + } + } + + /** Cancels a pending batch's scheduled retry wake-up, if it has one. */ + private cancelRetryTimer(batch: PendingBatch): void { + if (batch.timerId === undefined) return; + clearTimeout(batch.timerId); + this.retryTimers.delete(batch.timerId); + batch.timerId = undefined; + } + + private scheduleDrain(batch: PendingBatch): void { + const delay = Math.max(0, batch.nextAttemptAt - Date.now()); + const timer = setTimeout(() => { + this.retryTimers.delete(timer); + batch.timerId = undefined; + void this.drainPending(); + }, delay); + // Don't keep the process alive for a best-effort retry (CLI exits promptly). + if (typeof timer === "object" && timer !== null && "unref" in timer) { + (timer as { unref(): void }).unref(); + } + this.retryTimers.add(timer); + batch.timerId = timer; + } + + /** Re-sends every pending batch whose `nextAttemptAt` is due. */ + private async drainPending(): Promise { + if (this.pending.length === 0) return; + const now = Date.now(); + const due: PendingBatch[] = []; + const waiting: PendingBatch[] = []; + for (const batch of this.pending) { + (batch.nextAttemptAt <= now ? due : waiting).push(batch); + } + this.pending = waiting; + for (const batch of due) { + await this.attemptSend(batch); + } + } + + /** + * POSTs a serialized envelope, trying each endpoint in order. Returns the + * definitive outcome: `ok` on a 2xx; the HTTP `status` on a definitive non-2xx + * response (4xx incl. 429, which the shared API gateway fronts for every + * endpoint, so it is authoritative — stop here); or `network` when every + * endpoint threw. A transient upstream 5xx (502/503/504) does NOT stop the + * loop: a sibling endpoint may be healthy, so we fall through to it and only + * surface the last transient status if every endpoint returns one — matching + * the pre-retry loop's endpoint-fallback behavior. + */ + private async postEnvelope( + body: string, + ): Promise<{ kind: "ok" } | { kind: "status"; status: number; retryAfterMs?: number } | { kind: "network" }> { + const endpoints = this.resolveEndpoints(); + let lastTransient: { kind: "status"; status: number; retryAfterMs?: number } | undefined; for (const endpoint of endpoints) { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), SEND_TIMEOUT_MS); @@ -95,15 +413,32 @@ export class TelemetryClient { body, signal: controller.signal, }); - if (response.ok) { - return; + if (response.ok) return { kind: "ok" }; + const status = { kind: "status" as const, status: response.status, retryAfterMs: parseRetryAfterMs(response) }; + // Transient upstream 5xx: remember it and try the next endpoint. + if (this.isTransientServerStatus(response.status)) { + lastTransient = status; + continue; } + return status; } catch { - // Try the next built-in endpoint before dropping the batch. + // Network/timeout on this endpoint — try the next built-in endpoint. } finally { clearTimeout(timer); } } + // Every endpoint failed: surface the last transient status (still retryable) + // if we saw one, otherwise report a pure network failure. + return lastTransient ?? { kind: "network" }; + } + + /** Upstream 5xx that a healthy sibling endpoint may still be able to serve. */ + private isTransientServerStatus(status: number): boolean { + return status === 502 || status === 503 || status === 504; + } + + private warn(message: string): void { + console.warn(`[telemetry] ${message}`); } startPeriodicFlush(intervalMs: number = 60_000): void { @@ -122,6 +457,10 @@ export class TelemetryClient { clearInterval(this.flushInterval); this.flushInterval = null; } + for (const timer of this.retryTimers) { + clearTimeout(timer); + } + this.retryTimers.clear(); } hashPrivateRef(value: string): string { diff --git a/packages/shared/src/telemetry/config.ts b/packages/shared/src/telemetry/config.ts index 7596e3adea..dfcd4d692c 100644 --- a/packages/shared/src/telemetry/config.ts +++ b/packages/shared/src/telemetry/config.ts @@ -5,7 +5,7 @@ const CI_ENV_VARS = ["CI", "CONTINUOUS_INTEGRATION", "BUILD_NUMBER", "GITHUB_ACT /** * Single source of truth for telemetry soft caps + backoff. Kept as config * *defaults* (not hardcoded flush logic) so later work reads config, not - * literals. Exported so Impl-2's `client.ts` consumer resolves the same values. + * literals. Exported so the `client.ts` consumer resolves the same values. */ export const TELEMETRY_DEFAULTS: { readonly maxEventsPerBatch: number; @@ -32,17 +32,24 @@ export type TelemetryConfigOverrides = Partial< > >; -type ResolvedCaps = Pick< - TelemetryConfig, - "maxEventsPerBatch" | "maxBodyBytes" | "maxPendingRetryBatches" | "backoff" ->; +/** + * Fully-resolved caps + backoff — every field is present (defaults applied), so + * `client.ts` can consume them without re-defaulting. `TelemetryConfig`'s own + * cap fields stay optional (additive wire surface); this is the resolved view. + */ +export interface ResolvedTelemetryCaps { + maxEventsPerBatch: number; + maxBodyBytes: number; + maxPendingRetryBatches: number; + backoff: TelemetryBackoffConfig; +} /** * Resolves soft caps + backoff, applying `TELEMETRY_DEFAULTS` for any field the * caller did not override. One source of truth for both the config surface and - * Impl-2's future `client.ts` consumer. + * the `client.ts` consumer. */ -export function resolveCaps(overrides?: TelemetryConfigOverrides): ResolvedCaps { +export function resolveCaps(overrides?: TelemetryConfigOverrides): ResolvedTelemetryCaps { return { maxEventsPerBatch: overrides?.maxEventsPerBatch ?? TELEMETRY_DEFAULTS.maxEventsPerBatch, maxBodyBytes: overrides?.maxBodyBytes ?? TELEMETRY_DEFAULTS.maxBodyBytes, diff --git a/packages/shared/src/telemetry/types.ts b/packages/shared/src/telemetry/types.ts index 039d99118c..083d4341a7 100644 --- a/packages/shared/src/telemetry/types.ts +++ b/packages/shared/src/telemetry/types.ts @@ -13,8 +13,8 @@ export interface TelemetryState { /** * Exponential-backoff-with-jitter parameters for the (future) batched-retry * sender. Shape mirrors the plugin worker crash-recovery backoff - * (`server/src/services/plugin-worker-manager.ts`). Consumed by Impl-2; nothing - * reads it yet. + * (`server/src/services/plugin-worker-manager.ts`). Consumed by the + * batched-retry sender; nothing reads it yet. */ export interface TelemetryBackoffConfig { baseDelayMs: number; @@ -31,7 +31,7 @@ export interface TelemetryConfig { /** * Optional, additive soft caps + backoff. Defaulted centrally in * `resolveTelemetryConfig`; no wire/envelope change and no consumer today — - * Impl-2 (PAP-2853) is the first reader. + * the batched-retry sender is the first reader. */ maxEventsPerBatch?: number; maxBodyBytes?: number; @@ -56,6 +56,14 @@ export interface TelemetryEventEnvelope { installId: string; version: string; events: TelemetryEvent[]; + /** + * Deterministic, salt-free content-hash of `{installId, events}` used as the + * server idempotency key so a retried batch de-dupes (202 replay) instead of + * double-counting. Derived in `client.ts`; the server allow-set already + * accepts it. The hash is salt-free so it stays stable across installs and + * never leaks the per-install salt. + */ + batchId: string; } export type RegisteredPluginEventName = never; diff --git a/server/src/__tests__/telemetry-client-flush.test.ts b/server/src/__tests__/telemetry-client-flush.test.ts index 85826e443c..80ffc3b3a9 100644 --- a/server/src/__tests__/telemetry-client-flush.test.ts +++ b/server/src/__tests__/telemetry-client-flush.test.ts @@ -98,3 +98,51 @@ describe("TelemetryClient periodic flush", () => { client.stop(); }); }); + +describe("TelemetryClient retry integration", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("retries a 429 batch on the same batchId until it succeeds", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce({ ok: false, status: 429 }) + .mockResolvedValue({ ok: true }); + vi.stubGlobal("fetch", fetchMock); + + const state: TelemetryState = { + installId: "test-install", + salt: "test-salt", + createdAt: "2026-01-01T00:00:00Z", + firstSeenVersion: "0.0.0", + }; + const client = new TelemetryClient( + { + enabled: true, + endpoint: "http://localhost:9999/ingest", + backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 }, + }, + () => state, + "0.0.0-test", + () => 0.5, // seeded RNG -> zero jitter -> delay == baseDelayMs + ); + + client.track("install.started"); + await client.flush(); + expect(fetchMock).toHaveBeenCalledTimes(1); // attempt 1 -> 429 + + await vi.advanceTimersByTimeAsync(1_000); + expect(fetchMock).toHaveBeenCalledTimes(2); // attempt 2 -> 200 + + const first = JSON.parse(String((fetchMock.mock.calls[0]?.[1] as RequestInit).body)); + const second = JSON.parse(String((fetchMock.mock.calls[1]?.[1] as RequestInit).body)); + expect(second.batchId).toBe(first.batchId); + expect(second.events).toEqual(first.events); + client.stop(); + }); +});