A Redis Streams consumer group distributes new entries among named consumers and records deliveries awaiting acknowledgement. It does not guarantee exactly-once effects, prevent retention loss, or recover abandoned work by itself. Understanding the group position, pending entries, and retained stream data separately is essential before treating it as a work queue.

Position and pending work differ

XGROUP CREATE jobs workers 0-0 starts before the retained history; using $ instead starts after the current end. Workers in one group share new deliveries, while separate groups maintain independent delivery state. Use distinct names for concurrent workers. The group owns a last-delivered position, not a separate offset committed by each consumer.

For ordinary XREADGROUP without CLAIM or NOACK, > requests new entries and delivery adds them to the Pending Entries List, or PEL. The group position advances when new entries are delivered, before XACK. XPENDING reports delivered but unacknowledged entries, not the backlog that has never been delivered. A group can therefore have pending count zero while unread work exists.

XACK removes specified IDs from that group’s PEL. It does not delete their stream records or advance the last-delivered position. Acknowledgement can finish out of order, and another group can still read the retained entries. XREADGROUP with 0-0 reads this named consumer’s pending history instead of requesting new work; it does not retrieve another consumer’s pending messages.

Recovery needs an explicit decision

Repeated ordinary > reads do not retry an abandoned PEL entry. XAUTOCLAIM transfers eligible pending entries to the requesting consumer. Its arguments include the stream, group, consumer, minimum idle milliseconds, and starting ID. Continue scanning with the returned cursor until 0-0; an empty claimed batch alone does not mean the scan is complete. Later scans may find entries that have since become eligible.

Redis 8.4 also introduced XREADGROUP CLAIM as an explicit combined recovery option. The example does not enable it. Its XAUTOCLAIM idle threshold of zero is solely for deterministic local observation; production thresholds must reflect processing time and liveness policy. Claiming ownership cannot stop an old worker that is still running or undo its external side effects.

When it breaks

A crash after an external effect but before acknowledgement leaves work eligible for another delivery. Retrying can repeat that effect. Acknowledging before the effect instead risks losing required work. Use stable business identifiers and transactional deduplication where possible; a delivery counter records delivery attempts, not completed database transactions. Bound retries and preserve failing payloads before retiring work.

A PEL reference does not pin the stream payload. XDEL and default KEEPREF trimming can remove it while leaving a pending ID. XAUTOCLAIM reports and clears such missing records on Redis 7 and later; that is cleanup, not recovery of their contents. Choose retention with consumer outages in mind. Stream persistence and replication also require separate durability decisions.

Monitor unread lag, pending count, idle time, and repeated deliveries together. One count cannot establish queue health. NOACK omits pending tracking for new deliveries and is unsuitable when that tracking is required for recovery. Parallel workers can finish out of insertion order even when the stream IDs themselves remain ordered.

Check it quickly

Use a POSIX environment with a C compiler, make, curl, tar, and Python 3.10 or newer. This ran on macOS with Python 3.14.4, Redis 8.10.1, and redis-py 8.1.0. The commands build Redis core with libc and install the client in a temporary virtual environment, without optional modules or system-wide installation.

redis_lab=$(mktemp -d)
curl --fail --location --output "$redis_lab/redis.tar.gz" https://download.redis.io/releases/redis-8.10.1.tar.gz
tar -xzf "$redis_lab/redis.tar.gz" -C "$redis_lab"
make -C "$redis_lab/redis-8.10.1" build redis -j2 MALLOC=libc
export REDIS_SERVER="$redis_lab/redis-8.10.1/src/redis-server"
python3 -m venv "$redis_lab/venv"
"$redis_lab/venv/bin/python" -m pip install redis==8.1.0

Run the Python block with "$redis_lab/venv/bin/python" in the same shell. It creates a private Unix socket under /tmp, disables TCP, scheduled snapshots, and append-only persistence, then terminates only its own process and removes its runtime directory. Downloads require network access. No default local Redis is contacted.

from contextlib import contextmanager
import os
from pathlib import Path
import subprocess
from tempfile import TemporaryDirectory
import time

import redis
from redis.backoff import NoBackoff
from redis.retry import Retry


@contextmanager
def instance():
    with TemporaryDirectory(prefix="redis-lab-", dir="/tmp") as directory:
        endpoint = str(Path(directory) / "redis.sock")
        process = subprocess.Popen([
            os.environ["REDIS_SERVER"], "--port", "0",
            "--unixsocket", endpoint, "--unixsocketperm", "700",
            "--save", "", "--appendonly", "no", "--dir", directory,
            "--maxmemory-policy", "noeviction",
        ], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
        client = redis.Redis(unix_socket_path=endpoint, decode_responses=False,
                             socket_timeout=2, retry=Retry(NoBackoff(), 0))
        try:
            deadline = time.monotonic() + 5
            while True:
                assert process.poll() is None, "Redis exited during startup"
                try:
                    if client.ping():
                        break
                except redis.ConnectionError:
                    if time.monotonic() >= deadline:
                        raise
                    time.sleep(0.02)
            yield client
        finally:
            client.close()
            if process.poll() is None:
                process.terminate()
            try:
                _, errors = process.communicate(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
                process.communicate()
                raise
            assert process.returncode == 0, errors.decode()


with instance() as client:
    for entry_id in ("1-0", "2-0", "3-0"):
        client.xadd("jobs", {"job": entry_id}, id=entry_id)
    client.xgroup_create("jobs", "workers", id="0-0")
    client.xgroup_create("jobs", "tail", id="$")
    assert client.xpending("jobs", "workers")["pending"] == 0
    assert client.xreadgroup("tail", "observer", {"jobs": ">"}) == []
    first = client.xreadgroup("workers", "w1", {"jobs": ">"}, count=1)[0][1]
    second = client.xreadgroup("workers", "w2", {"jobs": ">"}, count=1)[0][1]
    assert first[0][0] == b"1-0" and second[0][0] == b"2-0"
    assert client.xpending("jobs", "workers")["pending"] == 2
    group = next(item for item in client.xinfo_groups("jobs") if item["name"] == b"workers")
    assert group["last-delivered-id"] == b"2-0"
    print("backlog starts with PEL=0; new deliveries split 1-0/2-0 and advance the group position")

    effects = [first[0][0]]
    assert client.xack("jobs", "workers", "2-0") == 1
    assert client.xlen("jobs") == 3
    group = next(item for item in client.xinfo_groups("jobs") if item["name"] == b"workers")
    assert group["last-delivered-id"] == b"2-0"
    assert client.xreadgroup("workers", "w2", {"jobs": "0-0"}) == [[b"jobs", []]]
    assert client.xreadgroup("workers", "w1", {"jobs": "0-0"})[0][1] == first
    third = client.xreadgroup("workers", "w2", {"jobs": ">"})[0][1]
    assert third[0][0] == b"3-0"
    assert client.xack("jobs", "workers", "3-0") == 1
    assert client.xreadgroup("workers", "w2", {"jobs": ">"}) == []
    print("XACK keeps stream data and read position; 0-0 reads own pending; > does not recover w1")

    cursor, claimed, deleted = client.xautoclaim(
        "jobs", "workers", "w2", min_idle_time=0, start_id="0-0", count=10,
    )
    assert cursor == b"0-0" and claimed == first and deleted == []
    pending = client.xpending_range("jobs", "workers", "-", "+", 10)
    assert len(pending) == 1 and pending[0]["consumer"] == b"w2"
    assert pending[0]["times_delivered"] >= 2
    effects.append(claimed[0][0])
    assert effects.count(b"1-0") == 2
    assert client.xack("jobs", "workers", "1-0") == 1
    assert client.xack("jobs", "workers", "1-0") == 0
    assert client.xpending("jobs", "workers")["pending"] == 0
    assert client.xlen("jobs") == 3
    print("explicit claim changes owner; simulated side effect repeats; ACK clears PEL, not the stream")

    client.xadd("jobs", {"job": "4-0"}, id="4-0")
    assert client.xreadgroup("tail", "observer", {"jobs": ">"})[0][1][0][0] == b"4-0"
    assert client.xreadgroup("workers", "w1", {"jobs": ">"})[0][1][0][0] == b"4-0"
    assert client.xdel("jobs", "4-0") == 1
    assert client.xpending("jobs", "workers")["pending"] == 1
    cursor, claimed, deleted = client.xautoclaim(
        "jobs", "workers", "w2", min_idle_time=0, start_id="0-0", count=10,
    )
    assert cursor == b"0-0" and claimed == [] and deleted == [b"4-0"]
    assert client.xpending("jobs", "workers")["pending"] == 0
    print("a $ group starts after old entries; XDEL loses pending payload; claim reports deleted ID")

The workers group begins with three unread entries and zero pending entries. Two reads distribute 1-0 and 2-0; the group position already reaches 2-0 before acknowledgement. Acknowledging the second entry leaves stream length and that position unchanged. Pending-history reads distinguish the two consumers, and a new-entry read returns 3-0 instead of retrying 1-0.

The effects list models a completed action without its acknowledgement. Explicit claiming moves 1-0 to w2, and processing it again produces two recorded effects. This is a simulation of duplicate business processing, not a transaction-safe implementation. Acknowledgement then clears the PEL while leaving the three stream records available.

The tail group sees only a later 4-0 entry. Deleting that entry while pending makes the subsequent claim return its ID in the deleted list without a payload. The checks exercise command-level state changes, not process-crash detection, automatic retry scheduling, durable deduplication, or failover recovery.

Sources