InventoryAllocation

Implementing Redis Streams for Inventory Polling

A polling loop that applies each delta inline is fragile: if the worker crashes mid-batch, the polled changes it had in memory are gone, and if one channel’s push is slow it stalls the whole poll. Putting a durable buffer between the poller and the appliers fixes both — the poll writes deltas to a log, and a pool of consumers drains that log at its own pace with crash recovery. Redis Streams with consumer groups is the right primitive for that buffer: it gives at-least-once delivery, horizontal fan-out, and per-key ordering without standing up Kafka. This guide implements that buffer for polled inventory deltas, as a companion to async polling for inventory updates, whose polling client feeds the stream.

The design has three moving parts: a producer that XADDs each polled delta, a consumer group whose members XREADGROUP and XACK, and a recovery path that reclaims entries a dead consumer never acknowledged. Ordering per saleable triplet (property_id, room_type_code, rate_plan_code) is preserved by routing each triplet to a stable consumer.

Prerequisites and environment

The stream key is per-property (inv:deltas:{property_id}) so one property’s backlog cannot head-of-line-block another, and the consumer group is created once with MKSTREAM so the first producer does not race the group creation.

Polled deltas buffered in a Redis Stream and fanned out to a consumer group A polling client on the left produces inventory deltas with XADD into a Redis Stream in the centre, which retains entries as an ordered append-only log. A consumer group named inv-appliers reads from the stream with XREADGROUP, distributing entries across two consumers. Each consumer applies the delta to a channel and acknowledges it with XACK, which removes it from the pending list. Below, a recovery path uses XAUTOCLAIM to reclaim entries that a crashed consumer read but never acknowledged, so at-least-once delivery is preserved. A durable buffer between polling and applying XADD produces · XREADGROUP fans out · XACK confirms · XAUTOCLAIM recovers crashes. Polling client httpx snapshot diff Redis Stream · inv:deltas e1 e2 e3 append → Consumer group · inv-appliers consumer A XREADGROUP → XACK consumer B XREADGROUP → XACK XADD read Pending-entries recovery · XAUTOCLAIM reclaim entries a crashed consumer read but never acked → at-least-once
Figure 1: The stream decouples polling from applying; the consumer group fans out work, and XAUTOCLAIM guarantees a crashed consumer's unacked entries are reprocessed.

Step-by-step implementation

Step 1 — Validate then XADD each polled delta

The producer validates each polled row into a canonical delta and appends it to the property’s stream. Cap the stream with MAXLEN ~ so it cannot grow without bound, and let Redis assign the entry ID so ordering is server-authoritative.

python
import redis.asyncio as redis
from pydantic import BaseModel, Field
from datetime import date

class InventoryDelta(BaseModel):
    property_id: str = Field(pattern=r"^[A-Z]{3}-[A-Z]{3}-\d{2}$")  # "LON-STJ-01"
    room_type_code: str = Field(pattern=r"^[A-Z0-9_]+$")            # "DLX_KING"
    rate_plan_code: str = Field(pattern=r"^[A-Z0-9_]+$")            # "BAR"
    channel: str
    stay_date: date
    available_units: int = Field(ge=0)
    version: int = Field(ge=0)     # monotonic per triplet

async def publish_delta(r: redis.Redis, delta: InventoryDelta) -> str:
    stream = f"inv:deltas:{delta.property_id}"
    # approximate MAXLEN (~) is far cheaper than exact trimming on every add
    return await r.xadd(stream, delta.model_dump(mode="json"),
                        maxlen=100_000, approximate=True)

Using approximate=True on MAXLEN lets Redis trim in whole macro-nodes rather than counting exactly on every write, which keeps XADD on the hot polling path cheap while still bounding memory over the long run.

Step 2 — Create the group and consume with XREADGROUP

Create the consumer group idempotently with MKSTREAM, then read new entries with XREADGROUP using > (only messages never delivered to this group). Each consumer has a stable name so ownership of unacked entries is attributable.

python
async def ensure_group(r: redis.Redis, stream: str, group: str = "inv-appliers") -> None:
    try:
        # MKSTREAM creates the stream if absent so the first consumer never races.
        await r.xgroup_create(stream, group, id="0", mkstream=True)
    except redis.ResponseError as exc:
        if "BUSYGROUP" not in str(exc):     # group already exists → fine
            raise

async def consume(r: redis.Redis, stream: str, consumer: str,
                  group: str = "inv-appliers") -> list:
    # ">" = deliver only entries not yet handed to any consumer in this group.
    resp = await r.xreadgroup(group, consumer, {stream: ">"}, count=64, block=2000)
    return resp[0][1] if resp else []       # [(entry_id, fields), ...]

Reading with the special > ID is what makes the group a work queue rather than a broadcast: each new entry is delivered to exactly one consumer, so two consumers draining the same stream share the load instead of both processing every delta.

Step 3 — Apply, then XACK exactly once per entry

Process each entry and acknowledge it only after the apply durably succeeds. Acking before the apply would lose the entry on a crash; acking after guarantees at-least-once, so the apply itself must be idempotent.

python
import structlog
log = structlog.get_logger()

async def handle_batch(r: redis.Redis, stream: str, entries: list,
                       group: str = "inv-appliers") -> None:
    for entry_id, fields in entries:
        delta = InventoryDelta(**fields)
        try:
            await apply_to_channel(delta)   # MUST be idempotent (at-least-once)
        except TransientApplyError:
            # Do NOT ack — leave it pending so recovery or a retry reprocesses it.
            log.warning("delta.apply_deferred", entry_id=entry_id,
                        triplet=_triplet(delta))
            continue
        await r.xack(stream, group, entry_id)   # remove from the pending list
        log.info("delta.acked", entry_id=entry_id, triplet=_triplet(delta))

Acking strictly after a successful apply is the crux of at-least-once: a crash between apply and ack leaves the entry pending and it is reprocessed, so the applied write must be idempotent — which is why every delta carries a version the destination can use to drop a duplicate.

Step 4 — Reclaim pending entries a dead consumer abandoned

If a consumer dies after reading but before acking, its entries sit in the Pending Entries List forever. A janitor uses XAUTOCLAIM to transfer entries idle longer than a threshold to a live consumer, which then processes and acks them.

python
async def reclaim_stale(r: redis.Redis, stream: str, consumer: str,
                        group: str = "inv-appliers", min_idle_ms: int = 60_000) -> int:
    cursor, reclaimed = "0-0", 0
    while True:
        cursor, entries, _ = await r.xautoclaim(
            stream, group, consumer, min_idle_time=min_idle_ms, start_id=cursor, count=64)
        if entries:
            await handle_batch(r, stream, entries)   # same apply+ack path
            reclaimed += len(entries)
        if cursor == "0-0":                          # scan complete
            break
    if reclaimed:
        log.info("delta.reclaimed", stream=stream, count=reclaimed)
    return reclaimed

Setting min_idle_time to a value comfortably above a healthy consumer’s processing time is what prevents the janitor from stealing entries a slow-but-alive consumer is still working — reclaiming too eagerly causes double processing, the opposite of the recovery you wanted.

Gotchas and production notes

Verification snippet

Confirm the two durability guarantees — an unacked entry stays pending and is recoverable, and an acked entry is gone from the PEL.

python
import pytest

@pytest.mark.asyncio
async def test_unacked_entry_is_recoverable(r):
    stream = "inv:deltas:LON-STJ-01"
    await ensure_group(r, stream)
    delta = InventoryDelta(property_id="LON-STJ-01", room_type_code="DLX_KING",
                           rate_plan_code="BAR", channel="booking_com",
                           stay_date="2026-08-14", available_units=2, version=1010)
    await publish_delta(r, delta)

    # Consumer A reads but "crashes" before ack → entry sits in the PEL.
    read = await consume(r, stream, "consumer-A")
    assert len(read) == 1
    pending = await r.xpending(stream, "inv-appliers")
    assert pending["pending"] == 1               # unacked → still pending

    # Recovery reclaims and processes it, then the PEL is empty.
    await reclaim_stale(r, stream, "consumer-B", min_idle_ms=0)
    pending_after = await r.xpending(stream, "inv-appliers")
    assert pending_after["pending"] == 0         # reclaimed, applied, acked

The pending count going from one to zero across a simulated crash is the assertion that matters: it proves an entry a consumer read but never acknowledged is not lost but reprocessed, which is the entire reason for putting a stream between polling and applying. In production, alert whenever XPENDING stays above zero for longer than one poll interval.

← Back to Async Polling for Inventory Updates