Redis streams, and the broker you might not need

The list had been the queue for three years and it had lost messages twice, both times during a deploy. Nobody could prove it at the time because there was no record — that is the entire problem with a list. Redis 5.0 shipped in October with a data type that keeps one.

The symptom

$ redis-cli LLEN jobs
(integer) 0

$ grep -c 'job.started'   /var/log/app/worker.log
41208
$ grep -c 'job.completed' /var/log/app/worker.log
41194

# 14 jobs started and never finished. the list is empty.
# there is nothing to retry, because BRPOP removed them.

Fourteen orders that never had a receipt sent, discovered by counting log lines. The list itself has no state to inspect — a message is either in it or it is gone, and “in flight” is not a thing it can represent.

Why it happens

// the classic list queue
$redis->rPush('jobs', json_encode($payload));

$raw = $redis->blPop('jobs', 5);    // atomically removes it
$this->handle(json_decode($raw[1], true));
// ^ a crash here loses the message. permanently.

BLPOP is atomic and destructive, which is the right primitive for a work queue with reliable consumers and the wrong one for consumers that can die. The RPOPLPUSH pattern — moving the item to a processing list — is the traditional mitigation and it works, at the cost of maintaining the processing list, a recovery timer and a way to distinguish one consumer’s in-flight items from another’s. That is a broker, hand-rolled.

The fix

XADD, and reading without removing

> XADD orders '*' order_id 91204 total 4900
"1539172800123-0"

> XLEN orders
(integer) 1

> XRANGE orders - + COUNT 2
1) 1) "1539172800123-0"
   2) 1) "order_id" 2) "91204" 3) "total" 4) "4900"

# reading did not remove it. the entry is still there.

The * asks Redis to generate the id from the current millisecond plus a sequence number, which makes ids sortable, unique and roughly meaningful as timestamps. Entries persist after being read, which is what makes replay and multiple independent consumers possible — and also means the stream grows forever unless something trims it.

The fields are flat key-value pairs rather than a nested structure, so anything with shape is still your own serialisation problem. Storing a JSON blob in one field works and gives up the ability to filter, which nothing in Redis does anyway — so it is a smaller loss than it appears.

Consumer groups and the pending entries list

// once, at deploy
$redis->rawCommand('XGROUP', 'CREATE', 'orders', 'workers', '0', 'MKSTREAM');

// the worker loop
while (true) {
    $batch = $redis->rawCommand('XREADGROUP',
        'GROUP', 'workers', $this->name,
        'COUNT', 10, 'BLOCK', 5000,
        'STREAMS', 'orders', '>');

    foreach ($batch[0][1] as $entry) {
        list($id, $fields) = $entry;

        $this->handle($this->toArray($fields));

        $redis->rawCommand('XACK', 'orders', 'workers', $id);
    }
}

The > means “entries never delivered to this group”, which is what distributes work between members. An entry delivered and not acknowledged stays in the pending entries list indefinitely — that list is the whole mechanism, and it is what the list-based queue could not represent.

BLOCK is the other thing worth having: the worker waits rather than polling, so the latency floor is delivery time rather than the poll interval and the idle cost is nothing. Forgetting XACK produces a system that works perfectly and accumulates an unbounded pending list, which is a slow leak with no symptom until somebody looks.

Claiming from a worker that died

// on a timer, in every worker
$stale = $redis->rawCommand('XPENDING', 'orders', 'workers',
    'IDLE', 60000, '-', '+', 10);

foreach ($stale as $entry) {
    list($id, $consumer, $idleMs, $deliveries) = $entry;

    if ($deliveries > 5) {
        $this->deadLetter($id);
        $redis->rawCommand('XACK', 'orders', 'workers', $id);
        continue;
    }

    $redis->rawCommand('XCLAIM', 'orders', 'workers', $this->name, 60000, $id);
}

This is the recovery that the list queue could not have, and in Redis 5 it is code you write — XAUTOCLAIM arrives in 6.2 and collapses it into one command. The IDLE filter on XPENDING is what makes it cheap; without it you fetch everything pending and filter in PHP.

The delivery counter is the part that prevents an infinite loop. An entry that has been claimed five times is one that kills whatever picks it up, and continuing to reclaim it means every worker dies in turn. Acknowledging it after recording it elsewhere is the equivalent of burying, and there is no built-in dead-letter state to do it for you.

Trimming, because a stream does not empty itself

// exact, and O(n) — do not do this on a hot path
$redis->rawCommand('XTRIM', 'orders', 'MAXLEN', 100000);

// approximate, trims whole nodes, effectively free
$redis->rawCommand('XTRIM', 'orders', 'MAXLEN', '~', 100000);

// or on write, which is where it belongs
$redis->rawCommand('XADD', 'orders', 'MAXLEN', '~', 100000,
    '*', 'order_id', $id);

The ~ is not optional in practice. Exact trimming has to walk to find the boundary; approximate trimming removes whole macro nodes and stops when the next one would take it below the limit, which is bounded work. The result is a stream slightly longer than the limit, which is exactly what anyone wants.

Verifying it worked

$ php artisan stream:seed 10000
$ for i in 1 2 3; do php artisan stream:work & done

# kill one mid-flight
$ kill -9 $(pgrep -f stream:work | head -1)

$ redis-cli XPENDING orders workers
1) (integer) 7                    # seven in flight on a dead consumer
2) "1539172800123-0"

# 60 seconds later, after the claim timer
$ redis-cli XPENDING orders workers
1) (integer) 0

$ mysql -Nse 'SELECT COUNT(*), COUNT(DISTINCT order_id) FROM receipts_sent'
10000  10000                      # every one, exactly once

Killing a consumer mid-message and watching another finish the work is the assertion the whole change exists to make. The count and the distinct count being equal is the other half — recovery that delivers a message twice is only acceptable because the handler is idempotent, and this is the check that says it actually is.

What this costs

Another thing in Redis, and Redis was already load-bearing. A cache under memory pressure with allkeys-lru will happily evict a stream, which is data loss rather than a cache miss — so this wants its own instance with its own persistence settings, and that is a second thing to run and monitor rather than a free addition to something already there.

The larger cost is that the pending entries list, the claim timer, the delivery counter and the dead-letter handling are all yours to write, and that is perhaps eighty lines that a real broker ships. For a team already running Redis and needing one reliable queue, that is a good trade against operating RabbitMQ. For a team that needs routing, fanout or delayed delivery, it is eighty lines on the way to reimplementing something that already exists.