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

Celery Beat is a separate scheduler process that sits alongside your Celery workers. It reads a schedule — crontab entries or timedelta intervals in beat_schedule — computes which tasks are due, and dispatches them to the broker. Because Beat and workers are decoupled, teams often treat them as independent concerns. But three Beat behaviors — the singleton requirement violated during Kubernetes rolling updates, the catch-up dispatch fired by PersistentScheduler on restart after downtime, and the absence of any deduplication between Beat-dispatched tasks and manually-triggered apply_async() calls — introduce Stripe billing failure modes that Celery's retry primitives and Stripe's restricted keys do not prevent on their own.

This post covers all three failure modes with Beat configuration and Python task 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 task graph.

Failure mode 1: Two Beat pods during Kubernetes rolling update independently dispatch the billing task

The Celery documentation is explicit: "You must ensure that only one Beat instance is running at any given time, otherwise you will end up with duplicate tasks." This warning is easy to satisfy in a VM-based setup where Beat runs as a single supervisor process. It becomes a real failure mode in a Kubernetes deployment where Beat is managed as a Deployment with replicas: 1. The default rolling update strategy — maxUnavailable: 0, maxSurge: 1 — means the rollout creates the new Beat pod before terminating the old one. During the rollout window (typically 10–60 seconds, depending on image pull time and readiness probes), two Beat pods run simultaneously.

Each Beat pod has its own ephemeral filesystem. The default PersistentScheduler stores its state in a local SQLite file (celerybeat-schedule or the path specified by --schedule). With separate SQLite files, neither pod knows the other exists. Both read the same beat_schedule configuration, both independently determine that the monthly billing cron is due (or was due in the last loop iteration), and both call apply_async() against the same Redis broker. Two billing task messages with different task IDs land in the queue. Both are dequeued by workers. Both call stripe.charges.create() for all 800 customers. 800 customers are charged twice.

# tasks.py — UNSAFE: billing task dispatched by Beat, no idempotency key
import stripe
import os
from celery import shared_task
from myapp.models import Customer

@shared_task
def run_monthly_billing(billing_period: str):
    customers = Customer.objects.filter(plan__isnull=False)
    for customer in customers:
        # UNSAFE: no idempotency key.
        # If two Beat pods dispatch this task simultaneously, two task
        # executions run in parallel. Both call stripe.charges.create()
        # for every customer. All 800 customers are charged twice.
        stripe.Charge.create(
            amount=customer.amount_cents,
            currency="usd",
            customer=customer.stripe_customer_id,
            description=f"Billing {billing_period}",
        )

# celery_app.py
app.conf.beat_schedule = {
    "monthly-billing": {
        "task": "myapp.tasks.run_monthly_billing",
        "schedule": crontab(0, 0, day_of_month=1),
        "args": ["2026-07"],  # hardcoded per-period or computed
    },
}

The fix has two parts. First, change the Kubernetes rollout strategy for the Beat deployment to Recreate rather than RollingUpdate. Recreate terminates the existing pod before creating the new one, guaranteeing that at most one Beat pod runs at any time. Accept the brief (10–30 second) downtime during the rollout — for a monthly billing task, missing a few seconds of Beat availability is far less damaging than double-charging 800 customers. Second, use a content-hash idempotency key in the billing task itself. The idempotency key is the second line of defense: even if two Beat pods do dispatch the same task in the rare window where both are running (e.g., due to a readiness probe that passes before the new pod's container process is fully initialized), Stripe deduplicates the charges.

# kubernetes/beat-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-beat
spec:
  replicas: 1
  # SAFE: Recreate terminates old pod before creating new one.
  # Brief downtime during rollout is acceptable for a monthly billing scheduler.
  strategy:
    type: Recreate
  selector:
    matchLabels:
      app: celery-beat
  template:
    metadata:
      labels:
        app: celery-beat
    spec:
      containers:
      - name: beat
        image: my-registry/myapp:latest
        command: ["celery", "-A", "myapp", "beat", "--loglevel=info"]
        env:
        - name: KEYBRAKE_API_KEY
          valueFrom:
            secretKeyRef:
              name: keybrake-credentials
              key: api-key

# tasks.py — SAFE: content-hash idempotency key + vault key
import stripe
import os
import hashlib
from celery import shared_task
from myapp.models import Customer, BillingRecord

KEYBRAKE_PROXY_URL = os.environ.get("KEYBRAKE_PROXY_URL", "https://proxy.keybrake.com")

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

@shared_task
def run_monthly_billing(billing_period: str, vault_key: str):
    stripe.api_key = vault_key
    stripe.api_base = f"{KEYBRAKE_PROXY_URL}/stripe"

    customers = Customer.objects.filter(plan__isnull=False)
    for customer in customers:
        # Pre-flight: skip customers already billed this period.
        if BillingRecord.objects.filter(
            customer=customer, billing_period=billing_period
        ).exists():
            continue

        idempotency_key = billing_idempotency_key(
            customer.stripe_customer_id,
            customer.amount_cents,
            billing_period,
        )

        try:
            charge = stripe.Charge.create(
                amount=customer.amount_cents,
                currency="usd",
                customer=customer.stripe_customer_id,
                description=f"Billing {billing_period}",
                idempotency_key=idempotency_key,
            )
        except stripe.error.CardError:
            continue
        except stripe.error.InvalidRequestError:
            continue

        BillingRecord.objects.get_or_create(
            customer=customer,
            billing_period=billing_period,
            defaults={"charge_id": charge["id"], "amount_cents": customer.amount_cents},
        )

Failure mode 2: PersistentScheduler fires catch-up dispatches on Beat restart after downtime exceeds the billing interval

Celery Beat's default PersistentScheduler writes a last_run_at timestamp to a local SQLite file each time it dispatches a scheduled task. On restart, Beat reads last_run_at and computes how many intervals have passed since the last dispatch. If the Beat process was down for a period longer than the billing interval, Beat fires catch-up dispatches immediately on startup — one for each missed interval — before returning to the regular schedule.

For an hourly billing schedule (run_every=timedelta(hours=1)), a two-hour Beat outage causes Beat to fire two catch-up dispatch calls immediately on restart. Both billing task messages land in the broker queue with different task IDs and are executed by workers concurrently. Because neither worker knows about the other's execution and neither task uses an idempotency key, both workers call stripe.charges.create() for all customers at the same time. If the billing task takes 45 minutes, the first catch-up task is still running when the second catch-up task starts — both tasks attempt to INSERT the same billing records and both attempt to charge the same customers. The customer sees two charges on the same day for the same billing period.

The catch-up behavior is controlled by app.conf.beat_max_loop_interval and the number of missed intervals stored in the celerybeat-schedule file. With run_every=crontab(0, 0, day_of_month=1) (monthly), Beat restoring from a 35-day outage fires two catch-up dispatches (one for the missed July 1 trigger, one for the missed August 1 trigger) and then schedules the next at September 1. Both catch-up tasks execute the billing script with billing_period = datetime.now().strftime("%Y-%m"), which resolves to "2026-08" for both tasks at restart time — both tasks bill for August, one month at a time, charging every customer twice for August.

# UNSAFE: run_every=timedelta(hours=1) with PersistentScheduler.
# A 2-hour Beat outage fires two catch-up dispatches immediately on restart.
# Both tasks execute concurrently. Both bill all customers for the same period.
app.conf.beat_schedule = {
    "hourly-usage-billing": {
        "task": "myapp.tasks.run_hourly_billing",
        "schedule": timedelta(hours=1),
    },
}

@shared_task
def run_hourly_billing():
    billing_period = datetime.utcnow().strftime("%Y-%m-%dT%H:00")
    customers = get_billable_customers()
    for customer in customers:
        # UNSAFE: billing_period computed from datetime.utcnow() at task execution time.
        # Two catch-up tasks running simultaneously both resolve billing_period to the
        # same current hour → same customers charged twice for the same period.
        stripe.Charge.create(
            amount=customer["usage_cents"],
            currency="usd",
            customer=customer["stripe_id"],
            description=f"Usage {billing_period}",
        )

The fix is two-part. First, set app.conf.beat_catchup = False (Celery 5.3+) to disable catch-up dispatch entirely — Beat restarts and resumes from the next scheduled interval rather than firing for missed ones. This is the correct behavior for billing: a missed billing dispatch is better handled by an explicit backfill process than by firing stale tasks after an unpredictable delay. For older Celery versions, set run_every to a value larger than your maximum expected Beat downtime, or use a job store that supports atomic "claim and dispatch" semantics (Redis job store with a distributed lock). Second, use a billing-period parameter passed to the task at dispatch time — not computed inside the task with datetime.now() — combined with a content-hash idempotency key. A billing task dispatched with billing_period="2026-07-13T12:00" will produce the same idempotency key regardless of when the task actually executes.

# SAFE: beat_catchup=False + billing_period passed as argument + idempotency key
from celery.schedules import crontab
from datetime import datetime, timezone

app.conf.beat_catchup = False  # Celery 5.3+: do not fire for missed intervals

app.conf.beat_schedule = {
    "hourly-usage-billing": {
        "task": "myapp.tasks.run_hourly_billing",
        "schedule": crontab(minute=0),
        # Pass billing_period as argument at dispatch time so the task
        # doesn't compute it from datetime.now() at execution time.
        # Beat computes this via a custom schedule entry or a Beat plugin.
    },
}

# Use a custom periodic task that passes billing_period at dispatch:
from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)

@shared_task
def run_hourly_billing(billing_period: str, vault_key: str):
    """billing_period is passed at dispatch time by the Beat schedule entry."""
    stripe.api_key = vault_key
    stripe.api_base = f"{KEYBRAKE_PROXY_URL}/stripe"

    customers = get_billable_customers_for_period(billing_period)
    for customer in customers:
        # Pre-flight: skip period already billed for this customer
        if billing_record_exists(customer["id"], billing_period):
            logger.info(f"Skipping {customer['id']} — already billed for {billing_period}")
            continue

        idempotency_key = billing_idempotency_key(
            customer["id"], customer["usage_cents"], billing_period
        )
        try:
            charge = stripe.Charge.create(
                amount=customer["usage_cents"],
                currency="usd",
                customer=customer["stripe_id"],
                idempotency_key=idempotency_key,
            )
            write_billing_record(customer["id"], billing_period, charge["id"])
        except stripe.error.CardError:
            logger.warning(f"Card declined for {customer['id']}")
        except stripe.error.InvalidRequestError as e:
            logger.error(f"Invalid request for {customer['id']}: {e}")

Failure mode 3: Manual apply_async() alongside Beat-scheduled task creates two untracked billing executions

Celery Beat dispatches tasks automatically according to the schedule. In most engineering teams, there is also a path to trigger the same billing task manually — an admin Django view, a management command, a Slack bot, or an on-call runbook that calls billing_task.apply_async(args=[billing_period, vault_key]) directly. This manual trigger is used when an alert fires for a missed billing period, when a customer reports not receiving an invoice, or when ops wants to run billing early for a subset of customers.

The problem is that Celery has no knowledge of why a task was dispatched. A task message sent by Beat and a task message sent by apply_async() are identical from the broker's perspective — both are JSON payloads with different task IDs sitting in the same queue. Celery workers dequeue and execute both. If the Beat schedule fired the billing task at midnight and an ops engineer fires run_monthly_billing.apply_async(args=["2026-07", vault_key]) at 9 AM because they didn't see the midnight execution complete in the Flower dashboard, two task executions process the same customer list for the same billing period. Without an idempotency key and without a pre-flight check, both task executions call stripe.charges.create() for all customers. The second execution, nine hours after midnight, charges customers who were already charged at midnight.

# management/commands/trigger_billing.py — UNSAFE
# An ops engineer runs: python manage.py trigger_billing --period 2026-07
from django.core.management.base import BaseCommand
from myapp.tasks import run_monthly_billing

class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("--period", required=True)

    def handle(self, *args, **options):
        billing_period = options["period"]
        # UNSAFE: dispatches a new task message with a new task_id.
        # Celery has no knowledge of whether Beat already dispatched this
        # task. If Beat fired at midnight and the operator runs this at 9 AM,
        # a second billing execution starts. No dedup → 800 duplicate charges.
        run_monthly_billing.apply_async(args=[billing_period])
        self.stdout.write(f"Billing task dispatched for {billing_period}")

The fix is to enforce deduplication at the task level rather than at the dispatch level. A pre-flight guard in the billing task itself checks whether a billing run for this period has already started, using a database record or a distributed lock. Any dispatch path — Beat, management command, admin view, Celery Beat plugin, or another task — runs the same pre-flight check and exits early if a run is already in progress or completed. This pattern is called "task idempotency at the business logic level" and is the correct approach for any task that can be triggered from multiple paths.

# tasks.py — SAFE: pre-flight guard with Redis distributed lock
import redis
from contextlib import contextmanager

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

@contextmanager
def billing_period_lock(billing_period: str, timeout: int = 7200):
    """Distributed lock: at most one billing execution per period at a time."""
    lock_key = f"billing_lock:{billing_period}"
    lock = redis_client.lock(lock_key, timeout=timeout)
    acquired = lock.acquire(blocking=False)
    try:
        yield acquired
    finally:
        if acquired:
            lock.release()

@shared_task
def run_monthly_billing(billing_period: str, vault_key: str):
    stripe.api_key = vault_key
    stripe.api_base = f"{KEYBRAKE_PROXY_URL}/stripe"

    with billing_period_lock(billing_period) as acquired:
        if not acquired:
            # Another billing execution is already running for this period.
            # This task was a duplicate dispatch (Beat catch-up, manual trigger, etc.).
            # Exit immediately — the running execution will complete the billing.
            logger.info(f"Billing lock held for {billing_period} — skipping duplicate task")
            return

        # Check persistent record: has this period already been billed?
        if BillingRecord.objects.filter(billing_period=billing_period, status="completed").exists():
            logger.info(f"Billing already completed for {billing_period}")
            return

        customers = Customer.objects.filter(plan__isnull=False)
        for customer in customers:
            if BillingRecord.objects.filter(
                customer=customer, billing_period=billing_period
            ).exists():
                continue

            idempotency_key = billing_idempotency_key(
                customer.stripe_customer_id,
                customer.amount_cents,
                billing_period,
            )
            try:
                charge = stripe.Charge.create(
                    amount=customer.amount_cents,
                    currency="usd",
                    customer=customer.stripe_customer_id,
                    idempotency_key=idempotency_key,
                )
                BillingRecord.objects.get_or_create(
                    customer=customer,
                    billing_period=billing_period,
                    defaults={"charge_id": charge["id"], "amount_cents": customer.amount_cents},
                )
            except stripe.error.CardError:
                logger.warning(f"Card declined for {customer.stripe_customer_id}")
            except stripe.error.InvalidRequestError as e:
                logger.error(f"Invalid request: {e}")

        BillingRecord.objects.filter(billing_period=billing_period, status="in_progress").update(
            status="completed"
        )

Approach comparison

Gap analysis: four scenarios the above does not cover

1. django-celery-beat DatabaseScheduler with two Beat instances and no distributed lock. The DatabaseScheduler from django-celery-beat stores schedule state in the Django database rather than a local SQLite file. Two Beat instances sharing the same database are both aware of the shared schedule. However, the DatabaseScheduler does not use SELECT FOR UPDATE when reading last_run_at — there is a race window between "read last_run_at" and "update last_run_at + dispatch task" where both instances can read the same stale last_run_at and both dispatch. The distributed lock in the billing task is the necessary second line of defense even when using DatabaseScheduler.

2. Beat deployed as a StatefulSet with persistent volume does not prevent catch-up on pod rescheduling. Storing the celerybeat-schedule file on a persistent volume (PVC) rather than ephemeral storage preserves scheduler state across pod restarts. But when the pod is rescheduled to a different node, the PVC must unmount from the old node and mount on the new node before Beat starts. If there is a delay and Beat starts before the PVC is fully mounted (a race common with ReadWriteOnce PVCs), Beat falls back to an in-memory scheduler with no last_run_at history and fires catch-up dispatches for all due tasks. beat_catchup=False prevents this; the billing task lock is the backup.

3. Vault key TTL expires during a long billing run triggered by a catch-up dispatch. If Beat issues a vault key at dispatch time with a TTL sized for the expected billing run (2 hours), but the billing task is delayed in the broker queue for 90 minutes before a worker picks it up, the vault key may have only 30 minutes of remaining TTL when the worker starts executing. A large customer cohort billing run that takes 45 minutes will reach the TTL limit at the 30-minute mark, and subsequent Stripe calls through the proxy will return 401. Size vault key TTL to expected_queue_delay + expected_run_duration + 30 minutes, and monitor proxy 401 rates per task type to detect TTL undersizing before it causes partial billing runs.

4. Beat with beat_schedule defined in two places (app config and a Beat plugin) fires duplicate dispatches without any runtime warning. Some projects define the billing schedule in app.conf.beat_schedule and also register it in a django-celery-beat PeriodicTask record in the database (added via a data migration or admin). Both definitions produce dispatch events at the scheduled time — one from the Beat process reading app.conf.beat_schedule and one from the DatabaseScheduler reading the PeriodicTask record. Celery does not detect or warn about this duplication. The pre-flight distributed lock is the only protection.

Enforcement tests

import hashlib
import pytest
from unittest.mock import patch, MagicMock

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

def test_idempotency_key_is_deterministic_across_task_executions():
    key1 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    key2 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    assert key1 == key2
    assert len(key1) == 32

def test_idempotency_key_differs_per_customer_and_period():
    key1 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    key2 = billing_idempotency_key("cus_def", 4999, "2026-07")
    key3 = billing_idempotency_key("cus_abc", 4999, "2026-08")
    assert key1 != key2
    assert key1 != key3

def test_billing_lock_prevents_concurrent_execution(mocker):
    """Simulate two concurrent task invocations — second should exit immediately."""
    lock_acquired = [True, False]  # first call acquires, second does not
    acquire_calls = iter(lock_acquired)

    mock_lock = MagicMock()
    mock_lock.acquire.side_effect = lambda blocking: next(acquire_calls)
    mocker.patch("myapp.tasks.redis_client.lock", return_value=mock_lock)

    # Second invocation (lock not acquired) should return early without billing
    with patch("myapp.tasks.stripe.Charge.create") as mock_charge:
        result = run_monthly_billing_inner("2026-07", acquired=False)
        mock_charge.assert_not_called()

def test_preflight_skips_completed_billing_period(mocker):
    mocker.patch(
        "myapp.tasks.BillingRecord.objects.filter",
        return_value=MagicMock(exists=lambda: True)
    )
    with patch("myapp.tasks.stripe.Charge.create") as mock_charge:
        # Task should detect completed period and return early
        mock_charge.assert_not_called()

def test_card_decline_does_not_abort_billing_loop():
    """CardError for one customer should log and continue, not raise."""
    try:
        raise stripe.error.CardError("Card declined", "card_declined", "card_declined", 402)
    except stripe.error.CardError:
        pass  # correct: caught and continued, not re-raised

Frequently asked questions

Can I run two Beat instances with a shared Redis job store instead of a local SQLite file?

Redis-backed Beat schedulers (e.g., redbeat) store schedule state in Redis and use a distributed lock to ensure only one Beat instance dispatches at a time. This prevents the singleton violation at the scheduler level and is a safer architecture for Kubernetes deployments than the default PersistentScheduler. You still need idempotency keys and the pre-flight guard in the billing task — redbeat's lock covers Beat dispatch dedup, not task execution dedup. A task re-dispatched by a management command or admin view bypasses redbeat's lock entirely.

Does Celery's task_always_eager mode (used in tests) affect Beat behavior?

task_always_eager=True causes apply_async() to execute the task synchronously in the calling process rather than sending it to the broker. In tests, this means the billing task runs immediately when Beat would dispatch it. Beat itself still computes the schedule and calls apply_async() — in eager mode the task runs in-process. For testing Beat scheduling logic (catch-up behavior, dispatch interval), use task_always_eager=False and inspect the broker's task queue rather than observing side effects in the calling process.

How should billing-period be passed to the billing task from Beat?

The cleanest pattern is a Beat schedule entry with a custom args callable that computes the billing period at dispatch time: "args": lambda: [datetime.utcnow().strftime("%Y-%m")]. This requires a custom Beat schedule entry class or a thin dispatcher task that computes the period and dispatches the billing task as a subtask. Avoid computing the billing period inside the billing task with datetime.now() — catch-up tasks and delayed tasks will compute a different period than the one they were originally dispatched for, creating billing for the wrong month.

What happens to the vault key if the Beat pod restarts and dispatches a catch-up task after the original vault key was issued?

Vault keys are issued at task dispatch time by the calling code that calls apply_async(), not by Beat internally. If Beat fires a catch-up dispatch, the catch-up task call site (the custom dispatcher or Beat plugin) issues a new vault key for the new task. The original vault key (if one was issued for the original dispatch that never reached a worker) expires naturally at its TTL. Two catch-up dispatches each receive their own vault key — the distributed billing lock in the task ensures only one proceeds past the pre-flight guard.

Does the Keybrake proxy's spend cap apply per vault key or per Stripe account?

The spend cap is enforced per vault key. Each billing task execution is issued a distinct vault key with a cap sized to cohort_total_usd × 1.10 + 1 — 10% headroom for rounding and one dollar buffer. A duplicate billing dispatch that somehow bypasses the distributed lock will use its own vault key, which will be capped at the same amount. Two simultaneous billing executions each consuming their full cap would spend 2× the expected amount — the lock is the primary protection against this, and the idempotency key prevents the actual double charge at Stripe even if both executions reach the proxy.

Is concurrency=1 on the Celery worker a substitute for the Beat singleton requirement?

concurrency=1 means the worker processes one task at a time. It does not prevent two task messages for the same billing period from existing in the queue simultaneously — it only prevents them from executing in parallel within that one worker. The first execution dequeues message A, runs the billing task, and completes. The worker then dequeues message B for the same period and starts a second billing run. Without the pre-flight guard, the second execution re-charges all customers whose DB records were committed by the first execution before the second starts. Use concurrency=1 for other reasons (rate limiting, database connection pooling) but not as a substitute for the distributed lock or the idempotency key.