Transit Fare Revenue Reconciliation Automation

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. This page solves one concrete task for the Python automation builder: pull a full day of tap events out of a throttling legacy gateway without losing a single record, using request pacing, stateful checkpointing, and defensive parsing.

This is the resilient-fetch problem inside AFC API Data Extraction — the extraction hop of the broader Fare Data Ingestion & GTFS-RT Sync pipeline. Where modern cursor-paginated endpoints let the extractor stream freely, a legacy gateway forces the extractor to negotiate a rate ceiling it was never told about. Every dropped or duplicated page here becomes a farebox-recovery variance the revenue analyst has to explain at month-end close, so pacing and checkpointing are treated as correctness requirements, not performance tuning.

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 — a boundary owned upstream by AFC System Security Boundaries.

The practical consequence is that the extractor must self-govern. It cannot rely on the vendor to describe its own limit, so it paces every request against a conservative target rate, treats any 429/5xx as a signal to back off rather than an error to surface, and persists a durable cursor so a mid-run block never forces a full re-pull. Flat-file exporters that sidestep the API entirely are handled instead by CSV Batch Parsing Workflows; this page assumes you are stuck with the HTTP gateway.

Decision Flow: Pacing, Backoff, and Escalation

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

Paced fetch decision flow: pacing gate, response branch, backoff loop, and checkpoint A single fetch acquires the concurrency semaphore, passes through a token-bucket pacing sleep of one over the target request rate, then issues a GET for a fare batch. The vendor response branches three ways. A 200 with a valid schema advances the checkpoint cursor and saves the checkpoint file. A 429 sleeps for the Retry-After hint and then backs off. A 5xx, timeout, or truncated payload backs off with jitter. Both backoff paths reach an attempts-less-than-five check: while attempts remain the flow loops back to the pacing gate so every retry is re-paced, and when they are exhausted the cycle raises and halts. 200 + valid schema 429 5xx · timeout · truncated attempts remain · re-pace every retry no Acquire semaphore Token-bucket pacing sleep 1 / rps GET fare batch Advance checkpoint cursor Save checkpoint file Sleep Retry-After, then back off Exponential backoff + jitter Attempts < 5 ? Raise · halt cycle
The pacing gate sits inside the retry loop: every backoff re-enters token-bucket pacing before the next GET, so a burst of 429s can never compound into a faster second burst.

The single most important property of this flow is that the pacing sleep sits inside the retry loop, not before it. Every retry re-enters the token-bucket gate, so a burst of 429s cannot compound into a second, faster burst that gets the source IP blocked outright.

Step-by-Step Implementation

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 implementation below uses a semaphore-controlled async client, structured audit logging, and persistent checkpointing so a run that dies halfway through resumes from the last confirmed cursor instead of re-pulling — and re-billing — the whole day.

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 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}
    ))

The key design choices, from top to bottom: _enforce_pacing serializes the pacing sleep under a lock so concurrent tasks cannot all fire in the same millisecond; the tenacity decorator only retries on the transient exception types (throttle and timeout), so a genuine 4xx schema/auth error fails fast instead of hammering the gateway five times; and the 429 branch honours the vendor’s Retry-After before raising, so the exponential backoff stacks on top of the vendor’s own hint rather than ignoring it.

Validation & Test Cases

Because the gateway is legacy and undocumented, you cannot trust it in an integration test — you assert against a mocked transport that reproduces its worst behaviours. The cases below cover one normal path and the two failure modes that most often corrupt reconciliation.

import asyncio
from unittest.mock import AsyncMock, MagicMock
import aiohttp
import pytest

def _mock_response(status: int, json_body=None, headers=None):
    resp = MagicMock()
    resp.status = status
    resp.headers = headers or {}
    resp.request_info = MagicMock()
    resp.history = ()
    resp.json = AsyncMock(return_value=json_body)
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=resp)
    cm.__aexit__ = AsyncMock(return_value=False)
    return cm

@pytest.mark.asyncio
async def test_success_advances_cursor(tmp_path):
    """Normal case: a 200 with a well-formed page returns data and records success."""
    client = AFCRateLimitClient("https://gw.test/v1", target_rps=1000.0)
    client.checkpoint.checkpoint_file = tmp_path / "cp.json"
    session = MagicMock()
    session.get = MagicMock(return_value=_mock_response(
        200, {"data": [{"tap_id": "T1"}, {"tap_id": "T2"}],
              "metadata": {"next_cursor": "cur_002"}}))

    payload = await client.fetch_fare_batch(session, "/fare-events", {"limit": 500})

    assert len(payload["data"]) == 2
    assert client.checkpoint.request_count == 1
    assert client.checkpoint.last_success is not None

@pytest.mark.asyncio
async def test_429_then_success_is_transparent(tmp_path):
    """Edge case: a 429 is retried and the caller never sees the throttle."""
    client = AFCRateLimitClient("https://gw.test/v1", target_rps=1000.0)
    client.checkpoint.checkpoint_file = tmp_path / "cp.json"
    session = MagicMock()
    session.get = MagicMock(side_effect=[
        _mock_response(429, headers={"Retry-After": "0"}),
        _mock_response(200, {"data": [{"tap_id": "T9"}], "metadata": {}}),
    ])

    payload = await client.fetch_fare_batch(session, "/fare-events", {"limit": 500})
    assert payload["data"][0]["tap_id"] == "T9"

@pytest.mark.asyncio
async def test_truncated_payload_is_rejected(tmp_path):
    """Error case: a 200 missing the 'data' key is treated as truncation, not success."""
    client = AFCRateLimitClient("https://gw.test/v1", target_rps=1000.0)
    client.checkpoint.checkpoint_file = tmp_path / "cp.json"
    session = MagicMock()
    session.get = MagicMock(return_value=_mock_response(200, {"status": "ok"}))

    with pytest.raises(ValueError, match="Unexpected payload schema"):
        await client.fetch_fare_batch(session, "/fare-events", {"limit": 500})
    assert client.checkpoint.request_count == 0  # never counted as a success

Expected results:

Test Input Expected outcome
test_success_advances_cursor 200 with a two-record data array Returns both records; request_count == 1; last_success set
test_429_then_success_is_transparent 429 (Retry-After: 0) then 200 Retry is invisible to caller; final tap T9 returned
test_truncated_payload_is_rejected 200 whose body lacks data Raises ValueError; request_count stays 0

The third case is the one that matters most for revenue integrity: a legacy gateway that returns 200 OK with a half-written body is far more dangerous than one that returns 503, because a naive client counts it as a successful page and silently skips the missing taps.

Transit-Specific Debugging Steps

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

  1. 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.

  2. 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.

  3. 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.

  4. 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.

  5. 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 the official asyncio documentation for event loop scheduling and tenacity retry patterns for deterministic backoff tuning.

Integration Note

This task is the resilience layer of the parent AFC API Data Extraction component: once a page survives pacing and checkpointing, its raw records are still untrusted and must pass the boundary contract enforced by Schema Validation Pipelines before any of them can move money. In practice the data array returned by fetch_fare_batch is handed straight to the models described in Implementing Pydantic Models for AFC Event Streams, which reject malformed or duplicate taps the throttled gateway may have double-delivered on a retry. Field semantics — what fare_media_type and its media-specific columns actually mean — come from Smart Card Schema Mapping, so this extractor binds raw hardware fields to that taxonomy rather than inventing its own.

Production Readiness Checklist

  • target_rps
  • All PII is masked in AUDIT_LOG
  • A guard halts the cycle on repeated empty data

FAQ

The vendor sends no Retry-After header — what value should I back off by?
Fall back to the exponential schedule and ignore the missing hint. The code already defaults Retry-After to 5 seconds, but the real protection is wait_exponential(min=2, max=60) on the tenacity decorator, which grows the delay on every attempt regardless of headers. For a gateway that never sends the header, tune max upward (90–120s) rather than raising target_rps, since a silent limiter punishes bursts far harder than slow steady traffic.
Should I raise max_concurrent to pull the day faster?
Rarely. On a token-bucket gateway, concurrency past the bucket refill rate just produces more 429s and longer backoffs, so throughput goes down, not up. Keep max_concurrent at 2–3 and instead widen the query window per request (a larger limit) so each successful call returns more taps. Concurrency helps only when the limiter is per-connection rather than per-account, which you should confirm against the SLA before tuning.
How do I resume after a crash without re-pulling or skipping taps?
Persist the cursor only after a page validates, which is exactly what run_ingestion_cycle does — checkpoint.save() runs after fetch_fare_batch returns cleanly. On restart, CheckpointState.load() restores last_fetched_id and the next request continues from it. Because saving is downstream of validation, a crash mid-page re-pulls that one page rather than skipping it; downstream idempotency (unique transaction IDs) absorbs the duplicate.
A 200 response with an empty data array keeps arriving — is the pull finished?
Not necessarily. Legacy gateways sometimes emit empty pages under load instead of a proper end-of-stream cursor, so a single empty page is ambiguous. Treat end-of-data as a positive signal (an explicit next_cursor of null or a total-count match), and treat three consecutive empty pages as a fault that halts the cycle for investigation rather than a clean finish.

Part of AFC API Data Extraction.