Change Data Capture for Graph Attribute Sync

The symptom that brings teams to this page is a routing graph that quietly diverges from its system of record: a point of interest that closed weeks ago still resolves as an active waypoint, a category correction never lands, and a full re-import — the usual sledgehammer fix — takes hours and thrashes the page cache every night. The root cause is treating the upstream database as something you periodically re-read in full, when what you actually have is a stream of changes. A change-data-capture (CDC) feed emits an ordered log of inserts, updates, and deletes with a monotonic sequence number per row, and applying only those deltas is orders of magnitude cheaper than re-scanning the source. The trap is that CDC delivery is at-least-once and only ordered per key, so a naive consumer replays stale events over fresh ones and never handles deletes. This page resolves that with one runnable async consumer that applies CDC deltas idempotently: a version-guarded MERGE/SET for upserts, a tombstone for deletes, and per-key deduplication so out-of-order redelivery collapses to the correct terminal state. It is the event-log counterpart to the general write-back approach in syncing external attribute changes to graph nodes.

Prerequisites & Versions

The consumer is transport-agnostic — it reads CDCEvent records from any async source (Kafka with Debezium, Neo4j’s own CDC feed, a polled outbox table). Only the async driver and a uniqueness constraint on the business key are required.

Library	Min version	Install / note
Python	3.10+	`dataclass`, `asyncio`, union typing
`neo4j` async driver	5.14	`pip install "neo4j>=5.14"`
Neo4j server	5.x	`docker run -p7687:7687 -e NEO4J_AUTH=neo4j/password neo4j:5`

This layer assumes the PoiNode records already exist with a stable external_id and a location, established upstream by the OSM data ingestion pipelines stage. CDC is a delta write-back on top of that graph; if you are still loading the base topology, build that first.

Implementation

Two Cypher statements do the mutation: a version-guarded upsert for creates and updates, and a version-guarded tombstone for deletes. Both MERGE on external_id so the match is an index seek, seed a sentinel cdc_version of -1 on first sight, and only mutate when the incoming version strictly exceeds the stored one. The CDCApplier reads events from an async stream, deduplicates each batch to the highest version per key, applies upserts and deletes, then commits the source offset — in that order — so an at-least-once redelivery re-runs harmlessly.

import asyncio
import logging
from dataclasses import dataclass
from typing import Any, AsyncIterator, Awaitable, Callable

from neo4j import AsyncGraphDatabase

logging.basicConfig(level=logging.INFO)

UPSERT = """
UNWIND $events AS ev
MERGE (n:PoiNode {external_id: ev.key})
  ON CREATE SET n.cdc_version = -1
WITH n, ev
WHERE ev.version > n.cdc_version
SET n.name        = ev.after.name,
    n.category    = ev.after.category,
    n.location     = point({srid: 4326, longitude: ev.after.lon, latitude: ev.after.lat}),
    n.deleted     = false,
    n.cdc_version = ev.version
"""

TOMBSTONE = """
UNWIND $events AS ev
MERGE (n:PoiNode {external_id: ev.key})
  ON CREATE SET n.cdc_version = -1
WITH n, ev
WHERE ev.version > n.cdc_version
SET n.deleted = true, n.cdc_version = ev.version
"""


@dataclass
class CDCEvent:
    key: str                 # business key (external_id) — CDC orders per key
    version: int             # monotonic log sequence number from the source
    op: str                  # "c" (create), "u" (update), or "d" (delete)
    after: dict[str, Any] | None
    offset: int              # position in the CDC log, committed after apply


class CDCApplier:
    def __init__(
        self,
        uri: str,
        auth: tuple[str, str],
        *,
        database: str = "neo4j",
        batch_size: int = 1_000,
    ) -> None:
        self.driver = AsyncGraphDatabase.driver(
            uri, auth=auth,
            max_connection_pool_size=20,
            connection_acquisition_timeout=10.0,
        )
        self.database = database
        self.batch_size = batch_size
        self.log = logging.getLogger("cdc")

    async def close(self) -> None:
        await self.driver.close()

    async def ensure_schema(self) -> None:
        async with self.driver.session(database=self.database) as s:
            await s.run(
                "CREATE CONSTRAINT poi_extid IF NOT EXISTS "
                "FOR (n:PoiNode) REQUIRE n.external_id IS UNIQUE"
            )
            await s.run(
                "CREATE POINT INDEX poi_location IF NOT EXISTS "
                "FOR (n:PoiNode) ON (n.location)"
            )

    @staticmethod
    def _dedupe_latest(events: list[CDCEvent]) -> list[CDCEvent]:
        """Collapse each key to its highest-version event so one batch commits
        the correct terminal state regardless of intra-batch delivery order."""
        latest: dict[str, CDCEvent] = {}
        for ev in events:
            cur = latest.get(ev.key)
            if cur is None or ev.version > cur.version:
                latest[ev.key] = ev
        return list(latest.values())

    async def _apply_batch(self, events: list[CDCEvent]) -> None:
        events = self._dedupe_latest(events)
        upserts = [
            {"key": e.key, "version": e.version, "after": e.after}
            for e in events if e.op in ("c", "u") and e.after is not None
        ]
        deletes = [{"key": e.key, "version": e.version} for e in events if e.op == "d"]
        async with self.driver.session(database=self.database) as s:
            if upserts:
                await (await s.run(UPSERT, events=upserts)).consume()
            if deletes:
                await (await s.run(TOMBSTONE, events=deletes)).consume()

    async def run(
        self,
        stream: AsyncIterator[CDCEvent],
        commit_offset: Callable[[int], Awaitable[None]],
    ) -> None:
        await self.ensure_schema()
        buffer: list[CDCEvent] = []
        last_offset = -1
        async for ev in stream:
            buffer.append(ev)
            last_offset = ev.offset
            if len(buffer) >= self.batch_size:
                await self._apply_batch(buffer)
                await commit_offset(last_offset)  # commit only after the write is durable
                buffer.clear()
        if buffer:
            await self._apply_batch(buffer)
            await commit_offset(last_offset)
        self.log.info("CDC stream drained; last committed offset %d", last_offset)


async def demo_stream() -> AsyncIterator[CDCEvent]:
    """Stand-in for a Kafka/Debezium consumer. Note the out-of-order v3 and the delete."""
    lon, lat = -87.6298, 41.8781  # Chicago
    events = [
        CDCEvent("poi:1001", 5, "u", {"name": "Depot A", "category": "hub", "lon": lon, "lat": lat}, 5),
        CDCEvent("poi:1002", 7, "c", {"name": "Kiosk 7", "category": "retail", "lon": lon + 0.01, "lat": lat}, 6),
        CDCEvent("poi:1001", 3, "u", {"name": "OLD NAME", "category": "hub", "lon": lon, "lat": lat}, 7),
        CDCEvent("poi:1002", 9, "d", None, 8),
    ]
    for ev in events:
        yield ev


async def main() -> None:
    applier = CDCApplier("bolt://localhost:7687", ("neo4j", "password"))
    committed = {"offset": -1}

    async def commit_offset(offset: int) -> None:
        committed["offset"] = offset  # in production: commit to Kafka / checkpoint store

    try:
        await applier.run(demo_stream(), commit_offset)
    finally:
        await applier.close()
    print(f"last committed offset: {committed['offset']}")


if __name__ == "__main__":
    asyncio.run(main())

How It Works

Three mechanics make the consumer correct under at-least-once, per-key-ordered delivery, and each maps to a specific line:

The monotonic version guard (WHERE ev.version > n.cdc_version). This is optimistic concurrency without locks. Every event carries the source log sequence number in version; the SET fires only when that number strictly beats the stored cdc_version. A redelivered or reordered event with an older version matches the node but fails the guard and does nothing — an idempotent no-op. Replaying the same offset after a crash therefore cannot corrupt state, which is what lets the consumer commit offsets safely and resume from the last checkpoint. The guard shape mirrors the one detailed in syncing external attribute changes to graph nodes; here the version is the CDC log position rather than an application counter.
Per-key deduplication (_dedupe_latest). CDC guarantees ordering only within a key, and a single poll can contain several events for the same row. Collapsing each key to its highest-version event before writing means the batch commits the correct terminal state in one pass, regardless of the order events sit in the buffer, and turns a chatty burst into one write per node. The dedupe also resolves the update-versus-delete race: if a key has both an update at v5 and a delete at v9 in the same batch, the delete wins because it carries the higher version.
Tombstone deletes, not hard deletes. The delete path does not DETACH DELETE; it sets deleted = true and bumps cdc_version. Hard-deleting the node would also destroy the stored version, so a later-arriving stale create for the same key would resurrect a row that should stay gone. The tombstone keeps the version record alive as a gravestone, so out-of-order redelivery still fails the guard. A separate compaction job can physically remove tombstoned nodes past a retention window once no older offsets remain in flight.

The location is written from the CDC payload in Neo4j’s point({longitude, latitude}) convention (WGS84 / EPSG:4326) so it stays seekable by the spatial indexing strategies layer.

Common Failure Patterns

1. Out-of-order events overwriting newer data. Without the guard, an event that is redelivered or arrives late clobbers a fresher value, because whichever write commits last wins — and CDC gives no global ordering promise. The version comparison must live in Cypher, not in application code where a concurrent worker can still race it:

// WRONG — last delivery to commit wins, even if it is older
MATCH (n:PoiNode {external_id: ev.key}) SET n.name = ev.after.name
// RIGHT — only a strictly newer log position mutates the node
MERGE (n:PoiNode {external_id: ev.key}) ON CREATE SET n.cdc_version = -1
WITH n, ev WHERE ev.version > n.cdc_version
SET n.name = ev.after.name, n.cdc_version = ev.version

2. Missing delete handling leaves ghost nodes. A consumer that only processes c and u events silently ignores every d, so deleted rows linger in the graph forever and routing keeps returning waypoints that no longer exist. Route the d operation to a tombstone that participates in the same version guard, rather than dropping it or issuing an unguarded delete that a stale replay can undo.

3. Committing the offset before the write is durable. Advancing the CDC offset first and writing second means a crash between the two loses the delta permanently — the source considers it delivered, but the graph never applied it. Always apply the batch, confirm the write with consume(), and only then commit the offset; on restart the consumer safely re-applies the last batch because the guard makes re-application a no-op.

Performance Notes

CDC cost tracks the volume of delivered events, while committed writes track the number of distinct keys after deduplication. For a batch of $B$ raw events spanning $K$ distinct keys with an at-least-once redundancy factor $A = B / K$, the dedupe step reduces the work sent to the database:

$$W_{\text{commit}} \le K, \qquad C_{\text{apply}} = K \cdot c_{\text{idx}}, \qquad \text{lag} = o_{\text{head}} - o_{\text{commit}}$$

where $c_{\text{idx}}$ is the cost of one index-seek MERGE. As $A$ climbs — a hot row re-emitted many times per poll — deduplication saves proportionally more, because $B - K$ events never reach a transaction. Consumer lag, the gap between the log head offset and the last committed offset, is the metric to alarm on: a steadily growing lag means apply throughput has fallen behind the source’s change rate, and the fix is a larger batch_size (fewer round trips) or profiling the write plan before adding consumers, since more concurrency only helps if the MERGE is genuinely index-bound rather than lock-bound. Confirm that with the techniques in optimizing Cypher query plans for spatial data. Keep memory flat by bounding batch_size; peak buffer is roughly batch_size × bytes_per_event regardless of total stream volume.

This guide is part of Attribute Synchronization Techniques, within the broader Spatial Graph Construction & OSM Ingestion guide.

Change Data Capture for Graph Attribute Sync

Prerequisites & Versions

Implementation

How It Works

Common Failure Patterns

Performance Notes

Related

Related pages

Siblings