Batch Idempotency Key Generation

A batch of inventory deltas is only safe to retry if the retry is provably the same batch. The moment a worker crashes after committing but before acknowledging, the broker redelivers, and a naive pipeline posts every Schedule II quantity movement a second time — inflating the perpetual count and manufacturing a discrepancy that a DEA inspector will read as diversion. The defence is an idempotency key: a deterministic fingerprint that lets the ledger recognise a batch it has already applied and turn the second write into a no-op. This recipe builds that key correctly, and it operates within the Async Batch Processing for Inventory Updates workflow, whose broker layer claims the key with an atomic SET NX before any worker mutates a controlled-substance balance.

The hard part is not the hashing — it is what you hash. Two byte streams that a human reads as “the same batch” will produce two different SHA-256 digests if the field order shifts, a float is formatted differently, or a timestamp carries a different timezone offset. A key that changes when the payload is semantically identical is worse than no key at all: it defeats deduplication silently, and the batch double-posts anyway. Correct key generation is therefore a canonicalization problem first and a hashing problem second.

Problem framing & prerequisites

The upstream layer accepts each canonical event and forwards it toward the append-only ledger described in Defining Audit Boundaries for Controlled Substances. Between acceptance and commitment there is a redelivery window — any at-least-once broker (RabbitMQ, SQS, Redis Streams) will replay an unacknowledged message, and that is by design, because the alternative is losing the batch entirely. The idempotency key is what makes at-least-once delivery safe for a 21 CFR § 1304.11 perpetual record: the batch is delivered one or more times but applied exactly once.

Environment for the code below:

• Python 3.11+ — the implementation uses @dataclass(frozen=True), the union-type syntax, and decimal.Decimal for quantity normalization.
• Standard library only: hashlib, json, decimal, datetime, logging. No third-party dependency is needed to generate a key; add pytest for the verification block.
• Regulatory grounding you should already hold: 21 CFR § 1304.11 (perpetual inventory must be complete and accurate — no double-count), 21 CFR § 1304.04 (records must be reproducible for inspection), and 45 CFR § 164.312(b) (audit controls over electronic records). The key itself carries no patient data, which keeps the broker inside the HIPAA transmission boundary.

Implementation: canonicalize, then hash

The rule is absolute: never hash a Python object or a default json.dumps string directly. Route every batch through one canonicalization function so that any two semantically identical batches serialize to identical bytes. The function fixes the three things that silently break determinism — key order, numeric formatting, and timestamp representation — then hashes the result.

import hashlib
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal, ROUND_HALF_UP

logger = logging.getLogger("pharmacy.batch.idempotency")

# Units are normalized to a single canonical token so "ea"/"EA"/"each"
# cannot produce three different keys for the same physical movement.
_UNIT_CANON = {
    "ea": "EA", "each": "EA", "EA": "EA",
    "ml": "ML", "ML": "ML",
    "gm": "GM", "g": "GM", "GM": "GM",
    "tab": "TAB", "TAB": "TAB",
    "cap": "CAP", "CAP": "CAP",
}
_QTY_QUANTUM = Decimal("0.001")  # fixed 3-dp scale for every quantity


@dataclass(frozen=True)
class InventoryItem:
    ndc11: str          # 11-digit normalized product identity
    lot: str
    qty_delta: Decimal  # signed movement, carried as Decimal end to end
    unit: str
    event_type: str     # dispense | receipt | adjustment | waste


def _canon_qty(value: Decimal) -> str:
    """Quantities become fixed-scale decimal *strings* — never floats.

    0.1 + 0.2 as a float is 0.30000000000000004; formatted two ways it yields
    two hashes for one batch. Quantizing to a fixed quantum removes that.
    """
    return str(Decimal(value).quantize(_QTY_QUANTUM, rounding=ROUND_HALF_UP))


def _canon_ts(ts: datetime) -> str:
    """Every timestamp is coerced to UTC and rendered with a fixed format.

    A naive datetime is rejected: an unknown offset makes the key ambiguous.
    """
    if ts.tzinfo is None:
        raise ValueError("timestamp must be timezone-aware to key deterministically")
    return ts.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def canonical_item(item: InventoryItem) -> dict[str, str]:
    """One item as a dict of canonical primitives — sorted at serialization."""
    unit = _UNIT_CANON.get(item.unit.strip(), item.unit.strip().upper())
    return {
        "ndc11": item.ndc11,
        "lot": item.lot,
        "qty_delta": _canon_qty(item.qty_delta),
        "unit": unit,
        "event_type": item.event_type,
    }


def canonical_bytes(facility_uuid: str, batch_ts: datetime,
                    items: list[InventoryItem]) -> bytes:
    """Deterministic byte string for an entire batch.

    Items are sorted by their canonical identity so that the *order* deltas
    arrive in the payload cannot change the key. json.dumps uses sort_keys and
    compact separators so whitespace never enters the digest.
    """
    canon_items = sorted(
        (canonical_item(i) for i in items),
        key=lambda d: (d["ndc11"], d["lot"], d["event_type"], d["qty_delta"]),
    )
    envelope = {
        "facility_uuid": facility_uuid,
        "batch_ts": _canon_ts(batch_ts),
        "items": canon_items,
    }
    return json.dumps(envelope, sort_keys=True, separators=(",", ":")).encode("utf-8")


def batch_idempotency_key(facility_uuid: str, batch_ts: datetime,
                          items: list[InventoryItem]) -> str:
    """Per-batch key: one SHA-256 over the canonical batch, namespaced."""
    digest = hashlib.sha256(canonical_bytes(facility_uuid, batch_ts, items)).hexdigest()
    key = f"idem:batch:{facility_uuid}:{digest}"
    logger.info("BATCH_KEY_GENERATED key=%s item_count=%d", key, len(items))
    return key


def item_idempotency_key(facility_uuid: str, item: InventoryItem) -> str:
    """Per-item key: use when items commit independently rather than atomically."""
    payload = json.dumps(
        {"facility_uuid": facility_uuid, **canonical_item(item)},
        sort_keys=True, separators=(",", ":"),
    ).encode("utf-8")
    return f"idem:item:{facility_uuid}:{hashlib.sha256(payload).hexdigest()}"

Per-batch vs per-item keys. Choose per-batch when the whole batch commits inside one atomic ledger transaction — the single key guards the all-or-nothing write, and a redelivery of the batch is rejected wholesale. Choose per-item keys when items are applied independently (for example, a partial batch where some items quarantine and others commit): each item then carries its own key, so a replay re-applies only the items that never landed. Do not mix the two for one write path — the deduplication guarantee only holds if the key granularity matches the commit granularity.

Because the key is derived purely from the payload, it survives a worker restart, a rolling deploy, and a broker failover. The same batch produces the same key on any host, at any time, which is exactly the property the offline path depends on: batches buffered during a disconnect and replayed on reconnect through the Fallback Routing for Offline Sync layer carry their original keys, so a delta partially delivered before the outage cannot double-post when connectivity returns.

Verification & testing

Two properties must be proven: that semantically identical batches with different surface representations collapse to one key, and that a genuinely different batch produces a different key. The block below shuffles item order, varies key order, mixes unit spellings, and shifts timezone offset — every case must yield the identical digest.

import pytest
from datetime import timedelta

def _item(ndc, lot, qty, unit="EA", ev="receipt"):
    return InventoryItem(ndc11=ndc, lot=lot, qty_delta=Decimal(qty),
                         unit=unit, event_type=ev)

FAC = "5f1e6c2a-0000-4000-8000-000000000abc"
TS = datetime(2026, 7, 16, 14, 30, 0, tzinfo=timezone.utc)

def test_item_order_does_not_change_key():
    a = [_item("00093-0058-01", "L1", "10"), _item("00185-0674-01", "L2", "5")]
    b = list(reversed(a))
    assert batch_idempotency_key(FAC, TS, a) == batch_idempotency_key(FAC, TS, b)

def test_unit_spelling_is_normalized():
    a = [_item("00093-0058-01", "L1", "10", unit="ea")]
    b = [_item("00093-0058-01", "L1", "10", unit="EACH")]
    assert batch_idempotency_key(FAC, TS, a) == batch_idempotency_key(FAC, TS, b)

def test_timezone_offset_is_normalized():
    est = timezone(timedelta(hours=-4))
    ts_est = TS.astimezone(est)              # same instant, different offset
    items = [_item("00093-0058-01", "L1", "10")]
    assert batch_idempotency_key(FAC, TS, items) == batch_idempotency_key(FAC, ts_est, items)

def test_float_style_quantities_are_stable():
    a = [_item("00093-0058-01", "L1", "0.3")]
    b = [_item("00093-0058-01", "L1", str(Decimal("0.1") + Decimal("0.2")))]
    assert batch_idempotency_key(FAC, TS, a) == batch_idempotency_key(FAC, TS, b)

def test_different_ndc_yields_different_key():
    a = [_item("00093-0058-01", "L1", "10")]
    b = [_item("00185-0674-01", "L1", "10")]
    assert batch_idempotency_key(FAC, TS, a) != batch_idempotency_key(FAC, TS, b)

def test_naive_timestamp_is_rejected():
    with pytest.raises(ValueError):
        batch_idempotency_key(FAC, datetime(2026, 7, 16, 14, 30, 0),
                              [_item("00093-0058-01", "L1", "10")])

The emitted audit line carries the key and item count but never a patient identifier or a raw quantity that could be linked to a dispensing event:

INFO pharmacy.batch.idempotency BATCH_KEY_GENERATED \
     key=idem:batch:5f1e6c2a-0000-4000-8000-000000000abc:9f2c7d...e41a item_count=2

Storing the key alongside each committed ledger row makes the inspection check trivial: re-derive the key from the persisted canonical fields and confirm it matches. A mismatch proves the stored payload was altered after commit, which is the tamper signal 21 CFR § 1304.04 reproducibility is meant to surface.

Gotchas & compliance pitfalls

• Non-canonical JSON is the number-one silent failure. json.dumps(payload) without sort_keys=True orders keys by insertion, so the same batch built by two code paths hashes differently and double-posts. Always route through one canonicalization function; never hash an ad-hoc serialization.
• Floats do not round-trip. 0.1 + 0.2 formats as 0.30000000000000004 in one place and 0.3 in another. Carry quantities as Decimal, quantize to a fixed scale, and hash the string. A float anywhere in the key path is a latent duplicate.
• Timezone in timestamps. A batch_ts of 14:30-04:00 and 18:30Z are the same instant but different strings. Coerce to UTC with a fixed format, and reject naive datetimes outright — an unknown offset makes the key non-deterministic and therefore unusable.
• Key reuse across drug NDCs. If the key omits the ndc11 (or derives only from a batch counter), two batches touching different products can collide, and the second is silently dropped as a “duplicate” — a lost Schedule II movement. The NDC, lot, and signed quantity must all feed the digest.
• Namespacing per facility. A multi-site deployment must include facility_uuid in the key. Two sites can legitimately submit an identical item list in the same second; without the facility namespace one site’s batch suppresses the other’s.
• TTL shorter than the retry window. The broker claims the key with an expiry. If the TTL is shorter than the maximum retry-plus-backoff window, an old key expires and a late redelivery re-commits. Set the key TTL to comfortably exceed the longest possible retry horizon.

Frequently Asked Questions

Related

• Async Batch Processing for Inventory Updates — parent workflow whose broker claims the key before committing.
• Async Retry with Exponential Back-off — the retry engine that must tie every attempt to a stable key.
• Fallback Routing for Offline Sync — replay of buffered batches carrying their original keys.
• Defining Audit Boundaries for Controlled Substances — the append-only ledger the key protects from double-posting.