Using a Transactional Outbox for Spatial Events

Write the feature row and the outbox row in the same PostGIS transaction, keep the geometry out of the outbox and reference it by id and version, then relay with FOR UPDATE SKIP LOCKED — publishing before marking sent gives at-least-once delivery, which is the guarantee the consumer’s deduplication already assumes.

This guide sits under Delivery Guarantees & Ordering, within Queue Management, Retries & Delivery Guarantees. It is the explicit alternative to Capturing PostGIS Changes with Logical Replication, and the two catch different sets of changes.

When to use this pattern

  • The application is the only writer that matters, so an application-level record catches every change worth publishing.
  • Events need intent — an event type, a routing key, a schema version — that a decoded row does not carry.
  • Operating a replication slot is undesirable, which for a managed database with restricted permissions it often is.

Two failure modes, in opposite directions

Whichever order you pick without an outbox, one of these happens Three orderings are traced for a single feature edit. Writing the feature and then publishing means that if the publish fails — a broker timeout, a network partition, a process death — the database holds a change nobody downstream ever hears about. Tiles stay stale, derived indexes drift, and no error is raised anywhere because the write itself succeeded, so the gap is discovered later by someone comparing counts. Publishing and then writing means that if the transaction rolls back, consumers have already acted on a change that does not exist: tiles are rebuilt from uncommitted data and a notification goes out about an edit that never happened. With the outbox, the feature row and the outbox row are written in one transaction, so either both exist or neither does, and a separate relay publishes from the outbox afterwards. A relay crash delays the event; it cannot lose it, because the row is still there. A relay that publishes twice produces a duplicate, which the consumer's deduplication already handles. write, then publish — an event can be lost COMMIT feature publish fails the database holds a change nobody hears about · tiles stale, indexes drift no error anywhere — the write succeeded publish, then write — an event can be invented publish ROLLBACK consumers act on a change that does not exist · tiles rebuilt from uncommitted data a notification about an edit that never happened outbox — neither is possible COMMIT feature + outbox row relay publishes later both exist or neither does · a relay crash delays the event, it cannot lose it a relay that publishes twice produces a duplicate, which dedup already handles The outbox does not remove the failure — it moves it from "lost or invented" to "late or duplicated", and the pipeline already has an answer for both.
Figure 1. The trade is not correctness for cost. It is exchanging two failures nothing downstream can handle for two the pipeline is already built to absorb.

Complete runnable implementation

sql 
CREATE TABLE feature_outbox (
    id            bigserial PRIMARY KEY,
    feature_id    bigint      NOT NULL,
    feature_version int       NOT NULL,   -- what the relay must publish
    event_type    text        NOT NULL,   -- intent: created | moved | retired
    routing_key   text        NOT NULL,   -- H3 cell, computed at write time
    schema_version int        NOT NULL DEFAULT 2,
    created_at    timestamptz NOT NULL DEFAULT now(),
    published_at  timestamptz
    -- Deliberately NO geometry column: an outbox row carrying a full
    -- multipolygon doubles the write volume of every edit and makes this
    -- table larger than the one it serves.
);

-- The relay only ever scans unpublished rows, so index for exactly that.
CREATE INDEX feature_outbox_unpublished
    ON feature_outbox (id) WHERE published_at IS NULL;
python 
import json
from datetime import datetime, UTC

import h3
from prometheus_client import Counter, Gauge

RELAYED = Counter("outbox_rows_relayed_total", "Rows published", ("event_type",))
BACKLOG = Gauge("outbox_unpublished_rows", "Unpublished outbox rows")

BATCH_SIZE = 200


async def edit_feature(conn, feature_id: int, geometry: dict,
                       event_type: str) -> None:
    """Write the feature and its outbox row in ONE transaction.

    Either both are durable or neither is. That is the whole pattern; every
    other detail here is about cost.
    """
    async with conn.transaction():
        version = await conn.fetchval(
            """
            UPDATE features
               SET geom = ST_GeomFromGeoJSON($2),
                   version = version + 1,
                   updated_at = now()
             WHERE id = $1
            RETURNING version
            """,
            feature_id, json.dumps(geometry),
        )

        centroid = await conn.fetchrow(
            "SELECT ST_Y(ST_Centroid(geom)) AS lat, ST_X(ST_Centroid(geom)) AS lon "
            "FROM features WHERE id = $1",
            feature_id,
        )
        routing_key = h3.latlng_to_cell(centroid["lat"], centroid["lon"], 8)

        await conn.execute(
            """
            INSERT INTO feature_outbox
                (feature_id, feature_version, event_type, routing_key)
            VALUES ($1, $2, $3, $4)
            """,
            feature_id, version, event_type, routing_key,
        )


async def relay(conn, publish) -> int:
    """Publish unsent rows. Safe to run in several processes at once."""
    async with conn.transaction():
        rows = await conn.fetch(
            """
            SELECT o.id, o.feature_id, o.feature_version, o.event_type,
                   o.routing_key, o.schema_version,
                   ST_AsGeoJSON(f.geom) AS geometry, f.version AS current_version
              FROM feature_outbox o
              JOIN features f ON f.id = o.feature_id
             WHERE o.published_at IS NULL
             ORDER BY o.id
             LIMIT $1
               FOR UPDATE OF o SKIP LOCKED
            """,
            BATCH_SIZE,
        )

        for row in rows:
            # The relay reads whatever version is CURRENT, which may be newer
            # than the row's. Carrying both lets a consumer tell that it is
            # seeing a later state than the event described.
            await publish({
                "schema_version": row["schema_version"],
                "action": row["event_type"],
                "feature_id": str(row["feature_id"]),
                "event_version": row["feature_version"],
                "geometry_version": row["current_version"],
                "routing_key": row["routing_key"],
                "crs": "EPSG:4326",
                "geometry": json.loads(row["geometry"]),
                "occurred_at": datetime.now(UTC).isoformat(),
            })
            RELAYED.labels(event_type=row["event_type"]).inc()

        # Marked sent AFTER publishing. A crash between the two republishes
        # on the next pass — at-least-once, which the consumer already expects.
        if rows:
            await conn.execute(
                "UPDATE feature_outbox SET published_at = now() WHERE id = ANY($1)",
                [r["id"] for r in rows],
            )
    return len(rows)
SKIP LOCKED is what lets the relay scale Three relay processes poll the same outbox table. Without FOR UPDATE SKIP LOCKED, all three select the same oldest rows: either they block on each other, serialising the relay to the throughput of one process while consuming three connections, or — if they read without locking — all three publish the same events, tripling the duplicate rate the consumer has to absorb. With SKIP LOCKED each transaction claims rows the others have not locked and skips over the rest, so the three processes take disjoint batches and throughput scales with the number of relays. Ordering is preserved only within a batch, not across them, which means two edits to the same feature can be published out of order if they land in different batches taken by different relays. That is why the routing key is computed at write time and why the event carries its version: ordering per feature has to be reconstructed by the consumer rather than assumed from the relay. without SKIP LOCKED relay 1 relay 2 relay 3 the same oldest rows blocked on each other — one relay's throughput, three connections or, reading unlocked, all three publish the same events tripling the duplicate rate the consumer must absorb with FOR UPDATE ... SKIP LOCKED relay 1 rows 1–200 relay 2 rows 201–400 relay 3 rows 401–600 disjoint batches · throughput scales with relay count but ordering holds only WITHIN a batch, not across them so two edits to one feature can publish out of order — which is why the event carries its version
Figure 2. Scaling the relay costs per-feature ordering, so the consumer has to reconstruct it. Pretending otherwise is how a single-relay design quietly becomes load-bearing.

Parameter reference

Name Type Spatial constraint Default
feature_version int The version the event describes; compared against current at relay —
routing_key text H3 cell computed at write time, from the geometry as it then was —
Geometry column — Absent by design; relay joins the feature table —
BATCH_SIZE int Ordering holds within a batch only 200
SKIP LOCKED — Required for more than one relay; without it they serialise or duplicate —
Partial index — On id WHERE published_at IS NULL — the only query the relay runs —

Gotchas and spatial edge cases

  1. The relay publishes the current geometry, not the one at the time of the edit. Two rapid edits mean the first event carries the second edit’s shape. Carrying both event_version and geometry_version lets a consumer detect it; storing the geometry in the outbox would avoid it entirely, at a write cost most pipelines will not pay.

  2. Computing the routing key at write time is deliberate. A feature that moves between cells would otherwise be routed by wherever it ended up rather than where the edit happened, so a consumer subscribed to the origin area never learns the feature left. The same reasoning drives the previous-geometry field in Tile Update Event Pipelines.

  3. Published rows must be deleted, not merely marked. An outbox retaining every event forever becomes the largest table in the database and slows the partial index scan through sheer bloat. Delete on a schedule once the topic has the events, and keep the deletion window longer than any replay you might run.

  4. A stalled relay is invisible without a backlog metric. The application keeps working — writes succeed, transactions commit — while nothing reaches consumers. The outbox_unpublished_rows gauge and an age-of-oldest-unpublished-row metric are the only signals, and they belong in the freshness objective described in SLOs & Alerting for Spatial Webhook Pipelines.

  5. The outbox misses every writer that is not the application. A bulk import, a migration or a desktop GIS client editing the table directly produces no outbox row, and the pipeline silently has no idea those features changed. If that is a real risk, logical replication is the pattern that catches them.

  6. ST_Centroid of a multipolygon can fall outside the feature. For a routing key that is usually acceptable, but for a crescent-shaped or multi-part feature the centroid can land in a cell the feature does not touch. Use ST_PointOnSurface where the key must be inside the geometry.

A stalled relay looks like nothing at all from the application The relay process stops — a crash, a bad deploy, a database permission change. From the application's point of view nothing has happened: writes still succeed, transactions still commit, outbox rows are still inserted, and every request returns normally. From the consumers' point of view the stream has stopped entirely, and because no error is raised anywhere in between, the only evidence is the outbox table growing and the age of its oldest unpublished row increasing. Neither of those is visible on an application dashboard, which watches request rates and error rates, both of which are healthy. The two metrics that catch it are the count of unpublished rows and the age of the oldest one, and the second is the more useful because it is unaffected by traffic volume: a backlog of ten thousand rows during a bulk import is normal, while an oldest row from four hours ago never is. the application writes succeed · transactions commit outbox rows inserted · requests 200 every dashboard is green the relay — stopped a crash, a bad deploy, a permission change no error is raised anywhere in between the consumers the stream has stopped entirely tiles stale, indexes drifting, no signal Two metrics catch it: unpublished row count, and the age of the oldest unpublished row. Prefer the second — it is unaffected by volume, so a ten-thousand-row backlog during a bulk import reads as normal while a four-hour-old row never does.
Figure 3. The outbox converts a lost event into a delayed one, which is only an improvement if somebody is watching the delay.

Verification

python 
import pytest

SQUARE = {"type": "Polygon", "coordinates": [[[13.40, 52.52], [13.40, 52.53],
                                              [13.41, 52.53], [13.41, 52.52],
                                              [13.40, 52.52]]]}


@pytest.mark.asyncio
async def test_rollback_leaves_no_outbox_row(conn):
    """The invented-event failure, asserted."""
    with pytest.raises(RuntimeError):
        async with conn.transaction():
            await edit_feature(conn, 4471, SQUARE, "moved")
            raise RuntimeError("simulated failure after the edit")

    assert await conn.fetchval(
        "SELECT count(*) FROM feature_outbox WHERE feature_id = 4471") == 0


@pytest.mark.asyncio
async def test_relay_crash_republishes_rather_than_losing(conn):
    """Publish-then-mark gives at-least-once, deliberately."""
    await edit_feature(conn, 4471, SQUARE, "moved")

    async def failing_publish(event):
        raise ConnectionError("broker unavailable")

    with pytest.raises(ConnectionError):
        await relay(conn, failing_publish)

    published = []
    assert await relay(conn, published.append) == 1
    assert published[0]["feature_id"] == "4471"


@pytest.mark.asyncio
async def test_two_relays_take_disjoint_batches(conn, conn2):
    """Without SKIP LOCKED this either blocks or double-publishes."""
    for i in range(400):
        await edit_feature(conn, 4471, SQUARE, "moved")

    a, b = [], []
    await asyncio.gather(relay(conn, a.append), relay(conn2, b.append))
    ids_a = {e["event_version"] for e in a}
    ids_b = {e["event_version"] for e in b}
    assert not (ids_a & ids_b)


@pytest.mark.asyncio
async def test_outbox_row_carries_no_geometry(conn):
    """The cost decision, guarded so it does not drift back."""
    columns = await conn.fetch(
        "SELECT column_name FROM information_schema.columns "
        "WHERE table_name = 'feature_outbox'")
    assert not any("geom" in c["column_name"] for c in columns)

The second test is the one that documents the guarantee rather than the implementation: it asserts that a failed publish leads to a republish, which is what makes the consumer’s deduplication load-bearing rather than decorative.