Background jobs in practice: queues, retries, idempotency, and dead-letter failures

// Broken, and it looks completely reasonable
await db.transaction(async (tx) => {
  const order = await tx.order.create({ data: input })
  await queue.publish('order.created', { orderId: order.id })   // ← outside the tx
  await tx.inventory.decrement({ sku: input.sku })              // ← this throws
})
// Transaction rolls back. The message does not.
// A worker now picks up a job for an order that was never created.

CREATE TABLE outbox (
    id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    topic        text        NOT NULL,
    payload      jsonb       NOT NULL,
    created_at   timestamptz NOT NULL DEFAULT now(),
    published_at timestamptz
);

CREATE INDEX outbox_unpublished_idx ON outbox (id)
    WHERE published_at IS NULL;

// Now atomic: either both happen or neither does
await db.transaction(async (tx) => {
  const order = await tx.order.create({ data: input })
  await tx.inventory.decrement({ sku: input.sku })
  await tx.outbox.create({
    data: { topic: 'order.created', payload: { orderId: order.id } },
  })
})

// A relay process, running continuously
async function relay() {
  const rows = await db.$queryRaw`
    SELECT id, topic, payload FROM outbox
    WHERE published_at IS NULL
    ORDER BY id
    FOR UPDATE SKIP LOCKED
    LIMIT 100`

  for (const row of rows) {
    await queue.publish(row.topic, row.payload, { messageId: String(row.id) })
    await db.outbox.update({
      where: { id: row.id },
      data: { published_at: new Date() },
    })
  }
}

-- Not idempotent: running twice charges twice
UPDATE accounts SET balance = balance - 100 WHERE id = $1;

-- Idempotent: the second run is a no-op
UPDATE orders SET status = 'shipped', shipped_at = now()
WHERE id = $1 AND status = 'paid';

-- Idempotent by construction
INSERT INTO notifications (user_id, kind, sent_at)
VALUES ($1, $2, now())
ON CONFLICT (user_id, kind) DO NOTHING;

async function handle(job: Job) {
  const key = `job:done:${job.idempotencyKey}`

  // Claim the key. NX means only the first caller wins.
  const claimed = await redis.set(key, 'processing', 'NX', 'EX', 86400)
  if (!claimed) {
    metrics.increment('jobs.duplicate_skipped')
    return
  }

  try {
    await doTheWork(job)
    await redis.set(key, 'done', 'EX', 86400)
  } catch (err) {
    await redis.del(key)   // release so a retry can proceed
    throw err
  }
}

CREATE TABLE job_executions (
    idempotency_key text PRIMARY KEY,
    job_type        text        NOT NULL,
    result          jsonb,
    completed_at    timestamptz NOT NULL DEFAULT now()
);

async function chargeCustomer(job: ChargeJob) {
  return db.transaction(async (tx) => {
    // If this insert fails, the whole transaction aborts and nothing happens twice
    try {
      await tx.jobExecution.create({
        data: { idempotencyKey: job.idempotencyKey, jobType: 'charge' },
      })
    } catch (err) {
      if (isUniqueViolation(err)) return { skipped: true }
      throw err
    }

    await tx.payment.create({ data: { orderId: job.orderId, cents: job.cents } })
    await tx.order.update({
      where: { id: job.orderId },
      data: { status: 'paid' },
    })
  })
}

class PermanentError extends Error {}       // never retry
class TransientError extends Error {}       // retry with backoff
class RateLimitError extends Error {
  constructor(public retryAfterMs: number) { super('rate limited') }
}

async function processJob(job: Job) {
  try {
    await handle(job)
  } catch (err) {
    if (err instanceof PermanentError) {
      await deadLetter(job, err)             // straight to the DLQ, no retries
      return
    }
    if (err instanceof RateLimitError) {
      await requeue(job, err.retryAfterMs)   // their schedule, not ours
      return
    }
    throw err                                // normal backoff path
  }
}

function nextDelayMs(attempt: number): number {
  const base = Math.min(1000 * 2 ** attempt, 3_600_000)   // cap at one hour
  return Math.random() * base                             // full jitter
}

// attempt 0 →  0–1s
// attempt 1 →  0–2s
// attempt 2 →  0–4s
// attempt 6 →  0–64s
// attempt 12 → 0–1h

resource "aws_sqs_queue" "orders" {
  name                       = "orders"
  visibility_timeout_seconds = 300
  message_retention_seconds  = 1209600   # 14 days, the maximum

  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.orders_dlq.arn
    maxReceiveCount     = 5
  })
}

resource "aws_sqs_queue" "orders_dlq" {
  name                      = "orders-dlq"
  message_retention_seconds = 1209600
}

-- A DLQ table for a database-backed queue, with the context that matters
CREATE TABLE dead_letters (
    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    job_type        text        NOT NULL,
    payload         jsonb       NOT NULL,
    idempotency_key text,
    attempts        integer     NOT NULL,
    last_error      text        NOT NULL,
    error_class     text,
    first_failed_at timestamptz NOT NULL,
    dead_lettered_at timestamptz NOT NULL DEFAULT now(),
    redriven_at     timestamptz
);

CREATE INDEX dead_letters_pending_idx ON dead_letters (job_type, dead_lettered_at)
    WHERE redriven_at IS NULL;

CREATE TABLE jobs (
    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    job_type        text        NOT NULL,
    payload         jsonb       NOT NULL,
    idempotency_key text UNIQUE,
    run_after       timestamptz NOT NULL DEFAULT now(),
    attempts        integer     NOT NULL DEFAULT 0,
    max_attempts    integer     NOT NULL DEFAULT 5,
    locked_until    timestamptz,
    locked_by       text
);

CREATE INDEX jobs_ready_idx ON jobs (run_after)
    WHERE locked_until IS NULL;

-- Claim atomically. SKIP LOCKED is what lets many workers run without collision.
UPDATE jobs SET
    locked_until = now() + interval '5 minutes',
    locked_by    = $1,
    attempts     = attempts + 1
WHERE id = (
    SELECT id FROM jobs
    WHERE run_after <= now()
      AND (locked_until IS NULL OR locked_until < now())
      AND attempts < max_attempts
    ORDER BY run_after
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
RETURNING id, job_type, payload, idempotency_key;

// A Temporal-style workflow: linear code, durable across failures
export async function onboardCustomer(input: OnboardInput) {
  const payment = await activities.chargeCard(input.paymentMethod, input.cents)

  const account = await activities.provisionAccount(input.email)

  await activities.sendWelcomeEmail(account.id)

  // The process sleeps for a week. No worker is blocked; no timer table is yours.
  await sleep('7 days')

  if (!(await activities.hasCompletedSetup(account.id))) {
    await activities.sendNudgeEmail(account.id)
  }

  return { accountId: account.id, paymentId: payment.id }
}