Replaying Dead-Letter Queue Records in Clinical ETL

A vendor outage has been fixed and 4,000 subject records are sitting in your dead-letter queue, waiting to be re-driven — but replaying them naively risks two regulatory-grade defects at once: writing duplicate clinical observations, or breaking the audit chain that ties every record back to its original failure. That is the exact problem this page resolves. Clinical data managers, biotech developers, and Python ETL engineers all hit it the moment a real recovery is needed, because a queue that captures failures is only half a solution — the replay is where duplicates, phantom records, and orphaned audit trails are actually created. This deep-dive sits under Dead-Letter Queues and Error Recovery in Clinical ETL, part of the broader Automated EDC Ingestion & Sync Pipelines discipline, and it treats replay as a controlled, idempotent, fully-audited operation rather than a “just re-run it” button — the only way to re-drive Electronic Data Capture (EDC) records that stays defensible under 21 CFR Part 11.

Replay Flow #

A replay never blindly re-sends the stored bytes: it selects a bounded batch by error_class, re-validates each record against the current schema, upserts idempotently on the natural key, and marks the original failure resolved with a link back to it.

Root Cause: Why a Replay Is Riskier Than the Original Ingest #

The original ingest ran once against a known-good schema, in a single execution, with the source system as the authority. A replay violates all three of those assumptions, and each violation is a distinct hazard.

First, time has passed. The record failed against the schema that existed when it entered the queue, but the target SDTM domains, controlled terminology, and cross-form rules may have moved on since. A replay that re-uses the stored bytes without re-checking them against today’s schema can write a record that no longer conforms — a defect that is invisible until analysis. Second, the original write may have partially succeeded. A record can reach the DLQ because a warehouse write timed out after the row was committed; a blind replay then inserts a second copy, creating the duplicate clinical observation that the whole error-recovery design exists to prevent. Third, the audit trail must stay continuous. If a replay writes a fresh record with no link back to the original failure, an inspector sees a record that appears from nowhere with no explanation of the outage it survived. Deterministic replay therefore rests on three guarantees mirroring the ingest primitives in Building Retry Logic for EDC API Timeouts: every write is idempotent on a natural key, every record is re-validated against the current schema, and every replay is bounded and linked to the failure it resolves.

Step-by-Step Implementation #

1. Select records by error_class under a replay cap #

Never replay the whole queue at once. Select a bounded batch filtered to the error_class you actually fixed — replaying transient_infra records after a vendor recovery is safe, but sweeping up poison or unresolved schema_invalid rows in the same pass just re-fails them and burns the replay budget.

# 21 CFR Part 11 relevance: the selection filter is the audit-traceable scope of a
# replay run — which records were re-driven, and why, is reconstructable from it.
import sqlalchemy as sa


def select_replayable(conn, table: sa.Table, error_class: str,
                      replay_cap: int, batch_size: int = 500) -> list[dict]:
    """Pick only open records of one class that are still under the replay cap."""
    stmt = (
        sa.select(table)
        .where(table.c.status == "open")
        .where(table.c.error_class == error_class)   # only what was actually fixed
        .where(table.c.replay_count < replay_cap)    # never re-drive a poison loop
        .order_by(table.c.first_seen)                # oldest failures first
        .limit(batch_size)                           # bounded blast radius
        .with_for_update(skip_locked=True)           # safe under concurrent workers
    )
    return [dict(row) for row in conn.execute(stmt).mappings()]

2. Re-validate each record against the current schema #

Before a single byte is written, run the stored payload back through the current validation model — not the schema captured when it failed. A record that was invalid at capture may be valid now (the mapping was fixed), and a record that was merely stuck on infrastructure must still prove it conforms today.

# ALCOA+ requirement: Accurate — a replay re-checks conformance against the schema
# in force NOW, so recovery can never quietly admit a non-conforming record.
import json
from pydantic import ValidationError
from mypipeline.schema import CurrentSubjectRecord   # the live, versioned model


def revalidate(raw: bytes) -> tuple[bool, dict | str]:
    """Validate stored bytes against the current schema; return (ok, parsed|error)."""
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError as exc:
        return False, f"unparseable on replay: {exc}"   # still poison -> stays parked
    try:
        model = CurrentSubjectRecord.model_validate(payload)  # today's rules
    except ValidationError as exc:
        return False, exc.json()                         # still invalid -> re-quarantine
    return True, model.model_dump(mode="json")

3. Upsert idempotently on the natural key #

Write with an upsert keyed on the record’s natural clinical identity — the tuple that uniquely names the observation regardless of how many times it is replayed. This is what makes a re-drive after a partially-succeeded original write safe: a second replay updates the existing row instead of inserting a duplicate.

# ALCOA+ requirement: Original + Consistent — the natural key collapses any replay
# of one observation to a single warehouse row, so a re-drive cannot duplicate it.
from sqlalchemy.dialects.postgresql import insert


NATURAL_KEY = ["studyid", "usubjid", "domain", "visitnum", "testcd"]


def upsert_observation(conn, obs_table: sa.Table, record: dict, run_id: str) -> None:
    """Insert-or-update on the clinical natural key — never a blind INSERT."""
    stmt = insert(obs_table).values(**record, ingest_run_id=run_id)
    stmt = stmt.on_conflict_do_update(
        index_elements=NATURAL_KEY,        # the clinical identity of the observation
        set_={                             # replay refreshes, never re-inserts
            "value": stmt.excluded.value,
            "ingest_run_id": run_id,
            "superseded_at": sa.func.now(),
        },
    )
    conn.execute(stmt)

4. Mark the failure resolved with a link back and an audit entry #

Closing the loop is a regulatory step, not housekeeping. Update the DLQ row to resolved, stamp the replay run_id and a pointer to the warehouse row it produced, and emit an audit event that ties the resolution to the original failure so the chain from outage to recovery is continuous.

# 21 CFR Part 11 §11.10(e): the resolution is a durable, attributable, time-stamped
# event that links the recovered record back to the exact failure it resolves.
import structlog

log = structlog.get_logger("clinical_etl.replay")


def mark_resolved(conn, dlq: sa.Table, dlq_id: int, *, payload_sha256: str,
                  natural_key: dict, replay_run_id: str) -> None:
    conn.execute(
        sa.update(dlq).where(dlq.c.id == dlq_id).values(
            status="resolved",
            replay_count=dlq.c.replay_count + 1,   # counts toward the cap
            resolved_run_id=replay_run_id,
            resolved_at=sa.func.now(),
        )
    )
    log.info("dlq_resolved", dlq_id=dlq_id, payload_sha256=payload_sha256,
             natural_key=natural_key, replay_run_id=replay_run_id)

5. Cap replays to prevent poison loops #

The replay_count increment in step 4, combined with the replay_count < cap filter in step 1, forms a closed guardrail: a record that keeps failing is re-driven at most cap times before it falls out of every selection and must be released by a human. This is the mechanism that stops one stubborn record from cycling through the pipeline forever.

# ALCOA+ requirement: Available — a record that exhausts its replay budget is parked
# for engineering, so a single poison payload cannot starve the recovery pipeline.
def park_if_exhausted(conn, dlq: sa.Table, dlq_id: int, replay_cap: int) -> bool:
    row = conn.execute(sa.select(dlq.c.replay_count)
                       .where(dlq.c.id == dlq_id)).scalar_one()
    if row >= replay_cap:
        conn.execute(sa.update(dlq).where(dlq.c.id == dlq_id)
                     .values(status="needs_engineering"))
        log.error("replay_cap_reached", dlq_id=dlq_id, replay_count=row)
        return True                         # terminal: only a human releases it
    return False

Verification and Audit Trail #

A replay is GxP-relevant software, so success must be demonstrable from the record, not asserted. Confirm two properties against a fixture queue: replaying the same record twice produces exactly one warehouse row (idempotency held), and every resolved DLQ row carries an unbroken link back to its original failure (audit continuity held). Capture, per replayed record, a structured event routed to a write-once store.

To prove idempotency concretely, replay a record whose original write had partially succeeded and assert the warehouse row count is unchanged; the upsert must have updated in place. Records that fail re-validation on replay route back into the quarantine and become trackable discrepancies through Automated Clinical Query Generation, while the schema they are checked against reconciles with CDISC ODM vs CDASH Schema Mapping. The end-to-end design these steps close out is documented in Dead-Letter Queues and Error Recovery in Clinical ETL.

Edge Cases and Vendor-Specific Gotchas #

The original write half-succeeded. The most dangerous replay is the one that follows a warehouse write which committed the row and then timed out. Never assume a dead-lettered record means “nothing was written.” The upsert on the natural key is what makes this safe — but only if the natural key genuinely uniquely identifies the observation. Validate that your key includes visitnum and testcd (or the domain-appropriate discriminators) so two legitimately different measurements are never collapsed into one.

The schema changed while the record waited. A record dead-lettered before a controlled-terminology update can pass its old validation but fail the current one — or vice versa. Always re-validate against the live model at replay time. If the record is now valid because the mapping was fixed, that is the intended recovery; if it is now invalid because terminology tightened, re-quarantining it for CDM review is the correct, auditable outcome, not a silent write.

Concurrent replay workers. If two operators trigger a replay simultaneously, both can select the same DLQ rows and double-write. Guard the selection with SELECT ... FOR UPDATE SKIP LOCKED so each row is claimed by exactly one worker, and rely on the idempotent upsert as the second line of defense. Never run an unlocked bulk replay from an ad-hoc script against production.

Frequently Asked Questions #

Related #

• Dead-Letter Queues and Error Recovery in Clinical ETL — the parent design that captures the failures this page replays.
• Building Retry Logic for EDC API Timeouts — the bounded retry layer that populates the queue in the first place.
• Automated Clinical Query Generation — where records that fail re-validation on replay become trackable discrepancies.
• CDISC ODM vs CDASH Schema Mapping — the schema definition a replay re-validates each record against.
• Automated EDC Ingestion & Sync Pipelines — the parent reference for this discipline.