feat(ingester): at-least-once sink delivery with bounded retry buffer

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_<status>`, `error`) — raw
+  transport errors stay in server logs, never in the unauthenticated
+  health response.

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` |

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"
 	},

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) |

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),

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> = {},
+): 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);
+});

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

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";