Handling Rate Limits on Legacy AFC Vendor APIs

Legacy Automated Fare Collection (AFC) systems rarely expose modern, well-documented REST interfaces. Transit operators and revenue analysts frequently interact with SOAP gateways, paginated XML endpoints, or early-generation JSON APIs that enforce rigid, undocumented, or static rate limits. When these limits are breached, vendors return 429 Too Many Requests or 503 Service Unavailable responses, causing silent data drops, reconciliation drift, and broken downstream analytics. For mobility tech developers and Python automation builders, surviving these constraints requires a disciplined approach to request pacing, stateful checkpointing, and defensive parsing.

Understanding Vendor Throttling Mechanics

Legacy AFC vendors typically implement token-bucket or fixed-window rate limiting at the API gateway level. Unlike modern cloud APIs, they rarely expose X-RateLimit-Remaining or Retry-After headers in a consistent format. Throttling manifests as delayed responses, truncated payloads, or abrupt TCP resets. From a compliance standpoint, aggressive polling can trigger vendor-side IP blocks, violate data-sharing SLAs, or breach PCI-DSS logging requirements if raw request/response dumps are retained without PII masking.

To maintain operational continuity, ingestion pipelines must treat rate limits as a first-class architectural constraint rather than an edge case. This is especially critical when synchronizing fare validation events with Fare Data Ingestion & GTFS-RT Sync pipelines, where delayed tap data directly impacts vehicle occupancy models and passenger crowding dashboards.

Python Implementation: Async Throttling & Exponential Backoff

The most reliable mitigation strategy combines asyncio concurrency control with deterministic backoff logic. Synchronous requests loops will inevitably saturate vendor limits under high-volume conditions. The following implementation uses a semaphore-controlled async client, structured audit logging, and persistent checkpointing to guarantee exactly-once processing semantics.

The flow below shows how a single fetch passes through pacing, then branches on the vendor response into retry, escalation, or success:

flowchart TD A["Acquire semaphore"] --> B["Token-bucket pacing<br/>(sleep 1/rps)"] B --> C["GET fare batch"] C -->|"200 + valid schema"| D["Advance checkpoint<br/>cursor"] C -->|"429 Too Many Requests"| E["Sleep Retry-After<br/>+ exponential backoff"] C -->|"5xx / timeout"| F["Exponential backoff<br/>+ jitter"] C -->|"truncated payload"| F E --> G{"Attempts < 5?"} F --> G G -->|"yes"| B G -->|"no"| H["Raise / halt cycle"] D --> I["Save checkpoint file"]

import asyncio
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Optional

import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type, before_sleep_log

# ---------------------------------------------------------------------------
# Audit & Checkpoint Infrastructure
# ---------------------------------------------------------------------------
AUDIT_LOG = logging.getLogger("afc_reconciliation.audit")
AUDIT_LOG.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter(
    "%(asctime)s | %(levelname)s | %(message)s", datefmt="%Y-%m-%dT%H:%M:%S%z"
))
AUDIT_LOG.addHandler(_handler)

@dataclass
class CheckpointState:
    last_fetched_id: str = ""
    request_count: int = 0
    last_success: Optional[datetime] = None
    checkpoint_file: Path = Path("afc_reconciliation_checkpoint.json")

    def load(self) -> None:
        if self.checkpoint_file.exists():
            try:
                data = json.loads(self.checkpoint_file.read_text(encoding="utf-8"))
                self.last_fetched_id = data.get("last_fetched_id", "")
                self.request_count = data.get("request_count", 0)
                if data.get("last_success"):
                    self.last_success = datetime.fromisoformat(data["last_success"])
            except (json.JSONDecodeError, ValueError) as e:
                AUDIT_LOG.warning(f"Corrupt checkpoint file, resetting state: {e}")
                self.checkpoint_file.unlink(missing_ok=True)

    def save(self) -> None:
        payload = {
            "last_fetched_id": self.last_fetched_id,
            "request_count": self.request_count,
            "last_success": self.last_success.isoformat() if self.last_success else None
        }
        self.checkpoint_file.write_text(json.dumps(payload, indent=2), encoding="utf-8")

# ---------------------------------------------------------------------------
# Rate-Limited AFC Client
# ---------------------------------------------------------------------------
class AFCRateLimitClient:
    def __init__(self, base_url: str, max_concurrent: int = 3, target_rps: float = 8.0):
        self.base_url = base_url.rstrip("/")
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.token_interval = 1.0 / target_rps
        self._lock = asyncio.Lock()
        self.checkpoint = CheckpointState()
        self.checkpoint.load()

    async def _enforce_pacing(self) -> None:
        """Token-bucket approximation to prevent burst saturation."""
        async with self._lock:
            await asyncio.sleep(self.token_interval)

    @retry(
        wait=wait_exponential(multiplier=1.0, min=2, max=60),
        stop=stop_after_attempt(5),
        retry=retry_if_exception_type((aiohttp.ClientResponseError, asyncio.TimeoutError)),
        before_sleep=before_sleep_log(AUDIT_LOG, logging.WARNING)
    )
    async def fetch_fare_batch(self, session: aiohttp.ClientSession, endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]:
        async with self.semaphore:
            await self._enforce_pacing()
            url = f"{self.base_url}{endpoint}"
            
            try:
                async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                    if resp.status == 200:
                        try:
                            payload = await resp.json()
                        except aiohttp.ContentTypeError as e:
                            AUDIT_LOG.error(f"INVALID_CONTENT_TYPE | url={url} | error={e}")
                            raise aiohttp.ClientResponseError(request_info=resp.request_info, history=resp.history, status=resp.status, message="Malformed JSON/XML response")
                        
                        # Defensive check for legacy truncation
                        if not isinstance(payload, dict) or "data" not in payload:
                            raise ValueError("Unexpected payload schema from legacy AFC gateway")

                        self.checkpoint.request_count += 1
                        self.checkpoint.last_success = datetime.now(timezone.utc)
                        AUDIT_LOG.info(f"SUCCESS | endpoint={endpoint} | status=200 | records={len(payload.get('data', []))}")
                        return payload

                    elif resp.status == 429:
                        retry_after = int(resp.headers.get("Retry-After", 5))
                        AUDIT_LOG.warning(f"THROTTLED | endpoint={endpoint} | retry_after={retry_after}s")
                        await asyncio.sleep(retry_after)
                        raise aiohttp.ClientResponseError(
                            request_info=resp.request_info, history=resp.history, status=429, message="Rate limit exceeded"
                        )
                    elif resp.status >= 500:
                        AUDIT_LOG.error(f"VENDOR_SERVER_ERROR | endpoint={endpoint} | status={resp.status}")
                        raise aiohttp.ClientResponseError(
                            request_info=resp.request_info, history=resp.history, status=resp.status, message="Vendor server error"
                        )
                    else:
                        AUDIT_LOG.error(f"CLIENT_ERROR | endpoint={endpoint} | status={resp.status}")
                        resp.raise_for_status()
            except asyncio.TimeoutError as e:
                AUDIT_LOG.error(f"TIMEOUT | endpoint={endpoint} | error={e}")
                raise

# ---------------------------------------------------------------------------
# Execution Runner
# ---------------------------------------------------------------------------
async def run_ingestion_cycle(base_url: str, endpoint: str, params: Dict[str, Any]) -> None:
    client = AFCRateLimitClient(base_url=base_url, max_concurrent=3, target_rps=8.0)
    
    async with aiohttp.ClientSession() as session:
        try:
            data = await client.fetch_fare_batch(session, endpoint, params)
            client.checkpoint.last_fetched_id = data.get("metadata", {}).get("next_cursor", client.checkpoint.last_fetched_id)
            client.checkpoint.save()
            AUDIT_LOG.info(f"CYCLE_COMPLETE | checkpoint_saved={client.checkpoint.checkpoint_file}")
        except Exception as e:
            AUDIT_LOG.critical(f"INGESTION_FAILED | error={e}")
            raise

if __name__ == "__main__":
    # Example invocation aligned with [AFC API Data Extraction](/fare-data-ingestion-gtfs-rt-sync/afc-api-data-extraction/) workflows
    asyncio.run(run_ingestion_cycle(
        base_url="https://legacy-afc-gateway.transit-agency.gov/v1",
        endpoint="/fare-events/paginated",
        params={"since": "2024-01-01T00:00:00Z", "limit": 500}
    ))

Transit-Specific Debugging Steps

When reconciliation drift occurs despite rate-limit mitigation, isolate the failure vector using these targeted steps:

Correlate 429/503 Timestamps with Farebox Logs Cross-reference audit log THROTTLED or TIMEOUT entries with raw farebox validation timestamps. If gaps align with peak boarding windows (e.g., 07:00–09:00 local), reduce target_rps and increase max_concurrent semaphore limits to prevent queue starvation.
Validate Checkpoint Continuity Inspect afc_reconciliation_checkpoint.json for stale last_fetched_id values. Legacy gateways occasionally return 200 OK with empty data arrays instead of proper pagination cursors. Implement a guard clause to halt execution if len(payload["data"]) == 0 across three consecutive successful requests.
Detect Silent Payload Truncation Legacy AFC vendors frequently truncate XML/JSON responses mid-stream when connection pools exhaust. Enable aiohttp’s raise_for_status() and wrap JSON parsing in explicit try/except aiohttp.ContentTypeError. Compare Content-Length headers against parsed byte sizes to flag incomplete transfers before they corrupt downstream GTFS-RT feeds.
Sanitize Logs for PCI-DSS Compliance Ensure audit trails strip or hash PANs, card serial numbers, and exact tap coordinates before writing to disk. Use structured logging filters to mask sensitive fields while preserving endpoint, status, and retry_after values for vendor SLA reporting.
Align with Vendor SLA Windows Many legacy contracts restrict bulk data pulls to off-peak maintenance windows (e.g., 01:00–04:00). Schedule async runners using cron or systemd timers rather than continuous polling. Reference official asyncio documentation for event loop scheduling and tenacity retry patterns for deterministic backoff tuning.

Production Readiness Checklist

target_rps `target_rps` matches vendor-contracted limits (typically 5–10 RPS for legacy gateways)
Checkpoint file is backed up to immutable storage (S3/Glacier) before each cycle
Retry logic caps at 5 attempts with exponential backoff to prevent cascade failures
All PII is masked in AUDIT_LOG All PII is masked in `AUDIT_LOG` output before persistence
Pipeline integrates with downstream Fare Data Ingestion & GTFS-RT Sync Pipeline integrates with downstream [Fare Data Ingestion & GTFS-RT Sync](/fare-data-ingestion-gtfs-rt-sync/) validation hooks to flag occupancy anomalies