Warren
Guide

Dead letters in Go: amqp091-go consumers that fail without losing messages

2026-10-12 · 5 min read · Viktor Baumann

A Go consumer on RabbitMQ has three ways to fail on a message: the handler returns an error, the handler panics, or the connection drops while it works. With autoAck the first two lose the message. With manual acks a panic takes the process down, the message is requeued, the restarted pod gets it again and panics again. And a dropped connection closes the deliveries channel, so a for range loop simply ends and the consumer is gone without a word. Here is a consumer with amqp091-go that sends failures to a dead-letter queue instead, and how to put them back.

Declare a dead-letter route and consume with manual acks

conn, err := amqp.Dial(os.Getenv("AMQP_URL"))
if err != nil {
	log.Fatal(err)
}
ch, err := conn.Channel()
if err != nil {
	log.Fatal(err)
}

// the dead-letter route
must(ch.ExchangeDeclare("orders.dlx", "direct", true, false, false, false, nil))
_, err = ch.QueueDeclare("orders.dlq", true, false, false, false, nil)
must(err)
must(ch.QueueBind("orders.dlq", "orders.dlq", "orders.dlx", false, nil))

_, err = ch.QueueDeclare("orders.process", true, false, false, false, amqp.Table{
	"x-dead-letter-exchange":    "orders.dlx",
	"x-dead-letter-routing-key": "orders.dlq",
})
must(err)

must(ch.Qos(20, 0, false))
deliveries, err := ch.Consume("orders.process", "", false /* autoAck */, false, false, false, nil)
must(err)

for d := range deliveries {
	if err := safeHandle(d); err != nil {
		log.Printf("order %s failed, dead-lettering: %v", d.MessageId, err)
		_ = d.Nack(false, false) // multiple=false, requeue=false
		continue
	}
	_ = d.Ack(false)
}
log.Print("deliveries channel closed: connection or channel is gone")

Nack(false, false) hands the message to the queue's dead-letter exchange; RabbitMQ adds an x-death header with the queue, the reason (rejected), the time, and the original exchange and routing key.

Queue arguments are fixed at creation. If orders.process exists without them, QueueDeclare fails with PRECONDITION_FAILED and the channel is closed. For existing queues use a policy:

rabbitmqctl set_policy orders-dlx '^orders\.process$' \
  '{"dead-letter-exchange":"orders.dlx","dead-letter-routing-key":"orders.dlq"}' \
  --apply-to queues

A panic is a failure, not a crash

A panic in the handler kills the process. The broker requeues the unacknowledged message, Kubernetes restarts the pod, the pod receives the same message first, and you get CrashLoopBackOff with the message as the cause. Turn a panic into an error so the message goes to the dead-letter queue:

func safeHandle(d amqp.Delivery) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("panic: %v", r)
		}
	}()
	return handle(d.Body)
}

On a quorum queue in RabbitMQ 4 the crash loop does end on its own after 20 deliveries, the default delivery limit, when the message is dropped or dead-lettered. On a classic queue it does not.

The loop that ends quietly

deliveries is closed when the channel or the connection closes: a broker restart, a network blip, a PRECONDITION_FAILED from a double ack. The for range ends and, unless you check, the service runs on without consuming. amqp091-go does not reconnect by itself. Watch for the close and rebuild connection, channel and consumer:

closed := conn.NotifyClose(make(chan *amqp.Error, 1))
// ... consume loop in a goroutine ...
if err := <-closed; err != nil {
	log.Printf("connection closed: %v, reconnecting", err)
	// back off, then dial again and redeclare the consumer
}

Messages that were delivered but not acknowledged when the connection died are requeued by the broker and delivered again to the next consumer, with Redelivered set. Make the handler idempotent, because that redelivery is normal, not a failure.

Retry with a delay, counted in x-death

A wait queue with a TTL whose dead-letter exchange points back at the work queue gives you delayed retries without holding the message in your process. Every round increments the count of the same x-death entry, so the consumer can give up after a few. amqp091-go decodes the header as []interface{} of amqp.Table, with count as int64:

func rounds(h amqp.Table, queue string) int64 {
	deaths, _ := h["x-death"].([]interface{})
	for _, x := range deaths {
		death, _ := x.(amqp.Table)
		if death["queue"] == queue && death["reason"] == "rejected" {
			n, _ := death["count"].(int64)
			return n
		}
	}
	return 0
}

In the consume loop: a temporary error with fewer than five rounds is Nack(false, false) into orders.retry.wait (x-message-ttl: 30000, no consumers, dead-lettering back to orders.process); after that, publish the message to orders.dlq on a confirm channel and ack it. To see what such a message has been through, paste it into the x-death decoder.

Put them back

Read from the dead-letter queue, publish to the original route with confirms, ack the dead-lettered copy only after the confirm:

must(ch.Confirm(false))
for {
	d, ok, err := ch.Get("orders.dlq", false)
	must(err)
	if !ok {
		break
	}
	deaths, _ := d.Headers["x-death"].([]interface{})
	death := deaths[0].(amqp.Table)
	keys := death["routing-keys"].([]interface{})
	headers := amqp.Table{}
	for k, v := range d.Headers {
		if !strings.HasPrefix(k, "x-death") && !strings.HasPrefix(k, "x-first-death-") &&
			!strings.HasPrefix(k, "x-last-death-") && k != "x-delivery-count" && k != "x-acquired-count" {
			headers[k] = v
		}
	}
	conf, err := ch.PublishWithDeferredConfirmWithContext(ctx, death["exchange"].(string), keys[0].(string), true, false,
		amqp.Publishing{Headers: headers, ContentType: d.ContentType, MessageId: d.MessageId, DeliveryMode: d.DeliveryMode, Body: d.Body})
	must(err)
	if !conf.Wait() {
		log.Fatalf("broker refused %s", d.MessageId)
	}
	must(d.Ack(false))
}

Copy the other properties you rely on (CorrelationId, Type, Timestamp) the same way. Send one message first and wait before the rest; Replay one message first explains how long.

Checklist