Broadcast Traffic Scheduling Automation

Backpressure and Retry for High-Volume Traffic Uploads

When a station group pushes a night’s worth of traffic logs at the ingestion API — thousands of spot records across dozens of markets, all released at midnight — the naive uploader fires every request at once, exhausts the connection pool, and buries the Avstar endpoint under a thundering herd it answers with 429 Too Many Requests and 503. This guide solves one exact task: applying backpressure so the uploader never has more work in flight than the endpoint can absorb, and retrying transient failures with exponential backoff and jitter behind a circuit breaker that stops the batch when the endpoint is genuinely down. It is the resilience procedure beneath Async Batch Processing for High-Volume Logs, part of Avion & Avstar Ingestion Pipelines. It matters for audit because a dropped or double-posted upload corrupts the as-run record that billing reconciles against, and an un-throttled retry storm can trip the API’s own abuse protection and lock out the whole station group.

The core requirement is bounded concurrency with graceful degradation: slow down under load rather than fail, retry only what is safely retryable, and stop entirely before a partial batch corrupts downstream state. That is a different concern from raw throughput — the parent guide on optimizing asyncio for traffic file uploads maximizes how fast a healthy endpoint drains work; this page keeps the uploader correct when the endpoint is not healthy.

Prerequisites

  • Python 3.11+ — required for asyncio.TaskGroup, X | None union syntax, and enum.StrEnum.
  • httpx — pin httpx==0.27.0 for its async client and per-request timeout support; a synchronous client cannot express backpressure across concurrent uploads.
  • Pydantic v2 — pin pydantic==2.7.1 to validate each upload envelope before it is sent, so malformed records fail locally rather than wasting a retry budget.
  • A valid Avstar bearer token — obtained and kept fresh per Refreshing Avstar OAuth Tokens Automatically; an expired token turns every upload into a 401 the retry loop must not blindly hammer.
  • Idempotency keys on every upload — a stable per-record key so a retried request the server already applied is deduplicated, never double-posted.
  • Known endpoint limits — the published rate ceiling and burst allowance from Avstar API Authentication and Rate Limits, used to size the semaphore and queue below.

Step-by-Step Implementation

The uploader is built from three cooperating controls: a semaphore that caps in-flight requests, a bounded queue that makes producers wait when consumers fall behind (that is the backpressure), and a retry decorator with exponential backoff plus jitter, all guarded by a circuit breaker that halts the batch after sustained failure.

Backpressure and retry pipeline for traffic uploads Traffic records are enqueued by a producer into a bounded queue. When the queue is full the producer blocks, which is backpressure that slows ingestion to match capacity. A pool of worker consumers each acquire a slot from a concurrency semaphore before sending an upload through the retry layer. The retry layer classifies the response: a 2xx success acknowledges the record and resets the circuit breaker; a retryable 429 or 503 waits an exponential backoff interval with jitter and tries again up to a maximum; a permanent 4xx failure is dead-lettered. Repeated failures feed the circuit breaker, which opens after a threshold and halts the whole batch with no further sends. Producer enqueue records blocks if full Bounded queue maxsize=N Semaphore cap in-flight worker worker status? retry layer Ack + reset breaker 2xx Backoff + jitter 429 / 503 · retry Dead-letter / OPEN 4xx or threshold ok retry fail

Figure — Backpressure pipeline: a bounded queue slows the producer to match capacity, a semaphore caps in-flight uploads, and the retry layer acknowledges successes, backs off on 429/503, and dead-letters or trips the breaker on hard failure.

Step 1 — Structured logging and the upload envelope

Goal: emit greppable audit lines in the timestamp | level | module | spot_id shape and validate each upload before it consumes a retry budget. A malformed record should fail locally, never on the wire.

python 
from __future__ import annotations

import asyncio
import logging
import random
from enum import StrEnum

import httpx
from pydantic import BaseModel, Field

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

# 429/503 are transient (retry); 408/425 are transport hiccups worth one retry.
RETRYABLE_STATUS = frozenset({408, 425, 429, 500, 502, 503, 504})


class UploadStatus(StrEnum):
    ACKED = "acked"
    RETRYING = "retrying"
    DEAD_LETTERED = "dead_lettered"
    HALTED = "halted"          # circuit breaker open


class TrafficUpload(BaseModel):
    spot_id: str
    idempotency_key: str = Field(min_length=8)  # dedupes a server-applied retry
    payload: dict[str, str]

Step 2 — A retry-safe circuit breaker

Goal: protect the endpoint from a retry storm. After failure_threshold consecutive failures the breaker opens and no further uploads are sent; after recovery_timeout it admits one probe and closes on the first success. This is the same breaker discipline used across the ingestion and scheduling stacks.

python 
import time


class CircuitBreaker:
    """closed -> open -> half-open state machine that halts cascading failures."""

    def __init__(self, failure_threshold: int = 8, recovery_timeout: float = 30.0) -> None:
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.opened_at = 0.0
        self.state = "closed"

    def allow(self) -> bool:
        if self.state == "open":
            if time.monotonic() - self.opened_at >= self.recovery_timeout:
                self.state = "half-open"      # admit a single probe
                return True
            return False
        return True

    def record_success(self) -> None:
        self.failure_count = 0
        self.state = "closed"

    def record_failure(self) -> None:
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            self.opened_at = time.monotonic()
            logger.error("circuit breaker OPENED after %d failures", self.failure_count)

Step 3 — Exponential backoff with jitter

Goal: compute a wait that grows exponentially with the attempt number but is randomized so a fleet of uploaders does not retry in lockstep and re-create the herd it is recovering from. Full jitter — a random value between zero and the ceiling — decorrelates the retries.

python 
def backoff_delay(attempt: int, base: float = 0.5, cap: float = 20.0) -> float:
    # Exponential ceiling capped, then "full jitter" to decorrelate the fleet.
    ceiling = min(cap, base * (2 ** attempt))
    return random.uniform(0.0, ceiling)

For attempt 0, 1, 2, 3 the ceiling is 0.5, 1, 2, 4 seconds; the actual sleep is a random fraction of that, so two workers that failed together do not wake together.

Step 4 — The backpressured uploader

Goal: compose the controls. A bounded queue applies backpressure to the producer, a semaphore caps concurrency, and each worker uploads with retry, honouring a server Retry-After header when present.

python 
class BackpressuredUploader:
    def __init__(self, client: httpx.AsyncClient, url: str,
                 concurrency: int = 8, queue_size: int = 200,
                 max_attempts: int = 5) -> None:
        self.client = client
        self.url = url
        self.sem = asyncio.Semaphore(concurrency)   # cap in-flight requests
        self.queue: asyncio.Queue[TrafficUpload] = asyncio.Queue(maxsize=queue_size)
        self.max_attempts = max_attempts
        self.breaker = CircuitBreaker()

    async def _send_once(self, item: TrafficUpload) -> httpx.Response:
        # Idempotency-Key lets the server dedupe a retry it already applied.
        return await self.client.post(
            self.url, json=item.payload, timeout=10.0,
            headers={"Idempotency-Key": item.idempotency_key},
        )

    async def _upload(self, item: TrafficUpload) -> UploadStatus:
        for attempt in range(self.max_attempts):
            if not self.breaker.allow():
                logger.critical("%s | breaker open, halting upload", item.spot_id)
                return UploadStatus.HALTED
            async with self.sem:                     # bounded concurrency
                try:
                    resp = await self._send_once(item)
                except (httpx.TimeoutException, httpx.TransportError) as exc:
                    self.breaker.record_failure()
                    logger.warning("%s | transport error attempt=%d: %s",
                                   item.spot_id, attempt, exc)
                    await asyncio.sleep(backoff_delay(attempt))
                    continue
            if resp.status_code < 300:
                self.breaker.record_success()
                logger.info("%s | acked status=%d", item.spot_id, resp.status_code)
                return UploadStatus.ACKED
            if resp.status_code in RETRYABLE_STATUS:
                self.breaker.record_failure()
                # Respect an explicit Retry-After; else exponential + jitter.
                wait = float(resp.headers.get("Retry-After", backoff_delay(attempt)))
                logger.warning("%s | retryable status=%d wait=%.2fs",
                               item.spot_id, resp.status_code, wait)
                await asyncio.sleep(wait)
                continue
            logger.error("%s | permanent status=%d, dead-lettering",
                         item.spot_id, resp.status_code)
            return UploadStatus.DEAD_LETTERED
        logger.error("%s | exhausted %d attempts", item.spot_id, self.max_attempts)
        return UploadStatus.DEAD_LETTERED

    async def _worker(self, results: list[UploadStatus]) -> None:
        while True:
            item = await self.queue.get()
            try:
                results.append(await self._upload(item))
            finally:
                self.queue.task_done()

    async def run(self, items: list[TrafficUpload], workers: int = 8) -> list[UploadStatus]:
        results: list[UploadStatus] = []
        async with asyncio.TaskGroup() as tg:
            consumers = [tg.create_task(self._worker(results)) for _ in range(workers)]
            for item in items:
                await self.queue.put(item)           # blocks when queue is full
            await self.queue.join()                  # drain before cancelling
            for c in consumers:
                c.cancel()
        return results

A representative log line reads 2026-07-17T00:03:11+00:00 | WARNING | traffic.ingest.upload_backpressure | SP-2026-0717-0042 | retryable status=429 wait=1.30s.

Verification & Testing

Correctness rests on two properties: backpressure (the producer cannot outrun the workers) and bounded retry (a transient failure is retried, a permanent one is not). Both are assertable with a stub transport that scripts responses.

python 
import httpx

# Script: first call 503 (retry), second 200 (ack).
seq = iter([httpx.Response(503), httpx.Response(200, json={"ok": True})])
transport = httpx.MockTransport(lambda req: next(seq))

async def _test() -> None:
    async with httpx.AsyncClient(transport=transport) as client:
        up = BackpressuredUploader(client, "https://avstar.example/v1/logs",
                                   concurrency=2, queue_size=4)
        item = TrafficUpload(spot_id="SP-1", idempotency_key="abc12345",
                             payload={"house_number": "WXYZ-0000482"})
        results = await up.run([item], workers=1)
    assert results == [UploadStatus.ACKED]           # recovered after one retry

asyncio.run(_test())

# Backpressure: a queue of maxsize 1 cannot hold 3 items at once.
q: asyncio.Queue[int] = asyncio.Queue(maxsize=1)
assert q.maxsize == 1                                 # producer must wait on put()

Run this against a golden fixture of scripted status sequences so a change to RETRYABLE_STATUS or the backoff schedule that would silently drop or storm the endpoint fails the build. Assert that a 429 is retried and a 422 is dead-lettered, not the reverse.

Edge Cases & Failure Handling

  • Retry-After that exceeds the batch window. A server may return a Retry-After of minutes during a maintenance window, longer than the nightly upload has to complete. Cap the honoured wait and, past the cap, dead-letter the record for the next run rather than blocking a worker for the whole window — the idempotency key means the deferred retry is safe.
  • Non-idempotent double-post. Under at-least-once retry, a request the server applied but whose response was lost will be retried. The Idempotency-Key header makes the second POST a no-op at the receiver; without it, a retried upload double-posts the spot and corrupts the as-run that as-run reconciliation later checks. Never retry a write that lacks a stable key.
  • Breaker flapping on a slow endpoint. An endpoint that is slow but not down can trip the breaker on timeouts, then immediately fail the half-open probe, oscillating. Raise recovery_timeout and the request timeout together so a probe has a fair chance, and confirm the token is fresh via handling Avstar session timeouts before blaming the endpoint.

FAQ

What is the difference between backpressure and rate limiting?

Rate limiting caps the number of requests per unit time; backpressure caps the amount of work in flight and pushes the slowdown back onto the producer when consumers fall behind. The bounded queue is the backpressure mechanism — a full queue blocks the producer’s put() — while the semaphore enforces the concurrency limit. Together they keep a burst from overwhelming the endpoint even before its published rate ceiling is hit, which complements the throughput tuning in optimizing asyncio for traffic file uploads.

Why add jitter instead of plain exponential backoff?

Because a fleet of uploaders that all failed at the same instant will, under plain exponential backoff, all retry at the same instant too, re-creating the thundering herd. Full jitter randomizes each wait between zero and the exponential ceiling, decorrelating the retries so the endpoint sees a smooth trickle instead of synchronized waves. It is the single most effective change you can make to a retry loop that talks to a shared rate-limited API.

Should I retry a 401 the same way as a 429?

No. A 429 is transient — back off and retry. A 401 means the token is invalid or expired, so retrying with the same token just burns the budget. Refresh the credential first, following Refreshing Avstar OAuth Tokens Automatically, then retry once. That is why 401 is deliberately absent from RETRYABLE_STATUS.

When does the circuit breaker help versus just retrying?

Retry handles a single transient blip; the breaker handles a sustained outage. Without a breaker, a down endpoint drives every record through its full retry budget, multiplying load exactly when the endpoint can least handle it. The breaker opens after a consecutive-failure threshold and halts the batch, so no partial upload corrupts downstream state — the same halt semantics the scheduler applies in Spot Scheduling Validation & Rule Engines.

Related content

Up one levelAsync Batch Processing for High-Volume Logs RelatedOptimizing Asyncio for Traffic File Uploads