diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 688e15d..b9bfed9 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -221,6 +221,7 @@ When `AUTTER_SINK_URL` is set, each ingest batch POSTs: ```json { "version": 1, + "batchId": "5f0c9e7a-…", "orgId": "...", "repositoryId": "...", "occurrences": [ @@ -249,5 +250,42 @@ Batches also carry `metrics` (1-minute usage rollup points) and `llmCalls` `userId`, `status`, `startedAt`) whenever the ingest produced them — same shapes as their ClickHouse rows, additive to the v1 payload. -Delivery is best-effort fire-and-forget (the ingester is not a queue); the -consumer should treat ClickHouse as the recovery source for missed batches. +### Delivery semantics + +Delivery is **at-least-once within a process lifetime**: batches queue in +memory and retry with exponential backoff (1 s → 60 s, `SINK_MAX_ATTEMPTS` +tries, ~8 min by default) on network errors, timeouts, 408/429, and 5xx. +Other 4xx responses mean the consumer rejected the batch — those drop +immediately and are logged. The retry buffer is bounded +(`SINK_MAX_BUFFERED_BATCHES` / `SINK_MAX_BUFFERED_MB`); on overflow the +oldest batch of the tenant holding the most buffered bytes drops first — +one flooding org cannot evict everyone else — and every drop is logged +with its signal time range. A single batch larger than the whole buffer +is dropped alone rather than flushing the queue. Retrying batches keep +their enqueue-age position, so eviction order stays oldest-first even +under sustained failure. + +Consequences for consumers: + +- **Deduplicate on `batchId`** (and per-occurrence on `occurrenceId`): + a batch can arrive more than once — e.g. the consumer processed it but + the 2xx response was lost, so the ingester retried. +- **`occurrenceId` is content-derived, not random.** An OTLP exporter + that retries an export (after a 503 from a partially-failed ClickHouse + write, or a lost 2xx) reproduces the same ids, so per-occurrence dedupe + holds across transport retries too — and duplicated ClickHouse rows + share an id, so replays and row counts should use distinct ids. +- **ClickHouse is the recovery source.** Every forwarded signal was + written to ClickHouse before it was queued (ingest returns 503 + otherwise), so a crashed ingester, an exhausted retry budget, or a + buffer overflow never loses data — the consumer reconciles by replaying + the affected time range from `runtime_error_occurrences` / + `runtime_metrics_1m`. Occurrence rows carry the same `occurrence_id` + the sink payload does, so replays deduplicate exactly. +- `/healthz` exposes delivery counters (`sink.queued`, `sink.delivered`, + `sink.retried`, `sink.droppedOverflow`, `sink.droppedPermanent`, + `sink.lastFailureAt`, …) for missed-batch monitoring and alerting. + Failure detail is a fixed category (`sink.lastFailureReason`: + `timeout`, `connection_error`, `http_`, `error`) — raw + transport errors stay in server logs, never in the unauthenticated + health response. diff --git a/docs/PLAN.md b/docs/PLAN.md index f679d30..5bb97c8 100644 --- a/docs/PLAN.md +++ b/docs/PLAN.md @@ -31,7 +31,9 @@ Self-hostable ingest service, `packages/otlp-ingester`. with a 60-second in-process cache. - **Sink webhook** (optional): fingerprinted occurrences are forwarded to `AUTTER_SINK_URL` so a backend can do issue grouping/alerting in Postgres. - The ingester itself only writes ClickHouse. + The ingester itself only writes ClickHouse. Delivery is at-least-once + (in-memory retry buffer, `batchId` for consumer dedupe); ClickHouse + remains the replay source for anything the buffer cannot save. - Payload cap (default 1 MB), per-key fixed-window rate limit, graceful degrade when ClickHouse is unreachable (503 on ingest, never crash). @@ -128,4 +130,5 @@ initialisation in the Node SDKs: | `/v1/traces`, `/v1/metrics` OTLP/HTTP | OTLP spec-stable | | `/v1/browser` payload (`version: 1`) | additive-only changes | | ClickHouse table schemas | additive-only; TTLs configurable via env | -| Sink webhook payload (`version: 1`) | additive-only changes (`llmCalls` added additively) | +| Sink webhook payload (`version: 1`) | additive-only (`llmCalls`, `batchId`) | +| Sink webhook delivery | at-least-once; dedupe on `batchId`/`occurrenceId` | diff --git a/examples/express-app/package.json b/examples/express-app/package.json index 46812aa..de6be65 100644 --- a/examples/express-app/package.json +++ b/examples/express-app/package.json @@ -4,6 +4,7 @@ "version": "0.0.0", "type": "module", "scripts": { + "build": "node --check server.js", "prestart": "mkdir -p public/sdk && cp ../../packages/runtime-browser/dist/index.js public/sdk/index.js", "start": "node server.js" }, diff --git a/packages/otlp-ingester/README.md b/packages/otlp-ingester/README.md index 63ae958..ad1e4c1 100644 --- a/packages/otlp-ingester/README.md +++ b/packages/otlp-ingester/README.md @@ -54,8 +54,11 @@ The validator webhook may return the same extra fields: | `AUTTER_INGEST_KEYS` | — | JSON: `[{"key":"...","orgId":"...","repositoryId":"..."}]` | | `AUTTER_KEY_VALIDATOR_URL` | — | Webhook: `POST {key}` → `{orgId, repositoryId}` (60 s cache) | | `AUTTER_KEY_VALIDATOR_TOKEN` | — | Bearer token sent to the validator | -| `AUTTER_SINK_URL` | — | Webhook receiving fingerprinted occurrences for issue grouping | +| `AUTTER_SINK_URL` | — | Issue-grouping webhook; at-least-once (`docs/ARCHITECTURE.md`) | | `AUTTER_SINK_TOKEN` | — | Bearer token sent to the sink | +| `SINK_MAX_ATTEMPTS` | `12` | Delivery attempts per batch (1–60 s backoff) | +| `SINK_MAX_BUFFERED_BATCHES` | `1000` | Retry buffer cap; oldest drops are logged | +| `SINK_MAX_BUFFERED_MB` | `64` | Sink retry buffer cap (memory) | | `MAX_BODY_BYTES` | `1048576` | Request body cap | | `RATE_LIMIT_PER_MINUTE` | `300` | Per-key fixed window (server keys) | | `CLIENT_RATE_LIMIT_PER_MINUTE` | `120` | Per-key fixed window (client keys) | diff --git a/packages/otlp-ingester/src/config.ts b/packages/otlp-ingester/src/config.ts index 0f2230d..45d1d5c 100644 --- a/packages/otlp-ingester/src/config.ts +++ b/packages/otlp-ingester/src/config.ts @@ -23,6 +23,11 @@ export interface IngesterConfig { /** Optional webhook receiving fingerprinted occurrences for issue grouping. */ sinkUrl: string | null; sinkToken: string | null; + /** Sink delivery attempts per batch before giving up (backoff-capped ~8 min). */ + sinkMaxAttempts: number; + /** Bounds for the in-memory sink retry buffer; oldest batches drop first. */ + sinkMaxBufferedBatches: number; + sinkMaxBufferedMb: number; maxBodyBytes: number; /** Per-key requests per minute (server keys). */ rateLimitPerMinute: number; @@ -72,6 +77,11 @@ export function loadConfig(): IngesterConfig { keyValidatorToken: process.env.AUTTER_KEY_VALIDATOR_TOKEN || null, sinkUrl: process.env.AUTTER_SINK_URL || null, sinkToken: process.env.AUTTER_SINK_TOKEN || null, + // 12 attempts with 1s..60s exponential backoff spans ~8 minutes — long + // enough to ride out a routine consumer deploy without unbounded memory. + sinkMaxAttempts: intEnv("SINK_MAX_ATTEMPTS", 12), + sinkMaxBufferedBatches: intEnv("SINK_MAX_BUFFERED_BATCHES", 1000), + sinkMaxBufferedMb: intEnv("SINK_MAX_BUFFERED_MB", 64), maxBodyBytes: intEnv("MAX_BODY_BYTES", 1024 * 1024), rateLimitPerMinute: intEnv("RATE_LIMIT_PER_MINUTE", 300), clientRateLimitPerMinute: intEnv("CLIENT_RATE_LIMIT_PER_MINUTE", 120), diff --git a/packages/otlp-ingester/src/fingerprint.test.ts b/packages/otlp-ingester/src/fingerprint.test.ts new file mode 100644 index 0000000..dab28c1 --- /dev/null +++ b/packages/otlp-ingester/src/fingerprint.test.ts @@ -0,0 +1,70 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { fingerprintOccurrence, occurrenceIdFor } from "./fingerprint.js"; +import type { RuntimeOccurrenceInput } from "./types.js"; + +function input( + overrides: Partial = {}, +): RuntimeOccurrenceInput { + return { + source: "server", + severity: "error", + service: "svc", + environment: "prod", + release: null, + errorType: "TypeError", + message: "user 42 not found", + stack: null, + route: "/users/42", + method: "GET", + statusCode: 404, + traceId: "trace-1", + sessionId: null, + attributes: null, + occurredAt: new Date("2026-01-01T00:00:00.000Z"), + ...overrides, + }; +} + +test("occurrenceIdFor is a pure function of the signal", () => { + const scope = { orgId: "org", repositoryId: "repo" }; + const a = occurrenceIdFor(scope, input(), "fp", 0); + // Same signal (an exporter retry of the same batch) → same id. + assert.equal(a, occurrenceIdFor(scope, input(), "fp", 0)); + assert.match(a, /^[0-9a-f]{32}$/); +}); + +test("occurrenceIdFor separates distinct signals", () => { + const scope = { orgId: "org", repositoryId: "repo" }; + const a = occurrenceIdFor(scope, input(), "fp", 0); + // Different batch position (identical twin events in one batch). + assert.notEqual(a, occurrenceIdFor(scope, input(), "fp", 1)); + // Different millisecond. + assert.notEqual( + a, + occurrenceIdFor( + scope, + input({ occurredAt: new Date("2026-01-01T00:00:00.001Z") }), + "fp", + 0, + ), + ); + // Different trace. + assert.notEqual( + a, + occurrenceIdFor(scope, input({ traceId: "trace-2" }), "fp", 0), + ); + // Different tenant. + assert.notEqual( + a, + occurrenceIdFor({ orgId: "org2", repositoryId: "repo" }, input(), "fp", 0), + ); +}); + +test("fingerprint groups per-value message variants into one issue", () => { + const a = fingerprintOccurrence(input({ message: "user 42 not found" })); + const b = fingerprintOccurrence(input({ message: "user 7 not found" })); + assert.equal(a, b); + const c = fingerprintOccurrence(input({ errorType: "RangeError" })); + assert.notEqual(a, c); +}); diff --git a/packages/otlp-ingester/src/fingerprint.ts b/packages/otlp-ingester/src/fingerprint.ts index 40b15f7..831d0b8 100644 --- a/packages/otlp-ingester/src/fingerprint.ts +++ b/packages/otlp-ingester/src/fingerprint.ts @@ -93,6 +93,48 @@ export function fingerprintOccurrence(input: RuntimeOccurrenceInput): string { return createHash("sha256").update(parts.join(" ")).digest("hex").slice(0, 32); } +/** + * Deterministic per-occurrence identity (as opposed to the fingerprint, + * which is the per-ISSUE identity shared by every occurrence of a defect). + * + * The id must be a pure function of the signal, not a fresh UUID per + * request: OTLP exporters retry whole batches (after a 503 from a partial + * ClickHouse write, or when only the 2xx was lost), and both the ClickHouse + * rows and the sink consumer's dedupe ledger key on this id — random ids + * would turn every transport retry into a duplicate error downstream. + * + * Identical signals within one batch stay distinct through their batch + * position, which is stable across retries because exporters re-send the + * same serialized batch. Residual coalescing risk: two occurrences from + * DIFFERENT requests that share the same millisecond, message, and batch + * position while carrying neither a traceId nor a sessionId — accepted, as + * server signals virtually always carry a traceId and browser signals a + * sessionId. + */ +export function occurrenceIdFor( + scope: { orgId: string; repositoryId: string }, + input: RuntimeOccurrenceInput, + fingerprint: string, + batchIndex: number, +): string { + const parts = [ + "v1", + scope.orgId, + scope.repositoryId, + fingerprint, + String(input.occurredAt.getTime()), + input.traceId ?? "", + input.sessionId ?? "", + input.message.slice(0, 1000), + String(batchIndex), + ]; + // NUL-joined so a free-text field can never bleed into its neighbour. + return createHash("sha256") + .update(parts.join("\u0000")) + .digest("hex") + .slice(0, 32); +} + /** * Derived, aggregation-ready fields, computed from the SAME normalisers the * fingerprint hashes — so a stored fingerprint can always be explained by diff --git a/packages/otlp-ingester/src/index.ts b/packages/otlp-ingester/src/index.ts index 8505980..f5fee12 100644 --- a/packages/otlp-ingester/src/index.ts +++ b/packages/otlp-ingester/src/index.ts @@ -2,7 +2,7 @@ import { loadConfig } from "./config.js"; import { createIngesterApp } from "./server.js"; const config = loadConfig(); -const { app, store } = createIngesterApp(config); +const { app, store, sink } = createIngesterApp(config); const server = app.listen(config.port, () => { console.log( @@ -23,6 +23,17 @@ if (store.configured) { async function shutdown(signal: string) { console.log(`${signal} received, shutting down`); + if (sink) { + const pending = sink.pendingCount(); + sink.stop(); + if (pending > 0) { + // The retry buffer is memory-only; everything in it is already in + // ClickHouse, so the consumer's reconciliation replays it. + console.warn( + `${pending} sink batch(es) undelivered at shutdown — recoverable via ClickHouse replay`, + ); + } + } server.close(() => { void store.close().finally(() => process.exit(0)); }); @@ -34,4 +45,5 @@ process.on("SIGINT", () => void shutdown("SIGINT")); export { createIngesterApp } from "./server.js"; export { loadConfig } from "./config.js"; +export { SinkForwarder, type SinkStats, type SinkTuning } from "./sink.js"; export * from "./types.js"; diff --git a/packages/otlp-ingester/src/server.ts b/packages/otlp-ingester/src/server.ts index 4f24ed6..b4842df 100644 --- a/packages/otlp-ingester/src/server.ts +++ b/packages/otlp-ingester/src/server.ts @@ -1,4 +1,3 @@ -import { randomUUID } from "node:crypto"; import express, { type Express, type Request, @@ -7,7 +6,11 @@ import express, { import { KeyResolver, RateLimiter } from "./auth.js"; import { ClickHouseStore } from "./clickhouse.js"; import type { IngesterConfig } from "./config.js"; -import { deriveFields, fingerprintOccurrence } from "./fingerprint.js"; +import { + deriveFields, + fingerprintOccurrence, + occurrenceIdFor, +} from "./fingerprint.js"; import { browserPayloadSchema, normalizeBrowserPayload, @@ -19,10 +22,9 @@ import { type OtlpTraceRequest, } from "./normalize-otlp.js"; import { decodeMetricsRequest, decodeTraceRequest } from "./otlp-proto.js"; +import { SinkForwarder } from "./sink.js"; import type { IngestContext, - RuntimeLlmCall, - RuntimeMetricPoint, RuntimeOccurrence, RuntimeOccurrenceInput, } from "./types.js"; @@ -30,10 +32,16 @@ import type { export interface IngesterApp { app: Express; store: ClickHouseStore; + /** Present when AUTTER_SINK_URL is configured. */ + sink: SinkForwarder | null; } export function createIngesterApp(config: IngesterConfig): IngesterApp { const store = new ClickHouseStore(config); + // Fingerprinted occurrences feed the consumer's issue grouping, metric + // points feed the request/error-rate rollups, LLM calls feed spend + // watching. Delivery is at-least-once with bounded retries — see sink.ts. + const sink = config.sinkUrl ? new SinkForwarder(config) : null; const keys = new KeyResolver(config); const serverRateLimiter = new RateLimiter(config.rateLimitPerMinute); const clientRateLimiter = new RateLimiter(config.clientRateLimitPerMinute); @@ -84,15 +92,18 @@ export function createIngesterApp(config: IngesterConfig): IngesterApp { }); app.get("/healthz", async (_req, res) => { + const sinkStats = sink ? { sink: sink.stats() } : {}; if (!store.configured) { - res.status(200).json({ ok: true, clickhouse: "unconfigured" }); + res.status(200).json({ ok: true, clickhouse: "unconfigured", ...sinkStats }); return; } try { const ok = await store.ping(); - res.status(ok ? 200 : 503).json({ ok, clickhouse: ok ? "up" : "down" }); + res + .status(ok ? 200 : 503) + .json({ ok, clickhouse: ok ? "up" : "down", ...sinkStats }); } catch { - res.status(503).json({ ok: false, clickhouse: "down" }); + res.status(503).json({ ok: false, clickhouse: "down", ...sinkStats }); } }); @@ -148,64 +159,23 @@ export function createIngesterApp(config: IngesterConfig): IngesterApp { return ctx; } + /** Ids are content-derived (occurrenceIdFor), NOT random: an exporter + * that retries a batch — after a 503 from a partially-failed ClickHouse + * write, or when only our 2xx got lost — must produce the same ids, so + * the sink consumer's per-occurrence dedupe holds across transport + * retries and duplicated ClickHouse rows stay identifiable. */ function fingerprintAll( + ctx: IngestContext, inputs: RuntimeOccurrenceInput[], ): RuntimeOccurrence[] { - return inputs.map((input) => ({ - ...input, - occurrenceId: randomUUID(), - fingerprint: fingerprintOccurrence(input), - ...deriveFields(input), - })); - } - - /** - * Best-effort forward for the cloud dashboard: fingerprinted occurrences - * feed issue grouping, metric points feed the request/error-rate rollups, - * LLM calls feed spend/anomaly watching. - */ - function forwardToSink( - ctx: IngestContext, - occurrences: RuntimeOccurrence[], - metricPoints: RuntimeMetricPoint[] = [], - llmCalls: RuntimeLlmCall[] = [], - ) { - if (!config.sinkUrl) return; - if ( - occurrences.length === 0 && - metricPoints.length === 0 && - llmCalls.length === 0 - ) { - return; - } - void fetch(config.sinkUrl, { - method: "POST", - headers: { - "content-type": "application/json", - ...(config.sinkToken - ? { authorization: `Bearer ${config.sinkToken}` } - : {}), - }, - body: JSON.stringify({ - version: 1, - orgId: ctx.orgId, - repositoryId: ctx.repositoryId, - occurrences: occurrences.map((o) => ({ - ...o, - occurredAt: o.occurredAt.toISOString(), - })), - metrics: metricPoints.map((p) => ({ - ...p, - bucketAt: p.bucketAt.toISOString(), - })), - llmCalls: llmCalls.map((c) => ({ - ...c, - startedAt: c.startedAt.toISOString(), - })), - }), - signal: AbortSignal.timeout(10_000), - }).catch((err) => { - console.warn("sink forward failed (non-fatal):", err?.message ?? err); + return inputs.map((input, index) => { + const fingerprint = fingerprintOccurrence(input); + return { + ...input, + occurrenceId: occurrenceIdFor(ctx, input, fingerprint, index), + fingerprint, + ...deriveFields(input), + }; }); } @@ -240,7 +210,15 @@ export function createIngesterApp(config: IngesterConfig): IngesterApp { } const { occurrences, spans, metricPoints, llmCalls } = normalizeTraces(request); - const fingerprinted = fingerprintAll(occurrences); + const fingerprinted = fingerprintAll(ctx, occurrences); + // ClickHouse has no cross-table transaction, so these four inserts can + // partially commit. Recovery boundary: any failure → 503 → the exporter + // retries the whole batch. Deterministic occurrence ids make the retry + // idempotent downstream (consumer dedupes per id; duplicate ClickHouse + // rows share an id, and the consumer's reconciler counts distinct ids), + // and nothing reaches the sink queue unless every insert succeeded — + // signals persisted by a partial write are picked up by the consumer's + // ClickHouse reconciliation instead. try { await Promise.all([ store.insertOccurrences(ctx, fingerprinted), @@ -252,7 +230,7 @@ export function createIngesterApp(config: IngesterConfig): IngesterApp { storageError(res, err); return; } - forwardToSink(ctx, fingerprinted, metricPoints, llmCalls); + sink?.enqueue(ctx, fingerprinted, metricPoints, llmCalls); otlpSuccess(req, res); }); @@ -277,7 +255,7 @@ export function createIngesterApp(config: IngesterConfig): IngesterApp { storageError(res, err); return; } - forwardToSink(ctx, [], metricPoints); + sink?.enqueue(ctx, [], metricPoints); otlpSuccess(req, res); }); @@ -303,7 +281,7 @@ export function createIngesterApp(config: IngesterConfig): IngesterApp { return; } const { occurrences, metricPoints } = normalizeBrowserPayload(parsed.data); - const fingerprinted = fingerprintAll(occurrences); + const fingerprinted = fingerprintAll(ctx, occurrences); try { await Promise.all([ store.insertOccurrences(ctx, fingerprinted), @@ -313,7 +291,7 @@ export function createIngesterApp(config: IngesterConfig): IngesterApp { storageError(res, err); return; } - forwardToSink(ctx, fingerprinted, metricPoints); + sink?.enqueue(ctx, fingerprinted, metricPoints); res.status(202).json({ accepted: fingerprinted.length }); }); @@ -339,5 +317,5 @@ export function createIngesterApp(config: IngesterConfig): IngesterApp { }, ); - return { app, store }; + return { app, store, sink }; } diff --git a/packages/otlp-ingester/src/sink.test.ts b/packages/otlp-ingester/src/sink.test.ts new file mode 100644 index 0000000..5367de7 --- /dev/null +++ b/packages/otlp-ingester/src/sink.test.ts @@ -0,0 +1,293 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import type { IngesterConfig } from "./config.js"; +import { SinkForwarder, type SinkTuning } from "./sink.js"; +import type { IngestContext, RuntimeOccurrence } from "./types.js"; + +/** + * SinkForwarder unit tests — fake fetch, no network, no ClickHouse. Timing + * is compressed through the tuning seam so retry/backoff paths run in + * milliseconds. + */ + +const TUNING: Partial = { + retryBaseMs: 4, + retryCapMs: 8, + requestTimeoutMs: 500, + healthyConcurrency: 4, +}; + +function cfg(overrides: Partial = {}): IngesterConfig { + return { + port: 0, + clickhouseUrl: null, + clickhouseUser: "default", + clickhousePassword: "", + clickhouseDatabase: "autter_runtime", + ingestKeys: [], + keyValidatorUrl: null, + keyValidatorToken: null, + sinkUrl: "http://sink.local/hook", + sinkToken: "sink-token", + sinkMaxAttempts: 5, + sinkMaxBufferedBatches: 1000, + sinkMaxBufferedMb: 64, + maxBodyBytes: 1_048_576, + rateLimitPerMinute: 300, + clientRateLimitPerMinute: 120, + occurrenceTtlDays: 14, + spanTtlDays: 7, + metricsTtlDays: 90, + llmCallTtlDays: 90, + ...overrides, + }; +} + +function ctx(orgId = "org-a"): IngestContext { + return { orgId, repositoryId: "repo-1", scope: "server", allowedOrigins: [] }; +} + +function occ(overrides: Partial = {}): RuntimeOccurrence { + return { + source: "server", + severity: "error", + service: "svc", + environment: "prod", + release: null, + errorType: "TypeError", + message: "boom", + stack: null, + route: "/x", + method: "GET", + statusCode: 500, + traceId: "trace-1", + sessionId: null, + attributes: null, + occurredAt: new Date("2026-01-01T00:00:00.000Z"), + occurrenceId: "occ-1", + fingerprint: "fp-1", + routeNormalized: "/x", + messageNormalized: "boom", + topFrames: [], + firstFrame: "", + ...overrides, + }; +} + +interface Deferred { + promise: Promise; + resolve: (value: T) => void; + reject: (err: unknown) => void; +} + +function deferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (err: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +const ok = () => new Response("", { status: 200 }); + +async function until(cond: () => boolean, ms = 2000): Promise { + const deadline = Date.now() + ms; + while (!cond()) { + if (Date.now() > deadline) throw new Error("condition not met in time"); + await new Promise((r) => setTimeout(r, 2)); + } +} + +test("delivers a batch with batchId, bearer auth, and ISO timestamps", async () => { + const calls: Array<{ url: string; init: RequestInit }> = []; + const fetchImpl = (async (url: unknown, init?: RequestInit) => { + calls.push({ url: String(url), init: init ?? {} }); + return ok(); + }) as typeof fetch; + const sink = new SinkForwarder(cfg(), fetchImpl, TUNING); + + sink.enqueue(ctx(), [occ()]); + await until(() => sink.stats().delivered === 1); + + assert.equal(calls.length, 1); + assert.equal(calls[0]?.url, "http://sink.local/hook"); + const headers = calls[0]?.init.headers as Record; + assert.equal(headers.authorization, "Bearer sink-token"); + const body = JSON.parse(String(calls[0]?.init.body)); + assert.equal(body.version, 1); + assert.match(body.batchId, /^[0-9a-f-]{36}$/); + assert.equal(body.orgId, "org-a"); + assert.equal(body.repositoryId, "repo-1"); + assert.equal(body.occurrences[0].occurredAt, "2026-01-01T00:00:00.000Z"); + assert.equal(body.occurrences[0].occurrenceId, "occ-1"); +}); + +test("retries 5xx and reports a sanitized http_ reason", async (t) => { + t.mock.method(console, "warn", () => {}); + let attempts = 0; + const fetchImpl = (async () => { + attempts += 1; + return attempts === 1 ? new Response("", { status: 503 }) : ok(); + }) as typeof fetch; + const sink = new SinkForwarder(cfg(), fetchImpl, TUNING); + + sink.enqueue(ctx(), [occ()]); + await until(() => sink.stats().delivered === 1); + + const stats = sink.stats(); + assert.equal(attempts, 2); + assert.equal(stats.retried, 1); + assert.equal(stats.consecutiveFailures, 0); + assert.equal(stats.lastFailureReason, "http_503"); + assert.ok(!JSON.stringify(stats).includes("sink responded")); +}); + +test("non-retryable 4xx drops immediately without retry", async (t) => { + t.mock.method(console, "error", () => {}); + let attempts = 0; + const fetchImpl = (async () => { + attempts += 1; + return new Response("", { status: 400 }); + }) as typeof fetch; + const sink = new SinkForwarder(cfg(), fetchImpl, TUNING); + + sink.enqueue(ctx(), [occ()]); + await until(() => sink.stats().droppedPermanent === 1); + + assert.equal(attempts, 1); + assert.equal(sink.stats().delivered, 0); + assert.equal(sink.stats().lastFailureReason, "http_400"); +}); + +test("network failures give up after sinkMaxAttempts with a safe category", async (t) => { + t.mock.method(console, "warn", () => {}); + t.mock.method(console, "error", () => {}); + let attempts = 0; + const fetchImpl = (async () => { + attempts += 1; + throw new TypeError("fetch failed: super secret internal detail"); + }) as typeof fetch; + const sink = new SinkForwarder(cfg({ sinkMaxAttempts: 2 }), fetchImpl, TUNING); + + sink.enqueue(ctx(), [occ()]); + await until(() => sink.stats().droppedPermanent === 1); + + const stats = sink.stats(); + assert.equal(attempts, 2); + assert.equal(stats.lastFailureReason, "connection_error"); + // Raw exception text must never surface in stats (/healthz is public). + assert.ok(!JSON.stringify(stats).includes("super secret")); +}); + +test("a concurrent stale success does not clear an active failure episode", async (t) => { + t.mock.method(console, "warn", () => {}); + const pending: Deferred[] = []; + const fetchImpl = (() => { + const d = deferred(); + pending.push(d); + return d.promise; + }) as typeof fetch; + const sink = new SinkForwarder(cfg(), fetchImpl, TUNING); + + // Two batches go in flight concurrently (healthy concurrency). + sink.enqueue(ctx(), [occ({ message: "first" })]); + sink.enqueue(ctx(), [occ({ message: "second" })]); + await until(() => pending.length === 2); + + // First request fails — the forwarder enters its failure episode. + pending[0]?.reject(new TypeError("down")); + await until(() => sink.stats().consecutiveFailures === 1); + + // Second request succeeds, but it STARTED before the failure: it must + // not reset the failure state and reopen full concurrency. + pending[1]?.resolve(ok()); + await until(() => sink.stats().delivered === 1); + assert.equal(sink.stats().consecutiveFailures, 1); + + // The failed batch retries (serially) and succeeds — now it resets. + await until(() => pending.length === 3); + pending[2]?.resolve(ok()); + await until(() => sink.stats().delivered === 2); + assert.equal(sink.stats().consecutiveFailures, 0); +}); + +test("overflow evicts the heaviest org's oldest batch, sparing small tenants", async (t) => { + const warn = t.mock.method(console, "warn", () => {}); + // Never resolves: four org-a batches occupy the in-flight slots, the + // rest sit in the queue where eviction applies. + const fetchImpl = (() => deferred().promise) as typeof fetch; + const sink = new SinkForwarder( + cfg({ sinkMaxBufferedBatches: 2 }), + fetchImpl, + TUNING, + ); + + for (let i = 0; i < 6; i += 1) { + sink.enqueue(ctx("org-a"), [ + occ({ occurredAt: new Date(Date.UTC(2026, 0, 1, 0, i)) }), + ]); + } + // 4 in flight + 2 queued for org-a; org-b's batch overflows the queue. + sink.enqueue(ctx("org-b"), [occ()]); + + assert.equal(sink.stats().droppedOverflow, 1); + const warned = warn.mock.calls.map((c) => c.arguments.join(" ")).join("\n"); + assert.ok(warned.includes("org org-a"), warned); + assert.ok(!warned.includes("org org-b"), warned); +}); + +test("a retrying batch keeps its enqueue-age position for eviction", async (t) => { + const warn = t.mock.method(console, "warn", () => {}); + t.mock.method(console, "error", () => {}); + const fetchImpl = (async () => { + throw new TypeError("down"); + }) as typeof fetch; + const sink = new SinkForwarder( + cfg({ sinkMaxBufferedBatches: 1, sinkMaxAttempts: 10 }), + fetchImpl, + { ...TUNING, retryBaseMs: 60, retryCapMs: 120 }, + ); + + sink.enqueue(ctx(), [occ({ occurredAt: new Date("2026-01-01T00:00:00Z") })]); + await until(() => sink.stats().retried >= 1); + + // The older (failed, backing-off) batch must be the eviction victim — + // not pushed behind this newer one. + sink.enqueue(ctx(), [occ({ occurredAt: new Date("2026-02-02T00:00:00Z") })]); + await until(() => sink.stats().droppedOverflow === 1); + + const warned = warn.mock.calls.map((c) => c.arguments.join(" ")).join("\n"); + assert.ok(warned.includes("2026-01-01"), warned); + assert.ok(!warned.includes("2026-02-02"), warned); +}); + +test("a single batch larger than the whole buffer is dropped alone", async (t) => { + const warn = t.mock.method(console, "warn", () => {}); + let called = false; + const fetchImpl = (async () => { + called = true; + return ok(); + }) as typeof fetch; + const sink = new SinkForwarder(cfg({ sinkMaxBufferedMb: 1 }), fetchImpl, TUNING); + + sink.enqueue(ctx(), [occ({ message: "x".repeat(2 * 1024 * 1024) })]); + + assert.equal(sink.stats().droppedOverflow, 1); + assert.equal(sink.stats().queued, 0); + assert.equal(called, false); + const warned = warn.mock.calls.map((c) => c.arguments.join(" ")).join("\n"); + assert.ok(warned.includes("exceeds the buffer cap"), warned); +}); + +test("enqueue after stop is a no-op and pendingCount reflects the queue", async () => { + const fetchImpl = (() => deferred().promise) as typeof fetch; + const sink = new SinkForwarder(cfg(), fetchImpl, TUNING); + + sink.enqueue(ctx(), [occ()]); + assert.equal(sink.pendingCount(), 1); + sink.stop(); + sink.enqueue(ctx(), [occ()]); + assert.equal(sink.pendingCount(), 1); +}); diff --git a/packages/otlp-ingester/src/sink.ts b/packages/otlp-ingester/src/sink.ts new file mode 100644 index 0000000..eac21c6 --- /dev/null +++ b/packages/otlp-ingester/src/sink.ts @@ -0,0 +1,431 @@ +import { randomUUID } from "node:crypto"; +import type { IngesterConfig } from "./config.js"; +import type { + IngestContext, + RuntimeLlmCall, + RuntimeMetricPoint, + RuntimeOccurrence, +} from "./types.js"; + +/** + * At-least-once delivery to the sink webhook. + * + * The sink feeds the consumer's issue grouping and incident detection, so a + * lost batch means silently missing error occurrences — a single + * fire-and-forget POST is not enough (a routine consumer deploy is longer + * than one request timeout). Batches therefore queue in memory and retry + * with exponential backoff until delivered, permanently rejected, or the + * bounded buffer overflows. + * + * Durability boundary: the queue is in-memory only. Everything forwarded + * here was already written to ClickHouse (ingest 503s otherwise), so after + * a process crash or an overflow/permanent drop the consumer recovers by + * replaying the logged time range from ClickHouse — see + * docs/ARCHITECTURE.md "Sink webhook". Every batch carries a unique + * `batchId` so consumers can deduplicate retried deliveries. + */ + +/** Delivery timing/concurrency knobs; overridable for tests. */ +export interface SinkTuning { + /** Base delay before the second attempt; doubles per attempt to the cap. */ + retryBaseMs: number; + retryCapMs: number; + requestTimeoutMs: number; + /** Deliveries run concurrently while healthy, serially while failing. */ + healthyConcurrency: number; +} + +const DEFAULT_TUNING: SinkTuning = { + retryBaseMs: 1000, + retryCapMs: 60_000, + requestTimeoutMs: 10_000, + healthyConcurrency: 4, +}; + +interface QueuedBatch { + batchId: string; + /** Tenant that produced the batch — buffer pressure is charged per org. */ + orgId: string; + body: string; + bytes: number; + /** Enqueue order; the queue stays sorted by this so "oldest" is queue[0]. */ + seq: number; + /** Delivery attempts made so far. */ + attempts: number; + nextAttemptAt: number; + /** ISO range of the signals inside — the replay hint when dropped. */ + signalsFrom: string | null; + signalsTo: string | null; + enqueuedAt: number; +} + +/** + * Operational counters for /healthz. Failure detail is reduced to a fixed + * category (`timeout`, `connection_error`, `http_`, `error`) — the + * health endpoint is unauthenticated, so raw transport/exception text stays + * in server-side logs only. + */ +export interface SinkStats { + queued: number; + queuedBytes: number; + inFlight: number; + delivered: number; + retried: number; + droppedOverflow: number; + droppedPermanent: number; + consecutiveFailures: number; + lastFailureAt: string | null; + lastFailureReason: string | null; + oldestQueuedSince: string | null; +} + +/** Fixed failure category — safe for the unauthenticated health response. */ +function failureReason(err: unknown): string { + if (err instanceof Error) { + if (err.name === "TimeoutError" || err.name === "AbortError") { + return "timeout"; + } + // fetch surfaces DNS/TLS/socket failures as TypeError. + if (err instanceof TypeError) return "connection_error"; + } + return "error"; +} + +export class SinkForwarder { + /** Always sorted by `seq`: queue[0] is the oldest batch. */ + private readonly queue: QueuedBatch[] = []; + private queuedBytes = 0; + /** Per-org share of queuedBytes — overflow evicts from the heaviest org. */ + private readonly queuedBytesByOrg = new Map(); + private inFlight = 0; + private timer: NodeJS.Timeout | null = null; + private stopped = false; + private seqCounter = 0; + + private delivered = 0; + private retried = 0; + private droppedOverflow = 0; + private droppedPermanent = 0; + private consecutiveFailures = 0; + /** + * Bumped on every failure. A success only clears consecutiveFailures if + * no failure happened after that request STARTED — a concurrent success + * that overlapped a failure proves nothing about current sink health and + * must not reopen full concurrency mid-outage. + */ + private failureEpoch = 0; + private lastFailureAt: string | null = null; + private lastFailureReason: string | null = null; + + private readonly tuning: SinkTuning; + + constructor( + private readonly config: IngesterConfig, + private readonly fetchImpl: typeof fetch = fetch, + tuning: Partial = {}, + ) { + this.tuning = { ...DEFAULT_TUNING, ...tuning }; + } + + /** Queue a batch for delivery. No-op when there is nothing to send. */ + enqueue( + ctx: IngestContext, + occurrences: RuntimeOccurrence[], + metricPoints: RuntimeMetricPoint[] = [], + llmCalls: RuntimeLlmCall[] = [], + ): void { + if (!this.config.sinkUrl || this.stopped) return; + if ( + occurrences.length === 0 && + metricPoints.length === 0 && + llmCalls.length === 0 + ) { + return; + } + const batch = this.buildBatch(ctx, occurrences, metricPoints, llmCalls); + // A batch bigger than the whole buffer could never be admitted without + // evicting everyone else — and the consumer's body cap would reject it + // anyway. Drop it alone instead of letting it flush the queue. + if (batch.bytes > this.maxBufferedBytes()) { + this.droppedOverflow += 1; + console.warn( + `sink batch ${batch.batchId} (org ${batch.orgId}) exceeds the buffer cap (${batch.bytes} bytes) — dropped; replay signals ${batch.signalsFrom ?? "?"} .. ${batch.signalsTo ?? "?"} from ClickHouse`, + ); + return; + } + this.queue.push(batch); + this.track(batch); + this.enforceBounds(); + this.pump(); + } + + stats(): SinkStats { + const oldest = this.queue[0]?.enqueuedAt ?? null; + return { + queued: this.queue.length, + queuedBytes: this.queuedBytes, + inFlight: this.inFlight, + delivered: this.delivered, + retried: this.retried, + droppedOverflow: this.droppedOverflow, + droppedPermanent: this.droppedPermanent, + consecutiveFailures: this.consecutiveFailures, + lastFailureAt: this.lastFailureAt, + lastFailureReason: this.lastFailureReason, + oldestQueuedSince: oldest ? new Date(oldest).toISOString() : null, + }; + } + + pendingCount(): number { + return this.queue.length + this.inFlight; + } + + /** Stop scheduling new deliveries (shutdown). Queued batches are logged + * by the caller — they are recoverable from ClickHouse, not from here. */ + stop(): void { + this.stopped = true; + if (this.timer) { + clearTimeout(this.timer); + this.timer = null; + } + } + + /** Serialize one delivery payload; `seq` pins its age for ordering. */ + private buildBatch( + ctx: IngestContext, + occurrences: RuntimeOccurrence[], + metricPoints: RuntimeMetricPoint[], + llmCalls: RuntimeLlmCall[], + ): QueuedBatch { + const batchId = randomUUID(); + const body = JSON.stringify({ + version: 1, + batchId, + orgId: ctx.orgId, + repositoryId: ctx.repositoryId, + occurrences: occurrences.map((o) => ({ + ...o, + occurredAt: o.occurredAt.toISOString(), + })), + metrics: metricPoints.map((p) => ({ + ...p, + bucketAt: p.bucketAt.toISOString(), + })), + llmCalls: llmCalls.map((c) => ({ + ...c, + startedAt: c.startedAt.toISOString(), + })), + }); + const range = signalRange(occurrences, metricPoints, llmCalls); + return { + batchId, + orgId: ctx.orgId, + body, + bytes: Buffer.byteLength(body), + seq: ++this.seqCounter, + attempts: 0, + nextAttemptAt: Date.now(), + signalsFrom: range.from, + signalsTo: range.to, + enqueuedAt: Date.now(), + }; + } + + private maxBufferedBytes(): number { + return this.config.sinkMaxBufferedMb * 1024 * 1024; + } + + private track(batch: QueuedBatch): void { + this.queuedBytes += batch.bytes; + this.queuedBytesByOrg.set( + batch.orgId, + (this.queuedBytesByOrg.get(batch.orgId) ?? 0) + batch.bytes, + ); + } + + private untrack(batch: QueuedBatch): void { + this.queuedBytes -= batch.bytes; + const left = (this.queuedBytesByOrg.get(batch.orgId) ?? 0) - batch.bytes; + if (left > 0) this.queuedBytesByOrg.set(batch.orgId, left); + else this.queuedBytesByOrg.delete(batch.orgId); + } + + /** Re-insert a retrying batch at its age position (queue is seq-sorted), + * so overflow eviction still drops the genuinely oldest signals first. */ + private insertBySeq(batch: QueuedBatch): void { + const at = this.queue.findIndex((b) => b.seq > batch.seq); + if (at === -1) this.queue.push(batch); + else this.queue.splice(at, 0, batch); + } + + /** + * Oldest-first eviction, charged to the heaviest tenant: the freshest + * signals matter most to grouping, and one org flooding the shared + * buffer must not evict everyone else's batches. + */ + private enforceBounds(): void { + const maxBytes = this.maxBufferedBytes(); + while ( + this.queue.length > this.config.sinkMaxBufferedBatches || + (this.queuedBytes > maxBytes && this.queue.length > 1) + ) { + const dropped = this.evictOne(); + if (!dropped) break; + this.droppedOverflow += 1; + console.warn( + `sink buffer overflow: dropped batch ${dropped.batchId} (org ${dropped.orgId}, signals ${dropped.signalsFrom ?? "?"} .. ${dropped.signalsTo ?? "?"}) — replay this range from ClickHouse`, + ); + } + } + + /** The oldest batch of the org holding the most buffered bytes. */ + private evictOne(): QueuedBatch | null { + let heaviest: string | null = null; + let heaviestBytes = -1; + for (const [orgId, bytes] of this.queuedBytesByOrg) { + if (bytes > heaviestBytes) { + heaviest = orgId; + heaviestBytes = bytes; + } + } + const at = heaviest + ? this.queue.findIndex((b) => b.orgId === heaviest) + : 0; + const [dropped] = this.queue.splice(at === -1 ? 0 : at, 1); + if (!dropped) return null; + this.untrack(dropped); + return dropped; + } + + private currentLimit(): number { + return this.consecutiveFailures > 0 ? 1 : this.tuning.healthyConcurrency; + } + + private pump(): void { + if (this.stopped || !this.config.sinkUrl) return; + const limit = this.currentLimit(); + const now = Date.now(); + while (this.inFlight < limit) { + const index = this.queue.findIndex((b) => b.nextAttemptAt <= now); + if (index === -1) break; + const [batch] = this.queue.splice(index, 1); + if (!batch) break; + this.untrack(batch); + this.inFlight += 1; + void this.send(batch).finally(() => { + this.inFlight -= 1; + this.pump(); + }); + } + this.scheduleWake(); + } + + private scheduleWake(): void { + if (this.timer) { + clearTimeout(this.timer); + this.timer = null; + } + if (this.stopped || this.queue.length === 0) return; + // At capacity there is nothing to wake for: every completion pumps + // again anyway, and a 0 ms timer here would spin the event loop while + // a ready batch waits on a slow in-flight request. + if (this.inFlight >= this.currentLimit()) return; + const next = Math.min(...this.queue.map((b) => b.nextAttemptAt)); + this.timer = setTimeout( + () => { + this.timer = null; + this.pump(); + }, + Math.max(0, next - Date.now()), + ); + this.timer.unref(); + } + + private async send(batch: QueuedBatch): Promise { + batch.attempts += 1; + const epochAtStart = this.failureEpoch; + let reason: string | null = null; + /** Raw failure text — logged server-side, never surfaced in stats. */ + let detail: string | null = null; + let permanent = false; + try { + const res = await this.fetchImpl(this.config.sinkUrl as string, { + method: "POST", + headers: { + "content-type": "application/json", + ...(this.config.sinkToken + ? { authorization: `Bearer ${this.config.sinkToken}` } + : {}), + }, + body: batch.body, + signal: AbortSignal.timeout(this.tuning.requestTimeoutMs), + }); + // Drain the body so keep-alive sockets are reusable; its content is + // irrelevant to delivery. + await res.text().catch(() => {}); + if (res.ok) { + this.delivered += 1; + if (epochAtStart === this.failureEpoch) { + this.consecutiveFailures = 0; + } + return; + } + reason = `http_${res.status}`; + detail = `sink responded ${res.status}`; + // 4xx (except timeout/rate-limit) means the consumer rejected the + // batch — retrying the same body cannot succeed. + permanent = res.status < 500 && res.status !== 408 && res.status !== 429; + } catch (err) { + reason = failureReason(err); + detail = err instanceof Error ? err.message : String(err); + } + + this.failureEpoch += 1; + this.consecutiveFailures += 1; + this.lastFailureAt = new Date().toISOString(); + this.lastFailureReason = reason; + + if (permanent || batch.attempts >= this.config.sinkMaxAttempts) { + this.droppedPermanent += 1; + console.error( + `sink delivery gave up after ${batch.attempts} attempt(s) (${detail}): batch ${batch.batchId} (signals ${batch.signalsFrom ?? "?"} .. ${batch.signalsTo ?? "?"}) — replay this range from ClickHouse`, + ); + return; + } + + this.retried += 1; + // Full jitter avoids retry stampedes when the consumer comes back. + const backoff = Math.min( + this.tuning.retryCapMs, + this.tuning.retryBaseMs * 2 ** (batch.attempts - 1), + ); + batch.nextAttemptAt = Date.now() + backoff / 2 + Math.random() * (backoff / 2); + this.insertBySeq(batch); + this.track(batch); + if (batch.attempts === 1 || batch.attempts % 5 === 0) { + console.warn( + `sink delivery failed (attempt ${batch.attempts}/${this.config.sinkMaxAttempts}, will retry): ${detail}`, + ); + } + this.enforceBounds(); + this.scheduleWake(); + } +} + +/** ISO range across every signal in the batch — the replay hint on drops. */ +function signalRange( + occurrences: RuntimeOccurrence[], + metricPoints: RuntimeMetricPoint[], + llmCalls: RuntimeLlmCall[], +): { from: string | null; to: string | null } { + const timestamps = [ + ...occurrences.map((o) => o.occurredAt), + ...metricPoints.map((p) => p.bucketAt), + ...llmCalls.map((c) => c.startedAt), + ].map((d) => d.getTime()); + if (timestamps.length === 0) return { from: null, to: null }; + return { + from: new Date(Math.min(...timestamps)).toISOString(), + to: new Date(Math.max(...timestamps)).toISOString(), + }; +} diff --git a/packages/otlp-ingester/tsconfig.json b/packages/otlp-ingester/tsconfig.json index d916cf9..e846272 100644 --- a/packages/otlp-ingester/tsconfig.json +++ b/packages/otlp-ingester/tsconfig.json @@ -10,5 +10,6 @@ "skipLibCheck": true, "forceConsistentCasingInFileNames": true }, - "include": ["src"] + "include": ["src"], + "exclude": ["src/**/*.test.ts"] }