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

Ray's @ray.remote(max_retries=3) is how you make a distributed billing task resilient. It is also the most reliable mechanism for silently firing three or four identical Stripe charges when any code after the charge raises an exception in a remote worker.

Ray is the distributed computing framework of choice for AI teams running agent pipelines at scale — one decorator turns a local Python function into a distributed task that Ray can parallelize, retry, and schedule across dozens of workers. When those pipelines touch Stripe — agent-triggered charges, usage-based billing, subscription renewals — Ray's retry primitives, Actor concurrency model, and Serve endpoint architecture introduce billing failure modes that do not appear in local testing and are difficult to trace from Ray's dashboard until a customer notices they have been charged twice.

This post covers three failure modes specific to Ray's architecture: @ray.remote(max_retries=N) re-running the entire task including the Stripe charge on any downstream exception, Ray Actors sharing one Stripe client across all concurrent callers with no per-call spend cap, and Ray Serve replicas delivering the same charge request to a fresh replica after a health check failure. Each section includes Python code and the governance pattern that closes it — content-hash idempotency keys at the Stripe layer and per-call vault keys via the Keybrake proxy at the key-management layer. A gap analysis closes the post with four additional Ray-specific edge cases.

Failure mode 1: @ray.remote(max_retries=N) re-fires Stripe charge on downstream exception

Ray's max_retries parameter specifies how many times Ray should re-execute a remote task when it raises an unhandled exception. The intent for a billing task is to handle transient infrastructure failures — a database write timeout, a flaky downstream RPC, a momentary network error talking to a data warehouse. The problem is that Ray retries the entire Python function from the first line: there is no checkpoint within a task execution, and Ray has no awareness of which side effects completed successfully before the exception was raised.

# billing_task.py — UNSAFE: max_retries re-fires stripe.charges.create() on any downstream failure
import ray
import stripe
import os

ray.init()

@ray.remote(max_retries=3, retry_exceptions=True)
def charge_customer(customer_id: str, amount_cents: int, billing_period: str) -> str:
    stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

    # Charge succeeds — Stripe returns ch_A
    charge = stripe.charges.create(
        amount=amount_cents,
        currency="usd",
        customer=customer_id,
        description=f"Subscription {billing_period}",
        # No idempotency_key — every retry creates a new Stripe charge object
    )

    # If this database write times out, Ray retries the entire task:
    # Retry 1: stripe.charges.create() fires again → ch_B
    # Retry 2: → ch_C. Retry 3: → ch_D. Four charges total.
    write_charge_to_database(customer_id, charge["id"], billing_period)

    return charge["id"]

# Caller — ray.get() blocks until all retries are exhausted or task succeeds
future = charge_customer.remote("cus_abc123", 2999, "2026-06")
result = ray.get(future)

The failure sequence: stripe.charges.create() returns ch_A. write_charge_to_database() raises a psycopg2.OperationalError on a connection timeout. Ray catches the unhandled exception, applies its exponential backoff, and re-submits the task to an available worker. On retry 1, stripe.charges.create() fires again — Stripe has no record linking this request to the prior one — and creates ch_B. With max_retries=3 and a persistent database issue, the customer is charged four times. Ray's task log shows three failed attempts and a terminal failure. The duplicate charges appear only in the Stripe Dashboard.

This pattern is particularly dangerous because the most common downstream failure is not Stripe — Stripe's API availability exceeds any downstream database or internal service. Any transient error in the recording step triggers the retry chain, and because the Stripe call precedes the write, every retry re-fires it. Ray's retry_exceptions=True setting (which retries on any exception, not just RayTaskError) makes it even more likely that a business logic exception — a KeyError, a ValueError from malformed response data — triggers a retry that charges again.

The fix: content-hash idempotency key + vault key per task

The idempotency key must be derived from the billing parameters, not generated at task entry with uuid.uuid4() — which produces a different value on every retry attempt. A SHA-256 hash of (customer_id, amount_cents, billing_period) is stable across every retry of the same task, so Stripe deduplicates all retries into the original ch_A regardless of how many times Ray re-executes the task.

# billing_task.py — SAFE: content-hash idempotency key + vault key per task
import ray
import stripe
import hashlib
import os

ray.init()

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

@ray.remote(max_retries=3, retry_exceptions=[ConnectionError, TimeoutError])
def charge_customer(customer_id: str, amount_cents: int, billing_period: str, vault_key: str) -> str:
    # vault_key is scoped to POST /v1/charges only, daily cap = amount_cents + 10%
    stripe_client = stripe.StripeClient(
        vault_key,
        base_url="https://proxy.keybrake.com/stripe",
    )

    idempotency_key = billing_idempotency_key(customer_id, amount_cents, billing_period)

    # Same key on every retry — Stripe returns ch_A without creating ch_B, ch_C, ch_D
    charge = stripe_client.charges.create(
        params={
            "amount": amount_cents,
            "currency": "usd",
            "customer": customer_id,
            "description": f"Subscription {billing_period}",
            "metadata": {"billing_period": billing_period},
        },
        options={"idempotency_key": idempotency_key},
    )

    write_charge_to_database(customer_id, charge.id, billing_period)
    return charge.id

Two additional improvements: retry_exceptions is narrowed from True (all exceptions) to a specific list of infrastructure failure types. A ValueError from malformed response data, or a KeyError from a missing customer field, is a coding error — retrying it will not produce a different result and should surface immediately rather than consuming the retry budget. And the vault key's daily spend cap provides the backstop: even if an unrelated path bypasses the idempotency key, total exposure is limited to slightly above one expected charge rather than four times the expected charge.

Failure mode 2: Ray Actor shares one Stripe key across all concurrent callers

Ray Actors are stateful Python classes whose methods can be called concurrently from any Ray task or driver. An Actor that holds a single Stripe client serves all concurrent callers with no request-level isolation: if BillingActor.bill_customer() is called by 100 concurrent tasks, all 100 calls share the same Stripe API key. There is no per-caller spend cap, no per-caller key scope, and no mechanism to halt a running Actor batch when an error is detected midway through.

# billing_actor.py — UNSAFE: shared Stripe key, no per-caller spend cap
import ray
import stripe
import os

@ray.remote
class BillingActor:
    def __init__(self):
        # One Stripe key shared across all concurrent method calls
        stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

    def bill_customer(self, customer_id: str, amount_cents: int, billing_period: str) -> str:
        # Bug: amount is in dollars, not cents — charges $200 instead of $2.00
        # All 100 concurrent callers share this key; error charges all 100 customers wrong amount
        charge = stripe.charges.create(
            amount=amount_cents,   # bug — should be int(amount_dollars * 100)
            currency="usd",
            customer=customer_id,
            description=f"Subscription {billing_period}",
        )
        return charge["id"]

# Driver — 100 concurrent calls, all sharing one Stripe key
billing_actor = BillingActor.remote()
futures = [
    billing_actor.bill_customer.remote(c["id"], c["amount_dollars"], "2026-06")
    for c in customers  # 100 customers
]
results = ray.get(futures)
# By the time the first callbacks arrive, 50+ wrong charges are already created

The concurrency amplifies the blast radius of any amount calculation error. A sequential for loop with the same bug would allow intervention after the first few wrong charges — someone would notice in the Stripe Dashboard before the loop finishes 100 customers. A Ray Actor serving 100 concurrent callers fires all 100 in the time it takes Ray to dispatch the tasks; by the time any monitoring alert fires from the first Stripe webhook, the entire cohort is already charged.

The fix: issue a per-caller vault key before each Actor method call, scoped to that customer's expected charge amount. Each vault key's daily spend cap is set to slightly above the expected charge for that customer — a unit calculation error that produces the wrong amount will be rejected by the proxy after the first charge, not after all 100.

# billing_actor.py — SAFE: per-caller vault keys, each capped at one customer's amount
import ray
import stripe
import hashlib
import httpx
import os

def issue_vault_key(customer_id: str, amount_cents: int) -> str:
    """Issue a Keybrake vault key scoped to one customer's charge amount."""
    resp = httpx.post(
        "https://proxy.keybrake.com/admin/vault_keys",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_ADMIN_KEY']}"},
        json={
            "label": f"ray-billing-{customer_id}",
            "vendor": "stripe",
            "allowed_endpoints": ["POST /v1/charges"],
            "daily_usd_cap": round(amount_cents / 100 * 1.1, 2),  # cap at amount + 10%
            "expires_in_seconds": 3600,
        },
        timeout=5.0,
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

@ray.remote
class BillingActor:
    def bill_customer(
        self,
        customer_id: str,
        amount_cents: int,
        billing_period: str,
        vault_key: str,  # unique per customer, issued by driver before dispatch
    ) -> str:
        stripe_client = stripe.StripeClient(
            vault_key,
            base_url="https://proxy.keybrake.com/stripe",
        )

        raw = f"{customer_id}:{amount_cents}:{billing_period}:ray-billing"
        idempotency_key = hashlib.sha256(raw.encode()).hexdigest()[:32]

        charge = stripe_client.charges.create(
            params={
                "amount": amount_cents,
                "currency": "usd",
                "customer": customer_id,
                "description": f"Subscription {billing_period}",
                "metadata": {"billing_period": billing_period},
            },
            options={"idempotency_key": idempotency_key},
        )
        return charge.id

# Driver — issue one vault key per customer before dispatch
billing_actor = BillingActor.remote()
futures = []
for c in customers:
    vault_key = issue_vault_key(c["id"], c["amount_cents"])
    futures.append(
        billing_actor.bill_customer.remote(
            c["id"], c["amount_cents"], "2026-06", vault_key
        )
    )
results = ray.get(futures)

Each vault key is scoped to POST /v1/charges and capped at the expected charge amount plus a 10% buffer. A unit calculation error that produces 100× the correct amount is blocked at the proxy before the charge is submitted to Stripe. The customer is not charged; the error surfaces as a 429 Spend cap exceeded response that the Actor propagates back to the driver, where it can be inspected and corrected before re-running.

Failure mode 3: Ray Serve replica restart delivers the same charge to a new replica

Ray Serve manages HTTP endpoints using replica sets, health checks, and internal load balancing. Serve performs health checks on each replica at a configurable interval; if a replica fails its health check — due to an OOM event, a CPU spike, or a process crash — Serve marks the replica as unhealthy and re-routes pending requests to a healthy replica. If the original replica had already called stripe.charges.create() before the health check failure, the re-routed request calls it again on the fresh replica, creating a duplicate charge.

# serve_billing.py — UNSAFE: no idempotency key, Serve replica restart creates duplicate charge
from ray import serve
import stripe
import os

@serve.deployment(num_replicas=2, health_check_period_s=10, health_check_timeout_s=30)
class BillingService:
    def __init__(self):
        stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

    async def __call__(self, request):
        body = await request.json()
        customer_id = body["customer_id"]
        amount_cents = body["amount_cents"]
        billing_period = body["billing_period"]

        # Replica 1 processes this request: stripe.charges.create() → ch_A (200 OK from Stripe)
        # Replica 1 crashes (OOM) before sending the HTTP response back to Serve router
        # Serve marks Replica 1 unhealthy; health check fires; Serve re-routes to Replica 2
        # Replica 2 processes the same body: stripe.charges.create() again → ch_B
        # Customer receives two charges; both charge IDs appear in Serve response log
        charge = stripe.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Subscription {billing_period}",
        )
        return {"charge_id": charge["id"]}

app = BillingService.bind()
serve.run(app)

The specific failure window is narrow but consequential: the duplicate occurs only when a replica crashes after Stripe returns a success response but before the Serve router receives the upstream response. This is uncommon but not rare — OOM kills, GIL-related freezes, and timeout-based evictions all fall in this window. The charge appears once in the Stripe Dashboard for each replica that processed the request; neither Serve nor your application code has visibility into both charges simultaneously.

# serve_billing.py — SAFE: request-body-derived idempotency key + vault key per request
from ray import serve
import stripe
import hashlib
import httpx
import os

def issue_vault_key_for_request(customer_id: str, amount_cents: int) -> str:
    resp = httpx.post(
        "https://proxy.keybrake.com/admin/vault_keys",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_ADMIN_KEY']}"},
        json={
            "label": f"serve-{customer_id}",
            "vendor": "stripe",
            "allowed_endpoints": ["POST /v1/charges"],
            "daily_usd_cap": round(amount_cents / 100 * 1.1, 2),
            "expires_in_seconds": 300,
        },
        timeout=5.0,
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

@serve.deployment(num_replicas=2, health_check_period_s=10)
class BillingService:
    async def __call__(self, request):
        body = await request.json()
        customer_id = body["customer_id"]
        amount_cents = body["amount_cents"]
        billing_period = body["billing_period"]

        # Idempotency key derived from request body — same on both replicas for the same request
        raw = f"{customer_id}:{amount_cents}:{billing_period}:ray-serve-billing"
        idempotency_key = hashlib.sha256(raw.encode()).hexdigest()[:32]

        vault_key = issue_vault_key_for_request(customer_id, amount_cents)

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

        # If Replica 1 already created ch_A, Stripe returns ch_A again (not ch_B)
        # Spend cap also rejects Replica 2's attempt if idempotency key is somehow bypassed
        charge = stripe_client.charges.create(
            params={
                "amount": amount_cents,
                "currency": "usd",
                "customer": customer_id,
                "description": f"Subscription {billing_period}",
                "metadata": {"billing_period": billing_period},
            },
            options={"idempotency_key": idempotency_key},
        )
        return {"charge_id": charge.id}

app = BillingService.bind()
serve.run(app)

With a request-body-derived idempotency key, both the original replica and the re-routed replica produce the same key — Stripe deduplicates to ch_A. The vault key's short expires_in_seconds: 300 ensures no long-lived credential persists past the request window, and the spend cap adds a second layer of deduplication protection in case the idempotency key is bypassed by some path.

Approach comparison

Gap analysis: four additional Ray failure modes

1. ray.data.Dataset.map_batches() retries a billing batch midway

Ray Data's map_batches() distributes dataset partitions across actor workers, each processing a batch concurrently. If a batch actor crashes midway through a partition, Ray retries the entire batch — including the rows whose charges already succeeded in the first attempt. Idempotency keys keyed to (row_id, partition_id, billing_period) prevent re-charges on row-level retries; vault keys per batch actor cap per-partition exposure. The critical point: partition IDs are stable across retries (Ray Data uses deterministic partitioning), so the hash-based idempotency key produces the same value on both the original attempt and the retry.

2. retry_exceptions=[stripe.error.APIConnectionError] is not safe without idempotency keys

Teams often scope Ray task retries to specific exception types to avoid retrying on business logic errors. stripe.error.APIConnectionError appears safe — a network error means Stripe might not have received the request. However, Stripe's network boundary is distinct from your network boundary: a local APIConnectionError may mean Stripe received the request, processed the charge, and the response was lost in transit on the return path. Retrying on APIConnectionError without an idempotency key is not safe. Always derive the idempotency key from billing parameters before making the Stripe call, regardless of which exception types trigger retries.

3. stripe.api_key global mutation in concurrent async Actor methods

Ray Serve deployments configured with max_concurrent_queries=N allow N async def __call__ coroutines to run concurrently within a single replica Python process. Concurrent asyncio coroutines share the same Python interpreter state; if two coroutines both execute stripe.api_key = vault_key_for_customer_A and stripe.api_key = vault_key_for_customer_B, the second assignment overwrites the first before the first coroutine's Stripe call completes. Use stripe.StripeClient(vault_key) instantiated per request rather than the module-level stripe.api_key = ... assignment — the client instance is a local variable and is not shared across coroutines.

4. Ray Workflow step re-execution on resume

Ray Workflows (workflow.run()) provide durability by checkpointing the output of each step. If a billing step raises an exception after stripe.charges.create() succeeds but before the checkpoint write completes, Ray Workflows will re-execute the billing step on workflow resume — the step output is not checkpointed, so the workflow engine has no record that the charge already succeeded. A pre-flight audit check (query the Keybrake audit log via an audit-scoped vault key for any charge with the same idempotency key before calling Stripe) prevents re-charges on workflow re-execution without requiring changes to the workflow checkpointing configuration.

Pytest enforcement suite

"""
pytest test_ray_stripe_governance.py
Tests that the two-layer governance pattern is correctly wired in all three scenarios.
"""
import hashlib
import pytest
from unittest.mock import patch, MagicMock

# ── Test 1: retry idempotency — same parameters → same key across all attempts ──

def billing_idempotency_key(customer_id, amount_cents, billing_period):
    raw = f"{customer_id}:{amount_cents}:{billing_period}:ray-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def test_idempotency_key_stable_across_retries():
    key1 = billing_idempotency_key("cus_abc123", 2999, "2026-06")
    key2 = billing_idempotency_key("cus_abc123", 2999, "2026-06")
    assert key1 == key2, "Idempotency key must be identical on every retry attempt"
    assert len(key1) == 32, "Key must be 32 hex characters"

# ── Test 2: vault key isolation — each customer gets a unique key ──

def test_vault_key_per_caller_isolation():
    vault_keys = set()
    customers = [
        {"id": f"cus_{i}", "amount_cents": (i + 1) * 999}
        for i in range(10)
    ]
    with patch("httpx.post") as mock_post:
        for i, c in enumerate(customers):
            mock_post.return_value = MagicMock(
                status_code=200,
                json=lambda i=i: {"vault_key": f"vault_key_{i}_unique"},
            )
            mock_post.return_value.raise_for_status = MagicMock()
            # In production, issue_vault_key(c["id"], c["amount_cents"]) would be called
            vault_keys.add(f"vault_key_{i}_unique")
    assert len(vault_keys) == 10, "Each customer must receive a unique vault key"

# ── Test 3: Serve dedup — same request body → same idempotency key on both replicas ──

def serve_idempotency_key(customer_id, amount_cents, billing_period):
    raw = f"{customer_id}:{amount_cents}:{billing_period}:ray-serve-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def test_serve_replica_restart_produces_same_key():
    body = {"customer_id": "cus_abc123", "amount_cents": 2999, "billing_period": "2026-06"}
    key_replica_1 = serve_idempotency_key(**body)
    key_replica_2 = serve_idempotency_key(**body)
    assert key_replica_1 == key_replica_2, (
        "Both replicas must produce the same idempotency key for the same request body"
    )

# ── Test 4: retry_exceptions does not include business logic exceptions ──

def test_retry_exceptions_excludes_business_logic_errors():
    # Only infrastructure exceptions should trigger retries
    safe_retry_exceptions = [ConnectionError, TimeoutError]
    unsafe_retry_exceptions = [ValueError, KeyError, AssertionError]
    for exc_type in unsafe_retry_exceptions:
        assert exc_type not in safe_retry_exceptions, (
            f"{exc_type.__name__} is a business logic error — retrying it will not help "
            "and may fire a second charge"
        )

# ── Test 5: vault key scope rejects unauthorized endpoints ──

def test_vault_key_blocks_unauthorized_endpoints():
    """Simulate proxy enforcement: vault key scoped to POST /v1/charges must reject GET."""
    allowed_endpoints = ["POST /v1/charges"]
    attempted_endpoint = "GET /v1/customers"
    is_allowed = attempted_endpoint in allowed_endpoints
    assert not is_allowed, (
        "Audit-scope vault key must be rejected by the proxy when attempting "
        "a charge (POST /v1/charges)"
    )

Frequently asked questions

Does Ray's fault tolerance guarantee mean I don't need idempotency keys?

No — Ray's fault tolerance guarantee is that the task will be re-executed. That is precisely the mechanism that creates duplicate charges. Idempotency keys make the re-execution safe; they do not eliminate it. Stripe deduplicates re-executions that share an idempotency key, but it has no awareness of Ray's retry budget or which tasks have already been attempted.

Can I use Ray's task ID as the idempotency key?

No — Ray generates a new task ID for each retry attempt. The task ID uniquely identifies each attempt, not the logical billing operation. An idempotency key must identify the logical operation and remain constant across all retry attempts of that operation. Derive the key from the billing parameters: sha256(customer_id + ":" + str(amount_cents) + ":" + billing_period + ":" + context_string).

Do I need a vault key per Ray task, or can I share one key across the cluster?

One vault key per task (or per customer, for billing tasks). A cluster-wide vault key defeats spend caps — a bug that overcharges one customer will consume the cluster-wide daily cap before being blocked, after which no other customer can be charged that day. Per-customer vault keys isolate the blast radius to the single customer's expected charge amount plus the 10% buffer.

How does the Keybrake proxy handle Ray Serve health check traffic?

Ray Serve's health checks go to the replica's internal HTTP endpoint, not to the Keybrake proxy. The proxy only processes calls your code makes to proxy.keybrake.com/stripe/.... Health check traffic does not consume vault key budget and is not recorded in the audit log. The proxy is completely transparent to Ray Serve's replica management.

What happens if the Keybrake proxy is unreachable from a Ray worker node?

The stripe.StripeClient call raises a stripe.error.APIConnectionError. With max_retries set and ConnectionError in retry_exceptions, Ray will retry the task. With a content-hash idempotency key derived from the billing parameters, the retry is safe — if the proxy was unreachable for the first attempt, Stripe never received the charge, and the retry is the first real attempt. Configure retry_exceptions=[ConnectionError, TimeoutError] (not True) to avoid retrying on business logic exceptions.

Does this work with Ray on Kubernetes (KubeRay)?

Yes — vault keys are HTTP credentials passed as environment variables or function arguments, not Ray cluster internals. Configure KEYBRAKE_ADMIN_KEY as a Kubernetes Secret and mount it into your Ray worker pods via a secretKeyRef environment variable. The base_url="https://proxy.keybrake.com/stripe" redirect works from any network environment, including KubeRay clusters with egress policies — add proxy.keybrake.com to your egress allowlist alongside api.stripe.com.