Incremental Sync for AFC REST APIs with Cursor-Based Pagination

This page builds a resumable incremental sync that pulls AFC vendor transactions over a REST API using cursor-based pagination and a persisted watermark, so every run resumes exactly where the last one stopped and is safe to retry without double-counting fares. It handles the two failure modes that break naive pagination loops in production: a cursor that expires mid-run, and pages that arrive out of order. It lives under AFC API Data Extraction within Fare Data Ingestion & GTFS-RT Sync, and it owns the checkpointing seam: extraction captures a page, this component decides what “already have it” and “resume from here” mean.

Incremental sync flow

The loop is a small state machine: load the last checkpoint, request a page from the cursor it holds, persist a new watermark only after the page is durably written, and repeat until the vendor stops handing back a next cursor. Two branches leave the happy path — an expired cursor rewinds to the last watermark, and a page whose records were already applied is acknowledged but not re-written. The flow below traces those transitions.

Step 1 — The checkpoint contract

A resumable sync is only as safe as its checkpoint. The checkpoint stores two things: the opaque cursor the vendor gave us to fetch the next page, and a watermark — the highest transaction timestamp we have durably persisted. The cursor is an optimization for the common resume; the watermark is the source of truth that lets us recover when the cursor is gone. Persisting the checkpoint is a small, explicit operation, not a side effect buried in the fetch loop.

import json
import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

logger = logging.getLogger("afc.sync")
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")


@dataclass(frozen=True)
class SyncCheckpoint:
    """Durable resume point for one AFC sync stream."""
    stream_id: str
    watermark: datetime          # highest tap_timestamp durably persisted
    cursor: Optional[str]        # opaque next-page token, may expire
    last_run_utc: datetime


class CheckpointStore:
    """Atomic file-backed checkpoint. Swap for a row in Postgres in production."""

    def __init__(self, path: Path) -> None:
        self._path = path

    def load(self, stream_id: str) -> Optional[SyncCheckpoint]:
        if not self._path.exists():
            return None
        try:
            raw = json.loads(self._path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as exc:
            logger.error("Checkpoint unreadable for %s: %s", stream_id, exc)
            raise
        return SyncCheckpoint(
            stream_id=raw["stream_id"],
            watermark=datetime.fromisoformat(raw["watermark"]),
            cursor=raw.get("cursor"),
            last_run_utc=datetime.fromisoformat(raw["last_run_utc"]),
        )

    def save(self, cp: SyncCheckpoint) -> None:
        # Write to a temp sibling then atomically replace, so a crash mid-write
        # never leaves a half-serialized checkpoint that would poison the resume.
        payload = asdict(cp)
        payload["watermark"] = cp.watermark.isoformat()
        payload["last_run_utc"] = cp.last_run_utc.isoformat()
        tmp = self._path.with_suffix(".tmp")
        tmp.write_text(json.dumps(payload), encoding="utf-8")
        tmp.replace(self._path)
        logger.info("Checkpoint saved: %s watermark=%s", cp.stream_id, cp.watermark.isoformat())

The atomic tmp.replace() matters: if the process is killed between writing and renaming, the previous good checkpoint survives intact. A watermark that got half-written would be worse than no checkpoint at all, because it would silently skip a slice of taps on the next run.

Step 2 — The resumable paginated client

The client wires the checkpoint into a fetch loop. Each iteration asks the transport for a page keyed by the current cursor, drops any record at or below the watermark, writes the rest, then advances both the watermark and the cursor. Cursor expiry is a first-class branch, not an exception that aborts the run: when the vendor signals the cursor is stale, the client rewinds to the watermark and re-derives a starting cursor from it.

from typing import Callable, Iterable, Protocol


class CursorExpired(Exception):
    """Raised by the transport when the vendor rejects a stale cursor."""


@dataclass(frozen=True)
class Page:
    records: list[dict]
    next_cursor: Optional[str]


class Transport(Protocol):
    def fetch(self, cursor: Optional[str]) -> Page: ...
    def cursor_from_watermark(self, watermark: datetime) -> Optional[str]: ...


class IncrementalSyncClient:
    """Cursor-paginated AFC sync with watermark recovery and idempotent writes."""

    def __init__(
        self,
        stream_id: str,
        transport: Transport,
        store: CheckpointStore,
        sink: Callable[[Iterable[dict]], None],
        max_expiry_rewinds: int = 3,
    ) -> None:
        self._stream_id = stream_id
        self._transport = transport
        self._store = store
        self._sink = sink
        self._max_expiry_rewinds = max_expiry_rewinds

    def _tap_time(self, record: dict) -> datetime:
        return datetime.fromisoformat(record["tap_timestamp"]).astimezone(timezone.utc)

    def run(self, cold_start: datetime) -> SyncCheckpoint:
        cp = self._store.load(self._stream_id)
        if cp is None:
            logger.info("Cold start for %s from %s", self._stream_id, cold_start.isoformat())
            cp = SyncCheckpoint(self._stream_id, cold_start, None, datetime.now(timezone.utc))

        watermark = cp.watermark
        cursor = cp.cursor
        rewinds = 0

        while True:
            try:
                page = self._transport.fetch(cursor)
            except CursorExpired:
                rewinds += 1
                if rewinds > self._max_expiry_rewinds:
                    logger.error("Cursor kept expiring for %s; aborting run", self._stream_id)
                    raise
                logger.warning("Cursor expired; rewinding to watermark %s", watermark.isoformat())
                cursor = self._transport.cursor_from_watermark(watermark)
                continue

            # Idempotent dedup: out-of-order or replayed rows at/below the
            # watermark are dropped so a retry never double-writes a fare.
            fresh = [r for r in page.records if self._tap_time(r) > watermark]
            if fresh:
                self._sink(fresh)
                watermark = max(self._tap_time(r) for r in fresh)

            cp = SyncCheckpoint(self._stream_id, watermark, page.next_cursor,
                                datetime.now(timezone.utc))
            self._store.save(cp)   # persist ONLY after the sink write succeeds

            if page.next_cursor is None:
                logger.info("Sync complete for %s at watermark %s",
                            self._stream_id, watermark.isoformat())
                return cp
            cursor = page.next_cursor

Two ordering rules make this safe under retry. First, the sink write happens before store.save, so a crash after writing but before checkpointing simply re-reads and re-dedups the same page — at-least-once delivery with idempotent application. Second, the watermark only ever advances past records we actually wrote, so a page that arrives out of order (older rows after newer ones) cannot pull the watermark backwards; those older rows are dropped as duplicates on the next pass.

Step 3 — Deriving a cursor from the watermark

Cursor expiry recovery depends on the vendor supporting a timestamp-seeded start. Most AFC back-office APIs expose an updated_since or from_ts parameter precisely so a client can bootstrap a cursor without one. The transport translates the watermark into that first request.

import httpx


class HttpTransport:
    """Thin AFC REST transport: cursor pages plus watermark-seeded restart."""

    def __init__(self, client: httpx.Client, base_url: str, page_size: int = 500) -> None:
        self._client = client
        self._base_url = base_url
        self._page_size = page_size

    def fetch(self, cursor: Optional[str]) -> Page:
        params = {"limit": self._page_size}
        if cursor is not None:
            params["cursor"] = cursor
        resp = self._client.get(self._base_url, params=params)
        if resp.status_code == 410:                 # Gone: vendor's expired-cursor signal
            raise CursorExpired(f"cursor rejected: {cursor}")
        resp.raise_for_status()
        body = resp.json()
        return Page(records=body["transactions"], next_cursor=body.get("next_cursor"))

    def cursor_from_watermark(self, watermark: datetime) -> Optional[str]:
        # Re-seed the stream from the last durable timestamp. Returning None here
        # tells fetch() to start from the server's from_ts anchor on the next call.
        resp = self._client.get(self._base_url,
                                params={"limit": self._page_size,
                                        "from_ts": watermark.isoformat()})
        resp.raise_for_status()
        # The re-seed response itself carries the first fresh page's cursor.
        return resp.json().get("next_cursor")

Rate-limit handling is deliberately out of scope here — fetch assumes each call succeeds or raises. Wrapping it with the backoff and Retry-After handling from Handling Rate Limits on Legacy AFC Vendor APIs is how this client survives a throttling vendor without corrupting the checkpoint.

Validation & Test Cases

A fake transport lets us prove the three properties that matter — a full multi-page sync, a resume from mid-point, and a duplicate page being ignored — without a live vendor.

from datetime import timedelta


class FakeTransport:
    """Serves fixed pages; can be told to expire the cursor once."""

    def __init__(self, pages: list[Page], expire_on: Optional[str] = None) -> None:
        self._pages = pages
        self._expire_on = expire_on
        self._expired_once = False

    def fetch(self, cursor: Optional[str]) -> Page:
        if cursor == self._expire_on and not self._expired_once:
            self._expired_once = True
            raise CursorExpired(cursor)
        idx = 0 if cursor is None else int(cursor)
        return self._pages[idx]

    def cursor_from_watermark(self, watermark: datetime) -> Optional[str]:
        return None  # restart from the first page in this fake


def _rec(ts: str, tx: str) -> dict:
    return {"tap_timestamp": ts, "transaction_id": tx, "amount_cents": 275}


base = datetime(2026, 7, 3, 8, 0, tzinfo=timezone.utc)
pages = [
    Page([_rec("2026-07-03T08:00:00+00:00", "T1")], "1"),
    Page([_rec("2026-07-03T08:05:00+00:00", "T2")], "2"),
    Page([_rec("2026-07-03T08:10:00+00:00", "T3")], None),
]

# Case 1: full three-page sync writes every record exactly once.
import tempfile
with tempfile.TemporaryDirectory() as d:
    written: list[dict] = []
    store = CheckpointStore(Path(d) / "cp.json")
    client = IncrementalSyncClient("s1", FakeTransport(pages), store, written.extend)
    final = client.run(cold_start=base - timedelta(hours=1))
    assert [r["transaction_id"] for r in written] == ["T1", "T2", "T3"]
    assert final.watermark == datetime(2026, 7, 3, 8, 10, tzinfo=timezone.utc)

# Case 2: resume from a mid-point checkpoint skips already-synced pages.
with tempfile.TemporaryDirectory() as d:
    written = []
    store = CheckpointStore(Path(d) / "cp.json")
    store.save(SyncCheckpoint("s1", datetime(2026, 7, 3, 8, 5, tzinfo=timezone.utc),
                              "2", datetime.now(timezone.utc)))
    client = IncrementalSyncClient("s1", FakeTransport(pages), store, written.extend)
    client.run(cold_start=base)
    assert [r["transaction_id"] for r in written] == ["T3"]   # T1, T2 already applied

# Case 3: a replayed page below the watermark is ignored (idempotent retry).
with tempfile.TemporaryDirectory() as d:
    written = []
    store = CheckpointStore(Path(d) / "cp.json")
    dup_pages = [Page([_rec("2026-07-03T08:00:00+00:00", "T1")], None)]
    store.save(SyncCheckpoint("s1", datetime(2026, 7, 3, 8, 0, tzinfo=timezone.utc),
                              None, datetime.now(timezone.utc)))
    client = IncrementalSyncClient("s1", FakeTransport(dup_pages), store, written.extend)
    client.run(cold_start=base)
    assert written == []   # record timestamp == watermark, dropped as duplicate

Case 1 asserts the cold run writes T1, T2, T3 once and lands the watermark at 08:10. Case 2 seeds a checkpoint at page 2 and proves only T3 is written — the resume skips what was already synced. Case 3 replays a page whose only record equals the watermark and asserts nothing is written, which is the idempotency guarantee a retried run depends on. To exercise expiry, construct FakeTransport(pages, expire_on="1") and confirm the run still completes with all three records after one rewind.

Edge Cases & Gotchas

1. Watermark ties. Two taps can share a timestamp to the second. A strictly-greater comparison (> watermark) drops the second one on resume. Key the watermark on a (timestamp, transaction_id) tuple, or store the last seen ID set for the boundary second, when the vendor’s clock resolution is coarse.
2. Clock skew at the vendor. If the API assigns timestamps from a server whose clock drifts, a late page can carry a timestamp below an already-advanced watermark and be silently dropped. Reconcile counts against the vendor manifest, and treat a sustained drop rate as a skew alarm rather than normal dedup.
3. Cursor and watermark disagreeing. After a rewind, the cursor is fresh but the watermark is old; that is intentional. Never persist a next_cursor without the watermark it corresponds to — the pair must move together or a crash can resume from a cursor that outran the durable data.
4. Poison page. If one page repeatedly fails to write, the checkpoint never advances and the sync loops. Cap retries per page and route a persistently failing page to a dead-letter sink so the stream can advance past it under audit.

Integration Note

This client is the stateful layer on top of the stateless fetcher in the parent AFC API Data Extraction component: extraction owns capture and validation of a single page, while the checkpoint and watermark here decide which pages to request and which records are new. Its closest sibling is Handling Rate Limits on Legacy AFC Vendor APIs, which supplies the backoff and Retry-After discipline that must wrap fetch so a throttled run pauses cleanly instead of expiring cursors under retry storms. Together they give exactly-once application over an at-least-once transport.

FAQ

Related

• AFC API Data Extraction — the parent component whose stateless fetcher this sync client wraps with checkpoint state
• Handling Rate Limits on Legacy AFC Vendor APIs — the backoff and Retry-After layer that must wrap the transport
• Schema Validation Pipelines — model-level validation each synced page passes through before the sink
• CSV Batch Parsing Workflows — the flat-file path that reconciles against the same taps this sync pulls over REST
• GTFS-RT Realtime Sync — where synced tap timestamps align against live vehicle positions

Part of AFC API Data Extraction, within Fare Data Ingestion & GTFS-RT Sync.