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

RabbitMQ is a widely-deployed AMQP message broker that decouples usage event producers from billing consumers using exchanges, bindings, and queues with configurable delivery semantics. Teams wire RabbitMQ into Stripe billing pipelines as an event queue: a producer publishes a usage event message to a RabbitMQ exchange, a billing consumer reads from the bound queue, and calls stripe.charges.create() per customer. Three RabbitMQ-specific behaviors introduce Stripe double-charge failure modes that RabbitMQ’s management console, queue depth metrics, and unacknowledged message counters do not surface as billing risk.

This post covers those three failure modes with pika Python consumer code, content-hash idempotency keys, per-run vault keys via a spend-cap proxy, and pre-flight database checks — the two-layer governance pattern that eliminates all three without restructuring your RabbitMQ topology or changing exchange bindings.

Failure mode 1: prefetch window requeue on channel close — the next consumer re-charges customers already billed

RabbitMQ consumers use the basic_qos prefetch setting to control how many unacknowledged messages a consumer can hold at a time. With prefetch_count=100, RabbitMQ delivers up to 100 messages to the consumer before waiting for acks. This is the standard configuration for high-throughput billing consumers: the consumer reads a batch of 100 usage event messages, calls stripe.charges.create() for each customer, then sends basic_ack for each processed message (or a batch ack via multiple=True).

The failure window: the consumer has received all 100 messages from the prefetch buffer and is processing them sequentially — it has called stripe.charges.create() for customers 1 through 70. At this point, the consumer channel (or the underlying AMQP connection) closes. This happens during a Kubernetes pod eviction, a spot instance interruption, an OOM kill, a rolling deployment restart, or a simple network partition. RabbitMQ receives the channel close event and immediately requeues all 100 unacknowledged messages — including the 70 messages for customers already charged.

import pika, stripe, os, json

AMQP_URL = os.environ["AMQP_URL"]
STRIPE_SECRET_KEY = os.environ["STRIPE_SECRET_KEY"]

def run_billing_consumer_unsafe():
    connection = pika.BlockingConnection(pika.URLParameters(AMQP_URL))
    channel = connection.channel()

    # Prefetch 100 messages: RabbitMQ holds 100 unacked messages in this consumer's buffer.
    # All 100 are requeued if the channel closes before basic_ack is sent.
    channel.basic_qos(prefetch_count=100)

    stripe.api_key = STRIPE_SECRET_KEY  # UNSAFE: unrestricted key, no spend cap

    def on_message(ch, method, properties, body):
        data = json.loads(body)
        stripe.Charge.create(
            amount=data["amount_cents"],
            currency="usd",
            customer=data["stripe_customer_id"],
            description=f"Usage billing for {data['billing_period']}",
            # no idempotency_key — requeued message creates ch_B for same customer
        )
        # If channel closes between stripe.Charge.create() and this ack,
        # RabbitMQ requeues this message. The next consumer creates ch_B.
        ch.basic_ack(delivery_tag=method.delivery_tag)

    channel.basic_consume(
        queue="billing.usage-events",
        on_message_callback=on_message,
        auto_ack=False,  # manual ack required — but the window still exists
    )
    channel.start_consuming()

What RabbitMQ’s management console shows: after the channel close, the queue’s “Unacked” counter drops from 100 to 0, and the queue depth shows 100 messages ready for delivery. The management console records a consumer cancellation event. A second consumer (or the restarted first consumer) connects, receives all 100 messages as new deliveries, and processes them without any indication from RabbitMQ that these messages were previously consumed. method.redelivered is set to True on the redelivered messages, but in most consumer implementations this flag is not checked before calling stripe.charges.create().

The batch ack pattern (basic_ack(delivery_tag=last_tag, multiple=True)) compounds this. Instead of acking each message immediately after processing, the billing consumer accumulates all 100 messages and sends a single ack for all of them at the end. If the channel closes between the first stripe.charges.create() call and the final batch ack, all 100 messages are requeued — even though 99 of them were already charged. The batch ack pattern reduces AMQP traffic and appears cleaner in management console metrics, but it creates a larger requeue window than per-message acking.

The prefetch window also creates an intra-run duplicate risk under high load. With prefetch_count=100 and no per-message deduplication, if the same customer’s usage event is published to the queue twice (publisher retry, double-publish from an upstream pipeline), both messages land in the same prefetch batch and both are processed in the same billing run. RabbitMQ delivers both messages to the same consumer within the same prefetch window. The consumer calls stripe.charges.create() for the customer twice, seconds apart — without an idempotency key, Stripe creates ch_A and ch_B. The duplicate source (upstream pipeline, publisher retry) is invisible to the billing consumer.

Failure mode 2: dead letter exchange retry loop re-fires billing after a Stripe timeout where the charge was created server-side

RabbitMQ’s dead letter mechanism routes messages that are rejected (via basic_nack or basic_reject with requeue=False) or that expire (via per-message or per-queue TTL) to a dead letter exchange (DLX). Teams implement billing retries by chaining DLX queues: the billing queue’s DLX routes rejected messages to a retry queue with a TTL (e.g., 60 seconds); when the TTL fires, the retry queue’s own DLX routes the message back to the original billing exchange. The billing function processes the message again. This retry loop is correct for genuinely failed Stripe calls — but incorrect for stripe.error.Timeout where the charge was created server-side before the timeout response was lost.

import pika, stripe, os, json

def setup_billing_topology(channel):
    # Retry queue: messages wait here for 60s before being routed back to billing exchange
    channel.queue_declare(
        queue="billing.retry-60s",
        arguments={
            "x-message-ttl": 60_000,             # 60 second TTL
            "x-dead-letter-exchange": "billing",  # route back to billing exchange on expiry
        }
    )
    # Dead letter queue for permanent failures (after max retries)
    channel.queue_declare(queue="billing.dead")
    # Billing exchange DLX: rejected messages go to retry queue
    channel.exchange_declare(exchange="billing", exchange_type="direct")
    channel.queue_declare(
        queue="billing.usage-events",
        arguments={
            "x-dead-letter-exchange": "",         # default exchange
            "x-dead-letter-routing-key": "billing.retry-60s",
        }
    )

def on_message_unsafe(ch, method, properties, body):
    data = json.loads(body)
    stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

    # Track retry count from headers (RabbitMQ adds x-death header on each DLX routing)
    death_count = 0
    if properties.headers and "x-death" in properties.headers:
        death_count = sum(d.get("count", 0) for d in properties.headers["x-death"])

    if death_count >= 3:
        # Route to permanent dead letter queue after 3 retries
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        return

    try:
        stripe.Charge.create(
            amount=data["amount_cents"],
            currency="usd",
            customer=data["stripe_customer_id"],
            description=f"Usage billing for {data['billing_period']}",
            # no idempotency_key — retry creates ch_B if ch_A was created server-side
        )
        ch.basic_ack(delivery_tag=method.delivery_tag)

    except stripe.error.Timeout:
        # DANGER: the charge may have been created server-side before the timeout.
        # basic_nack(requeue=False) routes message to billing.retry-60s via DLX.
        # After 60s, the message routes back to the billing exchange.
        # The billing function calls stripe.Charge.create() again.
        # If ch_A was created on the first attempt, ch_B is now created 60s later.
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

    except stripe.error.CardError:
        # Card errors are permanent failures — correct to dead-letter without retry.
        # But nacking here with requeue=False also routes to DLX,
        # and if the DLX routing is misconfigured, it still hits the retry queue.
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

    except Exception:
        # Catch-all nack routes all unexpected exceptions to retry queue.
        # If any Stripe exception inherits from a broad base class and the billing
        # function uses bare except, all stripe errors including Timeout get retried.
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

The Stripe timeout ambiguity is the core of this failure mode. A stripe.error.Timeout occurs when the billing function’s HTTP client times out waiting for a Stripe API response. Stripe’s documentation distinguishes between two cases: the request was received and the charge was created (server-side success, response lost in transit), and the request was never fully received (client-side timeout before Stripe processed the request). For network-level timeouts, the billing function cannot distinguish which occurred — stripe.error.Timeout covers both. Retrying without an idempotency key always risks a duplicate charge when the original request succeeded.

The 60-second retry delay makes this failure mode hard to diagnose. RabbitMQ’s management console shows 1 message on the retry queue, then 1 message on the billing queue 60 seconds later. The billing function’s logs show a Timeout exception at T=0 and a successful charge (ch_B) at T=60. Stripe’s dashboard shows two charges for the customer in the billing period — ch_A (from T=0, created server-side before the timeout) and ch_B (from T=60, the retry). Nothing in RabbitMQ’s metrics flags the duplicate: one message consumed (with a retry detour), one message acked. Correct behavior by the messaging layer; two charges for one customer at the billing layer.

Per-message TTL creates a distinct variant of this failure mode. If the billing event is published with a short expiry (expiration field in AMQP message properties — for example, expiration="30000" for 30 seconds), and the message expires in the billing queue before the consumer processes it (billing consumer slow, queue backed up), RabbitMQ routes the expired message to the DLX without the consumer ever calling basic_nack. The DLX routes it to the retry queue, which routes it back to the billing exchange. If the billing function is now available and processes it, this is correct behavior — but if the billing function was slow because it was in the middle of processing the same message (it took more than 30 seconds), the DLX delivers a second copy while the first copy is still being processed. Two concurrent stripe.charges.create() calls for the same customer in the same billing period — the idempotency key is the only thing that prevents two charges.

Failure mode 3: AMQP connection-level multiplexing — a TCP connection drop requeues all in-flight messages across all channels simultaneously

AMQP 0-9-1 (RabbitMQ’s native protocol) multiplexes multiple logical channels over a single TCP connection. A billing consumer that processes messages in parallel typically opens one connection with multiple channels: each channel has its own prefetch window, its own consumer tag, and its own set of unacknowledged messages. This is the standard pika pattern for concurrent billing processing — one connection, multiple channels, multiple concurrent billing goroutines or threads each consuming from the same queue.

The failure mode specific to connection-level multiplexing: when the TCP connection drops (not just one channel), RabbitMQ requeues all unacknowledged messages across all channels on that connection simultaneously. A billing consumer with four channels and prefetch_count=25 per channel has up to 100 messages in flight. When the TCP connection drops — from a network partition, a broker restart, a Kubernetes node drain, or a TLS handshake timeout on reconnect — all 100 messages are requeued at once. If stripe.charges.create() was already called for 80 of them, those 80 customers will be charged again when the next consumer picks them up.

import pika, stripe, os, json, threading

AMQP_URL = os.environ["AMQP_URL"]

# UNSAFE pattern: multiple channels on a single connection
# A TCP connection drop requeues all in-flight messages across ALL channels at once.
def run_parallel_billing_unsafe(num_channels=4, prefetch_per_channel=25):
    connection = pika.BlockingConnection(pika.URLParameters(AMQP_URL))

    # Four channels on the same TCP connection.
    # Total in-flight window: 4 channels * 25 prefetch = 100 unacked messages.
    channels = []
    for i in range(num_channels):
        ch = connection.channel()
        ch.basic_qos(prefetch_count=prefetch_per_channel)
        channels.append(ch)

    # Each channel runs in its own thread (pika BlockingConnection thread model)
    def consume_on_channel(ch, thread_id):
        def on_message(ch, method, properties, body):
            data = json.loads(body)
            stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
            stripe.Charge.create(
                amount=data["amount_cents"],
                currency="usd",
                customer=data["stripe_customer_id"],
                description=f"Usage billing for {data['billing_period']}",
                # no idempotency_key
            )
            # If TCP connection drops between stripe.Charge.create() and basic_ack:
            # - This channel's unacked messages are requeued
            # - ALL other channels' unacked messages are ALSO requeued
            # - A single TCP drop requeues up to 100 messages across all 4 channels
            # - Not 25 messages (one channel) but 100 messages (all channels)
            ch.basic_ack(delivery_tag=method.delivery_tag)

        ch.basic_consume(
            queue="billing.usage-events",
            on_message_callback=on_message,
        )
        ch.start_consuming()

    threads = [
        threading.Thread(target=consume_on_channel, args=(ch, i))
        for i, ch in enumerate(channels)
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    # If the TCP connection drops during billing:
    # - connection.close() fires a connection-level close event at the broker
    # - broker requeues all unacked messages across ALL four channels simultaneously
    # - Next consumer run receives all 100 messages as new deliveries
    # - Up to 100 duplicate stripe.Charge.create() calls without idempotency keys

The important distinction: a channel close (one channel errors, is closed, or is cancelled) requeues only that channel’s unacked messages — up to prefetch_count messages. A connection close requeues all channels’ unacked messages simultaneously — up to num_channels × prefetch_count messages. The billing risk from a connection drop is multiplicative relative to a single-channel consumer.

RabbitMQ’s management console shows the queue depth increase from 0 to 100 (or whatever the total prefetch window was) when the TCP connection drops. The “Unacked” counter drops to 0. A new consumer connects and sees 100 messages ready for delivery. The management console records a consumer disconnection event. Nothing distinguishes “never processed” from “already processed by the disconnected consumer.” The redelivered flag is set to True on all 100 messages — but unless the billing consumer checks this flag and queries the billing database before calling Stripe, the flag alone does not prevent duplicate charges.

The connection-level drop scenario is common in Kubernetes environments. A rolling deployment restarts all billing consumer pods: each pod closes its AMQP connection during preStop or on SIGTERM. If the billing consumer does not have a graceful shutdown handler that drains its prefetch buffer (acking or nacking all in-flight messages before closing the connection), the connection close requeues all in-flight messages. A deployment rolling through four pods with 25 messages each can requeue 100 messages simultaneously, with no guarantee that any of them are genuinely unprocessed.

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

All three failure modes share the same root cause: the billing function calls stripe.charges.create() multiple times for the same customer in the same billing period, and neither RabbitMQ nor Stripe’s API layer prevents the subsequent calls from creating duplicate charges. The fix has two independent layers that close all three modes without changing the RabbitMQ topology, exchange bindings, prefetch settings, or DLX configuration.

Layer 1: content-hash idempotency key stable across RabbitMQ redeliveries

The Stripe idempotency key must be derived from the business event’s stable fields — fields that do not change across RabbitMQ message redeliveries, DLX routing, or connection-level requeues. The key must be scoped to (customer_id, billing_period) and must not include any RabbitMQ message metadata that changes across deliveries: the delivery tag, the redelivery count (tracked via the x-death header), the message timestamp, or the routing key suffix added by DLX chains.

import hashlib, stripe, os, psycopg2, pika, json

def make_idempotency_key(customer_id: str, billing_period: str) -> str:
    # Derived from stable business identity: customer + billing period only.
    # Must NOT include: RabbitMQ delivery tag, x-death count, message timestamp,
    # routing key suffix from DLX chain, consumer tag, or connection ID.
    # All of these change across RabbitMQ redeliveries, DLX routing, and
    # connection-level requeues from TCP drops or pod evictions.
    raw = f"{customer_id}:{billing_period}:rabbitmq-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

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

    idempotency_key = make_idempotency_key(customer_id, billing_period)

    cur = conn.cursor()

    # Pre-flight check: was this customer already billed for this period?
    # This closes the gap for redeliveries older than Stripe's 24-hour idempotency window.
    # The billing_records table is the authoritative source — not the RabbitMQ queue state.
    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 acking the RabbitMQ message.
    # If the write succeeds and the ack fails (TCP drop between write and ack),
    # the message is requeued — but the pre-flight check above will skip it.
    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"}

The idempotency key construction excludes x-death count and routing key suffix deliberately. RabbitMQ’s DLX mechanism appends x-death headers on each routing through a dead letter exchange — a message that has been DLX-routed three times has an x-death header with count: 3. If the idempotency key includes the DLX count, each retry attempt generates a different key, defeating Stripe’s 24-hour deduplication. The key must be derived from the business event content only.

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 — typically set at 110% of the expected billing total. If the prefetch requeue triggers 100 duplicate charges and the billing function has already charged all customers once, the vault key’s cap prevents the second round of charges from executing: after expected_total × 1.10 has been charged through this vault key, the proxy rejects further requests with a 429.

# Vault key configuration via Keybrake API (before starting the billing run)
# Per-billing-period key capped at expected total * 1.10

import requests, os, 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"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=4)
            ).isoformat() + "Z",
        },
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

# Per-billing-period cap closes the runaway loop:
# - Expected total: $10,000 (1,000 customers at $10 each)
# - Vault key cap: $11,000 (10% buffer for legitimate variance)
# - After $11,000 charged: proxy rejects further Stripe calls with 429
# - A prefetch requeue that would re-charge 100 customers ($1,000) is blocked
#   if $10,000+ has already been charged through this vault key this period

Safe message handler: combining both layers

def make_safe_consumer(conn, vault_key: str):
    def on_message(ch, method, properties, body):
        data = json.loads(body)

        # Check redelivered flag first — fast path to pre-flight check
        if method.redelivered:
            result = process_billing_message(conn, data)
            if result.get("skipped"):
                # Already billed — ack the redelivered message to clear it from the queue
                ch.basic_ack(delivery_tag=method.delivery_tag)
                return

        try:
            result = process_billing_message(conn, data)
            ch.basic_ack(delivery_tag=method.delivery_tag)

        except stripe.error.Timeout:
            # Nack safely: idempotency key makes DLX retry safe regardless of
            # whether the charge was created server-side on the first attempt.
            # basic_nack(requeue=False) routes to DLX → retry queue → back to billing exchange.
            # The retry will hit the pre-flight check if the charge was already recorded,
            # or use the same idempotency key if it was not.
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

        except stripe.error.CardError as e:
            # Card decline: permanent failure, record it and ack the message.
            # Do NOT nack — nacking routes to DLX for retry, but card declines
            # are not transient. Acking + recording prevents DLT accumulation.
            cur = conn.cursor()
            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
                """,
                (data["customer_id"], data["billing_period"], data["amount_cents"], str(e)),
            )
            conn.commit()
            ch.basic_ack(delivery_tag=method.delivery_tag)

    return on_message

Comparison: protection levels across all three failure modes

FAQ

Should I use requeue=True or requeue=False in my nack handler?

basic_nack(requeue=False) routes the message to the DLX (if configured) — this is the correct choice for Stripe billing errors, because requeue=True puts the message back at the front of the queue immediately, which can create a tight requeue loop that hammers Stripe with rapid retries during a transient outage. Route to DLX with a TTL-based delay queue to get controlled retry spacing. The idempotency key makes DLX retries safe for Stripe timeouts. The pre-flight check makes them safe for retries beyond the 24-hour idempotency window.

Does the redelivered flag on the RabbitMQ message tell me if the customer was already charged?

No. method.redelivered = True tells you the message was previously delivered to a consumer that did not ack it — it does not tell you whether that consumer called stripe.charges.create() before failing. A consumer can fail before making any Stripe calls (redelivered, safe to process), or after completing all Stripe calls (redelivered, already charged). The only authoritative source is the billing_records table. The redelivered flag is a useful fast-path hint to trigger the pre-flight check, but it cannot be used as a substitute for the pre-flight check itself.

How does the idempotency key interact with Stripe’s 24-hour deduplication window?

Stripe’s idempotency keys deduplicate requests within 24 hours of the first successful request for that key. If the DLX retry or connection requeue happens within 24 hours of the original charge, the idempotency key returns the original charge response (ch_A) without creating a new charge. If the requeue happens after 24 hours (unusual for billing retries, but possible if a message is stuck in a DLX queue with a long TTL or the broker has a backlog), the idempotency key no longer deduplicates — and only the pre-flight billing_records check prevents a duplicate charge. This is why both layers are required: the idempotency key handles the common case, and the pre-flight check handles the long-tail case beyond the 24-hour window.

Can I use a quorum queue to avoid the prefetch requeue failure mode?

Quorum queues improve RabbitMQ’s durability and broker-side failure handling — they replicate queue state across a quorum of nodes so a broker node failure does not lose messages. But quorum queues do not change the consumer-side prefetch behavior: the consumer still holds unacked messages in its prefetch buffer, and a channel or connection close still requeues all of them. Quorum queues are the correct choice for production billing queues because of their durability guarantees — but they do not replace the idempotency key and pre-flight check at the consumer layer.

What is the safe order of operations for writing to billing_records and acking the RabbitMQ message?

Write to billing_records first, commit the database transaction, then ack the RabbitMQ message. If the database write succeeds but the ack fails (TCP drop between the database commit and the basic_ack), the message is requeued — but the pre-flight check in the next delivery will find the record in billing_records and skip the Stripe call. If the ack is sent before the database write, a crash between the ack and the write leaves no record: the message is consumed (and will not be redelivered), but there is no record in billing_records, and the next billing_records-based billing check will think this customer was not billed this period. Write-then-ack is the safe ordering.