At-Least-Once Delivery, Ordering, Idempotent Consumers, Replay, and Schema/Model Evolution

AtlasMart deploys a CDC consumer that updates a search read model. During a network interruption the target transaction commits, but the consumer process dies before its checkpoint file is written. On restart the source legitimately sends the same change again. If the handler performs a blind CREATE, duplicate products or duplicate notifications appear. This is the canonical at-least-once failure: the source did nothing wrong; the consumer lacked an idempotency boundary.

Learning outcomes

Lab contract and assumptions

1. Exactly once is an end-to-end claim, not a source checkbox

2. Make receipt + target mutation one atomic Neo4j transaction

On the target graph, a unique SyncReceipt.eventId constraint turns duplicate detection into an executable invariant. The application checks the receipt and applies the business-key upsert/deletion in the same transaction that creates the receipt. A separate file/Kafka offset may still lag behind, but a replay then sees the receipt and becomes a no-op.

CYPHER 25
CREATE CONSTRAINT ch20_product_replica_id IF NOT EXISTS
FOR (p:ProductReplica) REQUIRE p.productId IS UNIQUE;
CREATE CONSTRAINT ch20_customer_replica_id IF NOT EXISTS
FOR (c:CustomerReplica) REQUIRE c.customerId IS UNIQUE;
CREATE CONSTRAINT ch20_order_replica_id IF NOT EXISTS
FOR (o:OrderReplica) REQUIRE o.orderId IS UNIQUE;
CREATE CONSTRAINT ch20_receipt_id IF NOT EXISTS
FOR (r:SyncReceipt) REQUIRE r.eventId IS UNIQUE;

from neo4j import GraphDatabase

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "atlasmart-course-2026")

def apply_event(tx, ev):
    # Receipt and graph mutation share one Neo4j transaction.
    already = tx.run(
        "MATCH (r:SyncReceipt {eventId:$id}) RETURN count(r) AS n",
        id=ev["id"],
    ).single()["n"]
    if already:
        return "duplicate"

    if ev["entity"] == "Product":
        if ev["op"] == "d":
            tx.run("MATCH (p:ProductReplica {productId:$key}) DETACH DELETE p", key=ev["key"])
        else:
            tx.run("""
                MERGE (p:ProductReplica {productId:$key})
                SET p += $after, p.labTag='ch20'
            """, key=ev["key"], after=ev["after"])
    # CustomerReplica / OrderReplica follow the same constrained-business-key pattern.
    tx.run("CREATE (:SyncReceipt {eventId:$id, labTag:'ch20', appliedAt:datetime()})", id=ev["id"])
    return "applied"

with GraphDatabase.driver(URI, auth=AUTH) as driver:
    driver.verify_connectivity()
    # Use session.execute_write(apply_event, event) in the consumer loop.

3. Controlled crash proves the replay path

The simulator writes target state before it writes the checkpoint. --crash-after exits in that gap. Restarting therefore replays sim-2004; the durable receipt converts the replay into DUPLICATE and the consumer continues.

$ python cdc_sim.py reset
$ python cdc_sim.py consume --crash-after sim-2004
...
APPLIED id=sim-2004 txId=505 seq=0 key=P-2001
CRASH after durable target apply, before checkpoint advance

$ python cdc_sim.py consume
DUPLICATE id=sim-2004 txId=505 seq=0 key=P-2001
APPLIED id=sim-2005 txId=506 seq=0 key=P-2002
CHECKPOINT ordinal=5 id=sim-2005

# The replay is expected. Receipt-based idempotency turns the duplicate into a no-op.

4. Ordering: scope the guarantee to the business invariant

Within a Neo4j transaction, seq establishes event order. Across transactions, the CDC result is chronological, but downstream systems may introduce partitions, retries or parallelism. Decide the smallest ordering domain your business needs. Inventory for one SKU may require per-productId order; unrelated SKUs should not be serialized behind each other.

5. Gap/replay recovery uses cursor evidence, not txId arithmetic

Real CDC may raise Neo.DatabaseError.ChangeDataCapture.ScanFailure when the requested log entry has been pruned, or Neo.ClientError.ChangeDataCapture.InvalidIdentifier for a wrong/obsolete cursor. Restores also invalidate old cursors. At that point “retry harder” is unsafe: stop incremental application, establish a new source snapshot/backfill, reconcile target state, then anchor a new current/earliest cursor according to the runbook.

$ python cdc_sim.py reset
$ python cdc_sim.py consume --drop sim-2003
APPLIED id=sim-2000 txId=500 seq=0 key=P-2001
APPLIED id=sim-2001 txId=500 seq=1 key=P-2002
APPLIED id=sim-2002 txId=502 seq=0 key=C-2001
GAP expectedOrdinal=3 got=4 -> BACKFILL_REQUIRED

$ python cdc_sim.py backfill
BACKFILL entities=3 products=1 customers=1 orders=1
CHECKPOINT ordinal=5 id=sim-2005

6. Schema and graph-model evolution are part of the event contract

{
  "contract": "atlasmart.product-change",
  "schemaVersion": 2,
  "sourceDatabase": "neo4j",
  "sourceEpoch": "deployment-specific",
  "sourceChangeId": "opaque-db-cdc-id",
  "eventId": "opaque-db-cdc-id",
  "businessKey": {"productId": "P-2001"},
  "operation": "u",
  "payload": {"price": 179.0, "currency": "USD"}
}

7. Deliberately wrong retry strategy

“Catch Exception, sleep one second, retry forever” conflates malformed schema, authorization denial, invalid cursor, target uniqueness violation, transient network loss and a poison event. Classify errors into retryable, quarantine/dead-letter, backfill-required and operator-action categories. Bound retries by an SLO; preserve the original event and trace identifiers for diagnosis.

8. Production judgment

9. Cleanup/reset

CYPHER 25
MATCH (n) WHERE n.labTag = 'ch20' DETACH DELETE n;
DROP CONSTRAINT ch20_product_replica_id IF EXISTS;
DROP CONSTRAINT ch20_customer_replica_id IF EXISTS;
DROP CONSTRAINT ch20_order_replica_id IF EXISTS;
DROP CONSTRAINT ch20_receipt_id IF EXISTS;

rm -f .atlasmart_ch20_state.json .atlasmart_ch20_checkpoint.json cdc_sim.py

Summary and next step

An at-least-once pipeline is reliable when duplicate application is boring, gaps are detected rather than guessed away, and model changes are versioned. Lesson 3 moves the same correctness contract into Kafka Connect, where connector offsets, topic partitioning, serialization and feedback-loop topology add new failure surfaces.

Authoritative references

• Current Neo4j versions — Current database release 2026.07.1 and current 5.26 LTS patch 5.26.30.
• CDC introduction — Current CDC availability and purpose; CDC is a change feed, not an exact database-copy mechanism.
• CDC on self-managed Neo4j — Enterprise-only enablement, OFF/DIFF/FULL txLogEnrichment modes, security, retention, disk and unrecorded-change boundaries.
• CDC on Aura — Current AuraDB Business Critical / Virtual Dedicated Cloud enablement, pause/resume and snapshot-reset behavior.
• CDC procedures — db.cdc.earliest/current/query semantics, exclusive cursors, txId/seq, metadata and change identifiers.
• CDC event schema — Current node/relationship event envelope including operation, keys, before/after state and metadata.
• CDC selectors — Server-side entity/operation/label/type/key/property and metadata filtering.
• CDC examples — Official cursor-management examples and empty-result/current-cursor retention guidance.
• CDC backup/restore behavior — Why restores invalidate old change identifiers and require consumer state/backfill/reinitialization planning.
• CDC troubleshooting — Disabled, scan-failure and invalid-identifier error mechanisms including retention and wrong-database cursors.
• CDC known issues — ElementId and change-identifier instability across restore/import/copy/pause-resume operations.
• CDC changelog — Current db.cdc.* namespace and deprecation history of cdc.* procedures.
• Neo4j Connector for Kafka — Current source/sink connector architecture and CDC/query source strategies.
• Kafka source CDC quickstart — Current CDC source configuration, txLogEnrichment prerequisite, event fields, offsets, topics and failure modes.
• Kafka connector installation — Current connector release 5.5.3 and Kafka Connect plugin deployment model.
• Kafka source query quickstart — Community-compatible polling source alternative and its delete/soft-delete limitation.
• Kafka sink CDC quickstart — Neo4j-to-Neo4j CDC sink pattern, source header requirement, key constraints and loop warning.
• Python driver 6.3 API — Official driver 6.3 and Bolt compatibility used for optional application examples.