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

Dask is a flexible Python library for parallel and distributed computing that scales workflows from a single laptop to a multi-node cluster without changing the API. Teams reach for it when a billing pipeline outgrows what a single Python process can handle comfortably: a Dask bag or delayed graph reads a large DataFrame of customer records, applies usage-based charge computations in parallel across workers, calls stripe.charges.create() for each customer, and writes results to a database. But three Dask behaviors — worker death that causes the scheduler to retry tasks that may have already called Stripe, dask.delayed graph recomputation that re-executes the entire billing pipeline when .compute() is called again after a downstream failure, and Client.submit() with retries=N that resubmits billing functions that succeeded on Stripe but raised a downstream exception — introduce duplicate-charge failure modes that neither Dask's task graph nor Stripe's Restricted Key feature surfaces automatically.

This post covers all three failure modes with Dask Distributed Python code, and the two-layer governance pattern — content-hash idempotency keys plus per-task vault keys via a spend-cap proxy — that eliminates all three without restructuring your Dask pipeline.

Failure mode 1: Dask worker death retries a billing task that already called Stripe

Dask Distributed runs tasks on a pool of workers managed by a central scheduler. When a worker fails — due to an out-of-memory kill, a preempted spot or preemptible instance, a network partition that causes the worker to stop heartbeating, or a SIGKILL from the OS — the scheduler marks all tasks that were assigned to that worker as failed and resubmits them to live workers. The scheduler's view of task state at the moment of worker failure is binary: a task either returned a result (success) or it did not (failure/unknown). There is no "in-progress" state that the scheduler can distinguish from "not started" — if the worker dies before returning the result, the task is retried regardless of how far it progressed.

The billing failure mode is: a worker running a billing task calls stripe.charges.create() for a customer (Stripe's API receives the request and creates the charge), but the worker is killed (OOM, preemption) before it sends the task result back to the Dask scheduler. The scheduler does not receive a result. It marks the task as failed and resubmits it to a live worker. The live worker receives the same task arguments — same customer ID, same amount — and calls stripe.charges.create() again. Without an idempotency key, Stripe creates a second charge. The Dask task graph eventually reports success (the live worker completed the task), and nothing in Dask's state signals that a duplicate charge was created on the dead worker.

This failure mode is especially common in teams that run Dask on AWS Spot instances or GCP Preemptible VMs to reduce cost — billing pipelines are often scheduled during off-peak hours precisely when spot reclamation is more frequent, and the per-customer billing loop is often the longest-running phase of the pipeline where mid-task worker death is most likely to occur.

# UNSAFE: billing task with no idempotency key on a Dask worker
# If the worker dies after stripe.charges.create() succeeds but before returning
# the result, the scheduler retries the task and Stripe is called again.

from dask.distributed import Client
import stripe

stripe.api_key = "sk_live_..."  # UNSAFE: unrestricted key, no per-task spend cap

def charge_customer(customer: dict) -> dict:
    # UNSAFE: no idempotency key — worker death after this call causes a second charge
    charge = stripe.Charge.create(
        amount=customer["amount_cents"],
        currency="usd",
        customer=customer["stripe_customer_id"],
        description=f"Billing for {customer['billing_period']}"
    )
    return {"customer_id": customer["id"], "charge_id": charge.id}

def run_billing(customer_list: list) -> list:
    client = Client("tcp://dask-scheduler:8786")
    futures = client.map(charge_customer, customer_list)
    # If a worker dies mid-task, the scheduler resubmits the task
    # stripe.charges.create() will be called again on the new worker
    return client.gather(futures)  # UNSAFE: duplicate charges on worker death

The fix is a content-hash idempotency key computed inside the billing task function from values that are stable across all task executions: customer ID, amount in cents, and billing period. The key sha256(customer_id:amount_cents:billing_period:dask-billing)[:32] is identical whether the task runs on the original worker or the retry worker — so when the retry worker calls stripe.charges.create() with the same idempotency key, Stripe returns the existing charge object created by the dead worker's call rather than creating a new one. The charge was already created; the idempotency key makes the retry a no-op at the Stripe level.

# SAFE: content-hash idempotency key + per-task vault key via Keybrake proxy
# Worker death retries are safe: Stripe deduplicates on the idempotency key

import hashlib
import stripe
from dask.distributed import Client

VAULT_KEY    = "vk_live_..."       # per-billing-run vault key from Keybrake
BILLING_PERIOD = "2026-07"         # passed as a constant to all tasks — NOT computed at runtime

def charge_customer(customer: dict) -> dict:
    stripe.api_key  = VAULT_KEY
    stripe.api_base = "https://proxy.keybrake.com/stripe"  # spend-cap proxy

    customer_id  = customer["id"]
    amount_cents = customer["amount_cents"]
    stripe_cus   = customer["stripe_customer_id"]

    # Content-hash idempotency key — identical on original execution and retry
    raw_key = f"{customer_id}:{amount_cents}:{BILLING_PERIOD}:dask-billing"
    idempotency_key = hashlib.sha256(raw_key.encode()).hexdigest()[:32]

    try:
        charge = stripe.Charge.create(
            amount=amount_cents,
            currency="usd",
            customer=stripe_cus,
            description=f"Billing for {BILLING_PERIOD}",
            idempotency_key=idempotency_key  # Stripe deduplicates across worker retries
        )
    except stripe.error.CardError as e:
        # Permanent failure — return an error dict, do NOT raise (would trigger Dask retry)
        return {"customer_id": customer_id, "error": str(e), "charge_id": None}
    except stripe.error.InvalidRequestError as e:
        return {"customer_id": customer_id, "error": str(e), "charge_id": None}

    return {
        "customer_id": customer_id,
        "charge_id": charge.id,
        "amount_cents": amount_cents,
        "idempotency_key": idempotency_key
    }

def run_billing(customer_list: list) -> list:
    client = Client("tcp://dask-scheduler:8786")
    futures = client.map(charge_customer, customer_list)
    return client.gather(futures)  # worker death retries are now safe

Catching CardError and InvalidRequestError and returning an error dict (rather than raising) is critical for Dask billing tasks: raising an exception inside a Dask task causes the scheduler to mark the task as failed, which triggers Dask's own retry logic if retries is set on the submission (see failure mode 3). Permanent per-customer failures should not trigger task retries. Only transient infrastructure errors — network timeouts on the Stripe API itself — should propagate as exceptions that justify a retry.

Failure mode 2: dask.delayed graph recomputation re-executes every Stripe charge after a downstream failure

The dask.delayed API lets you build a computation graph of Python functions that is not executed until you call .compute(). This is useful for billing pipelines: you can compose load_customers(), compute_charges(), call_stripe(customer), and write_to_database(results) as delayed functions, build the dependency graph, and call .compute() to execute the entire pipeline. The failure mode is what happens when .compute() fails partway through and the team calls .compute() again.

Dask's delayed graph has no built-in checkpoint mechanism. When you call .compute() on a delayed graph, Dask executes every node of the graph in dependency order on the available workers. If a node fails — for example, the write_to_database() step raises a database connection error — .compute() raises an exception, and the intermediate results (including all the Stripe charge objects returned by the call_stripe() nodes) exist only in Dask's in-memory task graph. If the workers are still alive, those results are cached in worker memory. But if the team is running this pipeline from a script or a notebook and the exception propagates to the top level, the standard response is to fix the error and call .compute() again — either in the same session (if workers are still running and memory is warm) or in a new session (if the script was restarted).

In a new session — which is the more common case when the pipeline is run as a scheduled job and fails — the delayed graph has no memory of what was computed in the previous session. Calling .compute() re-executes the entire graph: load_customers() loads the same customer list, call_stripe(customer) runs for every customer again, and every stripe.charges.create() call fires again. Without idempotency keys, every customer in the billing period is charged twice.

# UNSAFE: dask.delayed billing pipeline with no idempotency keys
# Calling .compute() again after a database write failure re-charges all customers

import dask
import stripe

stripe.api_key = "sk_live_..."  # UNSAFE: unrestricted key

@dask.delayed
def load_customers(billing_period: str) -> list:
    # Reads customer records from S3 or a database
    return fetch_customer_records(billing_period)

@dask.delayed
def call_stripe(customer: dict) -> dict:
    # UNSAFE: no idempotency key — graph recomputation creates a second charge
    charge = stripe.Charge.create(
        amount=customer["amount_cents"],
        currency="usd",
        customer=customer["stripe_customer_id"]
    )
    return {"customer_id": customer["id"], "charge_id": charge.id}

@dask.delayed
def write_to_database(results: list) -> None:
    # If this raises, .compute() fails — calling .compute() again re-runs call_stripe()
    db_insert_all(results)

def run_billing(billing_period: str):
    customers  = load_customers(billing_period)
    charges    = [call_stripe(c) for c in customers.compute()]  # materializes customer list
    write      = write_to_database(charges)
    write.compute()  # if this fails and you call .compute() again: double charges

The fix is the same content-hash idempotency key pattern, combined with a pre-flight DynamoDB or database check at the start of the billing task. When .compute() is called again on the delayed graph, the call_stripe() nodes re-execute for every customer. The idempotency key makes Stripe's response identical to the first call (within the 24-hour window). The pre-flight database check skips customers where a billing record already exists — protecting beyond the 24-hour Stripe idempotency window and avoiding the Stripe API call entirely for already-billed customers on graph recomputation.

# SAFE: dask.delayed billing pipeline with idempotency keys + pre-flight check
# Graph recomputation after database failure skips already-billed customers

import hashlib
import dask
import stripe
import boto3

VAULT_KEY      = "vk_live_..."
BILLING_PERIOD = "2026-07"

dynamodb      = boto3.resource("dynamodb")
billing_table = dynamodb.Table("billing_records")

@dask.delayed
def call_stripe(customer: dict) -> dict:
    stripe.api_key  = VAULT_KEY
    stripe.api_base = "https://proxy.keybrake.com/stripe"

    customer_id  = customer["id"]
    amount_cents = customer["amount_cents"]
    stripe_cus   = customer["stripe_customer_id"]

    # Pre-flight check: skip if this customer was already billed
    # Protects on graph recomputation after the write_to_database() step fails
    existing = billing_table.get_item(
        Key={"customer_id": customer_id, "billing_period": BILLING_PERIOD}
    ).get("Item")
    if existing:
        return {"customer_id": customer_id, "charge_id": existing["charge_id"], "skipped": True}

    raw_key = f"{customer_id}:{amount_cents}:{BILLING_PERIOD}:dask-billing"
    idempotency_key = hashlib.sha256(raw_key.encode()).hexdigest()[:32]

    try:
        charge = stripe.Charge.create(
            amount=amount_cents,
            currency="usd",
            customer=stripe_cus,
            description=f"Billing for {BILLING_PERIOD}",
            idempotency_key=idempotency_key
        )
    except stripe.error.CardError as e:
        return {"customer_id": customer_id, "error": str(e), "charge_id": None}
    except stripe.error.InvalidRequestError as e:
        return {"customer_id": customer_id, "error": str(e), "charge_id": None}

    # Write billing record immediately — before write_to_database() step
    # This record survives a downstream write_to_database() failure
    billing_table.put_item(Item={
        "customer_id": customer_id,
        "billing_period": BILLING_PERIOD,
        "charge_id": charge.id,
        "amount_cents": amount_cents,
        "idempotency_key": idempotency_key
    })

    return {"customer_id": customer_id, "charge_id": charge.id, "amount_cents": amount_cents}

An additional architectural pattern: write each billing record to the database (or DynamoDB) immediately inside the call_stripe() delayed task, not in a separate write_to_database() downstream node. When the write is co-located with the Stripe call, a downstream aggregation failure cannot leave customers in a "charged but not recorded" state. If write_to_database() is a separate step that aggregates all results and writes them in bulk, a failure in that aggregation leaves all per-customer DynamoDB records intact (since they were written inside call_stripe()) while only the bulk aggregation needs to be retried — and no Stripe charges will be re-issued on retry because the pre-flight check finds the DynamoDB records.

Failure mode 3: Client.submit() with retries=N resubmits billing tasks that succeeded on Stripe but raised a downstream exception

Dask Distributed's Client.submit(func, *args, retries=N) parameter instructs the scheduler to automatically resubmit a failed task up to N additional times if the task raises an exception. This is a common pattern for computational tasks: setting retries=3 makes a data-loading or aggregation task resilient to transient worker failures, network blips, or resource exhaustion. The failure mode appears when teams apply the same retries parameter to billing tasks without accounting for what "resubmit the task" means when the task calls a non-idempotent external API.

The specific sequence: a billing task is submitted with retries=2. The task calls stripe.charges.create() — this succeeds, Stripe creates the charge and returns a Charge object. The task then attempts a database write using the returned charge.id. The database write fails (connection error, write conflict, schema mismatch). The task raises an exception. The Dask scheduler sees an exception from a task with retries=2 and resubmits the entire task function to a live worker. The retry worker calls stripe.charges.create() again — with no idempotency key, Stripe creates a second charge. The database write succeeds on the retry, writing one charge ID — from the second Stripe call. The charge ID from the first Stripe call is not in the database and not visible to billing reconciliation.

# UNSAFE: Client.submit() with retries=N on a billing task that calls Stripe
# A downstream database write failure causes Dask to resubmit the entire task,
# re-executing stripe.charges.create() and creating a second charge.

from dask.distributed import Client
import stripe

stripe.api_key = "sk_live_..."

def charge_and_record(customer: dict) -> str:
    # Step 1: charge Stripe (succeeds)
    charge = stripe.Charge.create(         # UNSAFE: no idempotency key
        amount=customer["amount_cents"],
        currency="usd",
        customer=customer["stripe_customer_id"]
    )

    # Step 2: write to database (may fail)
    db_write(customer["id"], charge.id, customer["billing_period"])  # raises on DB error

    return charge.id

client = Client("tcp://dask-scheduler:8786")

# UNSAFE: retries=2 causes charge_and_record() to be called up to 3 times total
# if the database write fails — Stripe creates a new charge on each call
futures = client.map(charge_and_record, customer_list, retries=2)
results = client.gather(futures)

The fix is identical in structure to the previous failure modes — content-hash idempotency key plus per-customer DynamoDB write immediately after the Stripe call — but the ordering of operations inside the task function is especially important here. The database write that is prone to failure must happen after the billing record has been durably stored somewhere. If the DynamoDB billing record write succeeds and the downstream database write fails, Dask retries the task, the pre-flight DynamoDB check finds the existing billing record, skips the Stripe call, and proceeds directly to the database write. The retry is now safe regardless of how many times Dask resubmits it.

# SAFE: Client.submit() with retries=N — task is safe to resubmit
# DynamoDB write after Stripe call + pre-flight check = safe retry regardless of retries=N

import hashlib
import stripe
import boto3
from dask.distributed import Client

VAULT_KEY      = "vk_live_..."
BILLING_PERIOD = "2026-07"

dynamodb      = boto3.resource("dynamodb")
billing_table = dynamodb.Table("billing_records")

def charge_and_record(customer: dict) -> dict:
    stripe.api_key  = VAULT_KEY
    stripe.api_base = "https://proxy.keybrake.com/stripe"

    customer_id  = customer["id"]
    amount_cents = customer["amount_cents"]
    stripe_cus   = customer["stripe_customer_id"]

    # Pre-flight DynamoDB check: skip Stripe call if already billed
    existing = billing_table.get_item(
        Key={"customer_id": customer_id, "billing_period": BILLING_PERIOD}
    ).get("Item")
    if existing:
        charge_id = existing["charge_id"]
    else:
        raw_key = f"{customer_id}:{amount_cents}:{BILLING_PERIOD}:dask-billing"
        idempotency_key = hashlib.sha256(raw_key.encode()).hexdigest()[:32]

        try:
            charge = stripe.Charge.create(
                amount=amount_cents,
                currency="usd",
                customer=stripe_cus,
                description=f"Billing for {BILLING_PERIOD}",
                idempotency_key=idempotency_key
            )
        except stripe.error.CardError as e:
            return {"customer_id": customer_id, "error": str(e)}
        except stripe.error.InvalidRequestError as e:
            return {"customer_id": customer_id, "error": str(e)}

        # Write to DynamoDB BEFORE the downstream database write that may fail
        billing_table.put_item(Item={
            "customer_id": customer_id,
            "billing_period": BILLING_PERIOD,
            "charge_id": charge.id,
            "amount_cents": amount_cents,
            "idempotency_key": idempotency_key
        })
        charge_id = charge.id

    # This is the step that may fail — after this, Dask retries are safe
    db_write(customer_id, charge_id, BILLING_PERIOD)  # raises on DB error → Dask retries

    return {"customer_id": customer_id, "charge_id": charge_id}

client = Client("tcp://dask-scheduler:8786")
# retries=2 is now safe: pre-flight check + idempotency key prevent double-charges
futures = client.map(charge_and_record, customer_list, retries=2)
results = client.gather(futures)

The two-layer governance pattern for all three failure modes

Each failure mode has a specific structural fix. The same two-layer pattern closes all three and provides defense in depth for Dask executor behaviors you haven't encountered yet.

Layer 1: content-hash idempotency keys + pre-flight check. Before every stripe.charges.create() call, query a durable store (DynamoDB, PostgreSQL) for an existing billing record. If one exists, return the existing charge ID and skip the Stripe call. If no record exists, compute sha256(customer_id:amount_cents:billing_period:dask-billing)[:32] and pass it as idempotency_key. Write the billing record to the durable store immediately after the Stripe call succeeds — before any downstream computation that might fail and trigger a Dask retry. billing_period must be a constant value passed into every task, not computed from datetime.now() at execution time.

Layer 2: per-run vault keys via a spend-cap proxy. Issue a Keybrake vault key before the billing run — scoped to the total expected billing amount across all customers in the current period — and pass it to every Dask task as an environment constant or task argument. Set stripe.api_key = vault_key and stripe.api_base = "https://proxy.keybrake.com/stripe" inside each task. The proxy enforces the per-run USD cap across all concurrent worker calls, forwards the real Stripe key server-side, and logs every proxied request with its idempotency key, timestamp, and HTTP status — giving you a cross-worker audit trail that Dask's distributed task history does not provide.

Dask-specific audit gaps that make Stripe billing failures hard to detect

Dask Distributed's dashboard (served at localhost:8787 by default, or the scheduler address) shows real-time task progress, worker memory usage, and task durations. It records which tasks succeeded and which failed, grouped by function name. It does not record application-level events inside task functions — which customers were charged, which Stripe charge IDs were returned, or whether a given task execution was a retry of a previous execution. When a customer reports a double charge, Dask's task graph gives you no signal: both the original task execution (on the dead worker) and the retry execution (on the live worker) appear as successful tasks, and there is no way to distinguish them in Dask's history.

Dask's Future and distributed.client.futures_of() APIs let you inspect task status at a point in time, but they do not retain historical state across sessions. If the billing script is run as a scheduled job that exits after client.gather(futures), the task history is gone. The only durable record of what happened is in your application-level storage: the DynamoDB billing table written inside each task function, and the Keybrake proxy audit log recording every proxied Stripe request. These two records — written at the moment of each Stripe call — survive worker death, session restart, and Dask cluster teardown.

Stopping an in-flight Dask billing run requires calling client.cancel(futures) on all pending futures. Tasks already executing on workers are not interrupted by cancel() — they run to completion or until the worker process is killed. Tasks that are queued but not yet started on a worker are cancelled. If you need to stop all in-flight Stripe calls immediately — regardless of where they are in the Dask task lifecycle — revoking the Keybrake vault key takes effect at the proxy layer in under a second, blocking all requests that use that vault key regardless of which Dask worker or thread issues them.

FAQ

Can I use the Dask task key as the Stripe idempotency key?

The Dask task key (the string Dask uses internally to identify a unique task in the graph, typically function_name-hash_of_args) is stable across resubmissions of the same task if the function name and arguments are the same. This makes it a tempting candidate for the Stripe idempotency key. However, there are two problems. First, Dask's task key includes a hash of the Python arguments, and the hash algorithm and output format are internal implementation details that can change across Dask versions. Second, the task key is not passed into the task function by default — you would need to use dask.distributed.get_task_stream() or other introspection APIs to retrieve it, which is fragile. The content-hash approach based on business-meaningful inputs (customer ID, amount, billing period) is simpler, version-stable, and correct by construction.

Should I set retries=0 on billing task submissions to be safe?

retries=0 (the default for Client.submit()) prevents the Dask scheduler from automatically resubmitting failed tasks, which eliminates failure mode 3. But it does not prevent failure mode 1 (worker death) — worker death task reassignment happens at the scheduler level regardless of the retries parameter, because the scheduler considers an unacknowledged task to be lost, not "failed" in the sense that retries applies to. Use content-hash idempotency keys as the safety mechanism, not retries=0, since worker death retries are unavoidable in any distributed system.

Does Dask's .persist() help prevent graph recomputation?

dask.persist() submits the computation immediately and holds the results in distributed worker memory. If the downstream write_to_database() step fails after persist() has been called, the Stripe charge results are already in worker memory and calling .compute() on the persisted future does not re-execute the Stripe calls — it just re-materializes the in-memory results. This protects against failure mode 2 as long as the Dask workers are still running and their memory is intact. If the workers are restarted (cluster teardown, new session), the in-memory results are lost and the entire graph must be recomputed. Use .persist() as a performance optimization and immediate-retry safeguard, but use the DynamoDB pre-flight check as the durable protection across sessions and cluster restarts.

How does the Keybrake vault key TTL interact with long-running Dask billing jobs?

Issue the vault key before submitting billing tasks to the Dask cluster, and set its TTL to the expected total runtime of the billing run — the time from first task submission to the last expected client.gather() completion — plus a buffer. Dask's parallel execution means a billing run over 1,000 customers with 20 workers might complete in 5 minutes even if sequential execution would take 50 minutes. Size the TTL to the actual parallel runtime, not the sequential upper bound. If the vault key expires before all tasks complete (because the cluster is under load or workers are slow), tasks that call the proxy after expiry receive a 401 and raise a stripe.error.AuthenticationError. Handle this error class in the task function the same as CardError — return an error dict, do not raise, so the task does not trigger a Dask retry that would also fail against the expired key.

How do I correlate Dask task IDs with Keybrake audit log entries?

Include the Dask task key in the Stripe charge metadata or as an HTTP header in the proxied request. Inside the task function, call dask.distributed.get_client().id to get the current client ID (stable per Dask client instance), and pass it in the description field of stripe.Charge.create() or in a custom header if the proxy supports it. Alternatively, include it in the DynamoDB billing record written inside the task — the Keybrake audit log and the DynamoDB billing table together give you a complete view: the audit log shows every proxied request by idempotency key and timestamp, the DynamoDB table shows the business-level outcome. Cross-referencing the two gives you the full picture of what happened during worker death, graph recomputation, or task resubmission events.