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

Faust is a Python stream processing library built on Apache Kafka, designed to bring stream processing idioms into Python with coroutine-based agents, distributed Tables backed by Kafka changelog topics, and periodic tasks via @app.timer and @app.cron. When Faust agents process billing usage events from a Kafka topic and call stripe.charges.create() per event, three Faust-specific behaviors introduce Stripe double-charge failure modes that Kafka consumer group lag metrics and Faust’s own monitoring do not surface as billing risk.

This post covers those three failure modes with Faust Python agent code, content-hash idempotency keys stable across Kafka rebalances and offset resets, 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 Kafka topics, Faust app configuration, or consumer group topology.

Failure mode 1: Faust commits Kafka offsets on a timer — a rebalance during the Stripe call re-delivers the message to the new partition owner

Faust commits Kafka offsets on a configurable timer (commit_interval, default 3 seconds), not after each message is processed. This is a deliberate design choice — per-message commits would make throughput-sensitive stream processing prohibitively expensive. For most stream processing workloads, where processing is idempotent or eventual consistency is acceptable, timer-based commits work correctly. For billing workloads where each Stripe call costs real money, they introduce a specific failure mode.

The failure scenario: a Faust billing agent receives a usage event from the billing_events topic at Kafka offset 100. The agent calls stripe.charges.create() at T=0. The Stripe HTTP call is in-flight. At T=2 (before the 3-second commit interval fires), a Kafka rebalance is triggered — a new Faust worker pod comes online during a Kubernetes rolling deployment, causing the consumer group to rebalance. Kafka reassigns the partition to the new worker. The new worker starts consuming from the last committed offset, which is 99 (the commit that fired before offset 100 was fetched). The new worker receives offset 100 as a new message and calls stripe.charges.create() for the same customer and same billing period. The first worker’s Stripe call returns ch_A at T=4; the new worker’s call returns ch_B at T=6. Both workers log success. Stripe’s dashboard shows two charges for the customer.

import faust
import stripe
import os

app = faust.App(
    "billing-app",
    broker=os.environ["KAFKA_BROKER"],
    # Default commit_interval=3.0 seconds.
    # Offsets are committed every 3 seconds, not per-message.
    # A rebalance between the last commit and the current message
    # causes the new partition owner to re-process from the committed offset.
)

billing_topic = app.topic("billing.usage.events", value_type=bytes)

@app.agent(billing_topic)
async def process_billing_events(stream):
    async for event_bytes in stream:
        import json
        event = json.loads(event_bytes)

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

        # This Stripe HTTP call may take 5-30 seconds under load.
        # If a Kafka rebalance happens before Faust's 3-second commit_interval
        # fires for this offset, the new partition owner starts from the last
        # committed offset and re-delivers this message.
        charge = stripe.Charge.create(
            amount=event["amount_cents"],
            currency="usd",
            customer=event["stripe_customer_id"],
            description=f"Usage billing for {event['billing_period']}",
            # no idempotency_key — new partition owner creates ch_B
        )

        # Offset is committed 0-3 seconds after this point by the commit timer.
        # If rebalance happens in that window, next worker re-processes this offset.
        print(f"Charged {event['customer_id']}: {charge['id']}")

The rebalance-during-Stripe-call window is larger than it appears. A Kafka rebalance is not a rare event in containerized deployments: Kubernetes rolling updates, liveness probe timeouts, horizontal pod autoscaler scale-downs, and node drains all trigger consumer group rebalances. A billing pipeline that processes 500 usage events per billing period, each with a 5-10 second Stripe call, has a Stripe call in-flight for 2,500-5,000 seconds of total processing time. With a 3-second commit interval and 30-second Kafka session timeout, the window during which a rebalance produces a duplicate is non-trivial.

Faust’s commit_interval can be reduced to 0.5 seconds, but this does not eliminate the failure mode — it only tightens the window. As long as the Stripe call takes longer than commit_interval, a rebalance in that window re-delivers the message. Reducing commit_interval also increases Kafka commit overhead and broker load, and still does not protect against offset resets (auto.offset.reset=earliest during a consumer group recreation).

Failure mode 2: Faust Table changelog replay after agent restart shows charged customers as “pending” — the billing agent re-charges them

Faust Tables are distributed in-memory key-value stores backed by a Kafka changelog topic. Each Table write appends a record to the changelog; when a Faust worker starts or receives a rebalanced partition, it replays the changelog from the beginning to rebuild the in-memory state. This is Faust’s durability mechanism — the Table survives worker restarts because the changelog is durable in Kafka. The failure mode occurs when the changelog is missing an update because the agent crashed between calling Stripe and writing to the Table.

The failure scenario: a billing agent uses a Faust Table to track billing status: billing_status_table[customer_id] = "billed" after a successful Stripe charge. The agent processes a usage event, calls stripe.charges.create() at T=0, receives ch_A at T=8, and is about to call billing_status_table[customer_id] = "billed" at T=8.1. The Faust worker process is killed (OOM kill, Kubernetes pod eviction, Python uncaught exception) at T=8.05 — after ch_A is returned by Stripe but before the Table write completes. The changelog does not contain the "billed" update for this customer.

When the Faust worker restarts, it replays the changelog and rebuilds the Table. The customer’s entry shows the state as of the last successful Table write before the crash — either "pending" (if the agent wrote pending before calling Stripe) or absent (if no entry was written yet). A second Faust agent pass, a periodic reconciliation cron, or the same event being replayed from an unacknowledged Kafka offset now sees the customer as unbilled and calls stripe.charges.create() again. ch_B is created alongside ch_A.

import faust
import stripe
import os
import json

app = faust.App(
    "billing-app",
    broker=os.environ["KAFKA_BROKER"],
    store="rocksdb://",   # persistent Table store; changelog topic still required
)

billing_topic = app.topic("billing.usage.events", value_type=bytes)

# Faust Table backed by a Kafka changelog topic.
# Replay is the recovery mechanism: on restart, Faust replays
# the changelog to rebuild this in-memory dict.
billing_status_table = app.Table(
    "billing_status",
    default=str,
    partitions=8,
)

@app.agent(billing_topic)
async def process_billing_events(stream):
    async for event_bytes in stream:
        event = json.loads(event_bytes)
        customer_id = event["customer_id"]

        # Check Table for billing status.
        # UNSAFE: Table state may not reflect the most recent Stripe charge
        # if the worker crashed after stripe.charges.create() but before
        # the Table write committed to the changelog.
        if billing_status_table[customer_id] == "billed":
            continue   # skips correctly when Table is up-to-date

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

        # Stripe call succeeds, ch_A is created.
        charge = stripe.Charge.create(
            amount=event["amount_cents"],
            currency="usd",
            customer=event["stripe_customer_id"],
            description=f"Usage billing for {event['billing_period']}",
            # no idempotency_key
        )

        # If the worker dies here (OOM, eviction, exception) before the
        # next line executes, the changelog never gets the "billed" update.
        # After restart, billing_status_table[customer_id] is still "" (not "billed").
        # The customer's usage event is re-delivered from the uncommitted Kafka offset.
        # The agent sees "" in the Table, skips the "billed" check, and calls
        # stripe.charges.create() again — ch_B is created alongside ch_A.
        billing_status_table[customer_id] = "billed"

The changelog inconsistency window is not limited to agent crashes. Faust Table writes to the changelog are asynchronous — the write is appended to a Kafka producer buffer and flushed on a configurable interval. Between the moment billing_status_table[customer_id] = "billed" is called in Python and the moment the changelog record is durably written to Kafka and visible to a restarting consumer, there is a window. A SIGKILL that hits the Faust process in this window, before the Kafka producer buffer is flushed, produces the same outcome as the crash scenario above: the changelog does not have the update, and the next agent restart treats the customer as unbilled.

The Table-as-billing-state antipattern also creates a subtler issue: changelog compaction. Kafka compacts changelog topics by retaining only the most recent value per key. If the sequence of writes is customer_id -> "pending", customer_id -> "billed", and compaction runs before a replay, only "billed" is retained. This is correct behavior — but if compaction runs after the crash (which left the changelog at "pending" with no subsequent "billed" write), compaction retains "pending" indefinitely. Every new agent replay that includes this customer shows "pending", and every agent that uses the Table as its billing guard re-charges the customer.

Failure mode 3: Faust @app.cron and @app.timer fire on every worker replica simultaneously — N workers create N charges per customer per billing period

Faust supports periodic tasks via @app.timer(interval=3600.0) for interval-based execution and @app.cron("0 0 * * *") for cron-scheduled execution. These decorators register background tasks that fire on the configured schedule. The critical behavior: these tasks fire on every Faust worker instance that is running when the timer or cron triggers. There is no built-in leader election or distributed locking in Faust’s timer implementation. Every worker executes the task independently, at the same time.

The failure scenario: a billing system uses @app.cron("0 0 1 * *") (first day of the month at midnight) to run the monthly billing job. The Faust app is deployed on Kubernetes with 3 replicas for availability. At midnight on the first of the month, all three replicas fire the cron task simultaneously. Each replica iterates over the customer list and calls stripe.charges.create() for each customer. Without coordination between the replicas, each customer is charged three times — once by each replica. With 1,000 customers at $10 each, the billing run creates $30,000 in Stripe charges instead of $10,000.

import faust
import stripe
import os
import json
from datetime import datetime

app = faust.App(
    "billing-app",
    broker=os.environ["KAFKA_BROKER"],
)

# UNSAFE: fires on ALL running worker instances simultaneously.
# Three Kubernetes replicas = three stripe.charges.create() calls per customer.
@app.cron("0 0 1 * *")
async def run_monthly_billing():
    customers = await fetch_all_customers()

    for customer in customers:
        stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

        # Called once per replica per customer per month.
        # With three replicas: three charges per customer, not one.
        charge = stripe.Charge.create(
            amount=customer["monthly_fee_cents"],
            currency="usd",
            customer=customer["stripe_customer_id"],
            description=f"Monthly billing for {datetime.now().strftime('%Y-%m')}",
            # no idempotency_key — each replica creates an independent charge
        )
        print(f"Charged {customer['customer_id']}: {charge['id']}")


# The timer variant has the same problem.
# @app.timer(interval=3600.0) also fires on all replicas simultaneously.
@app.timer(interval=86400.0)
async def run_daily_usage_billing():
    # Runs every 24 hours on every connected Faust worker.
    # Replicas are not coordinated — all start at app startup + interval.
    # If three workers start within 10 seconds of each other (Kubernetes
    # rolling deploy), the timer fire time drifts by only 10 seconds.
    # All three fire within a 10-second window and call Stripe simultaneously.
    customers = await fetch_all_customers()
    for customer in customers:
        await bill_customer(customer)

The timer replica problem is not obvious from Faust’s documentation or error output. Each replica logs a successful billing run with unique Stripe charge IDs. No exception is raised, no conflict is detected, and Faust’s internal task monitoring shows all three cron tasks completing successfully. The duplicate charges are only visible in Stripe’s charge list for each customer, or in a reconciliation report that compares Stripe’s total charges against the expected billing amount.

The timer drift behavior makes this failure mode intermittent in interval-based tasks. Workers that start at different times fire their timers at different offsets — a worker that starts at T=0 fires at T=86400, while a worker that started at T=300 fires at T=86700. In a three-replica deployment where replicas start within 30 seconds of each other (typical Kubernetes rolling deploy), the timers fire within 30 seconds of each other — close enough that all three Stripe calls hit the same customer before any idempotency key would deduplicate them across the 24-hour window. Cron-based tasks have no drift: all replicas fire exactly at the scheduled time, simultaneously, with no offset.

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 Faust’s stream processing layer 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 your Faust app configuration, Kafka topic layout, or deployment replica count.

Layer 1: content-hash idempotency key stable across Kafka rebalances and offset resets

The Stripe idempotency key must be derived from the business event’s stable fields — fields that do not change across Kafka redeliveries, Faust worker restarts, or offset resets. The key must be scoped to (customer_id, billing_period) and must not include any Faust or Kafka message metadata that changes across deliveries: the Kafka offset, the consumer group ID, the Faust worker node name, the partition assignment, or the timestamp at which the event was fetched by the agent.

import hashlib
import stripe
import os
import 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: Kafka offset, consumer group ID, Faust worker node name,
    # partition ID, event fetch timestamp, or Faust Table state.
    # All of these change across Kafka rebalances, worker restarts, and
    # offset resets — including them generates different keys per attempt,
    # defeating Stripe's 24-hour deduplication window.
    raw = f"{customer_id}:{billing_period}:faust-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

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

    idempotency_key = make_idempotency_key(customer_id, billing_period)

    # Pre-flight check: was this customer already billed for this period?
    # This is the authoritative source — not the Faust Table state,
    # not the Kafka offset, and not the consumer group lag metric.
    # The Table changelog inconsistency failure mode is closed here:
    # even if the Table shows "pending" because the worker crashed before
    # writing "billed" to the changelog, this database check correctly
    # skips the customer if billing_records has a row from the prior charge.
    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 the Faust stream acknowledges the event
    # or the Table is updated. If the worker dies after this write, the
    # pre-flight check above will skip the customer on the next processing pass.
    # ON CONFLICT DO NOTHING closes the concurrent cron replica race condition
    # at the database level: the second and third replica's INSERT attempts
    # are silently ignored, and the unique constraint prevents duplicate rows.
    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 Kafka offset and Faust consumer group ID deliberately. The Kafka offset for a given message changes across consumer group resets (auto.offset.reset=earliest during consumer group recreation) and is not a stable identifier for the billing event itself. The Faust worker node name and partition assignment change every rebalance. If the idempotency key included any of these, a rebalance-triggered redelivery would produce a different key on the new partition owner — the second Stripe call would not match the first in Stripe’s deduplication table, and ch_B would be created. 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 cron replica failure mode — where three worker replicas all run the billing cron simultaneously and call stripe.charges.create() for every customer three times — the vault key provides a hard cap that the idempotency key partially addresses (Stripe’s 24-hour deduplication closes same-second duplicates) but does not fully protect against if there is any timing drift between replicas.

import requests
import os
import datetime

def create_billing_vault_key(
    billing_period: str,
    expected_total_cents: int,
) -> str:
    resp = requests.post(
        "https://proxy.keybrake.com/vault/keys",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
        json={
            "vendor": "stripe",
            "label": f"faust-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 cron replica scenario (3 replicas × full customer list):
# - Expected billing total: $10,000 (1,000 customers × $10)
# - Vault key cap: $11,000
# - After $11,000 charged: proxy returns 429 for all subsequent Stripe calls
# - Replicas 2 and 3 see 429 from the vault proxy before their charges complete
# - Combined with idempotency key: same-period charges from replica 2 match
#   Stripe's deduplication within 24h; vault cap provides a hard-stop backstop
# - Issue a NEW vault key per billing period — a shared key is exhausted
#   by the first period's charges and blocks subsequent periods

Safe Faust agent: rebalance-aware stream handler

import faust
import stripe
import os
import json
import asyncpg

app = faust.App(
    "billing-app",
    broker=os.environ["KAFKA_BROKER"],
    commit_interval=1.0,    # tighter window for offset commits; does not eliminate
                            # the race but reduces it from 3s to 1s
)

billing_topic = app.topic("billing.usage.events", value_type=bytes)

_db_conn = None

async def get_db():
    global _db_conn
    if _db_conn is None:
        _db_conn = await asyncpg.connect(os.environ["DATABASE_URL"])
    return _db_conn

@app.agent(billing_topic)
async def process_billing_events(stream):
    conn = await get_db()

    async for event_bytes in stream:
        event = json.loads(event_bytes)

        try:
            result = await process_billing_event(conn, event)

            if result.get("skipped"):
                # Pre-flight check: already billed, safe to advance offset
                continue

        except stripe.error.Timeout:
            # Do not advance the Kafka offset — let Faust redeliver this event.
            # The idempotency key makes redelivery safe: if the charge was
            # created server-side before the timeout, the next attempt returns ch_A.
            # Raising here causes Faust to back off and retry the message.
            raise

        except stripe.error.CardError as e:
            # Permanent failure: write the decline to billing_records and continue.
            # Continuing (not raising) advances the Kafka offset past this event.
            conn = await get_db()
            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
                """,
                event["customer_id"], event["billing_period"],
                event["amount_cents"], str(e),
            )


# SAFE cron: acquire a distributed lock before running billing.
# Faust itself does not provide leader election for cron tasks.
# Use Redis SETNX, PostgreSQL advisory locks, or a lightweight distributed
# lock to ensure only one replica executes the billing run per period.
@app.cron("0 0 1 * *")
async def run_monthly_billing_safe():
    import aioredis
    redis = await aioredis.from_url(os.environ["REDIS_URL"])

    billing_period = get_current_billing_period()
    lock_key = f"billing_lock:{billing_period}"

    # SET key value NX EX — atomic set-if-not-exists with 6-hour expiry.
    # Only the first replica to execute this line acquires the lock.
    # All other replicas see False and skip the billing run.
    acquired = await redis.set(lock_key, "1", nx=True, ex=21600)
    if not acquired:
        return   # another replica is running billing for this period

    try:
        conn = await get_db()
        customers = await fetch_all_customers(conn)
        vault_key = create_billing_vault_key(
            billing_period=billing_period,
            expected_total_cents=await get_expected_billing_total(conn),
        )
        os.environ["KEYBRAKE_VAULT_KEY"] = vault_key

        for customer in customers:
            await process_billing_event(conn, {
                "customer_id": customer["customer_id"],
                "stripe_customer_id": customer["stripe_customer_id"],
                "amount_cents": customer["monthly_fee_cents"],
                "billing_period": billing_period,
            })
    finally:
        await redis.delete(lock_key)

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

Comparison: protection levels across all three failure modes

FAQ

Can I set commit_interval to 0 to commit offsets after every message?

Faust’s commit_interval controls how often the offset commit timer fires, not whether offsets are committed per-message. Setting it to a very small value (0.1 seconds) reduces the rebalance window but does not eliminate it — a rebalance that occurs in the 100ms window between the Stripe call completing and the next commit tick still re-delivers the message. More importantly, very frequent Kafka offset commits increase broker load and can cause consumer group instability at scale. The correct approach is to treat offset redelivery as inevitable — which it is in any Kafka-based system — and make processing idempotent via the idempotency key and pre-flight check, rather than trying to tighten the commit window to zero.

Should I use a Faust Table or an external database for billing state?

Use an external database (PostgreSQL, MySQL) as the authoritative billing record, not a Faust Table. Faust Tables are designed for stream processing state that is rebuilt correctly from the changelog on restart. The changelog consistency guarantee applies to clean shutdowns — where the Faust producer flushes its buffer before the process exits. It does not apply to hard kills (OOM, SIGKILL, node failure) where the Kafka producer buffer may not flush. For billing state, the only safe source of truth is a transactional database with a unique constraint on (customer_id, billing_period). The Faust Table can be used as a fast in-memory cache for read-heavy lookups, but the database is the write authority. Never make billing decisions based solely on Faust Table state.

How do I prevent Faust @app.cron from running on all replicas?

Faust does not provide built-in leader election for cron tasks. The standard pattern is to use a distributed lock — Redis SET NX EX (set-if-not-exists with expiry), PostgreSQL advisory locks (pg_try_advisory_lock), or a dedicated lock service — acquired at the start of the cron handler. Only the first replica to acquire the lock runs the billing logic; all others exit immediately. Set the lock expiry to the maximum expected billing run duration (e.g., 6 hours) to prevent a stalled lock from blocking subsequent monthly runs. The try/finally pattern ensures the lock is released even if the billing run raises an exception, allowing the next cron fire to acquire the lock and retry.

Does Stripe’s idempotency key deduplication window cover the Faust offset redelivery case?

Stripe’s idempotency key deduplication window is 24 hours. If a Faust rebalance redelivers an offset within 24 hours of the original charge, and the second worker uses the same idempotency key, Stripe returns the original charge (ch_A) without creating ch_B. This covers the most common case: a Kubernetes rolling deploy that rebalances the consumer group and redelivers offsets from the past few minutes. It does not cover a auto.offset.reset=earliest consumer group reset that redelivers billing events from prior days, a changelog replay after a week-long maintenance window, or a cron that runs monthly (where the original charge and the duplicate attempt are 30 days apart). The pre-flight database check is the defense for those cases.

How does the pre-flight check handle the race condition between two concurrent billing replicas?

Without a database-level constraint, two replicas can both pass the pre-flight SELECT ... WHERE customer_id = $1 AND billing_period = $2 check simultaneously — both find no row, both proceed to call stripe.charges.create(). The Stripe idempotency key closes this within 24 hours: if both calls use the same key, the second returns ch_A. But the ON CONFLICT DO NOTHING unique constraint on (customer_id, billing_period) in the billing_records INSERT is the database-level guarantee: only one row can exist per pair, and the second INSERT from the concurrent replica silently does nothing. The combination — idempotency key (Stripe layer) + unique constraint (database layer) + pre-flight check (application layer) — closes the race at all three layers simultaneously. The idempotency key handles the concurrent Stripe call; the unique constraint handles the concurrent DB write; the pre-flight check handles sequential redeliveries.