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

NATS JetStream is a lightweight, cloud-native persistent messaging layer built on NATS Core, used in microservices and agent pipelines for at-least-once delivery with durable consumers, stream replay, and per-consumer acknowledgment semantics. When billing workers subscribe to a JetStream stream and call stripe.charges.create() per usage event, three NATS-specific behaviors introduce Stripe double-charge failure modes that NATS’s monitoring endpoints, consumer info metrics, and stream lag counters do not surface as billing risk.

This post covers those three failure modes with nats-py Python consumer code, content-hash idempotency keys stable across AckWait redeliveries and consumer restarts, 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 NATS subjects, stream configuration, or consumer topology.

Failure mode 1: JetStream AckWait expiry while the Stripe HTTP call is in-flight — the message becomes available for the next Fetch before the first charge returns

NATS JetStream consumers require explicit acknowledgment for each delivered message. Each consumer has an AckWait parameter — the maximum time the server will wait for an ack before considering the message unacknowledged and eligible for redelivery. The NATS default is 30 seconds. A billing consumer that receives a usage event message, calls stripe.charges.create(), and sends the ack after the Stripe call completes is safe — as long as the Stripe call completes within AckWait. When it does not, NATS redelivers the message before the ack arrives.

The failure scenario: a billing worker calls stripe.charges.create() at T=0. Stripe is under load; the API response takes 35 seconds. At T=30 (AckWait), NATS marks the in-flight message as unacknowledged and makes it available for redelivery. A second billing worker running msg = await sub.next_msg(timeout=1.0) in a pull consumer loop, or a second subscriber in a push consumer queue group, receives the same message. The second worker calls stripe.charges.create() at T=30 for the same customer and same billing period — without an idempotency key, Stripe creates ch_B. The first worker receives ch_A at T=35, sends its ack, and logs a successful charge. The duplicate — ch_B created at T=30 by the second worker — does not appear in either worker’s error log.

import asyncio, nats, stripe, os, json
from nats.js.api import ConsumerConfig, AckPolicy, DeliverPolicy

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

async def billing_consumer_ackwait_unsafe():
    nc = await nats.connect(NATS_URL)
    js = nc.jetstream()

    # Pull consumer with default AckWait.
    # If stripe.charges.create() takes longer than AckWait seconds,
    # NATS makes this message available for the next Fetch() call —
    # the second worker that fetches it calls stripe.charges.create() again.
    psub = await js.pull_subscribe(
        "billing.usage.events",
        durable="billing-worker",
        config=ConsumerConfig(
            ack_policy=AckPolicy.EXPLICIT,
            ack_wait=30,             # 30s — Stripe P99 latency can exceed this
            max_deliver=3,           # 3 redelivery attempts before consumer gives up
        ),
    )

    while True:
        try:
            msgs = await psub.fetch(batch=10, timeout=1.0)
        except nats.errors.TimeoutError:
            continue

        for msg in msgs:
            data = json.loads(msg.data)
            stripe.api_key = STRIPE_SECRET_KEY   # unrestricted key, no spend cap

            # This call may take longer than ack_wait under Stripe API load.
            # When it does, NATS delivers this message to the next Fetch()
            # while this call is still pending — duplicate charge incoming.
            charge = 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 — second worker creates ch_B
            )

            # Ack arrives at NATS after AckWait has already expired.
            # NATS accepts the late ack but the message was already redelivered.
            await msg.ack()

The AckWait failure mode is particularly insidious because the first worker’s Stripe call eventually succeeds. From the first worker’s perspective, the billing run completed normally — every customer was charged, every message was acked. From the second worker’s perspective, it received a valid usage event and charged the customer correctly. No exception was raised, no error was logged, and NATS’s consumer info shows 0 pending messages after both workers complete. Stripe’s dashboard shows two charges for the customer in the billing period, but that is only visible if you query Stripe directly for the customer’s charge list.

The 30-second default AckWait is not conservative for billing workloads. Stripe’s API has a 99th-percentile response time that regularly exceeds 10 seconds under load, and with timeout retries at the HTTP client layer, a single stripe.charges.create() call can block for 20–40 seconds before raising stripe.error.Timeout. A billing pipeline that processes 1,000 customers and hits a Stripe latency spike after processing customer 400 will trigger AckWait expiry on all 600 remaining in-flight messages if those messages were fetched in the same batch. Teams that extend AckWait to 120 or 300 seconds reduce the risk but introduce a different problem: stuck messages that should be redelivered (because the worker crashed) are not redelivered for 2–5 minutes, stalling the billing run.

Failure mode 2: NATS Core direct subscription without a queue group — all connected billing workers receive every usage event simultaneously

NATS Core (the base messaging layer beneath JetStream) uses interest-based delivery: a message published to a subject is delivered to every subscriber currently connected to that subject. This is the correct behavior for fan-out patterns (one event, multiple consumers), but incorrect for billing pipelines where exactly one worker should process each usage event. NATS Core queue groups address this: a queue group delivers each message to exactly one subscriber in the group, load-balanced across all connected members.

The failure mode occurs when the billing worker uses a direct subscription instead of a queue group subscription. A developer adds a second billing worker pod for redundancy without realizing that nc.subscribe("billing.usage.events", cb=...) creates a new independent subscriber, not a member of a load-balanced group. NATS delivers each usage event to both pods. Both pods call stripe.charges.create() for the same customer within milliseconds. Without an idempotency key, Stripe creates two charges. With three pods, three charges. The NATS monitoring endpoint (/varz, /subsz) shows three subscribers on the subject — correct behavior from the NATS perspective; the number of subscribers is the number of billing runs per event.

import asyncio, nats, stripe, os, json

NATS_URL = os.environ["NATS_URL"]

async def start_billing_worker_unsafe(nc):
    # UNSAFE: direct subscription without queue group.
    # NATS delivers each message to EVERY subscriber on this subject.
    # Three billing worker pods = three stripe.charges.create() calls per usage event.
    sub = await nc.subscribe(
        "billing.usage.events",
        cb=handle_billing_message_unsafe,
    )
    # Expected: one worker processes each billing event.
    # Actual:   all connected workers process every billing event simultaneously.
    return sub

async def handle_billing_message_unsafe(msg):
    data = json.loads(msg.data)
    stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
    # Called once per connected subscriber per usage event.
    # With N billing workers, called N times per customer per billing period.
    charge = 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 — concurrent duplicates create ch_A, ch_B, ch_C
    )

# SAFE: queue group subscription — NATS delivers each message to exactly one subscriber
async def start_billing_worker_safe(nc):
    sub = await nc.subscribe(
        "billing.usage.events",
        queue="billing-workers",   # queue group name — exactly one member processes each message
        cb=handle_billing_message_safe,
    )
    return sub

The JetStream equivalent of this failure mode occurs when two separate durable consumers with different names are both subscribed to the same stream subject. Each durable consumer maintains an independent delivery sequence, so each receives every message published to the stream. If a team deploys a new version of the billing worker (creating a new durable consumer name, e.g., billing-worker-v2) without deleting the old one (billing-worker-v1), both consumers are active simultaneously during the deployment window. Every usage event published during the rollover is processed by both consumers. Without an idempotency key, each consumer’s stripe.charges.create() call creates an independent charge.

The NATS JetStream consumer info API (nats consumer info BILLING billing-worker-v1) shows both consumers healthy, with 0 pending messages and correct ack sequences. Nothing in the monitoring output distinguishes “two workers processing two different events” from “two workers processing the same event twice.” The billing database (if it exists) is the only place where the duplication is visible: two rows for the same (customer_id, billing_period) with different Stripe charge IDs.

Failure mode 3: durable consumer deletion resets delivery sequence to the stream beginning — all retained billing events are reprocessed

NATS JetStream durable consumers are named consumers that persist their delivery sequence across reconnections and restarts. When a durable consumer reconnects after a pod restart, it resumes from the last acknowledged message — unprocessed messages are delivered, already-processed messages are not redelivered. This is the core value proposition of durable consumers over ephemeral ones. The failure mode occurs not on restart, but on deletion.

When a durable consumer is deleted — via nats consumer delete BILLING billing-worker, a Kubernetes job cleanup that calls the NATS management API, or an infrastructure reset that recreates the JetStream environment from scratch — the consumer’s delivery sequence is permanently lost. When the billing system creates a new consumer (same name or a new name) without specifying a delivery policy, DeliverPolicy.ALL is the NATS default. The new consumer starts from stream sequence 1: the first message ever published to the stream, within the stream’s retention window.

import asyncio, nats, stripe, os, json
from nats.js.api import ConsumerConfig, AckPolicy, DeliverPolicy

NATS_URL = os.environ["NATS_URL"]

async def recreate_consumer_unsafe(js):
    # The old durable consumer was deleted during a Kubernetes rollout cleanup.
    # Its delivery sequence — where it last stopped reading — is gone.
    # This new consumer with DeliverPolicy.ALL starts from stream sequence 1.
    # The stream has a 7-day retention window.
    # All billing events from the past 7 days are redelivered.

    psub = await js.pull_subscribe(
        "billing.usage.events",
        durable="billing-worker",       # same name, new consumer — fresh sequence
        config=ConsumerConfig(
            ack_policy=AckPolicy.EXPLICIT,
            deliver_policy=DeliverPolicy.ALL,   # DEFAULT — starts from sequence 1
            # deliver_policy=DeliverPolicy.NEW  # would be safe: only future messages
            # deliver_policy=DeliverPolicy.LAST_PER_SUBJECT  # also safe for some patterns
        ),
    )
    # Every usage event in the stream's 7-day retention window is now
    # redelivered as a new message. The billing function processes each one.
    # Customers billed on day 1 through day 6 are charged again today.
    return psub

async def billing_loop_unsafe(psub):
    while True:
        try:
            msgs = await psub.fetch(batch=50, timeout=1.0)
        except nats.errors.TimeoutError:
            continue

        for msg in msgs:
            data = json.loads(msg.data)
            stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

            # This message may have already been processed by the previous consumer.
            # The new consumer has no knowledge of what the deleted consumer processed.
            # No pre-flight check here — Stripe creates ch_B for a customer
            # who was already charged as ch_A 3 days ago.
            charge = 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
            )
            await msg.ack()

The stream’s retention policy determines the blast radius. A NATS stream with MaxAge=7d and high billing traffic might contain hundreds of thousands of usage events. When the new consumer starts from sequence 1 and processes all of them, every customer in the stream is charged for every billing event published in the past week — regardless of whether those events were already processed. NATS’s stream info shows a correct sequence count, consumer info shows the consumer processing messages normally, and Stripe’s API accepts every charge without error. The duplicate charges accumulate silently until a customer disputes a charge or a reconciliation report flags the anomaly.

The consumer deletion scenario is more common than it appears. Kubernetes Jobs that create ephemeral NATS consumers and clean up after themselves sometimes delete the consumer before the billing run confirms all charges are recorded. Infrastructure-as-code tools (Helm, Terraform) that manage NATS JetStream resources delete and recreate consumers when consumer configuration changes — an AckWait change or a filter subject update triggers a consumer delete + recreate cycle. NATS JetStream’s own pruning (consumers that have been inactive for InactiveThreshold) deletes idle consumers automatically. In each case, the new consumer starts from the stream beginning unless an explicit DeliverPolicy other than ALL is specified.

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() for the same (customer_id, billing_period) pair more than once, and neither NATS nor Stripe’s API layer prevents the subsequent calls from creating new charges. The fix has two independent layers that close all three modes without changing NATS stream configuration, consumer topology, or AckWait settings.

Layer 1: content-hash idempotency key stable across NATS redeliveries and consumer restarts

The Stripe idempotency key must be derived from the business event’s stable fields — fields that do not change across NATS redeliveries, AckWait expiry, or consumer deletion and recreation. The key must be scoped to (customer_id, billing_period) and must not include any NATS message metadata that changes across deliveries: the JetStream sequence number (each redelivery uses the same sequence, but derived fields like delivery count increment), the delivery timestamp, the consumer name, or the NATS server-assigned message ID.

import hashlib, stripe, os, json
import asyncpg

def make_idempotency_key(customer_id: str, billing_period: str) -> str:
    # Derived from stable business identity: customer + billing period only.
    # Must NOT include: NATS sequence number, delivery timestamp, consumer name,
    # AckWait retry count, or NATS server message ID (Nats-Sequence, Nats-Time-Stamp).
    # All of these change across AckWait redeliveries, consumer recreation,
    # and queue group rebalancing — including them generates different keys
    # per attempt, defeating Stripe's 24-hour deduplication.
    raw = f"{customer_id}:{billing_period}:nats-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

async 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)

    # Pre-flight check: was this customer already billed for this period?
    # Covers the consumer deletion case: a 7-day-old billing event that was
    # processed by the deleted consumer is skipped here, not re-charged.
    # The billing_records table is the authoritative source —
    # not the NATS stream position, consumer sequence, or stream lag counter.
    existing = await conn.fetchrow(
        """
        SELECT charge_id, status FROM billing_records
        WHERE customer_id = $1 AND billing_period = $2
        """,
        customer_id, billing_period,
    )
    if existing:
        return {"skipped": True, "reason": "already_billed", "charge_id": existing["charge_id"]}

    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 NATS message.
    # If the write succeeds and the ack fails (AckWait expiry between write and ack),
    # NATS redelivers the message — the pre-flight check above skips it.
    await conn.execute(
        """
        INSERT INTO billing_records
            (customer_id, billing_period, charge_id, amount_cents, status, created_at)
        VALUES ($1, $2, $3, $4, 'charged', NOW())
        ON CONFLICT (customer_id, billing_period) DO NOTHING
        """,
        customer_id, billing_period, charge["id"], amount_cents,
    )

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

The idempotency key excludes the NATS delivery count and AckWait retry counter deliberately. NATS JetStream tracks how many times a message has been delivered via the Nats-Num-Delivered header on push consumers, and the consumer info API shows a delivery count per message. If the idempotency key included the delivery count, each AckWait retry attempt would generate a different key — the first delivery uses key K1, the AckWait redelivery uses key K2, and the DLQ (MaxDeliver) re-attempt uses K3. Each key is unique to Stripe: three separate charge requests, each creating a new charge. The key must be derived from the billing event’s business 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, set at 110% of the expected total. For the consumer deletion failure mode, where a new consumer reprocesses all 7 days of billing events, the vault key provides a hard cap that the idempotency key cannot: Stripe’s 24-hour idempotency window does not protect charges from 3 days ago. The pre-flight check is the primary defense against that case, but the vault key provides a spending circuit breaker if the pre-flight check is bypassed or the billing_records table is also recreated during the infrastructure reset.

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"nats-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"]

# For the consumer deletion scenario (7 days of reprocessed events):
# - Expected total for this billing period: $10,000 (1,000 customers × $10)
# - Vault key cap: $11,000 (10% buffer)
# - After $11,000 charged: proxy rejects further Stripe calls with 429
# - Even if the consumer deletion reprocesses 6 prior billing periods,
#   the vault key blocks charges after the first period's total is exceeded.
# - Important: issue a NEW vault key per billing period, not one shared key
#   for all periods — a shared key would be exhausted by the first period's
#   legitimate charges and block subsequent periods.

Safe consumer: AckWait-aware message handler

import asyncio, nats, stripe, os, json
import asyncpg
from nats.js.api import ConsumerConfig, AckPolicy, DeliverPolicy

async def run_safe_billing_consumer():
    nc = await nats.connect(os.environ["NATS_URL"])
    js = nc.jetstream()
    conn = await asyncpg.connect(os.environ["DATABASE_URL"])

    # Set AckWait high enough to cover the Stripe API P99 latency under load.
    # 120s covers Stripe timeout (30s per call × up to 3 retry attempts with
    # exponential backoff), plus database write and ack round-trip.
    # This reduces AckWait redelivery risk without eliminating the idempotency key.
    psub = await js.pull_subscribe(
        "billing.usage.events",
        durable="billing-worker",
        config=ConsumerConfig(
            ack_policy=AckPolicy.EXPLICIT,
            ack_wait=120,             # 120s: generous for Stripe P99 latency
            max_deliver=3,            # 3 attempts, then move to DLQ subject
            deliver_policy=DeliverPolicy.NEW,  # ONLY future messages — prevents
                                               # consumer recreation replay
        ),
    )

    vault_key = create_billing_vault_key(
        billing_period=current_billing_period(),
        expected_total_cents=get_expected_billing_total(),
    )
    os.environ["KEYBRAKE_VAULT_KEY"] = vault_key

    while True:
        try:
            msgs = await psub.fetch(batch=50, timeout=1.0)
        except nats.errors.TimeoutError:
            continue

        for msg in msgs:
            data = json.loads(msg.data)

            try:
                result = await process_billing_message(conn, data)
                await msg.ack()

            except stripe.error.Timeout:
                # Idempotency key makes AckWait redelivery safe:
                # if the charge was created server-side before the timeout,
                # the retry uses the same idempotency key and returns ch_A.
                # nak() makes the message available for immediate redelivery;
                # use nak(delay=30) for a 30-second backoff between attempts.
                await msg.nak(delay=30)

            except stripe.error.CardError as e:
                # Card decline: permanent failure, record it and ack.
                # Do NOT nak — naking routes to redelivery + eventually DLQ.
                # Acking + recording prevents DLQ accumulation of unresolvable messages.
                await conn.execute(
                    """
                    INSERT INTO billing_records
                        (customer_id, billing_period, charge_id, amount_cents, status, error, created_at)
                    VALUES ($1, $2, NULL, $3, 'card_declined', $4, NOW())
                    ON CONFLICT (customer_id, billing_period) DO NOTHING
                    """,
                    data["customer_id"], data["billing_period"],
                    data["amount_cents"], str(e),
                )
                await msg.ack()

def current_billing_period() -> str:
    from datetime import date
    d = date.today()
    return f"{d.year}-{d.month:02d}"

async def get_expected_billing_total() -> int:
    # Query your billing database for the expected total for this period.
    # Used to set the vault key spend cap at expected_total * 1.10.
    pass

Comparison: protection levels across all three failure modes

FAQ

How do I set AckWait correctly for a Stripe billing consumer?

Set AckWait to at least 3× the maximum expected Stripe API call duration, including retries. Stripe’s stripe-python default timeout is 80 seconds per request. With up to 2 retry attempts (the default max_network_retries=2), a single stripe.charges.create() call can block for up to 240 seconds before raising a final exception. Add your database write time and ack round-trip: 300 seconds (5 minutes) is a reasonable AckWait for billing consumers with Stripe retries enabled. Long AckWait values delay redelivery of genuinely stuck messages, but the idempotency key and pre-flight check make redelivery safe regardless of AckWait duration — so err on the side of a longer wait.

Does the NATS idempotency mechanism (Nats-Msg-Id) protect against duplicate Stripe charges?

NATS JetStream’s Nats-Msg-Id header deduplicates published messages at the stream level — it prevents the same message from being stored in the stream twice (within a configurable deduplication window). It does not prevent a stored message from being delivered to multiple consumers, or from being redelivered after AckWait expiry. The NATS deduplication window (default 2 minutes) prevents publisher-side duplicates from entering the stream, but has no effect on consumer-side redeliveries, AckWait expiry, or consumer deletion replays. It is a complement to, not a substitute for, Stripe-level idempotency keys and pre-flight checks.

Should I use push consumers or pull consumers for billing workloads?

Pull consumers are safer for billing workloads. A pull consumer fetches messages explicitly via psub.fetch(batch=N, timeout=T), giving the billing function control over how many messages are in-flight at a time. This makes it straightforward to bound the AckWait window: fetch only as many messages as the worker can process within AckWait, process them serially, and ack each before fetching the next batch. Push consumers with queue groups can deliver messages faster, but the billing function must track which messages are in-flight and ensure AckWait is long enough to cover the processing time for the entire batch. For billing pipelines where correctness matters more than throughput, pull consumers with a small batch size (10–50) and serial processing within each batch are the recommended pattern.

If my consumer is deleted and recreated, how do I avoid reprocessing historical events?

Use DeliverPolicy.NEW (deliver only messages published after the consumer is created) or DeliverPolicy.BY_START_SEQUENCE with the current stream last-sequence. Never use DeliverPolicy.ALL for a consumer that is recreating after a deletion — it always starts from stream sequence 1. For a safer ops workflow: before deleting a consumer, record its current ack_floor.stream_seq from the consumer info API, and use DeliverPolicy.BY_START_SEQUENCE with that value when recreating. If the consumer info was not recorded before deletion, use DeliverPolicy.NEW and accept that any messages queued in the stream between the old consumer’s last ack and the new consumer’s creation are not processed — this is safer than reprocessing weeks of billing history.

Does the billing_records pre-flight check race condition allow two concurrent workers to both charge the same customer?

Yes, without a database-level constraint. If two billing workers both pass the SELECT ... WHERE customer_id = $1 AND billing_period = $2 pre-flight check simultaneously (before either has written to billing_records), both proceed to call stripe.charges.create(). This is the race condition that the NATS direct-subscription fan-out creates. Closing it requires two things: a unique constraint on (customer_id, billing_period) in the billing_records table (the ON CONFLICT DO NOTHING in the insert handles the race at the database level), and the Stripe idempotency key (which deduplicates concurrent requests within 24 hours at the Stripe layer). The pre-flight check handles the common case — sequential redeliveries and consumer restarts — while the unique constraint closes the concurrent-write race. Both layers are needed.