Exactly-Once Processing with Idempotency Keys in Grant Pipelines

This guide is part of the Async Batch Processing Pipelines section within the broader Data Ingestion & Grant Parsing Workflows framework, and it solves one narrow problem: your message broker guarantees at-least-once delivery, so the same grant submission will eventually be handed to a consumer twice — how do you make sure it is only processed once, without ever double-posting an award to the ledger?

There is no such thing as exactly-once delivery over a network. A broker that has handed you a message but never received your acknowledgement has exactly one safe option: redeliver. What you can build is exactly-once effect — an “effectively-once” consumer that recognizes a redelivered submission and refuses to do the side effect a second time. The mechanism is a deterministic idempotency key claimed atomically in a dedup store before any work happens, paired with a transactional outbox so the business write and the dedup record commit or roll back as one unit.

When to Use This Approach #

Reach for idempotency-key deduplication when all three conditions hold:

• Your consumer performs a non-idempotent side effect. Recording an award disbursement, incrementing an obligated-funds total, or emitting a downstream compliance event are all operations where “run twice” means “wrong.” If your consumer only overwrites a row with a full-state upsert keyed by award_id, redelivery is already harmless and you do not need this machinery. The gate matters precisely when a second execution would corrupt a running total or duplicate an artifact.
• The broker is at-least-once, which is nearly all of them. Kafka, RabbitMQ, SQS, and Redis Streams all redeliver on a missing acknowledgement by design. The concurrent dispatch that feeds this consumer is built in Building Async Batch Processors for Grant Submissions; this page assumes those workers are already pulling submissions off such a broker.
• The side effect is auditable. Grant disbursements sit inside a system of internal control governed by 2 CFR §200.303. A double-post is not a cosmetic bug; it is a control failure an auditor can find. The dedup record is itself an evidentiary artifact, so it must be retained for the period required by 2 CFR §200.334.

Broker configuration, partition assignment, consumer-group rebalancing, and retry backoff are explicitly out of scope — concurrency and dispatch live in the parent Async Batch Processing Pipelines section, and failure triage lives in Error Categorization & Logging. This page covers only the claim-gate contract that sits between “message received” and “message processed.”

Step-by-Step Implementation #

The reference implementation targets Python 3.10+ and a Postgres dedup store via psycopg 3. Redis with SETNX is shown as an alternative claim primitive. Install the two clients:

pip install "psycopg[binary]==3.2.1" "redis==5.0.7"

One contract to internalize first: the key is claimed before the work, and the work commits before the acknowledgement. That ordering is the entire safety argument. If you ack first and work second, a crash loses the message. If you work first and claim second, a crash between them double-executes on redelivery.

Step 1: Derive a deterministic idempotency key #

The key must be a pure function of the submission’s identity, so that any redelivery of the same submission produces byte-for-byte the same key on a different worker, a different process, or a week later. Combine the business identity — award_id and submission_timestamp — with a SHA-256 of the raw payload, so a resubmission that genuinely changes the payload gets a distinct key and is not suppressed.

import hashlib
import logging

AUDIT_LOGGER = logging.getLogger("grant_pipeline.idempotency")
AUDIT_LOGGER.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
AUDIT_LOGGER.addHandler(_handler)


def derive_idempotency_key(award_id: str, submission_timestamp: str, payload: bytes) -> str:
    """Return a deterministic idempotency key for one grant submission."""
    if not award_id or not submission_timestamp:
        raise ValueError("award_id and submission_timestamp are required to derive a key")
    payload_digest: str = hashlib.sha256(payload).hexdigest()
    material: str = f"{award_id}|{submission_timestamp}|{payload_digest}"
    return hashlib.sha256(material.encode("utf-8")).hexdigest()

Never derive the key from a broker-assigned message id or delivery timestamp — those change on every redelivery, so two copies of the same submission would get different keys and both would run. The key is about the submission’s identity, not the delivery’s.

Step 2: Claim the key atomically before doing work #

The claim is a single atomic write that either succeeds (first time this key is seen) or conflicts (already seen). In Postgres, an INSERT ... ON CONFLICT DO NOTHING against a unique-keyed table does this in one statement — cursor.rowcount is 1 on a fresh claim and 0 on a duplicate. There is no read-then-write race because the uniqueness check is the write.

from enum import Enum
import psycopg


class ClaimResult(Enum):
    CLAIMED = "claimed"
    DUPLICATE = "duplicate"


def claim_key(conn: psycopg.Connection, key: str, award_id: str) -> ClaimResult:
    """Atomically claim an idempotency key. Does not commit — caller owns the tx."""
    try:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO processed_messages (idempotency_key, award_id, claimed_at) "
                "VALUES (%s, %s, now()) ON CONFLICT (idempotency_key) DO NOTHING",
                (key, award_id),
            )
            if cur.rowcount == 1:
                return ClaimResult.CLAIMED
            AUDIT_LOGGER.info("Duplicate key %s for award %s; already processed", key[:12], award_id)
            return ClaimResult.DUPLICATE
    except psycopg.Error as exc:
        AUDIT_LOGGER.error("Claim failed for key %s: %s", key[:12], exc)
        raise RuntimeError(f"idempotency claim failed for {key[:12]}") from exc

The Redis equivalent is SET key value NX EX <ttl>, which returns truthy only if the key did not exist. Redis is faster but volatile: a SETNX claim is a dedup cache, not a durable audit record. For a regulated grant ledger, prefer the Postgres table so the claim itself survives as retained evidence under 2 CFR §200.334; use Redis only as a fast front-line filter in front of it.

Step 3: Skip and acknowledge on a duplicate claim #

A duplicate claim means the work already ran and committed in a previous delivery. The correct response is to do nothing to the business state and simply acknowledge the broker so it stops redelivering. Skipping without acking is the classic bug — the message loops forever.

def handle_duplicate(conn: psycopg.Connection, ack) -> None:
    """On a duplicate claim, roll back any open work and ack so redelivery stops."""
    conn.rollback()  # release the aborted claim attempt; touch no business rows
    try:
        ack()
    except Exception as exc:
        AUDIT_LOGGER.error("Ack failed on duplicate skip: %s", exc)
        raise RuntimeError("failed to ack a duplicate message") from exc
    AUDIT_LOGGER.info("Duplicate acknowledged and skipped with no side effect")

The ack argument is the broker-specific acknowledgement callable — msg.ack() for RabbitMQ, an offset commit for Kafka, delete_message for SQS. Passing it in keeps this function broker-agnostic and testable with a stub.

Step 4: Commit the side effect and the claim with a transactional outbox #

The claim record and the business write must land together or not at all. If the claim commits but the disbursement does not, a redelivery sees a duplicate and skips work that never happened — a silent lost update. The fix is the transactional outbox: within one database transaction, write the claim, write the business row, and insert a row into an outbox table describing the downstream event. A separate relay process publishes outbox rows to the broker afterward. Because all three writes share one commit, they are atomic.

import json
from typing import Any, Mapping


def process_in_transaction(
    conn: psycopg.Connection, key: str, submission: Mapping[str, Any]
) -> ClaimResult:
    """Claim, apply the side effect, and enqueue the outbox event in one commit."""
    result: ClaimResult = claim_key(conn, key, submission["award_id"])
    if result is ClaimResult.DUPLICATE:
        return result

    try:
        with conn.cursor() as cur:
            # 1. the business side effect (idempotent within this tx via the claim)
            cur.execute(
                "INSERT INTO disbursements (award_id, amount_cents, idempotency_key) "
                "VALUES (%s, %s, %s)",
                (submission["award_id"], submission["amount_cents"], key),
            )
            # 2. the downstream event, published later by the outbox relay
            cur.execute(
                "INSERT INTO outbox (event_type, payload, created_at) VALUES (%s, %s, now())",
                ("disbursement.recorded", json.dumps({"award_id": submission["award_id"]})),
            )
        conn.commit()  # claim + disbursement + outbox row commit atomically
    except psycopg.Error as exc:
        conn.rollback()
        AUDIT_LOGGER.error("Work tx failed for key %s: %s", key[:12], exc)
        raise RuntimeError(f"processing failed for {key[:12]}") from exc

    AUDIT_LOGGER.info("Processed award %s under key %s", submission["award_id"], key[:12])
    return ClaimResult.CLAIMED

Publishing the downstream event directly inside the consumer instead of through the outbox reopens the dual-write problem — the publish could succeed while the transaction rolls back, or vice versa. The outbox makes “record the disbursement” and “announce the disbursement” a single durable fact; the relay’s own delivery to the broker is then just another at-least-once hop that its consumers dedup the same way.

Step 5: Survive the crash-after-work-before-ack window #

There is exactly one window this design cannot eliminate: the transaction has committed but the process dies before the acknowledgement reaches the broker. This is the annotated span in the diagram. The broker, having seen no ack, redelivers. What makes it safe is Step 1 plus Step 2: the redelivery re-derives the same key, the claim INSERT hits the existing row, ClaimResult.DUPLICATE comes back, and the consumer acks and skips. The committed work is never repeated.

def consume_once(conn: psycopg.Connection, submission: Mapping[str, Any], ack) -> None:
    """Full effectively-once handling of a single delivery."""
    key: str = derive_idempotency_key(
        submission["award_id"], submission["submission_timestamp"], submission["raw_payload"]
    )
    result: ClaimResult = process_in_transaction(conn, key, submission)
    if result is ClaimResult.DUPLICATE:
        handle_duplicate(conn, ack)
        return
    try:
        ack()  # if the process dies before this line, redelivery -> DUPLICATE -> skip
    except Exception as exc:
        AUDIT_LOGGER.error("Ack failed after commit for key %s: %s", key[:12], exc)
        raise RuntimeError("commit succeeded but ack failed; redelivery will be skipped") from exc

The ordering is load-bearing: commit first, ack second. A crash before the commit loses nothing because the transaction rolls back and the message redelivers as a genuine first attempt. A crash after the commit but before the ack is caught by the duplicate gate. There is no ordering that also survives a broker that never acknowledges — but every failure resolves to “processed exactly once.”

Verification #

Confirm the gate is effectively-once with four checks:

1. The key is deterministic. Call derive_idempotency_key twice with identical award_id, submission_timestamp, and payload bytes and assert the two 64-character hex strings are equal; then change one payload byte and assert the key differs. This proves redelivery collapses to one key while a real resubmission does not.
2. A duplicate claim is rejected atomically. Insert a key, then call claim_key again on the same key and assert it returns ClaimResult.DUPLICATE with no new row — the ON CONFLICT did the dedup, not application logic.
3. The crash window is covered. Commit a process_in_transaction call, then replay the identical submission (simulating redelivery before ack) and assert the second call returns DUPLICATE and the disbursements row count stayed at one.
4. The outbox and disbursement share a commit. Force the outbox insert to fail and assert the disbursement row is also absent after rollback — proving the two writes are atomic.

A compliant run emits one INFO audit line per processed submission carrying the truncated key and award id, and one INFO line per suppressed duplicate. Ship those to a write-once tier so the dedup trail satisfies the retention period under 2 CFR §200.334.

key_a = derive_idempotency_key("A-2026-014", "2026-07-01T09:00:00Z", b"{...}")
key_b = derive_idempotency_key("A-2026-014", "2026-07-01T09:00:00Z", b"{...}")
assert key_a == key_b and len(key_a) == 64
first = process_in_transaction(conn, key_a, submission)
second = process_in_transaction(conn, key_a, submission)  # redelivery
assert first is ClaimResult.CLAIMED and second is ClaimResult.DUPLICATE

Common Errors & Fixes #

Related #

• Parent section: Async Batch Processing Pipelines
• The workers that feed this gate: Building Async Batch Processors for Grant Submissions
• Where suppressed-duplicate and failed-claim events are triaged: Error Categorization & Logging