Designing a quarantine table for rejected parcel records

A validation failure has three possible destinations and only one of them is right. Dropping the record loses data the county published. Writing it to the production table with a valid = false flag contaminates every query that forgets the predicate. Logging it and moving on means the record exists only in a text file that rotates away in fourteen days. The fourth option — a quarantine table — keeps the raw payload, the reason it was rejected, and enough context to replay it once the parser or the contract is fixed. This is the durable half of error handling and retry logic: retries handle transient failures, and quarantine handles the permanent ones.

Diagnosis: what a missing quarantine table costs

Three symptoms mean rejected records have nowhere to go.

The parcel count for a jurisdiction is lower than the county’s own published figure and nobody can say which parcels are missing. The ingestion log shows WARNING: skipping malformed geometry 1,847 times with no identifiers attached. And a parser fix ships, but there is no way to apply it retroactively — the records it would have handled are gone, so the fix only helps records that arrive after the deploy.

Four destinations for a rejected record A four-row by three-column matrix of destinations for a rejected parcel record. Rows are dropping the record, flagging it invalid in the production table, logging it, and a quarantine table. Columns are whether the raw payload survives, whether a later parser fix can recover it, and the cost. Only the quarantine table preserves the payload and supports replay. Four destinations for a rejected record raw payload survives a parser fix can replay cost Drop the record no no silent data loss Flag invalid in production partially awkward every query needs a predicate Log and continue as text, for 14 days no unreadable at volume Quarantine table yes, unmodified yes, by reason and version one table, one index set yes partial no
Replay is the column that decides it. Three of the four destinations lose the original bytes within days, and once the bytes are gone a parser fix helps only records that arrive after the deploy — the history stays broken permanently.

The last one is the expensive one. A parser improvement should recover history, and without a quarantine table it cannot.

Step-by-step implementation

1. Design the schema around replay 

CREATE TABLE parcel_quarantine (
    quarantine_id     BIGSERIAL PRIMARY KEY,

    -- Provenance: enough to find the record in the source again.
    jurisdiction_id   TEXT        NOT NULL,
    source_url        TEXT        NOT NULL,
    fetched_at        TIMESTAMPTZ NOT NULL,
    ingestion_run_id  UUID        NOT NULL,

    -- Identity, best-effort. A record can be rejected precisely BECAUSE the
    -- identifier could not be parsed, so this is nullable by design.
    source_parcel_id  TEXT,

    -- The payload exactly as received. Never normalised, never re-serialised:
    -- the point is to be able to re-run a fixed parser over the original bytes.
    raw_payload       JSONB       NOT NULL,
    raw_geometry_wkt  TEXT,

    -- Why it was rejected.
    reason_code       TEXT        NOT NULL,
    reason_detail     TEXT        NOT NULL,
    validator         TEXT        NOT NULL,     -- which check rejected it
    validator_version TEXT        NOT NULL,     -- so a fix can target a version

    -- Lifecycle.
    status            TEXT        NOT NULL DEFAULT 'quarantined',
    replay_attempts   INT         NOT NULL DEFAULT 0,
    last_replay_at    TIMESTAMPTZ,
    resolved_at       TIMESTAMPTZ,
    resolution        TEXT,

    CONSTRAINT quarantine_status_valid CHECK (
        status IN ('quarantined', 'replaying', 'resolved', 'rejected_permanently')
    )
);

CREATE INDEX ON parcel_quarantine (jurisdiction_id, status);
CREATE INDEX ON parcel_quarantine (reason_code, validator_version);
CREATE INDEX ON parcel_quarantine (ingestion_run_id);

Three columns do most of the work. raw_payload must be the bytes as received — the moment you normalise before quarantining, a replay tests the normaliser rather than the source. validator_version lets a parser fix target exactly the records the old version rejected, instead of replaying everything. And status distinguishes “not yet retried” from “retried and still wrong,” which is the difference between a queue and a graveyard.

2. Assign a stable reason code 

from enum import StrEnum


class ReasonCode(StrEnum):
    """Stable codes. Renaming one breaks the dashboards and the replay filters,
    so add rather than rename."""

    GEOMETRY_INVALID       = "geometry_invalid"        # self-intersection, unclosed ring
    GEOMETRY_MISSING       = "geometry_missing"
    GEOMETRY_OUT_OF_BOUNDS = "geometry_out_of_bounds"  # outside the jurisdiction envelope
    CRS_UNKNOWN            = "crs_unknown"
    PARCEL_ID_UNPARSEABLE  = "parcel_id_unparseable"
    PARCEL_ID_DUPLICATE    = "parcel_id_duplicate"
    ZONE_CODE_UNMAPPED     = "zone_code_unmapped"      # not in the taxonomy
    SCHEMA_CONTRACT        = "schema_contract"         # a required field absent
    ENCODING               = "encoding"


# Every code needs a documented disposition, so that a growing quarantine table
# is a work queue rather than an unbounded pile.
DISPOSITION = {
    ReasonCode.GEOMETRY_INVALID:       "repair via ST_MakeValid, then replay",
    ReasonCode.GEOMETRY_MISSING:       "await a corrected county publication",
    ReasonCode.GEOMETRY_OUT_OF_BOUNDS: "check the CRS assumption before repairing",
    ReasonCode.CRS_UNKNOWN:            "add the authority code, then replay",
    ReasonCode.PARCEL_ID_UNPARSEABLE:  "extend the identifier parser, then replay",
    ReasonCode.PARCEL_ID_DUPLICATE:    "decide a precedence rule, then replay both",
    ReasonCode.ZONE_CODE_UNMAPPED:     "extend the taxonomy mapping, then replay",
    ReasonCode.SCHEMA_CONTRACT:        "confirm whether the contract or feed changed",
    ReasonCode.ENCODING:               "fix the decode, then replay",
}
A quarter of quarantined records by reason code Lollipop chart of quarantined parcel records by reason code over one quarter. An unmapped zone code accounts for about 4,100 records, an unparseable parcel identifier about 1,350, an invalid geometry about 880, a schema contract violation about 410, an unknown CRS about 190, an encoding failure about 95 and a missing geometry about 40. A dashed rule marks a thousand records, above which a reason code is a systematic gap rather than a handful of bad rows. A quarter of quarantined records by reason code 0 1000 2000 3000 4000 5000 a thousand — above this it is a systematic gap geometry missing 40 records encoding 95 records CRS unknown 190 records schema contract 410 records geometry invalid 880 records parcel id unparseable 1350 records zone code unmapped 4100 records Quarantined records this quarter
The distribution is the work queue. The top two codes are 82% of the backlog and both are one-afternoon parser fixes with a replay behind them; the long tail is genuinely bad source data and belongs in a conversation with the county.

3. Quarantine inside the same transaction as the write 

def ingest_batch(conn, run_id, records, validator_version):
    """Accepted records and quarantined records commit together. If they commit
    separately, a crash between them leaves a batch that is neither ingested nor
    quarantined — the exact gap the table exists to close."""
    accepted, quarantined = [], []

    for raw in records:
        try:
            accepted.append(parse_and_validate(raw))
        except ValidationFailure as exc:
            quarantined.append(QuarantineRow(
                jurisdiction_id=raw.jurisdiction_id,
                source_url=raw.source_url,
                fetched_at=raw.fetched_at,
                ingestion_run_id=run_id,
                source_parcel_id=exc.best_effort_id,     # may be None
                raw_payload=raw.payload,                 # unmodified
                raw_geometry_wkt=raw.geometry_wkt,
                reason_code=exc.reason_code,
                reason_detail=str(exc)[:2000],
                validator=exc.validator,
                validator_version=validator_version,
            ))

    with conn.transaction():
        upsert_parcels(conn, accepted)
        insert_quarantine(conn, quarantined)
        record_run_summary(conn, run_id,
                           accepted=len(accepted), quarantined=len(quarantined))

    return len(accepted), len(quarantined)

Committing both together is what makes the invariant checkable: for any run, accepted + quarantined equals the number of records the source published. That single equation catches silently dropped records, which is the failure mode quarantine is meant to eliminate.

4. Replay against a fixed validator 

def replay(conn, *, reason_code, from_validator_version, to_validator_version,
           limit=5000, dry_run=True):
    """Re-run the current parser over records an earlier version rejected."""
    rows = conn.execute("""
        SELECT quarantine_id, raw_payload, raw_geometry_wkt, jurisdiction_id
          FROM parcel_quarantine
         WHERE status = 'quarantined'
           AND reason_code = %s
           AND validator_version = %s
         ORDER BY quarantine_id
         LIMIT %s
    """, (reason_code, from_validator_version, limit)).fetchall()

    recovered, still_failing = [], []
    for row in rows:
        try:
            recovered.append((row.quarantine_id, parse_and_validate(row.raw_payload)))
        except ValidationFailure as exc:
            still_failing.append((row.quarantine_id, exc.reason_code))

    if dry_run:
        return ReplayReport(len(rows), len(recovered), len(still_failing))

    with conn.transaction():
        upsert_parcels(conn, [p for _, p in recovered])
        mark_resolved(conn, [qid for qid, _ in recovered],
                      resolution=f"replayed under {to_validator_version}")
        # A record that fails again is re-stamped with the new version, so the next
        # replay of the same reason code does not pick it up a second time.
        restamp(conn, still_failing, validator_version=to_validator_version)

    return ReplayReport(len(rows), len(recovered), len(still_failing))

Always dry-run first. A replay that recovers 4% of a reason code means the fix targeted the wrong thing, and finding that out before writing to production is free.

What a dry-run replay tells you before it writes Stacked bar chart of four dry-run replays, each split into records recovered and records still failing. The taxonomy fix recovers 3,980 of 4,100. The identifier parser fix recovers 1,290 of 1,350. The geometry repair recovers 610 of 880. The CRS fix recovers 24 of 190, which means the fix targeted the wrong cause and should not be applied. What a dry-run replay tells you before it writes 0 1000 2000 3000 4000 5000 3980 taxonomy fix 1290 id parser fix 610 geometry repair CRS fix Records in the replay still failing after replay recovered A low recovery rate is a signal to revise the fix, not to run it anyway.
The fourth bar is why the dry run exists: a 13% recovery rate means the fix addressed something other than the actual cause, and finding that out before writing to production costs nothing.

5. Bound the table

Quarantine is a queue with a disposition per reason code, not an archive. Give it a retention policy: records resolved for more than 90 days can be deleted, and records still quarantined after 180 days should either be marked rejected_permanently with a documented reason or escalated to the county. An unbounded quarantine table that nobody reads is the same failure as a log file, just more expensive.

-- The operational view: what is waiting, and how old is the oldest thing waiting?
CREATE VIEW quarantine_backlog AS
SELECT jurisdiction_id, reason_code, validator_version,
       count(*)                                   AS pending,
       min(fetched_at)                            AS oldest,
       now() - min(fetched_at)                    AS age_of_oldest
  FROM parcel_quarantine
 WHERE status = 'quarantined'
 GROUP BY 1, 2, 3
 ORDER BY pending DESC;

Verification & testing 

def test_accepted_plus_quarantined_equals_source_count(conn):
    records = load_fixture("mixed_batch.json")        # 40 good, 7 malformed
    accepted, quarantined = ingest_batch(conn, run_id, records, "v3")
    assert accepted + quarantined == len(records)


def test_raw_payload_is_stored_unmodified(conn):
    raw = {"PARCEL": "0714221 ", "ZONE": "r-1", "geom": "POLYGON((0 0,1 0,1 1,0 0"}
    ingest_batch(conn, run_id, [as_record(raw)], "v3")
    stored = conn.execute(
        "SELECT raw_payload FROM parcel_quarantine ORDER BY quarantine_id DESC LIMIT 1"
    ).fetchone().raw_payload
    assert stored == raw          # trailing space and lowercase zone preserved


def test_quarantine_and_upsert_share_a_transaction(conn, monkeypatch):
    monkeypatch.setattr("ingest.insert_quarantine", boom)
    with pytest.raises(RuntimeError):
        ingest_batch(conn, run_id, load_fixture("mixed_batch.json"), "v3")
    assert conn.execute("SELECT count(*) FROM parcels").scalar() == 0


def test_replay_dry_run_writes_nothing(conn, quarantined_rows):
    before = conn.execute("SELECT count(*) FROM parcels").scalar()
    report = replay(conn, reason_code="zone_code_unmapped",
                    from_validator_version="v3", to_validator_version="v4")
    assert report.recovered > 0
    assert conn.execute("SELECT count(*) FROM parcels").scalar() == before


def test_replay_restamps_records_that_fail_again(conn, quarantined_rows):
    replay(conn, reason_code="geometry_invalid", from_validator_version="v3",
           to_validator_version="v4", dry_run=False)
    remaining = conn.execute("""SELECT DISTINCT validator_version
                                  FROM parcel_quarantine
                                 WHERE status = 'quarantined'
                                   AND reason_code = 'geometry_invalid'""").fetchall()
    assert [r.validator_version for r in remaining] == ["v4"]


def test_every_reason_code_has_a_disposition():
    assert set(DISPOSITION) == set(ReasonCode)

That last test is the one that keeps the table from becoming a graveyard: a new reason code cannot be added without deciding what happens to the records it catches.

Failure recovery

A quarantine table growing without bound. Read quarantine_backlog by reason code rather than in aggregate. One code with 40,000 rows is a parser gap worth an afternoon; forty codes with a thousand rows each is a contract that has drifted from the feed.

A replay that made things worse. Because replay upserts under the ordinary path, the parcel history carries a new transaction-time row — the previous state is still readable. Correct forward: fix the parser, mark the incorrectly-resolved rows quarantined again, and replay. Never delete the quarantine row to “start clean”; the raw payload is the only copy.

Records that were quarantined but never counted. If a run’s accepted + quarantined does not match the source count, the gap is code that raises an exception outside the ValidationFailure catch — usually a parse error thrown before the record is assembled. Wrap that path and quarantine with reason_code = "encoding" rather than letting it propagate.

Frequently asked questions

Why not a valid boolean on the production table?

Because every query then needs WHERE valid and one of them will forget. A rejected record also has no useful normalised form — that is precisely why it was rejected — so it would occupy a row with null geometry, an unmapped zone code and an unparseable identifier, all of which break the constraints the production table depends on. Separate table, separate schema, separate lifecycle.

Should the raw payload be stored as JSONB or as bytes?

JSONB when the source is JSON and round-trips losslessly for your purposes; a BYTEA column when the source is a shapefile chunk, a fixed-width extract, or anything where encoding is itself a suspect. If a meaningful number of records are quarantined with reason_code = "encoding", the payload column is the wrong type — JSONB requires a successful decode before storage, so the very records you most need are the ones that cannot be stored.

How does this relate to a dead-letter queue?

A dead-letter queue is the message-broker version of the same idea, and it is the right tool when the consumer is a stream. It is a poor fit here for two reasons: DLQ retention is typically days rather than months, and a replay wants a SQL predicate over reason code and validator version rather than a re-drive of everything. If a broker is already in the path, let the DLQ handle transport failures and let the quarantine table handle validation failures.

Does a quarantined record count as missing data downstream?

Yes, and it should be reported that way. A jurisdiction with 1,800 quarantined parcels has incomplete coverage, and the API should say so rather than serving a silently short answer — surface the pending count per jurisdiction alongside the parcel count so a consumer can judge completeness for themselves.