RQ (Redis Queue) Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

RQ (Redis Queue) is the most widely used lightweight Python job queue, valued for its minimal API surface — q.enqueue(func, *args) is the entire dispatch model — and its straightforward fork-per-job worker process. When billing tasks run inside RQ jobs, three framework-specific failure modes emerge that have nothing to do with your Stripe integration itself. They come from how RQ retries failed jobs, how its multi-worker model shares process environment across concurrent dequeues, and how rq-scheduler's recurring dispatch interacts with manually triggered runs.

This post covers three RQ-specific failure modes — job retry re-execution, multi-worker key sharing, and rq-scheduler duplicate dispatch — each of which causes a second stripe.charges.create() call on an already-billed customer, and the two-layer governance pattern that closes all three: content-hash idempotency keys plus per-batch vault keys via a spend-cap proxy, without changing your worker topology.

Failure mode 1: Retry re-queues the job function from line 1 after stripe.charges.create() has already succeeded

RQ supports automatic job retry via the Retry class. When a job function raises an unhandled exception, RQ moves the job to the failed queue and — if Retry(max=N) is configured — re-enqueues it for a fresh execution. The job function is called from the beginning with the original arguments. There is no mid-function resume capability — each retry is a full re-invocation of the same callable with the same args and kwargs.

The failure window is between stripe.charges.create() returning successfully and the job function completing without error. Consider a billing job that charges a customer, then writes the charge ID to a PostgreSQL table. If the database connection times out after the Stripe call succeeds — or any subsequent operation in the job function raises — RQ catches the exception, applies the configured retry interval, and re-enqueues the job. The Stripe call fires again with the same arguments. Without an idempotency key, Stripe creates ch_B. Both charge objects are valid; the customer is charged twice; and the RQ dashboard shows the job succeeded on attempt 2 with no indication of the duplicate charge.

# jobs/billing.py — UNSAFE: no idempotency key
import stripe
import psycopg2

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

def charge_customer(customer_id: str, amount_cents: int, billing_period: str) -> dict:
    # If this succeeds but the DB write below raises → job retried → ch_B created
    charge = stripe.charges.create(
        amount=amount_cents,
        currency="usd",
        customer=customer_id,
        description=f"Subscription {billing_period}",
    )

    # DB connection timeout here → RQ re-enqueues job → charge_customer runs again
    with db.connect() as conn:
        conn.execute(
            "INSERT INTO charges (charge_id, customer_id, billing_period) VALUES (%s, %s, %s)",
            (charge.id, customer_id, billing_period),
        )

    return {"charge_id": charge.id, "status": "charged"}

The fix is a content-hash idempotency key derived from the job arguments that are stable across all retry attempts. RQ passes the same args and kwargs on every retry — customer_id, amount_cents, and billing_period are immutable for the lifetime of the enqueued job — so a key derived from these inputs is identical on every attempt. Stripe returns the original ch_A rather than creating ch_B. Permanent Stripe errors (card declined, invalid customer) should be caught and returned as result dicts — not re-raised — so RQ does not re-enqueue the job for a charge that will never succeed.

# jobs/billing.py — SAFE: stable idempotency key + permanent errors as result dicts
import hashlib
import stripe

def charge_customer(customer_id: str, amount_cents: int, billing_period: str,
                    vault_key: str) -> dict:
    idempotency_key = hashlib.sha256(
        f"{customer_id}:{amount_cents}:{billing_period}:rq-billing".encode()
    ).hexdigest()[:32]

    # Use vault key from job kwargs — issued before batch dispatch, not os.environ
    stripe_client = stripe.Stripe(
        vault_key,
        base_url="https://proxy.keybrake.com/stripe/v1",
    )

    try:
        charge = stripe_client.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Subscription {billing_period}",
            idempotency_key=idempotency_key,
        )
        charge_id = charge.id
        status = "charged"
    except stripe.error.CardError as e:
        # Permanent — record and return without re-raising; RQ will not retry
        with db.connect() as conn:
            conn.execute(
                "INSERT INTO charges (customer_id, billing_period, status)"
                " VALUES (%s, %s, 'card_declined') ON CONFLICT DO NOTHING",
                (customer_id, billing_period),
            )
        return {"status": "card_declined", "error": str(e)}
    except stripe.error.InvalidRequestError as e:
        return {"status": "invalid_request", "error": str(e)}
    # stripe.error.APIConnectionError / APIError → re-raise → RQ retries
    # idempotency key is stable; Stripe returns ch_A, not ch_B

    with db.connect() as conn:
        conn.execute(
            "INSERT INTO charges (charge_id, customer_id, billing_period, status)"
            " VALUES (%s, %s, %s, %s) ON CONFLICT DO NOTHING",
            (charge_id, customer_id, billing_period, status),
        )
    return {"charge_id": charge_id, "status": status}

Failure mode 2: Multiple rq worker processes share one unrestricted STRIPE_SECRET_KEY across concurrent dequeues

RQ uses a fork-per-job execution model. When a worker dequeues a job, it forks a child process to run the job function. If you run multiple rq worker processes — either by starting several worker instances pointing at the same queue, or by using the --num-workers flag in newer RQ versions — all worker processes inherit os.environ from the parent, including STRIPE_SECRET_KEY. Each forked child process reads the same environment variable with no per-job key isolation.

The blast radius of a data error scales with your worker count. If amount_cents has a unit bug — 299900 (dollars denominated as cents) instead of 2999 — every worker process concurrently dequeuing billing jobs from the same queue charges its respective customer the wrong amount before any single worker surfaces an error. With 500 customers dispatched as 500 individual RQ jobs and 8 worker processes running, 8 incorrect charges fire simultaneously in the first wave. Without a spend cap on the key, there is no circuit breaker to halt the remaining 492 jobs once the first 8 complete with errors.

# UNSAFE: all concurrent worker processes share one unrestricted STRIPE_SECRET_KEY
# dispatch.py
import stripe
from rq import Queue
from redis import Redis

# Every forked rq worker child process reads this same env var
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

r = Redis()
q = Queue("billing", connection=r)

for customer in customers:
    # job_id auto-generated UUID — no dedup between duplicate dispatches
    q.enqueue(
        charge_customer,
        customer["id"],
        customer["amount_cents"],
        billing_period,
        retry=Retry(max=3, interval=[10, 30, 60]),
    )

The fix is to issue a per-batch vault key before dispatching jobs to the queue, cap it at the total expected billing amount for the period, and pass it as a keyword argument in each job. The vault key is scoped to POST /v1/charges and carries a daily spend cap equal to the sum of all customer amounts multiplied by a 1.10 buffer. Even if 8 worker processes dequeue jobs simultaneously, the proxy's spend cap is enforced across all concurrent calls sharing that key — the circuit breaker is at the proxy layer, not the job layer.

# dispatch.py — SAFE: vault key issued once before batch, passed as job kwarg
import hashlib
import requests
from rq import Queue, Retry
from redis import Redis

def dispatch_billing_batch(billing_period: str, customers: list[dict]) -> None:
    total_cents = sum(c["amount_cents"] for c in customers)
    max_spend_usd = round((total_cents / 100) * 1.10, 2)

    # Issue one vault key covering the entire batch
    resp = requests.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": max_spend_usd,
            "expires_in_seconds": 7200,
            "label": f"rq-billing-{billing_period}",
        },
        timeout=10,
    )
    resp.raise_for_status()
    vault_key = resp.json()["vault_key"]

    r = Redis()
    q = Queue("billing", connection=r)

    for customer in customers:
        # Deterministic job_id: RQ skips enqueue if a job with this ID already exists
        job_id = hashlib.sha256(
            f"{customer['id']}:{customer['amount_cents']}:{billing_period}".encode()
        ).hexdigest()[:32]

        q.enqueue(
            charge_customer,
            customer["id"],
            customer["amount_cents"],
            billing_period,
            vault_key=vault_key,  # per-batch vault key, not os.environ["STRIPE_SECRET_KEY"]
            job_id=job_id,        # dedup: second enqueue for same customer+period is ignored
            retry=Retry(max=3, interval=[10, 30, 60]),
        )

Failure mode 3: rq-scheduler recurring cron and a manual q.enqueue() call create two independent billing executions for the same period

RQ's companion library rq-scheduler enables recurring job dispatch via scheduler.cron(). When the cron fires, the scheduler calls q.enqueue() internally, producing a new job with an auto-generated ID. If an operator, monitoring script, or at-least-once webhook delivery also calls q.enqueue() for the same billing period — believing the scheduled run stalled — RQ creates a second independent job with a different auto-generated UUID. Both jobs enter the queue, both are dequeued by separate workers, and both execute the billing job function with the same billing_period argument. There is no built-in deduplication between a scheduler-dispatched job and a manually dispatched job for the same period.

This failure mode is particularly dangerous because it is invisible at the RQ level. The RQ dashboard shows two separate job records, both with status finished, both for the same billing period. Every customer in the cohort has been charged twice — once by each worker — but neither job record shows anything wrong. The Stripe dashboard will have two charge objects per customer, created seconds or minutes apart, both with valid status: succeeded.

# UNSAFE: rq-scheduler cron + manual enqueue() create two independent jobs
from rq_scheduler import Scheduler
from redis import Redis

r = Redis()
scheduler = Scheduler(queue_name="billing", connection=r)

# Schedule monthly billing
scheduler.cron(
    "0 0 1 * *",
    func=run_billing_for_period,
    args=["2026-07"],
    # No job_id → rq-scheduler generates an internal ID per cron tick
)

# Operator: billing not yet visible at 00:05 UTC, triggers manual run
from rq import Queue
q = Queue("billing", connection=r)
q.enqueue(
    run_billing_for_period,
    "2026-07",
    # No job_id → RQ generates a UUID → new, independent job
    # Both scheduler job and this job now live in the queue
)

# Result: two separate workers execute run_billing_for_period("2026-07")
# independently → N customers × 2 charges each without idempotency keys

Two layers close this failure mode. The first is a deterministic job_id computed from the billing period and passed to every q.enqueue() call — including the one rq-scheduler makes internally. When a job with that job_id already exists in any state (queued, started, finished), RQ silently skips the duplicate enqueue. The second is the idempotency key in the job function itself, which handles the race where two jobs enter the started state before either has written its result. Providing a job_id to scheduler.cron() requires passing it in the meta field — but using a stable job_id on the manual path alone already closes the most common failure scenario (operator re-trigger while scheduler job is in flight).

# SAFE: deterministic job_id closes the scheduler + manual enqueue collision
import hashlib
from rq import Queue, Retry
from rq_scheduler import Scheduler
from redis import Redis

r = Redis()
q = Queue("billing", connection=r)
scheduler = Scheduler(queue_name="billing", connection=r)

def billing_job_id(billing_period: str) -> str:
    return hashlib.sha256(
        f"monthly-billing:{billing_period}".encode()
    ).hexdigest()[:32]

def enqueue_billing(billing_period: str, vault_key: str) -> None:
    """Single entry point for both scheduled and manual billing dispatch."""
    job_id = billing_job_id(billing_period)
    existing = q.fetch_job(job_id)
    if existing and existing.get_status() in ("queued", "started", "finished"):
        # Dedup: job already in flight or completed for this period
        return

    q.enqueue(
        run_billing_for_period,
        billing_period,
        vault_key=vault_key,
        job_id=job_id,
        retry=Retry(max=3, interval=[10, 30, 60]),
    )

# Schedule via cron — enqueue_billing issues vault_key and calls q.enqueue() with stable job_id
scheduler.cron(
    "0 0 1 * *",
    func=enqueue_billing,
    args=["2026-07", os.environ["KEYBRAKE_API_KEY"]],
)

# Manual operator trigger uses the same function → same deterministic job_id
# If scheduler job already exists, enqueue_billing returns early
enqueue_billing("2026-07", vault_key=vault_key)

Approach comparison

Gap analysis

1. RQ's job_timeout can move a job to failed while stripe.charges.create() is in-flight, creating an ambiguous charge state

RQ's default job_timeout is 180 seconds. If the job function takes longer than job_timeout to complete — whether due to a slow database write, a large batch, or network congestion — RQ sends SIGALRM to the forked child process, raising JobTimeoutException inside the job function at whatever line it is currently executing. If SIGALRM fires after stripe.charges.create() returns but before the result is captured, the charge exists in Stripe with no charge ID recorded locally, and the job is moved to the failed queue. With Retry(max=N), it is re-enqueued and re-executes from line 1. Without an idempotency key, ch_B is created because ch_A has no local record to guard against the retry. The content-hash idempotency key closes this: Stripe returns ch_A regardless of where SIGALRM fires. Size job_timeout conservatively — for billing jobs that write to a database after charging, set it to at least 3× the expected Stripe P99 latency plus your database write P99 to reduce the likelihood of mid-function kills.

2. RQ's failure_ttl controls how long failed job records persist, which affects pre-flight deduplication windows

When a job exhausts all retries, RQ moves it to the failed job registry. Failed job records persist for failure_ttl seconds (default: one year). For billing jobs where the job function returns early on card decline (instead of raising), the job record ends up in the finished registry with a card_declined result. If result_ttl (default: 500 seconds) causes that finished record to expire before the next scheduler cron tick, a second dispatch for the same job_id is treated as a new job — the deterministic job_id dedup is no longer effective. For billing queues, set result_ttl=-1 (never expire) or a value larger than your retry window plus your billing period length to ensure job records persist long enough to guard against re-dispatch.

3. SimpleWorker (non-forking) changes the concurrency model but not the key-sharing problem

RQ's SimpleWorker runs job functions in the same process without forking, useful for environments where os.fork() is unavailable (some container runtimes, Windows). SimpleWorker executes jobs sequentially by default — there is no concurrent job execution within a single SimpleWorker instance. However, if you run multiple SimpleWorker instances pointing at the same queue, each instance reads os.environ['STRIPE_SECRET_KEY'] from its own process. The multi-process key-sharing problem is identical to the standard Worker case — multiple SimpleWorker instances concurrently dequeuing from the same billing queue share no spend boundary. The vault key pattern applies equally: pass vault_key as a job kwarg rather than reading it from the environment inside the job function.

4. rq-scheduler's scheduler.enqueue_at() and scheduler.enqueue_in() do not respect job_id dedup in the same way as q.enqueue()

Unlike q.enqueue(job_id=...) which checks for an existing job and skips the enqueue, rq-scheduler's time-based scheduling methods (scheduler.enqueue_at(), scheduler.enqueue_in()) schedule a future enqueue event in the scheduler's sorted set — not in the job queue directly. The dedup check against the job queue does not fire until the scheduled time arrives and the scheduler moves the job into the queue. If you call scheduler.enqueue_at() twice for the same time and function, two scheduled entries exist in the sorted set and both enqueue into the queue at the scheduled time, potentially with different auto-generated job IDs. Always use the central enqueue_billing() wrapper that checks for existing job records before calling q.enqueue(), regardless of which scheduling path is used to trigger the dispatch.

Enforcement tests

# tests/test_billing.py
import hashlib
import pytest
from unittest.mock import 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}:rq-billing".encode()
    ).hexdigest()[:32]

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

class TestRQBillingIdempotency:
    def test_idempotency_key_stable_across_retries(self):
        """Same inputs on retry 1 and retry 2 produce the same idempotency key."""
        key1 = make_idempotency_key("cus_ABC", 2999, "2026-07")
        key2 = make_idempotency_key("cus_ABC", 2999, "2026-07")
        assert key1 == key2
        assert len(key1) == 32

    def test_idempotency_key_unique_per_customer(self):
        """Different customer IDs produce different idempotency keys."""
        key1 = make_idempotency_key("cus_ABC", 2999, "2026-07")
        key2 = make_idempotency_key("cus_XYZ", 2999, "2026-07")
        assert key1 != key2

    def test_card_declined_returns_dict_not_raises(self):
        """CardError is caught and returned as a result dict — not re-raised."""
        import stripe
        with patch("stripe.Stripe") as mock_stripe_cls:
            mock_client = MagicMock()
            mock_stripe_cls.return_value = mock_client
            mock_client.charges.create.side_effect = stripe.error.CardError(
                "Your card was declined.", None, "card_declined"
            )

            with patch("jobs.billing.db") as mock_db:
                mock_db.connect.return_value.__enter__ = MagicMock()
                mock_db.connect.return_value.__exit__ = MagicMock(return_value=False)
                result = charge_customer("cus_FAIL", 2999, "2026-07", "vault_key_test")

            assert result["status"] == "card_declined"
            # Stripe was called exactly once (not re-raised for retry)
            assert mock_client.charges.create.call_count == 1

    def test_vault_key_cap_includes_ten_percent_buffer(self):
        """Per-batch vault key cap is sum of all amounts × 1.10."""
        customers = [{"id": f"cus_{i}", "amount_cents": 2999} for i in range(10)]
        total_cents = sum(c["amount_cents"] for c in customers)
        expected_cap = round((total_cents / 100) * 1.10, 2)
        assert expected_cap == pytest.approx(32.989)

    def test_deterministic_job_id_identical_for_same_inputs(self):
        """Same customer + amount + period always produces the same job ID."""
        id1 = make_job_id("cus_ABC", 2999, "2026-07")
        id2 = make_job_id("cus_ABC", 2999, "2026-07")
        id3 = make_job_id("cus_XYZ", 2999, "2026-07")
        assert id1 == id2        # stable across calls
        assert id1 != id3        # unique per customer
        assert len(id1) == 32

FAQ

Can I use the RQ job.id as the Stripe idempotency key?

No. RQ generates a UUID for each q.enqueue() call unless you explicitly pass job_id=. Even with deterministic job_id, the RQ job ID format is not guaranteed stable across all dispatch paths — rq-scheduler's internal enqueue may assign a different ID than your manual dispatch. Use a content-hash of the stable billing parameters instead: sha256(customer_id:amount_cents:billing_period:rq-billing)[:32]. This key is stable regardless of which dispatch path created the job and regardless of which attempt the retry is on.

Does Retry(max=3) become safe once idempotency keys are in place?

Yes — for transient failures. All 3 retry attempts compute the same idempotency key from the same job arguments, so Stripe returns ch_A on every retry attempt for an APIConnectionError or timeout. The only remaining risk is permanent failures: card declines and invalid request errors should be caught and returned as result dicts rather than re-raised, so RQ does not schedule further retry attempts for a charge that will never succeed regardless of how many times it is retried.

How do I handle billing the same customer legitimately twice in the same period?

Add a charge_reason or attempt_id parameter to the idempotency key hash to distinguish the two billing intents: sha256(f"{customer_id}:{amount_cents}:{billing_period}:{charge_reason}:rq-billing"). Common values for charge_reason: subscription, overage, upgrade-proration. This produces a distinct idempotency key for each billing intent so the second charge is not deduplicated against the first.

What happens if the vault key expires mid-batch?

Jobs that attempt to call stripe.charges.create() after the vault key expires receive a 401 Unauthorized from the proxy, which the Stripe Python SDK surfaces as stripe.error.AuthenticationError. RQ treats this as a transient error and retries the job if Retry(max=N) is configured — but the vault key is still expired on the next attempt. Size the vault key TTL to cover your largest expected batch duration with a 2× buffer. For 500 customers with 30ms average Stripe latency and 8 workers, the batch takes roughly 500 / 8 × 0.03 = 1.9 seconds, but add RQ scheduling overhead and DB write time. A 7200-second TTL is appropriate for most billing batches up to ~10,000 customers.

Is SimpleWorker compatible with the vault key pattern?

Yes. SimpleWorker executes jobs in the same process without forking, which means vault keys passed as job kwargs are accessed directly from the job's kwargs dictionary — no environment variable inheritance concern. The vault key pattern is actually simpler with SimpleWorker because there is no fork barrier between the dispatch code and the job function. Pass vault_key=vault_key as a kwarg in q.enqueue(), and the job function reads it from kwargs['vault_key'] or as a named parameter.

Do I need separate vault keys per customer or one per batch?

One vault key per batch works for most use cases — the spend cap covers the total expected charges plus a 10% buffer. The proxy enforces the cap across all concurrent worker calls sharing that key, so no individual worker can breach the batch-level budget. For high-risk scenarios where individual charge amounts vary significantly (some customers owe $50, others $5,000), consider per-customer vault keys each capped at customer_amount × 1.10 — this isolates the blast radius to a single customer rather than the entire batch if one job misbehaves.