Dead-Letter Queue Design for Inventory Events
A dead-letter quarantine for controlled-substance inventory events that fail validation or exhaust retries, with replay and poison-message isolation.
A dead-letter queue is the difference between a bug and a compliance gap. When a controlled-substance inventory event fails validation or exhausts its retries, an ordinary system drops it and moves on — but a dropped Schedule II movement is an unreconciled discrepancy the moment a DEA inspector counts the shelf. A dead-letter queue (DLQ), designed as a durable quarantine rather than a trash can, guarantees that a failed event is preserved with its full context until a human resolves it. This recipe designs that quarantine, and it operates within the Error Handling & Retry Mechanisms subsystem, which routes exhausted and hard-failed events into it.
The design has three jobs. It must capture enough context to reconstruct and replay the event (original payload, why it failed, how many times it was tried). It must let an operator redrive a corrected event back through ingestion without re-introducing the fault. And it must isolate a poison message — one that will fail forever — so it cannot loop endlessly and starve the queue. Get those three right and the DLQ becomes an auditable safety net; get them wrong and it becomes either a black hole or an infinite loop.
Problem framing & prerequisites
Two distinct failure classes land in the DLQ, and the design must serve both. Validation failures — a malformed NDC, a missing lot, a bad DEA schedule — are deterministic: they will fail identically on every attempt, so they arrive without ever entering the retry loop. Exhausted transient failures — a downstream that stayed down past the attempt ceiling of the Async Retry with Exponential Back-off loop — are the opposite: they might well succeed on redrive once the condition clears. The DLQ record must record which class it is, so an operator knows whether to fix the payload or simply re-drive it.
Environment:
- Python 3.11+ — the implementation uses
@dataclass(frozen=True),enum.Enum, and typed collections. - Standard library only:
hashlib,json,logging,dataclasses,datetime,enum. A production deployment persists these records in a WORM-retained store (object-lock S3 or a ledger table); the shape below is storage-agnostic. - Regulatory grounding:
21 CFR § 1304.11(a Schedule II movement must remain reconcilable — it cannot be silently dropped),21 CFR § 1304.04(records reproducible for inspection), and45 CFR § 164.312(b)(audit controls). The quarantine record binds each failed event into the same tamper-evident envelope as the ledger, so the audit boundary is continuous — a property defined by Audit Boundary Definition & Scope.
Implementation: a quarantine record with replay and poison isolation
The quarantine record is the whole design. It is immutable once written, carries the original payload verbatim (so the event can be reconstructed exactly), records the ordered attempt history, and seals itself with a SHA-256 hash so a tampered quarantine entry is detectable. A monotonically incremented redrive_count is the poison-message guard: past a cap, the event is isolated rather than re-queued.
import hashlib
import json
import logging
from dataclasses import dataclass, field, replace
from datetime import datetime, timezone
from enum import Enum
logger = logging.getLogger("pharmacy.dlq")
MAX_REDRIVES = 3 # past this, an event is a poison message and is isolated
class FailureClass(str, Enum):
VALIDATION = "validation" # deterministic — fix payload before redrive
EXHAUSTED = "exhausted" # transient outlived retries — redrive may succeed
POISON = "poison" # redrive cap hit — isolated, needs human root-cause
@dataclass(frozen=True)
class Attempt:
at_utc: str
reason: str
stage: str # "validate" | "commit" | "redrive"
@dataclass(frozen=True)
class QuarantineRecord:
dlq_id: str # stable id = the event's idempotency key
failure_class: FailureClass
original_payload: dict # verbatim, so the event is reconstructable
attempts: tuple[Attempt, ...]
redrive_count: int
quarantined_at_utc: str
audit_hash: str = ""
def sealed(self) -> "QuarantineRecord":
"""Return a copy with a SHA-256 seal over the canonical record body."""
body = {
"dlq_id": self.dlq_id,
"failure_class": self.failure_class.value,
"original_payload": self.original_payload,
"attempts": [a.__dict__ for a in self.attempts],
"redrive_count": self.redrive_count,
"quarantined_at_utc": self.quarantined_at_utc,
}
canonical = json.dumps(body, sort_keys=True, separators=(",", ":"))
digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
return replace(self, audit_hash=digest)
def quarantine(idempotency_key: str, payload: dict,
failure_class: FailureClass, attempts: tuple[Attempt, ...],
redrive_count: int = 0) -> QuarantineRecord:
"""Write a sealed quarantine record. NEVER drop the event.
A Schedule II event that reaches here is preserved in full so the perpetual
record under 21 CFR § 1304.11 can still be reconciled by a pharmacist.
"""
rec = QuarantineRecord(
dlq_id=idempotency_key,
failure_class=failure_class,
original_payload=payload,
attempts=attempts,
redrive_count=redrive_count,
quarantined_at_utc=datetime.now(timezone.utc).isoformat(),
).sealed()
# PHI-free audit line: identity, class, and hash only — never payload values.
logger.warning("DLQ_QUARANTINED dlq_id=%s class=%s redrives=%d audit_hash=%s",
rec.dlq_id, rec.failure_class.value, rec.redrive_count, rec.audit_hash)
return rec
class RedriveError(Exception):
"""Raised when a redrive is attempted on an event that must not be replayed."""
def redrive(rec: QuarantineRecord,
reingest: "callable[[dict], bool]") -> QuarantineRecord | bool:
"""Replay a corrected event through ingestion, guarding against poison loops.
Returns True on successful re-commit, or a new isolated QuarantineRecord
when the redrive cap is reached. Redrive ALWAYS re-runs validation — a
corrected payload is still an untrusted assertion until it passes the gate.
"""
if rec.redrive_count >= MAX_REDRIVES:
poisoned = replace(rec, failure_class=FailureClass.POISON).sealed()
logger.error("DLQ_POISON_ISOLATED dlq_id=%s redrives=%d",
poisoned.dlq_id, poisoned.redrive_count)
return poisoned
attempt = Attempt(at_utc=datetime.now(timezone.utc).isoformat(),
reason="redrive", stage="redrive")
try:
# reingest MUST send the event back through full validation + the
# idempotency-keyed commit, so a duplicate cannot be posted on replay.
if reingest(rec.original_payload):
logger.info("DLQ_REDRIVE_OK dlq_id=%s", rec.dlq_id)
return True
raise RedriveError("reingest returned False")
except Exception as exc: # noqa: BLE001 — re-quarantine with incremented count
return quarantine(
rec.dlq_id, rec.original_payload,
FailureClass.EXHAUSTED,
rec.attempts + (replace(attempt, reason=f"redrive_failed:{type(exc).__name__}"),),
redrive_count=rec.redrive_count + 1,
)
The critical invariant is in redrive: it re-runs the full validation gate, never a shortcut commit. A corrected payload is still an untrusted assertion, and replaying it straight to the ledger would let a still-malformed record slip past the very check that quarantined it. Equally important, reingest routes through the idempotency-keyed commit path, so a redrive of an event that was in fact partially applied cannot double-post.
Verification & testing
The tests prove the three guarantees: a controlled-substance event is preserved (never dropped) with a verifiable seal, redrive re-validates rather than committing blindly, and the poison cap halts an endlessly failing event.
import pytest
def _rec(redrives=0, cls=FailureClass.EXHAUSTED):
payload = {"ndc11": "00093005801", "lot": "L1", "qty_delta": "-2",
"unit": "EA", "event_type": "dispense"}
attempts = (Attempt("2026-07-16T14:30:00Z", "503", "commit"),)
return quarantine("idem:batch:x", payload, cls, attempts, redrives)
def test_event_is_preserved_with_verifiable_seal():
rec = _rec()
# The original payload survives verbatim for reconstruction.
assert rec.original_payload["ndc11"] == "00093005801"
# Re-sealing an untampered record reproduces the stored hash.
assert rec.sealed().audit_hash == rec.audit_hash
def test_tamper_breaks_the_seal():
rec = _rec()
tampered = replace(rec, original_payload={**rec.original_payload, "qty_delta": "-1"})
assert tampered.sealed().audit_hash != rec.audit_hash # edit is detectable
def test_redrive_revalidates_and_can_succeed():
seen = {}
def reingest(p): # stands in for the full validation + keyed commit
seen["validated"] = p
return True
assert redrive(_rec(), reingest) is True
assert seen["validated"]["ndc11"] == "00093005801"
def test_failed_redrive_requarantines_with_incremented_count():
def reingest(_): return False
out = redrive(_rec(redrives=0), reingest)
assert isinstance(out, QuarantineRecord)
assert out.redrive_count == 1
def test_poison_message_is_isolated_at_cap():
out = redrive(_rec(redrives=MAX_REDRIVES), reingest=lambda _: True)
assert isinstance(out, QuarantineRecord)
assert out.failure_class is FailureClass.POISON # never re-queued
A representative quarantine log line carries the identity, failure class, redrive count, and seal — no payload values, so a quarantined dispensing event never leaks PHI into the audit store:
WARNING pharmacy.dlq DLQ_QUARANTINED dlq_id=idem:batch:5f1e6c2a:9f2c7d \
class=exhausted redrives=0 audit_hash=3b7e...c012
Gotchas & compliance pitfalls
- Silently discarding a controlled-substance event. The cardinal sin. A DLQ that drops on overflow, or a handler that swallows the exception without writing a record, turns a transient failure into a permanent DEA discrepancy. The write to quarantine must be as durable as the ledger write it failed to make.
- Replaying without re-validating. A redrive that commits the stored payload directly bypasses the gate that quarantined it. Always route redrive through full validation and the idempotency-keyed commit, so a still-bad record cannot slip through and a good one cannot double-post.
- Infinite redrive loops. Without a
redrive_countcap, a poison message — one that fails deterministically forever — re-queues endlessly, burning workers and hiding the real failure. Cap redrives, then isolate for human root-cause. - Losing failure context. Storing only the payload, or only an error string, leaves an operator unable to tell a fixable validation error from a “just retry it” transient one. Capture the ordered attempt history and the failure class both.
- PHI in the quarantine. The original payload may legitimately need to be held for reconstruction, but it belongs in an encrypted, access-controlled quarantine store — never echoed into the searchable audit log. Log the id, class, and hash; hold the body separately.
- Unsealed quarantine records. If the DLQ entry is mutable and unhashed, an edit to a quarantined Schedule II event is undetectable. Seal each record with SHA-256 so the quarantine inherits the same tamper-evidence as the ledger.
Frequently Asked Questions
Why not just log the failure and move on?
Because a log line is not a recoverable record. A dropped Schedule II movement is an unreconciled discrepancy under 21 CFR § 1304.11, and a plain log entry is neither durable, replayable, nor tamper-evident. The DLQ preserves the whole event so a pharmacist can reconcile or redrive it, and seals it so the preservation itself is auditable.
How is a poison message different from an exhausted transient failure?
An exhausted transient failure might succeed on redrive once the downstream recovers — it is worth replaying. A poison message fails deterministically no matter how often it is replayed, so replaying it only wastes resources and masks the root cause. The redrive_count cap converts a repeatedly-failing event into an isolated poison record that demands human diagnosis.
Should redrive re-run validation, or trust the corrected payload?
Always re-run validation. A corrected payload is still an untrusted assertion until it passes the same gate every inbound event passes. Skipping validation on redrive is exactly how a malformed record slips into the ledger through the back door.
Where does the DLQ record’s audit hash fit the broader chain?
The quarantine record seals itself with the same SHA-256 discipline the ledger uses, so a failed event stays inside one continuous tamper-evident boundary rather than falling outside it. That boundary — what is inside scope and what its integrity guarantees are — is defined by the audit-boundary reference.
Related
- Error Handling & Retry Mechanisms — parent subsystem that routes exhausted and hard-failed events here.
- Async Retry with Exponential Back-off — the retry loop whose exhaustion feeds this quarantine.
- Audit Boundary Definition & Scope — the tamper-evident boundary the quarantine records extend.
- JSON Schema Validation for Drug Records — the validation gate a redrive must re-run before re-commit.