Late-Arriving Telemetry and Watermarks in Geofence Streams

A geofence stream never receives its telemetry in the order it was recorded. Pings buffer on the device during coverage gaps, retry over lossy radio, and traverse different network paths, so a position recorded at t=10.0 can arrive after one recorded at t=11.2. The evaluator needs a rule for when it is safe to decide that a device’s history up to some instant is complete — that rule is a watermark, and how it treats the pings that arrive after it has passed determines whether a straggler is silently lost or correctly reconciled. This page sits under Streaming vs Batch Geofence Evaluation and the broader Core Architecture & Latency Constraints reference, and it answers one narrow question: given a watermark, what do you do with a late ping, and what does that choice cost in trigger correctness? It builds directly on the ordering discipline in event-time ordering and clock skew, which establishes why the two timelines must stay separate; here we quantify the lateness policy itself.

Concept and specification #

A watermark is a monotonically advancing timestamp asserting that no event with an event-time at or below it should still be expected. For a stream whose maximum observed event-time is $E_{\max }$ and an allowed-lateness slack $L$, the watermark is:

$$W(t)=\underset{s\le t}{\max }{E}_{\text{event}}(s)-L$$

Any ping whose event-time $e$ satisfies $e\le W(t)$ at its arrival is late: the watermark has already advanced past it, and the transition it might have affected has already been committed. The slack $L$ is the single tuning knob. Too small and the watermark races ahead of legitimately delayed pings, marking valid data late and dropping it; too large and every transition waits $L$ before it can commit, inflating latency for no correctness gain once $L$ exceeds the real lateness distribution. The correct $L$ is the P99 (not the maximum) of the observed event-to-ingest delay — sized so 99% of pings land in time and the rare stragglers are handled explicitly rather than paid for by everyone.

Two disposition policies exist for a ping that crosses the watermark late. Drop discards it — cheapest, but any transition it would have corrected is lost. Side-output diverts it to a separate reconciliation stream that can re-emit or retract a trigger out-of-band — more correct, at the cost of a second processing path and eventual-consistency semantics downstream. The policy is chosen per trigger class: a low-value idle ping is dropped, a compliance-boundary crossing is side-outputted.

Step-by-step implementation #

Prerequisites: Python 3.11+, standard library only for the core evaluator (heapq, dataclasses, time); prometheus_client for the counters. Input is a stream of dual-timestamp telemetry records (event_time_ns, ingest_ns) as established in the event-time ordering reference; output is committed transitions plus a side-output stream of late compliance pings.

1. Generate the watermark from the running maximum event-time. The watermark trails $E_{\max }$ by the allowed-lateness slack and only ever moves forward.

from __future__ import annotations

from dataclasses import dataclass


@dataclass(slots=True)
class Watermark:
    """Monotonic watermark: max event-time seen minus allowed lateness."""
    allowed_lateness_ns: int          # L, e.g. 500ms = 500_000_000
    _max_event_ns: int = 0

    def observe(self, event_time_ns: int) -> None:
        # Monotonic: a stale ping can never pull the watermark backward.
        if event_time_ns > self._max_event_ns:
            self._max_event_ns = event_time_ns

    @property
    def current_ns(self) -> int:
        return self._max_event_ns - self.allowed_lateness_ns

    def is_late(self, event_time_ns: int) -> bool:
        # <= is deliberate: a ping exactly at the watermark is already covered.
        return event_time_ns <= self.current_ns

Gotcha: advance the watermark from the event-time you just observed, not from wall-clock time. Tying it to wall-clock makes an idle stream drift its watermark forward and mark a resuming device’s backlog late en masse.

2. Route each ping through the watermark before it reaches the evaluator. On-time pings enter the reorder heap; late pings branch by trigger class.

import heapq

def ingest_ping(
    p: TelemetryPoint,
    wm: Watermark,
    heap: list[tuple[int, TelemetryPoint]],
    late_out: list[TelemetryPoint],
    late_counter,                      # prometheus Counter
) -> None:
    wm.observe(p.event_time_ns)        # update W first
    if wm.is_late(p.event_time_ns):
        late_counter.inc()
        if _is_compliance(p):          # correctness-critical → reconcile
            late_out.append(p)         # side-output path
        # else: drop (low-value idle ping) — nothing committed for it
        return
    heapq.heappush(heap, (p.event_time_ns, p))   # on-time → ordered buffer

3. Drain and evaluate everything the watermark has passed, in event order. Only a prefix guaranteed complete is committed, so no transition is resolved against a partial history.

def drain_and_commit(
    wm: Watermark,
    heap: list[tuple[int, TelemetryPoint]],
    evaluate,                          # callable: TelemetryPoint -> None
) -> int:
    committed = 0
    w = wm.current_ns
    while heap and heap[0][0] <= w:    # everything at/below W is final
        _, point = heapq.heappop(heap)
        evaluate(point)                # commit ENTER/EXIT/DWELL
        committed += 1
    return committed

Gotcha: never feed a side-outputted late ping back into heap. Re-inserting it below the watermark would reopen an already-committed transition; reconciliation must emit a compensating trigger out-of-band instead.

Benchmark and verification #

The change to measure is trigger correctness and late-drop rate before and after introducing a watermark with a P99-sized allowed-lateness slack, replaying the same captured stream of 2M pings from a 40k-device fleet with injected coverage gaps (a store-and-forward burst distribution). Correctness is the fraction of committed transitions that match the ground-truth transition log computed from the fully-ordered stream.

import time, statistics

def bench_correctness(stream, evaluate, ground_truth, runs: int = 20) -> dict[str, float]:
    lat_samples: list[float] = []
    matched = total = 0
    for _ in range(runs):
        t0 = time.perf_counter()
        emitted = list(evaluate(stream))          # committed transitions
        lat_samples.append((time.perf_counter() - t0) / len(stream) * 1e3)  # ms/ping
        matched += sum(1 for e in emitted if e in ground_truth)
        total += len(emitted)
    lat_samples.sort()
    return {
        "correctness": matched / total,
        "p95_ms": lat_samples[int(0.95 * runs)],
        "p99_ms": lat_samples[int(0.99 * runs)],
    }

The 500ms watermark lifts correctness from 96.2% to 99.97% — a 100× reduction in phantom transitions per 100k — while dropping only 0.3% of pings, and every dropped ping is a low-value one the side-output would have reconciled anyway. The P99-sized slack is the sweet spot: the 200ms window still drops 1.8% because it undercuts the real lateness tail, while widening past 500ms only adds latency. Verify against the ground-truth log before promoting: replay through a shadow evaluator, require correctness parity above 99.9%, and confirm late_drop_rate stays under the 0.5% budget before shifting live traffic.

Failure modes and edge cases #

• Watermark too tight → valid pings dropped. If $L$ is set below the real lateness P99, legitimately delayed pings cross the watermark and are marked late; a compliance crossing routed to drop instead of side-output is then silently lost. Size $L$ from the measured event-to-ingest P99, and alert when late_drop_rate exceeds 0.5% — a rising drop rate is the first sign the slack undercuts the tail.
• Idle stream drifts the watermark. With no new pings, a wall-clock-driven watermark advances and marks a resuming device’s entire backlog late. Advance the watermark only from observed event-times, and give silent devices a bounded idle flush rather than a running clock.
• Non-monotonic event-times. A clock that jumps backward would pull $E_{\max }$ and the watermark backward if unguarded, reopening committed transitions. The observe guard keeps $E_{\max }$ monotonic; clamp implausible stamps per the clock-skew discipline before they reach the watermark.
• Side-output backlog unbounded. A flood of late compliance pings can grow the reconciliation queue without limit. Bound it and apply backpressure exactly as the pipeline sheds on-path load; reconciliation is best-effort and must not stall the primary evaluator.
• NaN or empty event-time. A record with a missing or NaN event-time cannot be ordered at all. Reject it at ingest to a dead-letter path rather than letting it corrupt $E_{\max }$.

Related #

• Streaming vs Batch Geofence Evaluation — the parent execution-model decision the watermark policy sits inside.
• Event-Time Ordering & Clock Skew in Geofence Triggers — why the two timelines stay separate and how the reorder buffer feeds the watermark.
• Windowed vs Continuous Geofence Evaluation Trade-offs — how the same lateness slack interacts with window sizing.
• Core Architecture & Latency Constraints — where the allowed-lateness delay fits in the end-to-end latency budget.