Park poison messages after three failed attempts #482

OpenSWSam Whitfieldwants to merge 4 commits intomain←sam/dead-letter-queue
12/12 checks
LMJW
Files changed5+91 −15
1 of 5 viewed2 open
  • src/queue
  • test
  • docs/webhooks
JK filesV viewed
N next threadR review
src/queue/consumer.ts+19−7
@@ -1,7 +1,8 @@
1import { Queue, type Message } from "./queue";
1import { Queue, type Message } from "./queue";
2import { deliver } from "../delivery";
2import { deliver } from "../delivery";
3−import { backoff } from "./retry";
3+import { backoff, MAX_ATTEMPTS } from "./retry";
  • LM
    Leo Marchetti2h

    Should MAX_ATTEMPTS come from the endpoint's retry policy? Ops will want to tune it for the noisy integrations.

  • SW
    Sam Whitfield1h

    Fair. I'd keep the constant as the default and read endpoint.retryPolicy in a follow-up, so this one stays small.

4+import { deadLetter } from "./dead-letter";
4import { metrics } from "../metrics";
5import { metrics } from "../metrics";
5
6
6const queue = new Queue("webhooks");
7const queue = new Queue("webhooks");
⋯@@ -38,15 +39,31 @@ export async function consume(batchSize = 50)
38export async function consume(batchSize = 50) {
39export async function consume(batchSize = 50) {
39 const messages = await queue.receive(batchSize);
40 const messages = await queue.receive(batchSize);
40
41
41 for (const message of messages) {
42 for (const message of messages) {
42− try {
43+ await handle(message);
43− await deliver(message.body);
44− await queue.ack(message);
45− } catch (error) {
46− await queue.retry(message, backoff(message.attempts));
47− }
48 }
44 }
49
45
50 metrics.gauge("webhooks.lag", await queue.lag());
46 metrics.gauge("webhooks.lag", await queue.lag());
51}
47}
48+
49+async function handle(message: Message) {
50+ try {
51+ await deliver(message.body);
52+ await queue.ack(message);
53+ } catch (error) {
54+ if (message.attempts + 1 >= MAX_ATTEMPTS) {
55+ // Park it so one bad payload can't hold up the partition.
56+ await deadLetter.put(message, error);
57+ await queue.ack(message);
58+ metrics.increment("webhooks.dead_lettered");
59+ return;
60+ }
61+ await queue.retry(message, backoff(message.attempts));
62+ }
63+}
src/queue/dead-letter.tsNew+35
@@ -0,0 +1,35 @@
1+import { db } from "../db";
2+import type { Message } from "./queue";
3+
4+const RETENTION_DAYS = 7;
5+
6+export interface DeadLetter {
7+ id: string;
8+ endpointId: string;
9+ payload: unknown;
10+ error: string;
11+ attempts: number;
12+ parkedAt: Date;
13+}
14+
15+export const deadLetter = {
16+ async put(message: Message, error: unknown) {
17+ await db.insert("dead_letters", {
18+ id: message.id,
19+ endpointId: message.endpointId,
20+ payload: message.body,
21+ error: String(error),
22+ attempts: message.attempts + 1,
23+ parkedAt: new Date(),
24+ });
25+ },
26+
27+ async replay(id: string) {
28+ const row = await db.take<DeadLetter>("dead_letters", id);
29+ return row && queue.publish(row.payload, { endpointId: row.endpointId });
  • LM
    Leo Marchetti48m

    queue isn't imported in this file, so replay will throw the first time someone clicks it.

30+ },
31+
32+ async sweep() {
33+ await db.deleteOlderThan("dead_letters", "parkedAt", RETENTION_DAYS);
34+ },
35+};
src/queue/retry.ts+4−6
@@ -1,11 +1,8 @@
1+export const MAX_ATTEMPTS = 3;
1const BASE_DELAY_MS = 1_000;
2const BASE_DELAY_MS = 1_000;
2−const MAX_DELAY_MS = 60_000;
3+const MAX_DELAY_MS = 30_000;
Resolved · Jonas: 30s cap feels low for endpoints that are down for a deploy.3h
3
4
4export function backoff(attempt: number) {
5export function backoff(attempt: number) {
5− const delay = BASE_DELAY_MS * 2 ** attempt;
6+ const delay = Math.min(MAX_DELAY_MS, BASE_DELAY_MS * 2 ** attempt);
6− if (delay > MAX_DELAY_MS) {
7+ return delay / 2 + Math.random() * (delay / 2);
7− return MAX_DELAY_MS;
8− }
9− return delay;
10}
8}
test/consumer.test.tsNew+26
@@ -0,0 +1,26 @@
1+import { describe, expect, it } from "vitest";
2+import { consume } from "../src/queue/consumer";
3+import { fakeQueue, failingEndpoint } from "./fixtures";
4+
5+describe("consume", () => {
6+ it("parks a message after three failed attempts", async () => {
7+ const queue = fakeQueue([{ id: "m1", attempts: 2, body: "{" }]);
8+ failingEndpoint("ep_live_42");
9+
10+ await consume();
11+
12+ expect(queue.deadLetters).toHaveLength(1);
13+ expect(queue.pending).toHaveLength(0);
14+ });
15+
16+ it("keeps delivering after a poison message", async () => {
17+ const queue = fakeQueue([
18+ { id: "bad", attempts: 2, body: "{" },
19+ { id: "ok", attempts: 0, body: "{}" },
20+ ]);
21+
22+ await consume();
23+
24+ expect(queue.delivered).toContain("ok");
25+ });
26+});
docs/webhooks/retries.md+7−2

That's every file. Press R to finish your review.