From 5c25eb543e80609ac88d9701c5634793c04be387 Mon Sep 17 00:00:00 2001 From: devinderstockholm3219gmailcom Date: Sun, 21 Jun 2026 13:14:25 +0300 Subject: [PATCH] feat(shared): retry with exponential backoff before dead-lettering A handler failure used to dead-letter on the first attempt, too aggressive for transient errors (a Redis blip during the idempotency check, a momentary DB hiccup). Failed messages now go to a per-queue retry queue with an exponential per-message TTL (1s/2s/4s) that dead-letters them back into the original queue for redelivery, up to 3 attempts, before falling through to the existing permanent DLQ. Also fixes a routing-key bug this surfaced: a message redelivered via the retry queue arrives through the default exchange with msg.fields.routingKey set to the queue name, not the original event's routing key. The true routing key now travels in a header instead. --- .claude/CONTEXT.md | 17 ++++++----- packages/shared/src/mq.ts | 59 +++++++++++++++++++++++++++++++++++---- 2 files changed, 64 insertions(+), 12 deletions(-) diff --git a/.claude/CONTEXT.md b/.claude/CONTEXT.md index c271177..4da479c 100644 --- a/.claude/CONTEXT.md +++ b/.claude/CONTEXT.md @@ -22,27 +22,30 @@ Full plan: `~/.claude/plans/lets-read-the-files-unified-galaxy.md` | 1e | Notification service — Postgres table (audit log) | ✅ Done | | 2 | Outbox pattern — order service | ✅ Done | | 3 | Redis idempotency — replace in-memory Idempotency class | ✅ Done | -| 4 | Retry-with-backoff in mq.ts | ⏳ Pending | +| 4 | Retry-with-backoff in mq.ts | ✅ Done | | 5 | Testcontainers integration tests | ⏳ Pending | | 6 | End-to-end test | ⏳ Pending | ## Current state -**Phase 3 complete, on branch `phase-3-redis-idempotency` (not yet merged).** The in-memory `Idempotency` class is deleted from `packages/shared/src/mq.ts`; replaced by `RedisIdempotency` in the new `packages/shared/src/idempotency.ts`, using an atomic `SET event:{namespace}:{eventId} 1 EX 86400 NX`. Kitchen and notification (the only two consumers — order never consumed anything) each construct their own instance with a distinct namespace: `new RedisIdempotency("kitchen")` / `new RedisIdempotency("notification")`. +**Phase 4 complete, on branch `phase-4-retry-backoff` (not yet merged) — this closes out the cross-cutting resilience work (Phases 2-4).** `setupConsumer()` in `packages/shared/src/mq.ts` no longer dead-letters on the first handler failure. Each queue now gets a companion `${queue}.retry` queue (no bindings — only ever reached via `sendToQueue`); on failure, the message is republished there with a per-message `expiration` (exponential backoff: 1s, 2s, 4s for retry 1/2/3) and an `x-retry-count` header, then dead-letters BACK into the original queue once that TTL expires (via the default exchange + `deadLetterRoutingKey`). After `MAX_RETRIES` (3) failed attempts, it falls through to the existing permanent `.dlq` exactly as before. No service `index.ts` files changed — entirely contained in `mq.ts`, same shape as the Phase 2 reconnect fix. -**Real bug caught by manual testing, fixed in the same PR**: the first version of `RedisIdempotency` keyed purely on `event:{eventId}`, with no namespace. Since kitchen and notification both consume the SAME `order.placed` event (same `eventId`, two separate queues), whichever service's idempotency check ran first would mark the bare eventId as seen — and the other service's check would then see a false duplicate and silently skip an event it never actually processed. Caught immediately because kitchen's `kitchen_orders` table had no row for a manually-placed test order. The old in-memory `Set` never had this problem because each service held its own private Set; Redis accidentally made the key space global. Fixed by namespacing the key per consumer. +**Real bug caught by manual testing, fixed in the same PR**: a message redelivered via the retry queue arrives through the default exchange with `msg.fields.routingKey` set to the QUEUE NAME, not the original event's routing key (e.g. `order.placed`) — discovered when a 2nd-attempt redelivery threw `TypeError: undefined is not an object (evaluating 'eventSchemas[routingKey].parse')`. Fixed by carrying the true routing key forward in an `x-original-routing-key` header, set on first failure and read in preference to `msg.fields.routingKey` on every subsequent attempt. -Verified manually: normal order flow still works (kitchen accepted→ready, notification logged all 3 events) with correctly-namespaced Redis keys (`event:kitchen:` and `event:notification:` both exist independently for the same eventId); confirmed a second `SET ... NX` for the same key genuinely returns nothing (no overwrite); killed kitchen mid-flight to force a RabbitMQ redelivery and confirmed exactly one `kitchen_orders` row resulted, no duplicate-insert errors. Snyk: 0 issues across `packages/shared`, `services/kitchen`, `services/notification`. Typecheck green. +Verified manually with a throwaway test consumer (not committed) on a dedicated test queue: a handler that fails twice then succeeds on attempt 3 retried at the correct 1s/2s intervals and acked cleanly with no DLQ involvement; a handler that always fails retried exactly 3 times (4 total attempts, t=0/1/3/7s matching the backoff) then landed in the permanent DLQ, with no 4th retry. Re-confirmed the full real order→kitchen→notification happy path afterward — zero retry/error log lines, fully unaffected. Snyk: 0 issues. Typecheck green. **Workflow reminder**: every phase gets its own branch off `main`, pushed with a PR via `gh pr create` — never commit straight to `main`. No Claude/Anthropic references in commits, PRs, or files (user's explicit preference). Repo: https://github.com/devinder-dev/quickbite (public). ## Next step -**Phase 4 — Retry-with-backoff in `mq.ts`.** Goal: don't dead-letter a message on the very first handler failure — retry a bounded number of times with exponential backoff first. Today, `consume()`'s try/catch (`packages/shared/src/mq.ts`) nacks straight to the DLQ on any throw, including transient failures like a brief Redis hiccup during the idempotency check just built in Phase 3. +**Phase 5 — Testcontainers integration tests.** Goal: real Postgres + RabbitMQ in tests, no mocks for the broker/db — this is the graded testing-pyramid requirement (CLAUDE.md: unit → integration → e2e). -Likely approach: track retry count via message headers (`x-retry-count`), republish to a delayed queue (or use a per-queue dead-letter TTL trick) up to `MAX_RETRIES`, then dead-letter only after that's exhausted. +Likely approach: +- Add `@testcontainers/postgresql` and `@testcontainers/rabbitmq` dev deps to order/kitchen/notification (and menu, for its Postgres+Redis path — also `@testcontainers/redis` there) +- Flesh out `services/order/tests/place-order.test.ts` (currently a pure unit test, no I/O) into a real integration test: spin up containers, `POST /orders`, assert the DB row + outbox row + eventual publish +- Similar integration tests for kitchen (consume `order.placed`, assert `kitchen_orders` row + events published) and notification (assert all 3 events logged) -Before starting: branch off `main` as `phase-4-retry-backoff` (after Phase 3's PR is merged). +Before starting: branch off `main` as `phase-5-testcontainers` (after Phase 4's PR is merged). ## Key files to know diff --git a/packages/shared/src/mq.ts b/packages/shared/src/mq.ts index b2532da..a8f225c 100644 --- a/packages/shared/src/mq.ts +++ b/packages/shared/src/mq.ts @@ -38,6 +38,7 @@ function createResilientChannel() { prefetch: (...args: Parameters) => current!.prefetch(...args), consume: (...args: Parameters) => current!.consume(...args), publish: (...args: Parameters) => current!.publish(...args), + sendToQueue: (...args: Parameters) => current!.sendToQueue(...args), ack: (...args: Parameters) => current!.ack(...args), nack: (...args: Parameters) => current!.nack(...args), }; @@ -45,6 +46,13 @@ function createResilientChannel() { export type ResilientChannel = ReturnType; +const MAX_RETRIES = 3; + +// Exponential backoff: 1s, 2s, 4s for retryCount 0, 1, 2. +function backoffMs(retryCount: number): number { + return 1000 * 2 ** retryCount; +} + // The actual queue/binding/consumer setup, extracted so it can run both for // a brand-new consume() call AND be replayed for every past registration // after a reconnect. All operations here are idempotent AMQP declarations — @@ -62,20 +70,61 @@ async function setupConsumer(channel: ResilientChannel, opts: ConsumeOptions, ha await channel.assertQueue(queue, { durable: true, deadLetterExchange: dlx }); for (const key of routingKeys) await channel.bindQueue(queue, EXCHANGE, key); + // The retry queue holds failed messages for a per-message TTL (set on + // each republish below, not on the queue itself — that's what lets each + // retry wait longer than the last). No bindings of its own: messages only + // ever arrive here via sendToQueue. When a message's TTL expires, + // RabbitMQ's default behavior for an expired message is to dead-letter it + // — here, to the original queue (via the nameless default exchange + + // deadLetterRoutingKey), which is exactly a delayed redelivery. + const retryQueue = `${queue}.retry`; + await channel.assertQueue(retryQueue, { + durable: true, + deadLetterExchange: "", + deadLetterRoutingKey: queue, + }); + await channel.prefetch(prefetch); await channel.consume(queue, async (msg) => { if (!msg) return; + // A message redelivered via the retry queue arrives through the default + // exchange with msg.fields.routingKey set to the QUEUE name, not the + // original event's routing key — so a retried message must carry its + // true routing key forward in a header instead of relying on that field + // a second time. Fresh (never-retried) deliveries fall back to the + // field, which is correct for them. + const routingKey = (msg.properties.headers?.["x-original-routing-key"] as EventName) ?? (msg.fields.routingKey as EventName); + try { - const routingKey = msg.fields.routingKey as EventName; const schema = eventSchemas[routingKey]; const event = schema.parse(JSON.parse(msg.content.toString())); await handler(event, msg); channel.ack(msg); } catch (err) { - // Reject without requeue -> message goes to the dead-letter queue. - // TODO: add bounded retry-with-backoff before dead-lettering. - console.error(`[mq] handler failed on ${msg.fields.routingKey}, dead-lettering`, err); - channel.nack(msg, false, false); + const retryCount = (msg.properties.headers?.["x-retry-count"] as number) ?? 0; + + if (retryCount < MAX_RETRIES) { + const delayMs = backoffMs(retryCount); + console.error(`[mq] handler failed on ${routingKey}, retry ${retryCount + 1}/${MAX_RETRIES} in ${delayMs}ms`, err); + channel.sendToQueue(retryQueue, msg.content, { + ...msg.properties, + headers: { + ...msg.properties.headers, + "x-retry-count": retryCount + 1, + "x-original-routing-key": routingKey, + }, + expiration: String(delayMs), + }); + // We've taken ownership of this message via the retry queue — ack + // it off the original queue so it isn't also sent to the DLX. + channel.ack(msg); + } else { + // Retries exhausted — nack without requeue, same as before Phase 4. + // The original queue's deadLetterExchange routes this to the + // permanent DLQ; that wiring is untouched. + console.error(`[mq] handler failed on ${routingKey} after ${MAX_RETRIES} retries, dead-lettering`, err); + channel.nack(msg, false, false); + } } }); }