Building Fault-Tolerant Billing Ingestion with Celery

Cloud billing APIs do not behave like the well-mannered REST endpoints Celery tutorials assume. They throttle aggressively during end-of-month reconciliation, return partial payloads when an export job is still finalizing, and drop TCP connections mid-stream on multi-gigabyte pulls. The specific bottleneck this page solves: a default Celery worker acknowledges a task the moment it is delivered, so a SIGKILL during a 12-minute Cost and Usage Report fetch silently destroys an in-flight chunk, and an at-least-once broker redelivering that same task double-counts every line item it does manage to persist. The goal is a Celery topology where neither a 429 Too Many Requests nor a pod eviction can lose or duplicate a single cost row across AWS CUR, GCP BigQuery exports, and Azure Cost Management endpoints.

This build is the distributed evolution of the in-process worker described in Async Billing Data Processing Patterns, and it implements the four-stage acquisition-to-persistence contract defined in Cloud Billing Data Ingestion & Parsing. Celery’s distributed task queue is the right tool once a single asyncio process can no longer hold the concurrency, but its defaults leak state on failure and assume homogeneous network behavior. Production billing ingestion needs explicit idempotency keys, chunked pagination, jittered exponential backoff, and a dead-letter routing strategy layered on top.

# Root Cause & Failure Modes

Three default behaviors make a naive Celery pipeline lose financial data the first time a provider misbehaves:

• Early acknowledgement. With task_acks_late=False (the default), the broker deletes a task from the queue before the worker finishes. A worker OOM-killed while decompressing a CUR manifest takes the task with it — the broker believes it succeeded, and that billing chunk never lands. At a few hundred accounts pulled nightly, worker churn alone guarantees this fires.
• Prefetch hoarding. The default worker_prefetch_multiplier=4 lets one worker reserve four tasks. During a throttling window where each fetch blocks on Retry-After, those reserved tasks sit idle behind a stalled request instead of being processed by an unblocked sibling, collapsing effective concurrency.
• No idempotency at the sink. Because the broker is at-least-once, every retry and every redelivery replays the write. Without a deduplication key, a single 503-triggered retry storm inflates unblended_cost aggregates by whatever fraction of rows got persisted twice.

The hard limits that drive the design are concrete. AWS S3 returns adaptive SlowDown/503 and CUR finalization lags 24–48 hours; the Azure Cost Management Query endpoint emits 429 with a Retry-After header and paginates by nextLink with finalization up to 72 hours; GCP BigQuery exports throw rateLimitExceeded 403s and settle within 24 hours. A worker that does not respect every one of these signals will either exhaust quota and get locked out, or trip downstream allocation models with gaps. The shared retry mathematics behind the backoff decorator below are derived in Handling Billing API Rate Limits & Retries.

# Production Pipeline Architecture

The pipeline decomposes into four phases, each pinned to its own Celery queue so backpressure in one stage never cascades into another:

1. Acquire — fetch_billing_chunk walks provider pagination under jittered backoff, emitting one raw payload per cursor page.
2. Normalize — normalize_payload collapses provider-specific field names into a single dimensional model and quarantines malformed rows.
3. Persist — persist_billing_chunk upserts on a deterministic idempotency key so redelivery is a no-op.
4. Reconcile / dead-letter — exhausted-retry tasks route to a DLQ; a nightly audit compares ingested counts against provider manifest totals.

Cursor state lives in a durable registry rather than the broker, so a restarted worker resumes from the exact nextLink or manifest offset instead of re-pulling from zero — the same watermark-resumption discipline used in Incremental Sync Strategies for GCP Billing Exports. The idempotent upsert at the persist stage solves the same double-counting class that Azure Cost API Pagination and Deduplication resolves at the pagination layer.

# Step-by-Step Python Implementation

The Celery app sets the four behaviors that make tasks survivable: late acknowledgement, single-task prefetch, rejection-on-lost-worker, and a child recycle ceiling that caps memory growth on large CUR files.

import hashlib
import logging
import random
from typing import Any, Dict, List, Optional

import requests
from celery import Celery

logger = logging.getLogger("billing.ingest")

app = Celery(
    "billing_ingest",
    broker="redis://redis:6379/0",
    backend="db+postgresql://user:pass@db:5432/billing",
)

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    task_acks_late=True,              # ack only after the task returns successfully
    task_reject_on_worker_lost=True,  # requeue if the worker is SIGKILLed mid-task
    worker_prefetch_multiplier=1,     # never hoard tasks behind a throttled request
    worker_max_tasks_per_child=500,   # recycle workers to bound CUR decompression memory
    task_default_retry_delay=10,
    task_max_retries=5,
    task_track_started=True,
    task_routes={                     # isolate each phase on its own queue
        "billing.fetch_chunk": {"queue": "acquire"},
        "billing.persist_chunk": {"queue": "persist"},
    },
)

The fetch task distinguishes provider-driven backoff (429 with Retry-After) from server errors (5xx, where we synthesize jittered exponential delay) and from permanent client errors (4xx, which fail fast to the DLQ rather than burning retries):

@app.task(bind=True, name="billing.fetch_chunk", max_retries=5, acks_late=True)
def fetch_billing_chunk(
    self,
    provider: str,
    endpoint_url: str,
    chunk_id: str,
    auth_headers: Dict[str, str],
    params: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """Fetch one paginated billing chunk with signal-aware retries."""
    try:
        resp = requests.get(endpoint_url, headers=auth_headers, params=params, timeout=30)
        resp.raise_for_status()
        logger.info("fetched chunk=%s provider=%s", chunk_id, provider)
        return resp.json()
    except requests.exceptions.HTTPError as exc:
        status = exc.response.status_code
        if status == 429:
            retry_after = int(exc.response.headers.get("Retry-After", 60))
            logger.warning("throttled chunk=%s retry_after=%ss", chunk_id, retry_after)
            raise self.retry(exc=exc, countdown=retry_after)
        if 500 <= status < 600:
            base = min(2 ** self.request.retries * 10, 300)
            countdown = base + random.uniform(0, base * 0.1)  # bounded jitter, no herd
            logger.warning("server err=%d chunk=%s retry_in=%.1fs", status, chunk_id, countdown)
            raise self.retry(exc=exc, countdown=countdown)
        logger.error("permanent http=%d chunk=%s -> DLQ", status, chunk_id)
        raise
    except requests.exceptions.RequestException as exc:
        countdown = 15 * (2 ** self.request.retries)
        logger.warning("network err chunk=%s retry_in=%ds", chunk_id, countdown)
        raise self.retry(exc=exc, countdown=countdown)

Normalization quarantines bad rows instead of poisoning the chunk, and persistence upserts on a SHA-256 idempotency key so an at-least-once redelivery collapses to a no-op:

from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError

engine = create_engine("postgresql://user:pass@db:5432/billing", pool_pre_ping=True)

# Provider field aliases collapsed to one dimensional model before persistence.
_FIELD_MAP = {"cost": ("cost", "unblended_cost"), "account": ("accountId", "billing_account_id")}


def normalize_payload(provider: str, chunk_id: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
    rows: List[Dict[str, Any]] = []
    for row in raw.get("items", []):
        try:
            rows.append({
                "provider": provider,
                "account_id": row.get("accountId") or row["billing_account_id"],
                "billing_period": raw["billing_period"],
                "service_name": row.get("serviceName") or row.get("service", "unknown"),
                "usage_quantity": float(row.get("usageAmount", 0)),
                "currency": row.get("currency", "USD"),
                "unblended_cost": float(row.get("cost") or row.get("unblended_cost", 0)),
                "chunk_id": chunk_id,
            })
        except (KeyError, TypeError, ValueError) as exc:
            logger.warning("schema violation chunk=%s skipped: %s", chunk_id, exc)
    return rows


@app.task(bind=True, name="billing.persist_chunk", max_retries=3, acks_late=True)
def persist_billing_chunk(self, items: List[Dict[str, Any]]) -> int:
    """Upsert normalized rows with idempotency guarantees."""
    if not items:
        return 0
    for idx, item in enumerate(items):
        seed = f"{item['provider']}:{item['account_id']}:{item['billing_period']}:{item['chunk_id']}:{idx}"
        item["idempotency_key"] = hashlib.sha256(seed.encode()).hexdigest()
    query = text("""
        INSERT INTO billing_line_items
            (provider, account_id, billing_period, service_name, usage_quantity,
             currency, unblended_cost, chunk_id, idempotency_key)
        VALUES
            (:provider, :account_id, :billing_period, :service_name, :usage_quantity,
             :currency, :unblended_cost, :chunk_id, :idempotency_key)
        ON CONFLICT (idempotency_key) DO NOTHING
    """)
    try:
        with engine.begin() as conn:
            result = conn.execute(query, items)
            logger.info("persisted rows=%d chunk=%s", result.rowcount, items[0]["chunk_id"])
            return result.rowcount
    except SQLAlchemyError as exc:
        logger.error("persist failed chunk=%s: %s", items[0]["chunk_id"], exc)
        raise self.retry(exc=exc, countdown=30)

Tasks that exhaust their retry budget must be captured, not dropped. A task_failure signal handler routes the poisoned payload to a dead-letter table for manual reconciliation and automated replay:

from celery.signals import task_failure


@task_failure.connect
def route_to_dead_letter(sender=None, task_id=None, exception=None, args=None, **_):
    logger.critical("task=%s permanently failed -> DLQ: %s", task_id, exception)
    with engine.begin() as conn:
        conn.execute(
            text("INSERT INTO billing_dlq (task_id, task_name, payload, error) "
                 "VALUES (:tid, :name, :payload, :err) ON CONFLICT (task_id) DO NOTHING"),
            {"tid": task_id, "name": getattr(sender, "name", "?"),
             "payload": str(args), "err": str(exception)},
        )


if __name__ == "__main__":
    # Local dispatch example: fetch -> normalize -> persist as a chained signature.
    from celery import chain
    raw = fetch_billing_chunk.s(
        provider="aws",
        endpoint_url="https://example.test/cur/chunk/0",
        chunk_id="2026-06-aws-0",
        auth_headers={"Authorization": "Bearer <token>"},
    )
    chain(raw | persist_billing_chunk.s()).apply_async(queue="acquire")

# Verification & Testing

Confirm fault tolerance with three deterministic checks before trusting the pipeline against real invoices:

• Idempotency assertion. Dispatch the same normalized chunk twice and assert the row count is stable: SELECT count(*) FROM billing_line_items WHERE chunk_id = :id must return the same value after the second run, proving ON CONFLICT DO NOTHING absorbed the redelivery.
• Crash-resume drill. Send SIGKILL to a worker mid-fetch and confirm the broker redelivers the task (because task_reject_on_worker_lost=True) and that the durable cursor resumes from the committed offset rather than restarting the pull.
• Reconciliation tolerance. Run a nightly audit task that sums unblended_cost per billing_period and compares it to the provider manifest total; alert if the absolute difference exceeds a 0.1% tolerance. A persistent shortfall localized to one account points at a silently dead-lettered task.

In CI, mock the provider with a server that returns a 429 plus Retry-After: 1 on the first call and 200 on the second, then assert fetch_billing_chunk.retries == 1 and the final payload is non-empty — this proves the backoff branch fires without burning the full retry budget.

# Common Pitfalls Checklist

• Injecting auth_headers into Celery config or task results. Tokens land in the result backend in plaintext — fetch them from Vault or AWS Secrets Manager at dispatch time and keep them out of serialized state.
• Leaving worker_prefetch_multiplier at the default 4. Throttled fetches hoard reserved tasks; pin it to 1 for I/O-bound, rate-limited billing work.
• Deriving the idempotency key from mutable fields. Hash only stable identifiers (provider, account, period, chunk, row index) — never unblended_cost or a timestamp, or restated amounts create phantom duplicates.
• Catching the retry exception. self.retry() raises Retry; a bare except Exception around it swallows the signal and silently drops the task.
• Running fetch, normalize, and persist on one queue. A slow Postgres write then backpressures acquisition — keep each phase on its own queue with independent concurrency.

# Frequently Asked Questions

# Related

• Async Billing Data Processing Patterns — the parent reference whose in-process asyncio worker defines the extraction contract this Celery build distributes.
• Cloud Billing Data Ingestion & Parsing — the four-stage acquisition-to-persistence pipeline and idempotent ingestion model implemented here.
• Handling Billing API Rate Limits & Retries — the per-provider retry matrices and backoff mathematics behind the fetch task’s countdown logic.
• Azure Cost API Pagination and Deduplication — resolves the same double-counting class at the pagination layer that the idempotency key resolves at the sink.
• Incremental Sync Strategies for GCP Billing Exports — the durable watermark-resumption pattern that mirrors this pipeline’s cursor-state registry.

Up: Async Billing Data Processing Patterns · Discipline root: Cloud Cost Optimization & FinOps Automation