Delivery Guarantees & Event Ordering

Spatial webhook pipelines rarely need global ordering, but they always need per-feature ordering: a feature-update event applied before its create, or an old create replayed after a delete, silently resurrects stale geometry — and the fix is a version-guarded idempotent consumer, not a stronger broker.

This topic sits within Queue Management, Retries & Delivery Guarantees, the section covering how spatial events move reliably from producer to consumer once you accept that networks drop, reorder, and duplicate messages.

Prerequisites

Before wiring ordering guarantees into your consumer, confirm your stack and mental model are in place. Check off each item:

• Python 3.11+ — required for tomllib, precise datetime handling, and modern asyncio task groups
• A partitioned broker — Kafka 3.x, Redis Streams, or equivalent that preserves order within a partition
• A stable ordering key per event — a feature ID, asset ID, or H3 cell that groups all mutations of one spatial entity
• pydantic 2.x — strict validation of observation_time and version before any state write
• A versioned state store — PostgreSQL/PostGIS row with a version column, or any store supporting conditional writes
• Clock discipline at the source — producers stamp observation_time; consumers never trust arrival time for ordering
• An idempotency layer — see Idempotency & Spatial Deduplication for the duplicate-suppression half of the problem

Architecture

The design separates three concerns that are often conflated: the delivery guarantee (how many times an event may arrive), the ordering key (which events must be sequenced relative to each other), and the staleness guard (how the consumer decides an arriving event is older than current state). Get these three right and the specific broker becomes an implementation detail.

Layer breakdown:

1. Delivery guarantee — the broker and consumer contract on how many times an event may surface: at-most-once (fire-and-forget, may lose events), at-least-once (retried until acknowledged, may duplicate), or effectively-once (at-least-once delivery plus an idempotent consumer). Spatial ingestion almost always targets effectively-once; losing a parcel edit is worse than reprocessing one.
2. Partitioning and ordering key — the producer sets a partition key equal to the spatial entity’s stable identifier (feature ID or H3 cell). All events for one entity land on one partition and preserve relative order there. Unrelated entities scale out across partitions. Partition strategy is covered in Broker Selection & Partitioning for Spatial Streams.
3. Staleness guard — even within a partition, a retry can reorder events across a rebalance or across two producers. The consumer stamps and compares a monotonic version (and observation_time as a tiebreaker) and applies an event only if it is newer than stored state.
4. Reconciliation buffer — for the narrow window where two events for the same key arrive nearly simultaneously and slightly out of order, a short per-key buffer sorts them before they hit the state store, avoiding needless rejections and re-fetches.

Step-by-Step Implementation

Step 1 — Model the Event with Ordering Metadata

Every spatial event must carry the two fields ordering depends on: observation_time (when the source measured the geometry, per the sensor clock) and version (a monotonic counter per feature). Validate both before the event reaches any state logic. Coordinates are EPSG:4326 (WGS84) per RFC 7946 (spec).

from datetime import datetime, timezone
from typing import Any, Literal
from pydantic import BaseModel, Field, field_validator


class SpatialEvent(BaseModel):
    feature_id: str                      # stable ordering key for one entity
    version: int = Field(ge=0)           # monotonic per feature_id
    op: Literal["create", "update", "delete"]
    observation_time: datetime           # source clock, NOT arrival time
    crs: str = "EPSG:4326"               # WGS84 per RFC 7946
    geometry: dict[str, Any] | None      # None only for delete

    @field_validator("observation_time")
    @classmethod
    def must_be_tz_aware(cls, v: datetime) -> datetime:
        # Naive timestamps make cross-source ordering ambiguous; force UTC.
        if v.tzinfo is None:
            raise ValueError("observation_time must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)

    @field_validator("geometry")
    @classmethod
    def geometry_present_unless_delete(cls, v, info):
        if info.data.get("op") != "delete" and not v:
            raise ValueError("geometry required for create/update")
        return v

The version field is what makes ordering deterministic when timestamps collide. Two edits made in the same millisecond by different operators still have a strict order if the producer assigns versions from a per-feature counter.

Step 2 — Publish with a Partition Key Equal to the Ordering Key

Ordering is only preserved within a partition, so the producer must route every event for one feature to the same partition by keying on feature_id. Using the H3 cell as the key instead groups all edits within a cell — appropriate when your consumer’s state is per-cell rather than per-feature. H3, S2, and Quadkey tradeoffs are compared in H3 vs S2 vs Quadkey for Spatial Partitioning.

import json
from aiokafka import AIOKafkaProducer


async def publish_event(producer: AIOKafkaProducer, event: SpatialEvent) -> None:
    """
    Publish keyed by feature_id so all mutations of one entity share a
    partition and therefore a preserved relative order. Kafka hashes the
    key to select the partition; the same key always maps to the same one.
    """
    payload = event.model_dump_json().encode("utf-8")
    key = event.feature_id.encode("utf-8")   # <-- ordering key == partition key
    await producer.send_and_wait("spatial-events", value=payload, key=key)

Do not key on a random UUID or the event ID — that scatters a feature’s events across partitions and destroys any ordering the broker could have given you. The single most common ordering bug is a partition key that is finer-grained than the entity you actually need ordered.

Step 3 — Reject Stale Events in an Idempotent Consumer

The core of the design: before applying an event, compare its version (with observation_time as a tiebreaker) against the version already stored for that feature. Apply only if strictly newer; otherwise drop it as stale. This single guard neutralizes both duplicates and out-of-order arrivals, and it is the pattern detailed further in Idempotent Consumers for Out-of-Order Spatial Events.

import asyncpg


async def apply_event(pool: asyncpg.Pool, event: SpatialEvent) -> str:
    """
    Conditionally upsert feature state. The WHERE clause makes the write a
    no-op when an equal-or-newer version already exists, so a replayed
    create (v1) arriving after an update (v2) cannot resurrect old geometry.
    Returns 'applied' or 'stale'.
    """
    geom_json = json.dumps(event.geometry) if event.geometry else None
    row = await pool.fetchrow(
        """
        INSERT INTO feature_state
            (feature_id, version, op, observation_time, geometry)
        VALUES ($1, $2, $3, $4, $5)
        ON CONFLICT (feature_id) DO UPDATE
            SET version          = EXCLUDED.version,
                op               = EXCLUDED.op,
                observation_time = EXCLUDED.observation_time,
                geometry         = EXCLUDED.geometry
            WHERE EXCLUDED.version > feature_state.version
               OR (EXCLUDED.version = feature_state.version
                   AND EXCLUDED.observation_time > feature_state.observation_time)
        RETURNING version
        """,
        event.feature_id, event.version, event.op,
        event.observation_time, geom_json,
    )
    return "applied" if row is not None else "stale"

The conditional WHERE on the ON CONFLICT branch is doing the real work: PostgreSQL performs the comparison and the write atomically, so two concurrent consumers processing the same feature cannot interleave a lost update. When apply_event returns "stale", the event was older than current state and was correctly ignored — this is the moment a phantom geometry state would otherwise have been created.

Step 4 — Buffer Near-Simultaneous Events per Key

A version guard rejects genuinely stale events, but for two events that arrive within milliseconds slightly out of order, rejection forces the producer to re-send. A short per-key reordering buffer sorts a small window by version before applying, so the natural case resolves without redelivery.

import asyncio
from collections import defaultdict


class PerKeyOrderingBuffer:
    """
    Holds events per feature_id for up to `window` seconds, then flushes them
    in version order. Bounds reordering work to one small buffer per active
    key rather than sorting the whole stream (which would need global order).
    """

    def __init__(self, window: float = 0.25) -> None:
        self.window = window
        self._buffers: dict[str, list[SpatialEvent]] = defaultdict(list)
        self._timers: dict[str, asyncio.Task] = {}

    async def submit(self, event: SpatialEvent, apply) -> None:
        key = event.feature_id
        self._buffers[key].append(event)
        # (Re)start the flush timer for this key only.
        if key in self._timers:
            self._timers[key].cancel()
        self._timers[key] = asyncio.create_task(self._flush_later(key, apply))

    async def _flush_later(self, key: str, apply) -> None:
        try:
            await asyncio.sleep(self.window)
        except asyncio.CancelledError:
            return
        events = sorted(
            self._buffers.pop(key, []),
            key=lambda e: (e.version, e.observation_time),
        )
        self._timers.pop(key, None)
        for event in events:          # apply() is itself the Step 3 guard
            await apply(event)

Keep the window tight — 100-500 ms is typical. The buffer is an optimization that reduces redelivery churn, not a correctness mechanism: the Step 3 version guard remains the authority, so even if the buffer flushes out of order, stale events are still rejected downstream.

Spatial Validation and Error Handling

Ordering metadata can be valid while the geometry is not. Validate topology and CRS after the version guard admits an event but before it becomes stored state, so you never persist an invalid shape or waste validation cycles on events you will reject.

from shapely.geometry import shape
from shapely.validation import explain_validity
from pyproj import CRS


def validate_spatial_payload(event: SpatialEvent) -> None:
    """
    Confirm CRS is the expected EPSG:4326 (WGS84) and the geometry is
    topologically valid before it is written as current feature state.
    Deletes carry no geometry and are skipped.
    """
    if event.op == "delete":
        return
    if CRS.from_user_input(event.crs).to_epsg() != 4326:
        raise ValueError(f"expected EPSG:4326 (WGS84), got {event.crs!r}")

    geom = shape(event.geometry)
    if geom.is_empty:
        raise ValueError("geometry empty after parsing")
    if not geom.is_valid:
        raise ValueError(f"invalid topology: {explain_validity(geom)}")

Route validation failures to a dead-letter path rather than dropping them, so a malformed producer does not silently lose data. A rejected-as-stale event is a normal, expected outcome and should be counted, not alerted on; a rejected-as-invalid event is a producer defect and deserves an alert. Distinguishing the two in your metrics is what keeps on-call noise low.

Retry, Backoff, and Delivery Guarantees

At-least-once delivery means the broker will redeliver on any un-acknowledged event, and those redeliveries are exactly what reorder your stream. The consumer must therefore be safe to run repeatedly on the same event — which the Step 3 guard already guarantees — and retries of processing failures should use exponential backoff with jitter to avoid synchronized retry storms across partitions. The dedicated mechanics live in Exponential Backoff & Jitter for Spatial Webhooks.

import random


async def consume_with_retry(buffer: PerKeyOrderingBuffer, pool, event: SpatialEvent,
                             max_attempts: int = 5, base: float = 0.3) -> None:
    """
    Apply an event through the ordering buffer with retry. Because apply_event
    is version-guarded, re-running it after a transient failure is safe:
    a duplicate or reordered redelivery converges to the same final state.
    """
    for attempt in range(1, max_attempts + 1):
        try:
            validate_spatial_payload(event)
            await buffer.submit(event, lambda e: apply_event(pool, e))
            return
        except ValueError:
            raise                                   # invalid payload: do not retry, dead-letter
        except (asyncpg.PostgresError, ConnectionError) as exc:
            if attempt == max_attempts:
                raise
            delay = random.uniform(0, base * (2 ** attempt))   # full jitter
            await asyncio.sleep(delay)

The guarantees form a ladder — pick the lowest rung that meets your correctness bar, since each step up costs latency or throughput:

For nearly every geospatial webhook pipeline, effectively-once with a per-key version guard is the correct target. It requires no distributed transaction coordinator, tolerates the reordering that retries inevitably introduce, and — crucially for spatial data — makes it impossible for an old create or update to overwrite newer geometry. When two concurrent edits genuinely conflict (not merely arrive out of order), version comparison alone is insufficient and you must escalate to a merge policy; those semantics are covered in Conflict Resolution Strategies.

Verification

Prove the guard behaves under the exact failure it exists to stop: a create arriving after the update that superseded it. The final state must reflect the newest version regardless of arrival order.

import pytest
from datetime import datetime, timezone


def _evt(version: int, op: str, secs: int) -> SpatialEvent:
    return SpatialEvent(
        feature_id="F7",
        version=version,
        op=op,
        observation_time=datetime(2026, 7, 13, 12, 0, secs, tzinfo=timezone.utc),
        geometry={"type": "Point", "coordinates": [-0.1276, 51.5074]},  # EPSG:4326
    )


@pytest.mark.asyncio
async def test_stale_create_after_update_is_rejected(pg_pool):
    # In-order: create v1 then update v2 both apply.
    assert await apply_event(pg_pool, _evt(1, "create", 0)) == "applied"
    assert await apply_event(pg_pool, _evt(2, "update", 1)) == "applied"

    # Out-of-order redelivery: the create (v1) arrives LAST.
    assert await apply_event(pg_pool, _evt(1, "create", 0)) == "stale"

    row = await pg_pool.fetchrow(
        "SELECT version, op FROM feature_state WHERE feature_id = 'F7'"
    )
    assert row["version"] == 2 and row["op"] == "update", (
        "stored state must remain at v2 — the late create must not resurrect old geometry"
    )


@pytest.mark.asyncio
async def test_duplicate_delivery_is_idempotent(pg_pool):
    assert await apply_event(pg_pool, _evt(3, "update", 2)) == "applied"
    assert await apply_event(pg_pool, _evt(3, "update", 2)) == "stale"  # exact dup

Run with pytest -v --asyncio-mode=auto against a containerized PostgreSQL with the feature_state table and its feature_id primary key. The first test is the one that matters: without the conditional WHERE, the replayed create returns "applied" and the assertion on the stored version fails — the precise signature of a phantom geometry state reaching production.

Troubleshooting

FAQ

Related

• Queue Management, Retries & Delivery Guarantees — the parent section covering reliable movement of spatial events from producer to consumer
• Idempotent Consumers for Out-of-Order Spatial Events — a deep dive on the version-guarded consumer pattern used in Step 3
• Broker Selection & Partitioning for Spatial Streams — choosing a broker and a partition key that preserves per-key order at scale
• Conflict Resolution Strategies — merge policies for genuinely concurrent edits that a version guard alone cannot resolve
• Idempotency & Spatial Deduplication — the duplicate-suppression discipline that pairs with ordering to reach effectively-once

Explore this section