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

By Keybrake · July 18, 2026

Apache Pulsar is a distributed pub-sub messaging platform with persistent message storage, multi-tenant topic management, and flexible subscription models — Exclusive, Shared, Failover, and Key_Shared. Teams wire Pulsar into Stripe billing pipelines as a usage event stream: producers publish usage events to a Pulsar topic, and a billing consumer reads the stream, aggregates per-customer usage, and calls stripe.charges.create() for each. Three Pulsar-specific behaviors introduce Stripe double-charge failure modes that Pulsar’s admin console, topic stats, and subscription cursor metrics do not surface as billing risk.

This post covers those three failure modes with Pulsar consumer configuration, 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 changing how your Pulsar topics or subscriptions are structured.

Failure mode 1: cumulative acknowledgment mode — consumer restart redelivers the entire unacked batch after Stripe charges were already created

Pulsar consumers support two acknowledgment modes: individual ack (each message is acknowledged separately as it is processed) and cumulative ack (acknowledging message N implicitly acknowledges all messages with ID ≤ N). Cumulative ack is appealing for high-throughput billing consumers because it reduces the number of ack RPCs to the Pulsar broker — one ack per batch instead of one per message. The billing function reads a batch of 500 usage event messages, processes all of them (calling stripe.charges.create() for each customer), then sends a single cumulative ack for message 500 to advance the subscription cursor.

The problem is acknowledgment timing relative to processing completion. If the billing consumer crashes, is evicted, or loses its connection to the broker after processing the 500 messages but before sending the cumulative ack, the Pulsar broker treats all 500 messages as unacknowledged. The subscription cursor does not advance. When the consumer restarts — or when the Pulsar subscription’s ack timeout fires — the broker redelivers all 500 messages. The billing function receives them as new messages. Without an idempotency key, stripe.charges.create() fires again for all 500 customers, creating 500 new charges (ch_B through ch_BZZZ) alongside the 500 charges already created in the first run (ch_A through ch_AZZZ).

import pulsar, stripe, os

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

client = pulsar.Client(PULSAR_SERVICE_URL)

consumer = client.subscribe(
    "persistent://billing/default/usage-events",
    subscription_name="billing-consumer",
    consumer_type=pulsar.ConsumerType.Exclusive,
    # Cumulative ack: send one ack for the whole batch at the end.
    # If the consumer crashes between the last stripe.charges.create() call
    # and consumer.acknowledge_cumulative(last_msg), the broker redelivers
    # all 500 messages to the next consumer instance.
)

# UNSAFE: processes all messages in the batch, then acks — crash window exists
def run_billing_batch_unsafe():
    stripe.api_key = STRIPE_SECRET_KEY  # UNSAFE: unrestricted key, no spend cap

    messages = []
    last_msg = None
    while len(messages) < 500:
        msg = consumer.receive(timeout_millis=5000)
        messages.append(msg)
        last_msg = msg

    for msg in messages:
        data = msg.value()
        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 — redelivered message creates ch_B for same customer
        )
        # crash here → broker redelivers entire batch → 500 duplicate charges

    # Cumulative ack sent only after all charges complete.
    # Any crash between first stripe.Charge.create() and this line
    # leaves all 500 messages unacked. Broker redelivers on reconnect.
    consumer.acknowledge_cumulative(last_msg)

Pulsar’s admin console shows the subscription cursor at its pre-batch position during the processing window. After a crash-and-restart, it shows the cursor still at the same position — which is correct, because the ack was never sent. The topic’s backlog counter shows 500 messages available for delivery. Nothing in Pulsar’s metrics distinguishes “not yet processed” from “processed but not yet acked.” Stripe’s dashboard shows two charges per customer for the billing period, both with different Request-Id headers and different timestamps.

The ack timeout compounds the risk. Pulsar subscriptions have a configurable ackTimeout (default: no timeout in most deployments, but often set to 30–60 seconds in billing pipelines to handle slow consumers). If the billing function takes longer than the ack timeout to process the full batch of 500 messages, Pulsar redelivers the earliest unacked messages even while the consumer is still running and still processing later messages. The consumer receives message 1 again while it is processing message 400. If the billing function does not check for already-billed customers before calling Stripe, message 1’s customer is charged twice within the same Lambda invocation.

Failure mode 2: Shared subscription rebalancing during consumer scale-down redistributes in-flight messages to surviving consumers that already processed them

Teams scale their Pulsar billing consumers horizontally using Shared subscriptions: multiple consumer instances connect to the same subscription, and Pulsar round-robin distributes messages across all connected consumers. In normal operation, each message goes to exactly one consumer. But Pulsar’s Shared subscription model provides at-least-once delivery, not exactly-once delivery. The distinction matters when consumers disconnect mid-batch.

The failure mode: three billing consumer pods are running a Shared subscription. Pod A receives messages 1–166, Pod B receives 167–333, Pod C receives 334–500. All three pods are mid-processing — Pod A has charged customers 1–100 and is processing customers 101–166. The Kubernetes node running Pod A is evicted (spot instance reclaimed, OOM eviction, or rolling deploy). Pod A disconnects from the Pulsar broker before sending any acks for its messages (1–166). Pulsar detects the disconnection and marks messages 1–166 as unacknowledged. It redistributes them across the surviving consumers — Pod B and Pod C — interleaved with their own in-flight messages.

# Shared subscription: multiple consumer instances, round-robin dispatch
# Each consumer processes its own partition of the message stream

# Pod A — processes messages 1–166 (round-robin assignment)
consumer_a = client.subscribe(
    "persistent://billing/default/usage-events",
    subscription_name="billing-shared",
    consumer_type=pulsar.ConsumerType.Shared,  # multiple consumers on same subscription
)

# Pod B — processes messages 167–333 (round-robin assignment)
consumer_b = client.subscribe(
    "persistent://billing/default/usage-events",
    subscription_name="billing-shared",
    consumer_type=pulsar.ConsumerType.Shared,
)

# When Pod A is evicted mid-batch (after processing msgs 1–100, before acking any):
# - Pulsar marks msgs 1–166 as unacknowledged (Pod A disconnected without acking)
# - Pulsar redistributes msgs 1–166 to Pod B and Pod C
# - Pod B now receives: msgs 167–333 (its own) + msgs 1–100 (redelivered from Pod A)
# - Pod B has no knowledge that msgs 1–100 were already processed by Pod A
# - Pod B calls stripe.charges.create() for customers 1–100 again
# - Stripe creates ch_B_001 through ch_B_100 alongside ch_A_001 through ch_A_100
# - Total: 200 charges for 100 customers. Pod B's consumer logs show 100 new successful charges.
# - Pulsar admin console shows 0 backlog after both runs complete: all messages acked.
# - Nothing flags the duplicate charges — Pulsar considers the redelivery correct behavior.

Pulsar’s Key_Shared subscription type reduces (but does not eliminate) this risk: messages with the same key (e.g., customer_id) are always routed to the same consumer, so a customer’s usage events are never split across consumers. But Key_Shared still redelivers unacked messages when a consumer disconnects — if a consumer that owns key range A–M disconnects mid-batch, its unacked messages are redistributed to a surviving consumer. If the surviving consumer already processed some of those messages in an earlier overlapping batch, the duplicates arrive without any distinguishing marker.

The scale-down scenario is common: Kubernetes horizontal pod autoscalers scale billing consumers down when the topic backlog drops (billing run nearing completion), pod disruption budgets allow evictions during rolling deployments, and spot instance interruptions are unpredictable. Any of these causes Pod A to disconnect mid-batch at billing run completion — exactly the moment when most messages have been processed but the subscription cursor has not advanced.

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

Pulsar’s retry mechanism works in layers: a consumer can negative-acknowledge (nack) a message, causing Pulsar to redeliver it after a configurable delay (using the retry topic); after maxRedeliverCount retries, Pulsar routes the message to a dead letter topic (DLT). Teams wire a DLT consumer to re-process failed billing events with additional logging or manual review. The billing function’s error handling treats both the retry topic and the DLT as “try billing again”, which is correct for genuinely failed Stripe calls — but incorrect for Stripe timeouts where the charge was created server-side before the timeout fired.

The failure scenario: the billing function calls stripe.charges.create() for customer cus_001 with amount 4,900 cents. The Stripe API receives the request, validates the payment method, creates the charge (ch_A), and begins writing the response. The network connection between the billing Lambda and Stripe drops during the response transmission. The billing function receives a stripe.error.Timeout (or stripe.error.APIConnectionError). The billing function catches the exception and nacks the Pulsar message — signaling to Pulsar that processing failed and the message should be retried.

import pulsar, stripe, os

consumer = client.subscribe(
    "persistent://billing/default/usage-events",
    subscription_name="billing-consumer",
    consumer_type=pulsar.ConsumerType.Exclusive,
    dead_letter_policy=pulsar.ConsumerDeadLetterPolicy(
        max_redeliver_count=3,
        dead_letter_topic="persistent://billing/default/usage-events-DLT",
    ),
    negative_ack_redelivery_delay_ms=30_000,  # 30 seconds between retries
)

# UNSAFE: nacking on Stripe timeout causes retry even if charge was created server-side
def process_message_unsafe(msg):
    data = msg.value()
    stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

    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 even if ch_A was created server-side
        )
        consumer.acknowledge(msg)  # ack on success
    except stripe.error.Timeout:
        # DANGER: the charge may have been created server-side before the timeout.
        # Nacking causes Pulsar to redeliver the message after 30 seconds.
        # Retry calls stripe.charges.create() again → ch_B for the same customer.
        consumer.negative_acknowledge(msg)
    except stripe.error.RateLimitError:
        # Also DANGER: nacking on rate limit retries the charge.
        # Stripe's rate limit response confirms the charge was NOT created,
        # but if the billing function nacks all stripe exceptions without checking,
        # a timeout that returns a 429-like error triggers the same retry loop.
        consumer.negative_acknowledge(msg)

# After 3 nacks, Pulsar routes the message to the DLT.
# The DLT consumer picks it up and calls process_message_unsafe() again.
# If ch_A was created on the first attempt and ch_B on the first retry,
# the DLT consumer creates ch_C — three charges for one customer, one billing period.
# Pulsar's DLT consumer log shows 1 message processed from DLT: normal behavior.
# Stripe's dashboard shows 3 charges for cus_001 in the billing period.

The Stripe timeout behavior is the core of this failure mode. Stripe’s API documentation distinguishes between client-side timeouts (where the request never reached Stripe) and server-side timeouts (where Stripe processed the request and the response was lost in transit). For stripe.error.Timeout, Stripe cannot guarantee which occurred — the client timed out waiting for a response, but the charge may or may not have been created. Retrying without an idempotency key always risks a duplicate charge when the original request succeeded server-side.

The retry-topic delay (30 seconds in the example) makes this particularly hard to debug: there are 30 seconds between the first nack and the first retry, during which Stripe’s system has a charge (ch_A) for cus_001. The billing function’s logs show a Timeout exception followed by a successful charge (ch_B) 30 seconds later. The Pulsar subscription metrics show 1 message redelivered. Nothing in the Pulsar admin console flags the double-charge — from Pulsar’s perspective, the first delivery failed (consumer nacked) and the retry succeeded (consumer acked). Correct behavior by the messaging layer; incorrect outcome at the billing layer.

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

All three failure modes share the same root cause: the billing function fires multiple times for the same customer in the same billing period, and nothing at the Stripe layer or the Pulsar layer prevents the subsequent calls from creating duplicate charges. The fix has two independent layers that close all three modes without changing Pulsar subscription type, ack mode, or retry configuration.

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

The Stripe idempotency key must be derived from the business event’s stable fields — fields that do not change across Pulsar message redeliveries, cumulative ack window resets, or DLT re-processing. The key must be scoped to (customer_id, billing_period) and must not include any Pulsar message metadata that changes across deliveries: message sequence ID, publish timestamp, redelivery count, or consumer name.

import hashlib, stripe, os, psycopg2, pulsar

def make_idempotency_key(customer_id: str, billing_period: str) -> str:
    # Derived from stable business identity: customer + billing period only.
    # Must NOT include: Pulsar message sequence ID, publish timestamp,
    # redelivery count, DLT attempt number, consumer pod ID, or
    # retry topic suffix. All of these change across Pulsar redeliveries.
    raw = f"{customer_id}:{billing_period}:pulsar-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def process_usage_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"]

    cur = conn.cursor()

    # Pre-flight: skip if already billed this period.
    # Survives cumulative ack redeliveries, Shared subscription redistributions,
    # and DLT re-processing beyond Stripe's 24h idempotency window.
    cur.execute(
        "SELECT 1 FROM billing_records WHERE customer_id = %s AND billing_period = %s",
        (customer_id, billing_period)
    )
    if cur.fetchone():
        return {"status": "skipped", "customer_id": customer_id}

    idempotency_key = make_idempotency_key(customer_id, billing_period)

    try:
        response = stripe.Charge.create(
            amount=amount_cents,
            currency="usd",
            customer=stripe_customer_id,
            description=f"Usage billing for {billing_period}",
            idempotency_key=idempotency_key,
        )
    except stripe.error.Timeout:
        # With an idempotency key set, retrying is safe:
        # if ch_A was created server-side, Stripe returns ch_A on retry (not ch_B).
        # Do NOT nack — raise and let the caller decide whether to ack or nack.
        # The billing function should check billing_records before retrying
        # to confirm whether the charge was committed on the first attempt.
        raise

    cur.execute(
        "INSERT INTO billing_records (customer_id, billing_period, charge_id, billed_at) "
        "VALUES (%s, %s, %s, NOW()) ON CONFLICT (customer_id, billing_period) DO NOTHING",
        (customer_id, billing_period, response["id"])
    )
    conn.commit()
    return {"status": "charged", "charge_id": response["id"]}

The idempotency key closes failure mode 1 (cumulative ack redelivery) and failure mode 3 (DLT retry on timeout) within Stripe’s 24-hour idempotency window. For failure mode 2 (Shared subscription redistribution), the pre-flight check provides defense-in-depth beyond the 24-hour window and eliminates unnecessary Stripe API calls for customers already committed to billing_records. The idempotency key closes the race window between the pre-flight check and the Stripe call for concurrent redeliveries within the window.

Layer 2: per-billing-period vault key capped at expected total × 1.10

Issue a vault key before the billing run capped at 110% of the expected total charges for the billing period. The vault key replaces the raw Stripe restricted key as the stripe.api_key for the billing function. Any run that exceeds the cap — because of a Shared subscription redistribution, a cumulative ack redelivery that processes customers twice, or a DLT retry loop that fires duplicate charges before the idempotency key closes the window — receives a 402 spend_cap_exceeded response from the proxy after the cap is hit.

import requests, os

PROXY_BASE = "https://proxy.keybrake.com"
PROXY_KEY = os.environ["KEYBRAKE_API_KEY"]

def issue_billing_vault_key(billing_period: str, expected_total_usd: float) -> str:
    cap = round(expected_total_usd * 1.10, 2)
    resp = requests.post(
        f"{PROXY_BASE}/vault/keys",
        headers={"Authorization": f"Bearer {PROXY_KEY}"},
        json={
            "label": f"pulsar-billing-{billing_period}",
            "vendor": "stripe",
            "daily_usd_cap": cap,
            "allowed_endpoints": ["/v1/charges"],
            "expires_at": f"{billing_period}-31T23:59:59Z",
        },
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

# Pulsar consumer with vault key and safe timeout handling:
def run_billing_consumer(billing_period: str, expected_total_usd: float):
    vault_key = issue_billing_vault_key(billing_period, expected_total_usd)
    stripe.api_key = vault_key  # vault key, not raw Stripe key

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

    consumer = client.subscribe(
        "persistent://billing/default/usage-events",
        subscription_name="billing-consumer",
        consumer_type=pulsar.ConsumerType.Exclusive,
        dead_letter_policy=pulsar.ConsumerDeadLetterPolicy(
            max_redeliver_count=3,
            dead_letter_topic="persistent://billing/default/usage-events-DLT",
        ),
    )

    while True:
        try:
            msg = consumer.receive(timeout_millis=10_000)
        except pulsar.Timeout:
            break  # no more messages — billing run complete

        data = msg.value()
        try:
            result = process_usage_message(conn, data)
            consumer.acknowledge(msg)  # ack regardless of skipped or charged
        except stripe.error.Timeout:
            # Idempotency key is set — safe to nack and let Pulsar retry.
            # The retry will either: (a) succeed with the same ch_A if charge was
            # created server-side (idempotency key returns existing charge), or
            # (b) create ch_A fresh if the original request was never processed.
            # Pre-flight check also protects: if ch_A was committed before the
            # timeout response arrived, the retry's pre-flight skips the Stripe call.
            consumer.negative_acknowledge(msg)
        except Exception as e:
            consumer.negative_acknowledge(msg)
            raise

Governance gaps specific to Apache Pulsar

Four Pulsar-specific behaviors reduce the effectiveness of billing governance patterns designed for other queue and streaming tools:

Pulsar’s subscription cursor tracks delivery position, not processing completion. The cursor advances only when a message is acknowledged. Between the first stripe.charges.create() call and the cumulative ack at the end of a batch, all processed messages appear as unacknowledged in Pulsar’s metrics. The billing database — not the Pulsar subscription cursor — is the authoritative source of truth for which customers have been charged. The pre-flight check reads from billing_records, not from Pulsar cursor position.
Pulsar’s Shared subscription provides at-least-once delivery, not exactly-once delivery. The exactly-once guarantee requires the application layer to implement idempotency. Pulsar does not deduplicate messages within a Shared subscription’s redelivery window based on business key. Two consumers in the same subscription can both receive the same message if one consumer disconnects mid-batch. Pulsar’s admin console correctly reports both deliveries as successful — this is expected at-least-once behavior, not a Pulsar bug.
Pulsar’s dead letter topic mechanism treats all nacks as “retry” regardless of whether the business operation completed. The DLT is designed for messages that consistently fail processing. If the billing function nacks a message after a Stripe timeout without first confirming whether the charge was created, the DLT retry re-fires billing as if the original attempt never happened. The idempotency key is the only layer that makes this retry safe — the DLT consumer has no access to the original Stripe API response.
Pulsar’s message sequence ID and publish timestamp change on redelivery. The message_id field in a redelivered message is the original message ID — but the publish_time and event_time metadata may differ between the original delivery and the retry topic delivery. Any idempotency key that incorporates Pulsar message metadata (sequence ID, topic, partition, ledger ID, entry ID) will be different across deliveries from the retry topic or DLT, making Stripe’s 24-hour deduplication ineffective for those retries.

Comparison: protection by governance layer

Approach	Cumulative ack redelivery	Shared subscription redistribution	DLT retry on Stripe timeout	Cost per billing record	Audit trail
No protection (no idempotency key, no pre-flight)	None — redelivered batch creates ch_B through ch_BZZZ for all customers	None — redistributed messages create duplicate charges independently	None — DLT retry creates ch_B even if ch_A was created server-side	None	None
Stripe idempotency key only (content-hash, customer + billing_period)	Full within 24h — same key on redelivery, Stripe returns ch_A	Full within 24h — same key on redistribution, Stripe returns ch_A	Full within 24h — same key on DLT retry, Stripe returns ch_A if already created	One hash per record	None
Pre-flight DB check only (no idempotency key)	Full — pre-flight skips customers already in billing_records from first delivery	Partial — concurrent redistributed consumers have race window before first consumer commits billing_records row	Full — pre-flight skips customer if billing_records row was committed before timeout response	One DB read per record	None
Content-hash key + vault key + pre-flight check (recommended)	Full — pre-flight skips; idempotency key closes race window; vault cap stops excess spend from undetected redeliveries	Full — idempotency key closes concurrent redistribution race; pre-flight skips beyond 24h window; vault cap bounds any excess charges that slip through	Full — idempotency key makes DLT retry safe regardless of server-side charge status; pre-flight skips if billing_records was committed; vault cap catches runaway DLT retry loops	One DB read + one hash per record + one vault key per billing period	Proxy audit log + billing DB

Safe nack handling for Stripe errors

The DLT failure mode requires a specific change to how the billing function handles Stripe exceptions. The default pattern — nack on any exception — is unsafe for network and timeout errors when no idempotency key is set. With a content-hash idempotency key, nacking on timeout is safe. But the billing function should also distinguish between Stripe errors that indicate the charge was never attempted (rate limits, invalid request errors) and errors where the charge status is unknown (timeouts, connection errors):

import stripe, psycopg2

def safe_nack_on_stripe_error(e: stripe.error.StripeError, customer_id: str,
                               billing_period: str, conn) -> bool:
    """
    Returns True if the message should be nacked (retried by Pulsar).
    Returns False if the message should be acked (billing complete or skip).

    With a content-hash idempotency key, nacking on any Stripe error is safe
    for retry within 24h. The pre-flight check protects beyond 24h.
    Without an idempotency key, nacking on Timeout or APIConnectionError
    may create ch_B on retry if ch_A was created server-side.
    """
    if isinstance(e, (stripe.error.RateLimitError, stripe.error.APIError)):
        # Transient: charge was not created. Safe to nack and retry.
        return True
    if isinstance(e, (stripe.error.Timeout, stripe.error.APIConnectionError)):
        # Ambiguous: charge may or may not have been created server-side.
        # With idempotency key: safe to nack. Stripe deduplicates on retry.
        # Without idempotency key: DO NOT nack — query Stripe for existing charge first.
        # Here we assume idempotency key is set:
        return True
    if isinstance(e, stripe.error.CardError):
        # Card declined: charge was not created. Consider acking and recording failure
        # in billing_records rather than nacking (repeated retries won't un-decline a card).
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO billing_records (customer_id, billing_period, charge_id, billed_at, status) "
            "VALUES (%s, %s, NULL, NOW(), 'card_declined') "
            "ON CONFLICT (customer_id, billing_period) DO NOTHING",
            (customer_id, billing_period)
        )
        conn.commit()
        return False  # ack the message — billing attempt recorded, don't retry
    # Unknown error: nack and let Pulsar's DLT handle after maxRedeliverCount
    return True

Card declines are worth special handling: nacking a declined card causes Pulsar to retry delivery up to maxRedeliverCount times before routing to the DLT. Three retries of a declined card add latency and Stripe API calls without any chance of success. Acking the message after recording the decline in billing_records with status = 'card_declined' keeps the subscription cursor moving and avoids DLT accumulation of legitimately declined messages mixed with genuinely failed billing attempts.

pytest enforcement suite

import hashlib, pytest
from unittest.mock import patch, MagicMock

# 1. Idempotency key is stable across Pulsar redeliveries and DLT re-processing
def test_idempotency_key_excludes_pulsar_message_metadata():
    # Key must not include Pulsar sequence ID, publish time, or redelivery count
    key1 = make_idempotency_key("cus_001", "2026-07")
    key2 = make_idempotency_key("cus_001", "2026-07")  # same customer, same period
    assert key1 == key2

    raw = "cus_001:2026-07:pulsar-billing"
    expected = hashlib.sha256(raw.encode()).hexdigest()[:32]
    assert key1 == expected

# 2. Pre-flight skips Stripe call for customers already in billing_records
def test_preflight_skips_cumulative_ack_redelivery(mock_db, mock_stripe):
    mock_db.cursor.return_value.fetchone.return_value = (1,)  # billing_records row exists

    result = process_usage_message(
        conn=mock_db,
        msg_data={
            "customer_id": "cus_001",
            "stripe_customer_id": "cus_stripe_001",
            "amount_cents": 4900,
            "billing_period": "2026-07",
        },
    )

    assert result["status"] == "skipped"
    mock_stripe.Charge.create.assert_not_called()

# 3. DLT retry after Stripe timeout: idempotency key returns existing charge
def test_dlt_retry_returns_existing_charge_on_timeout(mock_db, mock_stripe):
    mock_db.cursor.return_value.fetchone.return_value = None  # not in billing_records
    mock_stripe.Charge.create.side_effect = [
        stripe.error.Timeout("timeout"),    # first attempt: timeout
        MagicMock(id="ch_A"),               # second attempt (DLT retry): existing charge
    ]

    with pytest.raises(stripe.error.Timeout):
        process_usage_message(mock_db, {
            "customer_id": "cus_001",
            "stripe_customer_id": "cus_stripe_001",
            "amount_cents": 4900,
            "billing_period": "2026-07",
        })

    # DLT retry with same idempotency key:
    result = process_usage_message(mock_db, {
        "customer_id": "cus_001",
        "stripe_customer_id": "cus_stripe_001",
        "amount_cents": 4900,
        "billing_period": "2026-07",
    })
    assert result["charge_id"] == "ch_A"
    # Stripe.Charge.create called twice but only one charge created (idempotency key)
    assert mock_stripe.Charge.create.call_count == 2

# 4. Shared subscription redistribution: concurrent pre-flight race resolved by idempotency key
def test_shared_subscription_redistribution_idempotency(mock_db, mock_stripe):
    # Simulate two concurrent consumers both calling process_usage_message for cus_001
    mock_db.cursor.return_value.fetchone.return_value = None  # neither sees billing_records yet
    mock_stripe.Charge.create.return_value = MagicMock(id="ch_A")

    result_a = process_usage_message(mock_db, {
        "customer_id": "cus_001",
        "stripe_customer_id": "cus_stripe_001",
        "amount_cents": 4900,
        "billing_period": "2026-07",
    })
    result_b = process_usage_message(mock_db, {
        "customer_id": "cus_001",
        "stripe_customer_id": "cus_stripe_001",
        "amount_cents": 4900,
        "billing_period": "2026-07",
    })

    # Both calls use the same idempotency key — Stripe returns ch_A for both
    assert result_a["charge_id"] == "ch_A"
    assert result_b["charge_id"] == "ch_A"
    # Stripe called twice with same idempotency key — only one actual charge
    assert mock_stripe.Charge.create.call_count == 2

# 5. Vault key cap blocks excess charges from runaway DLT retry loops
def test_vault_key_cap_blocks_dlt_runaway(mock_proxy):
    mock_proxy.post.side_effect = (
        [MagicMock(status_code=200, json=lambda: {"id": f"ch_{i}"})] * 1000
        + [MagicMock(status_code=402, json=lambda: {"error": "spend_cap_exceeded"})] * 500
    )

    run_a = [make_charge_via_proxy(f"cus_{i}", 4900, "2026-07", mock_proxy) for i in range(1000)]
    assert all(r["status_code"] == 200 for r in run_a)

    # DLT runaway loop tries to re-bill all customers — vault cap blocks after expected total
    dlt_run = [make_charge_via_proxy(f"cus_{i}", 4900, "2026-07", mock_proxy) for i in range(500)]
    assert all(r["status_code"] == 402 for r in dlt_run)

FAQ

Q: Should I switch from Shared to Exclusive subscription to prevent redistribution failures?
A: Exclusive subscription prevents parallel consumer instances, which eliminates the redistribution failure mode. But it limits horizontal scaling of billing consumers — only one consumer can read the topic at a time. The trade-off depends on billing volume. For billing pipelines that process fewer than 50,000 customers per run, Exclusive subscription is the safer choice and eliminates failure mode 2 entirely. For pipelines that require horizontal scaling, Key_Shared subscription with a customer_id routing key reduces (but does not eliminate) the redistribution risk, and the content-hash idempotency key + pre-flight check covers the residual window.

Q: How do I detect charges from a cumulative ack redelivery that already went through?
A: Query Stripe for all charges in the billing period and join to your billing_records table. Any charge_id in Stripe that has no corresponding row in billing_records is a charge made during a batch that was redelivered before the ack was sent. These are not duplicates yet — they represent the first set of charges, and the redelivered batch is the source of potential duplicates. Insert the orphaned charges into billing_records first, then allow the redelivery to process: the pre-flight check will skip every customer whose charge is now in billing_records.

Q: Can I use Pulsar’s built-in message deduplication to prevent billing duplicates?
A: Pulsar’s broker-level deduplication (enabled with brokerDeduplicationEnabled=true on the namespace) prevents duplicate message delivery for messages with the same producer-assigned sequence ID within a configurable window. This covers producer-side duplicates (a producer retrying a publish). It does not cover consumer-side duplicates from redelivery — Pulsar’s deduplication operates at the broker’s receive layer, not the consumer’s ack layer. A message redelivered due to a consumer nack or cumulative ack timeout has the same message ID but is not deduplicated by broker-level deduplication; it is considered a legitimate redelivery.

Q: How should the billing function handle the Pulsar ack timeout to avoid the cumulative ack failure mode?
A: Two patterns work. First, individual ack mode: acknowledge each message immediately after its billing_records row is committed (not after the Stripe call), using the pre-flight check to skip already-committed customers on any redelivery. Second, reduce batch size to fit within the ack timeout: if the ack timeout is 30 seconds and billing takes 0.1 seconds per customer, a batch of 250 customers will complete in 25 seconds, safely within the ack timeout. The content-hash idempotency key provides defense-in-depth regardless of which ack strategy you choose.

Q: How does the vault key cap behave when the billing run legitimately exceeds 110% of expected total?
A: Issue a second vault key for the remaining amount before the cap is exhausted. The proxy’s audit log records which customers were charged under each vault key, giving you a complete per-run spend breakdown. Set an alert on vault key utilization reaching 90% so the billing function receives a warning before the cap blocks legitimate charges. For billing periods with high variance (new customer cohorts, plan upgrades), query the expected total from the billing database immediately before issuing the vault key rather than using the previous period’s total as the baseline.

Put the brakes on your Pulsar billing pipeline

Keybrake issues per-billing-period vault keys your Pulsar billing consumer can use against Stripe — with spend caps that stop cumulative ack redeliveries, Shared subscription redistributions, and DLT retry loops before they double-charge your customer base, plus a per-call audit log that shows every charge decision your consumer made. Drop-in replacement for the Stripe SDK base URL.