WMS & ERP Polling Strategies for Velocity Ingestion

Warehouse slotting only stays optimal while the velocity signals feeding it stay current, and for the majority of mid-tier WMS and ERP deployments those signals arrive by polling rather than by push. This guide is part of the Velocity Data Ingestion & WMS Sync Pipelines architecture, and it owns the extraction boundary specifically: how you pull inventory snapshots and movement transactions from a system that exposes only paginated REST or SOAP endpoints, how you stay under its rate limits, and how you turn each poll into a clean delta the downstream velocity engine can score. Get this layer wrong and everything above it inherits the damage — a stalled cursor makes yesterday’s demand masquerade as today’s, and a silent 429 drops SKUs that then read as zero-velocity slow movers.

What WMS/ERP Polling Is

Polling is the pattern of repeatedly querying a source system on an interval to detect state that has changed since the last query, in contrast to an event-driven model where the source pushes each change to a webhook as it happens. In a slotting context the source is the WMS inventory ledger and the ERP transaction log; the changed state is pick events, receipts, adjustments, and on-hand quantities; and the consumer is the velocity scoring engine that converts movement frequency into the tiers described in SKU Velocity Taxonomy Design.

Three properties separate a production poller from a while True loop that hits an endpoint on a fixed timer:

• Watermark-cursor extraction, not full re-reads. A production poller tracks a high-water mark — a monotonically increasing timestamp or sequence id — and asks the source only for records past it. Re-reading the full inventory table every cycle burns quota, saturates the ERP, and scales linearly with catalog size instead of with change volume.
• Adaptive intervals, not a hard-coded sleep. The gap between polls is a function of transaction volume and quota headroom, not a constant. During a receiving surge the poller tightens the loop; on a quiet third shift it widens it to conserve quota and ERP load.
• Idempotent snapshot reconciliation, not blind appends. Because a poll can overlap a previous one, return a record twice, or run after a partial failure, each record carries a key that lets the consumer dedupe and reconcile ERP financial inventory against WMS physical inventory deterministically.

Two collection shapes appear in the field. Incremental delta polling pulls only records past the watermark and is the default for high-cardinality transaction feeds. Full-snapshot polling pulls the entire on-hand table on a slow cadence and is reserved for reconciliation, where you need an absolute truth to correct accumulated delta drift. Most pipelines run both: frequent deltas for freshness, a nightly full snapshot for correctness.

The decision between polling and webhooks is not ideological. Use webhooks when the WMS vendor guarantees delivery and your infrastructure can absorb async event fan-out with TLS rotation, deduplication, and out-of-order handling. Fall back to polling when the platform exposes only paginated endpoints, when the event payload lacks the dimensional detail velocity scoring needs (a missing pick_type or unit_of_measure), or when audit trails demand deterministic replay. In practice, slotting recalculations run on scheduled optimization windows rather than per-transaction, so a well-tuned poller usually yields higher data integrity and far simpler debugging than a webhook fleet.

Input Data Requirements

The poller needs two things: the connection contract for the source endpoint, and a durable cursor it can advance. Everything else — field-level schema, unit-of-measure normalization — is the responsibility of layers downstream of extraction; this stage only guarantees transport correctness and at-least-once delivery of raw records.

Every record the poller returns must carry three fields the consumer relies on for idempotency and freshness: a stable business key (sku + location_id), a change timestamp that drives the watermark, and enough movement context to score velocity. The dataclasses below define that contract at the extraction boundary.

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone

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


@dataclass(frozen=True)
class PollConfig:
    """Immutable connection + cadence contract for one WMS/ERP feed."""
    base_url: str
    resource_path: str = "/inventory/snapshot"
    api_key: str = ""
    page_size: int = 500
    base_interval_s: float = 30.0
    min_interval_s: float = 10.0
    max_interval_s: float = 120.0
    rate_limit_tps: float = 5.0
    max_retries: int = 5


@dataclass
class InventoryDelta:
    """One extracted change record — the atomic unit the velocity engine consumes."""
    sku: str
    location_id: str
    quantity_on_hand: int
    last_movement_ts: datetime
    pick_type: str | None = None
    unit_of_measure: str | None = None
    ingested_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    @property
    def business_key(self) -> str:
        """Stable dedupe key for idempotent reconciliation downstream."""
        return f"{self.sku}:{self.location_id}"

Step-by-Step Implementation

The poller runs in four passes: advance a durable watermark cursor, poll under a rate-limit governor that treats 429 as an explicit signal, validate and normalize each snapshot into InventoryDelta records, then emit deltas to the velocity engine and adapt the next interval to observed volume. Each pass is isolated so a fault in one never corrupts the cursor.

1. Track and Advance a Durable Watermark Cursor

The cursor is the difference between polling change volume and polling catalog size. Persist the last successfully processed change timestamp per feed, request only records strictly after it, and advance it only after the batch is durably handed off — never mid-batch. Advancing early on a partial failure silently drops every record between the crash point and the new watermark.

from typing import Protocol


class CursorStore(Protocol):
    """Durable persistence for per-feed watermarks (DB row, KV, etc.)."""
    async def load(self, feed: str) -> datetime | None: ...
    async def save(self, feed: str, watermark: datetime) -> None: ...


async def resolve_watermark(store: CursorStore, feed: str, cfg: PollConfig) -> datetime:
    """Load the persisted watermark, or seed a cold start one interval back."""
    wm = await store.load(feed)
    if wm is None:
        from datetime import timedelta
        wm = datetime.now(timezone.utc) - timedelta(seconds=cfg.base_interval_s)
        logger.info("feed %s cold start; seeding watermark at %s", feed, wm.isoformat())
    return wm

2. Poll Under a Rate-Limit Governor

Aggressive polling degrades ERP performance and trips 429 Too Many Requests; conservative polling injects latency into slotting recalculations. A token-bucket governor bounds the request rate independently of loop timing, so a burst of pages during a busy window still respects the documented ceiling. When the source returns 429, honor its Retry-After header if present and otherwise back off exponentially with jitter. The deeper treatment of quota-blind sources — where no ceiling is published at all — lives in Polling WMS APIs Without Published Rate Limits.

import asyncio
import random
import time
from typing import Any

import aiohttp


class TokenBucket:
    """Refills at rate_limit_tps; acquire() blocks until a token is available."""

    def __init__(self, rate_tps: float, burst: int | None = None) -> None:
        self.rate = rate_tps
        self.capacity = burst if burst is not None else max(1, int(rate_tps))
        self.tokens = float(self.capacity)
        self.updated = time.monotonic()

    async def acquire(self) -> None:
        while True:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
            self.updated = now
            if self.tokens >= 1.0:
                self.tokens -= 1.0
                return
            await asyncio.sleep((1.0 - self.tokens) / self.rate)


async def fetch_page(
    session: aiohttp.ClientSession,
    bucket: TokenBucket,
    cfg: PollConfig,
    since: datetime,
    cursor: str | None,
) -> dict[str, Any]:
    """Fetch one page after `since`, respecting the rate ceiling and 429 backoff."""
    params: dict[str, Any] = {"limit": cfg.page_size, "since": since.isoformat()}
    if cursor:
        params["cursor"] = cursor
    for attempt in range(cfg.max_retries + 1):
        await bucket.acquire()
        async with session.get(cfg.resource_path, params=params) as resp:
            if resp.status == 429:
                retry_after = float(resp.headers.get("Retry-After") or 0) or \
                    min(2 ** attempt + random.uniform(0, 1), 60)
                logger.warning("rate-limited on %s; backing off %.1fs", cfg.resource_path, retry_after)
                await asyncio.sleep(retry_after)
                continue
            resp.raise_for_status()
            return await resp.json()
    raise RuntimeError(f"exhausted {cfg.max_retries} retries fetching {cfg.resource_path}")

3. Validate and Normalize Each Snapshot

Legacy ERPs return inconsistent field mappings, missing lot attributes, and deprecated SKU hierarchies. The poller coerces each raw row into an InventoryDelta and drops the malformed ones to a reject log rather than letting a single bad record abort the batch. This is deliberately lightweight transport-level validation; the authoritative field-contract enforcement, dead-lettering, and quarantine policy belong to Schema Validation for Inventory Feeds, which runs immediately downstream.

def normalize_rows(rows: list[dict[str, Any]]) -> tuple[list[InventoryDelta], int]:
    """Coerce raw endpoint rows into deltas; count and log rejects without aborting."""
    deltas: list[InventoryDelta] = []
    rejects = 0
    for row in rows:
        try:
            deltas.append(
                InventoryDelta(
                    sku=str(row["sku"]),
                    location_id=str(row["location_id"]),
                    quantity_on_hand=int(row["quantity_on_hand"]),
                    last_movement_ts=datetime.fromisoformat(row["last_movement_ts"]),
                    pick_type=row.get("pick_type"),
                    unit_of_measure=row.get("unit_of_measure"),
                )
            )
        except (KeyError, ValueError, TypeError) as exc:
            rejects += 1
            logger.debug("rejected malformed row %r: %s", row.get("sku", "?"), exc)
    if rejects:
        logger.warning("normalized %d rows, rejected %d", len(deltas), rejects)
    return deltas, rejects

4. Emit Deltas and Adapt the Next Interval

The orchestrator wires the passes together: resolve the watermark, page through every change since it under the governor, normalize, hand the deltas to the velocity engine, then advance the watermark to the newest change timestamp and size the next sleep to the volume just observed. A dense page means the source is busy, so tighten the loop; a sparse page means it is idle, so widen it and conserve quota.

from typing import Awaitable, Callable


async def poll_once(
    session: aiohttp.ClientSession,
    bucket: TokenBucket,
    store: CursorStore,
    cfg: PollConfig,
    feed: str,
    emit: Callable[[list[InventoryDelta]], Awaitable[None]],
) -> float:
    """Run one full poll cycle; return the adaptively computed next interval in seconds."""
    since = await resolve_watermark(store, feed, cfg)
    cursor: str | None = None
    high_watermark = since
    total = 0

    while True:
        payload = await fetch_page(session, bucket, cfg, since, cursor)
        deltas, _ = normalize_rows(payload.get("data", []))
        if deltas:
            await emit(deltas)
            high_watermark = max(high_watermark, max(d.last_movement_ts for d in deltas))
            total += len(deltas)
        cursor = payload.get("next_cursor")
        if not cursor:
            break

    # Advance the watermark only after the whole batch is emitted.
    if high_watermark > since:
        await store.save(feed, high_watermark)

    # Adapt: dense cycle -> poll sooner; sparse cycle -> back off.
    if total >= cfg.page_size * 0.8:
        nxt = max(cfg.min_interval_s, cfg.base_interval_s * 0.5)
    elif total == 0:
        nxt = min(cfg.max_interval_s, cfg.base_interval_s * 1.5)
    else:
        nxt = cfg.base_interval_s
    logger.info("feed %s: emitted %d deltas, watermark=%s, next poll in %.0fs",
                feed, total, high_watermark.isoformat(), nxt)
    return nxt


async def run_poller(
    cfg: PollConfig,
    store: CursorStore,
    feed: str,
    emit: Callable[[list[InventoryDelta]], Awaitable[None]],
) -> None:
    """Long-running poll loop with a shared session and token-bucket governor."""
    bucket = TokenBucket(cfg.rate_limit_tps)
    headers = {"Authorization": f"Bearer {cfg.api_key}", "Accept": "application/json"}
    timeout = aiohttp.ClientTimeout(total=30, connect=10)
    async with aiohttp.ClientSession(base_url=cfg.base_url, headers=headers, timeout=timeout) as session:
        while True:
            try:
                interval = await poll_once(session, bucket, store, cfg, feed, emit)
            except Exception as exc:  # noqa: BLE001 - loop must survive transient faults
                logger.error("feed %s poll cycle failed: %s", feed, exc)
                interval = cfg.base_interval_s
            await asyncio.sleep(interval)

The emit callback is the seam to the rest of the pipeline. In production it publishes deltas to a message broker or hands them directly to the recalculation runner in Async Batch Processing for Velocity, which triggers a scoring pass on watermark advance rather than on a blind timer.

Tuning & Calibration

Two parameters move outcomes most: base_interval_s and rate_limit_tps. The interval trades freshness against load — too short churns the ERP and inflates cost per useful record; too long lets the velocity window go stale and slotting recommendations lag real demand. The rate ceiling is not a tuning knob at all: set it from the vendor’s documented TPS and leave it, because exceeding it is what generates the 429 storms the governor exists to prevent. Externalize both per feed so a slow legacy SOAP endpoint and a fast cloud REST endpoint run under different profiles.

# polling.yaml — one profile per WMS/ERP feed
feed: wms_inventory_primary
connection:
  base_url: "https://wms.internal.example"
  resource_path: "/inventory/snapshot"
  page_size: 500              # rows per page; cap to the endpoint's max payload
cadence:
  base_interval_s: 30         # idle poll gap; the adaptive floor scales from here
  min_interval_s: 10          # tightest loop during a receiving surge
  max_interval_s: 120         # widest loop on a quiet shift
governor:
  rate_limit_tps: 5.0         # documented vendor ceiling; NOT a tuning knob
  max_retries: 5              # attempts before a page is surfaced as failed
reconciliation:
  full_snapshot_cron: "0 3 * * *"   # nightly absolute-truth read to correct drift

# Equivalent Python config dict consumed by the poller
POLLING = {
    "feed": "wms_inventory_primary",
    "connection": {
        "base_url": "https://wms.internal.example",
        "resource_path": "/inventory/snapshot",
        "page_size": 500,
    },
    "cadence": {"base_interval_s": 30, "min_interval_s": 10, "max_interval_s": 120},
    "governor": {"rate_limit_tps": 5.0, "max_retries": 5},
    "reconciliation": {"full_snapshot_cron": "0 3 * * *"},
}

Run a two-tier cadence. Incremental delta polling stays inside the min–max interval band and keeps the velocity window fresh, while a full-snapshot reconciliation runs nightly (03:00 local is the common slot, chosen so it never overlaps a picking wave) to correct any delta drift the watermark accumulated. Anchor the reconciliation cron to operational rhythm, not an arbitrary hour — a full read during a receiving surge competes with live transactions for the same rate budget.

Validation & Testing

Never ship a poller without asserting its two load-bearing invariants: the watermark advances monotonically and only after a successful emit, and a 429 triggers a retry rather than a crash. The pytest checks below encode both and run in the ingestion job’s CI gate.

import asyncio

import pytest


class _MemStore:
    def __init__(self) -> None:
        self._d: dict[str, datetime] = {}

    async def load(self, feed: str) -> datetime | None:
        return self._d.get(feed)

    async def save(self, feed: str, watermark: datetime) -> None:
        self._d[feed] = watermark


def test_watermark_advances_only_forward() -> None:
    store = _MemStore()

    async def _run() -> datetime | None:
        base = datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc)
        await store.save("f1", base)
        # A stale batch must never move the watermark backwards.
        older = base.replace(hour=11)
        if older > base:
            await store.save("f1", older)
        return await store.load("f1")

    assert asyncio.run(_run()).hour == 12


def test_normalize_drops_malformed_rows_without_aborting() -> None:
    rows = [
        {"sku": "A1", "location_id": "P-01", "quantity_on_hand": 5,
         "last_movement_ts": "2026-07-02T10:00:00+00:00"},
        {"sku": "BAD"},  # missing required fields
    ]
    deltas, rejects = normalize_rows(rows)
    assert len(deltas) == 1 and rejects == 1
    assert deltas[0].business_key == "A1:P-01"

A healthy run logs feed wms_inventory_primary: emitted N deltas, watermark=..., next poll in 30s, test_watermark_advances_only_forward confirms the cursor never regresses, and test_normalize_drops_malformed_rows_without_aborting keeps the one valid delta while quarantining the malformed row. If the watermark ever moves backward, fix cursor persistence before anything else — a regressing watermark re-emits and double-counts movement, inflating velocity for the affected SKUs.

Integration Points

The poller is the extraction boundary; it produces data but decides nothing. It sits between three sibling systems, each with a contract:

• Downstream contract enforcement. Raw deltas leave this stage and immediately hit Schema Validation for Inventory Feeds, which owns field-type checks, unit-of-measure conformance, and the dead-letter quarantine. Keep transport rejects (malformed rows dropped here) in a separate path from schema rejects so the two are diagnosed independently.
• Historical normalization. The velocity window a scored delta contributes to is only meaningful when sales history is normalized across channels, pack sizes, and promotions by Sales History Data Mapping. A poller that emits a raw pallet-unit quantity into a pipeline expecting eaches produces a phantom hyper-mover no amount of polling accuracy will fix.
• Compute trigger. Watermark advance is the natural trigger for Async Batch Processing for Velocity. Fire the recalculation on advance rather than a blind cron so the batch scores current demand; a frozen watermark should raise a staleness alert, not silently recompute yesterday.

Further downstream, the scored profiles those layers produce feed the tier and location logic in Location Assignment & ABC Classification Algorithms, which converts velocity coefficients into committed slot moves under weight, volume, and zone constraints.

Failure Modes & Edge Cases

• Watermark advanced before emit (data loss). Saving the cursor mid-batch, then crashing, permanently skips every record between the crash and the new watermark. Remediation: advance the watermark only after the full batch is durably handed off, as poll_once does.
• Silent 429 read as low demand. Pages that quietly fail to complete drop SKUs from the cycle, and those SKUs read as zero-velocity slow movers. Remediation: treat 429 as an explicit back-off-and-retry signal, honor Retry-After, and alert when a cycle’s page count collapses unexpectedly.
• Clock skew between poller and source. An since filter based on the poller’s clock, ahead of the ERP’s, skips records stamped in the gap. Remediation: drive the watermark from the source’s own change timestamps (as returned in the payload), never from local wall-clock.
• Duplicate delivery across overlapping cycles. A long cycle that overruns its interval can overlap the next and re-emit boundary records. Remediation: dedupe downstream on business_key + last_movement_ts, and guard against concurrent cycles per feed with a lock.
• Delta drift without reconciliation. Missed or reordered deltas accumulate into an on-hand count that diverges from physical truth. Remediation: run the nightly full-snapshot reconciliation to reset the absolute state and correct the drift.

FAQ

Should I poll or use webhooks for WMS velocity data?

Use webhooks only when the vendor guarantees delivery and your infrastructure can absorb async fan-out with TLS rotation, deduplication, and out-of-order handling. Poll when the platform exposes only paginated REST or SOAP endpoints, when the event payload lacks the dimensional detail velocity scoring needs (a missing pick_type or unit_of_measure), or when audit trails demand deterministic replay. Because slotting recalculations run on scheduled windows rather than per-transaction, a well-tuned poller usually delivers higher data integrity and simpler debugging than a webhook fleet.

How do I pick a polling interval?

Start at the freshness your slotting window actually needs, not the fastest the API allows. A 30-second base interval that adapts down to 10 seconds during receiving surges and up to 120 seconds on a quiet shift covers most facilities. Size the tightest interval from the vendor’s rate ceiling and your page size so a busy cycle still finishes before the next one starts. Recalculating tiers faster than demand genuinely shifts only churns slot assignments and floods handlers with move tasks.

What is a watermark cursor and why does it matter?

A watermark is a durably persisted high-water mark — a change timestamp or sequence id — that records how far the poller has consumed a feed. Requesting only records past it turns extraction cost into a function of change volume instead of catalog size, and advancing it strictly after a successful emit makes the poller crash-safe. Drive it from the source’s own change timestamps, never the poller’s local clock, or clock skew will silently skip records.

How do I handle a WMS with no documented rate limit?

Treat the absence of a published ceiling as a reason for more caution, not less: start with a low token-bucket rate, watch response latency and error codes, and let observed 429s or latency inflation pull the rate down adaptively. The full pattern — probing for the effective ceiling and self-throttling against it — is covered in Polling WMS APIs Without Published Rate Limits.

How does polling stay reconciled with physical inventory?

Incremental delta polling keeps the velocity window fresh but accumulates drift when deltas are missed or reordered. Pair it with a nightly full-snapshot read that pulls the entire on-hand table and resets the absolute state, correcting any divergence between the delta-derived count and physical truth. Keep the reconciliation cron off peak so it never competes with live transactions for the rate budget.

Related

• Polling WMS APIs Without Published Rate Limits — probing for the effective ceiling and self-throttling when no TPS limit is documented.
• Schema Validation for Inventory Feeds — the field-contract enforcement and dead-letter quarantine that runs on the deltas this poller emits.
• Sales History Data Mapping — normalizing channels, pack sizes, and units so the velocity window is comparable.
• Async Batch Processing for Velocity — the recalculation runner triggered on watermark advance.
• Velocity Data Ingestion & WMS Sync Pipelines — the parent architecture this extraction layer feeds.