Building a Write-Behind Buffer with Redis Streams

A traffic spike is driving ten thousand writes a second at an endpoint whose backing database comfortably sustains one thousand, and you cannot let the excess either block the request path or vanish. A write-behind buffer solves this by acknowledging the write into Redis immediately and flushing it to the system of record later, on a schedule the database can absorb. The naive version — a plain list or a fire-and-forget publish — loses data the moment a worker crashes mid-flush, because the mutation is gone from Redis before it ever reaches the database. This page builds the durable version on a Redis Stream: every mutation is appended with XADD, consumed by a group of workers with XREADGROUP, flushed to the database in a transaction, and only then acknowledged with XACK. Unacknowledged entries survive a crash in the stream's pending list, so a restarted worker reclaims and replays them instead of dropping them. For where this pattern sits against synchronous alternatives, see the write-behind trade-offs discussed on the parent page.

# Prerequisites

• Redis 7.x reachable from your application host; XAUTOCLAIM requires Redis 6.2+.
• redis-py 5.x on Python 3.10+ (pip install "redis>=5,<6"). Every example uses the redis.asyncio client.
• A system of record (Postgres, MySQL, or similar) whose write throughput you have measured — the flush rate must be tuned to it.
• Writes that are safe to apply at least once: the flush must be idempotent because Redis Streams guarantee at-least-once delivery, not exactly-once.
• A rough ceiling for how many buffered writes you can tolerate in memory, which becomes your MAXLEN cap.

# Step-by-Step Implementation

# 1. Append every mutation with XADD

On the write path, serialize the mutation and append it to a durable stream instead of touching the database. XADD returns a monotonic entry ID (<millis>-<seq>) that both orders the entry and acknowledges receipt to the caller in a single round trip.

import json
import redis.asyncio as redis

client = redis.Redis(host="redis-primary", port=6379, decode_responses=True)

STREAM = "wbstream:orders"

async def enqueue_write(entity_id: str, payload: dict) -> str:
    # MAXLEN with '~' trims approximately, letting Redis drop whole macro-nodes
    # cheaply instead of trimming to an exact length on every append.
    return await client.xadd(
        STREAM,
        {"entity_id": entity_id, "op": "upsert", "data": json.dumps(payload)},
        maxlen=100_000,
        approximate=True,
    )

# 2. Create the consumer group

A consumer group tracks, per group, which entries have been delivered and which have been acknowledged, so multiple workers share the stream without processing the same entry twice. Create it once at startup and swallow the BUSYGROUP error that means it already exists.

async def ensure_group(group: str = "flushers") -> None:
    try:
        # id="0" replays existing backlog; use "$" to consume only new entries.
        await client.xgroup_create(STREAM, group, id="0", mkstream=True)
    except redis.ResponseError as exc:
        if "BUSYGROUP" not in str(exc):
            raise  # a real error, not "group already exists"

# 3. Read batches with XREADGROUP

Each worker claims a batch under a unique consumer name. Reading with count bounds the batch size, and block parks the worker until new entries arrive instead of hot-looping. Entries handed out this way move into the group's pending entries list (PEL) until acknowledged.

async def read_batch(group: str, consumer: str, count: int = 200):
    # ">" means "entries never delivered to any consumer in this group".
    resp = await client.xreadgroup(
        group, consumer, streams={STREAM: ">"}, count=count, block=5000
    )
    # resp: [[stream_name, [(entry_id, {field: value}), ...]]] or None on timeout.
    return resp[0][1] if resp else []

# 4. Flush the batch to the system of record in a transaction

Apply the whole batch inside one database transaction so a partial failure rolls the entire batch back and leaves those entries unacknowledged for a later retry. Coalescing many buffered mutations into one transaction is exactly what lets a slow database absorb a burst.

async def flush_to_db(db_pool, entries) -> list[str]:
    flushed: list[str] = []
    async with db_pool.acquire() as conn:
        async with conn.transaction():  # commit-or-rollback for the whole batch
            for entry_id, fields in entries:
                await conn.execute(
                    "INSERT INTO orders (id, body) VALUES ($1, $2) "
                    "ON CONFLICT (id) DO UPDATE SET body = EXCLUDED.body",
                    fields["entity_id"], fields["data"],
                )
                flushed.append(entry_id)
    return flushed  # only reached if the transaction committed

# 5. Acknowledge only after a durable commit

XACK removes entries from the pending list, telling Redis they are safely persisted. Call it after the transaction commits — never before — so a crash between read and commit leaves the entries pending for recovery rather than silently lost.

async def ack(group: str, entry_ids: list[str]) -> int:
    if not entry_ids:
        return 0
    # Variadic XACK acknowledges the whole committed batch in one call.
    return await client.xack(STREAM, group, *entry_ids)

# 6. Reclaim stuck entries with XAUTOCLAIM

If a worker dies after reading but before acknowledging, its entries sit in the PEL forever unless another worker adopts them. Run a periodic sweep with XAUTOCLAIM (a single command that scans by idle time and reassigns) so any entry idle longer than a threshold is reclaimed and reprocessed.

async def reclaim(group: str, consumer: str, min_idle_ms: int = 60_000):
    # Reassigns entries idle > min_idle_ms to `consumer`, returning them to flush.
    cursor, claimed, _deleted = await client.xautoclaim(
        STREAM, group, consumer, min_idle_time=min_idle_ms, start_id="0-0", count=100
    )
    return claimed  # feed straight back into flush_to_db + ack

# 7. Dead-letter poison messages

An entry that fails every time — malformed payload, a row that violates a constraint — would otherwise be reclaimed forever and stall the buffer. Track delivery count via XPENDING, and once an entry exceeds a retry budget, copy it to a dead-letter stream and XACK the original so the group moves on.

async def dead_letter(group: str, entry_id: str, fields: dict, max_deliveries: int = 5):
    # XPENDING for one entry returns its per-group delivery counter.
    info = await client.xpending_range(STREAM, group, entry_id, entry_id, count=1)
    if info and info[0]["times_delivered"] >= max_deliveries:
        await client.xadd("wbstream:orders:dlq", {**fields, "failed_id": entry_id})
        await client.xack(STREAM, group, entry_id)  # release the poison entry
        return True
    return False

# 8. Cap stream growth with MAXLEN

The stream is durable, so nothing leaves it until you trim. Under sustained load an unbounded stream grows without limit; cap it with an approximate MAXLEN (already applied on XADD in step 1, and enforceable independently with XTRIM) sized so that the retained window always exceeds your worst-case flush lag.

async def trim() -> int:
    # Bound retained entries; '~' trims by whole macro-nodes for O(1) amortized cost.
    return await client.xtrim(STREAM, maxlen=100_000, approximate=True)

The worker loop stitches these together: read a batch, flush it, acknowledge it, and periodically reclaim and dead-letter. Because the acknowledgement is the last step, the buffer is durable across restarts by construction, in the same spirit as offloading purges to async invalidation queues with Celery — the difference is that a stream keeps the backlog inside Redis itself, replayable by entry ID.

# Failure Modes

Worker crash between read and acknowledge. A worker calls XREADGROUP, flushes half the batch, then the process is killed before XACK. Those entries stay in the pending list assigned to a consumer that no longer exists. Diagnose by inspecting the group's pending summary:

redis-cli XPENDING wbstream:orders flushers
# -> total pending, min/max IDs, and per-consumer counts; a dead consumer
#    with a nonzero, non-shrinking count is a leaked batch.

The fix is the step 6 sweep: XAUTOCLAIM reassigns any entry idle past min-idle-time to a live worker, which reprocesses and acknowledges it. Because the flush is idempotent (an ON CONFLICT upsert), replaying an entry that actually did commit before the crash is harmless.

Poison entry stuck in a reclaim loop. A malformed payload throws on every flush, so XAUTOCLAIM keeps handing it back and its times_delivered climbs without bound, blocking progress for everything behind it. Diagnose by ranking pending entries by delivery count:

redis-cli XPENDING wbstream:orders flushers - + 10
# entries with a high delivery count that never leave are poison.

The fix is the dead-letter path from step 7: once times_delivered exceeds the budget, copy the entry to wbstream:orders:dlq and XACK the original so the group advances. Triage the dead-letter stream out of band.

Stream growth outrunning the flushers. If the database slows and the flush rate drops below the ingest rate, the stream grows until it either hits its MAXLEN cap (silently dropping the oldest un-flushed entries) or exhausts memory. Diagnose by watching length against the last-delivered ID:

redis-cli XLEN wbstream:orders
redis-cli XINFO GROUPS wbstream:orders   # compare 'lag' / 'pending' per group

A steadily rising XLEN with growing lag means consumers are behind. Fix by scaling out workers, raising the batch count, or applying backpressure at the producer — and size MAXLEN generously so trimming never discards entries the flushers have not yet drained. A conservative TTL on any cache entries that mirror these writes keeps a downstream reader from trusting a value the buffer has not yet persisted.

# Verification

Confirm the group exists and is consuming from the right position:

redis-cli XINFO GROUPS wbstream:orders
# expect your group with 'pending' near zero in steady state.

Prove durability directly: enqueue an entry, read it into the group without acknowledging, and confirm it is pending; then reclaim and acknowledge it and confirm the pending count returns to zero.

redis-cli XADD wbstream:orders '*' entity_id t-1 op upsert data '{}'
redis-cli XREADGROUP GROUP flushers probe COUNT 1 STREAMS wbstream:orders '>'
redis-cli XPENDING wbstream:orders flushers   # expect total pending = 1

After a full read-flush-ack cycle under load, the pending count should hover near zero and the dead-letter stream should stay empty:

redis-cli XLEN wbstream:orders:dlq            # expect 0 in a healthy pipeline
redis-cli XPENDING wbstream:orders flushers   # expect a small, draining number

# FAQ

# Why use a Redis Stream instead of a list or Pub/Sub for the buffer?

A list (LPUSH/BRPOP) removes an item the instant a worker pops it, so a crash before the database commit loses that write with no record it existed. Pub/Sub is fire-and-forget: any message published while a subscriber is down is gone forever. A stream is a durable append log with per-group tracking — delivered entries stay in the pending list until explicitly acknowledged, so a crashed worker's in-flight batch is recoverable. That durability is the entire reason to prefer a stream for write-behind buffering.

# How do I make the flush idempotent given at-least-once delivery?

Redis Streams guarantee an entry is delivered at least once, so after a crash-and-reclaim the same mutation can be flushed twice. Make the database write idempotent: use an upsert (INSERT ... ON CONFLICT DO UPDATE) keyed on the entity's natural identity, or carry a monotonic version and apply the write only if it is newer. With an idempotent flush, replaying an already-committed entry is a harmless no-op rather than data corruption.

# Does a consumer group preserve ordering?

The stream itself is strictly ordered by entry ID, but a consumer group hands different batches to different workers that flush concurrently, so global ordering across workers is not preserved. If per-entity ordering matters, route all entries for one entity to the same consumer — for example by sharding across several streams on a hash of the entity ID — so a single worker applies that entity's writes in stream order. Do not rely on ordering across independent entities.

# How does MAXLEN trimming interact with unflushed entries?

MAXLEN trims from the head (oldest entries) regardless of whether they have been acknowledged, so a cap set too low can discard entries the flushers have not yet drained — silent data loss. Size the cap well above your worst-case backlog, monitor XLEN and per-group lag, and treat a stream approaching its cap as an incident. The approximate form (MAXLEN ~) trims by whole macro-nodes for cheap amortized cost; use it for the ceiling, not as a substitute for keeping consumers caught up.

# What exactly survives a crash, and what still needs recovery?

Everything appended by XADD is persisted in the stream according to your Redis persistence settings, and every entry delivered to the group but not yet XACKed remains in the pending entries list. On restart, workers do not automatically re-receive those pending entries — > only returns never-delivered ones. You must run XAUTOCLAIM (or XCLAIM after XPENDING) to adopt the orphaned entries from the dead consumer and replay them. Without that reclaim step the pending entries would sit unprocessed indefinitely.

---

Up: Write-Through vs Write-Behind Caching

# Related

• Building Async Invalidation Queues with Celery
• Asynchronous Invalidation Workflows
• Key Tagging Strategies for Bulk Updates
• TTL vs Explicit Invalidation: A Production Reliability Boundary