Redis Streams and Consumer Groups: Crashes, Re-delivery and Dead Letters

Pub/Sub delivers a message to whoever is listening at that moment and keeps nothing. A stream is the opposite: an append-only log stored in Redis, which readers can replay from any point and which tracks what each consumer has finished.

Setup

The same Redis and Python image as cache-aside, on one Docker network:

# setup.sh - Redis and a Python image with redis-py, on one Docker network
docker network create redis-lab > /dev/null 2>&1
docker rm -f redis-lesson > /dev/null 2>&1
docker run -d --name redis-lesson --network redis-lab -p 6379:6379 redis:8.10.2-alpine > /dev/null
printf 'FROM python:3.13-slim\nRUN pip install --no-cache-dir redis==8.1.0\n' \
  | docker build -q -t redis-py - > /dev/null
sleep 1
docker exec redis-lesson redis-cli INFO server | grep redis_version
docker run --rm redis-py python -c \
  "import redis, sqlite3, sys; print('redis-py', redis.__version__, '| sqlite', sqlite3.sqlite_version, '| python', sys.version.split()[0])"

$ bash setup.sh
redis_version:8.10.2
redis-py 8.1.0 | sqlite 3.46.1 | python 3.13.16

The shell examples call redis-cli through docker exec. The Python ones run in the image, through this shortcut (in Git Bash on Windows, use "$(pwd -W)" in place of "$PWD"):

py() { docker run --rm --network redis-lab -v "$PWD":/app -w /app redis-py python "$@"; }

XADD and XRANGE

# xadd.sh - append three entries to a stream and read them back
r() { docker exec -i redis-lesson redis-cli "$@"; }
r DEL orders > /dev/null
r XADD orders '*' item kettle qty 1
r XADD orders '*' item toaster qty 2
r XADD orders '*' item mug qty 4
r XLEN orders
r XRANGE orders - +

$ bash xadd.sh
1792490156021-0
1792490156946-0
1792490157836-0
3
1792490156021-0
item
kettle
qty
1
1792490156946-0
item
toaster
qty
2
1792490157836-0
item
mug
qty
4

* asks Redis to generate the ID: the server’s clock in milliseconds, a dash, and a sequence number for entries added in the same millisecond. Your IDs will differ. IDs only ever increase, so the ID is also the entry’s position in the log, and XRANGE orders - + reads from the smallest ID to the largest. Each entry is a small set of field-value pairs, like a hash. (The entries here are about a second apart only because each docker exec takes that long on this machine.)

The message that isn’t lost

This is the Pub/Sub experiment again: send with nobody reading, then start a reader:

# nobody_listening.sh - the Pub/Sub experiment again: send with no reader, read later
r() { docker exec -i redis-lesson redis-cli "$@"; }
r DEL alerts > /dev/null

echo "PUBLISH receivers: $(r PUBLISH alerts 'disk full')"
r XADD alerts '*' msg 'disk full' > /dev/null
echo "XADD stored it; stream length: $(r XLEN alerts)"

sleep 2                                      # a reader shows up later
echo "--- a reader that starts 2 s later, reading from the beginning (ID 0):"
r XREAD COUNT 10 STREAMS alerts 0

$ bash nobody_listening.sh
PUBLISH receivers: 0
XADD stored it; stream length: 1
--- a reader that starts 2 s later, reading from the beginning (ID 0):
alerts
1792490163456-0
msg
disk full

PUBLISH reached 0 receivers and the message was gone. The same message added with XADD was still there two seconds later. XREAD ... STREAMS alerts 0 means “entries with an ID greater than 0”, so a reader can always replay from the start, or from the last ID it saw.

Waiting for new entries: XREAD BLOCK

A reader that polls in a loop wastes round trips. BLOCK makes it wait on the server:

# xread_block.sh - a reader waits for new entries instead of polling
r() { docker exec -i redis-lesson redis-cli "$@"; }
r DEL orders > /dev/null

r XREAD BLOCK 10000 STREAMS orders '$' &     # '$' = only entries added after this call
sleep 2
echo "--- 2 s later the reader is still waiting; now add an entry:"
r XADD orders '*' item kettle qty 1 > /dev/null
wait

echo "--- nothing arrives within a 1 s timeout:"
r XREAD BLOCK 1000 STREAMS orders '$'

$ bash xread_block.sh
--- 2 s later the reader is still waiting; now add an entry:
orders
1792490172504-0
item
kettle
qty
1
--- nothing arrives within a 1 s timeout:

The background reader printed nothing for two seconds, then returned the moment the entry was added. The second XREAD timed out after one second and returned nil, which redis-cli shows as an empty line. $ means “the last ID in the stream right now”. Pass the ID of the last entry you processed instead, or a reader that restarts misses whatever arrived while it was down.

Consumer groups: splitting the work

XREAD gives every reader every entry. To share the work between several workers, create a consumer group:

# group.sh - a consumer group: two consumers split the entries between them
r() { docker exec -i redis-lesson redis-cli "$@"; }
r DEL orders > /dev/null
r XGROUP CREATE orders workers '$' MKSTREAM
for item in kettle toaster mug teapot; do
  r XADD orders '*' item "$item" > /dev/null
done

echo "--- alice asks for 2 new entries:"
r XREADGROUP GROUP workers alice COUNT 2 STREAMS orders '>'
echo "--- bob asks for 2 new entries:"
r XREADGROUP GROUP workers bob COUNT 2 STREAMS orders '>'
echo "--- alice asks again: nothing new is left"
r XREADGROUP GROUP workers alice COUNT 2 STREAMS orders '>'

echo "--- delivered but not acknowledged yet:"
r XPENDING orders workers

$ bash group.sh
OK
--- alice asks for 2 new entries:
orders
1792490179320-0
item
kettle
1792490180725-0
item
toaster
--- bob asks for 2 new entries:
orders
1792490181729-0
item
mug
1792490182657-0
item
teapot
--- alice asks again: nothing new is left

--- delivered but not acknowledged yet:
4
1792490179320-0
1792490182657-0
alice
2
bob
2

The special ID > means “entries never delivered to anyone in this group”, so alice and bob got different entries, and a third read found nothing left. The group stores its own position in the stream, so consumers don’t track IDs themselves.

XPENDING is the other half. All four entries are pending: delivered, but not acknowledged with XACK. The summary gives the count, the smallest and largest pending IDs, and how many each consumer holds. An entry stays pending until it’s acknowledged, which is what makes the next example recoverable.

Kafka’s consumer groups solve the same problem with partitions; see Kafka consumers and groups for the comparison. In Redis, any consumer in a group can take any entry, with no partition assignment.

A consumer crashes before XACK

alice takes two orders, acknowledges one and dies. Python makes the pending details easier to read:

# crash.py - alice reads an order and dies before XACK; bob claims it and finishes the job
import time
import redis

r = redis.Redis(host="redis-lesson", decode_responses=True)
r.delete("orders")
r.xgroup_create("orders", "workers", id="$", mkstream=True)
r.xadd("orders", {"item": "kettle"})
r.xadd("orders", {"item": "toaster"})

# alice takes both entries, acknowledges the first, then crashes
[[_, entries]] = r.xreadgroup("workers", "alice", {"orders": ">"}, count=2)
print("alice got:", [fields["item"] for _, fields in entries])
r.xack("orders", "workers", entries[0][0])
print("alice acked kettle, then crashed before acking toaster")

time.sleep(1.2)
for p in r.xpending_range("orders", "workers", min="-", max="+", count=10):
    print(f"pending: {p['message_id']} owner={p['consumer']} "
          f"idle={p['time_since_delivered']} ms deliveries={p['times_delivered']}")

# bob claims anything idle for more than 1 s
next_id, claimed, deleted = r.xautoclaim("orders", "workers", "bob", min_idle_time=1000, start_id="0")
print("bob claimed:", [fields["item"] for _, fields in claimed])
for p in r.xpending_range("orders", "workers", min="-", max="+", count=10):
    print(f"pending: owner={p['consumer']} deliveries={p['times_delivered']}")
r.xack("orders", "workers", claimed[0][0])
print("after bob's XACK, pending:", r.xpending("orders", "workers")["pending"])

$ py crash.py
alice got: ['kettle', 'toaster']
alice acked kettle, then crashed before acking toaster
pending: 1792490188639-1 owner=alice idle=1201 ms deliveries=1
bob claimed: ['toaster']
pending: owner=bob deliveries=2
after bob's XACK, pending: 0

The toaster order didn’t vanish with alice. It stayed pending, still owned by her, idle for 1.2 seconds and delivered once. Nothing hands it to another consumer automatically: if alice never comes back, it sits there forever.

XAUTOCLAIM is the recovery step. It scans the pending list from start_id and transfers every entry idle longer than min_idle_time to the caller. The owner changed to bob and the delivery count went up to 2. After bob’s XACK, nothing is pending. In production, every worker runs a claim like this every so often, with an idle time longer than any job should take, so it doesn’t steal work from a consumer that is just slow. A consumer that restarts with the same name can also re-read its own pending entries by reading from ID 0 instead of >, as the next example does.

A consumer that is still alive but knows it can’t finish an entry can hand it back itself. Redis 8.8 added XNACK key group SILENT IDS 1 <id>: on 8.10.2, an entry released that way stayed pending with no owner, and an XAUTOCLAIM with a 60-second minimum idle time picked it up at once, without waiting out the timeout.

This is at-least-once delivery: alice may have shipped the toaster before she crashed, and bob ships it again. Practice task 3 makes that harmless.

Dead-lettering a poison message

An entry that fails every time it’s processed, such as a malformed quantity, would be retried forever. The delivery count says when to give up:

# dead_letter.py - a message that fails every time: give up after 3 deliveries
import redis

r = redis.Redis(host="redis-lesson", decode_responses=True)
MAX_DELIVERIES = 3
r.delete("orders", "orders:dead")
r.xgroup_create("orders", "workers", id="$", mkstream=True)
r.xadd("orders", {"item": "kettle", "qty": "1"})
r.xadd("orders", {"item": "toaster", "qty": "not-a-number"})     # the poison message


def handle(fields):
    return int(fields["qty"])                # raises ValueError for the poison message


def work_once(consumer):
    """Retry our own pending entries first (ID 0), then take new ones ('>')."""
    for start in ("0", ">"):
        reply = r.xreadgroup("workers", consumer, {"orders": start}, count=10)
        for _, entries in reply:
            for msg_id, fields in entries:
                info = r.xpending_range("orders", "workers", min=msg_id, max=msg_id, count=1)[0]
                tries = info["times_delivered"]
                try:
                    handle(fields)
                    r.xack("orders", "workers", msg_id)
                    print(f"{fields['item']:8} delivery {tries}: done")
                except ValueError:
                    if tries >= MAX_DELIVERIES:
                        r.xadd("orders:dead", {**fields, "orig_id": msg_id, "error": "bad qty"})
                        r.xack("orders", "workers", msg_id)
                        print(f"{fields['item']:8} delivery {tries}: dead-lettered")
                    else:
                        print(f"{fields['item']:8} delivery {tries}: failed, left pending")


for round_no in range(1, 5):
    print(f"--- round {round_no}")
    work_once("alice")
print("pending:", r.xpending("orders", "workers")["pending"])
for _, fields in r.xrange("orders:dead"):
    print("dead letter:", {k: v for k, v in fields.items() if k != "orig_id"})

$ py dead_letter.py
--- round 1
kettle   delivery 1: done
toaster  delivery 1: failed, left pending
--- round 2
toaster  delivery 2: failed, left pending
--- round 3
toaster  delivery 3: dead-lettered
--- round 4
pending: 0
dead letter: {'item': 'toaster', 'qty': 'not-a-number', 'error': 'bad qty'}

Reading from ID 0 re-delivers the consumer’s own pending entries and increments their delivery count, the same counter XAUTOCLAIM increments. After the third failed delivery the worker copies the entry to a separate orders:dead stream, with the original ID and the error, and acknowledges it so the group can move on. Round 4 found nothing to do. Someone inspects the dead-letter stream later, fixes the data and re-adds it. The pattern itself, and why the limit matters, is in Dead-Letter Queues.

Trimming

A stream keeps every entry until you remove it, even after every group has acknowledged it:

# trim.sh - streams grow forever unless you trim them
r() { docker exec -i redis-lesson redis-cli "$@"; }
r DEL events > /dev/null
seq 1 1000 | sed 's/.*/XADD events * n &/' | r > /dev/null
echo "after 1,000 XADDs:          $(r XLEN events)"

r XTRIM events MAXLEN '~' 150 > /dev/null
echo "XTRIM MAXLEN ~ 150:         $(r XLEN events)"
r XTRIM events MAXLEN 150 > /dev/null
echo "XTRIM MAXLEN 150 (exact):   $(r XLEN events)"

for batch in 1 2 3 4; do
  seq 1 100 | sed 's/.*/XADD events MAXLEN ~ 150 * n &/' | r > /dev/null
  echo "100 more XADD MAXLEN ~ 150: $(r XLEN events)"
done
r CONFIG GET 'stream-node-max-*'

$ bash trim.sh
after 1,000 XADDs:          1000
XTRIM MAXLEN ~ 150:         200
XTRIM MAXLEN 150 (exact):   150
100 more XADD MAXLEN ~ 150: 200
100 more XADD MAXLEN ~ 150: 200
100 more XADD MAXLEN ~ 150: 200
100 more XADD MAXLEN ~ 150: 200
stream-node-max-bytes
4096
stream-node-max-entries
100

MAXLEN ~ 150 left 200 entries, not 150. Redis stores a stream in nodes, and a node closes when it reaches either limit the last command printed: 100 entries (stream-node-max-entries) or 4 KB (stream-node-max-bytes). The ~ form only drops whole nodes, which is cheap. The exact form trims to 150, but it has to delete individual entries inside a node, which costs more. Putting MAXLEN ~ on every XADD keeps the stream bounded as it grows; here it settled at 200. MINID trims by ID instead, which means by age with generated IDs.

By default, trimming ignores consumer groups. Redis 8.2 added an ACKED option that only trims entries every group has acknowledged:

# trim_acked.sh - plain XTRIM drops an unacknowledged entry; XTRIM ... ACKED (Redis 8.2+) keeps it
r() { docker exec -i redis-lesson redis-cli "$@"; }

setup() {
  r DEL orders > /dev/null
  r XGROUP CREATE orders workers '$' MKSTREAM > /dev/null
  r XADD orders '*' item kettle > /dev/null
  r XADD orders '*' item toaster > /dev/null
  r XADD orders '*' item mug > /dev/null
  r XREADGROUP GROUP workers alice COUNT 3 STREAMS orders '>' > /dev/null
  for id in $(r XRANGE orders - + COUNT 2 | grep -E '^[0-9]+-[0-9]+$'); do
    r XACK orders workers "$id" > /dev/null                    # kettle and toaster are done; mug isn't
  done
}

setup
r XTRIM orders MAXLEN 0 > /dev/null
echo "XTRIM MAXLEN 0:       length $(r XLEN orders), still pending $(r XPENDING orders workers | head -1)"

setup
r XTRIM orders MAXLEN 0 ACKED > /dev/null
echo "XTRIM MAXLEN 0 ACKED: length $(r XLEN orders), left: $(r XRANGE orders - + | grep -A1 '^item$' | tail -1)"

$ bash trim_acked.sh
XTRIM MAXLEN 0:       length 0, still pending 1
XTRIM MAXLEN 0 ACKED: length 1, left: mug

alice finished kettle and toaster but not mug. The plain XTRIM MAXLEN 0 removed all three, including the unfinished mug, and still left one entry pending: a reference to an entry that no longer exists, which is what the default KEEPREF behaviour means. With ACKED, the two acknowledged entries went and mug stayed for alice to finish. Without ACKED (or on Redis before 8.2), size the trim limit for your slowest consumer.

XACKDEL: acknowledge and delete

XACK removes an entry from the group’s pending list and leaves it in the stream. Redis 8.2 added XACKDEL, which does both in one command, for streams used purely as work queues:

# xackdel.sh - XACK leaves the entry in the stream; XACKDEL (Redis 8.2+) acknowledges and deletes it
r() { docker exec -i redis-lesson redis-cli "$@"; }
r DEL orders > /dev/null
r XGROUP CREATE orders workers '$' MKSTREAM > /dev/null
a=$(r XADD orders '*' item kettle)
b=$(r XADD orders '*' item toaster)
r XREADGROUP GROUP workers alice COUNT 2 STREAMS orders '>' > /dev/null

r XACK orders workers "$a" > /dev/null
echo "after XACK:    pending $(r XPENDING orders workers | head -1), length $(r XLEN orders)"
r XACKDEL orders workers IDS 1 "$b"
echo "after XACKDEL: pending $(r XPENDING orders workers | head -1), length $(r XLEN orders)"

$ bash xackdel.sh
after XACK:    pending 1, length 2
1
after XACKDEL: pending 0, length 1

After XACK the stream still held both entries. After XACKDEL the second entry was acknowledged and deleted, and its reply 1 is a per-ID result code meaning it was both acknowledged and deleted. With several groups reading the same stream, add ACKED to delete an entry only once every group has acknowledged it. Redis 8.2 also added XDELEX, which deletes entries by ID like XDEL and takes the same KEEPREF, DELREF and ACKED options.

Gotchas

• Pick the idle timeout for XAUTOCLAIM carefully. Too short and a slow but healthy consumer has its work taken and done twice; too long and a crashed consumer’s entries wait that long.
• Consumers are created on first use and never removed. A worker that picks a new random name on every restart leaves dead consumers behind. Use stable names, or XGROUP DELCONSUMER after claiming their pending entries.
• A stream is one key, so it lives on one cluster node. In Redis Cluster, spread load over several stream keys.
• Persistence still applies. Streams are only as durable as your persistence settings; an acknowledged entry can be lost in a crash just as a SET can.

Practice

1. How far behind is a group? Add 10 entries, let alice read 3 and bob 2, then use XINFO GROUPS and XINFO CONSUMERS to see the group’s lag and who holds what.

# p1_xinfo.sh - how far behind is the group, and who holds what?
r() { docker exec -i redis-lesson redis-cli "$@"; }
r DEL orders > /dev/null
r XGROUP CREATE orders workers '$' MKSTREAM > /dev/null
seq 1 10 | sed 's/.*/XADD orders * n &/' | r > /dev/null
r XREADGROUP GROUP workers alice COUNT 3 STREAMS orders '>' > /dev/null
r XREADGROUP GROUP workers bob COUNT 2 STREAMS orders '>' > /dev/null

echo "--- XINFO GROUPS orders"
r XINFO GROUPS orders | paste - - | grep -E '^(name|consumers|pending|entries-read|lag)'
echo "--- XINFO CONSUMERS orders workers"
r XINFO CONSUMERS orders workers | paste - - | grep -E '^(name|pending)'

$ bash p1_xinfo.sh
--- XINFO GROUPS orders
name	workers
consumers	2
pending	5
entries-read	5
lag	5
--- XINFO CONSUMERS orders workers
name	alice
pending	3
name	bob
pending	2

2. Two groups on one stream. Create a billing group and a shipping group on the same stream, add two orders, and read both groups. Which orders does each get?

# p2_two_groups.sh - two groups on one stream: each group gets every entry
r() { docker exec -i redis-lesson redis-cli "$@"; }
r DEL orders > /dev/null
r XGROUP CREATE orders billing '$' MKSTREAM > /dev/null
r XGROUP CREATE orders shipping '$' > /dev/null
r XADD orders '*' item kettle > /dev/null
r XADD orders '*' item toaster > /dev/null

echo "--- billing (one consumer):"
r XREADGROUP GROUP billing b1 COUNT 10 STREAMS orders '>' | grep -A1 '^item$' | grep -v -e '^item$' -e '^--$'
echo "--- shipping (one consumer):"
r XREADGROUP GROUP shipping s1 COUNT 10 STREAMS orders '>' | grep -A1 '^item$' | grep -v -e '^item$' -e '^--$'

$ bash p2_two_groups.sh
--- billing (one consumer):
kettle
toaster
--- shipping (one consumer):
kettle
toaster

3. Make re-delivery harmless. Repeat the crash: alice charges the order and crashes before XACK, and bob claims it. Count the charges, then make the consumer idempotent.

# p3_idempotent.py - a re-delivered entry is processed twice unless the consumer remembers it
import redis

r = redis.Redis(host="redis-lesson", decode_responses=True)


def run(idempotent):
    r.delete("orders", "charges", "processed")
    r.xgroup_create("orders", "workers", id="$", mkstream=True)
    r.xadd("orders", {"order": "A-1", "amount": "30"})

    def process(consumer, msg_id, fields, crash_before_ack):
        if idempotent and not r.sadd("processed", msg_id):
            print(f"  {consumer}: {msg_id} already processed, skipping")
        else:
            r.rpush("charges", fields["order"])                # the side effect: charge the card
            print(f"  {consumer}: charged {fields['order']}")
        if not crash_before_ack:
            r.xack("orders", "workers", msg_id)

    [[_, [(msg_id, fields)]]] = r.xreadgroup("workers", "alice", {"orders": ">"}, count=1)
    process("alice", msg_id, fields, crash_before_ack=True)    # charged, then crashed
    _, claimed, _ = r.xautoclaim("orders", "workers", "bob", min_idle_time=0, start_id="0")
    for msg_id, fields in claimed:
        process("bob", msg_id, fields, crash_before_ack=False)
    print("  charges:", r.lrange("charges", 0, -1))


print("without a processed set:")
run(idempotent=False)
print("with a processed set:")
run(idempotent=True)

$ py p3_idempotent.py
without a processed set:
  alice: charged A-1
  bob: charged A-1
  charges: ['A-1', 'A-1']
with a processed set:
  alice: charged A-1
  bob: 1792490263962-0 already processed, skipping
  charges: ['A-1']