API Polling Workers skill for Codex

Done well means: the worker never processes an item twice in a way anyone notices, never skips one after a crash or restart, slows down when the API pushes back, and an operator can see how far behind it is. Polling is the fallback; check for webhooks, streaming or a change feed first, because they are cheaper for both sides.

Before writing the loop

Collect these facts from the API docs (needs network; note them in the worker’s README):

Fact	Why it matters
Filter for changes (updated_since, since_id, cursor, change feed)	Decides between incremental polling and full scans
Sort order and tie-breaker	Without a stable order, items sharing a timestamp get skipped at page edges
Pagination style (cursor, keyset, offset)	Offset pagination over changing data skips and repeats rows
Rate limits and headers (Retry-After, X-RateLimit-Remaining, reset)	Sets the budget for interval and page size
ETag / Last-Modified support	A 304 costs almost nothing and often does not count against quota
Freshness the business needs	1 minute and 15 minutes are very different budgets

Interval rule of thumb: start at the slowest interval the user accepts, and make sure requests per poll x polls per hour stays under half the hourly rate limit, leaving headroom for retries and other clients sharing the key.

Worker shape

loop until shutdown:
    acquire/renew single-instance lock
    fetch pages since checkpoint (conditional request when possible)
    for each item (in stable order): dedup -> process -> record id
    after each page: write checkpoint atomically
    sleep(interval with jitter), waking early on shutdown

Reference implementation (Python, standard library + requests)

import json, os, random, signal, tempfile, time
from datetime import datetime, timedelta, timezone
from pathlib import Path
import requests

STATE = Path(os.environ.get("POLL_STATE", "state/orders.json"))
INTERVAL = int(os.environ.get("POLL_INTERVAL", "60"))
OVERLAP = timedelta(minutes=2)
stop = False

def _stop(*_):
    global stop
    stop = True
signal.signal(signal.SIGTERM, _stop)
signal.signal(signal.SIGINT, _stop)

def load_state() -> dict:
    return json.loads(STATE.read_text()) if STATE.exists() else {"cursor": None, "etag": None, "seen": []}

def save_state(state: dict) -> None:            # atomic: temp file + rename
    STATE.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=STATE.parent)
    with os.fdopen(fd, "w") as f:
        json.dump(state, f)
        f.flush(); os.fsync(f.fileno())
    os.replace(tmp, STATE)

def get_with_backoff(session, url, params, headers, max_tries=6):
    for attempt in range(max_tries):
        try:
            r = session.get(url, params=params, headers=headers, timeout=(5, 30))
        except (requests.ConnectionError, requests.Timeout):
            r = None
        if r is not None and r.status_code < 500 and r.status_code != 429:
            return r                                   # 2xx, 304, or a 4xx we must not retry
        retry_after = r.headers.get("Retry-After") if r is not None else None
        delay = float(retry_after) if retry_after and retry_after.isdigit() \
            else min(300, 2 ** attempt) * random.uniform(0.5, 1.0)   # capped, jittered
        time.sleep(delay)
    raise RuntimeError(f"gave up after {max_tries} attempts: {url}")

def poll_once(session, state) -> int:
    since = state["cursor"]
    params = {"sort": "updated_at,id", "limit": 100}
    if since:
        params["updated_since"] = (datetime.fromisoformat(since) - OVERLAP).isoformat()
    headers = {"If-None-Match": state["etag"]} if state.get("etag") else {}
    processed = 0
    url = "https://api.example.com/v1/orders"
    seen = set(state["seen"])
    while url and not stop:
        r = get_with_backoff(session, url, params, headers)
        if r.status_code == 304:
            return 0
        r.raise_for_status()                          # 4xx: fail loudly, do not retry
        body = r.json()
        for item in body["data"]:
            key = f'{item["id"]}:{item["updated_at"]}'
            if key in seen:
                continue                               # overlap re-read, already handled
            handle(item)                               # must be idempotent
            seen.add(key)
            state["cursor"] = max(state["cursor"] or item["updated_at"], item["updated_at"])
            processed += 1
        state["seen"] = list(seen)[-5000:]             # bounded dedup window
        state["etag"] = r.headers.get("ETag") if not body.get("next") else state.get("etag")
        save_state(state)                              # checkpoint per page
        url, params, headers = body.get("next"), None, {}
    return processed

The seen window only needs to cover the overlap period. Trimming a Python set by slicing a list is arbitrary order; for real workloads keep (key, timestamp) pairs and evict by age, or store processed keys in the database with a unique constraint.

Decisions

Situation	Do this
API supports a cursor or change feed	Store the opaque cursor; no overlap or timestamp logic needed
Only updated_since	Overlap window + dedup on (id, updated_at)
No change filter at all	Full fetch, hash each item, diff against stored hashes; poll rarely
Items must be processed exactly once downstream	Write results and the processed key in one DB transaction (or use an outbox)
Many tenants/accounts to poll	One scheduler, a queue of (tenant, due_at) jobs, per-tenant rate limit
Processing is slow	Poller only enqueues; workers process ($background-jobs)

Running it

• Single instance: two pollers on one cursor race and double-process. Use a DB advisory lock (pg_try_advisory_lock), a lease row with expiry, or a platform with replicas: 1 and no overlap during deploys (Recreate strategy in Kubernetes).
• State location: a local JSON file is fine on one long-lived machine; on containers or serverless, keep the checkpoint in Postgres, Redis or the platform’s KV.
• Shutdown: handle SIGTERM, finish the current item, checkpoint, exit within the platform’s grace period. Sleep in short slices so shutdown is prompt.
• Cron vs daemon: a cron job every N minutes is simpler and restarts cleanly; a daemon suits sub-minute intervals. Both need the same checkpoint and lock.
• Secrets: read API keys from the environment or secret manager; never write them into the state file or logs.

Observability

Log one line per poll: items fetched, processed, skipped as duplicates, HTTP status mix, current cursor, and lag (now - cursor). Alert when lag exceeds a threshold (for example 3x the interval) or when the poll has not completed in 10 intervals. A stuck cursor with successful polls usually means the API filter silently changed.

Gotchas

• Retry-After can be seconds or an HTTP date; handle both (email.utils.parsedate_to_datetime).
• Synchronised retries from many workers become a thundering herd; jitter is mandatory.
• updated_since with second precision plus > instead of >= drops items updated in the same second as the cursor.
• Some APIs return items in descending order by default; walking pages newest-first and checkpointing per page moves the cursor past older unprocessed pages.
• A 200 response with an error body ({"ok": false}) exists in several popular APIs; check the payload, not only the status.
• Clock skew between your host and the API: never compute the cursor from local time.

Verification

• Unit-test poll_once against recorded responses: empty page, 304, 429 with Retry-After, 500 then success, page boundary with equal timestamps.
• Kill the worker mid-page and restart: no duplicates downstream, nothing missing.
• Run against the real API with a dry-run handle and compare the item count with the API’s own UI or count endpoint for the same window.
• Watch X-RateLimit-Remaining over an hour; it should never approach zero.

Handover

**Worker:** orders-poller, polls /v1/orders every 60 s (+-20 % jitter)
**Checkpoint:** Postgres table poll_state (cursor = max updated_at, 2 min overlap)
**Dedup:** processed_orders (order_id, updated_at) unique constraint
**Limits:** ~65 requests/hour of 1,000 allowed; backs off on 429/5xx, honours Retry-After
**Alerts:** lag > 3 min, or no successful poll for 10 min
**Run:** `python -m workers.orders_poller` (single replica, SIGTERM-safe)