From 2c1f1737e791e6cc2e0fa685b06afb2e0da2e765 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Tue, 14 Jul 2026 13:14:02 +0000 Subject: [PATCH 1/2] fix(redis-worker): move unrecoverable queue items to the dead-letter queue SimpleQueue.dequeue() skipped items whose job field was not a string, whose job had no registered schema, or whose payload failed schema validation. Because the queue uses a visibility-timeout redelivery model (items are only removed on ack or by moving them to the DLQ), these skipped items were re-scored and redelivered on every visibility-timeout cycle indefinitely. These failures are unrecoverable, so route all three branches to moveToDeadLetterQueue() instead of silently skipping. Add tests covering the unknown-job and unparseable-payload cases. --- packages/redis-worker/src/queue.test.ts | 118 ++++++++++++++++++++++++ packages/redis-worker/src/queue.ts | 9 ++ 2 files changed, 127 insertions(+) diff --git a/packages/redis-worker/src/queue.test.ts b/packages/redis-worker/src/queue.test.ts index 8d72c2ca0b9..ef5a8a743de 100644 --- a/packages/redis-worker/src/queue.test.ts +++ b/packages/redis-worker/src/queue.test.ts @@ -484,6 +484,124 @@ describe("SimpleQueue", () => { } }); + redisTest( + "dequeue moves items with an unknown job to the DLQ", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const redisOptions = { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }; + + // Producer enqueues an item for the "test" job. + const producer = new SimpleQueue({ + name: "test-invalid-job", + schema: { + test: z.object({ + value: z.number(), + }), + }, + redisOptions, + logger: new Logger("test", "log"), + }); + + // Consumer shares the same queue name but doesn't know about the "test" job. + const consumer = new SimpleQueue({ + name: "test-invalid-job", + schema: { + other: z.object({ + value: z.number(), + }), + }, + redisOptions, + logger: new Logger("test", "log"), + }); + + try { + await producer.enqueue({ + id: "1", + job: "test", + item: { value: 1 }, + visibilityTimeoutMs: 2000, + }); + expect(await consumer.size()).toBe(1); + expect(await consumer.sizeOfDeadLetterQueue()).toBe(0); + + // Dequeue can't find a schema for the job, so it should be moved to the DLQ. + const dequeued = await consumer.dequeue(1); + expect(dequeued).toEqual([]); + expect(await consumer.size({ includeFuture: true })).toBe(0); + expect(await consumer.sizeOfDeadLetterQueue()).toBe(1); + + // The item must not resurface on a subsequent dequeue. + expect(await consumer.dequeue(1)).toEqual([]); + } finally { + await producer.close(); + await consumer.close(); + } + } + ); + + redisTest( + "dequeue moves items with an unparseable payload to the DLQ", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const redisOptions = { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }; + + // Producer enqueues a numeric payload. + const producer = new SimpleQueue({ + name: "test-invalid-payload", + schema: { + test: z.object({ + value: z.number(), + }), + }, + redisOptions, + logger: new Logger("test", "log"), + }); + + // Consumer expects a string payload for the same job, so validation fails. + const consumer = new SimpleQueue({ + name: "test-invalid-payload", + schema: { + test: z.object({ + value: z.string(), + }), + }, + redisOptions, + logger: new Logger("test", "log"), + }); + + try { + await producer.enqueue({ + id: "1", + job: "test", + item: { value: 1 }, + visibilityTimeoutMs: 2000, + }); + expect(await consumer.size()).toBe(1); + expect(await consumer.sizeOfDeadLetterQueue()).toBe(0); + + // The payload fails schema validation, so it should be moved to the DLQ. + const dequeued = await consumer.dequeue(1); + expect(dequeued).toEqual([]); + expect(await consumer.size({ includeFuture: true })).toBe(0); + expect(await consumer.sizeOfDeadLetterQueue()).toBe(1); + + // The item must not resurface on a subsequent dequeue. + expect(await consumer.dequeue(1)).toEqual([]); + } finally { + await producer.close(); + await consumer.close(); + } + } + ); + redisTest("cleanup orphaned queue entries", { timeout: 20_000 }, async ({ redisContainer }) => { const queue = new SimpleQueue({ name: "test-orphaned", diff --git a/packages/redis-worker/src/queue.ts b/packages/redis-worker/src/queue.ts index becfb3813ea..52b3865cace 100644 --- a/packages/redis-worker/src/queue.ts +++ b/packages/redis-worker/src/queue.ts @@ -198,6 +198,7 @@ export class SimpleQueue { const parsedItem = JSON.parse(serializedItem) as any; if (typeof parsedItem.job !== "string") { this.logger.error(`Invalid item in queue`, { queue: this.name, id, item: parsedItem }); + await this.moveToDeadLetterQueue(id, "Invalid item in queue: 'job' is not a string"); continue; } @@ -214,6 +215,10 @@ export class SimpleQueue { timestamp, availableJobs: Object.keys(this.schema), }); + await this.moveToDeadLetterQueue( + id, + `Invalid item in queue: no schema found for job "${parsedItem.job}"` + ); continue; } @@ -228,6 +233,10 @@ export class SimpleQueue { attempt: parsedItem.attempt, timestamp, }); + await this.moveToDeadLetterQueue( + id, + `Invalid item in queue: payload failed schema validation: ${validatedItem.error.message}` + ); continue; } From 6eefa5ca6e74ac13818a26ef63cf2aa528912d76 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Tue, 14 Jul 2026 13:17:42 +0000 Subject: [PATCH 2/2] chore: add changeset --- .changeset/quick-otters-dlq.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/quick-otters-dlq.md diff --git a/.changeset/quick-otters-dlq.md b/.changeset/quick-otters-dlq.md new file mode 100644 index 00000000000..edb86098d62 --- /dev/null +++ b/.changeset/quick-otters-dlq.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/redis-worker": patch +--- + +Move unrecoverable items (unknown job, missing schema, or invalid payload) to the dead-letter queue instead of redelivering them indefinitely