Quarantining Records That Fail Validation #

Most pipelines have two outcomes for a record: written or lost. The third — quarantined — is what makes the other two trustworthy. A quarantine store holds what failed, along with the raw input and the reason, so nothing is discarded silently and everything is recoverable once the bug is fixed. This guide builds one that is bounded, queryable and replayable, as part of schema validation with Pydantic in the Data Parsing & Transformation Pipelines section.

Problem Framing #

Without a quarantine, a validation failure has two possible handlings and both are bad. Drop and log loses the data: by the time someone reads the log, the page has changed and a recrawl produces something different. Write anyway poisons the dataset with records whose fields are missing or wrongly typed, indistinguishable from good ones once they are in the sink.

Without the third destination, a failure is a silent delete.

A quarantine gives a third option that preserves the evidence. But a quarantine that nobody can query, or that grows without bound, or that cannot be replayed, is just a slower form of dropping — so the design has to address all three from the start.

There is a fourth property people miss: a quarantine holds the same personal data as the curated store, and inherits the same retention obligations. A quarantine with no expiry is a compliance problem wearing an engineering hat.

Step-by-Step Implementation #

1. Capture enough to fix and replay #

A quarantine without the raw input is a slow delete with extra storage cost.

from dataclasses import dataclass, asdict, field
from datetime import datetime, timezone
import hashlib
import json

@dataclass
class QuarantinedRecord:
    quarantine_id: str
    source_url: str
    fetched_at: str
    extractor_version: str
    schema_version: str
    failure_kind: str                  # "validation" | "coercion" | "encoding" | "shape" | "conflict"
    failures: list[dict]               # structured, per field
    raw: dict                          # the input, exactly as the model received it
    quarantined_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat())
    retention_class: str = "quarantine_default"
    replay_count: int = 0

def make_id(source_url: str, raw: dict) -> str:
    payload = json.dumps(raw, sort_keys=True, default=str).encode("utf-8")
    return hashlib.sha256(source_url.encode("utf-8") + payload).hexdigest()[:32]

The identifier is derived from the URL and the payload, which makes quarantining idempotent: the same failing record re-encountered on a later crawl updates the existing entry rather than creating a duplicate. Without that, a persistent extraction bug produces one quarantine row per crawl per page, and the store grows by the crawl frequency rather than by the number of distinct problems.

2. Structure the failures so they group #

from pydantic import ValidationError

def to_failures(exc: ValidationError) -> list[dict]:
    return [{
        "field": ".".join(str(p) for p in err["loc"]),
        "kind": err["type"],                 # "missing", "int_parsing", "greater_than", ...
        "message": err["msg"],
        "input": str(err.get("input"))[:200],
    } for err in exc.errors()]

def validate_or_quarantine(raw: dict, context: dict, store):
    try:
        return Listing(**raw), None
    except ValidationError as exc:
        failures = to_failures(exc)
        record = QuarantinedRecord(
            quarantine_id=make_id(context["source_url"], raw),
            source_url=context["source_url"],
            fetched_at=context["fetched_at"],
            extractor_version=context["extractor_version"],
            schema_version=context["schema_version"],
            failure_kind="validation",
            failures=failures,
            raw=raw,
            retention_class=context.get("retention_class", "quarantine_default"),
        )
        store.upsert(record)
        metrics.quarantined_total.labels(
            field=failures[0]["field"], kind=failures[0]["kind"]).inc()
        return None, failures

Labelling the metric by field and error kind rather than by message keeps cardinality bounded while preserving the distinction that matters. missing on price is a broken selector and wants an engineer; int_parsing on price is a format change and wants a validator update; greater_than on rating is probably a real outlier and wants a look at the page.

3. Make the store queryable by the questions people ask #

CREATE TABLE quarantine (
    quarantine_id     text PRIMARY KEY,
    source_url        text        NOT NULL,
    host              text        NOT NULL,
    fetched_at        timestamptz NOT NULL,
    quarantined_at    timestamptz NOT NULL,
    extractor_version text        NOT NULL,
    schema_version    text        NOT NULL,
    failure_kind      text        NOT NULL,
    primary_field     text        NOT NULL,     -- denormalised for grouping
    primary_kind      text        NOT NULL,
    failures          jsonb       NOT NULL,
    raw               jsonb       NOT NULL,
    retention_class   text        NOT NULL,
    expires_at        timestamptz NOT NULL,
    replay_count      int         NOT NULL DEFAULT 0
);

CREATE INDEX quarantine_group_idx ON quarantine (host, primary_field, primary_kind);
CREATE INDEX quarantine_expiry_idx ON quarantine (expires_at);

Denormalising the primary field and kind into their own columns is what makes the useful query fast:

SELECT host, primary_field, primary_kind, count(*) AS n,
       min(fetched_at) AS first_seen, max(fetched_at) AS last_seen
FROM   quarantine
WHERE  quarantined_at > now() - interval '7 days'
GROUP  BY 1, 2, 3
ORDER  BY n DESC
LIMIT  20;

That single query turns a pile of failures into a ranked work list, and the first_seen column usually identifies the deployment or the site change that caused each group.

4. Replay as a first-class operation #

def replay(store, model_registry, sink, *, host: str | None = None,
           field: str | None = None, limit: int = 10_000) -> dict:
    """Re-validate quarantined records against the CURRENT model."""
    stats = {"attempted": 0, "recovered": 0, "still_failing": 0}
    for record in store.iter(host=host, primary_field=field, limit=limit):
        stats["attempted"] += 1
        model = model_registry["current"]
        try:
            validated = model(**record.raw)
        except ValidationError:
            store.bump_replay(record.quarantine_id)
            stats["still_failing"] += 1
            continue
        sink.write(validated)
        store.delete(record.quarantine_id)
        stats["recovered"] += 1
    return stats

Replay has to be one command, filterable by the same dimensions the grouping query uses. A quarantine that requires a bespoke script to drain will not be drained, and undrained quarantine is deferred data loss.

Bumping replay_count on a record that fails again is what identifies the permanently broken ones — a record replayed five times without success is not going to be recovered by a sixth attempt, and it should be reviewed or expired rather than reprocessed forever.

5. Expire it like any other store #

RETENTION = {
    "quarantine_default": 30,
    "quarantine_personal_data": 14,     # shorter: it holds what curated would have held
}

def set_expiry(record: QuarantinedRecord) -> datetime:
    days = RETENTION[record.retention_class]
    return datetime.fromisoformat(record.quarantined_at) + timedelta(days=days)

The quarantine is swept by the same scheduled job as every other store, and it appears in the same store list the erasure workflow iterates over. A quarantined record containing personal data is exactly as much personal data as a curated one, and it is the store most often forgotten in a deletion.

Verification & Testing #

The last rule is the one most pipelines would fail today.

def test_quarantine_is_idempotent(store):
    raw = {"title": "t"}                       # missing required price
    ctx = {"source_url": "https://example.com/1", "fetched_at": "2026-06-01T00:00:00Z",
           "extractor_version": "e1", "schema_version": "listing/3"}
    validate_or_quarantine(raw, ctx, store)
    validate_or_quarantine(raw, ctx, store)
    assert store.count() == 1

def test_replay_recovers_after_a_model_fix(store, sink):
    quarantine_a_record_missing_currency(store)
    register_model_with_currency_default()
    stats = replay(store, MODEL_REGISTRY, sink)
    assert stats["recovered"] == 1
    assert store.count() == 0

def test_quarantine_is_swept(store, clock):
    quarantine_a_record(store, retention_class="quarantine_personal_data")
    clock.advance(days=15)
    assert sweep(store) == 1
    assert store.count() == 0

def test_erasure_reaches_the_quarantine(store, erasure):
    quarantine_a_record_for_subject(store, "[email protected]")
    erasure.run("[email protected]")
    assert store.count() == 0

The last test is the one most pipelines would fail today, and it is worth writing before the quarantine is deployed rather than after the first erasure request.

Compliance & Operational Guardrails #

Quarantined records carry a retention class and an expiry, and are swept by the same job as every other store.
The quarantine appears in the store list the erasure workflow iterates over.
Quarantine rate is monitored per field and per kind; a rate of exactly zero means validation is too permissive.
Replay is a single filterable command, not a bespoke script.
Records that fail replay repeatedly are reviewed or expired rather than reprocessed indefinitely.

Common Mistakes #

Quarantining without the raw input. The record is unrecoverable, so the quarantine is a slow delete.
Letting it grow without expiry. It becomes the largest personal-data store nobody is managing.
Alerting on size rather than rate. A steady trickle is healthy; a step change is the signal.

Frequently Asked Questions #

What quarantine rate is healthy? #

Non-zero and stable. A rate of exactly zero across a large crawl usually means the schema is too permissive to be catching anything. Something in the region of a fraction of a percent, steady week to week, indicates validation that is doing real work. What matters is the derivative: a step change means something upstream moved.

Should quarantined records be visible to downstream consumers? #

No, and that is the point of the separation. A consumer querying the curated dataset should see only records that passed validation. Where a consumer genuinely needs to know that some records failed, expose the counts rather than the records — the raw payloads are for the team fixing the extractor.

How does this interact with the coverage metric? #

They catch different failures and both are needed. Coverage catches a selector that stopped matching where the field was optional, so no validation error fires at all. Quarantine catches records whose values are present but wrong. A pipeline monitoring only one of them has a blind spot, as covered in the parsing section.

Who should own the quarantine queue? #

The team that owns the extractor, with a named person on rotation. A quarantine with no owner grows until somebody deletes it wholesale, which is data loss dressed as housekeeping. Half an hour a week is usually enough: run the grouping query, fix the largest group, replay it, and move on. The groups are almost always few and large rather than many and small, which is what makes the time budget realistic.

Should a record be quarantined more than once? #

The idempotent identifier means it occupies one row however many times it is re-encountered, but the replay counter should rise each time a replay fails. A record that has failed five replays is telling you the raw input is genuinely unusable — a truncated response, a challenge page, a document in an unexpected language — and it should be reviewed and expired rather than reprocessed indefinitely. Tracking that counter is what keeps the queue converging rather than accumulating.

Schema Validation With Pydantic — the models whose failures land here.
Validating Scraped Data Against JSON Schema — collecting every error rather than only the first.
Enforcing Retention Windows With Scheduled Jobs — the sweep that keeps this store bounded.