Operational Context

A collections manager traces a nightly sync failure to a single mapping defect — the source system exported date_modified in a locale format the contract rejected, and every affected row landed in the dead-letter table with a schema error. The mapping is now fixed. Thousands of quarantined rows must re-enter the collection management system, but they must re-enter once each and only if they now pass the same contract that rejected them. This page resolves that exact task: draining the ingestion quarantine queue after a defect is corrected, without duplicating records, without re-failing on unrelated transient errors, and without looping forever on a genuinely poison row. It is the replay half of the durable dead-letter guarantee the rest of the automated record ingestion and sync workflows pipeline depends on — the quarantine table is only safe to fail into if there is a disciplined way to fail back out of it.

The naive instinct — re-run the whole failed batch — is precisely the move that corrupts a collection. Blind replay re-attempts records whose defect was never fixed, re-inserts rows a partial earlier run already committed, and burns retry budget on records that will never parse. Reprocessing has to be selective, idempotent, and bounded.

Root Cause Analysis

Unsafe replay fails along three concrete vectors, each with a single root cause rather than a scatter of unrelated bugs.

First, class-blind replay. A dead-letter table accumulates failures of different kinds: a schema mismatch that a code change just fixed, a transient timeout or deadlock that was never a data problem, and a permanently malformed poison record that no fix will rescue. Replaying the table indiscriminately re-fails the transient and permanent classes alongside the schema class you actually corrected, so the queue never drains and the reconciliation report never clears. The remedy is to select records by their recorded error class and replay only the class the fix addresses.

Second, non-idempotent replay. If the original run committed some rows before it died, and the dead-letter capture was not perfectly transactional, a replay that re-inserts on a plain INSERT raises a duplicate-key error — or worse, on a table without a unique constraint, silently writes the record twice. A collection with two accession rows for one object is a data-integrity incident, not a cosmetic one. Every replayed write must key on a stable idempotency key and upsert with ON CONFLICT DO UPDATE, exactly as the CSV-to-database sync commit path already does, so replaying an already-committed row is a harmless no-op.

Third, unbounded retry on poison. A record that fails validation for a reason the fix did not address — a truncated title, a rights value no classifier recognizes — will re-fail every replay. Without an attempt ceiling it cycles through the queue indefinitely, consuming throughput and masking real progress. The record needs to be re-validated through the same strict Pydantic v2 model that first rejected it, and after a bounded number of attempts it must escalate to a human rather than loop. Re-validating through a looser model on replay would defeat the entire contract — a record must never enter the collection through a gate weaker than the one that guards the front door.

Canonical Solution

The solution replays by error class, re-validates each record through the identical contract, upserts idempotently on the idempotency key, increments an attempt counter on each re-failure, and escalates once the attempt budget is spent or the record is already flagged permanent. Every record reaches exactly one of three terminal outcomes — committed, backed off for another pass, or escalated — and each maps to a single deterministic destination.

python
from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Iterable, Iterator

from pydantic import BaseModel, ConfigDict, ValidationError, field_validator

RIGHTS_URI_PREFIX = "https://rightsstatements.org/vocab/"


class ErrorClass(str, Enum):
    SCHEMA = "schema"        # the defect a schema fix targets — replay these first
    TRANSIENT = "transient"  # timeout / deadlock / upstream 503 — data was fine
    PERMANENT = "permanent"  # poison record no fix will rescue — escalate on sight


class Outcome(str, Enum):
    COMMITTED = "committed"  # passed the contract, upserted idempotently
    BACKOFF = "backoff"      # still failing, budget remains — re-queue
    ESCALATED = "escalated"  # permanent, or attempt budget exhausted


class CollectionRow(BaseModel):
    """The SAME strict contract the original ingest used, reused verbatim so a
    replayed record faces the identical gate — never a looser one."""
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    object_id: str
    title: str
    rights_status: str

    @field_validator("object_id", "title")
    @classmethod
    def not_blank(cls, value: str) -> str:
        if not value:
            raise ValueError("required field is blank")
        return value

    @field_validator("rights_status")
    @classmethod
    def must_be_rights_uri(cls, value: str) -> str:
        # A replayed record still has to carry a resolvable RightsStatements.org URI.
        if not value.startswith(RIGHTS_URI_PREFIX):
            raise ValueError("rights_status must be a RightsStatements.org URI")
        return value


@dataclass
class DeadLetter:
    idempotency_key: str              # stable per source row — dedupes every replay
    payload: dict                     # the raw row exactly as first rejected
    error_class: ErrorClass
    attempts: int = 0                 # incremented on each re-failure, never reset


@dataclass(frozen=True)
class ReplayPolicy:
    max_attempts: int = 5             # escalate a still-failing record after N tries


def reprocess(
    dl: DeadLetter,
    committed_keys: set[str],
    policy: ReplayPolicy,
) -> Outcome:
    # 1. Idempotency guard. If this key already committed on an earlier pass, the
    #    replay is a no-op — the same source row is never written to the CMS twice.
    if dl.idempotency_key in committed_keys:
        return Outcome.COMMITTED

    # 2. A permanently-classified poison record is escalated on sight. No amount of
    #    replay will make it parse, so looping on it only starves the real backlog.
    if dl.error_class is ErrorClass.PERMANENT:
        return Outcome.ESCALATED

    # 3. Re-validate through the identical model. The schema fix that motivated this
    #    replay lives inside the contract, so a formerly-SCHEMA row now either passes
    #    or fails for a genuinely new, still-unresolved reason.
    try:
        CollectionRow.model_validate(dl.payload)
    except ValidationError:
        dl.attempts += 1
        if dl.attempts >= policy.max_attempts:
            return Outcome.ESCALATED  # budget spent → treat as poison, hand to a human
        return Outcome.BACKOFF        # transient/unresolved → re-queue for a later pass

    # 4. Valid now. The caller upserts on the idempotency key (ON CONFLICT DO UPDATE),
    #    so recording the key here makes a duplicate replay in the same run inert too.
    committed_keys.add(dl.idempotency_key)
    return Outcome.COMMITTED


def destination(outcome: Outcome) -> str:
    # A match keeps routing exhaustive: adding an Outcome without a branch is a visible
    # omission rather than a silent fall-through into the collection.
    match outcome:
        case Outcome.COMMITTED:
            return "upsert:collection_records"
        case Outcome.BACKOFF:
            return "requeue:dead_letter"
        case Outcome.ESCALATED:
            return "queue:curator-review"


def replay_by_class(
    dead_letters: Iterable[DeadLetter],
    committed_keys: set[str],
    target: ErrorClass | None = None,
    policy: ReplayPolicy = ReplayPolicy(),
) -> Iterator[tuple[str, Outcome, str]]:
    # Replay only the class the fix addresses (usually SCHEMA after a mapping fix), so
    # an unrelated transient backlog is not dragged into this pass. target=None = all.
    for dl in dead_letters:
        if target is not None and dl.error_class is not target:
            continue
        outcome = reprocess(dl, committed_keys, policy)
        yield dl.idempotency_key, outcome, destination(outcome)

Because committed_keys is threaded through the whole pass and the write is an upsert on that same key, replaying a batch is idempotent at two layers: the guard short-circuits a key seen earlier in the run, and ON CONFLICT DO UPDATE absorbs a key that a prior run already persisted. The attempt counter lives on the durable dead-letter record, so a record that fails across several nights accumulates attempts and eventually escalates rather than cycling forever.

Dead-letter reprocessing decision flow: re-validate, then commit, back off, or escalate A dead-letter record selected by error_class enters a re-validation gate. On pass it exits up to an idempotent upsert and commits with no duplicate row. On failure it drops right into a budget gate. If transient with budget remaining it exits up to backoff and is re-queued with jitter; a dashed loop returns it to the dead-letter store for a later pass. If the budget is exhausted or the record is permanent it drops down to escalation for curator review. yes · valid now no · still fails yes · transient no · permanent re-queue Dead-letter record by error_class re-validate passes? budget left? Idempotent upsert commit · no duplicate row Backoff & re-queue transient · retry with jitter Escalate — permanent poison record · curator review

The dashed loop is the crucial safety property: a backed-off record is not lost, it returns to the dead-letter store with its attempt count incremented, so successive passes either eventually succeed as further fixes land or escalate once the budget is spent.

Edge Cases and Variants

  • CSV vs. XML vs. API origin. reprocess is agnostic to where the payload came from — a quarantined csv.DictReader row, an lxml.iterparse record, or a paged response body from the async ingestion pipeline all serialize to the same payload dict. Only the re-hydration of that dict differs; the class/re-validate/upsert loop is identical.
  • Schema-fix vs. transient vs. poison. These are the three error classes, and they demand different replays. A schema fix warrants a targeted replay_by_class(..., target=ErrorClass.SCHEMA) pass immediately after deploy. A transient backlog (timeouts, deadlocks) is better drained on a scheduled retry with backoff, since the data was always valid. A poison record should never be replayed automatically at all — flag it PERMANENT at capture time so it escalates on sight.
  • Batch size and throughput. Replay in bounded chunks against a lazy cursor over the dead-letter table, exactly as the ingest side streams, so peak memory stays flat regardless of backlog size. Cap the per-pass count so a multi-hundred-thousand-row backlog cannot starve the live nightly sync sharing the same database pool.
  • Dry-run diff. Before committing a replay, run it with the upsert disabled and tally the projected outcomes — how many would commit, back off, and escalate. A surprising escalation count means the deployed fix did not actually address the dominant failure, and the pass should be held.
  • Attempt-budget tuning. Set max_attempts against how many independent fixes a record realistically waits on. Too low escalates records a forthcoming fix would have rescued; too high lets genuine poison cycle for weeks. Five passes is a defensible default for a nightly cadence.

Validation

Prove the invariants that matter — a replay commits a fixed record exactly once, a permanent record escalates without wasting a re-validation, and a still-broken record backs off until its budget is spent — with an assert-based test that needs no database:

python
# test_reprocess.py  ->  run with:  pytest -q test_reprocess.py

def _dl(**over) -> DeadLetter:
    base = {
        "idempotency_key": "OBJ.1",
        "payload": {
            "object_id": "OBJ.1",
            "title": "Study of Hands",
            "rights_status": "https://rightsstatements.org/vocab/InC/1.0/",
        },
        "error_class": ErrorClass.SCHEMA,
    }
    base.update(over)
    return DeadLetter(**base)


def test_fixed_schema_record_commits():
    seen: set[str] = set()
    assert reprocess(_dl(), seen, ReplayPolicy()) is Outcome.COMMITTED
    assert "OBJ.1" in seen                              # recorded for idempotency


def test_replay_is_idempotent():
    seen: set[str] = set()
    first = reprocess(_dl(), seen, ReplayPolicy())
    second = reprocess(_dl(), seen, ReplayPolicy())     # same key, replayed
    assert first is second is Outcome.COMMITTED
    assert len(seen) == 1                               # committed exactly once


def test_permanent_record_escalates_without_retry():
    dl = _dl(error_class=ErrorClass.PERMANENT)
    assert reprocess(dl, set(), ReplayPolicy()) is Outcome.ESCALATED
    assert dl.attempts == 0                             # never re-validated


def test_still_broken_record_backs_off_then_escalates():
    dl = _dl(payload={"object_id": "", "title": "x",
                      "rights_status": "https://rightsstatements.org/vocab/InC/1.0/"})
    policy = ReplayPolicy(max_attempts=3)
    outcomes = [reprocess(dl, set(), policy) for _ in range(3)]
    assert outcomes[:2] == [Outcome.BACKOFF, Outcome.BACKOFF]
    assert outcomes[2] is Outcome.ESCALATED            # attempt budget spent
    assert dl.attempts == 3

A green run confirms that a record whose defect is fixed commits and is deduplicated on any subsequent replay, that a poison record escalates immediately instead of consuming a validation attempt, and that a record which keeps failing exhausts its budget and reaches a human rather than looping through the queue forever.

Standards Alignment

Reprocessing does not change what a record means, only when it enters the collection, so the standards contract is identical to first-pass ingest. Each committed row still resolves its rights assertion to a machine-readable RightsStatements.org URI before it is written, and an escalated record stays dark — never defaulted to open — until a curator adjudicates it. The interchange envelope remains LIDO: a replayed record populates the same <lido:recordWrap> and <lido:rightsWork> structures its siblings do, and the idempotency key maps to the record’s stable <lido:lidoRecID> so the upsert targets one canonical row. Because the replay re-runs the identical Pydantic v2 contract, no record can enter through reprocessing that could not have entered through the front door. See the Python enum documentation for the string-enum pattern the error classes and outcomes rely on.

FAQ

Should I replay the entire dead-letter table after a fix?

No. Select by the recorded error class and replay only the class your fix addressed — usually schema after a mapping correction. Replaying indiscriminately re-attempts transient and poison records that your fix never touched, so the queue never drains and the reconciliation report never clears. replay_by_class(..., target=ErrorClass.SCHEMA) scopes the pass to exactly the records the deploy can rescue.

How does replay avoid creating duplicate records?

Two layers. Every replayed write upserts on a stable idempotency key with ON CONFLICT DO UPDATE, so a row an earlier partial run already committed is updated in place rather than inserted again. Within a single pass, a committed_keys guard short-circuits a key already seen, making a duplicate replay in the same run inert. A record therefore reaches the collection exactly once no matter how many times it is replayed.

What stops a genuinely broken record from looping forever?

The attempt counter on the durable dead-letter record. Each re-failed validation increments it, and once it reaches max_attempts the record escalates to curator review instead of backing off again. A record the deployed fix did not actually repair is thus bounded — it cycles a fixed number of passes, then lands in front of a human.

Why re-validate through the same model instead of trusting the fix?

Because the fix might be incomplete, and a record must never enter the collection through a gate weaker than the one guarding first-pass ingest. Re-running the identical strict contract means a replayed record either genuinely passes now or fails for a real, still-unresolved reason — there is no looser path into the collection, which is the whole point of a validation contract.