Design an Idempotent Import/Upsert Workflow and Prove It Remains Correct Under Repeated Execution

Learning outcomes

AtlasMart's ingestion service can deliver the same product/order batch more than once because network retries, queue redelivery, and worker restarts are normal. The production contract is therefore not “the script usually runs once.” It is: one logical input produces one stable graph state even when the database transaction is retried or the same batch is replayed.

Re-establish the isolated Chapter 06 write fixture

The fixture deliberately reuses AtlasMart's stable business identifiers and Community-supported uniqueness constraints. It creates or normalizes one customer, one dedicated product, one dedicated order, and their PLACED/CONTAINS relationships. Because every identity is deterministic and the write pattern uses MERGE, rerunning the fixture should leave the target counts unchanged.

CYPHER 25CREATE CONSTRAINT customer_id IF NOT EXISTS FOR (c:Customer) REQUIRE c.customerId IS UNIQUE;CREATE CONSTRAINT product_id IF NOT EXISTS FOR (p:Product) REQUIRE p.productId IS UNIQUE;CREATE CONSTRAINT order_id IF NOT EXISTS FOR (o:Order) REQUIRE o.orderId IS UNIQUE;MERGE (c:Customer {customerId:'C-1001'})  ON CREATE SET c.name='Ava Chen', c.region='eu', c.createdAt=datetime('2026-09-09T00:00:00Z')  ON MATCH  SET c.lastSeenAt=datetime('2026-09-09T00:00:00Z');MERGE (p:Product {productId:'P-WRITE-1001'})  ON CREATE SET p.name='Atlas Travel Camera', p.price=149.90, p.status='active', p.createdAt=datetime('2026-09-09T00:00:00Z')  ON MATCH  SET p.name='Atlas Travel Camera', p.price=149.90, p.status='active';MERGE (o:Order {orderId:'O-WRITE-2001'})  ON CREATE SET o.status='pending', o.source='chapter06', o.createdAt=datetime('2026-09-09T00:00:00Z')  ON MATCH  SET o.source='chapter06';MERGE (c)-[:PLACED]->(o)MERGE (o)-[line:CONTAINS]->(p)  ON CREATE SET line.quantity=1, line.unitPrice=149.90  ON MATCH  SET line.quantity=1, line.unitPrice=149.90;

CYPHER 25MATCH (p:Product {productId:'P-WRITE-1001'})OPTIONAL MATCH (c:Customer {customerId:'C-1001'})-[placed:PLACED]->(o:Order {orderId:'O-WRITE-2001'})OPTIONAL MATCH (o)-[line:CONTAINS]->(p)RETURN count(DISTINCT p) AS products,       count(DISTINCT o) AS orders,       count(DISTINCT placed) AS placedRelationships,       count(DISTINCT line) AS containsRelationships;

The deterministic target for the dedicated slice is 1 / 1 / 1 / 1. Those counts do not imply the entire database contains only one Product or Order; previous chapters intentionally contain additional AtlasMart data.

1. Define the mutation contract before writing Cypher

Idempotency here does not mean “nothing changes on the second call.” Fields such as updatedAt can intentionally change each time. It means the business graph—identity/cardinality and authoritative payload values—does not accumulate duplicate logical entities or relationships.

2. A staged, parameterized upsert workflow

CYPHER 25MATCH (c:Customer {customerId:$customerId})MERGE (p:Product {productId:$product.productId})  ON CREATE SET p.createdAt=datetime()SET p.name=$product.name,    p.price=$product.price,    p.status=$product.status,    p.updatedAt=datetime()MERGE (o:Order {orderId:$order.orderId})  ON CREATE SET o.createdAt=datetime()SET o.status=$order.status,    o.source=$order.source,    o.updatedAt=datetime()MERGE (c)-[:PLACED]->(o)MERGE (o)-[line:CONTAINS]->(p)SET line.quantity=$line.quantity,    line.unitPrice=$line.unitPriceRETURN c.customerId AS customerId,       o.orderId AS orderId,       p.productId AS productId;

Each identity is matched separately. Mutable state is applied afterward. Relationship endpoints are already bound. The uniqueness constraints on customer/product/order business keys turn accidental duplicate identity into a database error rather than silent divergence.

3. Extend to a bounded batch with UNWIND

CYPHER 25UNWIND $rows AS rowMATCH (c:Customer {customerId:row.customerId})MERGE (p:Product {productId:row.product.productId})  ON CREATE SET p.createdAt=datetime()SET p += {  name:row.product.name,  price:row.product.price,  status:row.product.status}MERGE (o:Order {orderId:row.order.orderId})  ON CREATE SET o.createdAt=datetime()SET o += {  status:row.order.status,  source:row.order.source}MERGE (c)-[:PLACED]->(o)MERGE (o)-[line:CONTAINS]->(p)SET line.quantity=row.line.quantity,    line.unitPrice=row.line.unitPriceRETURN count(*) AS inputRowsProcessed;

This assumes one logical line per order/product pair. If AtlasMart needs multiple separate line items for the same product in one order, the relationship requires its own stable line identity or the model should reify OrderLine as a node. Idempotency follows the domain model; it cannot be added by syntax alone.

4. Prove replay invariants

Run the same parameter payload three times. After each run, capture the following query. Stable counts are the acceptance contract.

CYPHER 25MATCH (p:Product {productId:'P-WRITE-1001'})MATCH (o:Order {orderId:'O-WRITE-2001'})MATCH (c:Customer {customerId:'C-1001'})OPTIONAL MATCH (c)-[placed:PLACED]->(o)OPTIONAL MATCH (o)-[line:CONTAINS]->(p)RETURN count(DISTINCT p) AS productNodes,       count(DISTINCT o) AS orderNodes,       count(DISTINCT placed) AS placedCount,       count(DISTINCT line) AS lineCount,       max(line.quantity) AS quantity,       max(line.unitPrice) AS unitPrice;

The business cardinality target is 1 / 1 / 1 / 1, with line properties equal to the latest authoritative payload. Do not use total database counts as the assertion because previous chapters contain other AtlasMart entities.

5. Managed driver retries change the application design

Neo4j's official drivers can retry managed transaction functions after transient failures. The Python driver's execute_write() contract explicitly warns that the supplied transaction function may be invoked more than once and therefore must be idempotent. Keep payment capture, email sends, file writes, and irreversible HTTP calls out of the retryable callback unless they have independent idempotency keys and reconciliation.

from neo4j import GraphDatabaseURI = 'bolt://localhost:7687'AUTH = ('neo4j', 'atlasmart-course-password')UPSERT = '''CYPHER 25MATCH (c:Customer {customerId:$customerId})MERGE (p:Product {productId:$productId})  ON CREATE SET p.createdAt=datetime()SET p.name=$name, p.price=$price, p.status='active'MERGE (o:Order {orderId:$orderId})  ON CREATE SET o.createdAt=datetime()SET o.status='pending', o.source='chapter06'MERGE (c)-[:PLACED]->(o)MERGE (o)-[line:CONTAINS]->(p)SET line.quantity=$quantity, line.unitPrice=$priceRETURN o.orderId AS orderId'''def upsert_tx(tx, payload):    # Database work only. Do not send email/call payment APIs here.    return tx.run(UPSERT, **payload).single()['orderId']payload = {    'customerId':'C-1001',    'productId':'P-WRITE-1001',    'orderId':'O-WRITE-2001',    'name':'Atlas Travel Camera',    'price':149.90,    'quantity':1,}with GraphDatabase.driver(URI, auth=AUTH) as driver:    driver.verify_connectivity()    with driver.session(database='neo4j') as session:        print(session.execute_write(upsert_tx, payload))

6. Controlled concurrency: many callers, one logical graph

from concurrent.futures import ThreadPoolExecutorfrom neo4j import GraphDatabaseURI = 'bolt://localhost:7687'AUTH = ('neo4j', 'atlasmart-course-password')# Reuse UPSERT and payload from the previous example.def one_call(driver, _):    with driver.session(database='neo4j') as session:        return session.execute_write(upsert_tx, payload)with GraphDatabase.driver(URI, auth=AUTH) as driver:    with ThreadPoolExecutor(max_workers=8) as pool:        list(pool.map(lambda i: one_call(driver, i), range(32)))

Then rerun the replay acceptance query. The invariant must still be one Product, one Order, one PLACED, and one CONTAINS relationship for these IDs. If a deadlock/transient failure is retried internally, exact attempts/timing are environment-specific. If retries are exhausted, the application must surface failure and reconcile from business IDs—not generate a new random identity and hope.

7. Deliberately wrong: CREATE plus random identity in a retryable callback

CYPHER 25CREATE (o:Order {  orderId:randomUUID(),  externalOrderNumber:$externalOrderNumber,  status:'pending'})RETURN o.orderId;

If the callback is invoked again, it creates another Order because the generated identity changes. Even if the first transaction rolled back, this pattern also makes reconciliation harder after ambiguous client outcomes. Prefer a stable upstream idempotency/business key such as externalOrderNumber backed by an appropriate uniqueness contract, then MERGE on that identity.

8. Reconciliation is part of idempotency

A client can time out after sending a write and before learning whether it committed. The safe response is not automatically “run a different create.” Query by the stable business/request ID, inspect the committed state, and decide whether to retry, repair, or report success. For workflows spanning Neo4j and external systems, maintain an explicit operation/request ID and durable state machine or outbox/inbox pattern appropriate to the integration.

CYPHER 25MATCH (o:Order {orderId:$orderId})OPTIONAL MATCH (c:Customer)-[:PLACED]->(o)OPTIONAL MATCH (o)-[line:CONTAINS]->(p:Product)RETURN o.orderId,       o.status,       c.customerId,       collect({productId:p.productId, quantity:line.quantity, unitPrice:line.unitPrice}) AS lines;

9. Chapter 06 integration lab

Use one parameter payload and complete this verification sequence: establish constraints; run the upsert once; verify business counts; run the same upsert twice more; verify unchanged business counts; change only product price and line unitPrice, rerun, and prove attributes converge without creating new entities; run the optional concurrent replay test; verify SHOW CONSTRAINTS; finally remove only the dedicated concurrent-test entities you created beyond the shared fixture.

CYPHER 25MATCH (p:Product)WITH count(CASE WHEN p.productId IS NULL THEN 1 END) AS productsMissingIdMATCH (o:Order)WITH productsMissingId,     count(CASE WHEN o.orderId IS NULL THEN 1 END) AS ordersMissingIdMATCH (p:Product {productId:'P-WRITE-1001'})MATCH (o:Order {orderId:'O-WRITE-2001'})MATCH (c:Customer {customerId:'C-1001'})OPTIONAL MATCH (c)-[placed:PLACED]->(o)OPTIONAL MATCH (o)-[line:CONTAINS]->(p)RETURN productsMissingId,       ordersMissingId,       count(DISTINCT p) AS targetProducts,       count(DISTINCT o) AS targetOrders,       count(DISTINCT placed) AS targetPlaced,       count(DISTINCT line) AS targetLines;

Production judgment

Idempotency requires stable identities, constraints, mutation semantics, transaction boundaries, and an application contract for retries and ambiguous outcomes. Hot keys can serialize work; large batches can consume transaction memory; deadlock retries can reduce throughput; and external side effects require independent idempotency. Monitor retries, constraint failures, lock wait/deadlock signals, write latency distributions, transaction timeouts, and reconciliation counts. Migration and rollback plans must preserve the business keys used to recognize already-applied work.

Summary and next step

Chapter 06 turns mutation syntax into a correctness discipline: CREATE for truly new facts, MERGE for constraint-backed identity-aware convergence, SET/REMOVE with explicit patch/replacement semantics, deletes with bounded blast radius, and retry-safe workflows verified from committed state. Chapter 07 now returns to modeling and asks when a fact belongs on a node, property, relationship, intermediate entity, hierarchy, or temporal structure.

Authoritative references

• Current Neo4j versions — Release/LTS snapshot used to pin the course baseline.
• Cypher Manual — CREATE — CREATE semantics for nodes, relationships, parameters, and dynamic labels/types.
• Cypher Manual — MERGE — Match-or-create semantics, ON CREATE/ON MATCH, constraints, and concurrent relationship merges.
• Cypher Manual — SET — Property/label updates and map replacement versus mutation.
• Cypher Manual — REMOVE — Property and label removal semantics.
• Cypher Manual — DELETE — DELETE, NODETACH DELETE, DETACH DELETE, and large-delete guidance.
• Operations Manual — transactional behavior — ACID, read-committed isolation, locking, transaction logs, and deadlock behavior.
• Python Driver — transactions — Managed transactions, retry behavior, and session lifecycle.
• Python Driver — performance — Transaction overhead, grouping work, auto-commit tradeoffs, and connection-pool considerations.
• Python Driver API — execute_write callback retry/idempotency requirement.
• Concurrent data access — Locks, lost updates, deadlocks, and retry strategies.
• Transaction logging — Committed write logging and durability/recovery role.