API Polling Strategies for FSMA 204 Supplier Telemetry Ingestion

FSMA Rule 204 establishes a strict 24-hour window for reconstructing a complete chain of custody following an FDA traceability request. For food manufacturers, distributors, and logistics operators, that mandate eliminates manual reconciliation and requires automated, stateful ingestion of supplier telemetry. The specific engineering problem this page solves is how to fetch Critical Tracking Events (CTEs) from heterogeneous supplier APIs on a repeating cycle without losing records, duplicating events, or letting non-compliant payloads reach the traceability ledger. A deterministic polling strategy reliably extracts each CTE and maps it to the FDA-mandated Key Data Elements (KDEs) — the traceability_lot_code, quantities, locations, and timestamps that make a lot chain reconstructable. Within the broader Supplier Data Ingestion & Sync Automation pipeline, polling is the primary ingestion vector for suppliers that expose REST endpoints: ERP systems, warehouse management platforms, and agricultural IoT gateways that emit harvest records, cooling events, and shipment movements at scale.

The triggering requirement is concrete. A shipping CTE captured under 21 CFR 1.1340 is only useful if the poller retrieves it, validates its KDEs, and persists it before the next FDA traceback request. If a polling cycle silently skips records during a rate-limited window, or advances its cursor past events it never persisted, the lot chain fractures and the 24-hour reconstruction fails. Everything below is engineered to make that failure impossible.

Stateful Delta Polling and KDE Fidelity

Naive full-sync polling is computationally expensive, introduces unacceptable latency into recall simulations, and rapidly exhausts supplier API quotas. Production-grade systems must implement cursor-based or timestamp-driven delta polling. Each supplier ERP, WMS, or agricultural IoT gateway typically exposes a last_modified, updated_at, or incremental cursor field that serves as the polling anchor. The ingestion engine maintains a persistent state store tracking the last successfully processed cursor per supplier endpoint. On each execution cycle, the poller requests records where updated_at > last_cursor, applies strict KDE validation, and advances the cursor only after successful downstream persistence — never before.

Figure — Stateful delta-polling cycle:

The order of operations in this cycle is the compliance guarantee. Advancing the cursor is the last step, gated on successful persistence, so a crash mid-cycle re-fetches the same delta on the next run rather than skipping it. Because delta fetches are inherently idempotent — the same updated_at > last_cursor query returns the same records — re-processing is safe as long as downstream writes are keyed on a deterministic record hash.

The KDE Data Contract for Polled Payloads

FSMA 204 KDE mapping requires deterministic transformation of raw API payloads into standardized traceability records. Before any polled record is persisted, it must satisfy the contract below. Each field maps a raw supplier key onto a canonical KDE, applies a validation rule at the ingestion boundary, and cites the FDA mandate that requires it. This is the polling-scoped subset of the full KDE Field Mapping Guide, which covers the complete field catalog across every transport.

Any deviation in field naming, unit conversion, or timezone normalization fractures the traceability graph and triggers compliance audit failures. The polling layer must enforce this schema before downstream routing. For suppliers lacking modern REST endpoints, a parallel CSV/EDI Parser Setup ensures legacy data streams undergo identical validation and normalization before entering the traceability graph — the same contract, a different transport.

Resilient Execution and Adaptive Concurrency

Supplier systems operate under strict throughput constraints and varying uptime SLAs. Aggressive polling triggers HTTP 429 responses, IP rate bans, or degraded service that directly impacts recall readiness. Implementing robust rate-limit handling for food supply chains requires exponential backoff with randomized jitter, circuit breaker patterns, and dynamic interval scaling based on supplier tier classifications. The polling scheduler must respect Retry-After headers, maintain per-tenant concurrency pools to prevent cross-supplier resource starvation, and gracefully degrade when upstream systems become unresponsive.

The retry delay on attempt $n$ follows a capped exponential curve with bounded jitter, where $b$ is the base delay, $c$ the ceiling, $j$ the jitter magnitude, and $r\sim U(0,1)$:

The jitter term is not cosmetic. When many suppliers throttle at the same wall-clock moment — for example, at the top of an hour when scheduled ERP jobs fire — synchronized retries produce a thundering herd that re-triggers the throttle. Randomizing the delay spreads reconnection attempts across the recovery window and keeps each supplier’s concurrency pool below its published ceiling.

Production polling engines also require structured, audit-ready logging. Every request, validation pass or fail, cursor advancement, and retry event must be recorded with immutable timestamps, correlation IDs, and supplier metadata. This logging layer becomes the primary evidence source during FDA audits or internal compliance reviews, and it is subject to the same access controls described in the Security Boundaries for Trace Data guidance.

Production Implementation: Python Polling Engine

The following implementation demonstrates a production-ready polling engine featuring stateful cursor management, pydantic v2 KDE validation, structured audit logging, tenacity-based retry orchestration, and explicit quarantine routing. Records that fail validation are never dropped — they are isolated with full provenance for manual reconciliation. Validated records are designed to feed directly into downstream Async Batch Processing pipelines.

import os
import time
import json
import logging
import hashlib
from datetime import datetime, timezone
from decimal import Decimal
from typing import Any

import requests
from pydantic import BaseModel, Field, ValidationError, field_validator
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception_type,
)

# Configure structured, audit-ready logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)
logger = logging.getLogger("fsma204.poller")


class KDERecord(BaseModel):
    """Strict schema for FSMA 204 Key Data Elements (21 CFR Part 1, Subpart S)."""

    traceability_lot_code: str = Field(..., min_length=1, max_length=64)
    product_description: str = Field(..., min_length=1)
    quantity: Decimal = Field(..., gt=0)
    unit_of_measure: str = Field(..., pattern="^(kg|lb|ea|case|pallet|liter|gallon)$")
    origin_location_id: str = Field(..., min_length=1)
    destination_location_id: str = Field(..., min_length=1)
    event_timestamp: datetime
    event_type: str = Field(
        ..., pattern="^(harvest|cooling|packing|shipping|receiving|transformation)$"
    )

    @field_validator("event_timestamp", mode="before")
    @classmethod
    def normalize_utc(cls, v: Any) -> datetime:
        dt = datetime.fromisoformat(v.replace("Z", "+00:00")) if isinstance(v, str) else v
        if dt.tzinfo is None:
            raise ValueError("event_timestamp must be timezone-aware")
        return dt.astimezone(timezone.utc)


class SupplierPoller:
    def __init__(
        self,
        supplier_id: str,
        base_url: str,
        api_key: str,
        state_store: dict[str, str],
        quarantine: list[dict[str, Any]],
    ) -> None:
        self.supplier_id = supplier_id
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.state_store = state_store  # In production: Redis, PostgreSQL, or DynamoDB
        self.quarantine = quarantine     # In production: a durable dead-letter store
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Accept": "application/json",
            "User-Agent": "FSMA204-Traceability-Poller/1.0",
        })

    def _get_cursor(self) -> str | None:
        return self.state_store.get(f"cursor:{self.supplier_id}")

    def _save_cursor(self, cursor: str) -> None:
        self.state_store[f"cursor:{self.supplier_id}"] = cursor

    @retry(
        retry=retry_if_exception_type((requests.exceptions.RequestException,)),
        wait=wait_exponential_jitter(initial=2, max=30),
        stop=stop_after_attempt(4),
        reraise=True,
    )
    def _fetch_delta(self, cursor: str | None) -> requests.Response:
        params: dict[str, Any] = {"limit": 500}
        if cursor:
            params["updated_after"] = cursor

        logger.info(
            "Polling supplier delta | supplier_id=%s | cursor=%s",
            self.supplier_id, cursor,
        )

        response = self.session.get(f"{self.base_url}/api/v1/telemetry", params=params)

        # Honor upstream flow-control before raising; a 429 with Retry-After is a
        # directive, not noise. Sleeping here keeps tenacity's own backoff additive.
        if response.status_code == 429 and "Retry-After" in response.headers:
            delay = int(response.headers["Retry-After"])
            logger.warning(
                "Upstream throttle | supplier_id=%s | retry_after=%ss",
                self.supplier_id, delay,
            )
            time.sleep(delay)

        response.raise_for_status()
        return response

    def _quarantine_record(self, raw: dict[str, Any], record_id: str, errors: Any) -> None:
        """Isolate a non-compliant record with full provenance for reconciliation."""
        self.quarantine.append({
            "record_id": record_id,
            "supplier_id": self.supplier_id,
            "raw_payload": raw,
            "validation_errors": errors,
            "quarantined_at": datetime.now(timezone.utc).isoformat(),
        })
        logger.error(
            "Schema validation failed | record_id=%s | supplier_id=%s | errors=%s",
            record_id, self.supplier_id, errors,
        )

    def process_cycle(self) -> dict[str, Any]:
        cursor = self._get_cursor()
        validated_count = 0
        failed_count = 0

        try:
            response = self._fetch_delta(cursor)
            payload = response.json()
            records = payload.get("data", [])

            if not records:
                logger.info("No new records | supplier_id=%s", self.supplier_id)
                return {"status": "idle", "validated": 0, "failed": 0}

            for raw in records:
                record_id = hashlib.sha256(
                    json.dumps(raw, sort_keys=True).encode()
                ).hexdigest()[:12]
                try:
                    kde = KDERecord.model_validate(raw)
                    # In production: publish to the async worker pool / message queue.
                    logger.info(
                        "KDE validated | record_id=%s | lot_code=%s | event_ts=%s",
                        record_id,
                        kde.traceability_lot_code,
                        kde.event_timestamp.isoformat(),
                    )
                    validated_count += 1
                except ValidationError as exc:
                    failed_count += 1
                    self._quarantine_record(raw, record_id, exc.errors())

            # Advance the cursor ONLY after the full batch has been processed.
            new_cursor = payload.get("next_cursor") or payload.get("max_updated_at")
            if new_cursor:
                self._save_cursor(new_cursor)
                logger.info(
                    "Cursor advanced | supplier_id=%s | new_cursor=%s",
                    self.supplier_id, new_cursor,
                )

        except requests.exceptions.HTTPError as exc:
            status_code = exc.response.status_code if exc.response is not None else None
            logger.critical(
                "Polling cycle failed | supplier_id=%s | status_code=%s | error=%s",
                self.supplier_id, status_code, str(exc),
            )
            # Trigger circuit-breaker logic in production; leave the cursor untouched
            # so the next cycle safely re-fetches the same delta.
            return {"status": "error", "validated": validated_count, "failed": failed_count}

        return {"status": "completed", "validated": validated_count, "failed": failed_count}


# Example execution context
if __name__ == "__main__":
    STATE_STORE: dict[str, str] = {}   # Production: Redis / PostgreSQL / managed KV
    QUARANTINE: list[dict[str, Any]] = []  # Production: durable dead-letter store

    poller = SupplierPoller(
        supplier_id="SUP-8842",
        base_url="https://erp.supplier-domain.com",
        api_key=os.getenv("SUPPLIER_API_KEY", "dev-key"),
        state_store=STATE_STORE,
        quarantine=QUARANTINE,
    )

    result = poller.process_cycle()
    print(json.dumps(result, indent=2))

Error Handling and Quarantine Strategy

A polling engine that discards malformed records is worse than useless — it manufactures silent traceability gaps that only surface during an FDA investigation, when it is too late to recover the source data. The strategy above is fail-forward: every record that fails KDE validation is preserved in full, hashed for stable identity, annotated with the exact pydantic error paths, and routed to a durable quarantine store. Compliant records continue through the cycle uninterrupted, so one bad payload never stalls an entire supplier feed.

The quarantine record carries enough provenance to reconstruct what happened without re-querying the supplier: the original raw payload, the supplier ID, the structured validation errors, and a quarantine timestamp. Common triggers a reconciliation operator will see are a null or empty traceability_lot_code, a quantity that arrives as a string or a negative number, a naive (timezone-less) event_timestamp, or an event_type outside the recognized CTE vocabulary. Because the quarantine key is derived from a deterministic hash of the payload, re-delivery of the same broken record resolves to the same entry rather than flooding the queue. Records that are corrected upstream and re-polled flow through validation normally on the next cycle. This mirrors the broader error handling workflows used across the ingestion layer, where dead-letter routing and operator alerting are standardized.

Transport-level failures are handled separately from data-level failures. A 429, a timeout, or a 5xx is retried with jittered exponential backoff and, if it persists past the attempt ceiling, surfaces as a critical log event with the cursor left untouched — guaranteeing the next cycle re-fetches the same delta. A validation failure, by contrast, is a permanent defect in a specific record and goes straight to quarantine without retry. Conflating the two is the most common way polling pipelines lose data.

Integration with the Ingestion Pipeline

This polling engine is one ingestion vector inside the parent Supplier Data Ingestion & Sync Automation pipeline, and it is deliberately narrow: its only job is to fetch, validate, and hand off. Validated KDERecord objects are published to the message queue that fronts Async Batch Processing, where I/O-bound persistence and enrichment run on a worker pool decoupled from the polling cadence. This separation means a slow database write can never back-pressure the poller into missing a supplier’s rate-limit window.

The contract this page enforces is the same one applied by the schema validation rules at every other ingestion boundary, so a REST-polled record and a CSV-parsed record are indistinguishable once they reach the ledger. Downstream, the persisted KDE stream becomes the raw material for the FSMA 204 lot graph and its FDA-ready exports. Polling health — cursor lag, validation failure rate, quarantine depth per supplier — is emitted as telemetry that the data quality monitoring layer tracks against per-supplier SLAs, because ingestion lag beyond the 24-hour reconstruction window is itself a compliance exposure.

Operational Notes

Deploy the poller as a scheduled job (cron, Kubernetes CronJob, or a Celery beat task) rather than a long-lived loop, so that a crashed cycle is restarted cleanly by the scheduler and the state store remains the single source of truth for progress. Recommended runtime and dependency versions:

• Python 3.10+ (the code uses str | None union syntax and dict[str, str] generics).
• pydantic ≥ 2.5 — the v2 field_validator / model_validate API. Do not mix in v1 validator.
• tenacity ≥ 8.2 for wait_exponential_jitter.
• requests ≥ 2.31 for the HTTP session layer.

Configuration should come from the environment, never from code. At minimum, provide SUPPLIER_API_KEY per supplier, a STATE_STORE_URL (Redis or PostgreSQL DSN), and a QUARANTINE_STORE target. Tune three values per supplier tier: the poll interval (align it to the supplier’s CTE reporting cadence and keep it well inside the 24-hour window), the page limit (500 is a safe default; lower it for suppliers with tight response-size caps), and the retry ceiling. Store cursors and quarantine records in durable infrastructure — the in-memory dict and list in the example exist only to make the module runnable in isolation. In multi-instance deployments, guard each supplier’s cursor with a per-supplier lock so two workers cannot advance the same cursor concurrently and skip a delta.

Frequently Asked Questions

Related

• Handling API Rate Limits in Food Supply Chains — resolving 429 cascades with Retry-After parsing, semaphores, and circuit breakers.
• Async Batch Processing — the worker pool that consumes validated records from the poller.
• CSV/EDI Parser Setup — the same KDE contract applied to flat-file and EDI transports.
• Schema Validation Rules — enforcing the KDE contract at every ingestion boundary.
• Error Handling Workflows — dead-letter routing, retries, and operator alerting across the pipeline.

Up: Supplier Data Ingestion & Sync Automation