There are two completely different things people call “a message queue,” and confusing them is the fastest way to fail this interview. One is a log - Kafka-shaped, an append-only tape where the broker is dumb, never deletes on read, and every consumer tracks its own position. The other is a broker - RabbitMQ-shaped, a smart mailbox where the broker owns each message, hands it to exactly one consumer, waits for an acknowledgment, and redelivers if that ack never comes. This post builds the second one, because the requirements name it: at-least-once delivery, priority queues, dead-letter queues. Those are broker features. A log does not track per-message delivery state, cannot prioritize one message over another sitting behind it, and has no notion of “this message was rejected three times, send it somewhere else.” A smart broker does.

The reason a smart broker is hard is the exact thing that makes it useful: the broker, not the consumer, remembers the delivery state of every single in-flight message. Who has message #5719, was it acked, when does its ack time out, how many times has it been redelivered, does it outrank the message next to it, and when it finally gives up where does it go to die. Multiply that bookkeeping by a million messages a second, make it survive a broker crash without losing or duplicating beyond the at-least-once promise, and you have the real problem. A naive “linked list plus a lock” dies at the first of those. Let me build it properly, end to end.

Functional Requirements (FR)

In scope:

  • Publish messages. A producer sends a message (a routing key, a body, headers, optional priority) to the broker, which routes it into one or more queues and durably stores it. This is the write path.
  • Consume messages. A consumer subscribes to a queue and receives messages one or a batch at a time, processes them, and acknowledges each. The broker holds a message as “unacknowledged” until the ack lands. This is the read path, and it is push-based (the broker pushes to a subscribed consumer up to a prefetch limit), not pull-and-poll.
  • At-least-once delivery. No acknowledged-as-published message is lost. If a consumer crashes mid-processing without acking, or explicitly rejects (nack), the message is redelivered - to that consumer or another. A message may therefore be delivered more than once; consumers must be idempotent. At-least-once is the contract.
  • Routing via exchanges. Producers publish to an exchange, not directly to a queue. The exchange routes to queues by binding rules - direct (exact routing-key match), topic (wildcard patterns), fanout (broadcast to all bound queues). One publish can land in zero, one, or many queues.
  • Priority queues. A queue can be declared with priority support; a higher-priority message jumps ahead of lower-priority messages already waiting, so urgent work is not stuck behind a backlog of bulk work.
  • Dead-letter queues (DLQ). A message that is rejected too many times, exceeds a TTL, or overflows a full queue is routed to a configured dead-letter exchange instead of being dropped, so failures are captured for inspection and replay rather than lost.
  • Durability. Queues, exchanges, bindings, and messages marked persistent survive a broker restart or crash. A confirmed publish is not lost.

Explicitly out of scope (name it so you own the scope):

  • Long-term retention and replay of history. This is a broker, not a log. A message is deleted once acknowledged; there is no “rewind and re-read last week.” If you need infinite retention and replay, you want the Kafka-style log design, which is a different post.
  • Stream processing / transformation. Windowing, joins, aggregation over the stream sit in a separate layer; we move and deliver messages, we do not process payloads.
  • Exactly-once end to end. We guarantee at-least-once plus the machinery for idempotent consumption. True exactly-once into an arbitrary external sink requires that sink to dedup or transact; we hand it a stable dedup key and say so.
  • Cross-datacenter federation. Mirroring queues across regions (federation / shovel) is a real feature but a separate design; we build a single cluster and note the extension.
  • A full AMQP protocol surface. We design the semantics (publish, route, deliver, ack, nack, priority, dead-letter) and note where a wire protocol like AMQP 0-9-1 maps, without specifying every frame.

The one decision that drives everything: this is a smart broker that owns per-message delivery state, not a dumb log that consumers read from. The broker deletes on ack, redelivers on failure, orders by priority, and dead-letters on giving up - so the entire design is about storing and mutating the state of millions of individual in-flight messages fast and durably. Every hard part below is a consequence of that reframe.

Non-Functional Requirements (NFR)

  • Scale: 1,000,000 messages/sec published and consumed at peak across the cluster, spread over thousands of queues (no single queue need carry the full rate; queues are the unit of parallelism). Average message ~1 KB. A busy single queue might sustain tens of thousands of messages/sec. Read rate roughly tracks write rate here (unlike a log with fan-out), because each message in a work queue is consumed once - fanout exchanges multiply it, but the common case is one delivery per message.
  • Latency: publish-confirm p99 in the low single-digit to tens of milliseconds for persistent messages; end-to-end (publish to delivered) in single-digit milliseconds when consumers are keeping up and the queue is near-empty (the fast path is memory-to-memory). A backed-up queue trades latency for buffering.
  • Availability: 99.9%+ on publish and consume. A broker node failure must not lose confirmed persistent messages and must fail a queue over to a replica within seconds. Producers and consumers reconnect transparently.
  • Consistency: within a single queue, near-FIFO order (priority reorders deliberately; redelivery and requeue perturb strict order, which we accept). A confirmed publish is durably committed before the confirm returns. Delivery state (acked / unacked / redelivered count) is authoritative on the broker.
  • Durability: very high for persistent messages. Once the broker returns a publisher confirm, the message is written to disk and replicated to a quorum of nodes, so it survives the loss of a node with zero loss. Transient (non-persistent) messages are best-effort and may be lost on crash - an explicit, per-message trade of durability for speed.
  • Delivery semantics: at-least-once is the contract - no confirmed message is lost, but redelivery after failure can duplicate. At-most-once (auto-ack, fire and forget) is available per-consumer for cases that tolerate loss.

Back-of-the-Envelope Estimation (BoE)

Real numbers, because they set the queue count, the memory budget, and the disk throughput.

Throughput and bandwidth:

Target:               1,000,000 messages/sec published at peak
Average message size: ~1 KB (routing key + headers + body)
Write bandwidth:      1M/sec * 1 KB = 1 GB/sec inbound
Consume rate:         ~1M/sec (work queues consume once) => ~1 GB/sec outbound
Total broker I/O:     ~2 GB/sec across the fleet (ingress + egress), plus
                      replication traffic (each persistent msg copied to peers).

Per-queue and node sizing:

Sustainable per-queue throughput ~ 20K-50K msgs/sec (enqueue + persist +
   replicate + dispatch + ack bookkeeping, all serialized through one queue's
   ordering point).
To hit 1M/sec:  1M / ~30K = ~33 busy queues minimum; real deployments spread
   across hundreds-to-thousands of queues, so the per-queue rate stays modest.
Per node: sustain ~50K-100K msgs/sec of persistent traffic (disk + replication
   bound). 1M/sec => on the order of 15-30 broker nodes, plus headroom.

In-flight (unacknowledged) memory - the scarce resource a log does not have:

The broker holds delivery state for every UNACKED message: message ref,
   consumer, delivery tag, redelivery count, ack-deadline. Say ~200 B of
   tracking state per unacked message (plus the message body if kept in RAM).
If consumers lag and 10M messages are in flight (unacked or queued):
   10M * 200 B tracking = ~2 GB of pure bookkeeping, plus
   10M * 1 KB bodies   = ~10 GB if bodies are held in memory.
=> Backlog is a memory problem. Bodies must be pageable to disk; only hot
   messages and the ack-tracking index stay resident. Size for the backlog,
   not the steady state.

Durable storage (the message store on disk):

Persistent messages are written to a shared message store / WAL before confirm.
Steady state a queue drains to ~empty, so on-disk backlog is small IF consumers
   keep up. Size the disk for a bad day: consumers down for 1 hour at 1M/sec:
   1M/sec * 3600s * 1 KB = ~3.6 TB of buffered messages for that hour.
   Replicated 3x: ~11 TB. Provision disk for the outage window you promise,
   not the steady state (which is near-zero).

The headline shape: steady-state throughput is ~1 GB/sec each way and disk stays near-empty because consumers drain queues, but the moment consumers fall behind the system becomes a memory-and-disk backlog problem, and the genuinely scarce resource is the per-unacked-message delivery state the broker must hold and mutate. Size for the backlog and the bookkeeping, not the happy path.

High-Level Design (HLD)

One logical broker made of a cluster of nodes. Producers connect and publish to exchanges; exchanges route by bindings into queues; queues live on (and are replicated across) broker nodes; consumers subscribe to queues and get messages pushed to them up to a prefetch limit, then ack. A metadata/coordination layer (a Raft quorum) holds the cluster topology: which node hosts the leader replica of each queue, exchange and binding definitions, and node liveness.

   PRODUCERS                         BROKER CLUSTER                          CONSUMERS
  ┌─────────┐   publish(exchange,   ┌──────────────────────────────────────┐
  │producer │─── routing_key, ─────▶│  EXCHANGE (routing)                   │
  │         │◀── body, persistent)  │   direct / topic / fanout             │
  └─────────┘   ▲ publisher-confirm │   binding rules: key/pattern -> queue │
                │                   └───────────────┬──────────────────────┘
                │                          routes to │ (0, 1, or many queues)
                │                    ┌───────────────┼──────────────────────┐
                │                    ▼               ▼                       ▼
   ┌────────────┴──────┐   ┌──────────────┐  ┌──────────────┐      ┌──────────────┐
   │ Node 1 (LEADER of │   │  QUEUE  orders   │  QUEUE  orders │      │  QUEUE (priority)
   │  queue A)         │   │  ready msgs      │  ready msgs   │      │  hi > lo ordering
   │  ┌ ready ring ┐   │   │  ┌ unacked map ┐ │               │      │
   │  │ m9 m8 m7.. │───┼──▶│  │ tag->msg,   │ │  push up to   │─────▶│  consumer C1
   │  └────────────┘   │   │  │ deadline,   │ │  prefetch N   │      │  process -> ack
   │  ┌ unacked map ┐  │   │  │ redeliveries│ │               │      │  or nack/reject
   │  │ tag->msg... │  │   │  └─────────────┘ │               │      └──────┬───────┘
   │  └─────────────┘  │   └──────────────────┘               │             │ ack(tag)
   │  message store    │                                       │             ▼
   │  (WAL on disk)    │──replicate persistent msgs──▶ Node 2 (FOLLOWER replica of A)
   └───────────────────┘                               Node 3 (FOLLOWER replica of A)
        │                                                     ▲
        │  ┌──────────────── DEAD-LETTER EXCHANGE ────────────┼───────────┐
        │  │  rejected>max / TTL-expired / overflow msgs -> DLQ for replay │
        │  └──────────────────────────────────────────────────────────────┘
        │  ┌──────────────── CONTROLLER (Raft quorum) ────────────────────┐
        └─▶│  queue leadership + replicas, exchange/binding defs, liveness │
           └───────────────────────────────────────────────────────────────┘

Publish flow (a producer writing a message):

  1. The producer opens a connection and a channel (a lightweight virtual connection multiplexed over one TCP socket) and publishes to an exchange with a routing key, a body, headers, and a delivery_mode (persistent or transient), optionally a priority.
  2. The exchange evaluates its bindings and routes the message to the matching queue(s): direct matches the routing key exactly, topic matches a wildcard pattern, fanout ignores the key and copies to all bound queues. If nothing matches, the message is dropped or sent to an alternate exchange (explicit policy, not silent loss if configured).
  3. For each target queue, the message is handed to that queue’s leader node. If persistent, the leader writes it to the message store (WAL) and replicates it to a quorum of replicas before returning a publisher confirm to the producer. Now an acked publish survives a node loss.
  4. The message enters the queue’s ready set (ordered FIFO, or by priority if a priority queue), waiting to be dispatched.

Consume flow (a consumer reading and acking):

  1. A consumer subscribes to a queue with a prefetch count (QoS) - the max number of unacknowledged messages the broker will push to it at once. This is the flow-control knob.
  2. The queue pushes ready messages to subscribed consumers, round-robin, up to each consumer’s prefetch. On dispatch, the message moves from ready to unacknowledged: the broker records (delivery_tag -> message, consumer, ack_deadline, redelivery_count) in the unacked map. It is not deleted; it is loaned.
  3. The consumer processes the message and sends ack(delivery_tag). The broker deletes the message (from memory and, for persistent, marks it removable in the store) and frees a prefetch slot, letting the next ready message flow.
  4. If the consumer sends nack/reject (with requeue) or its ack deadline passes or its connection drops, the broker requeues the message (increments its redelivery count) so it is redelivered. If the redelivery count exceeds the limit (or a TTL expires, or the queue is full), it is routed to the dead-letter exchange instead.

Failure flow: if a node hosting a queue leader dies, the controller promotes a follower replica (which already has the persistent messages and the committed delivery state) to leader within seconds; in-flight unacked messages on the dead node are requeued and redelivered (hence at-least-once, and hence possible duplicates). Producers and consumers reconnect to the new leader.

The structural insight: the broker is a stateful loan-and-collect machine - it durably stores each message, loans it to a consumer as unacknowledged, and either collects an ack (delete) or takes it back (redeliver or dead-letter). Priority reorders the ready set, the DLQ catches the give-ups, and replication makes the whole ledger survive a node death.

Component Deep Dive

Four hard parts, naive-first then evolved: (1) tracking per-message delivery state for at-least-once at a million messages/sec; (2) priority queues without starving or blowing up; (3) dead-letter queues and redelivery policy; (4) durability and failover of the message store and the delivery ledger.

1. Delivery state and at-least-once: fire-and-forget -> loan, ack, and redeliver

The core of a smart broker is remembering, for every in-flight message, whether it has been acknowledged - and redelivering the ones that were not. This is the stated hard part.

Naive approach: pop the message off the queue on delivery, done

Treat the queue as a list. On consume, pop the head and send it to the consumer. Move on.

deliver(): msg = queue.popHead(); send(consumer, msg)   # message now gone from broker

Where it breaks:

  • A consumer crash loses the message. The broker already removed it on delivery. If the consumer dies before finishing, or the message never arrives over the network, it is gone - that is at-most-once, and it violates the at-least-once contract outright.
  • No redelivery. There is no record that the message is outstanding, so there is nothing to resend on failure.
  • No way to reject. A consumer that hits a poison message (malformed body) cannot say “not me, requeue it or dead-letter it” - the broker no longer knows the message exists.

Pop-on-delivery is at-most-once by construction. At-least-once requires the broker to keep the message until it is proven consumed.

Evolved approach: loan the message (unacked), collect an ack, redeliver on timeout or nack

On delivery, do not delete. Move the message from a ready set to an unacknowledged map keyed by a delivery tag, recording the consumer, an ack deadline, and a redelivery count. Delete only on ack. Requeue on nack, deadline expiry, or consumer disconnect.

ready:   ordered set of messages waiting for a consumer
unacked: map  delivery_tag -> { msg, consumer, ack_deadline, redelivery_count }

deliver(consumer):
    if consumer.inflight >= consumer.prefetch: skip        # QoS flow control
    msg = ready.pop()
    tag = next_delivery_tag()
    unacked[tag] = { msg, consumer, now()+ack_timeout, msg.redeliveries }
    send(consumer, msg, tag)

on ack(tag):        del unacked[tag]; free prefetch slot; delete msg (persistent: tombstone)
on nack(tag, requeue=true):
                    m = unacked.pop(tag); m.redeliveries += 1; requeue_or_deadletter(m)
on ack_deadline pass OR consumer disconnect:
                    for tag of that consumer: nack(tag, requeue=true)   # redeliver

requeue_or_deadletter(m):
    if m.redeliveries > max_retries: route to dead_letter_exchange
    else: ready.insert(m)   # back into the ready set (front, to preserve rough order)

Why this is right:

  • At-least-once holds. A message is deleted only after a positive ack. Any failure mode - consumer crash, network drop, explicit reject, ack timeout - results in requeue and redelivery, so no confirmed message is lost. The cost is possible duplicates, which is the honest at-least-once trade.
  • Prefetch is the flow-control lever. The broker only loans up to prefetch unacked messages per consumer, so a slow consumer does not get buried and a fast one is kept busy. Prefetch 1 gives strict one-at-a-time hand-off (fair, low throughput); higher prefetch pipelines for throughput at the cost of larger in-flight state and coarser fairness.
  • The delivery tag is the whole contract. It is a per-channel monotonic id that lets ack/nack refer to a specific loan. ack(multiple=true) can ack everything up to a tag in one round trip, cutting ack overhead at high rates.
  • Redelivery count enables giving up. Tracking redeliveries per message is what lets the broker stop redelivering a poison message forever and dead-letter it instead (deep dive 3).

The consumer must be idempotent because at-least-once means it will occasionally see a duplicate: dedup on a message id or a business key, or make the side effect naturally idempotent (upsert, not insert). The broker guarantees no loss; the consumer guarantees no double effect. Say this explicitly - it is the contract split.

2. Priority queues: one FIFO line -> bucketed sub-queues merged on dispatch

A priority queue must let an urgent message overtake lower-priority messages already waiting. A plain FIFO cannot.

Naive approach: one list, re-sort on every insert

Keep a single list; on each publish, insert the message in sorted priority order (or re-sort the list).

publish(m): ready.insert_sorted_by_priority(m)     # O(n) insert or O(n log n) re-sort
deliver():  send(ready.popHead())

Where it breaks:

  • O(n) per insert kills throughput. At 1M/sec into a backed-up queue of millions of messages, an O(n) sorted insert (or a full re-sort) per publish is quadratic behavior; the enqueue path collapses. Even a heap is O(log n) per op with poor cache locality on a huge structure and does not preserve FIFO within a priority level.
  • Starvation of low priority. If high-priority traffic never stops, low-priority messages at the tail are never dispatched. A single sorted structure has no built-in fairness valve.
  • Unbounded priority levels. Allowing arbitrary integer priorities means arbitrary comparison cost and unclear semantics. Real systems bound it (e.g. 0-9 or 0-255).

A single re-sorted list conflates ordering-within-priority with ordering-across-priority and pays O(n) for it.

Evolved approach: a small fixed set of FIFO sub-queues, one per priority level

Declare the queue with a bounded number of priority levels (say 0-9). Maintain one FIFO sub-queue per level. Publish appends to the sub-queue for the message’s level in O(1). Dispatch scans from the highest non-empty level down and pops its head - so priority ordering is across buckets, FIFO order is preserved within a bucket, and both are O(1) amortized.

levels: array[0..9] of FIFO sub-queues        # index = priority
publish(m): levels[clamp(m.priority, 0, 9)].append(m)     # O(1)
deliver():
    for p from HIGH downto LOW:
        if not levels[p].empty(): return levels[p].popHead()   # O(#levels) = O(1) bounded
    return none

# anti-starvation: with probability epsilon, serve a lower level instead of the top,
# so low priority still drains under sustained high-priority load (weighted fairness).

Why this is right:

  • O(1) enqueue and bounded-O(1) dispatch. No sorting. Publish is a tail append to one of a fixed number of lists; dispatch is at most a scan of the (small, fixed) number of levels. This sustains the enqueue rate that a sorted list cannot.
  • FIFO preserved within a level. Messages of equal priority keep their arrival order, which is what “priority queue” actually means to users - urgent first, but fair among equals.
  • Bounded levels bound the cost. A fixed 0-9 (or 0-255) range makes the dispatch scan constant and the memory overhead a fixed handful of list headers, not a per-message comparator.
  • Starvation is a tunable policy, not an accident. The epsilon weighted-fairness valve (or aging: bump a message’s effective priority the longer it waits) guarantees low-priority messages eventually drain even under relentless high-priority load. Without it, priority queues starve; name the mitigation.
  • Interacts cleanly with prefetch and unacked. The unacked map and redelivery are unchanged - priority only reorders the ready set. A requeued high-priority message re-enters its level’s head, so it stays ahead. A redelivered message can even be bumped a level to get it retried sooner.

Bucketed FIFO sub-queues give real priority with O(1) hot-path cost and an explicit anti-starvation lever - the sorted-list naive version has neither.

3. Dead-letter queues and redelivery: retry forever -> bounded retries then dead-letter

A message can fail to process: poison payload, a downstream that is down, an expired TTL, a full queue. The system must not lose it and must not retry it forever.

Naive approach: requeue on every failure, indefinitely

On nack/reject or timeout, always put the message back at the head of the ready set and try again.

on failure(m): ready.insert_front(m)        # forever

Where it breaks:

  • Poison-message loops. A message that always fails (bad body the consumer cannot parse) is redelivered forever, and because it goes to the head, it blocks every consumer behind it - a single poison message can wedge the entire queue (a “poison pill”).
  • No visibility into failures. Failed messages vanish back into the queue; an operator has no place to inspect what is failing or why, and no way to replay them after fixing the bug.
  • Retry storms. Immediate requeue hammers a struggling downstream with no backoff, turning a transient outage into a tight failure loop.

Infinite requeue turns one bad message into a stalled queue and hides every failure.

Evolved approach: count redeliveries, dead-letter on exhaustion, with TTL and backoff

Bound retries with the redelivery count from deep dive 1. When a message exceeds max_retries, or its per-message TTL expires while waiting, or it is rejected with requeue=false, or the queue overflows its length limit, route it to the queue’s configured dead-letter exchange (DLX), which lands it in a dead-letter queue with headers explaining why. Add delayed retry for transient failures so retries back off instead of hammering.

on failure(m):
    m.redeliveries += 1
    reason = classify(m)          # rejected | ttl_expired | max_retries | overflow
    if m.redeliveries > max_retries or reason in {ttl_expired, overflow, reject_no_requeue}:
        m.headers["x-death"] += { reason, queue, count, time }   # audit trail
        publish(dead_letter_exchange, m)         # -> DLQ, NOT dropped
    else:
        # delayed retry: park in a delay queue with TTL = backoff(m.redeliveries),
        # whose own dead-letter target is the original queue -> reappears after the delay
        publish(delay_exchange, m, ttl = backoff(m.redeliveries))

# operator later inspects the DLQ, fixes the bug, and re-publishes messages
# from the DLQ back to the original exchange (replay).

Why this is right:

  • Poison messages are contained, not looped. After max_retries, the message leaves the hot queue for the DLQ, so it stops blocking healthy traffic. The queue keeps flowing; the bad message is quarantined, not lost.
  • Failures become inspectable and replayable. The DLQ is a real queue an operator can drain, inspect (the x-death headers say which queue, how many times, and why), and - after fixing the consumer - re-publish from. Nothing is silently dropped; every give-up is captured.
  • TTL gives time-bound messages an exit. A message that is only useful for 30 seconds carries a TTL; when it expires in the queue it dead-letters automatically instead of being delivered stale. This is how you express “this work is worthless if late.”
  • Delayed retry with backoff protects downstreams. The delay-queue trick (park in a queue whose TTL expiry dead-letters back to the original) turns immediate retries into exponential backoff, so a transient outage gets breathing room instead of a retry storm. The delay queue is itself just a queue with a dead-letter target - no new machinery.
  • Overflow is a policy, not a crash. A max-length queue that overflows dead-letters (or drops-head, by policy) the excess rather than growing unbounded and taking the node’s memory with it. Bounded queues plus DLQ is how backpressure stays safe.

Bounded redelivery plus a dead-letter exchange is what makes failure a first-class, observable, recoverable event instead of an infinite loop or a silent loss - and it is built entirely from the same publish/route/TTL primitives.

4. Durability and failover: one in-memory queue -> replicated log-backed queue with a persistent ledger

A confirmed persistent message must survive a node crash, and the queue must fail over without losing messages or the delivery ledger. This is where “smart broker” gets genuinely hard, because both the messages and their delivery state must be durable and consistent across nodes.

Naive approach: keep the queue in memory on one node, flush to disk lazily

Hold ready and unacked messages in RAM on the node that owns the queue; periodically snapshot to local disk.

publish(m): ready.append(m) in memory; return confirm      # before it hits disk
crash: whatever was not yet flushed is gone; the whole queue is on one node.

Where it breaks:

  • Confirmed messages are lost on crash. If the confirm is returned before the message is durably on disk, a crash between confirm and flush loses an acknowledged publish - a direct durability violation.
  • Single node is a single point of failure. The queue lives on one node. That node dies, the queue and every message in it are unavailable (or lost) until the node comes back, if it ever does.
  • The delivery ledger is lost too. Even if messages survive, the unacked map (who has what, redelivery counts) is in RAM. On failover a replacement node has no idea which messages were in flight, so it either loses them or blindly redelivers everything, unbounded.

One in-memory copy on one node satisfies neither durability nor availability.

Evolved approach: quorum-replicated queue backed by a Raft log; confirm only after commit

Model each durable queue as a replicated state machine over a Raft group of (typically) 3 or 5 nodes: one leader, the rest followers. Every state-changing operation - enqueue, deliver (message goes unacked), ack (delete), nack/requeue, dead-letter - is an entry in the queue’s Raft log, committed only when a majority has it on disk. The publisher confirm is returned only after commit. Message bodies go to a shared message store (WAL) on each replica; the queue’s operations reference them.

queue Q as a Raft group {leader N1, followers N2, N3}:
   client publish(m) -> leader appends log entry ENQUEUE(m) to its WAL
                     -> replicates to N2, N3; on majority fsync -> COMMITTED
                     -> apply to state machine (m enters ready); return CONFIRM
   deliver/ack/nack are ALSO replicated log entries:
      DELIVER(tag, consumer)  -> ready->unacked, deadline recorded
      ACK(tag)                -> delete message
      REQUEUE(tag)/DEADLETTER(tag)
   => followers hold the same messages AND the same delivery ledger.

failover: leader N1 dies -> Raft elects a new leader from N2/N3 that already has
   all committed entries -> it rebuilds ready/unacked from the log ->
   requeues messages that were unacked on the dead node -> resumes in seconds.

log compaction: periodically snapshot the queue state (ready + unacked) and
   truncate the log prefix, so the log stays bounded to ~current queue size,
   not the full history of every enqueue/ack.

Why this is right:

  • Confirmed = committed = durable across a node loss. The confirm waits for a majority to fsync the enqueue, so an acknowledged message is on at least two nodes’ disks. Any single node can die with zero loss. Transient messages skip replication and fsync for speed - the explicit durability-for-latency trade, chosen per message via delivery_mode.
  • The delivery ledger is replicated too, not just the messages. Because deliver/ack/nack are themselves replicated log entries, a promoted follower knows exactly which messages were unacked and their redelivery counts, so it redelivers precisely the in-flight set - no blind full redelivery, no lost acks. This is the piece the naive version drops and the reason at-least-once holds across failover.
  • Fast, correct failover. Raft guarantees the new leader has every committed entry, so promotion is a metadata operation (seconds), not a data copy. Consumers reconnect and continue; messages that were unacked on the dead leader are requeued (hence a possible duplicate, consistent with at-least-once).
  • Compaction bounds the cost. Snapshotting ready+unacked and truncating the log keeps storage proportional to the live queue depth, not to the lifetime volume - critical because a healthy queue enqueues and acks millions of messages that must not accumulate as log forever.
  • Shared message store keeps big bodies out of the Raft log. The log carries operations and references; bodies live in a segmented on-disk store shared across queues on the node, so a 1 MB body is written once and paged, not copied into every log structure. (This is the classic message-store-plus-index split.)

A quorum-replicated, log-backed queue is what turns “a queue in RAM on one box” into a durable, highly available broker: messages and their delivery state are committed to a majority before confirm, and failover is a leader election over a log both replicas already hold.

API Design & Data Schema

Publish / admin API (write and control plane)

PUT  /exchanges/{name}     # declare an exchange
  { "type": "topic", "durable": true, "alternate_exchange": "unrouted" }

PUT  /queues/{name}        # declare a queue
  { "durable": true, "max_priority": 9, "max_length": 1000000,
    "message_ttl_ms": 60000, "dead_letter_exchange": "dlx",
    "replicas": 3, "overflow": "dead-letter" }

PUT  /bindings            # bind a queue to an exchange by routing pattern
  { "exchange": "events", "queue": "orders", "routing_key": "order.*.created" }

PUBLISH  publish(exchange="events", routing_key="order.us.created",
                 body=<bytes>, headers={...}, priority=5,
                 delivery_mode="persistent", expiration_ms=60000,
                 message_id="9f2a...", mandatory=true)
  -> publisher confirm { acked: true, delivery_tag: 880453 }   # after quorum commit
     (or basic.return if mandatory and unroutable)

Consume / ack API (read plane)

CONSUME  consume(queue="orders", prefetch=50, auto_ack=false,
                 consumer_tag="c1")           # broker pushes up to 50 unacked
   -> deliver { delivery_tag, redelivered: bool, exchange, routing_key,
                body, headers, priority, message_id }        # pushed, repeatedly

ACK      ack(delivery_tag, multiple=false)    # delete; multiple=true acks up to tag
NACK     nack(delivery_tag, requeue=true)     # requeue for redelivery
REJECT   reject(delivery_tag, requeue=false)  # -> dead-letter (no requeue)
CANCEL   cancel(consumer_tag)                 # stop delivery; unacked get requeued

Data model: message store, queue index, delivery ledger, metadata

The hot path is not a relational table - it is an append-only message store plus in-memory indexes rebuilt from a replicated log. A general SQL database is the wrong tool here: the workload is high-rate sequential append with delete-on-ack and per-message state mutation, which a purpose-built store handles far better than row-per-message with indexes. Metadata (durable, low-volume, needs strong consistency) lives in the Raft-backed store.

Message store (per node, shared across queues, append-only segments):

dir: /data/msgstore/
   segment-000123.log     # length-prefixed message records, append-only
   record: { msg_id, crc, size, headers, body }   # body written once, referenced
   acked messages tombstoned; segments GC'd when fully acked (compaction).

Queue log + derived state (Raft group per durable queue):

raft log entries: ENQUEUE(msg_ref, priority) | DELIVER(tag, consumer)
                | ACK(tag) | REQUEUE(tag) | DEADLETTER(tag, reason)
derived state machine (in memory, rebuilt from log on recovery):
   ready:   priority-bucketed FIFO of msg_refs      # levels[0..max_priority]
   unacked: map delivery_tag -> { msg_ref, consumer, ack_deadline, redeliveries }
snapshot: { ready[], unacked{} } taken periodically -> truncate log prefix

Delivery ledger entry (the per-unacked-message state, the scarce resource):

{ delivery_tag: uint64, msg_ref, consumer_tag, ack_deadline_ts,
  redelivery_count: uint8, priority: uint8 }        # ~40-200 B each, held per in-flight msg

Cluster metadata (Raft quorum, strongly consistent):

exchanges: name -> { type, durable, bindings[], alternate_exchange }
queues:    name -> { durable, max_priority, max_length, ttl, dlx,
                     leader_node, replica_nodes[], overflow_policy }
nodes:     node_id -> { host, status, last_heartbeat }

Why this storage split:

StoreHoldsWhy this choice
Append-only message storemessage bodiessequential append, write-once body, cheap segment GC on ack
Per-queue Raft log + snapshotenqueue/deliver/ack ops and derived ready+unackedreplicates messages AND delivery state; quorum commit = durability
In-memory delivery ledgerper-unacked-message statemust be mutated per ack at 1M/sec; rebuilt from the log on failover
Raft metadata quorumexchanges, bindings, queue defs, leadershipsmall, strongly consistent, needed for routing and failover

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each.

1. A single queue is a serialization point. Every message in a queue funnels through one ordering point (and one Raft leader), capping it at tens of thousands of msgs/sec. Fix: spread load across many queues - queues are the unit of parallelism. Shard a logical stream across N queues by a partition key (hash(key) % N) with a consistent-hash exchange, and run consumers per queue. Order is preserved per queue (per key), not globally - the same trade a log makes with partitions.

2. Backlog blows up memory when consumers lag. Unacked plus ready messages held in RAM (bodies + ledger) can exhaust the node. Fix: page message bodies to disk and keep only the ledger and hot messages resident; bounded queues (max_length) with overflow to DLQ or drop-head so a queue cannot grow without limit; alarm on queue depth and on memory watermark (block publishers when memory is high - flow control backpressure to producers).

3. Hot queue / hot consumer skew. One queue or one partition key gets a disproportionate share, hotspotting a node. Fix: higher-cardinality or salted partition keys so traffic spreads; more consumers on the hot queue with higher aggregate prefetch; detect via per-queue rate and depth metrics and rebalance queue leaders across nodes.

4. Ack overhead at high rate. One network round trip and one replicated log entry per message doubles the op rate. Fix: batch acks (ack multiple=true acks everything up to a tag in one call), batch enqueue commits in the Raft leader (group-commit fsync amortizes disk cost across many messages), and tune prefetch so consumers pipeline rather than stop-and-wait.

5. fsync / replication latency on the confirm path. Waiting for a majority fsync per persistent publish adds latency. Fix: group commit (one fsync flushes many messages), pipeline replication, and offer transient delivery mode for traffic that tolerates loss (skip fsync and replication). Persistent vs transient is a per-message durability-latency choice.

6. Poison messages wedge a queue. A message that always fails, requeued to the head, blocks everything behind it. Fix: bounded redelivery + dead-letter exchange (deep dive 3) so the poison message leaves the hot path after max_retries; delayed retry with backoff so retries do not storm.

7. Priority starvation. Relentless high-priority traffic starves low-priority messages. Fix: weighted fairness (serve a lower level with probability epsilon) or aging (raise a waiting message’s effective priority over time) so low priority still drains (deep dive 2).

8. Queue-leader node failure loses availability. The node hosting a queue’s Raft leader dies. Fix: quorum replication (RF 3/5) with Raft (deep dive 4) - a follower with all committed entries is promoted in seconds; unacked messages are requeued (at-least-once, possible duplicate). Spread queue leaders evenly so one node death does not move all leadership at once.

9. Controller / metadata as a single point of failure. The layer holding leadership, bindings, and liveness must not be a single node. Fix: run the controller as a Raft quorum (3 or 5 nodes); keep its state small (topology only, not messages) so it stays fast and always available for routing and failover decisions.

10. Connection and channel scale. Millions of producers/consumers means millions of TCP connections and heartbeats per node. Fix: multiplex channels over one connection, cap connections per node and front with load balancers / a connection-proxy tier, and offload TLS. Heartbeat tuning so a GC pause is not misread as a dead connection and does not trigger spurious redelivery.

Shard key, stated plainly: the cluster shards by queue, and a high-volume logical stream shards across queues by a partition key via hash(key) % N (consistent-hash exchange). The queue is simultaneously the unit of ordering (near-FIFO within a queue), the unit of durability (one Raft group per queue), and the unit of parallelism (one leader, consumers per queue) - so picking the partition key well is what lets a smart broker scale past a single node while keeping per-key order and at-least-once intact.

Wrap-Up

The trade-offs that define this design:

  • Smart broker over dumb log. The broker owns per-message delivery state - loan, ack, redeliver, dead-letter - which buys priority, per-message retries, and DLQs that a log cannot express, at the cost of no long-term retention or replay. We picked the broker because the requirements (at-least-once, priority, DLQ) are broker features. If you needed replay and infinite retention, you would pick the log instead.
  • At-least-once, made honest. A message is deleted only after a positive ack, and every failure requeues, so nothing confirmed is lost - but redelivery can duplicate, so consumers must be idempotent. The broker guarantees no loss; the consumer guarantees no double effect. That split is the contract.
  • Durability via a replicated log, not a single RAM copy. Each durable queue is a Raft group; enqueue/deliver/ack are replicated log entries committed to a majority before the confirm, so both the messages and the delivery ledger survive a node death and failover is a leader election in seconds. Transient mode trades that durability for latency, per message.
  • Priority and dead-lettering built from simple primitives. Priority is bucketed FIFO sub-queues with an anti-starvation valve (O(1), not a re-sorted list); dead-lettering is bounded redelivery plus a dead-letter exchange plus TTL and backoff (contain poison, capture failures, never loop forever). Both reuse the same publish/route/ack machinery rather than adding new subsystems.
  • Queues as the scaling unit. Throughput comes from spreading load across many queues sharded by a partition key, preserving per-queue order while scaling past one node - accepting that global ordering is given up, exactly as a partitioned log does.

One-line summary: a durable, quorum-replicated distributed broker where each queue is a Raft-backed state machine that loans every message to a consumer as unacknowledged and redelivers on failure for at-least-once delivery, orders its ready set by bucketed priority with an anti-starvation valve, dead-letters poison and expired messages to a configured exchange instead of losing or looping them, and scales to 1M messages/sec by sharding load across many independently-replicated queues.