Handling Poison Messages with Dead-Letter Queues #

Give every media work queue a redrive policy with a low maxReceiveCount (3–5) pointing at a dead-letter queue, make workers distinguish permanent errors (corrupt input, unsupported codec) from transient ones (throttling, timeouts) and fail the asset immediately on the first kind instead of letting the message cycle, alarm on DLQ depth, and use StartMessageMoveTask to redrive only after the cause is fixed.

A poison message is a job that fails every time it is processed. In a media pipeline it is almost always an upload: a truncated MP4 whose moov atom never arrived, a WAV with a lying header, a 200-megapixel PNG that makes the image library run out of memory. Without a dead-letter queue, the message returns to the queue after every visibility timeout, and every retry burns a full transcode attempt; with enough of them, workers spend their time failing on the same few files while good uploads wait. This page is part of media job orchestration in media processing and delivery pipelines. It builds on the queue from queueing transcode jobs with SQS and Lambda.

When to use this approach #

• Workers consume media jobs from SQS (or any queue with redelivery), and you have seen the same message fail over and over.
• Jobs are expensive — minutes of CPU — so each wasted retry is visible on the bill.
• You need a place where failed jobs wait for a human or a fix, without being lost and without blocking the healthy ones.

Prerequisites #

1. An SQS standard queue for jobs and a second standard queue as its DLQ, in the same account and region.
2. @aws-sdk/client-sqs v3 for the worker and the redrive tooling.
3. Workers that set message visibility longer than the slowest job, or extend it while working.
4. A place to record per-asset status — the same row the idempotent job claims use.

How a message becomes dead #

Implementation #

Configure the redrive policy once:

import { SQSClient, SetQueueAttributesCommand, GetQueueAttributesCommand } from "@aws-sdk/client-sqs";

const sqs = new SQSClient({});

export async function attachDlq(queueUrl: string, dlqArn: string, maxReceiveCount = 3): Promise<void> {
  await sqs.send(new SetQueueAttributesCommand({
    QueueUrl: queueUrl,
    Attributes: {
      RedrivePolicy: JSON.stringify({ deadLetterTargetArn: dlqArn, maxReceiveCount }),
      VisibilityTimeout: "900",          // longer than the slowest job
    },
  }));
  const { Attributes } = await sqs.send(new GetQueueAttributesCommand({
    QueueUrl: queueUrl, AttributeNames: ["RedrivePolicy"],
  }));
  console.log(Attributes?.RedrivePolicy);
}

Then make the worker classify errors, so input that can never succeed fails fast and does not wait three receives to reach the DLQ:

import {
  SQSClient, ReceiveMessageCommand, DeleteMessageCommand, ChangeMessageVisibilityCommand,
  type Message,
} from "@aws-sdk/client-sqs";

const sqs = new SQSClient({});
const QUEUE_URL = process.env.QUEUE_URL!;

/** Input that will fail identically on every attempt. */
export class PermanentError extends Error {
  constructor(message: string, readonly reason: string) { super(message); this.name = "PermanentError"; }
}

const PERMANENT_PATTERNS: [RegExp, string][] = [
  [/moov atom not found/i, "truncated-mp4"],
  [/Invalid data found when processing input/i, "unreadable-container"],
  [/Input image exceeds pixel limit/i, "image-too-large"],
  [/unsupported image format/i, "unsupported-format"],
  [/matches no streams/i, "missing-stream"],
];

export function classify(err: unknown): PermanentError | Error {
  const text = String((err as Error)?.message ?? err);
  for (const [re, reason] of PERMANENT_PATTERNS) {
    if (re.test(text)) return new PermanentError(text, reason);
  }
  return err instanceof Error ? err : new Error(text);
}

async function markAsset(assetId: string, status: "failed", reason: string): Promise<void> {
  console.log(JSON.stringify({ msg: "asset status", assetId, status, reason }));   // replace with DB update
}

export async function processOne(
  msg: Message,
  work: (body: { assetId: string }) => Promise<void>,
): Promise<void> {
  const body = JSON.parse(msg.Body ?? "{}") as { assetId: string };
  const receives = Number(msg.Attributes?.ApproximateReceiveCount ?? "1");

  // Heartbeat: keep the message invisible while a long job runs.
  const heartbeat = setInterval(() => {
    sqs.send(new ChangeMessageVisibilityCommand({
      QueueUrl: QUEUE_URL, ReceiptHandle: msg.ReceiptHandle!, VisibilityTimeout: 300,
    })).catch(() => { /* a failed heartbeat just lets the message reappear */ });
  }, 120_000);

  try {
    await work(body);
    await sqs.send(new DeleteMessageCommand({ QueueUrl: QUEUE_URL, ReceiptHandle: msg.ReceiptHandle! }));
  } catch (raw) {
    const err = classify(raw);
    if (err instanceof PermanentError) {
      // Record why, delete the message: retrying cannot help, and the DLQ is for surprises.
      await markAsset(body.assetId, "failed", err.reason);
      await sqs.send(new DeleteMessageCommand({ QueueUrl: QUEUE_URL, ReceiptHandle: msg.ReceiptHandle! }));
      return;
    }
    // Transient: back off before the next receive instead of retrying immediately.
    const delay = Math.min(900, 30 * 2 ** (receives - 1));
    await sqs.send(new ChangeMessageVisibilityCommand({
      QueueUrl: QUEUE_URL, ReceiptHandle: msg.ReceiptHandle!, VisibilityTimeout: delay,
    }));
    console.warn(JSON.stringify({ msg: "transient failure", assetId: body.assetId, receives, delay, err: err.message }));
  } finally {
    clearInterval(heartbeat);
  }
}

export async function poll(work: (body: { assetId: string }) => Promise<void>): Promise<void> {
  for (;;) {
    const { Messages = [] } = await sqs.send(new ReceiveMessageCommand({
      QueueUrl: QUEUE_URL, MaxNumberOfMessages: 1, WaitTimeSeconds: 20,
      MessageSystemAttributeNames: ["ApproximateReceiveCount"],
    }));
    for (const m of Messages) await processOne(m, work);
  }
}

Line-by-line on the decisions that matter #

• maxReceiveCount of 3. Low enough that a poison message stops costing money quickly, high enough to survive a genuinely transient failure and one worker crash. For jobs under a minute, 5 is fine; for multi-hour encodes, 2.
• PermanentError patterns from real FFmpeg and Sharp messages. moov atom not found means the MP4 index never arrived — a truncated upload that will never decode. These strings are stable across versions and cheap to match. Extend the list from what your DLQ actually collects.
• Delete on permanent failure. The asset row now carries failed with a reason the uploader can see. Sending known-bad input to the DLQ as well would bury the real surprises under hundreds of expected failures.
• Backoff via ChangeMessageVisibility. Without it, a transient failure makes the message visible again only when the original visibility timeout expires — possibly immediately if the worker crashed. Setting an exponential visibility spreads retries out, like client-side exponential backoff for failed chunks.
• The heartbeat. A 25-minute encode with a 15-minute visibility timeout gets redelivered to a second worker at minute 15 and runs twice. Extending visibility every two minutes keeps it private for exactly as long as the worker is alive.

Inspecting and redriving the DLQ #

When the DLQ alarm fires, look before you redrive. Redriving a batch of genuinely poison messages just burns three more attempts each.

import { SQSClient, ReceiveMessageCommand, StartMessageMoveTaskCommand } from "@aws-sdk/client-sqs";

const sqs = new SQSClient({});

export async function peekDlq(dlqUrl: string, max = 10): Promise<void> {
  const { Messages = [] } = await sqs.send(new ReceiveMessageCommand({
    QueueUrl: dlqUrl, MaxNumberOfMessages: max, VisibilityTimeout: 30,   // look, then let go
    MessageSystemAttributeNames: ["ApproximateReceiveCount", "SentTimestamp"],
  }));
  for (const m of Messages) {
    console.log(m.MessageId, new Date(Number(m.Attributes?.SentTimestamp)).toISOString(), m.Body);
  }
}

/** Move everything back to the source queue at a controlled rate, after the fix is deployed. */
export async function redrive(dlqArn: string, maxPerSecond = 5): Promise<string | undefined> {
  const res = await sqs.send(new StartMessageMoveTaskCommand({
    SourceArn: dlqArn,
    MaxNumberOfMessagesPerSecond: maxPerSecond,   // do not stampede the workers
  }));
  return res.TaskHandle;
}

Configuration gotchas #

The dead letter queue … must be the same type as the source queue. A FIFO queue needs a FIFO DLQ, a standard queue a standard DLQ. Create the DLQ with the same type.

Messages land in the DLQ that never failed. A worker received them and then took longer than the visibility timeout — so SQS redelivered, and the second worker’s receive counted too. Enough slow jobs and healthy messages hit maxReceiveCount. Add the heartbeat or raise the visibility timeout above the slowest job.

DLQ messages expire before anyone looks. The DLQ’s retention period is separate from the source queue’s, and messages keep their original enqueue timestamp. Set the DLQ’s MessageRetentionPeriod to the maximum (14 days) so an alarm on Friday is still actionable on Monday.

Lambda event source mappings retry differently. With an SQS-triggered Lambda, a batch fails as a unit unless you enable ReportBatchItemFailures and return the IDs of only the failed messages. Without it, one poison message in a batch of ten makes all ten retry — and all ten reach the DLQ together.

What to alarm on #

Verification #

# 1. Enqueue a job for a deliberately truncated file.
head -c 200000 good.mp4 > truncated.mp4
aws s3 cp truncated.mp4 s3://uploads/uploads/test-poison/original
aws sqs send-message --queue-url "$QUEUE_URL" --message-body '{"assetId":"test-poison"}'

# 2. Expect: asset marked failed with reason truncated-mp4, message deleted, DLQ still empty.
aws sqs get-queue-attributes --queue-url "$DLQ_URL" \
  --attribute-names ApproximateNumberOfMessages --query 'Attributes'

# 3. Simulate an unknown failure (worker throws a novel error) and confirm the message
#    reaches the DLQ after exactly maxReceiveCount receives.

Frequently Asked Questions