Building a Dead-Letter Queue for Invalid MRO Records

When a parsed maintenance record fails validation, dropping it is not an option: the original entry still carries legal weight and must survive for forensic review. This page shows how to build an idempotent dead-letter queue (DLQ) that quarantines schema-invalid records with full error context and a controlled re-drive path, extending the quarantine mechanics introduced in Schema Validation & Error Handling. The design target is zero silent loss: every rejected payload is preserved verbatim, fingerprinted, and replayable.

Prerequisites & inputs

The DLQ sits directly behind the validation gate. It consumes three artifacts for every failed record: the raw payload exactly as extracted (never a coerced or partially normalized copy), the structured validation error, and the source-document metadata (page hash, scan job ID, ingest timestamp). A concrete failing example from a narrowbody line check on N512DL might arrive as a JSON blob with an ata_chapter of 240 — outside the valid 21–100 band — and a part_number field the OCR layer split across two tokens.

The queue itself needs three guarantees before you write a line of handler code. First, durability: entries persist to append-only storage, not an in-memory buffer, so a worker crash cannot erase quarantined work. Second, idempotency: replaying the same failed record must not create a second DLQ entry or, after remediation, a duplicate ledger row. Third, traceability: each entry carries an immutable key derived from the source content, so an auditor can reconstruct exactly which physical logbook page produced which rejection. These map directly onto the retention obligations detailed in FAA Part-145 Recordkeeping Standards.

Quarantine flow

Dead-letter quarantine flowParsedrecordValidateDead-letterqueueRemediate+ re-drive

Step-by-step implementation

  1. Derive a stable quarantine key. Hash the raw payload bytes plus the source-page hash with SHA-256. This key is the deduplication anchor: identical content produces an identical key regardless of how many times the pipeline retries.
  2. Capture, do not mutate. Store the payload as the original UTF-8 string. Any normalization at this stage would destroy the evidentiary value the regulation demands.
  3. Attach classified error context. Serialize the Pydantic ValidationError into structured error_locations so a remediation engineer sees exactly which field failed and why.
  4. Write idempotently. Use the quarantine key as the primary key with an upsert that increments a seen_count instead of inserting a duplicate row.
  5. Expose a guarded re-drive. A record leaves the DLQ only after re-validation succeeds; the re-drive path checks whether the derived ledger key already exists before committing, so a replayed record can never double-write.
from __future__ import annotations

import hashlib
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("mro.dead_letter_queue")


class QuarantinedRecord(BaseModel):
    """Immutable DLQ entry. Preserves the raw payload verbatim for forensic review.

    Retention aligns with FAA 14 CFR § 145.219(c): records must be kept for at
    least two years after the work is performed and the aircraft is returned to service.
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    quarantine_key: str = Field(pattern=r"^[0-9a-f]{64}$")
    raw_payload: str
    source_page_hash: str
    error_locations: list[str]
    quarantined_at: datetime
    seen_count: int = Field(default=1, ge=1)

    @field_validator("quarantined_at")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        return v.astimezone(timezone.utc)


class MRORecord(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")

    aircraft_registration: str = Field(pattern=r"^[A-Z0-9\-]{3,10}$")
    ata_chapter: int = Field(ge=21, le=100)
    part_number: str = Field(min_length=3, max_length=32)
    work_order_id: str


@dataclass
class DeadLetterQueue:
    """Append-only, idempotent quarantine store with a guarded re-drive path."""

    _store: dict[str, QuarantinedRecord] = field(default_factory=dict)
    _ledger_keys: set[str] = field(default_factory=set)

    @staticmethod
    def _derive_key(raw_payload: str, source_page_hash: str) -> str:
        digest = hashlib.sha256()
        digest.update(raw_payload.encode("utf-8"))
        digest.update(source_page_hash.encode("utf-8"))
        return digest.hexdigest()

    def quarantine(self, raw_payload: str, source_page_hash: str, error: ValidationError) -> str:
        key = self._derive_key(raw_payload, source_page_hash)
        existing = self._store.get(key)
        if existing is not None:
            bumped = existing.model_copy(update={"seen_count": existing.seen_count + 1})
            self._store[key] = bumped
            logger.info("DLQ dedupe key=%s seen_count=%d", key[:12], bumped.seen_count)
            return key

        locations = [".".join(str(p) for p in err["loc"]) for err in error.errors()]
        entry = QuarantinedRecord(
            quarantine_key=key,
            raw_payload=raw_payload,
            source_page_hash=source_page_hash,
            error_locations=locations,
            quarantined_at=datetime.now(timezone.utc),
        )
        self._store[key] = entry
        logger.warning("DLQ quarantine key=%s fields=%s", key[:12], locations)
        return key

    def redrive(self, key: str, remediated_payload: dict[str, Any]) -> str:
        """Re-validate and commit exactly once; never double-write the ledger."""
        record = MRORecord.model_validate(remediated_payload)
        ledger_key = f"{record.work_order_id}:{record.aircraft_registration}"
        if ledger_key in self._ledger_keys:
            logger.info("Re-drive skipped, ledger_key=%s already committed", ledger_key)
            self._store.pop(key, None)
            return ledger_key
        self._ledger_keys.add(ledger_key)
        self._store.pop(key, None)
        logger.info("Re-drive committed ledger_key=%s", ledger_key)
        return ledger_key

Worked example

Feed the handler the failing N512DL electrical-power record. The extractor emitted {"aircraft_registration": "N512DL", "ata_chapter": 240, "part_number": "BAC27DFR12", "work_order_id": "WO-8841"}. Validation rejects ata_chapter because 240 violates the le=100 bound (ATA 24 is electrical power; the OCR layer appended a stray digit). The DLQ derives a key such as 9f3c…a17b, stores the raw string untouched, and records error_locations=["ata_chapter"]. A second identical retry does not create a new row — it bumps seen_count to 2, which is the log line an on-call engineer wants to see.

An engineer corrects the chapter to 24 and calls redrive. Re-validation passes, the ledger key WO-8841:N512DL is committed, and the DLQ row is removed. Replaying the same remediation a second time finds the ledger key already present and short-circuits — the record is never written twice. This exactly-once behavior is what lets you re-drive an entire day of quarantined records without fear of duplicating certifying-staff entries. The deeper conformance checks that a re-driven record must still satisfy are covered in Validating Parsed Data Against AMM Standards.

Edge cases & compliance gotchas

  • Mixed OEM encodings. A part number arriving as BAC27DFR12 from one OEM and BAC-27-DFR-12 from another hashes to different quarantine keys even though they denote the same component. Normalize the ledger key, never the quarantine key — the quarantine key must reflect the raw evidence.
  • Unicode diacritics in technician names. A certifying-staff field containing José Nuñez may fail a naive ASCII validator. Reject on a real rule, not on encoding: keep the diacritics in the preserved payload so the audit trail matches the physical signature block.
  • Timezone drift on quarantined_at. A worker running in local time can stamp an entry with an offset-naive timestamp; the enforce_utc validator normalizes to ISO 8601 UTC so DLQ ordering stays monotonic across regions.
  • Poison records. A payload that fails re-validation on every attempt must not loop forever. Cap seen_count, and once it exceeds an engineering threshold, escalate to manual review rather than re-driving — a deterministic schema failure will never fix itself.

A dead-letter queue is not an error sink; it is a controlled holding area with legal weight. Preserve the raw bytes, key on content, and gate the re-drive on an exactly-once commit, and you get a quarantine layer that satisfies the retention and non-repudiation obligations MRO auditors expect while keeping the main pipeline flowing at line-check speed.

Related pages