Redis Streams and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Redis Streams is a persistent, append-only log with consumer group semantics — XREADGROUP delivers messages, the Pending Entries List tracks what has not been acknowledged, and XAUTOCLAIM can reassign idle entries to a new consumer. When Streams drives billing, three Redis-specific behaviors introduce Stripe double-charge failure modes that XPENDING and XINFO GROUPS do not surface as billing risk.

This post covers those three failure modes with redis-py Python code, content-hash idempotency keys stable across XAUTOCLAIM redeliveries and AOF replay, per-billing-period vault keys via a spend-cap proxy, and per-event XACK patterns — the two-layer governance approach that closes all three without changing stream configuration, consumer group settings, or Redis persistence policy.

Failure mode 1: XAUTOCLAIM reclaims a billing message while stripe.charges.create() is still in-flight — Redis Streams has no lock-renewal mechanism

Every message delivered to a consumer group via XREADGROUP is added to the Pending Entries List (PEL) for that consumer. The PEL entry tracks: the message ID, the consumer name that last received it, the idle time (milliseconds since last delivery), and the delivery count. The message remains in the PEL until the consumer calls XACK with its message ID. Until that XACK, any caller with access to the stream can call XAUTOCLAIM or XCLAIM to transfer ownership of the PEL entry to a different consumer, provided the entry’s idle time exceeds the specified min-idle-time.

The critical difference from other message systems: Redis Streams has no way to extend a message’s processing window. In SQS, a worker can call change_message_visibility to push out the visibility timeout and buy more processing time. In Azure Service Bus, AutoLockRenewer periodically calls renew_message_lock() to extend the lock token. In Google Pub/Sub, the StreamingPullFuture lease manager thread calls ModifyAckDeadline on a schedule. Redis Streams offers none of these. The idle time for a PEL entry increments continuously from the moment of delivery. A consumer processing a billing event for 75 seconds has a PEL entry with idle=75000 ms, and any call to XAUTOCLAIM with min-idle-time=60000 at T=60 will claim that entry — without knowing whether the owning consumer is still actively processing it.

The failure scenario: a billing consumer reads a usage event from the stream via XREADGROUP. The message is added to the consumer’s PEL at T=0 with idle=0. The consumer calls stripe.charges.create(), which takes 90 seconds under load (stripe-python’s max_network_retries=2 with a 30-second per-attempt timeout can run to 90+ seconds in worst case). At T=60, a monitoring job or a second consumer in the group calls XAUTOCLAIM billing-stream billing-group consumer-B 60000 0-0. Redis transfers ownership of the PEL entry to consumer-B. Consumer-B reads the message body from the stream and calls stripe.charges.create() for the same customer. Consumer-A’s Stripe call returns ch_A at T=75. Consumer-B’s Stripe call creates ch_B at T=62. Consumer-A then calls XACK — which succeeds (Redis removes the PEL entry) — but ch_B already exists on Stripe. XPENDING shows 0 pending messages. XINFO GROUPS shows pending=0. Redis has no record that the message was delivered twice.

import redis
import stripe
import json
import os

r = redis.Redis.from_url(os.environ["REDIS_URL"])
STREAM = "billing-stream"
GROUP = "billing-group"
CONSUMER = "worker-1"

while True:
    # Read up to 1 message at a time from the consumer group.
    # The message enters the PEL the moment XREADGROUP returns it.
    # idle time starts counting from T=0. No mechanism to extend it.
    results = r.xreadgroup(GROUP, CONSUMER, {STREAM: ">"}, count=1, block=5000)

    if not results:
        continue

    for stream_name, messages in results:
        for msg_id, fields in messages:
            event = json.loads(fields[b"payload"])
            customer_id = event["customer_id"]
            billing_period = event["billing_period"]

            stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

            # This call can take 90 seconds under load.
            # If any other consumer or monitoring job calls XAUTOCLAIM
            # with min-idle-time=60000 after T=60, this message will be
            # transferred to that consumer, which calls stripe.charges.create()
            # for the same customer. Redis has no lock-renewal mechanism —
            # idle time counts continuously from XREADGROUP delivery.
            charge = stripe.Charge.create(
                amount=event["amount_cents"],
                currency="usd",
                customer=event["stripe_customer_id"],
                description=f"Usage billing for {billing_period}",
                # no idempotency_key — XAUTOCLAIM consumer creates ch_B
            )

            # XACK removes the PEL entry. If consumer-B already XACK'd
            # the reclaimed entry (after creating ch_B), this XACK
            # is a no-op: Redis returns 0 (entry already removed).
            # The duplicate charge is invisible at the stream level.
            r.xack(STREAM, GROUP, msg_id)
            print(f"Charged {customer_id}: {charge['id']}")

The failure is invisible in Redis monitoring. XPENDING billing-stream billing-group - + 10 returns 0 entries after both consumers have processed the message — the PEL is empty regardless of whether one or two Stripe charges were created. XINFO GROUPS billing-stream reports pending=0 and consumers=2, with no indication that both consumers processed the same message ID. Redis’s stream length (XLEN billing-stream) is unchanged: the message remains in the stream after XACK (stream entries are only removed by explicit XDEL or MAXLEN trimming). From Redis’s perspective, the message was delivered once to consumer-A, reclaimed by consumer-B, and acknowledged by consumer-B. From Stripe’s perspective, the customer was charged twice.

The XAUTOCLAIM failure is particularly likely in deployments that use a reaper or “stuck message” monitor — a common pattern where a separate process periodically calls XAUTOCLAIM to reclaim messages stuck in the PEL for longer than a threshold. If that threshold is set to 60 seconds (to detect workers that died mid-processing), it will also reclaim messages from workers that are alive but processing a Stripe call that takes longer than 60 seconds. The reaper cannot distinguish a dead consumer from a slow one: both look identical in the PEL (idle time > threshold, no XACK).

Failure mode 2: XREADGROUP with COUNT > 1 adds all batch messages to the PEL atomically — a partial batch failure leaves all messages pending, and XAUTOCLAIM re-delivers the entire batch including events already billed

XREADGROUP with COUNT=10 returns up to 10 messages from the stream in a single Redis round trip. All 10 messages are added to the consumer’s PEL simultaneously with delivery. Each message has its own PEL entry and its own message ID and must be individually XACK’d — there is no batch acknowledgment in Redis Streams. A consumer that processes the 10 messages sequentially and calls XACK per message as it completes is correct. A consumer that reads the batch, processes all 10 in sequence, and then calls XACK for all 10 only after processing the last one — or that calls XACK on multiple IDs in a single XACK billing-stream billing-group id1 id2 id3 ...id10 at the end — has no way to partially acknowledge the batch if processing fails mid-batch.

The failure scenario: a billing consumer calls XREADGROUP with COUNT=10 and reads billing events for customers 1–10. The consumer processes them sequentially: it calls stripe.charges.create() for customer 1, customer 2, ..., customer 6 (ch_A1 through ch_A6 are created on Stripe), and then receives stripe.error.Timeout on customer 7’s charge. The exception breaks out of the processing loop. Events 1–6 were billed successfully, but none of them were XACK’d because the consumer deferred XACK to the end of the loop (or the exception handler does not call XACK for processed entries). All 10 PEL entries age. When XAUTOCLAIM runs with min-idle-time=60000, it reclaims all 10 entries and delivers the entire batch to consumer-B. Consumer-B calls stripe.charges.create() for customers 1–10. Customers 1–6 each receive a second Stripe charge — ch_B1 through ch_B6 alongside ch_A1 through ch_A6.

import redis
import stripe
import json
import os

r = redis.Redis.from_url(os.environ["REDIS_URL"])
STREAM = "billing-stream"
GROUP = "billing-group"
CONSUMER = "worker-1"

results = r.xreadgroup(GROUP, CONSUMER, {STREAM: ">"}, count=10, block=5000)
if not results:
    exit()

for stream_name, messages in results:
    # messages contains up to 10 (msg_id, fields) tuples.
    # All 10 are already in the PEL from the moment xreadgroup returned.
    charged_ids = []

    for msg_id, fields in messages:
        event = json.loads(fields[b"payload"])
        customer_id = event["customer_id"]

        try:
            charge = stripe.Charge.create(
                amount=event["amount_cents"],
                currency="usd",
                customer=event["stripe_customer_id"],
                description=f"Usage billing for {event['billing_period']}",
                # no idempotency_key
                api_key=os.environ["STRIPE_SECRET_KEY"],
            )
            charged_ids.append(msg_id)
            # Does NOT call XACK here — defers to end of loop.

        except stripe.error.Timeout:
            # Processing aborts here. charged_ids holds msg IDs for
            # customers 1-6 who were successfully billed.
            # But XACK has not been called for ANY of them.
            # All 10 PEL entries will age until XAUTOCLAIM reclaims them.
            # The reclaiming consumer re-bills customers 1-6.
            raise

    # This line is never reached if Timeout is raised above.
    # Customers 1-6 are billed but never XACK'd.
    if charged_ids:
        r.xack(STREAM, GROUP, *charged_ids)

The batch-failure mode is compounded by the common practice of using COUNT > 1 for throughput. A consumer reading 50 billing events per batch and failing on event 30 leaves 29 successfully billed customers unacknowledged alongside 21 unbilled ones. When XAUTOCLAIM reclaims the idle batch, the new consumer re-bills those 29 customers. The damage scales with batch size: larger batches mean more customers re-billed per partial failure.

The failure also surfaces through a subtler path: a consumer that does call per-event XACK but whose process is OOM-killed or receives SIGKILL between processing event 6 and calling XACK for event 6. The kernel does not flush in-process state on OOM kill. If the consumer had called stripe.charges.create() for customer 6 and the charge succeeded server-side, but the XACK for event 6 was never sent to Redis, event 6 stays in the PEL and will be reclaimed. Even a consumer with correct per-event XACK logic has a crash window between the Stripe call returning and the XACK reaching Redis. The content-hash idempotency key and pre-flight database check close that window regardless of batch size.

Failure mode 3: appendfsync everysec creates a 1-second XACK data-loss window — after Redis primary failover, billing messages are restored to PEL as unacknowledged and re-delivered

Redis persistence has three main modes: appendfsync always (every write fsynced to disk before ACK, safest), appendfsync everysec (writes batched and fsynced to disk every second, default and most common in production), and appendfsync no (OS decides when to flush, fastest but least durable). The vast majority of Redis deployments — including managed Redis offerings on AWS ElastiCache, Google Cloud Memorystore, Azure Cache for Redis, and Redis Cloud — use appendfsync everysec by default.

appendfsync everysec explicitly trades up to 1 second of write durability for throughput. On an unclean shutdown (power loss, OOM kill, kernel panic), the Redis Append-Only File on disk may be missing up to 1 second of write operations. XACK is a write operation. If a consumer calls stripe.charges.create(), the charge succeeds server-side (ch_A), the consumer writes to billing_records, and then calls XACK — all within 1 second — and the Redis primary crashes before the next AOF fsync, the replica that becomes primary (via Redis Sentinel or Cluster failover) has the AOF as of the last fsync. That AOF does not include the XACK call. The PEL entry for that billing message is restored as still-pending. The new primary delivers the message to the next consumer in the group (or an XAUTOCLAIM job reclaims it). That consumer calls stripe.charges.create() for the same customer — creating ch_B alongside ch_A that Stripe processed before the Redis primary crashed.

import redis
import stripe
import json
import os
import psycopg2

r = redis.Redis.from_url(os.environ["REDIS_URL"])
STREAM = "billing-stream"
GROUP = "billing-group"
CONSUMER = "worker-1"

while True:
    results = r.xreadgroup(GROUP, CONSUMER, {STREAM: ">"}, count=1, block=5000)
    if not results:
        continue

    for stream_name, messages in results:
        for msg_id, fields in messages:
            event = json.loads(fields[b"payload"])
            customer_id = event["customer_id"]
            billing_period = event["billing_period"]

            # Step 1: stripe.charges.create() — ch_A created on Stripe servers.
            charge = stripe.Charge.create(
                amount=event["amount_cents"],
                currency="usd",
                customer=event["stripe_customer_id"],
                description=f"Usage billing for {billing_period}",
                # no idempotency_key
                api_key=os.environ["STRIPE_SECRET_KEY"],
            )

            # Step 2: write to billing_records (PostgreSQL commit, durable).
            conn = psycopg2.connect(os.environ["DATABASE_URL"])
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO billing_records (customer_id, billing_period, charge_id) "
                    "VALUES (%s, %s, %s)",
                    (customer_id, billing_period, charge["id"]),
                )
            conn.commit()

            # Step 3: XACK — this write may NOT reach the AOF before Redis primary crashes.
            # If the Redis primary crashes in the ~1-second window between this XACK
            # and the next appendfsync everysec flush, the replica that takes over
            # as primary will have the PEL entry as still-pending.
            # The next consumer to receive the re-delivered message calls
            # stripe.charges.create() again — no idempotency_key means ch_B is created.
            r.xack(STREAM, GROUP, msg_id)

            # billing_records already has charge_id from ch_A (PostgreSQL is durable).
            # But the next consumer doesn't know to check it — it calls stripe.charges.create()
            # without a pre-flight check and creates ch_B.
            print(f"Charged {customer_id}: {charge['id']}")

The AOF data-loss failure mode is particularly dangerous because it is completely invisible at the application layer. The billing consumer code above is correct: it creates a Stripe charge, writes to the database, and calls XACK. There is no bug in the consumer logic. The failure occurs at the Redis infrastructure layer, below the application. The billing engineer can review the consumer code, the database records, and the Redis stream state (before the crash) and find nothing wrong. The duplicate charge appears on customer statements hours or days after the incident, after the Redis primary has been replaced and its AOF has been rotated.

Managed Redis offerings complicate this further: Redis Cluster with appendfsync everysec across multiple shards means multiple shards can each lose up to 1 second of writes simultaneously during a datacenter event. A billing consumer group that distributes load across multiple shards (using hash slots) can create duplicate charges on multiple shards in parallel during a cluster-wide failover event, multiplying the number of affected billing messages. The failure rate is proportional to the number of XACK calls processed in the 1-second AOF window across all shards.

The fix: content-hash idempotency key + vault key + pre-flight check + per-event XACK

All three failure modes share the same root cause: stripe.charges.create() is called more than once for the same (customer_id, billing_period) pair, and neither Redis Streams’ PEL mechanics nor Stripe’s API layer prevents the second call from creating a new charge when the idempotency key is absent or rotated. The fix has three independent components that together close all three modes without changing stream configuration, consumer group settings, or appendfsync policy.

Layer 1: content-hash idempotency key stable across XAUTOCLAIM redeliveries and AOF replay

The Stripe idempotency key must be derived from the billing event’s stable business fields — fields present in the message body that represent the same business intent regardless of which Redis consumer processes them or how many times the PEL entry has been reclaimed. The key must not include any Redis Streams metadata that changes across deliveries or redeliveries. Fields to exclude: the Redis message ID (a millisecondTimestamp-sequenceNumber string assigned at XADD time — unique per insertion, not per business event; a re-published message from a DLQ-equivalent gets a new ID), delivery-count from XPENDING (incremented on each XCLAIM/XAUTOCLAIM transfer), and any consumer name from the PEL (changes on every XAUTOCLAIM transfer). These fields are envelope metadata, not business identity. Including any of them produces a different idempotency key on each redelivery or reclaim, defeating Stripe’s 24-hour deduplication window.

import hashlib
import stripe
import os
import json
import psycopg2

def make_idempotency_key(customer_id: str, billing_period: str) -> str:
    # Derived from stable business identity: customer + billing period only.
    # Must NOT include: Redis message ID (millisecondTimestamp-sequenceNumber
    # assigned at XADD; a re-added message gets a new ID), delivery count
    # from XPENDING (increments on each XAUTOCLAIM transfer), consumer name
    # from the PEL (changes on every XAUTOCLAIM reclaim), or stream name
    # (changes if the billing event is moved to a different stream).
    # All of these change across XAUTOCLAIM redeliveries and AOF replay.
    raw = f"{customer_id}:{billing_period}:redis-streams-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]


def process_billing_event(conn, event: dict) -> dict:
    customer_id = event["customer_id"]
    stripe_customer_id = event["stripe_customer_id"]
    amount_cents = event["amount_cents"]
    billing_period = event["billing_period"]

    idempotency_key = make_idempotency_key(customer_id, billing_period)

    # Pre-flight check: was this customer already billed for this period?
    # Authoritative source: billing_records (PostgreSQL — durable through Redis failover).
    # This closes all three Redis Streams failure modes:
    # (1) XAUTOCLAIM: if the first consumer wrote billing_records before the
    #     XAUTOCLAIM reclaim, the second consumer's pre-flight check skips the Stripe call.
    # (2) Batch partial failure: if billing_records was written for customers 1-6,
    #     the XAUTOCLAIM consumer's pre-flight check finds existing rows and skips those 6.
    # (3) AOF data loss: billing_records is in PostgreSQL (survives Redis failover).
    #     The consumer that re-receives the AOF-restored message finds the row and skips.
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT charge_id, status FROM billing_records
            WHERE customer_id = %s AND billing_period = %s
            """,
            (customer_id, billing_period),
        )
        existing = cur.fetchone()

    if existing:
        return {
            "skipped": True,
            "reason": "already_billed",
            "charge_id": existing[0],
        }

    vault_key = os.environ["KEYBRAKE_VAULT_KEY"]  # per-billing-period key with spend cap

    charge = stripe.Charge.create(
        amount=amount_cents,
        currency="usd",
        customer=stripe_customer_id,
        description=f"Usage billing for {billing_period}",
        idempotency_key=idempotency_key,
        api_key=vault_key,   # proxied through spend-cap layer
    )

    # Write to billing_records BEFORE calling XACK.
    # Write-before-ack: if XACK is lost (AOF data loss) and the message is
    # re-delivered, the next consumer's pre-flight check finds this row.
    # ON CONFLICT DO NOTHING closes the XAUTOCLAIM race:
    # if two consumers simultaneously passed the pre-flight check (XAUTOCLAIM
    # fired just after both consumers read no existing row), the unique constraint
    # ensures only one INSERT succeeds — the second is silently ignored.
    # The idempotency key ensures both consumers receive ch_A from Stripe.
    with conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO billing_records
                (customer_id, billing_period, charge_id, amount_cents, status, created_at)
            VALUES (%s, %s, %s, %s, 'charged', NOW())
            ON CONFLICT (customer_id, billing_period) DO NOTHING
            """,
            (customer_id, billing_period, charge["id"], amount_cents),
        )
    conn.commit()

    return {"charge_id": charge["id"], "status": "charged"}

Layer 2: vault key with per-billing-period spend cap

The vault key is a per-billing-period Stripe key issued through a spend-cap proxy. The vault key enforces a maximum spend for the billing run, set at 110% of the expected total. For the AOF data-loss failure mode — where a re-delivered message creates a second Stripe call after Stripe’s 24-hour idempotency window has closed (a message held in the PEL for more than 24 hours due to a persistent consumer failure, then reclaimed after AOF replay) — the vault key caps worst-case spend at expected_total × 1.10, providing a hard backstop even when the content-hash idempotency key’s 24-hour window has elapsed.

import requests
import os
import datetime

def create_billing_vault_key(
    billing_period: str,
    expected_total_cents: int,
) -> str:
    resp = requests.post(
        "https://proxy.keybrake.com/vault/keys",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
        json={
            "vendor": "stripe",
            "label": f"redis-streams-billing-run-{billing_period}",
            "daily_usd_cap": round((expected_total_cents * 1.10) / 100, 2),
            "allowed_endpoints": ["/v1/charges", "/v1/payment_intents"],
            "expires_at": (
                datetime.datetime.utcnow() + datetime.timedelta(hours=6)
            ).isoformat() + "Z",
        },
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

# For a billing run with 500 customers at $25 each = $12,500 expected total:
# - Vault key cap: $13,750 (110% of expected total)
# - XAUTOCLAIM race: two consumers both process 50 customers simultaneously
#   (50 × $25 × 2 = $2,500 potential duplicate in worst-case same-second race)
# - Idempotency key deduplicates charges within 24h of original attempt
# - Pre-flight check skips charges where billing_records was already written
# - Vault cap blocks charges past $13,750 (covers expired-idempotency edge case)

Safe Redis Streams billing consumer with per-event XACK and write-before-ack

import redis
import stripe
import psycopg2
import os
import json
import logging

logger = logging.getLogger(__name__)

r = redis.Redis.from_url(os.environ["REDIS_URL"])
STREAM = "billing-stream"
GROUP = "billing-group"
CONSUMER = f"worker-{os.getpid()}"

conn = psycopg2.connect(os.environ["DATABASE_URL"])

while True:
    # Read ONE message at a time to minimize XAUTOCLAIM blast radius.
    # COUNT=1 ensures that if XAUTOCLAIM reclaims this PEL entry while
    # we're in stripe.charges.create(), only one billing event is affected —
    # not a batch of 10 where events 1-9 were billed and only event 10 failed.
    results = r.xreadgroup(GROUP, CONSUMER, {STREAM: ">"}, count=1, block=5000)

    if not results:
        continue

    for stream_name, messages in results:
        for msg_id, fields in messages:
            try:
                event = json.loads(fields[b"payload"])
                customer_id = event["customer_id"]
                billing_period = event["billing_period"]

                result = process_billing_event(conn, event)

                if result.get("skipped"):
                    # Pre-flight check found existing billing record.
                    # XACK removes from PEL — no further redelivery.
                    r.xack(STREAM, GROUP, msg_id)
                    logger.info(
                        "Skipped %s/%s: already_billed (%s)",
                        customer_id,
                        billing_period,
                        result.get("charge_id"),
                    )
                    continue

                # billing_records write completed inside process_billing_event.
                # XACK AFTER the database write (write-before-ack pattern).
                # If this XACK is lost to AOF data loss (appendfsync everysec),
                # the re-delivered message hits the pre-flight check above —
                # billing_records row exists (PostgreSQL survived the Redis failover)
                # and the Stripe call is skipped.
                r.xack(STREAM, GROUP, msg_id)
                logger.info(
                    "Charged %s/%s: %s",
                    customer_id,
                    billing_period,
                    result.get("charge_id"),
                )

            except stripe.error.Timeout:
                # Stripe timed out. ch_A may already exist on Stripe's servers.
                # Do NOT call XACK — keep the message in PEL for redelivery.
                # On redelivery (XAUTOCLAIM or GROUP re-read from PEL):
                # - content-hash idempotency key returns ch_A within 24h
                # - pre-flight check skips if billing_records was written before timeout
                # Do NOT XACK after a Stripe timeout. Let XAUTOCLAIM handle redelivery.
                logger.warning(
                    "Stripe timeout for %s/%s — keeping in PEL for redelivery",
                    event.get("customer_id"),
                    event.get("billing_period"),
                )

            except stripe.error.CardError as e:
                # Card declined — permanent failure. Write to billing_records
                # with status='card_declined', then XACK to remove from PEL.
                # Do NOT leave in PEL: XAUTOCLAIM would re-deliver indefinitely,
                # creating repeated Stripe card-decline events visible to the customer.
                try:
                    with conn.cursor() as cur:
                        cur.execute(
                            """
                            INSERT INTO billing_records
                                (customer_id, billing_period, charge_id,
                                 amount_cents, status, error, created_at)
                            VALUES (%s, %s, NULL, %s, 'card_declined', %s, NOW())
                            ON CONFLICT (customer_id, billing_period) DO NOTHING
                            """,
                            (
                                event["customer_id"],
                                event["billing_period"],
                                event["amount_cents"],
                                str(e),
                            ),
                        )
                        conn.commit()
                finally:
                    r.xack(STREAM, GROUP, msg_id)

            except Exception as e:
                logger.error(
                    "Billing failed for %s: %s — keeping in PEL",
                    event.get("customer_id"),
                    e,
                )
                # For unknown errors, keep in PEL for XAUTOCLAIM redelivery.
                # The idempotency key and pre-flight check make the next
                # delivery safe regardless of what the current delivery did.

Comparison: protection levels across all three failure modes

FAQ

Does increasing min-idle-time in XAUTOCLAIM prevent duplicate Stripe charges from reclaimed messages?

It reduces the probability for short Stripe calls but does not eliminate the failure mode. If min-idle-time is set to 300,000 ms (5 minutes), a Stripe call that takes 90 seconds (stripe-python P99 under load) finishes before the threshold — assuming the consumer is still alive and the XAUTOCLAIM job runs exactly at the threshold boundary. But a consumer that is suspended by the OS scheduler (GC pause, CPU throttle on a shared Kubernetes node, cold-start latency in a serverless environment) can have its processing window pause while idle time continues to tick. A consumer with a 50ms GC pause at T=59,950 can miss the 60-second threshold by only 50 ms and be reclaimed. The only robust defense is the content-hash idempotency key and pre-flight database check — they make every XAUTOCLAIM redelivery safe regardless of how long the original consumer processed the message.

Can I use XREADGROUP’s NOACK flag to prevent PEL accumulation and avoid XAUTOCLAIM races?

NOACK mode (XREADGROUP GROUP g c STREAMS s > NOACK) delivers messages without adding them to the PEL, which means no XAUTOCLAIM redelivery is possible — eliminating failure mode 1 and the batch partial-failure aspect of failure mode 2. But NOACK is a fire-and-forget delivery: if the consumer crashes after receiving the message but before stripe.charges.create() returns, the message is permanently lost. There is no way to recover it from the stream (it was never added to the PEL, so there is nothing to reclaim). For billing pipelines, NOACK trades the risk of double-charging for the risk of missed billing — a different but equally dangerous failure mode. NOACK is appropriate for analytics events where loss is acceptable; it is not appropriate for Stripe billing calls where every usage event must be billed exactly once.

How does appendfsync always prevent the AOF data-loss failure mode, and why do most deployments not use it?

appendfsync always calls fsync() after every write command, ensuring that every XACK is durable on disk before the command returns to the client. With appendfsync always, a Redis primary crash cannot cause a completed XACK to be missing from the replica’s AOF. The replica that takes over as primary has the XACK in its AOF and the billing message is no longer in the PEL — no re-delivery occurs. Most deployments avoid appendfsync always because it reduces Redis write throughput by 10–100× depending on disk I/O speed — fsync() is a blocking syscall and Redis is single-threaded for command execution. A managed Redis instance on AWS ElastiCache processing 100,000 writes/second with appendfsync always may process fewer than 10,000 writes/second, making the setting impractical for high-throughput billing pipelines. The content-hash idempotency key and pre-flight database check close the AOF data-loss failure mode without requiring appendfsync always, because the pre-flight check queries PostgreSQL — which is durable through Redis failover — rather than Redis’s own state.

What happens if the Redis primary crashes between billing_records write and XACK — is the customer billed twice?

No, if the pre-flight check is in place. The write-before-ack pattern ensures billing_records is written to PostgreSQL (durable) before XACK is sent to Redis (potentially lossy). If the Redis primary crashes in that window, the failover replica restores the message to the PEL, and the next consumer to receive it runs the pre-flight check against PostgreSQL. PostgreSQL has the billing_records row from the first consumer’s write (PostgreSQL WAL is fsynced on commit by default — unlike Redis’s appendfsync everysec). The pre-flight check finds the existing row, skips stripe.charges.create(), and calls XACK to remove the message from the now-restored PEL. The idempotency key provides a second layer: even if the pre-flight check is not in place and the consumer calls stripe.charges.create(), the content-hash key returns ch_A from Stripe’s deduplication cache within 24 hours. The vault key spend cap provides a third layer for the edge case where both the DB write and the idempotency cache are unavailable.

Does Redis Cluster sharding change the failure modes for billing consumers?

Redis Cluster distributes stream keys across hash slots and shards. If a billing pipeline uses a single stream key (e.g., billing-stream), all messages are on one shard — a shard failover affects all pending messages in the PEL simultaneously. If the pipeline uses per-customer or per-region stream keys (e.g., billing-stream:{customer_id}), different customers’ streams are on different shards; a shard failover affects only the customers on that shard, but the failure mode (AOF data loss, XAUTOCLAIM reclaim) applies to each shard independently. The content-hash idempotency key is shard-agnostic: it is derived from business fields, not from the shard ID, slot number, or stream key name. The pre-flight database check queries PostgreSQL, which is not distributed across Redis shards. Both defenses work identically in single-node, Sentinel, and Cluster deployments. The only Cluster-specific consideration is connection handling: redis-py’s RedisCluster client routes commands to the correct shard automatically, but the consumer must handle MovedError and AskError during cluster rebalancing events, which can temporarily interrupt XACK calls and leave messages in the PEL longer than expected.