Every adapter in the automated record ingestion and sync workflows pipeline shares one failure boundary: the quarantine queue. When the CSV reader hits an unmapped rights value, when the XML harvester meets a record missing a required LIDO element, or when a database write times out mid-batch, none of those rows are dropped and none abort the run. They are written, whole, to a durable quarantine store — a dead-letter table that captures the raw payload, the structured reasons it failed, and enough context to reproduce the failure. This page is about what happens after a row lands there: how you read the queue, triage it by the class of error, report on it daily, alert when it grows faster than curators can clear it, and safely push fixed records back into the pipeline.

The quarantine queue is where ingestion reliability is actually proven. A pipeline that validates strictly but has no discipline around its rejects is not safer than one that writes garbage — it has simply moved the garbage into a table nobody reads. A well-run queue, by contrast, is a work list: each entry is a specific, reproducible defect with a known owner and a known remedy. The difference is operational, not architectural, and this guide specifies the operational side end to end.

Workflow Context

The three flat-file and network adapters upstream all converge on the same table. The CSV to database sync strategies adapter routes rows that fail its Pydantic contract or carry an unresolvable rights string. The XML and OAI-PMH harvest adapters route records that fail structural validation or arrive with a deleted-status header. The async API poller routes payloads that fail schema validation with Pydantic or that a transient 503 left half-fetched. Because every adapter writes the same record shape, one triage tool, one reconciliation report, and one requeue path serve all of them.

That convergence is deliberate. If each adapter invented its own reject log, a collections manager would have three inboxes with three formats and no single count of “how many records did not make it in last night.” Standardizing on one quarantine record — the same columns, the same error_class taxonomy, the same correlation_id threading back to the source event — turns three pipelines’ worth of failures into one queue that one person can clear. The queue is transport-agnostic on purpose, exactly as the ingestion coordinator that feeds it is.

Prerequisites

Before operating the queue, confirm the pieces that make it durable and diagnosable are in place:

  • A durable quarantine store, not an in-memory list or a log line. A PostgreSQL quarantine table (or an equivalent append-only store) that survives a worker restart, so a crash mid-batch never loses the rejects the batch had already produced.
  • Python 3.9+ for PEP 604 str | None unions and match statements used in the triage classifier below.
  • pydantic (v2) for the quarantine record model itself — the queue record is validated on write with the same rigor as the object records it holds, using the v2 API (model_config, field_validator, model_dump).
  • asyncpg (0.29+) for reading and updating quarantine rows off the async event loop the adapters already run on.
  • Structured error lists on every entry: the exact output of Pydantic’s exc.errors(include_url=False), preserved as JSON, never flattened to a string. The loc and type of each error are what let you classify without re-parsing prose.
  • A correlation_id per source event, minted at ingress and carried through validation to quarantine, so a quarantined row can be traced back to the file, offset, or OAI record it came from.
  • A rights normalization map (the RightsStatements.org lookup the adapters share) so the triage step can tell an unmapped-rights failure apart from a schema failure.

Schema & Field Reference

A quarantine record is a first-class object, not a stringified error. Modeling it strictly means the reconciliation report, the alerting query, and the requeue tool all read the same well-typed fields instead of grepping log text. Each adapter constructs one of these on rejection and persists it durably before moving to the next row.

Field Type Constraint Purpose
payload dict required, verbatim The raw row/record exactly as received, so a fix is reproducible
errors list[dict] required, non-empty Structured Pydantic/parse errors (loc, type, msg) driving classification
error_class str enum: schema | rights | transient Triage bucket; decides fix path vs escalation
source str enum: csv | xml | oai | api Which adapter produced the reject, for per-source reconciliation
correlation_id str required Traces the record back to its source event/offset
quarantined_at datetime required, UTC Drives queue-age alerting and daily report windows
status str enum: open | escalated | requeued | resolved Lifecycle state; only requeued rows re-enter the pipeline
python
from datetime import datetime, timezone
from typing import Literal

from pydantic import BaseModel, ConfigDict, field_validator

ErrorClass = Literal["schema", "rights", "transient"]
Source = Literal["csv", "xml", "oai", "api"]
Status = Literal["open", "escalated", "requeued", "resolved"]


class QuarantineRecord(BaseModel):
    """One durably-stored rejection from any ingestion adapter."""
    model_config = ConfigDict(extra="forbid", validate_assignment=True)

    payload: dict[str, object]
    errors: list[dict[str, object]]
    error_class: ErrorClass
    source: Source
    correlation_id: str
    quarantined_at: datetime
    status: Status = "open"

    @field_validator("errors")
    @classmethod
    def errors_not_empty(cls, value: list[dict[str, object]]) -> list[dict[str, object]]:
        # A quarantine row with no recorded reason is undiagnosable —
        # reject it at write time rather than store a mystery.
        if not value:
            raise ValueError("quarantine record must carry at least one error")
        return value

    @field_validator("quarantined_at")
    @classmethod
    def must_be_utc(cls, value: datetime) -> datetime:
        # Naive timestamps make queue-age math wrong across DST and hosts.
        if value.tzinfo is None:
            raise ValueError("quarantined_at must be timezone-aware (UTC)")
        return value.astimezone(timezone.utc)

With validate_assignment=True, advancing status from open to requeued re-runs validation, so an out-of-enum status can never be written by a careless update. The record is as strict as the object contract it failed — the queue does not get a laxer standard than the pipeline.

Step-by-Step Implementation

1. Classify each rejection into schema, rights, or transient

Triage is only tractable if every entry carries a bucket. Rather than let a human read every error list, derive error_class deterministically from the structured errors and the source. The three buckets map to three fundamentally different remedies: a schema defect needs a mapping or data fix, a rights defect needs a curator decision, and a transient defect needs only a retry.

python
TRANSIENT_TYPES = {"timeout", "connection_error", "http_5xx"}


def classify(errors: list[dict[str, object]]) -> str:
    """Bucket a rejection by inspecting its structured error list."""
    fields = {str(e.get("loc", ("",))[0]) for e in errors}
    types = {str(e.get("type", "")) for e in errors}

    # Transient infrastructure faults win: they are fixed by retry, not edit.
    if types & TRANSIENT_TYPES:
        return "transient"
    # A rights field that failed validation is a curatorial decision.
    if "rights_status" in fields or "rights_uri" in fields:
        return "rights"
    # Everything else — missing fields, bad types, unknown columns — is schema.
    return "schema"

Edge cases. For CSV, an unmapped free-text rights string surfaces as a rights_status value error and buckets to rights; a stray extra column surfaces as an extra_forbidden type and buckets to schema. For XML/OAI, a record missing a mandatory LIDO element buckets to schema, while an OAI 503 with a Retry-After header buckets to transient and should carry the retry deadline in its payload. For API sources, distinguish a 422 (the server rejected our shape — schema) from a 503 (transient) so you never retry an un-retryable request forever.

2. Build a daily reconciliation report

The report answers the collections manager’s one standing question: what did not make it in, and why. Group open entries by source and class, and reconcile the counts against the run’s expected total so unexplained rejects surface as a delta rather than hiding in a table.

python
import asyncpg


async def reconciliation_report(pool: asyncpg.Pool, since: datetime) -> list[dict]:
    """Summarize open quarantine entries by source and error class."""
    rows = await pool.fetch(
        """
        SELECT source, error_class, COUNT(*) AS n,
               MIN(quarantined_at) AS oldest
        FROM quarantine
        WHERE status = 'open' AND quarantined_at >= $1
        GROUP BY source, error_class
        ORDER BY n DESC
        """,
        since,
    )
    return [dict(r) for r in rows]

Edge cases. Report against status = 'open' only, so entries a curator already escalated or requeued do not inflate today’s backlog. Keep the oldest timestamp per group — a small but aging rights bucket is a bigger problem than a large transient bucket that will clear itself on the next retry. Emit the report even when it is empty; a silent report is indistinguishable from a broken cron job.

3. Alert on queue growth, not just queue size

A queue of 40 that has sat at 40 for a week is a known backlog; a queue that jumped from 5 to 400 overnight is an incident, usually a source-schema change that broke one adapter. Alert on the rate and on the oldest open age, not on an absolute threshold that pages you for a healthy backlog.

python
async def alert_signals(pool: asyncpg.Pool) -> dict[str, int]:
    """Return the two numbers worth paging on: recent inflow and stalest age."""
    row = await pool.fetchrow(
        """
        SELECT
          COUNT(*) FILTER (
            WHERE quarantined_at >= now() - interval '1 hour'
          ) AS last_hour,
          COALESCE(
            EXTRACT(EPOCH FROM now() - MIN(quarantined_at))::int, 0
          ) AS oldest_open_secs
        FROM quarantine
        WHERE status = 'open'
        """
    )
    return {"last_hour": row["last_hour"], "oldest_open_secs": row["oldest_open_secs"]}

Edge cases. Threshold last_hour against a per-source baseline: a nightly CSV sync legitimately produces a burst of schema rejects when a new export format arrives, whereas the API poller should almost never spike. A rising transient count is an infrastructure alert (page the engineer); a rising schema or rights count is a data alert (notify the curator). Route the two classes to two channels so the right person answers.

4. Requeue records after a fix

Reprocessing is the payoff. Once a curator assigns a rights URI, or an engineer corrects a column mapping, the fixed payload must re-enter the pipeline through the same validation gate — never a side door that skips validation. Because the upstream commit is idempotent, replaying a record that partially succeeded is harmless.

python
async def requeue(pool: asyncpg.Pool, entry_id: int, fixed_payload: dict) -> None:
    """Re-submit a repaired payload through the normal ingestion gate."""
    async with pool.acquire() as conn:
        async with conn.transaction():
            # Re-validate through the same adapter contract before re-entry.
            record = QuarantineRecord.model_validate(
                await conn.fetchrow(
                    "SELECT * FROM quarantine WHERE id = $1 FOR UPDATE", entry_id
                )
            )
            # Hand the corrected payload back to the pipeline's ingress,
            # then mark the quarantine row requeued for the audit trail.
            await enqueue_for_ingestion(conn, fixed_payload, record.correlation_id)
            await conn.execute(
                "UPDATE quarantine SET status = 'requeued' WHERE id = $1", entry_id
            )

Edge cases. Requeue inside a transaction with FOR UPDATE so two operators cannot replay the same entry twice. Preserve the original correlation_id on re-entry so the audit trail links the fixed record to its original failure. If the requeued record fails validation again, it lands back in quarantine as a fresh open entry — which is correct, and is exactly why the report tracks age: a record that has round-tripped three times needs escalation, not a fourth blind retry. The full reprocessing mechanics — batching replays, capping retry counts, and the poison-record cutoff — are the subject of the deep dive linked below.

Quarantine record lifecycle — entry, triage by class, fix or escalate, requeue, resolved A record enters the durable quarantine store. Triage reads its error_class and routes it to one of three buckets: schema class (fix field mapping), rights class (unmapped value sent to curator review), or transient class (backoff and retry). Schema and transient records flow to an idempotent requeue. Rights records escalate to a curator, and after the fix they rejoin the requeue via a dashed path. The requeue commits the repaired record to a resolved state. Unmapped-rights rows escalate and never default to public. Quarantine lifecycle — triage by class, fix or escalate, requeue, resolved Quarantine record enters Triage read error_class Schema class fix field mapping Rights class unmapped value Transient class backoff + retry Requeue idempotent write Escalate curator review Resolved committed fixed after fix commit

The lifecycle has exactly one entry (the durable store) and one success exit (resolved). Every path to resolution runs back through the requeue gate, so no repaired record is ever committed without re-validation — the queue cannot become a bypass around the contract that put records in it.

Rights and Access Routing

The rights bucket is the one class the queue must never resolve automatically. When a record lands in quarantine because its rights value did not map to a RightsStatements.org URI, the safe default is not public — it is escalated. An unmapped rights string means “we do not yet know what we are allowed to do with this asset,” and the only correct response is to withhold publication until a curator assigns a resolvable URI. The triage classifier routes these to the rights bucket precisely so they are handled by a person, not a fallback rule.

This is why the escalate path in the diagram is a distinct, curator-owned branch rather than a retry loop. A schema or transient record can be fixed and requeued by an engineer or a scheduler; a rights record embeds a policy decision that only a rights holder’s terms can settle. Once the curator assigns the URI, the record requeues through the same gate, and the resolved URI then governs the record’s access tier and is written into the downstream IIIF manifest rights property so every viewer enforces the same policy. Embargo timing — records that resolve to a valid rights URI but must stay dark until a release date — is a separate dimension handled in implementing embargo workflows, not in the queue. The queue’s only rights responsibility is to make sure an unmapped value blocks publication instead of leaking through it.

Verification and Testing

Before trusting the queue in production, prove the classifier buckets each source’s failures correctly and that a requeue re-validates rather than bypasses. The test below exercises a schema reject, a rights reject, and a transient reject against the deterministic classifier.

python
def test_classifier() -> None:
    # A missing required column is a schema failure.
    schema_err = [{"loc": ("title",), "type": "missing", "msg": "field required"}]
    assert classify(schema_err) == "schema"

    # An unmapped rights value is a curatorial (rights) failure.
    rights_err = [{"loc": ("rights_status",), "type": "value_error",
                   "msg": "not a RightsStatements.org URI"}]
    assert classify(rights_err) == "rights"

    # A gateway timeout is transient and must not be treated as bad data.
    transient_err = [{"loc": ("__source__",), "type": "timeout",
                      "msg": "OAI endpoint 503"}]
    assert classify(transient_err) == "transient"

    # Transient wins when mixed with a schema error: retry before editing.
    mixed = schema_err + transient_err
    assert classify(mixed) == "transient"


if __name__ == "__main__":
    test_classifier()
    print("classifier OK")

Then run the reconciliation report against a seeded quarantine table and confirm the counts, using a CLI dry run that reports without mutating any row:

bash
python -m quarantine.report --since 2026-07-16T00:00:00Z --group-by source,error_class

A healthy run shows the transient bucket trending to zero on its own between reports, the schema bucket clearing as mappings are fixed, and the rights bucket draining only through curator action — never on its own. Any open entry older than the escalation SLA is the report’s job to surface before it ages further.

Where to Go Deeper

  • Reprocessing Dead-Letter Records — the mechanics of replaying quarantined records at scale: batching requeues, capping retry counts per correlation_id, detecting poison records that fail on every attempt, and building the idempotent replay path that lets you re-run a whole night’s rejects without duplicating a single committed record.

FAQ

What is the difference between a quarantine queue and a dead-letter queue?

They are the same concept under two names. “Dead-letter queue” comes from message-broker terminology for messages that could not be delivered or processed; “quarantine” is the collections-domain framing for records held back from the catalog until a defect is cleared. This pipeline uses a durable table rather than a broker DLQ so that entries survive restarts, carry a structured errors list and error_class, and can be reported on and requeued with ordinary SQL.

How do I stop the same broken record from cycling through the queue forever?

Cap replays per correlation_id. A record that has requeued and failed a set number of times is a poison record: the fix that was applied did not work, and a further blind retry wastes cycles. Route it to escalated status for human diagnosis instead of leaving it open to be picked up by the next automated replay. The retry-count and poison-record cutoff logic is specified in the reprocessing dead-letter records deep dive.

Should an unmapped rights value be quarantined or given a safe default?

Quarantined, always. Defaulting an unknown rights value to public risks exposing an in-copyright or restricted asset, and defaulting it to private silently hides material that may be freely usable. Neither guess is acceptable, so the classifier buckets it to rights and escalates it to a curator who assigns a resolvable RightsStatements.org URI before the record can be requeued.

Why alert on queue growth rate instead of a size threshold?

Because absolute size conflates a healthy standing backlog with a live incident. A queue holding forty aging rights entries is a normal curatorial work list; a queue that gains four hundred schema entries in an hour is a source-format break that just took an adapter offline. Alerting on inflow rate and oldest-open age catches the incident while ignoring the backlog, and routing by error_class sends infrastructure spikes to an engineer and data spikes to a curator.

Does requeuing a fixed record risk duplicating it in the catalog?

No, provided the requeue path runs through the same idempotent commit the adapters use — an ON CONFLICT upsert keyed on the object’s accession number or manifest URI. A record that partially committed before it errored is updated in place on replay rather than inserted again, so replaying an entire night’s quarantine is safe even if some of those records had already landed.