From 0a69a33cc1f5abdfc3c332fb9d6a064bfe998c02 Mon Sep 17 00:00:00 2001 From: Frioh Date: Fri, 28 Aug 2026 13:42:48 +0100 Subject: [PATCH 1/2] fix(processing): persist dead-lettered events --- xstreamroll-processing/.env.example | 4 + xstreamroll-processing/.gitignore | 1 + .../__tests__/dead-letter-store.test.ts | 62 +++++++++++ .../__tests__/session.test.ts | 5 + xstreamroll-processing/src/config.ts | 1 + .../src/dead-letter-store.ts | 103 ++++++++++++++++++ xstreamroll-processing/src/metrics.ts | 26 ++++- xstreamroll-processing/src/session.ts | 28 +++-- xstreamroll-processing/src/worker.ts | 7 +- 9 files changed, 227 insertions(+), 10 deletions(-) create mode 100644 xstreamroll-processing/__tests__/dead-letter-store.test.ts create mode 100644 xstreamroll-processing/src/dead-letter-store.ts diff --git a/xstreamroll-processing/.env.example b/xstreamroll-processing/.env.example index 99f3206..567023e 100644 --- a/xstreamroll-processing/.env.example +++ b/xstreamroll-processing/.env.example @@ -28,6 +28,10 @@ POLL_INTERVAL_MS=5000 # Maximum number of stream sessions processed concurrently. Defaults to 32. MAX_CONCURRENT_SESSIONS=32 +# Durable local dead-letter file. Mount this path in production if records +# must survive worker replacement. GET /dead-letters exposes its contents. +DEAD_LETTER_STORE_PATH=./data/dead-letters.json + # ────────────────────────────────────────────────────────────────────────────── # Worker identification (issue #347) # diff --git a/xstreamroll-processing/.gitignore b/xstreamroll-processing/.gitignore index df5102b..c631da1 100644 --- a/xstreamroll-processing/.gitignore +++ b/xstreamroll-processing/.gitignore @@ -9,3 +9,4 @@ dist/ *.swp *.swo coverage/ +data/dead-letters.json diff --git a/xstreamroll-processing/__tests__/dead-letter-store.test.ts b/xstreamroll-processing/__tests__/dead-letter-store.test.ts new file mode 100644 index 0000000..b0d4abe --- /dev/null +++ b/xstreamroll-processing/__tests__/dead-letter-store.test.ts @@ -0,0 +1,62 @@ +import { mkdtemp, readFile, rm } from "fs/promises" +import { tmpdir } from "os" +import { join } from "path" + +import { FileDeadLetterStore } from "../src/dead-letter-store" +import { StreamEvent } from "../src/session" + +describe("FileDeadLetterStore", () => { + let directory: string + let store: FileDeadLetterStore + + beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), "xstreamroll-dlq-")) + store = new FileDeadLetterStore(join(directory, "dead-letters.json")) + }) + + afterEach(async () => { + await rm(directory, { recursive: true, force: true }) + }) + + it("persists a dead-letter and deduplicates the same event", async () => { + const event: StreamEvent = { + id: "event-1", + streamId: "stream-1", + data: { value: 1 }, + timestamp: "2026-08-28T00:00:00.000Z", + } + + await store.record(event, new Error("first failure"), 3) + await store.record(event, new Error("second failure"), 4) + + const records = await store.list() + expect(records).toHaveLength(1) + expect(records[0]).toMatchObject({ + key: "stream-1:event-1", + streamId: "stream-1", + event, + error: "second failure", + attemptCount: 4, + }) + expect(JSON.parse(await readFile(join(directory, "dead-letters.json"), "utf8"))) + .toHaveProperty("stream-1:event-1") + }) + + it("deduplicates events without an id using stable event content", async () => { + const event: StreamEvent = { + streamId: "stream-1", + data: { z: 2, a: 1 }, + timestamp: "2026-08-28T00:00:00.000Z", + } + const sameEventWithDifferentKeyOrder: StreamEvent = { + streamId: "stream-1", + data: { a: 1, z: 2 }, + timestamp: "2026-08-28T00:00:00.000Z", + } + + await store.record(event, new Error("failure"), 1) + await store.record(sameEventWithDifferentKeyOrder, new Error("failure"), 1) + + expect(await store.list()).toHaveLength(1) + }) +}) \ No newline at end of file diff --git a/xstreamroll-processing/__tests__/session.test.ts b/xstreamroll-processing/__tests__/session.test.ts index 7d136b6..6227357 100644 --- a/xstreamroll-processing/__tests__/session.test.ts +++ b/xstreamroll-processing/__tests__/session.test.ts @@ -267,6 +267,7 @@ describe("StreamSession — publish retry and dead-letter (issue #343)", () => { it("dead-letters after exhausting retries and continues processing remaining queue", async () => { const published: ProcessedStreamEvent[] = [] const deadLettered: StreamEvent[] = [] + const deadLetterAttempts: number[] = [] let callsForFirst = 0 // First event always fails; second event always succeeds @@ -281,6 +282,9 @@ describe("StreamSession — publish retry and dead-letter (issue #343)", () => { } published.push(e) }, + deadLetter: async (_event, _error, attempts) => { + deadLetterAttempts.push(attempts) + }, }, 1000, 2, @@ -306,6 +310,7 @@ describe("StreamSession — publish retry and dead-letter (issue #343)", () => { expect(callsForFirst).toBe(3) expect(deadLettered).toHaveLength(1) expect(deadLettered[0].data).toEqual({ seq: 1 }) + expect(deadLetterAttempts).toEqual([3]) // Second event was still published successfully expect(published).toHaveLength(1) diff --git a/xstreamroll-processing/src/config.ts b/xstreamroll-processing/src/config.ts index b4bb355..b2292bb 100644 --- a/xstreamroll-processing/src/config.ts +++ b/xstreamroll-processing/src/config.ts @@ -42,6 +42,7 @@ const envSchema = z.object({ .default("3") .transform((s) => Number(s)) .pipe(z.number().int().min(0)), + DEAD_LETTER_STORE_PATH: z.string().default("./data/dead-letters.json"), /** * Backend for the per-stream {@link EventFilter} config store * (issue #351). `memory` keeps every config in-process and matches diff --git a/xstreamroll-processing/src/dead-letter-store.ts b/xstreamroll-processing/src/dead-letter-store.ts new file mode 100644 index 0000000..afec2c0 --- /dev/null +++ b/xstreamroll-processing/src/dead-letter-store.ts @@ -0,0 +1,103 @@ +import { createHash } from "crypto" +import { mkdir, readFile, rename, writeFile } from "fs/promises" +import { dirname } from "path" + +import { StreamEvent } from "./session" + +export interface DeadLetterRecord { + key: string + streamId: string + event: StreamEvent + error: string + attemptCount: number + timestamp: string +} + +export interface DeadLetterStore { + record(event: StreamEvent, error: Error, attempts: number): Promise + list(): Promise +} + +/** File-backed dead-letter store used by the dependency-free worker. */ +export class FileDeadLetterStore implements DeadLetterStore { + private pendingWrite: Promise = Promise.resolve() + + constructor(private readonly filePath: string) {} + + async record( + event: StreamEvent, + error: Error, + attempts: number, + ): Promise { + this.pendingWrite = this.pendingWrite.catch(() => undefined).then(async () => { + const records = await this.readRecords() + const key = eventKey(event) + const existing = records[key] + records[key] = existing + ? { + ...existing, + error: error.message, + attemptCount: attempts, + timestamp: new Date().toISOString(), + } + : { + key, + streamId: event.streamId, + event, + error: error.message, + attemptCount: attempts, + timestamp: new Date().toISOString(), + } + await this.writeRecords(records) + }) + return this.pendingWrite + } + + async list(): Promise { + await this.pendingWrite + const records = await this.readRecords() + return Object.values(records).sort((left, right) => + left.timestamp.localeCompare(right.timestamp), + ) + } + + private async readRecords(): Promise> { + try { + const raw = await readFile(this.filePath, "utf8") + return JSON.parse(raw) as Record + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return {} + throw error + } + } + + private async writeRecords( + records: Record, + ): Promise { + await mkdir(dirname(this.filePath), { recursive: true }) + const temporaryPath = `${this.filePath}.${process.pid}.tmp` + await writeFile(temporaryPath, JSON.stringify(records, null, 2), "utf8") + await rename(temporaryPath, this.filePath) + } +} + +function eventKey(event: StreamEvent): string { + if (event.id) return `${event.streamId}:${event.id}` + return createHash("sha256") + .update(stableStringify(event)) + .digest("hex") +} + +function stableStringify(value: unknown): string { + if (Array.isArray(value)) { + return `[${value.map((item) => stableStringify(item)).join(",")}]` + } + if (value !== null && typeof value === "object") { + const object = value as Record + return `{${Object.keys(object) + .sort() + .map((key) => `${JSON.stringify(key)}:${stableStringify(object[key])}`) + .join(",")}}` + } + return JSON.stringify(value) ?? "null" +} diff --git a/xstreamroll-processing/src/metrics.ts b/xstreamroll-processing/src/metrics.ts index 933bb94..c528331 100644 --- a/xstreamroll-processing/src/metrics.ts +++ b/xstreamroll-processing/src/metrics.ts @@ -1,5 +1,7 @@ import { createServer, IncomingMessage, ServerResponse } from "http" +import { DeadLetterStore } from "./dead-letter-store" + export interface Metrics { messagesProcessed: number errors: number @@ -104,7 +106,10 @@ export function markReady(): void { live = true } -export function startMetricsServer(port = 3002): ReturnType { +export function startMetricsServer( + port = 3002, + deadLetterStore?: DeadLetterStore, +): ReturnType { const server = createServer((req: IncomingMessage, res: ServerResponse) => { const url = req.url || "" const accept = (req.headers.accept || "").toLowerCase() @@ -147,6 +152,25 @@ export function startMetricsServer(port = 3002): ReturnType return } + if (url === "/dead-letters" || url === "/dead-letters/") { + if (!deadLetterStore) { + res.writeHead(503, { "Content-Type": "application/json" }) + res.end(JSON.stringify({ error: "dead-letter store unavailable" })) + return + } + void deadLetterStore.list().then( + (records) => { + res.writeHead(200, { "Content-Type": "application/json" }) + res.end(JSON.stringify(records)) + }, + () => { + res.writeHead(500, { "Content-Type": "application/json" }) + res.end(JSON.stringify({ error: "dead-letter store unavailable" })) + }, + ) + return + } + if (url === "/metrics" || url === "/metrics/" || url === "/metrics/prometheus") { const wantsJson = accept.includes("application/json") && diff --git a/xstreamroll-processing/src/session.ts b/xstreamroll-processing/src/session.ts index 056ec80..a0ea4e3 100644 --- a/xstreamroll-processing/src/session.ts +++ b/xstreamroll-processing/src/session.ts @@ -26,6 +26,8 @@ export interface SessionHandlers { * caller back-pressure the queue when the API is slow. */ publish(event: ProcessedStreamEvent): Promise + /** Persists an event after its publish retry budget is exhausted. */ + deadLetter?(event: StreamEvent, error: Error, attempts: number): Promise /** Optional structured logger; defaults to console. */ logger?: Pick } @@ -46,13 +48,13 @@ export interface SessionHandlers { * errored ◀─── fail() called explicitly ──── * * Note: publish errors inside the pump loop are handled via - * exponential-backoff retry + dead-lettering (issue #343) and do - * NOT call `fail()`; the session keeps running after a dead-letter. + * exponential-backoff retry + durable dead-lettering (issue #343/#525) + * and do NOT call `fail()`; the session keeps running after a dead-letter. * * The session emits the following events for observability: * - `state` (next: SessionState, prev: SessionState) * - `processed` (ProcessedStreamEvent) - * - `dead-letter` (StreamEvent, Error) — emitted when retry budget exhausted + * - `dead-letter` (StreamEvent, Error) — emitted after durable recording * - `error` (Error) — only emitted by explicit `fail()` calls */ export class StreamSession extends EventEmitter { @@ -133,8 +135,8 @@ export class StreamSession extends EventEmitter { * repeated publish errors from the coordinator). Drops queued work * and marks the session errored so the registry can evict it. * - * Note: publish errors inside the pump loop are handled via - * exponential-backoff retry + dead-lettering (issue #343) and do + * Note: publish errors inside the pump loop are handled via + * exponential-backoff retry + durable dead-lettering (issue #343/#525) and do * NOT call this method; the session keeps running after a dead-letter. */ fail(err: Error): void { @@ -191,12 +193,22 @@ export class StreamSession extends EventEmitter { attempts++ if (attempts > maxRetries) { - // Dead-letter: log the event, emit the signal, then - // continue processing the rest of the queue. The session - // remains running so subsequent events still get a chance. + // Persist before emitting so an emitted dead-letter means the + // durable failure record already exists. this.logger.error( `[${this.workerId}] session ${this.id} publish FAILED after ${attempts} attempt(s) — dead-lettering event: ${error.message}`, ) + try { + await this.handlers.deadLetter?.(next, error, attempts) + } catch (deadLetterError) { + const message = + deadLetterError instanceof Error + ? deadLetterError.message + : String(deadLetterError) + this.logger.error( + `[${this.workerId}] failed to persist dead-letter: ${message}`, + ) + } this.emit("dead-letter", next, error) break // skip this event, carry on with the queue } diff --git a/xstreamroll-processing/src/worker.ts b/xstreamroll-processing/src/worker.ts index 1b8de78..420189d 100644 --- a/xstreamroll-processing/src/worker.ts +++ b/xstreamroll-processing/src/worker.ts @@ -7,6 +7,7 @@ import { env } from "./config" import { createLockManager, type LockManager } from "./leader-election" import { GracefulShutdown, ShutdownReason } from "./lifecycle" import { currentCorrelationId, newCorrelationId } from "./logger" +import { FileDeadLetterStore } from "./dead-letter-store" import { markShuttingDown, setQueueDepth, startMetricsServer } from "./metrics" import { EventFilter, @@ -54,6 +55,7 @@ const EVENT_FILTER_REDIS_URL: string | undefined = // Shared keep-alive agent so axios reuses TCP connections and we can // explicitly destroy the pool on graceful shutdown. export const httpAgent = new http.Agent({ keepAlive: true }) +export const deadLetterStore = new FileDeadLetterStore(env.DEAD_LETTER_STORE_PATH) /** * HTTP server that exposes worker metrics and probes to Kubernetes. @@ -63,7 +65,7 @@ export const httpAgent = new http.Agent({ keepAlive: true }) * below) can close it without losing the reference. */ export const metricsServer = - env.NODE_ENV !== "test" ? startMetricsServer(3002) : null + env.NODE_ENV !== "test" ? startMetricsServer(3002, deadLetterStore) : null // Axios instance that routes all requests through the shared agent. export const axiosInstance = axios.create({ httpAgent }) @@ -312,6 +314,9 @@ async function start(): Promise { async publish(event: ProcessedStreamEvent): Promise { await axiosInstance.post(`${API_URL}/streams/processed`, event) }, + async deadLetter(event, error, attempts): Promise { + await deadLetterStore.record(event, error, attempts) + }, }, { maxConcurrentSessions: MAX_CONCURRENT_SESSIONS, From ce4ddfab98161b1025c28fdca7af4c51cd315934 Mon Sep 17 00:00:00 2001 From: Frioh Date: Sat, 29 Aug 2026 16:28:02 +0100 Subject: [PATCH 2/2] fix(processing): align worker tests with paginated poll API --- .../__tests__/integration/pipeline.integration.test.ts | 1 + .../__tests__/integration/worker.integration.test.ts | 9 +++++++-- xstreamroll-processing/src/lifecycle.ts | 7 ++++++- 3 files changed, 14 insertions(+), 3 deletions(-) diff --git a/xstreamroll-processing/__tests__/integration/pipeline.integration.test.ts b/xstreamroll-processing/__tests__/integration/pipeline.integration.test.ts index 7955993..a6e6910 100644 --- a/xstreamroll-processing/__tests__/integration/pipeline.integration.test.ts +++ b/xstreamroll-processing/__tests__/integration/pipeline.integration.test.ts @@ -65,6 +65,7 @@ test("filtered events are not published (integration)", async () => { let sent = false nock("http://mock-api") .get("/streams/pending") + .query(true) .times(100) .reply(() => { if (!sent) { diff --git a/xstreamroll-processing/__tests__/integration/worker.integration.test.ts b/xstreamroll-processing/__tests__/integration/worker.integration.test.ts index 8b8e0c0..1383ac2 100644 --- a/xstreamroll-processing/__tests__/integration/worker.integration.test.ts +++ b/xstreamroll-processing/__tests__/integration/worker.integration.test.ts @@ -45,6 +45,7 @@ test("single event: polled -> session -> published", async () => { let sent = false nock("http://mock-api") .get("/streams/pending") + .query(true) .times(100) .reply(() => { if (!sent) { @@ -86,6 +87,7 @@ test("multiple events same stream -> routed to same session", async () => { let sentOnce = false nock("http://mock-api") .get("/streams/pending") + .query(true) .times(100) .reply(() => { if (!sentOnce) { @@ -127,6 +129,7 @@ test("capacity exceeded -> event dropped, not published", async () => { let once = false nock("http://mock-api") .get("/streams/pending") + .query(true) .times(100) .reply(() => { if (!once) { @@ -164,6 +167,7 @@ test("graceful shutdown flushes pending publishes", async () => { let sent = false nock("http://mock-api") .get("/streams/pending") + .query(true) .times(100) .reply(() => { if (!sent) { @@ -183,7 +187,7 @@ test("graceful shutdown flushes pending publishes", async () => { return "ok" }) - const workerMod = await import("../../src/worker") + workerMod = await import("../../src/worker") // give worker a moment to pick up the event await new Promise((r) => setTimeout(r, 100)) @@ -202,6 +206,7 @@ test("api error then recovery -> worker retries next poll", async () => { let calls = 0 nock("http://mock-api") .get("/streams/pending") + .query(true) .times(100) .reply(() => { calls++ @@ -219,7 +224,7 @@ test("api error then recovery -> worker retries next poll", async () => { }) }) - const workerMod = await import("../../src/worker") + workerMod = await import("../../src/worker") await awaitWithTimeout(publishedPromise, 5000, "publish timeout") await workerMod.shutdown("test") }) diff --git a/xstreamroll-processing/src/lifecycle.ts b/xstreamroll-processing/src/lifecycle.ts index 234d5ae..34d44bd 100644 --- a/xstreamroll-processing/src/lifecycle.ts +++ b/xstreamroll-processing/src/lifecycle.ts @@ -18,7 +18,12 @@ */ export type ShutdownReason = - "SIGINT" | "SIGTERM" | "uncaughtException" | "unhandledRejection" | "manual" + | "SIGINT" + | "SIGTERM" + | "uncaughtException" + | "unhandledRejection" + | "manual" + | "test" export interface ShutdownHook { /** Human-readable name for logging. */