CS Fundamentals Trainer · Guides

Message queues and delivery guarantees

At-most-once, at-least-once and exactly-once in practice: the transactional outbox, idempotent consumers, visibility timeouts and dead-letter queues, run on PostgreSQL.

14 min read

A message queue lets one part of a system hand work to another without waiting for it. The producer puts a message in; a consumer takes it out later. That decoupling gives you smoothing of traffic spikes, retries, and independent scaling, and it creates a new problem: networks and processes fail between the steps, so a message may be lost or delivered twice. This guide covers what a queue gives you, what "exactly once" really means, and the patterns (outbox, idempotent consumer, dead-letter queue) that make a system correct anyway. The two key patterns were run for real on PostgreSQL, and the results are shown.

1. Why use a queue

Diagram: A producer, a queue, and consumers working in parallel

The costs: more moving parts, no immediate result, messages out of order sometimes, duplicates, and a harder time debugging a flow that spans services. Do not add a queue where a plain function call or a database table would do.

2. Queues and logs

Two designs share the word "queue":

Work queue (SQS, RabbitMQ)Log (Kafka, Kinesis)
A message isremoved once handledkept for a retention period; consumers track their position (offset)
Several independent consumerseach needs its own queueeach consumer group reads the same data
Replay old datano (unless you saved it)yes, rewind the offset
Orderingusually best-effort, or per message group / FIFO queueordered within a partition
Typical usebackground jobsevent streams, data pipelines, event sourcing

For ordering, a log gives order within one partition, and you choose the partition by a key (all events of one order go to one partition). There is no total order across partitions, so design as if there were none.

3. Delivery guarantees

What happens when the consumer or the network fails between "I received it" and "I finished it"? There are three possible contracts [1]:

The uncomfortable truth: between two machines connected by an unreliable network, you cannot make delivery itself exactly-once, because the sender can never tell whether a lost acknowledgement means "not received" or "received, ack lost". What systems offer is exactly-once processing (or effect), built from at-least-once delivery plus de-duplication, or from a transaction that includes both the read of the message and the write of its result. Kafka, for example, provides idempotent producers and transactions that give exactly-once semantics inside Kafka (read-process-write between Kafka topics) [1], but once your consumer calls an outside system (an email service, a payment API), that call is again at-least-once unless that system supports idempotency too.

So the working rule is:

at-least-once delivery + an idempotent consumer = effectively exactly once

4. The visibility timeout (how a work queue handles a crash)

In a queue like SQS, when a consumer receives a message the message is not deleted: it becomes invisible to other consumers for the visibility timeout. If the consumer finishes, it deletes the message. If it crashes or takes too long, the timeout expires, the message becomes visible again, and another consumer gets it [2]. This gives at-least-once delivery. Consequences:

5. The dual-write problem and the outbox pattern

A very common bug: a service saves an order to its database and publishes an OrderPlaced event.

BEGIN; INSERT INTO orders ...; COMMIT;     -- 1
publish("order.placed", ...)               -- 2   <- the process dies here: the order exists, no event

Swap the order, and a crash gives an event for an order that was never saved. Two systems cannot be updated atomically with a plain write to each. The fix is the transactional outbox: write the event to an outbox table in the same database transaction as the business row, and have a separate relay read the table and publish [3].

Diagram: Transactional outbox: the event is saved with the order, then relayed

I ran this on PostgreSQL: three orders were placed, each with an outbox row in the same transaction. The first relay run crashed after publishing, before marking the row as published (the worst moment). It restarted and ran again. Result: the broker held 4 messages for 3 distinct ids, a duplicate, exactly as at-least-once predicts, and no order was lost.

The tables, and the relay that moves rows from the outbox to the broker (Python with psycopg; publish stands for your broker client):

CREATE TABLE orders    (id serial PRIMARY KEY, item text);
CREATE TABLE outbox    (id bigserial PRIMARY KEY, topic text, payload jsonb, published_at timestamptz);
CREATE TABLE processed (message_id text PRIMARY KEY);
CREATE TABLE shipments (order_id int, item text);
def place_order(conn, item):
    with conn.transaction():                       # one transaction: order and event together
        oid = conn.execute("INSERT INTO orders (item) VALUES (%s) RETURNING id", (item,)).fetchone()[0]
        conn.execute("INSERT INTO outbox (topic, payload) VALUES ('order.placed', %s)",
                     (json.dumps({"order_id": oid, "item": item}),))
    return oid

def relay_once(conn, publish):
    with conn.transaction():                       # the row locks last until commit
        rows = conn.execute(
            "SELECT id, topic, payload FROM outbox WHERE published_at IS NULL "
            "ORDER BY id LIMIT 10 FOR UPDATE SKIP LOCKED").fetchall()
        for id_, topic, payload in rows:
            publish(f"outbox-{id_}", topic, payload)   # may repeat if we crash before the UPDATE
            conn.execute("UPDATE outbox SET published_at = now() WHERE id = %s", (id_,))

The select, the publish and the update sit in one transaction, so the lock taken by FOR UPDATE is held until the rows are marked. If the relay dies between publish and the update, the transaction rolls back, the rows are unpublished again, and the next run publishes them a second time: that is the duplicate in the experiment.

The relay's query on its own claims rows without blocking other relays:

SELECT id, topic, payload
FROM outbox
WHERE published_at IS NULL
ORDER BY id
LIMIT 10
FOR UPDATE SKIP LOCKED;

FOR UPDATE SKIP LOCKED locks the rows being handled and makes other relay instances skip them instead of waiting, so several relays can run in parallel without taking the same row. (Rows that were skipped may be published slightly out of order, so consumers must not depend on strict ordering across rows.) An alternative to polling is change data capture (CDC): read the database's own write-ahead log and publish from it; tools such as Debezium do this [3].

6. The idempotent consumer

An operation is idempotent if doing it twice has the same effect as doing it once. "Set the status to shipped" is idempotent; "add 10 to the balance" is not. For the second kind, make it idempotent by remembering which messages you have handled. Give every message a unique id (the outbox row id works), and record it in the same transaction as the effect:

BEGIN;
INSERT INTO processed (message_id) VALUES ($1) ON CONFLICT DO NOTHING RETURNING message_id;
-- if no row came back: it was a duplicate, ROLLBACK or just COMMIT and stop
INSERT INTO shipments (order_id, item) VALUES ($2, $3);
COMMIT;

In my run the consumer handled the four messages in order and returned: shipped; duplicate (skipped); shipped; shipped. That is 3 shipments for 3 orders, despite the duplicate. Because the id and the shipment are in one transaction, a crash cannot leave one without the other.

Here is the consumer in code (pick a language on the tabs; your choice is remembered):

def handle(conn, message_id, order_id, item):
    with conn.transaction():
        row = conn.execute(
            "INSERT INTO processed (message_id) VALUES (%s) "
            "ON CONFLICT DO NOTHING RETURNING message_id",
            (message_id,),
        ).fetchone()
        if row is None:
            return "duplicate"                         # already handled: do nothing
        conn.execute("INSERT INTO shipments (order_id, item) VALUES (%s, %s)", (order_id, item))
        return "shipped"
func handle(ctx context.Context, pool *pgxpool.Pool, messageID string, orderID int, item string) (string, error) {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return "", err
	}
	defer tx.Rollback(ctx) // no-op after Commit

	tag, err := tx.Exec(ctx,
		`INSERT INTO processed (message_id) VALUES ($1) ON CONFLICT DO NOTHING`, messageID)
	if err != nil {
		return "", err
	}
	if tag.RowsAffected() == 0 {
		return "duplicate", nil
	}
	if _, err := tx.Exec(ctx,
		`INSERT INTO shipments (order_id, item) VALUES ($1, $2)`, orderID, item); err != nil {
		return "", err
	}
	return "shipped", tx.Commit(ctx)
}
// node-postgres; pool is a pg.Pool
async function handle(pool, messageId, orderId, item) {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const r = await client.query(
      "INSERT INTO processed (message_id) VALUES ($1) ON CONFLICT DO NOTHING",
      [messageId]
    );
    if (r.rowCount === 0) {
      await client.query("ROLLBACK");
      return "duplicate";
    }
    await client.query("INSERT INTO shipments (order_id, item) VALUES ($1, $2)", [orderId, item]);
    await client.query("COMMIT");
    return "shipped";
  } catch (e) {
    await client.query("ROLLBACK");
    throw e;
  } finally {
    client.release();
  }
}
// Npgsql
static async Task<string> Handle(NpgsqlDataSource db, string messageId, int orderId, string item)
{
    await using var conn = await db.OpenConnectionAsync();
    await using var tx = await conn.BeginTransactionAsync();

    await using var dedupe = new NpgsqlCommand(
        "INSERT INTO processed (message_id) VALUES (@m) ON CONFLICT DO NOTHING", conn, tx);
    dedupe.Parameters.AddWithValue("m", messageId);
    if (await dedupe.ExecuteNonQueryAsync() == 0)
    {
        await tx.RollbackAsync();
        return "duplicate";
    }

    await using var ship = new NpgsqlCommand(
        "INSERT INTO shipments (order_id, item) VALUES (@o, @i)", conn, tx);
    ship.Parameters.AddWithValue("o", orderId);
    ship.Parameters.AddWithValue("i", item);
    await ship.ExecuteNonQueryAsync();
    await tx.CommitAsync();
    return "shipped";
}

When the effect is on an outside system (send an email, charge a card), pass the message id as an idempotency key if the system supports one, so the second call is recognised as a repeat. The processed table grows forever, so delete old ids after a retention period longer than the longest redelivery you expect.

7. Failures, poison messages and the dead-letter queue

A poison message is one that always fails (bad data, a bug). Without a limit it is retried forever and blocks the queue behind it (in an ordered queue) or burns CPU. The standard defences:

8. Ordering, backpressure and monitoring

9. Questions interviewers like to add

"Can you get exactly-once delivery?"

Not across an unreliable network as a property of delivery alone. You can get exactly-once effect with at-least-once delivery plus idempotent handling, or with transactions that cover both the message offset and the result (as Kafka does inside its own topics).

"Why not just publish to the broker inside the database transaction?"

A database transaction cannot roll back a message already sent to a broker, and the broker cannot be part of the database's commit. That is the dual-write problem; the outbox moves the event into the database so one commit covers both.

"Queue or database table for jobs?"

A table with FOR UPDATE SKIP LOCKED is a perfectly good queue at moderate volume and gives you transactions with your data for free. A broker is better for very high throughput, fan-out to many consumers, or when you want a replayable log.

"How do you keep order with several consumers?"

Partition by a key so that everything for one entity goes to one consumer at a time, and give up global order.

"What if the consumer is slow?"

Scale consumers (up to the number of partitions in a log), make the work cheaper, split slow and fast jobs into different queues, and use backpressure upstream.

10. Things to mention in an interview

Quick recap

  1. A queue decouples producers and consumers, absorbs spikes and enables retries.
  2. Networks fail between steps, so delivery is at-most-once or at-least-once; "exactly once" is built on top.
  3. Acknowledge after processing and expect duplicates.
  4. Use the outbox so a state change and its event are saved atomically.
  5. Make consumers idempotent: record the message id in the same transaction as the effect.
  6. Limit retries, back off, and send poison messages to a dead-letter queue.
  7. Watch queue depth and the age of the oldest message.

Sources

  1. Confluent, Kafka design: delivery semantics. At most once, at least once, exactly once; idempotent producers and transactions.
  2. AWS, Amazon SQS visibility timeout.
  3. Chris Richardson, Pattern: Transactional outbox and Pattern: Transaction log tailing.
  4. PostgreSQL documentation, The Locking Clause: FOR UPDATE ... SKIP LOCKED.
  5. PostgreSQL documentation, INSERT ... ON CONFLICT.
  6. Debezium, Outbox event router.
  7. Martin Kleppmann, Designing Data-Intensive Applications, O'Reilly, 2017, chapter 11 (stream processing) and chapter 9 (consistency and consensus).

Practise this topic: Distributed Systems & Messaging System Design

More guides