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

Taskiq is a modern asyncio-native distributed task queue for Python, built around async/await from the ground up and widely adopted in FastAPI and Starlette services for background billing, webhook processing, and AI agent orchestration. Three deployment-level failure modes surface specifically when Taskiq tasks call Stripe — and all three produce duplicate charges that look like successful completions in worker logs.

This post covers three Taskiq-specific failure modes — retry re-execution from line 1, unrestricted key sharing across concurrent async workers, and scheduler-plus-manual-dispatch dedup gaps — each of which produces multiple stripe.charges.create() calls against already-billed customers, and the two-layer governance pattern that closes all three: content-hash idempotency keys plus per-task vault keys via a spend-cap proxy, without changing your task topology.

Failure mode 1: Taskiq task retry re-executes the async billing callable from line 1 after stripe.charges.create() has already succeeded

Taskiq's retry mechanism is configured via the @broker.task(max_retries=N, retry_on_error=True) decorator or the TaskiqDepends middleware chain. When the task raises any exception, Taskiq re-enqueues the task message with the same arguments and a new task ID. The new task instance starts execution from the top of the async function — not from the line that raised.

The critical failure window opens when stripe.charges.create() succeeds but a subsequent operation raises before the task completes. A common example: the Stripe call returns a charge object, but the await db.write_charge(charge.id, customer_id) call times out because the database is briefly unavailable. Taskiq sees an unhandled exception and re-queues the task. The retry starts from line 1: stripe.charges.create() fires again for the same customer with the same amount_cents and no idempotency key. A new charge object is created — ch_B — while ch_A already exists in the Stripe dashboard. Both charges succeed. The worker logs show one task failure and one task success. The Stripe dashboard shows two charges per customer created a few seconds apart.

# tasks.py — UNSAFE: retry re-executes stripe.charges.create() from line 1
import os
import stripe
from taskiq import TaskiqDepends
from app.broker import broker

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # shared, unrestricted

@broker.task(max_retries=3, retry_on_error=True)
async def charge_customer(customer_id: str, amount_cents: int, billing_period: str):
    # Retry starts here — stripe.charges.create() fires again on any exception
    charge = await stripe.charges.create_async(
        amount=amount_cents,
        currency="usd",
        customer=customer_id,
        description=f"Subscription {billing_period}",
    )
    # If this raises (DB timeout, network error), Taskiq re-queues the task
    # and stripe.charges.create() fires again → ch_B created alongside ch_A
    await db.write_charge(charge.id, customer_id, billing_period)
    return charge.id

Two fixes work in combination. The first is a content-hash idempotency key computed from the task's stable input arguments before the Stripe call. The key must be derived from inputs that do not change between the original execution and any retry — customer_id, amount_cents, and billing_period form a stable triple, and appending a service namespace string prevents collisions with idempotency keys from other systems. When the retry fires, stripe.charges.create() receives the same idempotency key and Stripe returns the original charge object rather than creating a new one. The second fix is distinguishing permanent errors from transient ones: CardError and InvalidRequestError cannot be resolved by retrying with the same arguments, so they should be caught and returned as structured result dicts rather than re-raised. Raising them causes Taskiq to retry the task, which will fail again in the same way and exhaust max_retries while potentially creating additional charges if the original charge somehow partially succeeded before the card error was returned.

# tasks.py — SAFE: content-hash idempotency key + permanent errors returned as dicts
import hashlib
import os
import httpx
import stripe
from taskiq import TaskiqDepends
from app.broker import broker

async def get_vault_key(label: str, allowed_cents: int) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://proxy.keybrake.com/keys",
            headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
            json={
                "vendor": "stripe",
                "allowed_endpoints": ["POST /v1/charges"],
                "daily_usd_cap": round((allowed_cents / 100) * 1.10, 2),
                "expires_in_seconds": 3600,
                "label": label,
            },
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()["vault_key"]

@broker.task(max_retries=3, retry_on_error=True)
async def charge_customer(customer_id: str, amount_cents: int, billing_period: str):
    # Idempotency key derived from stable inputs — identical on every retry
    idempotency_key = hashlib.sha256(
        f"{customer_id}:{amount_cents}:{billing_period}:taskiq-billing".encode()
    ).hexdigest()[:32]

    # Per-task vault key capped at the single customer's amount + 10%
    vault_key = await get_vault_key(
        label=f"taskiq-billing-{customer_id}-{billing_period}",
        allowed_cents=amount_cents,
    )

    stripe_client = stripe.AsyncStripe(
        vault_key,
        base_url="https://proxy.keybrake.com/stripe/v1",
    )

    try:
        charge = await stripe_client.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Subscription {billing_period}",
            idempotency_key=idempotency_key,
        )
    except stripe.error.CardError as e:
        # Permanent — retrying will not fix a declined card
        return {"status": "declined", "customer_id": customer_id, "code": e.code}
    except stripe.error.InvalidRequestError as e:
        # Permanent — retrying with the same args will fail the same way
        return {"status": "invalid", "customer_id": customer_id, "error": str(e)}

    await db.write_charge(charge.id, customer_id, billing_period)
    return {"status": "charged", "charge_id": charge.id}

Failure mode 2: Concurrent asyncio workers share one unrestricted STRIPE_SECRET_KEY across all parallel billing tasks with no per-task spend cap

Taskiq dispatches task messages from a broker (Redis, RabbitMQ, or any other supported backend) to workers. Each worker is a long-running asyncio process that pulls messages and executes tasks concurrently. The number of concurrent task executions per worker is controlled by the --tasks-concurrency flag (or equivalent broker setting). By default, a Taskiq asyncio worker runs tasks concurrently within a single event loop — up to the configured concurrency limit — using asyncio.gather-style scheduling inside the worker.

When a billing fan-out dispatches N charge_customer.kiq() calls in rapid succession, N task messages are picked up by workers. Each executing task reads STRIPE_SECRET_KEY from os.environ, which is a process-level dictionary shared across all concurrent coroutines. There is no per-task spend isolation: if a data pipeline bug produces an incorrect amount_cents value — say, $2,999 expressed as 2999 cents rather than the correct $29.99 — all N concurrent billing tasks send the wrong amount to Stripe simultaneously. The error propagates to the entire customer cohort before any single task raises an exception, because each task independently calls stripe.charges.create() with the bad amount and succeeds from Stripe's perspective. The total exposure equals N customers × bad_amount_cents with no circuit breaker.

# billing_fanout.py — UNSAFE: all concurrent tasks share one unrestricted key
import asyncio
from tasks import charge_customer

async def run_monthly_billing(customers: list[dict], billing_period: str):
    # N tasks dispatched with one unrestricted STRIPE_SECRET_KEY
    # A bug in amount_cents charges all N customers wrong simultaneously
    tasks = [
        charge_customer.kiq(
            customer_id=c["id"],
            amount_cents=c["amount_cents"],  # bug here → all N customers charged wrong
            billing_period=billing_period,
        )
        for c in customers
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

The fix is per-task vault keys scoped to each individual customer's amount. Before dispatching the task fan-out, issue one vault key per customer — each key capped at that customer's amount_cents × 1.10 and scoped to POST /v1/charges only. Pass the vault key as a task argument rather than reading a shared environment variable inside the task. Even if amount_cents is wrong in the task arguments, the proxy rejects any charge attempt that exceeds the vault key's per-key spend cap, stopping the damage at the first customer rather than allowing it to propagate across the entire cohort.

# billing_fanout.py — SAFE: per-customer vault keys issued before fan-out dispatch
import asyncio
import httpx
import os

async def issue_vault_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://proxy.keybrake.com/keys",
            headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
            json={
                "vendor": "stripe",
                "allowed_endpoints": ["POST /v1/charges"],
                "daily_usd_cap": round((amount_cents / 100) * 1.10, 2),
                "expires_in_seconds": 3600,
                "label": f"taskiq-billing-{customer_id}-{billing_period}",
            },
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()["vault_key"]

async def run_monthly_billing(customers: list[dict], billing_period: str):
    # Issue one vault key per customer before dispatching any tasks
    vault_keys = await asyncio.gather(*[
        issue_vault_key(c["id"], c["amount_cents"], billing_period)
        for c in customers
    ])

    # Each task receives its own vault key — spend cap enforced at proxy layer
    task_handles = [
        charge_customer_with_key.kiq(
            customer_id=c["id"],
            amount_cents=c["amount_cents"],
            billing_period=billing_period,
            vault_key=vault_key,
        )
        for c, vault_key in zip(customers, vault_keys)
    ]
    results = await asyncio.gather(*task_handles, return_exceptions=True)
    return results

@broker.task(max_retries=3, retry_on_error=True)
async def charge_customer_with_key(
    customer_id: str,
    amount_cents: int,
    billing_period: str,
    vault_key: str,
):
    idempotency_key = hashlib.sha256(
        f"{customer_id}:{amount_cents}:{billing_period}:taskiq-billing".encode()
    ).hexdigest()[:32]

    stripe_client = stripe.AsyncStripe(
        vault_key,
        base_url="https://proxy.keybrake.com/stripe/v1",
    )

    try:
        charge = await stripe_client.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Subscription {billing_period}",
            idempotency_key=idempotency_key,
        )
    except stripe.error.CardError as e:
        return {"status": "declined", "customer_id": customer_id, "code": e.code}
    except stripe.error.InvalidRequestError as e:
        return {"status": "invalid", "customer_id": customer_id, "error": str(e)}

    await db.write_charge(charge.id, customer_id, billing_period)
    return {"status": "charged", "charge_id": charge.id}

Failure mode 3: TaskiqScheduler cron combined with a manual task.kiq() dispatch creates two independent task messages for the same billing period with no built-in deduplication

Taskiq's scheduling layer — TaskiqScheduler with ScheduledTask objects — fires task messages on a cron cadence by calling task.kiq(**kwargs) at the scheduled time. Each scheduled firing generates a new task ID assigned by the broker. A manual await billing_task.kiq(billing_period="2026-07") call triggered from an admin endpoint, a management command, or a webhook handler also generates a new task ID. The two task IDs are different. The broker has no awareness that both messages represent a billing run for the same period — it dequeues and executes both independently.

The deduplication gap matters most in two scenarios. First, an operations engineer manually triggers the billing job after observing that the scheduled run appeared to stall — but the scheduled run was actually executing normally and simply taking longer than expected. Two billing runs now execute concurrently for the same billing_period. Second, a deployment that restarts the scheduler process fires an extra manual .kiq() call as a post-deploy health check, overlapping with the already-queued scheduled task message that the previous scheduler instance sent before it was terminated. In both cases, two task instances arrive at workers within seconds of each other, both with the same customer_id, amount_cents, and billing_period arguments, and both execute stripe.charges.create() independently without any cross-task coordination.

# scheduler.py — UNSAFE: scheduled + manual dispatch creates duplicate task messages
from taskiq_aio_pika import AioPikaBroker
from taskiq.schedule_sources import LabelScheduleSource
from taskiq_redis import ScheduledTaskManager

broker = AioPikaBroker(os.environ["RABBITMQ_URL"])

@broker.task(schedule=[{"cron": "0 0 1 * *"}])  # fires at 00:00 on 1st of each month
async def monthly_billing(billing_period: str = get_current_billing_period()):
    customers = await db.fetch_unpaid_customers(billing_period)
    for c in customers:
        # Scheduled run + manual run both execute this for the same billing_period
        await stripe.charges.create_async(
            amount=c["amount_cents"],
            currency="usd",
            customer=c["id"],
        )

# admin endpoint — triggers second task message for same period
# POST /admin/trigger-billing
await monthly_billing.kiq(billing_period="2026-07")
# → two workers each executing monthly_billing for "2026-07" concurrently

Three approaches close this gap and work best in combination. The first is the content-hash idempotency key: even when two task instances execute concurrently for the same period, every stripe.charges.create() call carries the same idempotency key for each customer. Stripe returns the original charge object on the second call rather than creating a new one — the duplicate execution is safe at the Stripe layer. The second approach is a Redis NX lock keyed on the billing period, acquired before billing begins and released when the run completes. The first task to acquire the lock proceeds; the second task to attempt the same lock period finds the key already set and exits cleanly. The third approach — for teams using Taskiq's label-based scheduler — is to use a deterministic task ID derived from the billing period so that identical-period dispatches are deduplicated at the broker level before any worker picks them up.

# tasks.py — SAFE: idempotency key + Redis NX lock for cross-task dedup
import hashlib
import os
import httpx
import stripe
import redis.asyncio as aioredis
from taskiq_aio_pika import AioPikaBroker
from app.broker import broker

redis_client = aioredis.from_url(os.environ["REDIS_URL"])

@broker.task(
    schedule=[{"cron": "0 0 1 * *"}],
    max_retries=3,
    retry_on_error=True,
)
async def monthly_billing(billing_period: str):
    # Redis NX lock prevents two concurrent billing runs for the same period
    lock_key = f"billing:lock:{billing_period}"
    acquired = await redis_client.set(lock_key, "1", nx=True, ex=7200)
    if not acquired:
        # Another task instance already holds the lock for this period
        return {"status": "skipped", "reason": "lock_held", "period": billing_period}

    try:
        customers = await db.fetch_unpaid_customers(billing_period)
        if not customers:
            return {"status": "no_customers", "period": billing_period}

        total_cents = sum(c["amount_cents"] for c in customers)
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://proxy.keybrake.com/keys",
                headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
                json={
                    "vendor": "stripe",
                    "allowed_endpoints": ["POST /v1/charges"],
                    "daily_usd_cap": round((total_cents / 100) * 1.10, 2),
                    "expires_in_seconds": 7200,
                    "label": f"taskiq-billing-{billing_period}",
                },
                timeout=10,
            )
            resp.raise_for_status()
            vault_key = resp.json()["vault_key"]

        stripe_client = stripe.AsyncStripe(
            vault_key,
            base_url="https://proxy.keybrake.com/stripe/v1",
        )

        results = []
        for c in customers:
            idempotency_key = hashlib.sha256(
                f"{c['id']}:{c['amount_cents']}:{billing_period}:taskiq-billing".encode()
            ).hexdigest()[:32]

            try:
                charge = await stripe_client.charges.create(
                    amount=c["amount_cents"],
                    currency="usd",
                    customer=c["id"],
                    description=f"Subscription {billing_period}",
                    idempotency_key=idempotency_key,
                )
                await db.write_charge(charge.id, c["id"], billing_period)
                results.append({"status": "charged", "charge_id": charge.id})
            except stripe.error.CardError as e:
                results.append({"status": "declined", "customer_id": c["id"], "code": e.code})
            except stripe.error.InvalidRequestError as e:
                results.append({"status": "invalid", "customer_id": c["id"], "error": str(e)})

        return {"status": "complete", "period": billing_period, "results": results}
    finally:
        # Release lock only after full run completes (not on retry)
        await redis_client.delete(lock_key)

Approach comparison

Gap analysis

1. Redis NX lock released in finally block can be deleted by a concurrent retry before the first task completes

The Redis NX lock pattern above uses a finally block to release the lock after the billing run. If the first task instance acquires the lock, begins executing, and then raises an exception that triggers a Taskiq retry, the finally block releases the lock before the retry starts. Taskiq re-queues the task message. If the retry starts quickly enough, it may acquire the lock again before the first instance's finally block executes — or the finally block may delete the lock that the retry already set. To close this race, use a Lua-script-based compare-and-delete: store the task ID as the lock value and only delete the lock if the stored value matches the current task's ID. Taskiq's TaskiqState can provide the current task's broker message ID for use as the lock value, ensuring that only the task instance that set the lock can release it.

2. Taskiq AsyncResultBackend caching can return a stale result on retry, masking a real failure

Taskiq's result backend (Redis, MongoDB, or custom) stores the return value of every completed task keyed by task ID. When a task is retried, the new task instance gets a new task ID — so it does not read the previous failure's cached result. However, if your billing orchestration layer stores the original task ID and polls for its result via await task_handle.get_result(), it reads the first failure's exception rather than waiting for the retry's result. Polling code that re-uses the original task ID handle will see the first failure and may incorrectly conclude that the billing run failed, even though the retry is still executing and will complete successfully. Always use the task handle returned by the most recent .kiq() call — not a stored handle from the first dispatch — to poll for retry results.

3. Taskiq SimpleRetryMiddleware retries on all exceptions including stripe.error.StripeError base class

Taskiq's SimpleRetryMiddleware, when added to the broker middleware chain, catches all exceptions and re-queues the task up to max_retries. If your billing code raises stripe.error.StripeError (the base class for all Stripe exceptions) without distinguishing between transient errors (APIConnectionError, RateLimitError) and permanent ones (CardError, InvalidRequestError), the middleware retries permanent failures. A customer with an invalid card is re-billed up to max_retries times. With a content-hash idempotency key, each retry returns the same Stripe error — the charge is never created twice — but the retry attempts consume worker capacity and log spurious "retry N of 3" entries that obscure real transient failures. Catch CardError and InvalidRequestError explicitly and return them as structured dicts rather than raising.

4. TaskiqState holding a shared stripe.AsyncStripe client across concurrent coroutines mutates global API key on reassignment

Taskiq's TaskiqState and TaskiqDepends dependency injection system allows you to define shared state that is initialized once per worker process and injected into task functions. If you store a single stripe.AsyncStripe client in TaskiqState and share it across concurrent task coroutines, any code that modifies client.api_key (e.g., to swap between test and live keys, or to inject a vault key for a specific task) affects all concurrent coroutines sharing that client. The mutation is not task-scoped. Use the per-task client pattern — stripe.AsyncStripe(vault_key, base_url=proxy) constructed inside each task invocation — so each coroutine holds its own client with its own key, and no mutation of one task's key can affect another task running concurrently in the same event loop.

Enforcement tests

# tests/test_taskiq_billing.py
import hashlib
import pytest
import pytest_asyncio
from unittest.mock import AsyncMock, patch, MagicMock

def make_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    return hashlib.sha256(
        f"{customer_id}:{amount_cents}:{billing_period}:taskiq-billing".encode()
    ).hexdigest()[:32]

class TestTaskiqBillingIdempotency:
    def test_idempotency_key_stable_across_retries(self):
        """Same billing args on original run and retry produce the same key."""
        key_run1 = make_idempotency_key("cus_ABC", 2999, "2026-07")
        key_run2 = make_idempotency_key("cus_ABC", 2999, "2026-07")
        assert key_run1 == key_run2
        assert len(key_run1) == 32

    def test_idempotency_key_unique_per_period(self):
        """Same customer in different billing periods produces different keys."""
        key_jul = make_idempotency_key("cus_ABC", 2999, "2026-07")
        key_aug = make_idempotency_key("cus_ABC", 2999, "2026-08")
        assert key_jul != key_aug

    @pytest.mark.asyncio
    async def test_card_declined_returns_dict_not_raises(self):
        """CardError is caught and returned as a dict — task does not retry."""
        import stripe
        with patch("tasks.stripe.AsyncStripe") as mock_cls:
            mock_client = AsyncMock()
            mock_cls.return_value = mock_client
            mock_client.charges.create.side_effect = stripe.error.CardError(
                "Your card was declined.", None, "card_declined"
            )
            with patch("tasks.get_vault_key", new=AsyncMock(return_value="vk_test")):
                with patch("tasks.db", new=AsyncMock()):
                    result = await charge_customer.fn(
                        customer_id="cus_FAIL",
                        amount_cents=2999,
                        billing_period="2026-07",
                    )
            assert result["status"] == "declined"
            assert result["customer_id"] == "cus_FAIL"

    @pytest.mark.asyncio
    async def test_redis_lock_prevents_duplicate_billing_run(self):
        """Second task for same period finds lock held and returns skipped."""
        with patch("tasks.redis_client") as mock_redis:
            # First call acquires the lock
            mock_redis.set = AsyncMock(side_effect=[True, None])
            mock_redis.delete = AsyncMock(return_value=1)
            with patch("tasks.db.fetch_unpaid_customers", new=AsyncMock(return_value=[])):
                result = await monthly_billing.fn(billing_period="2026-07")
                assert result["status"] == "no_customers"

            # Second call finds lock already held
            mock_redis.set = AsyncMock(return_value=None)
            result2 = await monthly_billing.fn(billing_period="2026-07")
            assert result2["status"] == "skipped"
            assert result2["reason"] == "lock_held"

    def test_vault_key_cap_includes_ten_percent_buffer(self):
        """Per-task vault key spend cap equals customer amount × 1.10."""
        amount_cents = 4999
        expected_cap = round((amount_cents / 100) * 1.10, 2)
        assert expected_cap == pytest.approx(54.989)

FAQ

Does Taskiq's task_id work as an idempotency key?

No. Taskiq assigns a new UUID as task_id each time a task is dispatched — the original task ID and the retry task ID are different. Passing task_id as the Stripe idempotency key means each retry generates a fresh idempotency key and Stripe treats it as a new charge request. The idempotency key must be derived from the stable business arguments of the billing operation — customer_id, amount_cents, and billing_period — not from Taskiq's message-level identifiers. A content-hash of the stable inputs produces the same key on every retry and every duplicate dispatch, which is what Stripe's deduplication window requires.

Can I use Taskiq's built-in task deduplication instead of a Redis NX lock?

Taskiq does not ship with native task-level deduplication out of the box in most broker backends. Some community broker implementations (e.g., taskiq-redis extensions) support dedup via a configurable task ID derived from arguments — but this requires instrumenting the broker layer, not the task function. The Redis NX lock inside the task function is portable across all Taskiq brokers and gives you explicit control over the dedup window (TTL) and the uniqueness key (billing period). It also handles cases where two dispatches arrive from different callers that do not coordinate at the broker level, such as a cron event and a webhook handler firing simultaneously.

What happens to the Redis NX lock if the worker process is killed mid-billing-run?

The EX 7200 TTL on the NX lock ensures the lock expires automatically after 2 hours even if the worker dies before the finally block runs. This prevents the lock from becoming permanently stuck after an OOM kill, SIGKILL, or node failure. Set the TTL to a value slightly longer than your expected worst-case billing run duration — if your largest customer cohort takes 30 minutes to process at Stripe's rate limits, use EX 3600 or EX 7200. The content-hash idempotency key ensures that when the billing run eventually retries after the lock expires, any customers already charged in the first run are safely skipped at the Stripe layer.

How do I issue per-customer vault keys for large billing cohorts without hitting Keybrake's key issuance rate limit?

For large cohorts, issue vault keys in batches rather than one-at-a-time in an asyncio.gather fan-out. A batch key covering the entire billing run — capped at total_customers × avg_amount_cents × 1.10 — works well when all customers in the cohort are billed at the same amount, because the cap is tight relative to the actual batch spend. For cohorts with variable amounts, use a per-tier key: one key per pricing tier, capped at the number of customers in that tier times their tier amount plus 10%. This reduces key issuance calls from N (one per customer) to T (one per tier), while still providing spend isolation at the tier level rather than trusting the full cohort amount.

Does the vault key pattern work with Taskiq's InMemoryBroker in tests?

Yes. The InMemoryBroker executes tasks synchronously in the test process, which means await task.kiq() blocks until the task completes. The vault key issuance call inside the task is just an httpx.AsyncClient.post() — mock it with AsyncMock returning a fixture vault key string, and patch stripe.AsyncStripe to return a mock charge object. The idempotency key computation and the Redis lock acquisition are testable without any running infrastructure. Use InMemoryBroker for unit tests of billing logic, and a real Redis + broker in integration tests that verify the cross-task dedup behavior.

How should I handle Stripe's rate limit errors in a Taskiq billing run?

Stripe's RateLimitError is a transient error — the correct response is to wait and retry with the same idempotency key. Let Taskiq's retry mechanism handle this: catch RateLimitError and re-raise it so Taskiq's max_retries applies. Add a delay between retries by using Taskiq's retry_delay parameter or by sleeping briefly before re-raising. Because the retry uses the same billing_period and the same customer_id and amount_cents, the content-hash idempotency key is identical — any charge that was successfully created before the rate limit was hit returns the existing charge object rather than creating a new one. Unlike CardError and InvalidRequestError, RateLimitError should be re-raised, not caught and returned as a dict.