Durability by design: How AMQP keeps your messages safe

How messages actually get lost

Understanding message safety starts with identifying exactly how systems can fail. Usually, the risks boil down to two things:

• The broker crashes right after a publish - The producer thinks the message is sent, but the broker never had a chance to save it to disk. The message is just gone.
• A consumer crashes mid-job - The broker delivers the message and immediately deletes it from the queue. If the consumer dies before finishing the work, that task vanishes.

A reliable system has to handle both scenarios. AMQP does this by closing each gap with a specific mechanism.

Publisher confirms: Closing the producer gap

Think of the publisher's confirmation as a receipt. It’s a mechanism that lets the producer know the broker has received and persisted the message. The broker only sends an acknowledgment back after the message is safely stored. If the broker crashes before sending that ack, the producer treats the message as unconfirmed and triggers a retry.

This logic is essentially a database COMMIT. The handshake ensures the message is never in "limbo". It’s either still with the producer or safely in the broker’s storage. There is no silent drop.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))

channel = connection.channel()

channel.confirm_delivery()

try:
    channel.basic_publish(
        exchange="",
        routing_key="tasks",
        body=b"do some work",
        properties=pika.BasicProperties(delivery_mode=2),  # persistent
    )
    print("Message confirmed by broker — safe on disk")
except pika.exceptions.UnroutableError:
    print("Message returned — no queue could receive it")

Confirms can also be used asynchronously for higher throughput: publish freely, track sequence numbers, and republish unconfirmed messages when the connection closes.

Durable queues and persistent messages: Surviving restarts

A confirmation only matters if the broker actually persists the message. In AMQP, durability is controlled at two levels: the queue and the message. A durable queue is saved to disk and survives a broker restart. A non-durable queue is gone when the broker stops.

LavinMQ writes all messages to disk regardless of delivery_mode, unlike some other brokers, there is no in-memory-only mode. The difference with non-durable queues is that those messages are flagged for deletion on restart. So for full durability, the queue must be declared durable:

channel.queue_declare(
    queue="tasks",
    durable=True,  # queue survives broker restart
)

channel.basic_publish(
    exchange="",
    routing_key="tasks",
    body=b"do some work",
    properties=pika.BasicProperties(
        delivery_mode=2,  # 2 = persistent, 1 = transient
    ),
)

You can still pass delivery_mode=2 for compatibility with other AMQP brokers, but in LavinMQ, the only thing that determines whether messages survive a restart is whether the queue is declared durable.

Consumer acknowledgments: Closing the consumer gap

Once a message is delivered to a consumer, the broker needs to know whether it was successfully processed. AMQP's consumer acknowledgment model handles this.

When a consumer receives a message, the broker marks it as "unacknowledged" and holds onto it internally. The consumer processes the work and sends an explicit ack. Only then does the broker discard the message. If the consumer crashes, disconnects, or has its channel closed for any reason before sending an ack, the broker requeues the message automatically and redelivers it to the next available consumer.

def on_message(channel, method, properties, body):
    try:
        process(body)
        channel.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        pass  # handle error if needed

The channel.basic_ack() method is key. If your code never reaches that line, maybe because of a crash or a power outage, the broker knows the work isn't done.

Consumer acknowledgments and retries

This isn't a retry policy you have to build yourself; it's just how the protocol works. You don't need a scheduler or a database table to track status, the broker handles the delivery state natively.

The broker tracks delivery state as a first-class concern. In-flight work cannot be lost without the consumer explicitly acknowledging it.

channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
channel.basic_consume(queue="tasks", on_message_callback=on_message)
channel.start_consuming()

Dead letter exchanges: Where failed messages go

When a message cannot be processed because it was rejected, exceeded its retry count, or sat in a queue past its time-to-live, AMQP provides a structured way to handle it: the dead letter exchange (DLX).

A queue can be configured with a dead letter exchange. When a message is dead-lettered, the broker routes it to that exchange, which in turn routes it to a dead letter queue. The original message is preserved along with metadata about why it was dead-lettered: the rejection reason, the original queue, and the number of times it was delivered.

Declare the dead letter queue and its exchange

channel.exchange_declare(exchange="dlx", exchange_type="direct")

channel.queue_declare(queue="tasks.failed", durable=True)

channel.queue_bind(queue="tasks.failed", exchange="dlx", routing_key="tasks")

# Declare the main queue, pointing failures at the DLX
channel.queue_declare(
    queue="tasks",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "dlx",
        "x-dead-letter-routing-key": "tasks",
    },
)

In the consumer, reject messages without requeueing so they are sent to the dead-letter queue:

def on_message(channel, method, properties, body):
    try:
        process(body)
        channel.basic_ack(delivery_tag=method.delivery_tag)
    except PermanentError:
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

This is basically a built-in "failed jobs" table. Instead of disappearing, failed messages land in a dedicated queue where you can actually deal with them. You can replay them manually, send them to an alerting system, or have a separate service handle the errors. The best part is that the broker handles all of this.

Prefetch and flow control: Preventing overload

One of the most important safety features in AMQP is the prefetch setting (also known as QoS). It limits the number of unacknowledged messages a broker will send to a single consumer at once.

Without prefetch, a broker will flood a consumer with as many messages as possible the moment it connects. If that consumer crashes while holding hundreds of messages, they all get thrown back into the queue at once, creating a massive spike. With prefetch, the broker sends messages in small, controlled batches, waiting for an ack before sending the next.

Deliver no more than 10 unacknowledged messages to this consumer at once

channel.basic_qos(prefetch_count=10)
channel.basic_consume(queue="tasks", on_message_callback=on_message)
channel.start_consuming()

This does two things. First, it limits the impact of a consumer crash; you don't have to wait for hundreds of messages to be requeued if one worker goes down. Second, it balances the load. Since a slow consumer can't hoard messages, the faster ones stay busy. Effectively, prefetch turns basic acknowledgments into a natural load balancer.

Heartbeats and connection safety

AMQP uses heartbeats to ensure both the client and the broker are still there. This is vital because TCP connections can fail silently; for example, a network hiccup might leave a connection "open" at the OS level even though no data is getting through.

Without heartbeats, a consumer could hang onto unacknowledged messages forever while the broker thinks it's still healthy. Heartbeats fix this: once a stale connection is detected, the broker kills it and immediately puts those messages back in the queue so someone else can process them.

connection = pika.BlockingConnection(
    pika.ConnectionParameters(
        host="localhost",
        heartbeat=60,  # send a heartbeat every 60 seconds
        blocked_connection_timeout=300,
    )
)

A heartbeat of 60 seconds means a dead connection is detected within two missed intervals. Any unacknowledged messages held by that consumer are requeued shortly after.

Summary: The built-in safety net

When you combine these features, you get a system where data safety isn't an afterthought. It’s the default:

• At the start: Publisher confirms ensure the broker actually has the message.
• At rest: Durable queues survive restarts.
• At the end: Consumer acks and DLXs ensure work is either finished or saved for later.
• On the wire: Prefetch and heartbeats protect against overloaded or dead workers.

You don't need to build a complex safety net on top of your broker using a database. The logic for reliability is already there, baked into the protocol. AMQP doesn't just move messages; it’s designed to make sure they actually get where they’re going, no matter what fails along the way.