Adaptive polling with ETag conditional requests #

Adaptive polling with ETag conditional requests is how a landside ingestion service keeps container status fresh against a Terminal Operating System (TOS) REST API while spending almost nothing on the cycles where nothing has moved. The technique pairs two HTTP mechanisms — a conditional request (If-None-Match against a stored ETag, or If-Modified-Since against a stored Last-Modified) that lets the server answer 304 Not Modified with an empty body, and a feedback controller that lengthens the interval when the resource is quiet and shortens it the moment new gate moves appear. Together they turn a fixed-cadence cron job into a change-rate-driven loop that survives a berth idle for six hours and a 40-move-per-hour discharge with the same client and the same rate budget.

Architecture Alignment #

This page is the control-loop layer of Terminal API Polling Strategies inside the Container Tracking & AIS Event Synchronization domain. It deliberately splits responsibilities with its sibling: Polling terminal operating systems via REST APIs specifies how one request is built on the wire — auth headers, pagination cursors, body parsing — whereas this page specifies the two feedback signals wrapped around that request: a per-resource validator cache that drives 304 short-circuits, and an interval controller that adapts cadence to observed change. Where a 304 or an outage leaves a stretch of missing sequence numbers, the gap is reconstructed by Backfilling terminal API gaps with AIS events. The typed events a 200 produces are still resolved downstream by the Container Status Mapping Rules engine before any state is committed; this layer only decides how often and how cheaply to ask.

Prerequisites & Environment Setup #

The controller targets Python 3.11+ for asyncio.timeout, precise datetime.fromisoformat, and structural pattern matching. Install the async HTTP client and the structured-logging stack; nothing else is required, because the validator cache and interval controller are plain typed objects:

pip install "httpx[http2]>=0.27" "structlog>=24.1"

Structured JSON logging is mandatory — a 304-heavy loop is invisible without it, and a port-authority audit needs to see exactly which cycles cost payload and which did not. Configure structlog once at process start:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),
    ]
)
log = structlog.get_logger("tos.conditional")

The loop is tuned entirely from environment, so cadence can be retuned per facility without a redeploy:

Step-by-step Implementation #

Step 1 — Cache validators per resource, not per client #

The ETag and Last-Modified a TOS returns are properties of one resource — /events for a specific facility, or a per-vessel work-queue endpoint — so they must be stored keyed by resource, never in a single shared slot. Sending the ETag from /events?facility=NLRTM on a request to a different endpoint guarantees a spurious 200 (the server sees a non-matching validator) and throws away the whole benefit. RFC 9110 also permits weak validators (W/"..."); they are perfectly valid in If-None-Match and must be echoed verbatim, prefix included, or the server will treat every poll as a cache miss.

from __future__ import annotations

from dataclasses import dataclass

import structlog

log = structlog.get_logger("tos.conditional")


@dataclass
class Validators:
    etag: str | None = None            # strong or weak (W/"..."), stored verbatim
    last_modified: str | None = None   # HTTP-date, echoed exactly as received


class ValidatorStore:
    """Per-resource cache of the latest ETag / Last-Modified validators."""

    def __init__(self) -> None:
        self._by_resource: dict[str, Validators] = {}

    def conditional_headers(self, resource: str) -> dict[str, str]:
        v = self._by_resource.get(resource)
        if v is None:
            return {}
        headers: dict[str, str] = {}
        if v.etag:
            headers["If-None-Match"] = v.etag           # preferred validator
        elif v.last_modified:
            headers["If-Modified-Since"] = v.last_modified
        return headers

    def update(self, resource: str, etag: str | None, last_modified: str | None) -> None:
        current = self._by_resource.setdefault(resource, Validators())
        if etag is not None:
            current.etag = etag
        if last_modified is not None:
            current.last_modified = last_modified
        log.debug("validators_stored", resource=resource,
                  etag=etag, has_last_modified=last_modified is not None)

Prefer If-None-Match over If-Modified-Since whenever an ETag is available: Last-Modified has one-second resolution, so two gate moves committed inside the same wall-clock second can share a timestamp and one will be missed by a date-only conditional. The ETag changes on every distinct representation and has no such blind spot.

Step 2 — Send the conditional request and classify the outcome #

One request now carries the stored validator and resolves to exactly one of four outcomes. A 304 is the cheap happy path — no body, no parse, no cost against the payload budget; a 200 refreshes the validator and yields a page; a 429 is an explicit throttle; anything else is an error the caller handles. Classifying here keeps the interval controller in Step 3 free of HTTP details.

from enum import Enum

import httpx


class PollOutcome(Enum):
    CHANGED = "changed"        # 200 with a fresh representation
    UNCHANGED = "unchanged"    # 304 Not Modified
    THROTTLED = "throttled"    # 429 Too Many Requests
    ERROR = "error"            # 4xx/5xx that is neither of the above


async def conditional_get(
    client: httpx.AsyncClient, resource: str, url: str,
    cursor: str, store: ValidatorStore,
) -> tuple[PollOutcome, httpx.Response]:
    headers = store.conditional_headers(resource)
    resp = await client.get(url, params={"cursor": cursor}, headers=headers)

    if resp.status_code == 304:
        outcome = PollOutcome.UNCHANGED
    elif resp.status_code == 200:
        store.update(resource, resp.headers.get("ETag"), resp.headers.get("Last-Modified"))
        outcome = PollOutcome.CHANGED
    elif resp.status_code == 429:
        outcome = PollOutcome.THROTTLED
    else:
        outcome = PollOutcome.ERROR

    log.info("conditional_get", resource=resource, status=resp.status_code,
             outcome=outcome.value, sent_conditional=bool(headers))
    return outcome, resp

Step 3 — Drive the interval with additive-increase / multiplicative-decrease #

The controller is the heart of the page: it treats each 304 as evidence the berth is quiet and grows the interval, and each 200 as evidence something is moving and shrinks it. Growing multiplicatively and shrinking multiplicatively (an AIMD-shaped loop, biased to react fast to change) means a resource that starts a discharge is polled aggressively within two or three cycles, while an idle facility relaxes toward the ceiling instead of hammering a 304 every fifteen seconds. Both ends are clamped to the phase-appropriate bounds the parent topic area defines — a busy vessel alongside never relaxes past min_interval, an off-peak resource never tightens past max_interval.

from dataclasses import dataclass


@dataclass
class IntervalController:
    """Adapt the poll interval to observed change rate, clamped to phase bounds."""

    interval: float
    min_interval: float
    max_interval: float
    speedup: float = 0.5     # multiplicative decrease on a 200
    backoff: float = 1.5     # multiplicative increase on a 304
    floor: float = 0.0       # server-imposed lower bound (Step 4)

    def observe(self, outcome: PollOutcome) -> float:
        if outcome is PollOutcome.CHANGED:
            self.interval = max(self.min_interval, self.interval * self.speedup)
        elif outcome is PollOutcome.UNCHANGED:
            self.interval = min(self.max_interval, self.interval * self.backoff)
        # ERROR / THROTTLED leave the base interval untouched; the caller
        # applies the server floor directly so a 429 cannot be sped up.
        effective = max(self.interval, self.floor)
        log.debug("interval_adjusted", outcome=outcome.value,
                  base_interval=round(self.interval, 1),
                  effective_interval=round(effective, 1))
        return effective

Step 4 — Honour Cache-Control and Retry-After as a hard floor #

Adaptive speed-up must never override an explicit server directive. A Retry-After on a 429 is a contractual back-off window; a Cache-Control: max-age tells the client the resource cannot change before that many seconds elapse, so polling sooner only burns budget on a guaranteed 304. Both become a floor under the controller’s computed interval — RFC 9110 allows Retry-After as either delta-seconds or an HTTP-date, so parse both forms and never let a malformed header raise into the loop.

import email.utils
import time


def server_floor(resp: httpx.Response) -> float:
    """Lower bound the next sleep must respect, from Retry-After or Cache-Control."""
    retry_after = resp.headers.get("Retry-After")
    if retry_after is not None:
        try:
            return max(0.0, float(retry_after))          # delta-seconds form
        except ValueError:
            parsed = email.utils.parsedate_to_datetime(retry_after)
            if parsed is not None:                        # HTTP-date form
                return max(0.0, parsed.timestamp() - time.time())

    cache_control = resp.headers.get("Cache-Control", "")
    for directive in cache_control.split(","):
        token = directive.strip().lower()
        if token.startswith("max-age="):
            try:
                return max(0.0, float(token.split("=", 1)[1]))
            except ValueError:
                return 0.0
    return 0.0

Step 5 — Compose the per-resource loop #

The loop wires the three parts together: one conditional request, a floor read from the response, a throttle short-circuit, then a controller-driven sleep. A 429 sleeps for the floor without touching the base interval, so the client stops leaning on a wall instead of gently speeding up into it; a 200 advances the cursor and hands the page to the typed validation boundary documented on the sibling REST page. Because every branch logs a structured event, a stalled resource shows up as an absence of conditional_get lines the alerting layer can page on.

The conditional-request loop and an explicit rate limiter are complementary, not redundant, and both belong in a production deployment. The interval controller reduces the number of requests a quiet resource makes; a shared token-bucket limiter, sized to the carrier’s published quota and drawn across every resource on the same credential, caps the aggregate rate when dozens of busy berths all speed up at once. Acquire a token immediately before each conditional_get and let the controller’s min_interval sit above the steady-state token refill so the two mechanisms never fight — the adaptive loop keeps individual resources honest, and the bucket keeps the whole facility integration inside the 429 boundary during a simultaneous multi-vessel rush.

import asyncio


async def poll_resource(
    client: httpx.AsyncClient, resource: str, url: str,
    store: ValidatorStore, controller: IntervalController,
) -> None:
    cursor = "0"
    while True:
        outcome, resp = await conditional_get(client, resource, url, cursor, store)
        controller.floor = server_floor(resp)

        if outcome is PollOutcome.THROTTLED:
            delay = controller.floor or controller.max_interval
            log.warning("throttled_backoff", resource=resource, delay=round(delay, 1))
            await asyncio.sleep(delay)
            continue

        if outcome is PollOutcome.CHANGED:
            cursor = resp.headers.get("X-Next-Cursor", cursor)
            # hand resp to the typed validation boundary (sibling REST page)

        await asyncio.sleep(controller.observe(outcome))

Edge Cases & Carrier Deviations #

A subtle TOS deviation: some port community systems return a fresh ETag on a 200 whose page is identical to the last one, because the validator is derived from a mutable server-side timestamp rather than the representation body. Left unchecked this defeats the 304 path entirely and the interval never grows. Guard against it by also carrying the monotonic sequence_id cursor from the sibling REST layer — if the max sequence has not advanced, treat the cycle as unchanged for controller purposes even when the HTTP status was 200, so a cosmetically rotating ETag cannot pin the loop at min_interval.

Verification & Testing #

Assert the two behaviours that define the page — the controller reacts correctly to each outcome and stays clamped, and the server floor parses both Retry-After forms — with pytest and synthetic httpx.Response objects. No live TOS is needed because every input is a header.

import httpx
import pytest


def _resp(status: int, headers: dict[str, str] | None = None) -> httpx.Response:
    return httpx.Response(status_code=status, headers=headers or {},
                          request=httpx.Request("GET", "https://tos/events"))


def test_interval_shrinks_on_change_and_grows_when_quiet() -> None:
    ctl = IntervalController(interval=60.0, min_interval=15.0, max_interval=900.0)
    assert ctl.observe(PollOutcome.CHANGED) == 30.0        # 60 * 0.5
    assert ctl.observe(PollOutcome.CHANGED) == 15.0        # clamped at min
    ctl.observe(PollOutcome.UNCHANGED)                     # 22.5
    assert ctl.interval == pytest.approx(22.5)


def test_interval_clamped_to_max_when_persistently_quiet() -> None:
    ctl = IntervalController(interval=700.0, min_interval=15.0, max_interval=900.0)
    assert ctl.observe(PollOutcome.UNCHANGED) == 900.0     # 1050 clamped to max


def test_server_floor_reads_retry_after_and_max_age() -> None:
    assert server_floor(_resp(429, {"Retry-After": "12"})) == 12.0
    assert server_floor(_resp(200, {"Cache-Control": "max-age=30"})) == 30.0
    assert server_floor(_resp(200, {})) == 0.0


def test_conditional_headers_are_resource_scoped() -> None:
    store = ValidatorStore()
    store.update("events:NLRTM", etag='W/"abc"', last_modified=None)
    assert store.conditional_headers("events:NLRTM") == {"If-None-Match": 'W/"abc"'}
    assert store.conditional_headers("events:DEHAM") == {}   # unseen resource

A healthy run emits one JSON line per cycle. A quiet resource looks like {"event": "conditional_get", "status": 304, "outcome": "unchanged", "sent_conditional": true, ...} followed by {"event": "interval_adjusted", "effective_interval": 90.0, ...}; the moment a discharge starts, the status flips to 200 and effective_interval collapses toward min_interval within two or three lines. Assert on those structured keys, never on a formatted string.

Frequently Asked Questions #

Related #

• Polling terminal operating systems via REST APIs — the on-the-wire request construction this control loop wraps.
• Backfilling terminal API gaps with AIS events — reconstructing the sequence gaps that outages and long 304 stretches can leave.
• Container Status Mapping Rules — deterministic state resolution the 200 pages feed into.
• Threshold Tuning for Alerts — turning missing-cycle telemetry into transient-versus-systemic alerts.
• AIS Data Stream Integration — the push-based feed that cross-checks landside pulls.

Up: Terminal API Polling Strategies — the cadence and fallback discipline this conditional-request loop plugs into.