Handling Late and Out-of-Order Spatial Events

Spatial events arrive late for ordinary reasons: a device buffered while out of coverage, a gateway retried, a mobile network delayed a batch, a clock drifted. None of that matters for a per-event check — a geometry is valid whenever it arrives. It matters enormously for any rule that compares events with each other, because “these two vehicles were within fifty metres” is a claim about a set of events, and a late arrival changes the set. This guide covers the mechanics: measuring the lateness distribution, choosing watermarks and allowed lateness from evidence rather than instinct, evaluating windowed spatial rules, and publishing a finality contract so downstream consumers know when a result has stopped changing. It completes the windowing half of Streaming Spatial Data Validation.

Prerequisites

• A stream processor with event-time semantics — Flink, Kafka Streams, or Spark Structured Streaming. The concepts below exist in all three under slightly different names.
• Events carrying a trustworthy event timestamp distinct from the ingestion timestamp. If only ingestion time is available, none of this applies: you have processing-time windows and should say so.
• Python 3.10+ with pandas 2.x and shapely 2.0+ for the offline lateness analysis and the windowed predicate examples.
• A source identifier on every event. Lateness is a property of the source, and a single pipeline-wide figure derived from mixed sources is dominated by the worst one.

Step-by-Step Procedure

Step 1 — Measure lateness before choosing any parameter

# lateness/step1_measure.py
import numpy as np
import pandas as pd


def lateness_profile(df: pd.DataFrame) -> pd.DataFrame:
    """Distribution of (arrival - event) time, per source."""
    out = df.copy()
    out["event_ts"] = pd.to_datetime(out["event_ts"], utc=True)
    out["arrival_ts"] = pd.to_datetime(out["arrival_ts"], utc=True)
    out["lateness_s"] = (out["arrival_ts"] - out["event_ts"]).dt.total_seconds()

    negative = out["lateness_s"] < -1
    if negative.any():
        # Events "arriving before they happened" are clock skew, not lateness.
        out.loc[negative, "lateness_s"] = np.nan

    return out.groupby("source_id")["lateness_s"].agg(
        n="size",
        median="median",
        p95=lambda s: float(np.nanpercentile(s, 95)),
        p99=lambda s: float(np.nanpercentile(s, 99)),
        p999=lambda s: float(np.nanpercentile(s, 99.9)),
        worst="max",
        clock_skew_events=lambda s: int(s.isna().sum()),
    ).round(1)

Verification: the table should show a clear separation between well-connected sources (median under a second, p99 in single-digit seconds) and store-and-forward sources (p99 in minutes or hours). If one source dominates the tail, consider giving it its own pipeline rather than degrading the whole stream’s latency to accommodate it.

Step 2 — Derive the parameters from the measurement

# lateness/step2_parameters.py
from dataclasses import dataclass


@dataclass(frozen=True)
class WindowPolicy:
    window_s: int
    watermark_delay_s: int      # how long after the window end we wait before firing
    allowed_lateness_s: int     # how long the window state is kept after firing
    source_id: str

    @property
    def finality_s(self) -> int:
        """Time from window end until the result can no longer change."""
        return self.watermark_delay_s + self.allowed_lateness_s


def policy_from_profile(row, window_s: int = 60) -> WindowPolicy:
    """p99 sets the watermark; the gap to p99.9 sets allowed lateness."""
    watermark = int(max(5, row["p99"]))
    allowed = int(max(0, row["p999"] - row["p99"]))
    return WindowPolicy(window_s=window_s, watermark_delay_s=watermark,
                        allowed_lateness_s=allowed, source_id=row.name)

Verification: compute finality_s for each source and check it against the latency budget the consumers were promised. If finality exceeds the budget, something has to give — a shorter watermark that accepts more late events, or a renegotiated budget. Discovering that tension here is much cheaper than discovering it in production.

Step 3 — Evaluate the windowed spatial rule

# lateness/step3_window.py
from collections import defaultdict

from shapely import STRtree


class ProximityWindow:
    """Accumulate events per window, evaluate proximity when the window fires."""

    def __init__(self, policy, threshold_m: float = 50.0):
        self.policy = policy
        self.threshold_m = threshold_m
        self.buffers: dict[int, list] = defaultdict(list)
        self.fired: set[int] = set()

    def window_id(self, event_ts: float) -> int:
        return int(event_ts // self.policy.window_s)

    def add(self, event: dict) -> str:
        wid = self.window_id(event["event_ts"])
        if wid in self.fired:
            window_end = (wid + 1) * self.policy.window_s
            if event["arrival_ts"] - window_end > self.policy.finality_s:
                return "too_late"          # side output; the window is closed for good
            self.buffers[wid].append(event)
            return "late_revision"
        self.buffers[wid].append(event)
        return "on_time"

    def evaluate(self, wid: int) -> list[dict]:
        events = self.buffers[wid]
        geoms = [e["geom"] for e in events]
        tree = STRtree(geoms)
        violations = []
        for i, geom in enumerate(geoms):
            for j in tree.query(geom.buffer(self.threshold_m)):
                if j <= i:
                    continue
                d = geom.distance(geoms[j])
                if d <= self.threshold_m:
                    violations.append({
                        "window_id": wid,
                        "a": events[i]["entity_id"], "b": events[j]["entity_id"],
                        "distance_m": round(float(d), 2),
                    })
        self.fired.add(wid)
        return violations

Verification: feed the same events in two different orders and confirm the evaluated result is identical. Order-independence within a window is the property that makes late arrivals safe to fold in; if a rule’s result depends on arrival order, it is not a windowed rule, it is a sequence rule and belongs in the per-entity stage.

Step 4 — Emit revisions, not silent corrections

# lateness/step4_revision.py
import json


def publish(producer, topic: str, wid: int, violations: list[dict],
            revision: int, is_final: bool) -> None:
    """Every emission names its window and its revision number."""
    payload = {
        "window_id": wid,
        "revision": revision,
        "final": is_final,
        "violation_count": len(violations),
        "violations": violations,
    }
    producer.produce(
        topic,
        key=str(wid).encode(),        # keyed by window: compaction keeps the latest
        value=json.dumps(payload).encode(),
        headers=[("final", b"1" if is_final else b"0")],
    )

Verification: consume the output topic with log compaction enabled and confirm that only the newest revision per window survives. Keying by window identifier is what makes revisions supersede rather than accumulate — an output topic where revision 1 and revision 3 both persist is one where every consumer must implement its own deduplication.

Step 5 — Publish the finality contract

# lateness/step5_contract.py
def contract(policy) -> dict:
    """The document downstream teams need in order to consume windowed results safely."""
    return {
        "source_id": policy.source_id,
        "window_seconds": policy.window_s,
        "first_result_after_window_end_s": policy.watermark_delay_s,
        "may_be_revised_until_s": policy.finality_s,
        "beyond_that": "events are counted on the too_late side output and folded in "
                       "by the nightly batch pass, never into the streamed window",
        "revision_key": "window_id",
        "final_flag": "header 'final' = 1 marks the last revision for a window",
    }

Verification: hand the contract to a downstream consumer team and ask what they would do with a revision. If the answer is “we would not notice it”, the integration is not complete regardless of what the pipeline does.

Interpreting Results

The most useful single metric is the revision rate: the share of windows whose result changed after the first emission. A revision rate near zero means the watermark is generous and finality could be earlier. A high revision rate means downstream consumers are acting on numbers that keep moving, which is worse than waiting slightly longer for a stable answer.

Gotchas & Edge Cases

Processing-time windows hide the problem instead of solving it. They always produce a stable, prompt answer — about an arbitrary set of events. For spatial rules this is actively misleading: two vehicles that were genuinely close will land in different windows if one of them reported late, and the violation simply never appears.

A single global watermark is set by the worst source. Mixing a wired sensor network with store-and-forward mobile devices in one watermark means the sensor results wait hours for the mobile tail. Separate the streams or accept the latency; there is no third option that is also correct.

Allowed lateness costs state. Keeping a window’s contents open for six hours means holding six hours of events in the state store for every open window. Size the state from finality_s × event rate before choosing generous parameters.

Clock skew is not lateness and must not be treated as such. Events with an event time in the future arrive “early”, advance the watermark prematurely, and cause every genuinely on-time event behind them to be classified as late. Clamp future-dated events at ingestion and report them as a device fault.

Window boundaries create artificial edges for spatial rules. Two vehicles close together at 10:59:58 and 11:00:02 fall in different one-minute windows and no proximity is reported. Sliding windows with an overlap of at least the interaction timescale solve it; tumbling windows do not, and the resulting misses are invisible.

Reprocessing history uses different parameters. A batch replay of last month’s events has no lateness at all — every event is available immediately. Running the same windowed rule over history therefore produces results the live stream never produced, which is expected and must be explained when the two are compared.

When to Escalate

• Finality cannot meet the consumer’s latency budget — that is a design conversation involving the consumer, not a parameter to be quietly tightened. Tightening it means dropping real events.
• One source’s lateness profile degrades persistently — escalate to the source owner with the measured distribution. A device fleet that has started buffering for hours is a fleet-management issue, and no downstream parameter fixes it.
• Revision rates above a few percent suggest the streaming window is the wrong instrument for that rule. Move it to the batch pass described in Batch Processing Large Spatial Datasets and let the stream keep the rules it can answer definitively.
• Disagreement between streamed and batch results should always be reconcilable by the too_late counts. If it is not, something is being dropped silently and that is a correctness bug, not a tuning question.

Frequently Asked Questions

Related

• Streaming Spatial Data Validation — rule classification and where windowing fits
• Running Geometry Checks in a Kafka Consumer — the stateless stage that runs before any window
• Observability and Lineage for Validation — emitting per-window lineage so revisions stay auditable

Back to Streaming Spatial Data Validation