Buffering Cost Metrics with Redis Streams

This page uses Redis Streams through redis.asyncio as a durable buffer that decouples metric pollers from aggregators — an XADD producer, an XREADGROUP consumer group with XACK, MAXLEN trimming to bound memory, and XAUTOCLAIM to reclaim entries stranded by a crashed consumer.

Back to: Real-Time Metric Streaming Setup

A cost pipeline has two halves that run at different speeds: pollers that emit signals in bursts (a CloudWatch tick fires dozens of records at once, a logical replication slot can burst on a bulk commit) and aggregators that fold those signals into rollups at a steadier pace. Wire them directly and a burst either drops records or stalls the poller. Redis Streams is the buffer in between — an append-only log with consumer groups, per-message acknowledgement, and at-least-once redelivery — that absorbs bursts and lets aggregators scale out and recover from crashes without losing a metric. It is deliberately lighter than a full Kafka cluster: when your throughput fits in one Redis instance and you already run Redis, Streams gives you most of the durable-log semantics with none of the broker operations. The records it buffers still cross the same schema validation for billing data contract before they are aggregated.

Memory is the constraint that makes trimming mandatory. An untrimmed stream grows linearly with ingest rate, so MAXLEN bounds it to a working set:

stream_bytes≈entry_size×min⁡(arrivals, MAXLEN)\text{stream\_bytes} \approx \text{entry\_size} \times \min(\text{arrivals},\ \text{MAXLEN})

Capping MAXLEN at a few times your worst-case backlog keeps memory bounded while still tolerating a consumer that briefly falls behind.

Prerequisites

Before running the buffer, confirm Redis and the client are ready.

  • Redis: 6.2 or newer. XAUTOCLAIM requires 6.2+; consumer groups and XADD MAXLEN work from 5.0 but the reclaim code below uses XAUTOCLAIM.

  • Python: 3.10+ with the unified redis package, which provides redis.asyncio (the former aioredis, now merged in).

    pip install "redis>=5.0"

  • Access control: in production use a Redis ACL user restricted to the stream keys and the stream commands, not the default all-powerful user — the same least-privilege posture as access control for cost data.

    ACL SETUSER cost_buffer on >SECRET ~cost:stream:* +xadd +xreadgroup +xack +xautoclaim +xgroup +xpending

  • Connection: read the URL from os.environ["REDIS_URL"]; never hard-code credentials.

Step-by-Step Implementation

The producer XADDs each cost record and lets Redis trim the stream. A consumer group lets multiple aggregators split the load; each reads with XREADGROUP, processes, and XACKs. A janitor uses XAUTOCLAIM to reclaim entries a crashed consumer left un-acked in the Pending Entries List.

Step 1 — Produce with XADD and bounded MAXLEN

Each record is a flat field/value map (Redis Streams values are strings), added with an approximate MAXLEN (~) so Redis trims efficiently at segment boundaries rather than on every insert. The auto-generated * ID is a millisecond timestamp plus sequence, so entries are globally ordered.

import os
import redis.asyncio as redis

STREAM = "cost:stream:metrics"
MAXLEN = 100_000                     # bound the buffer to ~100k entries

async def publish_metric(r: redis.Redis, record: dict[str, str]) -> str:
    """Append one cost record to the stream, trimming approximately."""
    # approximate=True (the '~') lets Redis trim at macro-node boundaries — cheaper.
    entry_id = await r.xadd(
        STREAM,
        record,
        maxlen=MAXLEN,
        approximate=True,
    )
    return entry_id

# Expected:
#   await publish_metric(r, {"db_instance": "prod-orders-1",
#                            "metric": "CPUUtilization", "value": "63.4",
#                            "ts": "2026-07-18T14:03:00Z"})
#   -> '1752847380000-0'   (millisecond-timestamp-sequence stream ID)

Step 2 — Consume with a group and acknowledge

A consumer group tracks a per-group cursor and a Pending Entries List (PEL) of delivered-but-un-acked entries per consumer. Create the group once with mkstream=True, then each aggregator reads with its own consumer name using the special > ID (undelivered entries only), processes, and XACKs so the entry leaves the PEL.

from redis.exceptions import ResponseError

GROUP = "aggregators"

async def ensure_group(r: redis.Redis) -> None:
    """Create the consumer group idempotently (id='0' reads from the start)."""
    try:
        await r.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
    except ResponseError as exc:
        if "BUSYGROUP" not in str(exc):     # already exists is fine
            raise

async def consume(r: redis.Redis, consumer: str, handle) -> None:
    """Read, process, and acknowledge cost records for one consumer."""
    while True:
        # '>' = only entries never delivered to this group.
        resp = await r.xreadgroup(
            GROUP, consumer, streams={STREAM: ">"},
            count=100, block=5000,          # block up to 5s for new entries
        )
        if not resp:
            continue                        # idle timeout, loop again
        for _stream, entries in resp:
            for entry_id, fields in entries:
                await handle(fields)        # fold into rollup / forward
                await r.xack(STREAM, GROUP, entry_id)   # remove from PEL

Expected: with two consumers agg-1 and agg-2 in the group, Redis splits new entries between them, and after processing each consumer’s PEL length (via XPENDING) returns to zero.

XPENDING cost:stream:metrics aggregators
1) (integer) 0        # no un-acked entries when consumers keep up

Step 3 — Reclaim stranded entries with XAUTOCLAIM

If a consumer crashes mid-batch, its entries stay in the PEL, delivered but never acked — invisible to > reads by other consumers. A janitor runs XAUTOCLAIM to transfer entries idle longer than a threshold to a live consumer, which reprocesses and acks them. This is what makes the buffer at-least-once across consumer failure.

The diagram below shows the buffer end to end: pollers XADD, the group fans entries to aggregators that XACK, and a janitor XAUTOCLAIMs entries a dead consumer stranded.

Redis Streams cost buffer: XADD producer, XREADGROUP fan-out with XACK, MAXLEN trimming, and XAUTOCLAIM reclaim of stranded entriesMetric pollers append records with XADD to a Redis stream trimmed with MAXLEN. A consumer group fans undelivered entries via XREADGROUP to two aggregator consumers that fold each record into a rollup and XACK it out of the pending entries list. One consumer has crashed, stranding its unacked entries; a janitor runs XAUTOCLAIM to move those idle entries to a live consumer that reprocesses and acknowledges them, giving at-least-once delivery across failure.MetricpollersRedis streamgroup: aggregatorsMAXLEN ~100kagg-1 (live)process + XACKagg-2 (crashed)entries stranded in PELJanitorXAUTOCLAIMXADDXREADGROUP >XACKreclaim idle entries → live consumer
import logging

async def reclaim_stranded(r: redis.Redis, consumer: str,
                           min_idle_ms: int = 60_000) -> int:
    """Move entries idle > min_idle_ms to `consumer`, reprocess, and ack."""
    reclaimed, cursor = 0, "0-0"
    while True:
        # XAUTOCLAIM returns (next_cursor, claimed_entries, deleted_ids).
        cursor, entries, _deleted = await r.xautoclaim(
            STREAM, GROUP, consumer,
            min_idle_time=min_idle_ms,
            start_id=cursor,
            count=100,
        )
        for entry_id, fields in entries:
            if fields:                       # None => entry was trimmed; skip
                await _reprocess(fields)
                await r.xack(STREAM, GROUP, entry_id)
                reclaimed += 1
        if cursor == "0-0":                  # cursor wraps to 0-0 when done
            break
    logging.info("reclaimed %d stranded entries", reclaimed)
    return reclaimed

async def _reprocess(fields: dict) -> None:
    ...  # idempotent fold — same handler the live consumers use

Expected janitor log after agg-2 crashes mid-batch:

INFO:root:reclaimed 37 stranded entries

Verification

Confirm the buffer holds and reclaims correctly before relying on it under burst.

  1. Check group lag and PEL depth:

    redis-cli XINFO GROUPS cost:stream:metrics
# -> shows 'lag' (undelivered entries) and 'pending' (delivered, un-acked) per group

  2. Expected steady state: with consumers keeping up, pending is near zero and lag stays well under MAXLEN. A pending count that only grows points at a consumer that reads but never acks.

  3. Fault-injection test. Kill a consumer mid-batch, confirm its entries appear in XPENDING ... IDLE, run the janitor, and verify they are reprocessed and the PEL drains to zero — proving no cost record is lost across a consumer crash.

Gotchas & Edge Cases

  • Reading with > never redelivers. The > ID returns only never-delivered entries, so a crashed consumer’s in-flight entries are invisible to everyone else until XAUTOCLAIM (or XCLAIM) moves them. Without a janitor, those records sit in the PEL forever and are silently lost from aggregation — running the reclaim loop is not optional.
  • MAXLEN can trim un-acked entries. Trimming removes the oldest entries regardless of ack state, so a consumer that falls far enough behind can have entries trimmed out from under it — XAUTOCLAIM then returns None fields for those IDs (handle by skipping and acking). Size MAXLEN above your worst-case backlog, and alarm on lag approaching it.
  • Approximate trimming keeps more than MAXLEN. With approximate=True the stream can hold somewhat more than MAXLEN because Redis trims at macro-node boundaries. That is the right trade for throughput; use exact trimming (approximate=False) only when a hard cap is a correctness requirement, and expect higher CPU.
  • Idempotent handlers are mandatory. At-least-once means a reclaimed entry may be processed twice (crash after handling, before ack). Make the fold idempotent — key rollups on the stream entry ID or the record’s natural key — the same discipline the retry logic for failed metric pulls relies on.
  • Redis persistence bounds durability. Streams live in memory; a crash loses anything since the last AOF fsync or RDB snapshot. For cost data you cannot lose, enable AOF with appendfsync everysec and accept that Streams is a buffer, not the system of record — the validated ledger downstream is.
  • Poison entries block a partition of work. A record that always throws will be reclaimed, reprocessed, and re-stranded forever. Cap delivery attempts (track via XPENDING’s delivery count) and route repeat offenders to a dead-letter stream rather than looping, exactly as the error-handling model prescribes.

Frequently Asked Questions

When should I use Redis Streams instead of Kafka?

Use Redis Streams when your throughput fits comfortably in one Redis instance, you already operate Redis, and you want durable-log semantics without running and tuning a broker cluster. Reach for Kafka when you need multi-terabyte retention, partition counts in the hundreds, multi-datacenter replication, or an existing Kafka-based ecosystem. For a cost-metric buffer between pollers and aggregators, Streams is usually the lighter, sufficient choice.

Does XREADGROUP guarantee each entry is processed exactly once?

No — it guarantees at-least-once. An entry is delivered to exactly one consumer initially, but if that consumer crashes before XACK, XAUTOCLAIM redelivers it, so the same entry can be handled more than once. Exactly-once is achieved by making your handler idempotent, not by the stream.

How do I size MAXLEN?

Set it to a few multiples of your worst-case backlog: peak ingest rate times the longest tolerable consumer outage. Too small and a brief consumer stall trims un-acked entries; too large and memory grows without benefit. Monitor the group’s lag; if it routinely climbs toward MAXLEN, either raise the cap or add consumers rather than letting trimming silently drop cost records.

What min-idle-time should the janitor use?

Set min_idle_time comfortably above your longest legitimate processing time so the janitor never steals an entry a healthy consumer is still working. Sixty seconds is a reasonable default for fast folds; if a handler can legitimately take minutes, raise it, or you will reclaim and double-process in-flight work.

Can multiple aggregators safely run the janitor?

Yes. XAUTOCLAIM is atomic, so if several janitors run concurrently each claims a disjoint set of entries — no entry is handed to two consumers at once. Still, one dedicated janitor (or leader-elected instance) is simpler to reason about and avoids redundant scans of the pending list.

Back to: Real-Time Metric Streaming Setup