Designing idempotent DAG tasks for parcel ingestion

Somebody re-ran last night’s ingest for Weld County because the first attempt died halfway through, and now the parcel count is 82 431 against an expected 41 216 — every parcel is in the table twice. Or the count is right and the change table has two identical rezone events, so two identical alerts went out. Or the re-run deleted parcels the first attempt had already written and the layer is now missing a subdivision. All three are the same defect: a task that is safe to run once and unsafe to run again. Since re-running is the normal recovery for every failure in a municipal pipeline, that defect turns every ordinary incident into a data-repair job. This guide makes the tasks re-runnable, which is the precondition the parent topic — scheduling & orchestration for municipal feed runs — assumes throughout.

Diagnosis: finding the parts that are not re-runnable

Idempotency is a property of each write, so audit the writes rather than the pipeline. Four questions locate almost every violation.

Does anything use an auto-generated key as its identity? A SERIAL primary key means the second run cannot recognise the row the first run wrote, so it inserts a duplicate. The identity has to come from the data — (source_id, parcel_id, effective_date) — with the surrogate key kept for joins only.

Does anything append unconditionally? INSERT INTO parcel_history ... with no conflict clause is a duplicate factory. So is appending to a change log keyed on nothing but a timestamp.

Does anything delete by absence? “Delete every parcel not in this payload” is correct on a complete payload and catastrophic on a truncated one — and a re-run after a partial fetch is exactly when the payload is truncated.

Does anything have an external side effect? Sending an email, posting to a webhook, or writing to a queue cannot be undone by a rollback. A task that writes rows and notifies is not idempotent even if the rows are, because the second run re-notifies.

A quick way to find them all: run the pipeline twice against the same fixture and diff the database. Anything that differs between one run and two is a violation, and the diff names it precisely.

Four write patterns, and what a re-run does to each A four-row by three-column matrix of write patterns. Rows are using an auto-generated key as identity, appending unconditionally, deleting rows absent from the payload, and performing an external side effect. Columns are what a second run does, whether the damage is reversible, and the fix. Auto-generated keys and unconditional appends duplicate data reversibly. Delete-by-absence destroys rows irreversibly when the payload is truncated. An external side effect cannot be rolled back at all. Four write patterns, and what a re-run does to each what a second run does reversible? fix Auto-generated key as identity inserts a duplicate yes — de-duplicate natural key + unique constraint Unconditional append appends again yes — de-duplicate ON CONFLICT DO NOTHING Delete rows absent from payload deletes live rows no — needs a re-fetch tombstone + completeness check External side effect in the task notifies again no — it already left outbox + idempotency key safe or fixable messy but reversible irreversible
The bottom two rows are the dangerous ones because they are irreversible. A duplicate can be de-duplicated; a parcel deleted because a truncated payload did not mention it is gone until someone re-fetches, and a notification sent twice cannot be unsent.

Step-by-step implementation

1. Give every row a natural key 

-- The identity comes from the data. The surrogate id exists for joins, not identity.
CREATE TABLE parcel_version (
    id             bigserial PRIMARY KEY,
    source_id      text        NOT NULL,
    parcel_id      text        NOT NULL,
    valid_from     date        NOT NULL,
    zoning_code    text        NOT NULL,
    overlay_codes  text[]      NOT NULL DEFAULT '{}',
    geom           geometry(MultiPolygon, 26913) NOT NULL,
    content_digest text        NOT NULL,
    run_id         uuid        NOT NULL,
    CONSTRAINT parcel_version_identity
        UNIQUE (source_id, parcel_id, valid_from)
);

2. Upsert, and make the upsert a no-op when nothing changed 

UPSERT = """
INSERT INTO parcel_version (source_id, parcel_id, valid_from, zoning_code,
                            overlay_codes, geom, content_digest, run_id)
VALUES (%(source_id)s, %(parcel_id)s, %(valid_from)s, %(zoning_code)s,
        %(overlay_codes)s, ST_GeomFromWKB(%(geom)s, 26913), %(digest)s, %(run_id)s)
ON CONFLICT (source_id, parcel_id, valid_from) DO UPDATE
   SET zoning_code   = EXCLUDED.zoning_code,
       overlay_codes = EXCLUDED.overlay_codes,
       geom          = EXCLUDED.geom,
       content_digest= EXCLUDED.content_digest,
       run_id        = EXCLUDED.run_id
 WHERE parcel_version.content_digest <> EXCLUDED.content_digest
RETURNING (xmax = 0) AS inserted;
"""


def upsert_parcels(cur, rows, run_id):
    """Returns (inserted, updated, unchanged). The WHERE clause on DO UPDATE is what
    makes a re-run cheap AND keeps run_id honest: a row whose content is identical is
    not touched, so it still names the run that actually produced its value."""
    inserted = updated = unchanged = 0
    for row in rows:
        cur.execute(UPSERT, {**row, "run_id": run_id})
        res = cur.fetchone()
        if res is None:
            unchanged += 1          # DO UPDATE ... WHERE matched nothing
        elif res["inserted"]:
            inserted += 1
        else:
            updated += 1
    return inserted, updated, unchanged

The WHERE parcel_version.content_digest <> EXCLUDED.content_digest clause is the detail that makes this genuinely idempotent rather than merely non-duplicating. Without it, a re-run rewrites every row with a new run_id, so the provenance now claims the re-run produced values it only re-confirmed — and every row looks freshly changed to anything watching run_id.

3. Stage per run, then swap

For work that cannot be expressed as a single upsert — a full-snapshot load, a rebuilt derived table — write into a run-scoped staging table and swap atomically. A re-run rebuilds its own staging table from scratch, so a partial first attempt leaves nothing behind.

def load_snapshot(conn, source_id, payload, run_id):
    staging = f"stg_parcel_{source_id}_{run_id.hex[:8]}"
    with conn.transaction():
        conn.execute(f"CREATE UNLOGGED TABLE {staging} (LIKE parcel_current INCLUDING ALL)")
        copy_into(conn, staging, payload)                  # bulk COPY
        assert_row_count_plausible(conn, staging, source_id)   # before the swap, not after
        conn.execute("DELETE FROM parcel_current WHERE source_id = %s", (source_id,))
        conn.execute(f"INSERT INTO parcel_current SELECT * FROM {staging}")
        conn.execute(f"DROP TABLE {staging}")

The DELETE is safe here only because it is inside the same transaction as the INSERT and behind a plausibility assertion. That assertion is doing the work that makes delete-by-absence acceptable at all.

4. Tombstone instead of deleting 

def apply_absences(cur, source_id, present_ids, run_id, payload_is_complete):
    """Mark parcels absent from a COMPLETE payload as retired. Never runs on a
    partial payload, because absence then means 'not fetched yet', not 'gone'."""
    if not payload_is_complete:
        return 0
    cur.execute("""
        UPDATE parcel_current
           SET retired_at = now(), retired_by_run = %(run_id)s
         WHERE source_id = %(source_id)s
           AND parcel_id <> ALL(%(present)s)
           AND retired_at IS NULL
        """, {"source_id": source_id, "present": list(present_ids), "run_id": run_id})
    return cur.rowcount

A tombstone is re-runnable — the second run matches retired_at IS NULL and updates nothing — while a DELETE is destructive and, on a truncated payload, unrecoverable without a re-fetch.

5. Move side effects behind an outbox 

def record_changes(cur, changes, run_id):
    """Changes go to an outbox keyed on the change identity, not on the run. A
    re-run inserts the same keys and conflicts, so nothing is notified twice."""
    for ch in changes:
        cur.execute("""
            INSERT INTO notification_outbox
                (idempotency_key, parcel_id, change_class, payload, run_id)
            VALUES (%(key)s, %(parcel_id)s, %(cls)s, %(payload)s, %(run_id)s)
            ON CONFLICT (idempotency_key) DO NOTHING
            """, {"key": f"{ch.parcel_id}:{ch.change_digest}", "parcel_id": ch.parcel_id,
                  "cls": ch.change_class, "payload": ch.as_json(), "run_id": run_id})

The key is (parcel_id, change_digest) and deliberately not the run: the same change discovered by two runs is one notification. A separate delivery worker drains the outbox and marks rows sent, so the ingestion task never performs the un-undoable act itself. This is the same idempotency-key design that zoning change alerting relies on.

Verification & testing

The test is mechanical and belongs in every pipeline: run the task twice, assert the database is identical.

Row counts after one run and after two, per write pattern Grouped bar chart comparing row counts after one run and after two identical runs, for four write patterns over a 1 000-parcel fixture. An unconditional append goes from 1 000 to 2 000. A serial-keyed insert also doubles. An upsert without the unchanged guard stays at 1 000 rows but rewrites all 1 000. A guarded upsert stays at 1 000 rows and rewrites none. The chart shows that row count alone cannot distinguish the third pattern from the fourth. Row counts after one run and after two, per write pattern 0 500 1000 1500 2000 2500 1000 2000 0 append 1000 2000 0 serial key insert 1000 1000 1000 upsert, no guard 1000 1000 0 upsert, guarded Rows rows after 1 run rows after 2 runs rows rewritten by run 2 Patterns three and four are identical by row count and different by provenance.
Row count catches the first two patterns and is blind to the third: an unguarded upsert holds the count steady while restamping every row with the new run id, so provenance now credits a re-run with values it only re-confirmed. That is why the test compares full table snapshots rather than counts.
def test_task_is_idempotent(db, fixture_payload):
    run_ingest(db, fixture_payload, run_id=uuid4())
    first = snapshot_all_tables(db)

    run_ingest(db, fixture_payload, run_id=uuid4())   # a DIFFERENT run id
    second = snapshot_all_tables(db)

    # run_id columns are allowed to differ only where content actually changed —
    # which for identical input is nowhere.
    assert first == second


def test_partial_first_attempt_leaves_no_residue(db, fixture_payload):
    with pytest.raises(InjectedFailure):
        run_ingest(db, fixture_payload, fail_after_rows=500)
    assert db.count("parcel_current") == 0            # staging was never swapped
    assert db.tables_matching("stg_parcel_%") == []   # and cleaned up


def test_outbox_does_not_duplicate_notifications(db, fixture_payload):
    run_ingest(db, fixture_payload, run_id=uuid4())
    n = db.count("notification_outbox")
    run_ingest(db, fixture_payload, run_id=uuid4())
    assert db.count("notification_outbox") == n

The second assertion in the first test is the one people leave out and then regret: comparing full table snapshots rather than row counts catches the case where a re-run rewrote every run_id, which count-based tests pass happily.

Failure recovery

Duplicates already in the table. De-duplicate on the natural key, keeping the row with the earliest run_id (the one that actually first produced the value), then add the unique constraint so it cannot recur. Adding the constraint first will fail while duplicates exist, which is a useful way to find out how many there are.

Recovering from each violation once it has happened A three-row by three-column matrix of recoveries. Rows are duplicates already in the table, rows deleted by a truncated payload, and duplicate notifications already sent. Columns are how to detect the damage, how to repair it, and what prevents a recurrence. Duplicates are found by grouping on the natural key and repaired by keeping the earliest run. Deleted rows need the last good snapshot or a re-fetch. Duplicate notifications cannot be repaired, only corrected, and one correction is better than several apologies. Recovering from each violation once it has happened detect it by repair prevented by Duplicates in the table group by the natural key keep the earliest run_id unique constraint Rows deleted by a truncated payload count against trailing median restore snapshot or re-fetch tombstone + assertion Duplicate notifications sent outbox rows without a key one correction, not apologies outbox idempotency key clean needs judgement costly or impossible
Only the first row is fully repairable from what you already have. The second needs a snapshot or a re-fetch — which is what makes the source archive worth its storage — and the third cannot be repaired at all, only corrected once.

Rows deleted by a truncated payload. Restore from the last good snapshot, then re-run with the completeness assertion in place. If no snapshot exists, re-fetch — this is the scenario that makes the source archive worth keeping.

Duplicate notifications already sent. Send one correction naming the duplicates rather than several apologies, and add the outbox key before anything else. Subscribers tolerate one correction; they stop reading after the third duplicate.

Frequently asked questions

Is ON CONFLICT DO UPDATE enough on its own?

It stops duplicates but does not make the task fully idempotent. Without the WHERE content_digest <> EXCLUDED.content_digest guard, a re-run touches every row and stamps it with the new run_id, so provenance claims that run produced values it merely re-confirmed, and anything watching run_id sees the whole county as freshly changed. The guard makes an unchanged row a genuine no-op.

Why tombstone instead of deleting?

Because absence from a payload is ambiguous: it means “retired” on a complete payload and “not fetched yet” on a truncated one, and a re-run after a failure is precisely when the payload is truncated. A tombstone is re-runnable and reversible; a delete is neither. Tombstones also keep the parcel’s history reachable, which a delete destroys along with the row.

Should the run id be part of the natural key?

No — that is the most common way to accidentally break idempotency. Including the run id makes every run’s rows distinct by construction, so nothing ever conflicts and every re-run duplicates the data. The run id belongs in a column that records which run last changed the value, not in the identity of the value itself.

What about tasks that call an external API?

Move the call out of the ingestion task and behind an outbox, so the task only writes rows and a separate worker performs the un-undoable act. Where the external system supports an idempotency key, pass the same key the outbox uses; where it does not, the outbox’s sent_at column is what stops a second delivery, and it must be written in the same transaction as the send is acknowledged.