From 6d1e44f206a2370bba1b2a28b3ed04554a89465b Mon Sep 17 00:00:00 2001 From: DongHyeonka Date: Thu, 13 Aug 2026 23:05:43 +0900 Subject: [PATCH] fix: terminate telemetry work on disposal Give the best-effort telemetry adapter a terminal ACTIVE/DISPOSED lifecycle. dispose() now removes the pagehide listener, clears the queue, invalidates scheduled callback generations and aborts the in-flight sink; emit after dispose is a no-op and a sink that ignores the abort cannot reschedule or update post-dispose state. flush() joins the active delivery instead of resolving early, and runtime infrastructure teardown disposes telemetry first. Telemetry and diagnostics capacities are validated at construction against a documented ceiling, so NaN or Infinity can no longer disable eviction. Co-Authored-By: Claude Opus 5 (1M context) --- ...D-07-diagnostics-and-telemetry-exporter.md | 14 +++ docs/operations/adapter-remediation-ledger.md | 4 +- .../diagnostics/bounded-diagnostics.ts | 10 +- .../telemetry/best-effort-telemetry.ts | 111 ++++++++++++---- src/bootstrap/runtime-adapters.ts | 4 + tests/unit/runtime-adapters.test.ts | 32 +++++ tests/unit/telemetry.test.ts | 119 ++++++++++++++++++ 7 files changed, 269 insertions(+), 25 deletions(-) diff --git a/docs/architecture/decisions/VD-07-diagnostics-and-telemetry-exporter.md b/docs/architecture/decisions/VD-07-diagnostics-and-telemetry-exporter.md index 28eb4c8..b9c30ac 100644 --- a/docs/architecture/decisions/VD-07-diagnostics-and-telemetry-exporter.md +++ b/docs/architecture/decisions/VD-07-diagnostics-and-telemetry-exporter.md @@ -58,6 +58,20 @@ vendor 결정 전에는 안전한 기본값이 아니다. 7. queue full, invalid event/context, serialization과 sink failure는 제한된 reason bucket으로 집계한다. drop observer의 failure는 다시 telemetry를 발행하지 않는 nonrecursive 경계다. +7-1. telemetry adapter lifecycle은 `ACTIVE | DISPOSED` 둘뿐이다. `dispose()`는 + 한 번만 전이하고 `pagehide` listener 제거, queue 비우기, scheduled callback + generation 무효화, in-flight sink `AbortController` abort를 모두 수행한다. + dispose 뒤 `emit()`은 no-op이고 새 flush는 스케줄되지 않으며, abort를 무시한 + sink가 늦게 settle해도 post-dispose delivery state를 갱신하거나 재스케줄하지 + 못한다. 종료 중 drop telemetry를 재귀적으로 발행하지 않는다. +7-2. `flush()`는 active delivery promise를 join한다. 이미 진행 중인 flush가 + 있으면 같은 promise를 반환하므로 `await flush()`는 실제 settle을 뜻한다. +7-3. runtime `infrastructure.dispose()`는 diagnostics/state dependency를 파괴하기 + 전에 `telemetry.dispose()`를 먼저 호출한다. +7-4. queue/entry capacity는 construction-time 계약이다. `Number.isSafeInteger`가 + 아니거나 1 미만이거나 문서화된 ceiling(각각 `MAX_TELEMETRY_QUEUE`, + `MAX_DIAGNOSTIC_ENTRIES` = 10,000)을 넘으면 `TypeError`로 거절한다. NaN/Infinity가 + 조용히 eviction을 비활성화하는 경로를 남기지 않는다. 8. diagnostics와 telemetry failure는 제품 흐름, HTTP 결과, route transition, storage/cache fallback과 React error surface를 바꾸지 않는다. 9. mount 전 bootstrap failure는 안전한 build/config/error kind만 별도 evidence로 diff --git a/docs/operations/adapter-remediation-ledger.md b/docs/operations/adapter-remediation-ledger.md index fa9ad1b..6535b53 100644 --- a/docs/operations/adapter-remediation-ledger.md +++ b/docs/operations/adapter-remediation-ledger.md @@ -75,14 +75,14 @@ Rollout state starts at `NOT_STARTED`; documented-unimplemented items start at | N-01 | Live V3 path | `corepack pnpm exec vitest run tests/integration/http-execution-v3-observability.test.ts` | `fix: restore V3 HTTP observability` | `FIXED_NOT_RELEASED` | diagnostics/telemetry producer gate regression | Red 5/5 failed → green 5/5; `check:diagnostics` PASS (8 diagnostics, 5 telemetry producers); `check:types` PASS; `check:architecture` PASS | | N-02 | Live V3 path | `corepack pnpm exec vitest run tests/integration/http-execution-v3-auth-profile.test.ts` | `fix: enforce installed HTTP auth profiles` | `FIXED_NOT_RELEASED` | authenticated request 4xx spike after profile enforcement | Red suite failed to load (`installRestAuthProfileRegistry` absent) → green 7/7; `check:types` PASS; `check:architecture` PASS; `lint` PASS; unit+integration+features 1560 passed with only the pre-existing environmental `ci-artifact-contract` failures | | N-03 | Live V3 path | `corepack pnpm exec vitest run tests/unit/http-execution-v3.test.ts -t "retry-time fence"` | `fix: preserve command effect certainty across retries` | `FIXED_NOT_RELEASED` | command effect verdict regression | Red reproduced `SCOPE_FENCED` with `NOT_STARTED` after one dispatched attempt → green `MAYBE_APPLIED`; lattice table 9/9; `check:types` PASS; `lint` PASS | -| N-04 | Live composition teardown | `corepack pnpm exec vitest run tests/unit/telemetry.test.ts` | — | `NOT_STARTED` | telemetry delivery loss after teardown change | — | +| N-04 | Live composition teardown | `corepack pnpm exec vitest run tests/unit/telemetry.test.ts tests/unit/runtime-adapters.test.ts` | `fix: terminate telemetry work on disposal` | `FIXED_NOT_RELEASED` | telemetry delivery loss after teardown change | Red 10 failed (5 lifecycle + 5 capacity) → green 34/34; `check:diagnostics` PASS; `check:types` PASS; `check:architecture` PASS; `lint` PASS | | N-05 | Rollout blocker (sidecar not composed) | `corepack pnpm exec vitest run tests/unit/conditional-validator-store.test.ts` | — | `NOT_STARTED` | persisted validator key incompatibility | — | | N-06 | Legacy V2 rollback seam | `corepack pnpm exec vitest run tests/integration/http-client.test.ts` | — | `NOT_STARTED` | legacy keyed command rejection spike | — | | N-07 | Legacy V2 rollback seam | `corepack pnpm exec vitest run tests/integration/auth-recovery.test.ts` | — | `NOT_STARTED` | legacy credential timeout regression | — | | N-08 | Legacy V2 rollback seam | `corepack pnpm exec vitest run tests/unit/bounded-json-compatibility.test.ts` | — | `NOT_STARTED` | legacy JSON failure-code drift | — | | N-09 | Live cross-context host | `corepack pnpm exec vitest run tests/unit/cross-tab-invalidation.test.ts` | — | `NOT_STARTED` | cross-tab invalidation drop | — | | N-10 | Cursor runtime `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/cursor-pagination-runtime.test.ts` | — | `NOT_STARTED` | pagination abort semantics change | — | -| N-11 | Live composition | `corepack pnpm exec vitest run tests/unit/diagnostics.test.ts` | — | `NOT_STARTED` | capacity rejection on valid composition | — | +| N-11 | Live composition | `corepack pnpm exec vitest run tests/unit/telemetry.test.ts -t capacity` | `fix: terminate telemetry work on disposal` | `FIXED_NOT_RELEASED` | capacity rejection on valid composition | Red 5/5 capacity cases → green; ceilings documented in VD-07 §7-4 | ### Storage and browser files (`docs/reviews/adapters/03-storage-and-browser-files.md`) diff --git a/src/adapters/diagnostics/bounded-diagnostics.ts b/src/adapters/diagnostics/bounded-diagnostics.ts index c0b8e13..cec5cac 100644 --- a/src/adapters/diagnostics/bounded-diagnostics.ts +++ b/src/adapters/diagnostics/bounded-diagnostics.ts @@ -6,11 +6,15 @@ import { type DiagnosticRecordInput, } from "../../contracts/diagnostics.ts"; import { projectTelemetryEvent } from "../../contracts/telemetry.ts"; +import { assertBoundedCapacity } from "../telemetry/best-effort-telemetry.ts"; export const noOpDiagnostics: DiagnosticsPort = Object.freeze({ record() {}, }); +/** N-11. Documented absolute ceiling for bounded in-memory evidence. */ +export const MAX_DIAGNOSTIC_ENTRIES = 10_000; + export function createDiagnosticsAdapter( options: Readonly<{ maxEntries?: number; @@ -18,7 +22,11 @@ export function createDiagnosticsAdapter( sink?: (record: DiagnosticRecord) => void; }> = {}, ) { - const maxEntries = Math.max(1, options.maxEntries ?? 100); + const maxEntries = assertBoundedCapacity( + options.maxEntries ?? 100, + MAX_DIAGNOSTIC_ENTRIES, + "diagnostics maxEntries", + ); const entries: DiagnosticRecord[] = []; const droppedReasons = new Map(); diff --git a/src/adapters/telemetry/best-effort-telemetry.ts b/src/adapters/telemetry/best-effort-telemetry.ts index 933aaa3..01e8b4a 100644 --- a/src/adapters/telemetry/best-effort-telemetry.ts +++ b/src/adapters/telemetry/best-effort-telemetry.ts @@ -37,6 +37,31 @@ export const noOpTelemetry: TelemetryAdapter = Object.freeze({ dispose: () => {}, }); +/** + * N-04. Disposal is terminal: there is no durable queue and no resurrection. + */ +type TelemetryLifecycle = "ACTIVE" | "DISPOSED"; + +/** N-11. Documented absolute ceiling for the in-memory best-effort queue. */ +export const MAX_TELEMETRY_QUEUE = 10_000; + +/** + * N-11. A non-finite or fractional capacity silently disables eviction, so it is + * a construction-time configuration error rather than a runtime drop. + */ +export function assertBoundedCapacity( + value: number, + ceiling: number, + label: string, +): number { + if (!Number.isSafeInteger(value) || value < 1 || value > ceiling) { + throw new TypeError( + `${label} must be a safe integer between 1 and ${ceiling}`, + ); + } + return value; +} + export function createTelemetryAdapter( options: TelemetryAdapterOptions, ): TelemetryAdapter { @@ -46,11 +71,19 @@ export function createTelemetryAdapter( const endpoint = options.endpoint; const fetcher = options.fetcher ?? fetch; - const maxQueue = Math.max(1, options.maxQueue ?? 100); + const maxQueue = assertBoundedCapacity( + options.maxQueue ?? 100, + MAX_TELEMETRY_QUEUE, + "telemetry maxQueue", + ); const schedule = options.schedule ?? queueMicrotask; const queue: TelemetryEvent[] = []; + let lifecycleState: TelemetryLifecycle = "ACTIVE"; + /** Scheduled callbacks captured before disposal must not run afterwards. */ + let scheduleGeneration = 0; let scheduled = false; - let flushing = false; + let activeFlush: Promise | null = null; + let activeSink: AbortController | null = null; let dropped = 0; const dropReasons = new Map(); let lastDeliveryEvidence: TelemetryEvent | null = null; @@ -93,9 +126,12 @@ export function createTelemetryAdapter( } function scheduleFlush(): void { - if (scheduled) return; + if (scheduled || lifecycleState === "DISPOSED") return; scheduled = true; + const generation = scheduleGeneration; schedule(() => { + // A callback captured before disposal belongs to a dead generation. + if (generation !== scheduleGeneration) return; scheduled = false; void flush(); }); @@ -105,6 +141,7 @@ export function createTelemetryAdapter( eventName: TelemetryEventName, attributes: Record, ): void { + if (lifecycleState === "DISPOSED") return; const projected = projectTelemetryEvent( eventName, attributes, @@ -124,26 +161,44 @@ export function createTelemetryAdapter( scheduleFlush(); } - async function flush(): Promise { - if (flushing || queue.length === 0) return; - flushing = true; - const batch = queue.splice(0, queue.length); - try { - const response = await fetcher(endpoint, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ events: batch }), - keepalive: true, - }); - if (!response.ok) recordDrop("sink-failure", batch.length); - } catch { - recordDrop("sink-failure", batch.length); - } finally { - flushing = false; - if (queue.length > 0) { - scheduleFlush(); - } + /** + * `flush()` joins the active delivery instead of resolving immediately, so an + * awaited flush really means "the in-flight batch has settled". + */ + function flush(): Promise { + if (activeFlush) return activeFlush; + if (lifecycleState === "DISPOSED" || queue.length === 0) { + return Promise.resolve(); } + const generation = scheduleGeneration; + const run = async () => { + const batch = queue.splice(0, queue.length); + const controller = new AbortController(); + activeSink = controller; + try { + const response = await fetcher(endpoint, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ events: batch }), + keepalive: true, + signal: controller.signal, + }); + // A sink that ignored the abort must not update post-dispose state. + if (generation !== scheduleGeneration) return; + if (!response.ok) recordDrop("sink-failure", batch.length); + } catch { + if (generation !== scheduleGeneration) return; + recordDrop("sink-failure", batch.length); + } finally { + if (activeSink === controller) activeSink = null; + activeFlush = null; + if (generation === scheduleGeneration && queue.length > 0) { + scheduleFlush(); + } + } + }; + activeFlush = run(); + return activeFlush; } const flushBeforePageExit = () => { @@ -151,8 +206,20 @@ export function createTelemetryAdapter( }; lifecycle?.addEventListener("pagehide", flushBeforePageExit); + /** + * N-04. Terminal disposal: one state transition, no further admission, no + * further scheduling, and no recursive drop telemetry while shutting down. + */ function dispose(): void { + if (lifecycleState === "DISPOSED") return; + lifecycleState = "DISPOSED"; + scheduleGeneration += 1; + scheduled = false; lifecycle?.removeEventListener("pagehide", flushBeforePageExit); + queue.length = 0; + activeSink?.abort(); + activeSink = null; + activeFlush = null; } return Object.freeze({ diff --git a/src/bootstrap/runtime-adapters.ts b/src/bootstrap/runtime-adapters.ts index 30711df..c4aac8b 100644 --- a/src/bootstrap/runtime-adapters.ts +++ b/src/bootstrap/runtime-adapters.ts @@ -480,6 +480,10 @@ export async function createRuntimeAdapters( crossContextInvalidationStatus: () => serverStateGeneration.getSnapshot().crossContextStatus(), dispose() { + // N-04. Telemetry is torn down first: it must stop scheduling and + // delivering before the diagnostics and state dependencies it observes + // are destroyed. + telemetry.dispose(); conditionalValidators.clear(); serverStateScope.dispose(); serverStateGeneration.dispose(); diff --git a/tests/unit/runtime-adapters.test.ts b/tests/unit/runtime-adapters.test.ts index 5b06df1..28ac1a9 100644 --- a/tests/unit/runtime-adapters.test.ts +++ b/tests/unit/runtime-adapters.test.ts @@ -138,6 +138,38 @@ describe("runtime adapter composition", () => { adapters.infrastructure.dispose(); }); + it("runtime infrastructure disposal disposes telemetry first", async () => { + const fetcher = vi.fn(async () => new Response(null, { status: 204 })); + const adapters = await createRuntimeAdapters({ + runtime: { + ...runtime, + config: { + ...runtime.config, + TELEMETRY_ENABLED: true, + TELEMETRY_ENDPOINT: "https://telemetry.test/events", + }, + } as Runtime, + release, + host: {}, + fetcher: fetcher as unknown as typeof fetch, + }); + + adapters.outputPorts.telemetry.emit("api.request.failed", { + error_kind: "SERVER_FAILURE", + http_status_group: "5xx", + attempt_count_bucket: "1", + route_id: "TEST_ROUTE", + }); + expect(adapters.outputPorts.telemetry.pendingCount()).toBe(1); + + adapters.infrastructure.dispose(); + + expect(adapters.outputPorts.telemetry.pendingCount()).toBe(0); + await Promise.resolve(); + await Promise.resolve(); + expect(fetcher).not.toHaveBeenCalled(); + }); + it("replaces the QueryClient and coordinator for each session generation", async () => { const adapters = await createRuntimeAdapters({ runtime, release, host: {} }); const previousClient = adapters.infrastructure.queryClient; diff --git a/tests/unit/telemetry.test.ts b/tests/unit/telemetry.test.ts index 19652cb..95ed743 100644 --- a/tests/unit/telemetry.test.ts +++ b/tests/unit/telemetry.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from "vitest"; +import { createDiagnosticsAdapter } from "../../src/adapters/diagnostics/bounded-diagnostics.ts"; import { createTelemetryAdapter, safeTraceparent, @@ -236,4 +237,122 @@ describe("best-effort telemetry adapter", () => { expect(adapter.pendingCount()).toBe(0); expect(remove).toHaveBeenCalledWith("pagehide", expect.any(Function)); }); + + it("drops queued events and invalidates scheduled callbacks on dispose", async () => { + const callbacks: Array<() => void> = []; + const fetcher = vi.fn(async () => new Response(null, { status: 204 })); + const adapter = createTelemetryAdapter({ + enabled: true, + endpoint: "https://telemetry.test/events", + schedule: (callback) => callbacks.push(callback), + fetcher, + }); + adapter.emit("api.request.failed", validAttributes); + expect(adapter.pendingCount()).toBe(1); + + adapter.dispose(); + expect(adapter.pendingCount()).toBe(0); + + for (const callback of callbacks) callback(); + await Promise.resolve(); + expect(fetcher).not.toHaveBeenCalled(); + // Disposal is a silent shutdown, not a recursive drop event. + expect(adapter.dropReasons()).toEqual({}); + }); + + it("ignores emit after dispose", async () => { + const fetcher = vi.fn(async () => new Response(null, { status: 204 })); + const adapter = createTelemetryAdapter({ + enabled: true, + endpoint: "https://telemetry.test/events", + schedule: () => {}, + fetcher, + }); + adapter.dispose(); + adapter.emit("api.request.failed", validAttributes); + await adapter.flush(); + + expect(adapter.pendingCount()).toBe(0); + expect(fetcher).not.toHaveBeenCalled(); + }); + + it("aborts an in-flight sink and prevents post-dispose rescheduling", async () => { + const callbacks: Array<() => void> = []; + let observedSignal: AbortSignal | undefined; + let releaseSink: (() => void) | undefined; + const fetcher = vi.fn(async (_input: unknown, init?: RequestInit) => { + observedSignal = init?.signal ?? undefined; + await new Promise((resolve) => { + releaseSink = resolve; + }); + return new Response(null, { status: 204 }); + }); + const adapter = createTelemetryAdapter({ + enabled: true, + endpoint: "https://telemetry.test/events", + schedule: (callback) => callbacks.push(callback), + fetcher: fetcher as unknown as typeof fetch, + }); + adapter.emit("api.request.failed", validAttributes); + callbacks.splice(0).forEach((callback) => callback()); + await vi.waitFor(() => expect(fetcher).toHaveBeenCalledOnce()); + + adapter.dispose(); + expect(observedSignal?.aborted).toBe(true); + + // The sink ignored the abort and settles late. + releaseSink?.(); + await Promise.resolve(); + await Promise.resolve(); + expect(callbacks).toHaveLength(0); + expect(fetcher).toHaveBeenCalledOnce(); + }); + + it("joins an already active flush", async () => { + let releaseSink: (() => void) | undefined; + const fetcher = vi.fn(async () => { + await new Promise((resolve) => { + releaseSink = resolve; + }); + return new Response(null, { status: 204 }); + }); + const adapter = createTelemetryAdapter({ + enabled: true, + endpoint: "https://telemetry.test/events", + schedule: () => {}, + fetcher: fetcher as unknown as typeof fetch, + }); + adapter.emit("api.request.failed", validAttributes); + + const first = adapter.flush(); + await vi.waitFor(() => expect(fetcher).toHaveBeenCalledOnce()); + let secondSettled = false; + const second = adapter.flush().then(() => { + secondSettled = true; + }); + await Promise.resolve(); + expect(secondSettled).toBe(false); + + releaseSink?.(); + await first; + await second; + expect(secondSettled).toBe(true); + adapter.dispose(); + }); + + it.each([Number.NaN, Number.POSITIVE_INFINITY, 0, -1, 1.5])( + "rejects invalid telemetry and diagnostics capacity %s", + (value) => { + expect(() => + createTelemetryAdapter({ + enabled: true, + endpoint: "https://telemetry.test/events", + maxQueue: value, + }), + ).toThrow(TypeError); + expect(() => createDiagnosticsAdapter({ maxEntries: value })).toThrow( + TypeError, + ); + }, + ); });