Backfilling terminal API gaps with AIS events #

When a Terminal Operating System (TOS) API goes dark — a maintenance window, a dropped credential, a silent 503 that never recovers — the container timeline stops advancing even though vessels keep berthing and sailing. Backfilling terminal API gaps with AIS events closes that hole: a staleness watermark detects that the TOS feed has stalled, an AIS-derived deriver reconstructs berth arrival and departure from navigational status and a berth geofence, and those reconstructed milestones enter the timeline as explicitly provisional, low-confidence events until the authoritative TOS recovers and overrides them. The goal is a continuous container-tracking record that never fabricates certainty it does not have.

Architecture Alignment #

This page is the degraded-mode counterpart of Terminal API Polling Strategies, the pull-side ingestion discipline inside the Container Tracking & AIS Event Synchronization domain. When the healthy path — the conditional, cursor-based loop described in adaptive polling with ETag conditional requests — stops returning fresh events, backfill takes over as the tertiary confidence tier. It leans entirely on the push-based vessel feed built in AIS Data Stream Integration: the same validated !AIVDM position and voyage stream that normally corroborates a terminal move is, during an outage, the only signal left. Backfill is deliberately advisory — it keeps dashboards and provisional yard planning alive without ever actuating a gate release, and every event it writes carries the provenance needed to unwind it cleanly on recovery.

Prerequisites & Environment Setup #

The backfill service targets Python 3.11+ (for datetime.UTC, exception groups, and the StrEnum used to tag provenance). It consumes the already-validated AIS stream, so it needs only the typed-modelling and structured-logging stack plus a Redis client for the shared cursor and timeline store:

pip install "structlog>=24.1" "redis>=5.0" pydantic

Structured JSON logging is mandatory here — bare print() and unstructured logging strings are unacceptable because a port authority auditing an outage must be able to prove exactly which timeline entries were AIS-derived, when they were superseded, and by which authoritative event. Configure structlog once, at service boot:

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("maritime.backfill")

Environment variables the service expects:

Set TOS_STALENESS_SECONDS from the alongside poll cadence, not a wall-clock guess: a berth polled every 15–30 s is genuinely stale after roughly three missed cycles, whereas an off-peak berth polled every 10 minutes is not. Backfill that fires too eagerly floods the timeline with provisional churn.

Step-by-step Implementation #

Step 1 — Detect a TOS gap with a staleness watermark #

A gap is not “the last request failed” — a single 429 is normal and the retry layer already handles it. A gap is the absence of fresh authoritative state for longer than the operational phase tolerates. Track the newest committed event_ts per berth as a watermark, and open a gap only when the watermark ages past TOS_STALENESS_SECONDS. Critically, a 304 Not Modified response must not advance the watermark — nothing changed, so the last real event is still the freshest thing you know. Advancing on a 304 would mask a genuine outage behind a healthy-looking cursor.

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timedelta

import structlog

log = structlog.get_logger("maritime.backfill.gap")


@dataclass
class StalenessWatermark:
    """Tracks the newest authoritative TOS event per berth to detect polling gaps."""

    max_staleness: timedelta
    _last_seen: dict[str, datetime] = field(default_factory=dict)

    def observe(self, berth_id: str, event_ts: datetime) -> None:
        # Only real events move the watermark; a 304 must never call this.
        prev = self._last_seen.get(berth_id)
        if prev is None or event_ts > prev:
            self._last_seen[berth_id] = event_ts

    def gap_open(self, berth_id: str, now: datetime) -> bool:
        last = self._last_seen.get(berth_id)
        if last is None:
            return False  # cold start, never observed — not a gap
        stale_for = now - last
        is_open = stale_for > self.max_staleness
        if is_open:
            log.warning(
                "tos_gap_detected",
                berth_id=berth_id,
                last_event_ts=last.isoformat(),
                stale_seconds=round(stale_for.total_seconds(), 1),
                threshold_seconds=self.max_staleness.total_seconds(),
            )
        return is_open

Step 2 — Derive berth arrival and departure from AIS nav status + geofence #

With a gap open, reconstruct the berthing milestone from the AIS feed. ITU-R M.1371 encodes navigational status as an integer; 5 means moored and 1 means at anchor. A berth arrival is the transition into moored while the reported position sits inside the berth geofence; a departure is the transition back out. The geofence is a haversine radius around the berth centroid — tight enough (typically 120–160 m) that the anchorage waiting area does not register as alongside. Nav status alone is never sufficient: crews routinely leave the status stale, so the geofence dwell is the tiebreaker.

from dataclasses import dataclass
from datetime import datetime
from enum import StrEnum
from math import asin, cos, radians, sin, sqrt

import structlog

log = structlog.get_logger("maritime.backfill.derive")

NAV_MOORED: int = 5   # ITU-R M.1371 navigational status
_EARTH_M: float = 6_371_000.0


class BerthEvent(StrEnum):
    ARRIVAL = "BERTH_ARRIVAL"
    DEPARTURE = "BERTH_DEPARTURE"


@dataclass(frozen=True)
class AISFix:
    mmsi: int
    nav_status: int
    lat: float
    lon: float
    ts: datetime


def _inside(lat: float, lon: float, fence: tuple[float, float, float]) -> bool:
    clat, clon, radius_m = fence
    dlat, dlon = radians(lat - clat), radians(lon - clon)
    a = sin(dlat / 2) ** 2 + cos(radians(clat)) * cos(radians(lat)) * sin(dlon / 2) ** 2
    return (2 * _EARTH_M * asin(sqrt(a))) <= radius_m


def derive_berth_event(
    prev: AISFix | None, cur: AISFix, fence: tuple[float, float, float]
) -> BerthEvent | None:
    cur_alongside = cur.nav_status == NAV_MOORED and _inside(cur.lat, cur.lon, fence)
    prev_alongside = (
        prev is not None
        and prev.nav_status == NAV_MOORED
        and _inside(prev.lat, prev.lon, fence)
    )
    if cur_alongside and not prev_alongside:
        log.info("berth_arrival_derived", mmsi=cur.mmsi, ts=cur.ts.isoformat())
        return BerthEvent.ARRIVAL
    if prev_alongside and not cur_alongside:
        log.info("berth_departure_derived", mmsi=cur.mmsi, ts=cur.ts.isoformat())
        return BerthEvent.DEPARTURE
    return None

Step 3 — Emit provenance-tagged provisional events #

A backfilled event must never be indistinguishable from an authoritative one. Every timeline entry carries its provenance, a confidence score, and a provisional flag, plus a deterministic idempotency_key. The key buckets the timestamp to the minute so that re-deriving the same arrival from a slightly later AIS fix collapses onto the same entry instead of duplicating it — the same idempotency discipline the healthy polling loop applies to redelivered pages. AIS-derived events are stamped at confidence 0.50, matching the advisory tier of the polling fallback cascade.

import hashlib
import os
from dataclasses import dataclass
from datetime import datetime
from enum import StrEnum

import structlog

log = structlog.get_logger("maritime.backfill.timeline")
AIS_CONFIDENCE: float = float(os.environ.get("AIS_DERIVED_CONFIDENCE", "0.50"))


class Provenance(StrEnum):
    TOS_AUTHORITATIVE = "TOS_AUTHORITATIVE"
    AIS_DERIVED = "AIS_DERIVED"


@dataclass(frozen=True)
class TimelineEvent:
    berth_id: str
    imo: str
    event_type: str
    event_ts: datetime
    provenance: Provenance
    confidence: float
    provisional: bool

    @property
    def idempotency_key(self) -> str:
        # Identity = (berth, vessel, milestone, minute-bucket) — stable across re-derivation.
        bucket = self.event_ts.strftime("%Y%m%dT%H%M")
        raw = f"{self.berth_id}|{self.imo}|{self.event_type}|{bucket}"
        return hashlib.sha256(raw.encode()).hexdigest()[:16]


def provisional_from_ais(
    berth_id: str, imo: str, event_type: str, ts: datetime
) -> TimelineEvent:
    event = TimelineEvent(
        berth_id=berth_id,
        imo=imo,
        event_type=event_type,
        event_ts=ts,
        provenance=Provenance.AIS_DERIVED,
        confidence=AIS_CONFIDENCE,
        provisional=True,
    )
    log.info(
        "provisional_event_emitted",
        idempotency_key=event.idempotency_key,
        provenance=event.provenance,
        confidence=event.confidence,
        event_type=event_type,
    )
    return event

Step 4 — Reconcile when the TOS recovers #

When polling resumes, the authoritative TOS event must supersede the provisional AIS-derived stand-in — but only the matching one, and idempotently. Reconciliation looks for a provisional event with the same berth, vessel, and milestone whose timestamp falls inside RECONCILE_WINDOW_SECONDS of the authoritative time (AIS typically fires the arrival a few minutes before the gate system records first-line-ashore). The provisional entry is deleted and the authoritative entry committed. A replay of an authoritative event that is already committed is a no-op, so at-least-once delivery from the recovered feed never corrupts the timeline.

import os
from datetime import timedelta

import structlog

log = structlog.get_logger("maritime.backfill.reconcile")
RECONCILE_WINDOW = timedelta(seconds=float(os.environ.get("RECONCILE_WINDOW_SECONDS", "2700")))


def reconcile(
    timeline: dict[str, TimelineEvent], authoritative: TimelineEvent
) -> dict[str, TimelineEvent]:
    key = authoritative.idempotency_key
    existing = timeline.get(key)

    # Idempotent replay of an already-authoritative milestone: nothing to do.
    if existing is not None and existing.provenance is Provenance.TOS_AUTHORITATIVE:
        log.debug("reconcile_noop_duplicate", idempotency_key=key)
        return timeline

    superseded = [
        k
        for k, ev in timeline.items()
        if ev.provisional
        and ev.berth_id == authoritative.berth_id
        and ev.imo == authoritative.imo
        and ev.event_type == authoritative.event_type
        and abs(ev.event_ts - authoritative.event_ts) <= RECONCILE_WINDOW
    ]
    for k in superseded:
        log.info(
            "provisional_superseded",
            provisional_key=k,
            authoritative_key=key,
            ts_delta_seconds=round(
                (timeline[k].event_ts - authoritative.event_ts).total_seconds(), 1
            ),
        )
        del timeline[k]

    timeline[key] = authoritative
    log.info(
        "authoritative_committed",
        idempotency_key=key,
        superseded_count=len(superseded),
        confidence=authoritative.confidence,
    )
    return timeline

Edge Cases & Carrier Deviations #

A recurring trap is the stale navigational status field. Many Class B and older Class A transponders leave nav_status at 0 (under way using engine) even while the vessel is alongside, because the watch officer never updated it. If your deriver trusts nav status alone it will miss the arrival entirely. Treat the geofence dwell as authoritative for position and use nav status only as a corroborating signal — an in-geofence vessel with a low or zero speed-over-ground for a sustained window is berthed regardless of what the status byte claims.

Verification & Testing #

Assert the three load-bearing behaviours — gap detection, arrival derivation, and authoritative supersession — with fixed fixtures and pytest. Use timezone-aware datetime objects so the reconcile window is exercised against real UTC deltas.

from datetime import datetime, timedelta, timezone

import structlog

UTC = timezone.utc


def test_gap_opens_only_past_threshold() -> None:
    wm = StalenessWatermark(max_staleness=timedelta(seconds=180))
    t0 = datetime(2026, 7, 17, 8, 0, tzinfo=UTC)
    wm.observe("NLRTM-EMX-07", t0)
    assert wm.gap_open("NLRTM-EMX-07", t0 + timedelta(seconds=120)) is False
    assert wm.gap_open("NLRTM-EMX-07", t0 + timedelta(seconds=240)) is True


def test_arrival_derived_on_moored_inside_fence() -> None:
    fence = (51.9500, 4.0200, 150.0)  # centroid + 150 m radius
    away = AISFix(mmsi=244660000, nav_status=0, lat=51.9600, lon=4.0400, ts=datetime.now(UTC))
    at = AISFix(mmsi=244660000, nav_status=5, lat=51.9501, lon=4.0201, ts=datetime.now(UTC))
    assert derive_berth_event(away, at, fence) is BerthEvent.ARRIVAL
    assert derive_berth_event(at, at, fence) is None  # no transition


def test_authoritative_supersedes_provisional() -> None:
    ts = datetime(2026, 7, 17, 9, 30, tzinfo=UTC)
    prov = provisional_from_ais("NLRTM-EMX-07", "9321483", "BERTH_ARRIVAL", ts)
    timeline = {prov.idempotency_key: prov}
    authoritative = TimelineEvent(
        "NLRTM-EMX-07", "9321483", "BERTH_ARRIVAL",
        ts + timedelta(minutes=6), Provenance.TOS_AUTHORITATIVE, 0.97, False,
    )
    merged = reconcile(timeline, authoritative)
    assert all(not ev.provisional for ev in merged.values())
    assert authoritative.idempotency_key in merged


def test_reconcile_replay_is_idempotent() -> None:
    cap = structlog.testing.LogCapture()
    structlog.configure(processors=[cap])
    ts = datetime(2026, 7, 17, 9, 36, tzinfo=UTC)
    auth = TimelineEvent("NLRTM-EMX-07", "9321483", "BERTH_ARRIVAL", ts,
                         Provenance.TOS_AUTHORITATIVE, 0.97, False)
    timeline = reconcile({}, auth)
    reconcile(timeline, auth)  # replay
    assert cap.entries[-1]["event"] == "reconcile_noop_duplicate"

A healthy backfill run emits one JSON line per derived event and one per supersession — for example {"event": "provisional_superseded", "provisional_key": "…", "authoritative_key": "…", "ts_delta_seconds": -360.0, "log_level": "info", "timestamp": "…"}. Assert on the structured keys, never on a formatted string.

Frequently Asked Questions #

Related #

• Terminal API Polling Strategies — the pull-side ingestion loop whose outages this backfill covers.
• Adaptive polling with ETag conditional requests — the healthy conditional-request loop whose 304 responses must not mask a gap.
• AIS Data Stream Integration — the validated vessel feed the deriver reads nav status and position from.
• Container Status Mapping Rules — the resolver that fuses provisional and authoritative events into committed container state.
• Threshold Tuning for Alerts — geofence dwell and hysteresis tuning that keeps derived berth events from flapping.

Up: Terminal API Polling Strategies — the topic area governing how landside services pull and reconcile Terminal Operating System state.