Quick wins for a faster PC:
Clear out junk files and repair common Windows errorsFree Scan →Fix the driver behind crashes, sound loss and screen glitchesFind Drivers →Redis Streams can hold events so that a worker crash does not lose them, a pool of workers can share them, and the history stays readable for a retention window you choose. The pattern uses four operations: XADD appends an event, XREADGROUP hands it to one member of a consumer group, XACK marks it done after the work succeeds, and XAUTOCLAIM takes over deliveries that a stopped worker left behind. The guarantee that results is at-least-once processing, so handlers must tolerate seeing the same event twice.
Redis’s own Python guide demonstrates this flow with redis-py. A separate PyPI package, wredis, advertises a Streams interface with similar vocabulary. This article walks through the Redis pattern in full, then explains what the WRedis project page documents and what it leaves unstated. Version notes reflect Redis’s documentation as of early October 2026.
As an Amazon Associate I earn from qualifying purchases.
How the pieces fit together
Redis’s documentation defines a stream as an append-only log of field/value entries with auto-generated, time-ordered IDs. Each entry is a small map of fields, such as type and order_id, stored under one Redis key. Four objects matter for reliable ingestion:
What’s actually slowing this PC down?
Pick the symptom - the matching free tool is one click away.
- Stream: the entries and their IDs. One key holds one stream.
- Consumer group: a named cursor over the stream, plus the record of what has been delivered through it.
- Consumer: a named member of a group. Each delivery is attributed to the consumer that received it.
- Pending entries list (PEL): deliveries the group has made that have not yet been acknowledged. Every group keeps its own PEL.
Because the cursor and the PEL belong to the group, two groups reading the same stream track their deliveries separately.
#1 Best Overall
The basic flow
1. Append events with XADD
Producers append with XADD. Passing * as the ID, which is the redis-py default, lets Redis generate it. The example caps the stream at roughly 100,000 entries; the retention section explains what that cap does and does not guarantee.
import redisnnr = redis.Redis(host="localhost", port=6379, decode_responses=True)nSTREAM = "orders:events"nnentry_id = r.xadd(n STREAM,n {"type": "order_created", "order_id": "o-1001"},n maxlen=100_000,n approximate=True,n)n2. Create the consumer group once
Create the group before workers start. The starting ID decides what the group sees first: $ means only entries added after the group is created, while 0 means every retained entry. MKSTREAM creates the stream if it does not exist yet. The equivalent command is XGROUP CREATE orders:events billing $ MKSTREAM.
GROUP = "billing"nntry:n r.xgroup_create(STREAM, GROUP, id="$", mkstream=True)nexcept redis.ResponseError as exc:n if "BUSYGROUP" not in str(exc): # group already existsn raisenChoose the starting ID deliberately. Starting at 0 on a stream with long retention sends the group a large burst of historical work on its first run.
3. Read as a group member
Each worker reads with XREADGROUP and the special ID >, which asks only for entries never delivered to this group before. Redis records each delivery in the PEL.
CONSUMER = "worker-1"nndef handle(fields):n ... # must be safe to run twice; see the idempotency sectionnndef run_worker():n while True:n batches = r.xreadgroup(GROUP, CONSUMER, {STREAM: ">"}, count=10, block=5000)n for _stream, messages in batches or []:n for msg_id, fields in messages:n handle(fields)n r.xack(STREAM, GROUP, msg_id)nIf handle raises, the XACK line is skipped and the entry stays pending. The uncaught exception also stops this worker, which is usually the safer default: the entry waits in the PEL until a reclaim routine picks it up.
4. Acknowledge only after the work succeeds
Call XACK after the side effect completes, not when the message is read. If the worker stops between the side effect and the acknowledgement, the entry remains in the PEL and will be delivered again. That boundary is what makes recovery possible, and it is also the source of duplicates. XACK removes the entry from the PEL; it does not delete the entry from the stream.
Recovering deliveries after a worker stops
Redis’s Python guide demonstrates crash simulation and recovery with XAUTOCLAIM. The recovery path has three parts: find pending entries, decide which are stalled, and move them to a healthy consumer.
Rank #2
Inspect pending entries
The extended form of XPENDING lists individual deliveries. This command shows up to ten entries from the billing group:
XPENDING orders:events billing - + 10Each row shows the entry ID, the consumer holding it, how long it has been idle in milliseconds, and how many times it has been delivered. The summary form, r.xpending(STREAM, GROUP), returns the total pending count, the lowest and highest pending IDs, and a per-consumer breakdown.
Reclaim idle deliveries with XAUTOCLAIM
XAUTOCLAIM scans the PEL for entries idle for at least min_idle_time milliseconds, reassigns them to the calling consumer, and returns them for processing. It returns a cursor, so a long scan can resume where it stopped. Its reply also lists IDs of entries that were deleted from the stream while still pending. The three-value unpacking in the example below requires Redis 7.0 or later; see the version notes.
def reclaim_stalled(min_idle_ms=120_000):n cursor = "0-0"n while True:n cursor, messages, _deleted = r.xautoclaim(n STREAM, GROUP, CONSUMER,n min_idle_time=min_idle_ms,n start_id=cursor,n count=100,n )n for msg_id, fields in messages:n handle(fields)n r.xack(STREAM, GROUP, msg_id)n if cursor == "0-0":n breaknRun this routine periodically in each worker. A handler that keeps raising on the same entry will block the loop at that point, so if your workload can contain such messages, catch exceptions per message and route repeat failures to a separate stream for inspection.
Do these 3 things before closing this tab:
1Clear out junk files and repair common Windows errors2Fix the driver behind crashes, sound loss and screen glitches3Repair Windows errors before they cause bigger problemsChoose the idle threshold from processing time
The idle threshold is the main safety control. If it is shorter than a legitimate handler run, a second worker can reclaim an entry that the first is still processing, and both will run it. Set it above the slowest normal handler time plus a margin. A handler that usually finishes in seconds but occasionally takes two minutes suggests a threshold of several minutes, adjusted from your own timing data. This is operational guidance, not a Redis default.
After a restart, a worker can recover its own unfinished work by reading its PEL history. Reading with ID 0 instead of > returns the entries this consumer already holds, as described in the XREADGROUP command reference.
What the guarantee is, and what it is not
This pattern gives at-least-once processing. Redis does not make an external side effect and the acknowledgement a single atomic step, so a crash after the work succeeds but before XACK runs causes the same entry to be processed again. Handlers must therefore produce the same result when they run twice. Two approaches work well:
Rank #3
- Make the write naturally idempotent. An upsert keyed by order ID, or a state transition that does nothing when already applied, needs no extra bookkeeping.
- Record an idempotency key in the same transaction as the side effect. When the side effect is a relational write, a table with a primary key on the event key does the job:
CREATE TABLE processed_events (n event_key TEXT PRIMARY KEY,n processed_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMPn);nInsert the event key and perform the write in one transaction. If the insert conflicts, the event was already applied, and the handler returns without repeating the work. Keeping the marker in Redis instead reintroduces the crash window, because a failure between the two writes leaves them out of step.
Recommended Free Tools
Also give each logical event its own identifier in the fields. Redis assigns a new stream ID on every XADD call, so a producer that retries after an unclear timeout can append the same event twice. Deduplicate on the producer-assigned identifier, not on the stream ID.
Choosing a reader model
Three read patterns are available, and they behave differently with respect to delivery tracking:
| Reader pattern | Who receives each entry | Tracks deliveries and acknowledgements | Use when |
|---|---|---|---|
XREAD direct tailing |
Every caller reads from its own position | No | Displaying or debugging a stream without recoverable processing |
XREADGROUP, one group with several members |
One member receives each new entry | Yes, in one PEL for the group | Workers share a single job stream |
XREADGROUP, several groups |
Each group receives its own full pass of the stream | Yes, separately for each group | Independent applications consume the same events |
Plain XREAD does not create group state, so it has no pending list to recover from. For recoverable processing, use a group.
Replay and retention
Replay with XRANGE
XRANGE reads entries by ID without touching any group’s cursor or PEL. It is the right tool for replay, debugging, and rebuilding a downstream store from history. To continue from the last ID you processed, use an exclusive lower bound with a leading parenthesis:
start = "1765000000000-0" # the last ID you processednpage = r.xrange(STREAM, min=f"({start}", max="+", count=500)nReplay can only reach entries still retained in the stream, which is where trimming comes in.
Trimming sets the replay window
A stream grows until something trims it. Trimming is the retention policy: anything trimmed is gone from the stream, and no group can read it again. Redis’s guide describes both approximate length trimming and trimming by minimum ID.
Rank #4
Cap by entry count. MAXLEN with approximate=True is cheaper, but the length can briefly exceed the cap because Redis removes entries in internal groups rather than one at a time. Treat the cap as approximate. Use exact trimming only when the precise count matters, since it costs more.
Cap by age. Stream IDs begin with a millisecond timestamp, so a minimum ID is effectively an age cutoff. This keeps roughly the last seven days:
The Tool Desk
Outbyte Driver Updater FREEScan for outdated or missing drivers - takes under a minuteDriver Scan →Outbyte PC Repair FREERepair Windows errors before they cause bigger problemsFix Now →import timenncutoff_ms = int((time.time() - 7 * 24 * 3600) * 1000)nr.xtrim(STREAM, minid=f"{cutoff_ms}-0", approximate=True)nTrimming does not account for pending deliveries. If an entry is trimmed while a group still holds it in its PEL, reclaim cannot return its payload, and XAUTOCLAIM reports the ID as deleted. Set the retention window longer than your longest realistic recovery delay.
Monitoring lag and pending entries
Two numbers describe different failures. Lag is the number of entries a group has not yet delivered. Pending count is the number of entries delivered but not acknowledged. Check both per group:
for g in r.xinfo_groups(STREAM):n print(g["name"], "consumers:", g["consumers"],n "pending:", g["pending"], "lag:", g.get("lag"))nUse this table to match a symptom to its likely cause:
| Symptom | Likely cause | First check |
|---|---|---|
| Lag rises, pending stays near zero | Producers outpace consumers | Consumer count in XINFO GROUPS; handler duration |
| Pending rises, lag stays low | Deliveries not acknowledged: an exception before XACK, a crash, or an acknowledgement with the wrong ID |
XPENDING per consumer; idle times; logs from the exception path |
| Pending entries age without moving | The consumer holding them is gone and nothing reclaims its entries | Whether the reclaim routine runs; the min_idle_time value |
| Same event processed twice | Normal at-least-once redelivery, or an idle threshold shorter than the handler’s run time | Handler idempotency; threshold compared with the slowest normal run |
| Replay returns fewer events than expected | Entries removed by MAXLEN or MINID trimming |
Stream length and first entry ID in XINFO STREAM |
Scaling past one stream key
A stream is a single key, so it lives on one shard in Redis Cluster. Adding members to a consumer group spreads new deliveries across them, but it does not spread the stream’s storage or write load across shards.
When one stream exceeds what its shard can handle, split it by tenant or entity, for example orders:events:{tenant_id}. Two consequences follow:
Best Value
- Ordering holds only within a partition. Any consumer that needs events from several partitions in a particular order needs a different design.
- Reclaim, lag, and pending counts are tracked per stream, so each partition’s group state must be monitored separately.
Where one group’s workload must not take capacity from another, run separate consumer pools. Separate groups on the same workers still compete for the same processes.
Version and client requirements
XREADGROUPis available since Redis Open Source 5.0.0, per its command reference.XAUTOCLAIMwas added in Redis 6.2.- Exclusive range bounds, written with a leading parenthesis, require Redis 6.2 or later.
- Redis’s Python guide lists Redis 7.0 or later, Python 3.9 or later, and
redis-py5.0 or later for its example. The example relies on the reply shape available from Redis 7.0. - Redis 8.2 adds
XACKDELandXDELEXalong with stream operations for coordination among groups. Redis 8.6 adds idempotent message processing, which the Redis Streams overview frames around at-most-once production and deduplication. Neither is available on earlier servers.
Confirm the server version with INFO server before you deploy, and confirm the redis-py version in your environment, because the reply shapes shown above depend on both.
WRedis: what its package documents
The wredis package on PyPI is separate from redis-py. Its project page documents a Streams interface shaped like this:
Free tools Windows power users keep installed
One-click scans. No signup required.
from wredis.streams import RedisStreamManagernnsm = RedisStreamManager(host="localhost")nsm.add_to_stream("events", {"action": "login", "user": "alice"})[email protected]_message("events", group_name="my_group", consumer_name="worker_1")ndef process(data):n print(data)nnsm.wait()nThe page lists add_to_stream, on_message, exist, read_from_stream, wait, and delete_stream as Streams methods. The on_message decorator takes the stream name and the group_name and consumer_name arguments, which correspond to the group and consumer concepts above. The table compares the two approaches on the points that matter for recovery.
| Concern | redis-py flow in Redis’s Python guide | WRedis RedisStreamManager |
|---|---|---|
| Where it is documented | Redis’s Python streaming guide | PyPI project page |
| Append | xadd |
add_to_stream |
| Group read | xreadgroup with > |
on_message with group_name and consumer_name |
| Acknowledgement | xack after the handler succeeds |
Not described on the project page |
| Stalled-delivery reclaim | xautoclaim with an idle threshold |
Not described on the project page |
| Retention | xtrim with maxlen or minid |
Not described on the project page |
What is not established about WRedis
The project page does not say when the wrapper sends an acknowledgement relative to the handler returning, what happens when the handler raises, whether pending entries from a stopped consumer are reclaimed, or which retry, error-handling, and trimming defaults apply. It also does not state the Redis server and redis-py versions it supports. Nothing in this article is based on running WRedis under worker failure, so it should not be assumed to match the recovery behavior described above. The page’s Queue and Pub/Sub modules have their own delivery semantics and should not be treated as interchangeable with consumer-group Streams.
Before using WRedis in a reliability-critical pipeline:
- Pin an exact WRedis version and read the source for that version, not only the project page.
- Confirm that acknowledgement happens after the handler returns successfully, and that a raised exception leaves the entry pending.
- Confirm that stalled entries are reclaimed, and identify the idle threshold used.
- Check its trimming behavior and whether it can remove entries still held in a pending list.
- Stop a worker in the middle of a handler and confirm another consumer receives the entry.
The Bottom Line
For recoverable processing today, build on the redis-py flow that Redis documents: append with XADD, read through a group with XREADGROUP, acknowledge after the work, and reclaim with XAUTOCLAIM. Consider WRedis only after its acknowledgement and reclaim behavior has been checked against that flow in the version you intend to deploy.
Quick Recap
Product prices and availability are accurate as of the date/time indicated and are subject to change. Any price and availability information displayed on Amazon at the time of purchase will apply.
