Redis Streams: Consumer Groups, the Pending Entry List, XAUTOCLAIM and Idempotent XADD

Version note: The examples in this post were run with Redis 8.10.3 (redis:8), Valkey 9.1.3, .NET 10, StackExchange.Redis 3.4.35 and Microsoft.Extensions.Hosting 10.0.13.

Guess First

In post 05 we sent the order confirmation e-mails of Bookstore, the online bookstore used as the shared example across this site's series, through a list, and with LMOVE we rescued a crashed worker's job in a processing list; but nothing recorded who took it, how long it had been waiting or how many times it had been tried. Let's build the same scenario with a Redis stream. emails-worker-1 in the emails consumer group reads three order events, acknowledges only the first (XACK) and crashes:

StreamEntry[] first = await db.StreamReadGroupAsync(Stream, "emails", "emails-worker-1", StreamPosition.NewMessages, count: 10);
await db.StreamAcknowledgeAsync(Stream, "emails", first[0].Id);
Console.WriteLine($"emails-worker-1 read {first.Length}, acknowledged {first[0]["orderId"]}, then crashed");

// 1.1 seconds later a second worker reads: XREADGROUP ... CLAIM 1000
StreamEntry[] second = await db.StreamReadGroupAsync(Stream, "emails", "emails-worker-2", StreamPosition.NewMessages,
    count: 10, noAck: false, claimMinIdleTime: TimeSpan.FromSeconds(1));

Meanwhile a new order, number 1004, arrives on the stream. Which orders does the second worker get, and in which order? What happened to the unacknowledged 1002 and 1003?

dotnet run --project src/Bookstore.RedisLab -- stream-crash

emails-worker-1 read 3, acknowledged 1001, then crashed
XPENDING: 2 pending, emails-worker-1=2
emails-worker-2 got order 1002: idle=1110 ms, delivered before=1
emails-worker-2 got order 1003: idle=1110 ms, delivered before=1
emails-worker-2 got order 1004: idle=0 ms, delivered before=0
XPENDING after: 0

Redis did not forget the two unacknowledged events: XPENDING showed both waiting at emails-worker-1. With a single read, the second worker first took over these two events that had been waiting for 1.1 seconds (along with the fact that each had been delivered once before), then took the new 1004. With the list in post 05 this needed a separate hash, a reaper and quite a bit of code. This post covers Redis Streams: the structure of a stream, consumer groups and the pending entry list (PEL), XAUTOCLAIM and Redis 8.4's XREADGROUP … CLAIM, 8.6's idempotent XADD, 8.8's XNACK, a BackgroundService consumer in .NET, poison messages and trimming.

The Structure of a Stream: XADD, IDs and XRANGE

A stream is an append-only log in which every entry has an ID and field-value pairs. Entries are not removed when they are read; that is the biggest difference from a list. To keep it readable I give the first two IDs by hand and let Redis generate the third:

127.0.0.1:6379> XADD demo:orders:events 1794650000001-0 type order-created orderId 1001 isbn 9780441013593 quantity 2
"1794650000001-0"
127.0.0.1:6379> XADD demo:orders:events 1794650000002-0 type order-created orderId 1002 isbn 9780141439518 quantity 1
"1794650000002-0"
127.0.0.1:6379> XADD demo:orders:events * type order-created orderId 1003 isbn 9780451524935 quantity 1
"1794655623244-0"
127.0.0.1:6379> XLEN demo:orders:events
(integer) 3
127.0.0.1:6379> XRANGE demo:orders:events - + COUNT 1
1) 1) "1794650000001-0"
   2) 1) "type"
      2) "order-created"
      3) "orderId"
      4) "1001"
      5) "isbn"
      6) "9780441013593"
      7) "quantity"
      8) "2"

The ID has two parts: the Unix millisecond when the entry was added and a sequence number within that millisecond. With * Redis generates the ID from its own clock and guarantees it is always greater than the previous one; an ID given by hand also has to be greater than the last one. So a stream is ordered by time and can be read by time range with XRANGE (- from the start, + to the end). XREAD, in contrast, reads everything after an ID; without groups, it suits cases where every reader should see everything (a live monitoring screen, for example). Inside, entries are stored as listpacks in the nodes of a radix tree, with the field names written once and IDs stored as differences from a previous entry; in the measurement, a million order events took 42.9 bytes per entry (below).

Consumer Groups and the Pending Entry List

A consumer group is a group of workers reading a stream. The group keeps how far into the stream it has handed out entries (last-delivered-id) and the entries it handed out but that have not been acknowledged yet (the PEL). Several groups can attach to the same stream and each group gets every entry once; the workers within a group share the work:

127.0.0.1:6379> XGROUP CREATE demo:orders:events emails 0
OK
127.0.0.1:6379> XGROUP CREATE demo:orders:events stock 0
OK
127.0.0.1:6379> XREADGROUP GROUP emails emails-worker-1 COUNT 2 STREAMS demo:orders:events >
1) 1) "demo:orders:events"
   2) 1) 1) "1794650000001-0"
         2) 1) "type"
            2) "order-created"
            3) "orderId"
            4) "1001"
            5) "isbn"
            6) "9780441013593"
            7) "quantity"
            8) "2"
      2) 1) "1794650000002-0"
         2) 1) "type"
            2) "order-created"
            3) "orderId"
            4) "1002"
            5) "isbn"
            6) "9780141439518"
            7) "quantity"
            8) "1"
127.0.0.1:6379> XREADGROUP GROUP stock stock-worker-1 COUNT 1 STREAMS demo:orders:events >
1) 1) "demo:orders:events"
   2) 1) 1) "1794650000001-0"
         2) 1) "type"
            2) "order-created"
            3) "orderId"
            4) "1001"
            5) "isbn"
            6) "9780441013593"
            7) "quantity"
            8) "2"
127.0.0.1:6379> XACK demo:orders:events emails 1794650000001-0
(integer) 1
127.0.0.1:6379> XPENDING demo:orders:events emails - + 10
1) 1) "1794650000002-0"
   2) "emails-worker-1"
   3) (integer) 150
   4) (integer) 1

XGROUP CREATE … 0 starts the group at the beginning of the stream; with $ it would only see what arrives from now on. > means "entries never handed out to this group". The stock group got 1001, which emails had already read: groups are independent of each other. After XACK only 1002 was left in the PEL of emails; the detailed form of XPENDING shows which consumer has it, that it has been waiting for 150 ms and that it was delivered once. An entry leaves the PEL only with XACK; if the worker crashes, it stays there and waits to be taken over.

Taking Over a Dead Worker's Work: XAUTOCLAIM, CLAIM and XNACK

Since Redis 6.2, XAUTOCLAIM key group consumer min-idle-time start hands entries that have waited longer than the given time over to another consumer. Redis 8.4 added this to the read command: according to the docs, XREADGROUP … CLAIM min-idle-time … > returns the entries that have waited long enough first (the longest waiting first), then new entries, and adds to every claimed entry how long it waited and how many times it was delivered before. The idle=1110 ms, delivered before=1 in Guess First are these extra fields; in StackExchange.Redis 3.4.35 they come as StreamEntry.IdleTime and StreamEntry.DeliveryCount. So the worker no longer needs a separate "collect orphaned work" loop.

Redis 8.8 added the opposite direction: XNACK. When a worker realizes it cannot process an entry, it hands it back to the group without waiting:

127.0.0.1:6379> XNACK demo:orders:events emails FAIL IDS 1 1794650000002-0
(integer) 1
127.0.0.1:6379> XPENDING demo:orders:events emails - + 10
1) 1) "1794650000002-0"
   2) ""
   3) (integer) -1
   4) (integer) 1
127.0.0.1:6379> XAUTOCLAIM demo:orders:events emails emails-worker-2 0 0-0 COUNT 10
1) "0-0"
2) 1) 1) "1794650000002-0"
      2) 1) "type"
         2) "order-created"
         3) "orderId"
         4) "1002"
         5) "isbn"
         6) "9780141439518"
         7) "quantity"
         8) "1"
3) (empty array)

The released entry's owner became empty and its idle time -1; because its last delivery time was reset, it can be claimed at once with any minimum idle time. There are three modes: SILENT decrements the delivery counter by one (work released on shutdown "does not count"), FAIL leaves the counter as it is (this worker failed, let someone else try), FATAL sets the counter to its maximum value (the message is permanently broken). Valkey 9.1.3 has no XNACK (ERR unknown command) and XREADGROUP … CLAIM gives a syntax error; there you build a separate recovery loop with XAUTOCLAIM.

Bookstore: Order Events and Idempotent XADD

Bookstore.Api writes every new order to the bookstore:orders:events stream. If the network drops, the producer cannot know whether the XADD reached the server, and it retries; a plain XADD then adds the same event twice. Redis 8.6's IDMP option takes a producer ID and an idempotency ID; a second XADD with the same pair returns the first entry's ID instead of adding a new entry. As the ID we use the Idempotency-Key header from post 03:

// XADD bookstore:orders:events IDMP bookstore-api <Idempotency-Key> MAXLEN ~ 100000 * ...
// A retried XADD with the same idempotency key returns the first entry's id instead of a duplicate (Redis 8.6+)
private static Task<RedisValue> PublishOrderCreatedAsync(IDatabase db, CreateOrderRequest request, long orderId, string idempotencyKey) =>
    db.StreamAddAsync(RedisKeys.OrderEvents,
        [
            new("type", "order-created"),
            new("orderId", orderId),
            new("customerId", request.CustomerId),
            new("isbn", request.Isbn),
            new("quantity", request.Quantity)
        ],
        new StreamIdempotentId("bookstore-api", idempotencyKey), maxLength: 100_000, useApproximateMaxLength: true);

dotnet run --project src/Bookstore.RedisLab -- idempotent-xadd

plain attempt 1: 1794655555019-0
plain attempt 2: 1794655555021-0
IDMP  attempt 1: 1794655555022-0
IDMP  attempt 2: 1794655555022-0
XLEN: 3
idmp-duration=100, idmp-maxsize=100, pids-tracked=1, iids-tracked=1, iids-added=1, iids-duplicates=1

The plain XADD created two entries; the second attempt with IDMP returned the same ID. idmp-duration=100 and idmp-maxsize=100 in XINFO STREAM match the defaults in the docs: Redis remembers the last 100 idempotency IDs per producer for at most 100 seconds; with XCFGSET they can be raised to 86,400 seconds and 10,000 IDs. The docs call this feature "at-most-once production": a producer's retry does not create a duplicate entry. But the window is limited; a repeat arriving after 100 seconds becomes a new entry. The real idempotency is still in the API, with SET … NX from post 03; IDMP is the second safeguard behind it. Valkey 9.1.3 has no IDMP: the command took IDMP for an ID and returned Invalid stream ID.

A .NET Worker: a Consumer with BackgroundService

A third project joined the snapshot: src/Bookstore.Worker. It holds two BackgroundServices reading the same stream: the emails group, which sends e-mails, and the stock group, which picks books in the warehouse. Both use the same base class:

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    IDatabase db = redis.GetDatabase();
    // A stable name: after a restart this instance is the same consumer again
    string consumer = $"{Group}-{config["Worker:ConsumerName"] ?? "worker-1"}";
    await EnsureGroupAsync(db);

    while (!stoppingToken.IsCancellationRequested)
    {
        // XREADGROUP GROUP g c COUNT 10 CLAIM 30000 STREAMS key >: idle entries of other consumers first, then new ones (Redis 8.4+)
        StreamEntry[] entries = await db.StreamReadGroupAsync(Stream, Group, consumer, StreamPosition.NewMessages,
            count: 10, noAck: false, claimMinIdleTime: ClaimAfter);
        if (entries.Length == 0)
        {
            await Task.Delay(TimeSpan.FromMilliseconds(500), stoppingToken);   // no BLOCK on a shared multiplexer
            continue;
        }

        foreach (StreamEntry entry in entries)
        {
            if (entry.DeliveryCount >= MaxDeliveries)
            {
                await DeadLetterAsync(db, entry);
                continue;
            }
            try
            {
                await HandleAsync(entry, stoppingToken);
                await db.StreamAcknowledgeAsync(Stream, Group, entry.Id);          // XACK
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                logger.LogWarning("{Group}: {Id} failed ({Error}), released for retry", Group, entry.Id, ex.Message);
                await db.StreamNegativeAcknowledgeAsync(Stream, Group, StreamNackMode.Fail, entry.Id); // XNACK ... FAIL (Redis 8.8+)
            }
        }
    }
}

There are three design decisions. The consumer name is stable and comes from configuration (Worker:ConsumerName); if a new random name were generated on every restart, the old name would stay in the PEL with orphaned entries and XINFO CONSUMERS would grow a little longer with each restart. The BLOCK option of XREADGROUP is not used: as we saw in post 05, blocking commands freeze the shared connection in StackExchange.Redis; waiting 500 ms after an empty read is enough. If you want, you can shorten that wait with a pub/sub signal as in post 09. Finally, Redis 8.10's MAXCOUNT and MAXSIZE options cap the total number of entries and the reply size when reading several streams in one command; for this worker, which reads a single stream, COUNT 10 is enough. The lifecycle and graceful shutdown of a BackgroundService are covered in "IHostedService and BackgroundService in .NET" in the Background Jobs series.

I ran the API and the worker together and placed two orders, then added a malformed event with redis-cli (quantity two):

dotnet run --project src/Bookstore.Worker
curl -X POST http://localhost:5080/orders -H "Content-Type: application/json" -H "Idempotency-Key: k-10-1" -d '{"customerId":7,"isbn":"9780441013593","quantity":2}'
{"orderId":1,"replayed":false}
curl -X POST http://localhost:5080/orders -H "Content-Type: application/json" -H "Idempotency-Key: k-10-2" -d '{"customerId":7,"isbn":"9780141439518","quantity":1}'
{"orderId":2,"replayed":false}
127.0.0.1:6379> XADD bookstore:orders:events * type order-created orderId 9999 customerId 7 isbn 9780451524935 quantity two
"1794655604137-0"

info: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604072-0 2 x 9780441013593 picked from the warehouse
info: Bookstore.Worker.EmailConsumer[0]
      emails: 1794655604072-0 confirmation e-mail for order 1 to customer 7
info: Bookstore.Worker.EmailConsumer[0]
      emails: 1794655604086-0 confirmation e-mail for order 2 to customer 7
info: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604086-0 1 x 9780141439518 picked from the warehouse
info: Bookstore.Worker.EmailConsumer[0]
      emails: 1794655604137-0 confirmation e-mail for order 9999 to customer 7
warn: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604137-0 failed (The input string 'two' was not in a correct format.), released for retry
warn: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604137-0 failed (The input string 'two' was not in a correct format.), released for retry
warn: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604137-0 failed (The input string 'two' was not in a correct format.), released for retry
fail: Bookstore.Worker.StockConsumer[0]
      stock: 1794655604137-0 moved to bookstore:orders:events:dead as 1794655604237-0 after 3 deliveries

The two groups processed the same events without knowing about each other. The malformed event caused no trouble in the emails group (that group does not look at the quantity), while in the stock group it was tried three times. On every failure XNACK FAIL released the event at once, and the next XREADGROUP … CLAIM brought it back with priority; there was no need to wait 30 seconds.

Poison Messages and a Dead-Letter Stream

An event that can never be processed keeps the worker busy and bloats the PEL if it is retried forever. The rule in the worker is simple: when the delivery count reaches 3, the event is copied to the bookstore:orders:events:dead stream and acknowledged. As in post 08, both happen in one MULTI/EXEC; if the copy succeeded and the acknowledgment were lost, the event would end up in the dead-letter stream twice:

// A poison message: copy it to the dead-letter stream and acknowledge it, in one MULTI/EXEC
private async Task DeadLetterAsync(IDatabase db, StreamEntry entry)
{
    NameValueEntry[] fields = [new("sourceId", entry.Id), new("group", Group), new("deliveries", entry.DeliveryCount), .. entry.Values];
    ITransaction tx = db.CreateTransaction();
    Task<RedisValue> added = tx.StreamAddAsync(DeadLetters, fields);
    Task<long> acked = tx.StreamAcknowledgeAsync(Stream, Group, entry.Id);
    await tx.ExecuteAsync();
    logger.LogError("{Group}: {Id} moved to {DeadLetters} as {DeadId} after {Count} deliveries",
        Group, entry.Id, DeadLetters, await added, entry.DeliveryCount);
    await acked;
}

127.0.0.1:6379> XRANGE bookstore:orders:events:dead - +
1) 1) "1794655604237-0"
   2)  1) "sourceId"
       2) "1794655604137-0"
       3) "group"
       4) "stock"
       5) "deliveries"
       6) "3"
       7) "type"
       8) "order-created"
       9) "orderId"
      10) "9999"
      11) "customerId"
      12) "7"
      13) "isbn"
      14) "9780451524935"
      15) "quantity"
      16) "two"

In the dead-letter stream the event sits with where it came from and how many times it was tried; it can be fixed and added back to the main stream. The process of monitoring, inspecting and reprocessing a dead-letter queue is covered in "Managing Dead-Letter Queues: Finding, Inspecting and Reprocessing Poison Messages" in the Working with Message Queues series. One more detail: stream delivery is at-least-once. If a worker processes an event and crashes right before XACK, the event is processed again on another worker; the e-mail may go out twice. Idempotency on the consumer side (an inbox table) is in "Idempotent Consumers and the Inbox" in the same series.

Trimming and Memory

Since a stream does not remove entries that have been read, it grows forever unless you set a limit. I measured with a million order events:

dotnet run --project src/Bookstore.RedisLab -- stream-memory

1,000,000 events: 42,896,854 bytes (42.9 bytes per event)
XTRIM MAXLEN ~ 900000:        removed 9,981, length 990,019 (1 ms)
XTRIM MAXLEN 800000:          removed 190,019, length 800,000 (0 ms)
XTRIM MAXLEN ~ 700000 LIMIT 0: removed 99,956, length 700,044 (0 ms)

A million events take 41 MB; because the field names are written once per node, that is under 43 bytes per entry. As real events carry larger values, this number grows. The really interesting part is the first line: the XADD doc page says approximate trimming with ~ may leave "slightly" more entries than the threshold; in the measurement, 990,019 entries were left against a target of 900,000. The reason is in another sentence on the same page: without LIMIT, Redis examines at most 100 × stream-node-max-entries (default 100), that is 10,000 entries, in a single command. If you use MAXLEN ~ on every XADD, the limit does not matter, because the stream is trimmed a little on every add; but if you try to shrink a stream that has already grown with a single XTRIM ~, you only remove 10,000 entries. LIMIT 0 removes the limit, and 99,956 entries were removed. StackExchange.Redis 3.4.35's StreamTrimAsync, however, rejected limit: 0 on the client side with an ArgumentOutOfRangeException, so I sent that command with ExecuteAsync. Exact trimming (=) was fast at this size too; but because it has to remove entries from the middle of nodes, it is more expensive than the approximate form when called on every add.

To limit by time instead of length there is MINID: since IDs are timestamps, XTRIM key MINID ~ <unix-ms-7-days-ago> keeps the last seven days. Redis 8.2 brought trimming options that take consumer groups into account: ACKED removes only entries acknowledged by every group; the default KEEPREF may also remove an entry still waiting in a PEL, in which case the worker sees nil instead of the content when it reads that entry.

In Production

There are three numbers to watch: each group's lag in XINFO GROUPS (the number of entries the group has not read yet; whether the workers keep up), its pending (entries read but not acknowledged; if it grows, workers are dying or failing) and the length of the dead-letter stream. At the end of the run above, lag and pending were 0 in both groups. Make sure the eviction policy of the Redis holding the stream will not remove this key; a stream silently removed when memory fills up means lost messages (noeviction in post 12). Durability depends on Redis's persistence settings: if the AOF is off, or runs with everysec, a crash can take the latest writes with it; the details are in post 12.

When Streams, and when RabbitMQ or Kafka? If you already run Redis, the message volume fits in memory and retention is measured in days, Streams is a practical choice: consumer groups, the PEL, claiming and replay are all there. If you need large disk-based event logs kept for months, routing and dead-lettering managed by the broker, or a ready integration ecosystem, Kafka or RabbitMQ fit better. The full comparison is in "Getting Started with Message Queues in .NET" in the Working with Message Queues series.

Common Pitfalls

Forgetting XACK, or reading everything with NOACK: in the first case the PEL grows forever, in the second a crashed worker's work is lost.

A stream that grows without limit: entries that have been read are not removed; set MAXLEN ~ or MINID.

Generating a random consumer name on every restart: old names stay in the PEL with orphaned entries.

Assuming a single XTRIM ~ call can shrink a large stream: because of the default limit, 9,981 entries were removed instead of 100,000.

No limit for poison messages: a malformed event is retried forever.

Assuming Redis 8.4–8.8's CLAIM, IDMP and XNACK on Valkey or on an older managed Redis.

A Small Experiment

★☆☆ Start the worker twice with two different Worker:ConsumerName values and place 20 orders. Does XINFO CONSUMERS bookstore:orders:events emails show how the work was shared?

★★☆ Add a transient failure to StockConsumer: it throws on the first attempt and succeeds on the second. Look at how many times the event was processed in the log, and how the delivery counter changes when you use SILENT instead of XNACK FAIL.

★★★ Kill the worker after it processes an order but before XACK (with Environment.FailFast, for example), then restart it. How many times was the event processed? Add a check with SET … NX on a key like bookstore:emails:sent:{orderId} to keep the e-mail from going out twice.

In the next post we move on to Redis's JSON type and the Query Engine: JSON documents, partial updates with JSONPath, indexes with FT.CREATE, catalog search with FT.SEARCH and FT.AGGREGATE, and NRedisStack.