# Quarantining and Replaying Rejected Cost Rows

A validation gate that only rejects is half a system. The moment a cost row fails the strict schema built in Schema Validation & Error Handling, two obligations open at once: the failing row must be preserved exactly as it arrived so a production accountant can see what was wrong and fix it, and the corrected row must be able to re-enter the ledger later without any risk that a row already posted gets posted a second time. Get the first half wrong and a malformed line vanishes into a log file; get the second half wrong and a well-meaning re-run double-books a vendor invoice a completion guarantor is watching to the dollar. This walkthrough builds both halves as one deterministic loop — write the original row verbatim to a quarantine store alongside a SHA-256 fingerprint and a machine-readable reason code, then replay corrected rows with an idempotent dedup on the hash so a retried batch can never post the same cost twice.

# Prerequisites and Context

This page picks up exactly where validation leaves off: a row has already been rejected by the boundary schema and its ValidationError reason is known. Quarantine and replay are the storage-and-recovery layer around that rejection, not a second validator. The implementation targets Python 3.11+ for standard-library zoneinfo, and uses the same restrained stack as the rest of Cost Ingestion & Data Parsing Workflows: hashlib for the deterministic row fingerprint, zoneinfo for timezone-aware audit stamps, and Pydantic v2 (model_validate, ConfigDict(frozen=True)) for the quarantine record itself. The append-only quarantine and ledger tables are typically fronted by SQLAlchemy 2.0, but nothing here depends on the backing store beyond its being append-only.

Two properties from upstream carry directly into this design. First, the row arrives as an untrusted dictionary and must be stored without mutation — the same forensic-preservation rule the malformed-input handling in Handling Malformed CSVs from Set Accountants enforces, because a quarantine entry an accountant edited in place is no longer evidence of what the vendor actually sent. Second, the quarantine store and the ledger both live under the write-once guarantees formalized in Audit Logging & Immutability: a resolved quarantine entry is superseded by a new record, never overwritten, so the original failure remains reconstructable for a guarantor’s examiner months after the close it delayed.

# Step-by-Step Implementation

The mechanism rests on a single stable identity for a row: a SHA-256 over its canonical serialization. That fingerprint is computed the same way at three moments — when the row is quarantined, when a corrected version is prepared, and when it is posted — so the ledger can carry the set of hashes it has already committed and reject any replay whose hash is already present. Quarantine writes the raw row verbatim; replay recomputes the hash on the corrected row, checks it against the posted set, validates, and posts only if it is genuinely new.

from __future__ import annotations

import hashlib
import json
from datetime import datetime
from typing import Callable
from zoneinfo import ZoneInfo

from pydantic import BaseModel, ConfigDict

LEDGER_TZ = ZoneInfo("America/Los_Angeles")


def row_fingerprint(row: dict) -> str:
    """Stable SHA-256 over a canonical serialization of a raw cost row.

    The exact same serialization rule must be used everywhere a row is
    hashed, or the dedup key is not comparable across quarantine and replay.
    """
    canonical = json.dumps(row, sort_keys=True, separators=(",", ":"), default=str)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


class QuarantineEntry(BaseModel):
    """A rejected cost row preserved verbatim for correction and replay."""

    model_config = ConfigDict(frozen=True)

    row_sha256: str
    raw_row: dict            # the original row, exactly as received — never mutated
    reason_code: str
    batch_id: str
    row_index: int
    quarantined_at: str
    resolved: bool = False


def quarantine_row(
    raw_row: dict, reason_code: str, batch_id: str, row_index: int
) -> QuarantineEntry:
    return QuarantineEntry(
        row_sha256=row_fingerprint(raw_row),
        raw_row=raw_row,
        reason_code=reason_code,
        batch_id=batch_id,
        row_index=row_index,
        quarantined_at=datetime.now(LEDGER_TZ).isoformat(),
    )


def replay_corrected(
    corrected_rows: list[dict],
    validate: Callable[[dict], None],   # raises on an invalid row
    post_to_ledger: Callable[[dict, str], None],
    posted_hashes: set[str],            # hashes already committed to the ledger
) -> dict:
    """Replay corrected rows idempotently. A row whose hash is already in the
    ledger is skipped, so a retried or overlapping batch never double-posts."""
    summary = {"posted": 0, "skipped_duplicate": 0, "still_invalid": 0}

    for row in corrected_rows:
        digest = row_fingerprint(row)

        # Idempotency guard: identical corrected bytes hash identically, so a
        # replay of an already-posted row is a no-op rather than a duplicate.
        if digest in posted_hashes:
            summary["skipped_duplicate"] += 1
            continue

        try:
            validate(row)
        except Exception:
            # Still not valid — leave it quarantined for another correction pass.
            summary["still_invalid"] += 1
            continue

        post_to_ledger(row, digest)
        posted_hashes.add(digest)
        summary["posted"] += 1

    return summary

The design turns replay from a risky bulk re-import into a safe, repeatable operation. posted_hashes is not an in-memory convenience — it is materialized from the hashes the ledger itself recorded on every committed row, so it survives a process restart and reflects the true posted state rather than one run’s memory. A corrected row that reproduces the same canonical bytes as an already-posted row produces the same digest and is skipped; a row still failing validation is counted and left in quarantine for a further correction pass rather than silently dropped. Nothing about the operation assumes it runs once: replaying the same corrected file twice, or replaying a batch that overlaps a previous one, converges to the same ledger state.

The path a rejected row travels from quarantine back to the ledger is a short decision flow with the dedup check as its pivot.

# Audit Trail Requirements

Both stores in this loop are append-only, and every transition through them is logged. A quarantine entry must carry, at minimum: the row_sha256 fingerprint of the original bytes; the raw_row verbatim; the machine-readable reason_code that names why validation failed; the batch_id and row_index that locate the row in its source file; and a timezone-aware quarantined_at stamp pinned to the studio’s booking day via zoneinfo rather than a bare offset. The quarantine record is written before the batch’s clean rows commit, so a crash between rejection and persistence leaves a replayable record of intent rather than a lost row.

Replay is logged with equal rigor. Each replay entry records the corrected row’s hash, the outcome (posted, skipped_duplicate, or still_invalid), a timezone-aware timestamp, and the identity of the operator who authorized the correction — SYS_AUTO is never valid on a replay, because a human decided what the fix was. Resolving a quarantine entry is itself an append: a new record marks the original resolved, and the original is never edited in place, so the failure, the correction, and the eventual post form an unbroken chain. That chain is what an examiner reads to confirm a re-run corrected a genuine error rather than quietly reposting a cost that was already in the ledger — the same immutable-ledger property specified in Audit Logging & Immutability.

# Gotchas and Production Edge Cases

Replay idempotency. The single guarantee this whole design exists to provide is that replaying a corrected batch cannot double-post. That guarantee holds only if posted_hashes reflects the ledger’s committed state, not one run’s memory — materialize it from the hashes the ledger recorded, and add a hash to it in the same transaction that posts the row, so a crash between “post” and “remember” cannot readmit the row on the next replay. Deduping in memory alone means a process restart forgets what it posted and reposts it.

Hash stability. A dedup key is only useful if the same logical row always produces the same hash. Serialize canonically — sorted keys, fixed separators, a deterministic rule for non-string values — everywhere a row is fingerprinted, at quarantine and at replay alike. If one call path stringifies a number as "1250" and another as "1250.0", or one preserves key order and another does not, the same corrected row yields two different hashes and the idempotency guard silently fails. Pin the serialization in one shared function, as row_fingerprint does, and never hash a row any other way.

Partial-batch retries. When a replay of five hundred corrected rows fails at row three hundred, the safe recovery is to replay the whole file again — which is only safe because the first three hundred rows are already in posted_hashes and skip cleanly on the second pass. Never track “where the batch stopped” as a row offset and resume from it; an offset drifts the moment the corrected file is re-sorted or a row is added, whereas the hash-based skip is correct regardless of order or batch boundaries.

Correction changes the hash, on purpose. The quarantine entry stores the hash of the original row; a genuine correction changes the bytes and therefore the hash, which is exactly why a corrected row is not mistaken for its own failed predecessor. Do not try to preserve the original hash across a correction — the value of two distinct hashes is that the audit trail shows a real edit occurred rather than a duplicate slipping through.

Transport failures are not row failures. A truncated download or a 503 from a vendor API is an infrastructure outcome handled at the boundary in CSV & API Sync Pipelines; it must never land in this row-level quarantine, or an operator will waste a correction pass on a row that was never malformed, only never fully delivered.

# Related Guides

• Schema Validation & Error Handling — the validation gate whose rejections feed this quarantine-and-replay loop.
• Handling Malformed CSVs from Set Accountants — the field-data defects that most often send a cost row into quarantine in the first place.
• Audit Logging & Immutability — the write-once ledger discipline that keeps every quarantine, correction, and post reconstructable.
• CSV & API Sync Pipelines — the transport layer that separates delivery failures from the row-level rejections handled here.
• EP/Showbiz Sync Parsing — a concrete source of the raw cost rows this gate accepts, rejects, and later replays.

Up one level: Schema Validation & Error Handling, part of Cost Ingestion & Data Parsing Workflows.