Form API Polling Strategies for Registration Ingestion

Form API polling is the deterministic entry point for attendee data and the first hard boundary in the Registration Ingestion & Payment Reconciliation pipeline. Unlike event-driven architectures that rely exclusively on real-time push, polling provides guaranteed state synchronization: it compensates for transient delivery failures, enforces strict idempotency before any downstream work begins, and gives operators an authoritative, replayable record of every registration the vendor API has ever returned. When payment webhook handling misses a delivery — a dropped connection, a signature-verification reject, a provider outage during an early-bird surge — the polling loop is what eventually reconciles the gap.

The failure mode this stage prevents is silent registration loss. A webhook-only ingestion has no memory: if the provider never redelivers, the attendee simply never exists in your system, and the miss surfaces only when someone is turned away at the check-in desk. By maintaining a persistent cursor and draining the vendor API in deterministic windows, the polling controller converts an unreliable push stream into an eventually-consistent pull stream. Its scope is deliberately narrow — acquire, validate, deduplicate, and hand off — so it stays stateless per cycle, safe to restart mid-drain, and easy to reason about during an incident. This page defines the data contract, the control loop, and the operational runbook for that boundary; for vendor-specific throttle handling it links out to Polling Eventbrite Web APIs Without Rate Limiting.

Scope Boundary Link to this section#

Polling is a stateless drain over a remote cursor. Each cycle owns data acquisition, contract validation, idempotent deduplication, and failure routing — and nothing else. It never renders a badge, settles a payment, or mutates a record after emit. Keeping this boundary explicit is what lets the controller crash, redeploy, or run concurrently against multiple events without double-ingesting.

Two rules keep the boundary clean. First, the controller marks each ingested record PENDING or COMPLETED from the vendor’s immediate response only — it never waits on downstream confirmation. Second, once a record is emitted with its idempotency key it is never mutated here; corrections arrive as fresh cycles or as webhook-driven mutations keyed on the same identifier. That single-writer discipline at ingress is what makes the rest of the pipeline safe to parallelize.

Data Contract Link to this section#

Vendor form APIs return heterogeneous payloads that rarely align with internal processing requirements. Relying on implicit type coercion at ingestion time introduces silent corruption that propagates through badge rendering and reconciliation. A rigid contract must be enforced immediately on receipt, before any persistence or queueing. The contract here is deliberately thin — enough to make routing and deduplication decisions safely — while the exhaustive structural rules live one stage over in schema validation pipelines.

import hashlib
from datetime import datetime, timezone
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator

PaymentStatus = Literal["PENDING", "COMPLETED", "REFUNDED", "FAILED"]


class RegistrationPayload(BaseModel):
    # Accept vendor aliases on input, reject unknown fields so drift surfaces loudly.
    model_config = ConfigDict(populate_by_name=True, extra="forbid")

    external_id: str = Field(validation_alias="id")
    email: EmailStr
    ticket_type: str = Field(validation_alias="ticket_sku")
    status: PaymentStatus = Field(validation_alias="payment_status")
    registered_at: datetime = Field(validation_alias="created_at")
    metadata: dict[str, Any] = Field(default_factory=dict)

    @field_validator("email", mode="before")
    @classmethod
    def normalize_email(cls, v: str) -> str:
        return v.strip().lower()

    @field_validator("status", mode="before")
    @classmethod
    def coerce_status(cls, v: str) -> str:
        canonical = str(v).strip().upper()
        allowed = {"PENDING", "COMPLETED", "REFUNDED", "FAILED"}
        # Unknown vendor states default to PENDING, never dropped silently.
        return canonical if canonical in allowed else "PENDING"

    @field_validator("registered_at", mode="before")
    @classmethod
    def coerce_utc(cls, v: str) -> datetime:
        dt = datetime.fromisoformat(str(v).replace("Z", "+00:00"))
        return dt.astimezone(timezone.utc)

    def idempotency_key(self) -> str:
        anchor = f"{self.external_id}:{self.email}:{self.registered_at.isoformat()}"
        return hashlib.sha256(anchor.encode()).hexdigest()

Each field earns its place in the contract:

• external_id — the vendor’s record key, mapped from the raw id. It anchors the idempotency hash, so a coercion failure here is a hard reject rather than a guess.
• email — normalized to lowercase, stripped, and RFC-5322-validated by EmailStr at parse time. Case-folding at ingress prevents duplicate attendee records for the same person registering twice.
• status — coerced to a canonical enum. Unknown vendor states resolve to PENDING instead of raising, because a badge must never be printed on an unconfirmed payment, and an unknown state is not evidence of capture.
• registered_at — parsed to a timezone-aware UTC datetime. Storing naive local timestamps is the single most common cause of cursor drift when a vendor silently shifts its offset.
• extra="forbid" — the drift guard. A renamed or newly added vendor field surfaces as a ValidationError at ingress, routed to rejection handling, instead of a silent fulfillment miss downstream.

Invalid records are never dropped. They are serialized into a structured rejection payload carrying the raw body, the Pydantic error codes, and a deterministic reconciliation key, so a bad batch can be replayed after a field_validator patch without re-polling the entire event.

Deterministic Implementation Link to this section#

The control loop composes contract validation, cursor-based pagination, an idempotency guard, exponential backoff on rate limits, circuit breaking on upstream 5xx, and structured rejection routing. Pagination must be cursor-based rather than offset-based: an offset shifts under concurrent writes during a live registration window, silently duplicating or skipping records. Every emitted record carries a correlation_id so a single registration can be traced from this loop through async batch processing and into the badge-render logs.

import logging
import time
import uuid
from typing import Any, Optional

import httpx
from pydantic import ValidationError

logger = logging.getLogger("registration.poller")


class PollingController:
    def __init__(self, api_base: str, token: str, state_store: dict[str, Any]):
        self.client = httpx.Client(
            base_url=api_base,
            headers={"Authorization": f"Bearer {token}"},
            timeout=15.0,
        )
        self.state = state_store
        self.cursor: Optional[str] = state_store.get("last_cursor")
        self.max_retries = 3
        self.backoff_base = 2.0
        self.page_size = 100  # bounded window; never request the whole event at once

    def _fetch_page(self, cursor: Optional[str]) -> dict[str, Any]:
        params: dict[str, Any] = {"limit": self.page_size}
        if cursor:
            params["cursor"] = cursor
        response = self.client.get("/registrations", params=params)
        response.raise_for_status()
        return response.json()

    def _process_record(self, raw: dict[str, Any], correlation_id: str) -> dict[str, Any]:
        try:
            record = RegistrationPayload.model_validate(raw)
        except ValidationError as e:
            reject_key = hashlib.sha256(
                repr(sorted(raw.items())).encode()
            ).hexdigest()
            logger.error(
                "record_rejected",
                extra={"correlation_id": correlation_id, "reject_key": reject_key,
                       "errors": e.errors()},
            )
            return {"status": "REJECTED", "reject_key": reject_key,
                    "raw_payload": raw, "validation_errors": e.errors(),
                    "correlation_id": correlation_id}
        return {"status": "VALID", "idempotency_key": record.idempotency_key(),
                "canonical": record.model_dump(mode="json"),
                "correlation_id": correlation_id}

    def run_poll_cycle(self) -> list[dict[str, Any]]:
        results: list[dict[str, Any]] = []
        current_cursor = self.cursor
        attempt = 0

        while attempt <= self.max_retries:
            try:
                page = self._fetch_page(current_cursor)
                records = page.get("data", [])
                if not records:
                    break

                for raw in records:
                    correlation_id = str(uuid.uuid4())
                    results.append(self._process_record(raw, correlation_id))

                # Advance the cursor only after the page fully processes.
                current_cursor = page.get("next_cursor")
                self.state["circuit_breaker_open"] = False
                attempt = 0  # reset the retry budget on any successful page
                if not current_cursor:
                    break

            except httpx.HTTPStatusError as e:
                code = e.response.status_code
                if code == 429:
                    wait = self.backoff_base ** attempt
                    retry_after = e.response.headers.get("Retry-After")
                    wait = float(retry_after) if retry_after else wait
                    logger.warning("rate_limited", extra={"wait_s": wait, "attempt": attempt})
                    time.sleep(wait)
                    attempt += 1
                elif 500 <= code < 600:
                    logger.error("upstream_5xx", extra={"code": code})
                    self.state["circuit_breaker_open"] = True
                    break  # do not burn the retry budget on a degraded upstream
                else:
                    logger.error("client_error", extra={"code": code})
                    break
            except httpx.TransportError as e:
                logger.exception("transport_error", extra={"error": str(e)})
                time.sleep(self.backoff_base ** attempt)
                attempt += 1

        # Persist the cursor exactly once, at the horizon we actually reached.
        self.state["last_cursor"] = current_cursor
        return results

Two design decisions are load-bearing. First, the cursor advances only after a page is fully processed and is persisted once at the end of the cycle — a crash mid-page replays that page rather than skipping it, and because every downstream side effect is keyed on the idempotency hash, the replay is a no-op. Second, an upstream 5xx trips the circuit breaker and breaks the loop rather than retrying: hammering a degraded vendor API deepens the outage, and the next scheduled cycle resumes cleanly from the persisted cursor.

Production Debugging and Observability Link to this section#

Resolution speed depends on deterministic tracing and explicit failure categorization. Every emitted record and every rejection carries correlation_id; valid records add idempotency_key, and rejections add reject_key. The controller logs JSON so records are queryable by field in the aggregation layer. A rejection on the wire looks like this:

{
  "level": "error",
  "event": "record_rejected",
  "correlation_id": "b1e0c7a2-9d4f-4c3a-8f21-6a0f5c2e1d90",
  "reject_key": "9f2c8a1d5e7b4c3a8f21...",
  "errors": [{"type": "extra_forbidden", "loc": ["promo_code_v2"], "msg": "Extra inputs are not permitted"}],
  "cursor_position": "eyJvZmZzZXQiOjE4NTAwfQ==",
  "dlq": "registration.poll.dlq"
}

Index the aggregation layer on the same field names the code emits. In Datadog, facet on @correlation_id, @reject_key, and @event, and page when sum:registration.poll.dlq{*}.as_count() breaches 5% of a cycle’s throughput. In an ELK stack, map correlation_id, idempotency_key, and reject_key as keyword fields so rejection spikes can be sliced by failure vector, and pin correlation_id as the cross-index correlation key so a single Kibana query joins the poll cycle to the batch worker and the render log. Failures cluster into three deterministic categories:

Malformed payloads are quarantined via the dead-letter queue routing mechanism rather than blocking the cycle, so one bad record never stalls capture for the whole event.

Performance and Memory Constraints Link to this section#

The controller is I/O-bound on the vendor API and the state store, but a naive loop still wastes API quota, buffers whole events into memory, or serializes CPU work behind the GIL. The mitigations below keep a single controller predictable before you scale to per-event workers.

Incident Triage Checklist Link to this section#

Target MTTR under 15 minutes. Work top to bottom; never force a cursor reset before confirming the drift, and never bulk-replay a rejected batch into an unrecovered vendor API.

1. Detect. Alert fires on DLQ rate > 5% of a cycle, or cursor lag past the vendor horizon. Confirm scope — one event/provider or all of them?
2. Inspect state and queue depth. Read the persisted cursor and the DLQ backlog directly:

   redis-cli HGET poll:state:evt_88 last_cursor
   redis-cli LLEN registration.poll.dlq
   

3. Isolate the vector. Group recent DLQ entries by error type to separate contract drift from an upstream outage, and spot-check whether a record already reached the downstream idempotency index:

   redis-cli --scan --pattern 'registration.poll.dlq:*' | head
   redis-cli GET "poll:idem:9f2c8a1d5e7b4c3a8f21..."
   

4. Contain. If the breaker is open, leave it — do not manually re-drive against a 5xx vendor. If it is a contract drift, pause the scheduler for this event only so a good cycle does not race the fix.
5. Resolve. For contract drift, add or relax the offending field_validator / alias and redeploy. For cursor drift beyond ~10 minutes, force a reset and trigger a bounded delta re-sync:

   redis-cli HSET poll:state:evt_88 last_cursor "$(date -u -d '15 min ago' +%s)"
   

6. Replay. Re-drive DLQ records in batches of 50 behind a manual approval gate; every record is keyed on its idempotency hash, so replaying already-captured records is a safe no-op.
7. Verify and roll back if needed. Confirm the rejection rate returns to baseline and audit a sample of re-synced registrations against the vendor’s record count. If a bad deploy caused it, roll back to the previous controller image and resume — because the cursor is persisted and every emit is idempotent, re-draining from the last horizon is safe.

By enforcing a rigid ingress contract, cursor-anchored replay, and deterministic rejection routing, form API polling turns an unreliable vendor push stream into an eventually-consistent, audit-ready capture layer — so badge generation never triggers on unvalidated data and no attendee is silently lost when a webhook goes missing.

Related Link to this section#

• Polling Eventbrite Web APIs Without Rate Limiting — vendor-specific throttle mitigation and header-aware backoff for the loop defined here.
• Async Batch Processing — the downstream drain that receives this stage’s validated, idempotency-keyed records and gates them on payment reconciliation.
• Payment Webhook Handling — the complementary real-time producer whose missed deliveries this polling loop reconciles.
• Schema Validation Pipelines — the deep contract governance stage that owns versioned structural rules beyond this ingress check.
• Registration Ingestion & Payment Reconciliation — the parent architecture that frames how capture, reconciliation, and fulfillment fit together.