Delivery guarantees and idempotent consumers

The problem: the counts don't add up

Snip's click counter reads click events from the log in batches, adds them to the totals in Postgres, and saves its offset. One evening a worker is restarted during a deploy, and an owner notices that a link's click count jumped by about 500 more than its traffic graph shows. Nothing was clicked twice. A batch was counted twice.

Every messaging system has to answer the same question: if something crashes partway through handling a message, is the message tried again (and possibly handled twice), or not (and possibly lost)?

The three guarantees

Kafka's documentation defines them plainly:

• At most once: messages may be lost but are never redelivered.
• At least once: messages are never lost but may be redelivered.
• Exactly once: each message is processed once and only once.

Duplicates come from two places. A producer that sends a message and doesn't hear back can't tell whether it arrived, so it sends again. And a consumer that does its work and then crashes before recording that it's done will be handed the same message again. Losses come from the opposite order: recording "done" before doing the work.

Try it

Without crashes, every order gives the right answer, which is why these bugs survive testing. With a crash between the two steps, "save, then count" loses a batch, and "count, then save" counts it twice. That's at most once and at least once. Most systems choose at least once, because a lost click can't be recovered, and then make the duplicate harmless.

"Exactly once" means exactly-once effects

No network can guarantee a message is delivered exactly once: the receiver can always crash just after doing the work and just before saying so. What systems can offer is that the effect happens once, even when the message is seen twice. Two kinds of help exist:

• Inside the messaging system. Since version 0.11, Kafka's producer can be idempotent: the broker gives each producer an ID and drops resent messages by sequence number, so producer retries don't create duplicates in the log. Kafka transactions let a consumer read, write its output to other topics and save its offset as one atomic step. That's exactly-once within Kafka.
• At the edges, in your code. As soon as the effect is outside the log (a row in Postgres, an email, a charge), the consumer has to make it safe itself. Kafka's documentation suggests the classic approach: store the consumer's position in the same place as its output.

Making consumers safe to repeat

1. Store the offset with the result. The click counter keeps its offset in Postgres, next to the counts, and updates both in one transaction. A redelivered batch finds its offsets already counted and skips them:

type Click = { offset: number; code: string };

export async function applyBatch(db: pg.Pool, partition: number, batch: Click[]) {
  const client = await db.connect();
  try {
    await client.query("BEGIN");
    const { rows } = await client.query(
      "SELECT next_offset FROM consumer_offsets WHERE consumer = 'click-counter' AND partition = $1 FOR UPDATE",
      [partition],
    );
    const fresh = batch.filter((m) => m.offset >= Number(rows[0].next_offset)); // skip what's already counted
    if (fresh.length > 0) {
      const totals = new Map<string, number>();
      for (const m of fresh) totals.set(m.code, (totals.get(m.code) ?? 0) + 1);
      for (const [code, n] of totals) {
        await client.query(
          `INSERT INTO click_counts (code, clicks) VALUES ($1, $2)
           ON CONFLICT (code) DO UPDATE SET clicks = click_counts.clicks + $2`,
          [code, n],
        );
      }
      await client.query("UPDATE consumer_offsets SET next_offset = $2 WHERE consumer = 'click-counter' AND partition = $1", [
        partition,
        fresh[fresh.length - 1].offset + 1,
      ]);
    }
    await client.query("COMMIT");
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

Tested against Postgres with the same batch delivered twice, overlapping batches, and three copies of a batch applied at once: the counts came out exactly right every time. The FOR UPDATE lock on the offset row makes concurrent deliveries of one partition wait their turn. On startup, the consumer reads its position from this table, not from the broker.

2. Remember processed message IDs. For a queue with no offsets, record each message's ID in a table with a unique constraint, in the same transaction as the effect: INSERT INTO processed (message_id) VALUES ($1) ON CONFLICT DO NOTHING. If no row was inserted, it's a repeat; skip it.

3. Make the operation naturally idempotent. "Set the link's status to blocked" can run twice harmlessly; "add 1" can't (03.10 Idempotency keys). Where you can, send states, not changes.

4. Pass an idempotency key to the next system. An email or payment provider that accepts an idempotency key can drop the repeat for you. Use the event's ID as the key.

Effects with no undo and no key, like a plain SMTP email, can't be made exactly once. Pick which way to fail: a duplicate email is usually better than a missing one.

Messages that always fail

At least once has a nasty corner: a message that crashes the consumer every time is redelivered for ever. Count attempts, and after a few failures move the message to a dead-letter queue with the error, alert someone, and carry on (07.04 Partitions, consumer groups and ordering covers how this interacts with ordering).

Recap

Interview questions