Idempotent Batch Reprocessing with Python

A trading-partner capture endpoint acknowledges the first three chunks of a nightly serialization batch, then the worker dies on chunk four before it records that acknowledgment. Naively resubmitting the whole batch on restart would write duplicate ObjectEvent rows for units already committed, and a compliance reviewer reconstructing that lot’s history would find two conflicting commission records for the same serial. This page shows how to replay a failed or partially-applied batch of DSCSA serialization events so every retry lands exactly once — never zero times, never twice. It is a code-first pattern within Async Batch Processing Pipelines, the reference architecture for chunking, dispatching, and retrying serialization event batches, and it protects the same repository the broader Serialization Data Ingestion & EPCIS Event Sync pipeline must keep gap-free for six years of audit retention.

Prerequisites #

• Python 3.10+ — the snippets use dataclass, dict generics, and modern asyncio.
• asyncpg (or psycopg 3 in async mode) — an async Postgres driver whose command tag lets you tell “inserted” from “skipped” after an ON CONFLICT upsert.
• Pydantic v2 — validates each event’s shape before it reaches the idempotency-key step; malformed events should never arrive here. That gate is covered in Schema Validation & Error Handling — this page assumes events are already well-formed.
• A batch dispatcher upstream — this reprocessor is the recovery half of the send pipeline in Building Async Batch Processors for Serialization Events; that guide chunks and transmits, this one guarantees a crashed or redelivered chunk never double-commits.
• DSCSA data prerequisites — each event needs a GTIN (01), serial (21), lot (10), expiry (17), a bizStep, and a timezone-aware eventTime, since these feed the idempotency key directly.
• A durable checkpoint table — a small Postgres table that survives process restarts, so the reprocessor never relies on an in-memory offset.

Step-by-Step Solution #

Step 1 — Derive a deterministic idempotency key per event #

Key the row on the event’s own content, not on the batch. A batch-level key only tells you whether an entire chunk was seen before; a per-event key lets you safely replay a chunk where three events already committed and the fourth did not. Hash the normalized GTIN, serial, UTC event time, and business step — enough fields that a commission and a later decommission of the same serial never collide, while a byte-identical resend always reduces to the same key.

import hashlib
from datetime import datetime, timezone


def derive_idempotency_key(event: dict) -> str:
    """Derive a deterministic idempotency key for one EPCIS ObjectEvent.

    Includes gtin, serial, a UTC-normalized event_time, and biz_step so a
    commission and a later decommission of the same unit never collide,
    while a byte-identical resend always hashes to the same key.
    """
    event_time = event["event_time"]
    if event_time.tzinfo is None:
        raise ValueError("event_time must be timezone-aware before hashing")
    normalized_time = event_time.astimezone(timezone.utc).isoformat()

    canonical = "|".join([
        event["gtin"],       # AI (01)
        event["serial"],     # AI (21)
        normalized_time,
        event["biz_step"],
    ])
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

DSCSA/GS1 note: interoperable exchange must be gap-free and non-duplicative; a content-derived key gives effectively-once persistence per unit even when the network forces a resend after an unacknowledged 2xx.

Step 2 — Persist through an ON CONFLICT DO NOTHING gate #

Give the idempotency key a unique constraint and let the database enforce exactly-once persistence, rather than checking “does this exist?” first — that two-step pattern has a race window under concurrent workers. A single INSERT ... ON CONFLICT DO NOTHING statement is atomic and its command tag tells you whether the row was new.

CREATE TABLE IF NOT EXISTS epcis_object_event (
    idempotency_key   TEXT PRIMARY KEY,
    gtin              TEXT NOT NULL,
    serial_number     TEXT NOT NULL,
    lot_number        TEXT NOT NULL,
    expiration_date   DATE NOT NULL,
    biz_step          TEXT NOT NULL,
    event_time        TIMESTAMPTZ NOT NULL,
    source_offset     BIGINT NOT NULL,
    committed_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

import asyncpg


async def upsert_event(
    conn: asyncpg.Connection, key: str, event: dict, offset: int
) -> bool:
    """Insert one EPCIS event idempotently; return True if newly committed."""
    result = await conn.execute(
        """
        INSERT INTO epcis_object_event (
            idempotency_key, gtin, serial_number, lot_number,
            expiration_date, biz_step, event_time, source_offset
        )
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
        ON CONFLICT (idempotency_key) DO NOTHING
        """,
        key,
        event["gtin"],
        event["serial"],
        event["lot"],
        event["expiration_date"],
        event["biz_step"],
        event["event_time"],
        offset,
    )
    return result.endswith("INSERT 0 1")

DSCSA/GS1 note: the PRIMARY KEY constraint is the single source of truth for “already committed” — a reviewer must be able to prove no serial’s history holds two commission records for one lifecycle event, and a database constraint proves that better than application logic ever can.

Step 3 — Track a processed-offset high-water mark #

The idempotency key protects individual rows; the high-water mark protects the reprocessor from redoing work it does not need to redo. Persist the highest source offset (a Kafka offset, a file line number, a paginated API cursor — whatever the upstream uses) that has been durably committed, and only ever move it forward.

CREATE_CHECKPOINT_TABLE = """
CREATE TABLE IF NOT EXISTS ingestion_checkpoint (
    source_name            TEXT PRIMARY KEY,
    last_committed_offset  BIGINT NOT NULL
);
"""


async def advance_checkpoint(
    conn: asyncpg.Connection, source: str, offset: int
) -> None:
    """Advance the high-water mark, but never let it move backward."""
    await conn.execute(
        """
        INSERT INTO ingestion_checkpoint (source_name, last_committed_offset)
        VALUES ($1, $2)
        ON CONFLICT (source_name) DO UPDATE
        SET last_committed_offset = GREATEST(
            ingestion_checkpoint.last_committed_offset, EXCLUDED.last_committed_offset
        )
        """,
        source,
        offset,
    )

DSCSA/GS1 note: a monotonic checkpoint answers an auditor’s question — “how do you know nothing was skipped?” — with a mechanism, not a promise; the offset and row counts are reconciled against the source system in Verification below.

Step 4 — Build the async reprocessor around one transaction #

Wrap each event’s upsert and the checkpoint advance in a single transaction per batch. This turns a crash mid-batch into a no-op: either every event and the new checkpoint commit together, or the transaction rolls back and the next run starts from the same place.

async def reprocess_batch(
    pool: asyncpg.Pool, source: str, events: list[dict]
) -> dict[str, int]:
    """Replay a batch of EPCIS events idempotently and advance the checkpoint.

    Safe to call repeatedly with the same batch: rows that already exist
    are skipped by the unique constraint, and the checkpoint only moves
    forward, so a redelivered or manually re-triggered batch is a no-op.
    """
    stats = {"inserted": 0, "skipped": 0}
    async with pool.acquire() as conn:
        async with conn.transaction():
            max_offset = -1
            for event in events:
                key = derive_idempotency_key(event)
                inserted = await upsert_event(conn, key, event, event["offset"])
                stats["inserted" if inserted else "skipped"] += 1
                max_offset = max(max_offset, event["offset"])
            if max_offset >= 0:
                await advance_checkpoint(conn, source, max_offset)
    return stats

DSCSA/GS1 note: grouping the upserts and checkpoint write in one transaction preserves the ordered, gap-free lifecycle EPCIS depends on: a decommission can never be recorded processed while its commission silently failed to persist.

Step 5 — Reconcile partial batches after a crash #

On restart, never trust an in-memory offset left over from the crashed process — read the durable checkpoint, then ask the upstream source for everything after it. The same batch may legitimately arrive twice (a consumer-group rebalance in a Kafka-backed pipeline is the classic trigger, the same redelivery scenario handled at the streaming layer in Deduplicating EPCIS Events in a Kafka Stream), so reconciliation must tolerate offsets at or before the checkpoint arriving again.

from typing import Awaitable, Callable

FetchAfter = Callable[[str, int], Awaitable[list[dict]]]


async def resume_from_checkpoint(
    pool: asyncpg.Pool, source: str, fetch_events_after: FetchAfter
) -> dict[str, int]:
    """On startup, resume from the last durable checkpoint, not from
    whatever offset the crashed worker last held in memory."""
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT last_committed_offset FROM ingestion_checkpoint "
            "WHERE source_name = $1",
            source,
        )
    last_offset = row["last_committed_offset"] if row else -1
    pending_events = await fetch_events_after(source, last_offset)
    return await reprocess_batch(pool, source, pending_events)

DSCSA/GS1 note: requesting “everything after the checkpoint,” not “everything the crashed worker held,” makes recovery deterministic: the repository ends up identical regardless of when the crash happened, which is exactly what an inspector reconstructing a unit’s lifecycle depends on.

Verification #

Confirm idempotency with a test that replays the same batch twice and asserts the second pass inserts nothing:

import pytest
from datetime import datetime, timezone


class FakeConnection:
    """Minimal in-memory stand-in for an asyncpg connection."""

    def __init__(self) -> None:
        self.rows: dict[str, dict] = {}
        self.checkpoint = -1

    async def execute(self, query: str, *args) -> str:
        if "INSERT INTO epcis_object_event" in query:
            key = args[0]
            if key in self.rows:
                return "INSERT 0 0"
            self.rows[key] = {"gtin": args[1], "offset": args[7]}
            return "INSERT 0 1"
        if "INSERT INTO ingestion_checkpoint" in query:
            self.checkpoint = max(self.checkpoint, args[1])
            return "INSERT 0 1"
        raise ValueError("unexpected query in fake connection")


def make_event(offset: int) -> dict:
    return {
        "gtin": "00312345000058",
        "serial": f"SN{offset:06d}",
        "lot": "LOT42",
        "expiration_date": "2027-12-31",
        "biz_step": "urn:epcglobal:cbv:bizstep:commissioning",
        "event_time": datetime(2026, 7, 1, 10, 0, tzinfo=timezone.utc),
        "offset": offset,
    }


@pytest.mark.asyncio
async def test_replaying_the_same_batch_is_idempotent():
    conn = FakeConnection()
    events = [make_event(o) for o in range(100, 106)]

    first_pass = {"inserted": 0, "skipped": 0}
    for event in events:
        key = derive_idempotency_key(event)
        inserted = await upsert_event(conn, key, event, event["offset"])
        first_pass["inserted" if inserted else "skipped"] += 1

    second_pass = {"inserted": 0, "skipped": 0}
    for event in events:  # simulate a broker redelivery of the same batch
        key = derive_idempotency_key(event)
        inserted = await upsert_event(conn, key, event, event["offset"])
        second_pass["inserted" if inserted else "skipped"] += 1

    assert first_pass == {"inserted": 6, "skipped": 0}
    assert second_pass == {"inserted": 0, "skipped": 6}
    assert len(conn.rows) == 6

Beyond the unit test, run two checks before trusting the reprocessor against live data. First, reconcile row counts: SELECT count(*) FROM epcis_object_event WHERE source_offset BETWEEN :start AND :end should equal what the upstream source reports for that range — a mismatch means events were dropped, not duplicated. Second, kill the worker mid-batch in staging and confirm on restart that the checkpoint reflects only fully-committed offsets, never a half-applied batch.

Gotchas & Edge Cases #

• Too little entropy in the key. Hashing only the GTIN and serial collapses every lifecycle event for that unit — commission, ship, decommission — onto one key, so later events silently vanish. Include the event time and business step, or an EPCIS eventID if the source provides one.
• UTC vs. local event_time when hashing. Normalizing inconsistently means the same physical event recorded with two different offsets produces two different keys, defeating deduplication. Always call .astimezone(timezone.utc) before hashing.
• GTIN leading zeros. Never cast the GTIN to int anywhere in the key pipeline; a stripped leading zero changes both the GS1 modulo-10 check digit and the hash, so the “same” unit produces a different key on replay.
• ON CONFLICT DO NOTHING hides payload corrections. A corrected retry with the same key but a different lot or expiry is silently discarded, keeping the original row. If corrections are expected, use ON CONFLICT DO UPDATE guarded by a payload hash comparison instead.
• Checkpoint advanced outside the transaction. A separate commit for the offset write lets a crash between the two permanently skip events that never persisted. Keep both writes in the same transaction, as in Step 4.

Related #

• Up to the parent topic: Async Batch Processing Pipelines
• Building Async Batch Processors for Serialization Events — the dispatch side this reprocessor recovers behind
• Deduplicating EPCIS Events in a Kafka Stream — the equivalent idempotency problem at the streaming layer
• Serialization Data Ingestion & EPCIS Event Sync — the pipeline this reprocessing pattern keeps gap-free