Async Billing Data Processing Patterns

Asynchronous billing processing is the decoupling layer that sits between the acquisition and normalization stages of the Cloud Billing Data Ingestion & Parsing pipeline. Its job is narrow but unforgiving: pull cost artifacts out of provider export channels that are slow, rate-limited, and eventually consistent, then hand bounded, normalized chunks to downstream consumers without blocking, double-counting, or running a worker out of memory. A synchronous extract-transform-load script collapses against these constraints — a single 12 GB AWS Cost and Usage Report (CUR) decompresses past a pod’s heap, an Azure 429 Too Many Requests mid-loop loses an uncommitted cursor, and a long BigQuery scan stalls the entire job. This page covers the queue topology, the cursor-state contract, the rate-limit-aware client, and the idempotent staging model that turn fragile extraction jobs into horizontally scalable workers. It leads with the architecture constraint, ships a runnable asyncio worker, and closes with the failure modes that take async pipelines down in production.

# Architecture Context & Data-Flow Position

Within the four-stage pipeline — acquisition, normalization, allocation, persistence — asynchronous processing wraps acquisition and feeds normalization. It owns three responsibilities the synchronous path cannot satisfy: it lets the slow, rate-limited acquisition stage scale independently of the CPU-bound transform; it provides a durable boundary where a crashed worker resumes from a committed cursor rather than restarting a multi-gigabyte pull; and it absorbs provider-specific behavior behind a uniform task contract so a GCP schema migration or an Azure throttling spike never starves AWS extraction.

The async layer must absorb three provider realities that directly dictate the worker design. The table maps each export channel to the mechanism a worker has to implement before any normalization happens.

The unifying abstraction is a task graph. Each task encodes a billing period, an account or scope, and a data type (usage, amortized, or reservation) as a routing key. A worker pulls a task, initializes a provider client under a token bucket, walks the cursor, writes bounded chunks to staging, and commits the cursor only after the write succeeds. Provider-specific extraction detail lives in the sibling references — the manifest-driven build in AWS CUR to Data Lake Pipeline, the partition-pruned incremental query in GCP BigQuery Billing Export Sync, and the scope-targeted cursor walk in Azure Cost Management API Integration — while the async layer described here owns only the orchestration contract that binds them.

# Core Implementation Patterns

# 1. Queue Topology & Worker Isolation

Deploy a message broker — Redis Streams, RabbitMQ, or AWS SQS — with a dedicated queue per provider, and bind each queue to a worker pool that holds only that provider’s credentials. Provider isolation has two payoffs: it prevents cross-cloud credential leakage (an AWS worker never sees a GCP service-account key), and it lets you set per-provider concurrency limits that match each API’s quota. Route tasks with a key that encodes period.account.datatype, so a reservation-amortization job and a raw-usage job for the same account run on separate consumers and fail independently. Isolation is what guarantees that an Azure throttling incident drains only the Azure pool while AWS extraction continues at full rate.

# Routing key encodes the unit of idempotent work: one cursor lives per key.
routing_key = f"{provider}.{account_id}.{billing_period}.{data_type}"
# e.g. "azure.sub-0a91.2026-06.usage"  ->  consumed only by the Azure pool

# 2. Rate-Limit-Aware Client Initialization

Provider SDK defaults are not safe under sustained extraction. Instantiate each client with explicit connect/read timeouts and a worker-level limiter that reads the response headers the provider actually sends: Retry-After (Azure, seconds), x-ratelimit-remaining, and adaptive SlowDown/503 signals (AWS S3). A token-bucket limiter at the worker, rather than blind SDK retries, keeps you under the per-account ceiling — Azure Cost Management allows roughly 1 query/sec per scope with bursts, and the AWS Cost Explorer GetCostAndUsage API defaults to 5 requests/second. Aggressive retries without a limiter trigger exponential lockouts; the full per-provider matrix and Retry-After parsing live in Handling Billing API Rate Limits & Retries.

# 3. Cursor-Based Pagination & Memory-Bounded Chunking

Replace offset pagination with cursor or timestamp iteration — offsets re-scan and drift when the provider restates rows mid-walk. Yield records in bounded chunks (10k rows or ~50 MB, whichever comes first) and flush each chunk to object storage immediately so the heap stays flat on a multi-gigabyte report. Persist the last successful cursor to a durable store (PostgreSQL, DynamoDB, or Redis) before acknowledging the broker message. That ordering — write chunk, commit cursor, then ack — is what delivers at-least-once delivery with exactly-once effect, because a redelivered task resumes from the committed cursor and the idempotent staging key absorbs the overlap.

# 4. Idempotent Transformation & Staging

Normalize each chunk into the canonical cost record before persistence, and stamp every row with a deterministic fingerprint over its composite key (account_id, resource_id, usage_start, meter, cost). Stage chunks in a write-optimized columnar format (Parquet or Avro) partitioned by billing period. Idempotency at the row level is non-negotiable: duplicate broker deliveries, overlapping daily partitions in BigQuery exports, and provider restatements all re-present data the pipeline has already seen, and a fingerprint-keyed upsert turns each of those into a no-op instead of inflated showback. This is the same idempotent ingestion contract formalized in the parent Cloud Billing Data Ingestion & Parsing reference, applied at async chunk granularity.

# Production-Grade Python Ingestion Engine

The module below is a self-contained async worker. It defines a token-bucket RateLimiter that adapts to provider headers, a CursorState dataclass persisted per routing key, a retry decorator with exponential backoff, structured logging, and a process_account loop that flushes memory-bounded chunks and commits cursors only after a successful staging write. Replace fetch_chunk with aiobotocore (AWS), google-cloud-bigquery run in an executor (GCP), or azure-mgmt-costmanagement (Azure) through the same adapter signature. Dependencies: standard library only for the reference; add the provider SDKs in production.

import asyncio
import functools
import hashlib
import logging
import time
from dataclasses import dataclass, field
from typing import Awaitable, Callable, Dict, List, Optional, Tuple

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("billing.async_worker")


def async_retry(
    attempts: int = 5,
    base_delay: float = 2.0,
    max_delay: float = 30.0,
    exceptions: Tuple[type, ...] = (ConnectionError, TimeoutError),
) -> Callable:
    """Exponential-backoff retry for async provider calls.

    Backoff is capped so a throttled endpoint never blocks a worker
    unboundedly; the final attempt re-raises so the task is dead-lettered
    rather than silently dropped.
    """

    def decorator(func: Callable[..., Awaitable]) -> Callable[..., Awaitable]:
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            last_exc: Optional[BaseException] = None
            for attempt in range(1, attempts + 1):
                try:
                    return await func(*args, **kwargs)
                except exceptions as exc:  # noqa: PERF203
                    last_exc = exc
                    delay = min(base_delay * 2 ** (attempt - 1), max_delay)
                    logger.warning(
                        "retry %d/%d after %s: %.1fs backoff",
                        attempt, attempts, type(exc).__name__, delay,
                    )
                    await asyncio.sleep(delay)
            assert last_exc is not None
            raise last_exc

        return wrapper

    return decorator


@dataclass
class RateLimiter:
    """Token-bucket limiter that adapts to provider throttling headers."""

    capacity: int = 50
    tokens: float = 50.0
    refill_rate: float = 5.0  # tokens per second; align to provider quota
    last_refill: float = field(default_factory=time.monotonic)

    async def acquire(self) -> None:
        now = time.monotonic()
        self.tokens = min(self.capacity, self.tokens + (now - self.last_refill) * self.refill_rate)
        self.last_refill = now
        if self.tokens < 1.0:
            await asyncio.sleep((1.0 - self.tokens) / self.refill_rate)
            self.tokens = 1.0
        self.tokens -= 1.0

    def update_from_headers(self, headers: Dict[str, str]) -> None:
        # Azure sends Retry-After in seconds; back the bucket off to 1 token / window.
        if "retry-after" in headers:
            self.refill_rate = max(0.1, 1.0 / max(float(headers["retry-after"]), 0.1))
        # Drain the bucket when the provider says zero requests remain.
        if headers.get("x-ratelimit-remaining") == "0":
            self.tokens = 0.0


@dataclass
class CursorState:
    """Durable per-task position. Persisted under the routing key."""

    account_id: str
    billing_period: str
    last_cursor: Optional[str] = None
    chunk_size: int = 10_000
    rows_committed: int = 0

    @property
    def key(self) -> str:
        return f"{self.account_id}:{self.billing_period}"


def fingerprint(row: Dict) -> str:
    """Row-level dedup key; identical line items across replays collapse."""
    basis = "|".join(
        str(row.get(k, "")) for k in ("account_id", "resource_id", "usage_start", "meter", "cost")
    )
    return hashlib.sha256(basis.encode("utf-8")).hexdigest()


class StagingStore:
    """Reference idempotent sink. Swap for S3/GCS multipart or a Parquet writer."""

    def __init__(self) -> None:
        self._seen: set = set()
        self.rows_written = 0

    async def write_chunk(self, rows: List[Dict], cursor: Optional[str]) -> int:
        await asyncio.sleep(0.01)  # simulate object-store latency
        fresh = [r for r in rows if fingerprint(r) not in self._seen]
        self._seen.update(fingerprint(r) for r in fresh)
        self.rows_written += len(fresh)
        logger.info("staged %d new / %d total rows | cursor=%s", len(fresh), len(rows), cursor)
        return len(fresh)


class CursorRegistry:
    """In-memory reference; replace with DynamoDB/Postgres for distributed locking."""

    def __init__(self) -> None:
        self._store: Dict[str, CursorState] = {}

    def load(self, state: CursorState) -> CursorState:
        return self._store.get(state.key, state)

    def commit(self, state: CursorState) -> None:
        self._store[state.key] = state


class AsyncBillingWorker:
    def __init__(self, registry: CursorRegistry, staging: StagingStore, limiter: RateLimiter):
        self.registry = registry
        self.staging = staging
        self.limiter = limiter

    @async_retry(attempts=5)
    async def fetch_chunk(
        self, state: CursorState, cursor: Optional[str]
    ) -> Tuple[List[Dict], Optional[str], Dict[str, str]]:
        """Provider adapter seam. Replace the body with aiobotocore / bigquery / azure."""
        await self.limiter.acquire()
        await asyncio.sleep(0.05)  # simulate network latency
        rows = [
            {
                "account_id": state.account_id,
                "resource_id": f"res-{i}",
                "usage_start": f"{state.billing_period}-01T00:00:00Z",
                "meter": "compute",
                "cost": round(i * 0.0137, 6),
            }
            for i in range(state.chunk_size)
        ]
        next_cursor = f"tok-{int(time.time() * 1000)}" if cursor != "FINAL" else None
        headers = {"x-ratelimit-remaining": "44", "retry-after": "1.0"}
        self.limiter.update_from_headers(headers)
        return rows, next_cursor, headers

    async def process_account(self, seed: CursorState) -> None:
        """Extraction loop: fetch -> stage -> commit cursor, resumable on crash."""
        state = self.registry.load(seed)
        cursor = state.last_cursor
        logger.info("start extract key=%s resume_cursor=%s", state.key, cursor)
        pages = 0
        while True:
            rows, next_cursor, _ = await self.fetch_chunk(state, cursor)
            written = await self.staging.write_chunk(rows, cursor)
            # Commit cursor ONLY after the staging write succeeds. A crash before
            # this line replays the page; the staging fingerprint absorbs the overlap.
            state.last_cursor = next_cursor
            state.rows_committed += written
            self.registry.commit(state)
            pages += 1
            if next_cursor is None:
                logger.info("done key=%s pages=%d rows=%d", state.key, pages, state.rows_committed)
                return
            cursor = "FINAL" if pages >= 2 else next_cursor  # demo termination guard


async def run_pipeline() -> None:
    registry = CursorRegistry()
    staging = StagingStore()
    limiter = RateLimiter(capacity=100, tokens=100.0, refill_rate=10.0)
    worker = AsyncBillingWorker(registry, staging, limiter)

    # Isolate accounts into independent tasks; asyncio.gather with return_exceptions
    # so one failing scope dead-letters without aborting the others.
    tasks = [
        worker.process_account(CursorState(account_id="acc-123", billing_period="2026-06")),
        worker.process_account(CursorState(account_id="acc-456", billing_period="2026-06")),
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for result in results:
        if isinstance(result, Exception):
            logger.error("task dead-lettered: %s", result)


if __name__ == "__main__":
    asyncio.run(run_pipeline())

The decisions that matter most: the cursor is committed after the staging write, never before, so an at-least-once broker plus a fingerprint-keyed sink yields exactly-once effect; the RateLimiter is the single place backoff pressure is applied, adapting to Retry-After rather than relying on SDK defaults; and asyncio.gather(..., return_exceptions=True) isolates per-account failures so one dead-lettered scope never aborts the batch. When you outgrow a single process, the same contract maps directly onto distributed task routing and result backends in Building Fault-Tolerant Billing Ingestion with Celery.

# Schema Reference Table

Each provider’s raw fields converge on the canonical record the staging store persists. The async worker performs this mapping inside its _normalize adapter before writing, so downstream allocation never sees provider-specific shapes.

# Operational Considerations

• Provider rate limits (exact numbers). AWS Cost Explorer GetCostAndUsage defaults to 5 requests/second per account; S3 GetObject tolerates high throughput but returns adaptive 503 SlowDown under burst. GCP BigQuery enforces query concurrency and a per-project bytes-billed budget — always set maximum_bytes_billed. Azure Cost Management is roughly 1 query/second per scope and returns 429 with a Retry-After header you must honor. Size each worker pool’s token bucket to the matching ceiling.
• Eventual-consistency windows. AWS finalizes 24–48 h after usage, GCP up to 24 h, Azure 24–72 h. Treat the trailing days as provisional and re-run the affected periods; idempotent staging makes re-ingestion safe.
• Daily partition overlap. GCP billing exports re-emit overlapping rows across daily partitions; the fingerprint upsert is what prevents double-counting on incremental sync.
• Currency normalization. Store the native currency and amount, convert at query time using a dated FX table, and retain both so historical totals stay reproducible.
• Monitoring hooks. Instrument workers with OpenTelemetry spans for fetch latency, chunk size, retry count, and staging throughput, and emit custom metrics for queue depth, rate-limit hits, and cursor age. A rising cursor age with flat rows_committed is the earliest signal a worker is wedged behind throttling.
• Backpressure. Watch staging IOPS and normalization latency; when either saturates, pause consumption (stop pulling from the broker) rather than buffering in memory. Backpressure propagating upstream is what prevents heap exhaustion during a provider slowdown.

# Troubleshooting

1. Worker OOM-killed mid-extraction. Root cause: a chunk buffer accumulating an entire decompressed CUR (10 GB+) before the first flush. Detection: container memory metric climbing monotonically until the OOM-killer fires; logs stop without a “staged” line. Remediation: enforce the byte-bounded flush — write each chunk as soon as it reaches chunk_size or ~50 MB and clear the buffer; for AWS, stream Parquet via pyarrow record batches rather than loading the object.

2. Duplicate costs in showback after a redeploy. Root cause: the broker redelivered an unacknowledged task and the cursor was committed before the staging write, so the page replayed into a non-idempotent sink. Detection: reconciliation total exceeds the provider invoice by a clean multiple of one chunk. Remediation: enforce write-then-commit ordering and a fingerprint-keyed upsert (INSERT ... ON CONFLICT DO NOTHING); backfill by replaying the affected period, which the dedup key now absorbs.

3. Azure extraction stalls and never completes. Root cause: 429 responses retried without honoring Retry-After, tripping a longer provider lockout. Detection: repeated retry ... 429 log lines with no forward cursor progress; x-ratelimit-remaining pinned at 0. Remediation: feed Retry-After into the limiter (update_from_headers) so the token bucket drains and refills at the provider’s stated rate instead of hammering the endpoint.

4. Cursor lost across a pod restart. Root cause: cursor state held in process memory rather than a durable registry, so a restart restarts the pull from zero. Detection: full re-extraction after every deploy; staging throughput spikes with near-zero net-new rows. Remediation: persist CursorState to DynamoDB or Postgres keyed by account_id:billing_period, and load() it at the top of process_account so the worker resumes from the committed boundary.

5. Silent gaps in a billing period. Root cause: a task that raised a non-retryable exception was dropped without dead-lettering, leaving its cursor range unprocessed. Detection: a reconciliation shortfall isolated to one account/period; no error surfaced to alerting. Remediation: route exhausted-retry tasks to a dead-letter queue that retains the raw payload and alerts FinOps engineers, and run a post-ingestion reconciliation that compares async totals against the provider invoice with a 0.1% tolerance.

# Frequently Asked Questions

# Related

• Cloud Billing Data Ingestion & Parsing — the parent reference defining the four-stage pipeline and idempotent ingestion model this async layer implements.
• AWS CUR to Data Lake Pipeline — manifest-driven S3 acquisition that feeds the AWS extraction queue described here.
• GCP BigQuery Billing Export Sync — partition-pruned incremental sync and the daily-overlap dedup the async dedup key resolves.
• Azure Cost Management API Integration — nextLink cursor walking and Retry-After handling for the Azure worker pool.
• Handling Billing API Rate Limits & Retries — the per-provider retry matrices behind the shared backoff decorator and token bucket.
• Building Fault-Tolerant Billing Ingestion with Celery — distributed task routing and result backends when a single async process is no longer enough.

Up: Cloud Billing Data Ingestion & Parsing · Discipline root: Cloud Cost Optimization & FinOps Automation