Building Fallback Validation Chains for MongoDB Ingestion

Within the broader Automated Schema Enforcement & Monitoring framework, a fallback validation chain is the pattern that keeps ingestion moving when a document fails the current schema instead of dropping it or halting the pipeline. This page delivers a complete, runnable blueprint: where the chain sits in the write path, the exact MongoDB and driver versions it needs, an idempotent deployment sequence for the tiered collections, a production PyMongo orchestrator with retry and quarantine handling, the precise error fingerprints you will grep for during an incident, and a rollback path back to single-schema enforcement. The reader outcome is a tiered ingestion boundary where compliant documents land in the strict collection at full velocity, transitional and legacy payloads are accepted under relaxed contracts with routing metadata, and unrecoverable documents are quarantined with enough context for asynchronous reconciliation.

A fallback chain is not a licence for schema drift. It is a deterministic state machine that decouples ingestion velocity from schema rigidity: when a document fails primary validation it cascades to progressively permissive tiers, each explicitly versioned and bounded, so write availability survives an upstream contract change without silently corrupting the strict collection. Native collection-level validators give you the per-tier enforcement primitive; the chain is the application-layer orchestration that decides which tier a document belongs in and records why.

Architectural Context & Enforcement Boundaries

A production-grade fallback chain operates as a deterministic sequence of tiers, each backed by its own collection and its own $jsonSchema validator, rather than a single try/catch wrapper. The primary tier enforces current business invariants (required fields, bsonType constraints, enum membership) with validationAction: "error". Secondary tiers relax specific constraints for known transitional payloads or deprecated API contracts. The terminal quarantine tier carries no validator at all: it accepts structurally malformed documents but tags them with validation_tier, schema_version, and failure_reason metadata so a downstream worker can reconcile them.

MongoDB’s native $jsonSchema validator provides the foundation for each tier, but the cascade itself must live in the ingestion service. Relying solely on validationAction: "warn" at the collection level obscures failure context, pollutes production logs, and destroys the audit trail — the write succeeds but you lose the signal that it was non-compliant. Keeping validationAction: "error" on the primary schema and delegating fallback routing to the application means database-level constraints stay uncompromised while the service owns degradation paths and retry boundaries. This is the same routing discipline described in fallback routing for invalid documents, applied as an ordered chain rather than a single divert.

The chain sits in front of the storage engine, not inside it. Each tier evaluation is a real round trip that either commits or returns a WriteError, so the chain’s cost is proportional to how many tiers a document must traverse before it is accepted. That trade-off — availability now, reconciliation later — is the defining property of the pattern and shapes every constraint below.

Prerequisites & Operational Requirements

The chain assumes tiered collections that already carry their validators, pinned driver and library versions so schema semantics do not shift under you between deploys, and a role that can both write documents and run collMod when you provision or evolve a tier.

Requirement	Minimum	Notes
MongoDB deployment	5.0+ (replica set or Atlas M10+)	$jsonSchema errInfo detail (schemaRulesNotSatisfied) is complete from 5.0; a replica set is required if a reconciliation worker uses Change Streams.
pymongo	>= 4.5	Provides WriteError.details, BulkWriteError, and the timeout exception types used below. Pin exactly in requirements.txt.
jsonschema	>= 4.18	Optional pre-flight validation before the write; ships Draft202012Validator for parity with the collection contract.
Role	readWrite on all tier collections + dbAdmin (or collMod) to provision validators	The runtime chain needs only readWrite; provisioning a new tier needs collMod.
Quarantine collection	TTL index on quarantine_timestamp	Prevents unbounded storage growth; pair with an archival job for documents flagged for manual review.

Version pinning matters because JSON Schema draft semantics differ: a schema authored under Draft 2019-09 can validate a borderline document differently than the same schema under Draft 4, and MongoDB’s validator maps to specific draft behaviour per server version. Align every tier — and any pre-flight Python integration for schema checks — to the draft used by your schema versioning strategy so the chain and the collection never disagree about what “valid” means.

Idempotent Implementation Workflow

Provision the tiers in a fixed, repeatable order. Every step is safe to re-run — re-applying it converges to the same state rather than duplicating resources or double-locking a collection.

1. Create the strict primary tier with an erroring validator. This collection carries the current contract and rejects anything non-compliant with code: 121:

   // mongosh
   db.createCollection("events_v2", {
     validator: { $jsonSchema: {
       bsonType: "object",
       required: ["event_id", "tenant_id", "created_at"],
       properties: {
         event_id:  { bsonType: "string" },
         tenant_id: { bsonType: "string" },
         created_at:{ bsonType: "date" }
       }
     } },
     validationLevel: "strict",
     validationAction: "error"
   });

2. Create the transitional tier with a relaxed validator. It accepts the deprecated shape (for example a string created_at) but still guarantees the fields the reconciliation worker depends on:

   // mongosh
   db.createCollection("events_v2_transitional", {
     validator: { $jsonSchema: {
       bsonType: "object",
       required: ["event_id", "tenant_id"],
       properties: { event_id: { bsonType: "string" } }
     } },
     validationLevel: "strict",
     validationAction: "error"
   });

3. Create the quarantine tier with no validator and a TTL index. It must accept anything, and expire on its own so storage never grows without bound:

   // mongosh
   db.createCollection("events_v2_quarantine");
   db.events_v2_quarantine.createIndex(
     { quarantine_timestamp: 1 },
     { expireAfterSeconds: 60 * 60 * 24 * 30, name: "ttl_quarantine" }
   );

4. Add a unique idempotency index to every tier. The chain retries on transient network failures, so a stable key is what makes those retries safe against duplicate writes:

   // mongosh — run against each tier collection
   db.events_v2.createIndex({ idempotency_key: 1 }, { unique: true, name: "idem" });

5. Wire the tiers into the chain in order, strict first. The orchestrator below takes an ordered list of ValidationTier objects; tier order is the only thing that determines cascade priority, so keep it explicit and version-controlled alongside the schemas.

Because step 1–3 use createCollection (a no-op if the collection exists with the same options) and steps 4–5 rely on unique indexes, re-running the whole sequence after a partial failure reconciles to the intended state instead of corrupting it.

Production-Ready Automation Implementation

The following orchestrator is idempotent and explicitly failing. It intercepts MongoDB validation errors (WriteError code: 121) and routes the document to the next tier without swallowing exceptions or masking network faults. Duplicate-key, authorization, and connectivity errors are never treated as validation failures — they re-raise so the caller can alert rather than silently demote a document. Retries are bounded and scoped to a stable idempotency_key, and every accepted or quarantined document carries routing metadata for downstream reconciliation.

import logging
import time
from typing import Dict, Any, List, Optional, Tuple
from pymongo import errors
from pymongo.collection import Collection
from pymongo.errors import WriteError, PyMongoError

logger = logging.getLogger(__name__)

class ValidationTier:
    """Represents a single validation stage in the fallback chain."""
    def __init__(self, name: str, collection: Collection, max_retries: int = 2):
        self.name = name
        self.collection = collection
        self.max_retries = max_retries

class FallbackValidationChain:
    """
    Orchestrates tiered schema validation for MongoDB documents.
    Each tier maps to a separate collection with its own $jsonSchema
    validator applied via collMod. Tier 1 enforces the current production
    schema; Tier 2 a relaxed transitional schema; the quarantine tier has
    no validator.
    """
    def __init__(self, tiers: List[ValidationTier]):
        if not tiers:
            raise ValueError("At least one validation tier is required.")
        self.tiers = tiers

    def _attempt_insert(self, doc: Dict[str, Any], tier: ValidationTier) -> Tuple[bool, Optional[str]]:
        """
        Attempt to insert a document into a tier's collection.
        Returns (success, failure_reason).
        WriteError code 121 -> validation failure -> try next tier.
        All other errors -> re-raise immediately.
        """
        enriched = {**doc, "validation_tier": tier.name}
        try:
            tier.collection.insert_one(enriched)
            return True, None
        except WriteError as e:
            if e.code == 121:
                return False, e.details.get("errmsg", "Schema validation failed")
            raise  # Duplicate key, auth, etc. — not a validation issue
        except PyMongoError as e:
            logger.error("Database operation failed at tier %s: %s", tier.name, e)
            raise

    def execute(self, document: Dict[str, Any]) -> Dict[str, Any]:
        """
        Route the document through validation tiers in order.
        Returns the final document with routing metadata attached.
        """
        if "idempotency_key" not in document:
            raise ValueError("Document must carry an idempotency_key before entering the chain.")

        last_error = None

        for tier in self.tiers:
            for attempt in range(tier.max_retries + 1):
                try:
                    success, reason = self._attempt_insert(document, tier)
                    if success:
                        logger.info("Document validated at tier: %s", tier.name)
                        return {**document, "validation_tier": tier.name, "status": "accepted"}

                    last_error = reason
                    logger.warning(
                        "Tier %s rejected document (attempt %d/%d): %s",
                        tier.name, attempt + 1, tier.max_retries + 1, reason,
                    )
                    break  # Validation failure is deterministic — move to next tier
                except errors.DuplicateKeyError:
                    # Idempotent replay: this document was already accepted here.
                    logger.info("Idempotent replay detected at tier %s; treating as accepted.", tier.name)
                    return {**document, "validation_tier": tier.name, "status": "accepted"}
                except (errors.NetworkTimeout, errors.ServerSelectionTimeoutError) as net_err:
                    if attempt == tier.max_retries:
                        raise RuntimeError(f"Network failure at tier {tier.name}") from net_err
                    time.sleep(1.0 * (2 ** attempt))

        # All tiers exhausted — quarantine
        logger.error("Document quarantined after exhausting all validation tiers: %s", last_error)
        return {
            **document,
            "validation_tier": "quarantine",
            "status": "rejected",
            "failure_reason": last_error,
            "quarantine_timestamp": time.time(),
        }

Key Operational Safeguards

• Error-code discrimination. Only WriteError code: 121 (DocumentValidationFailure) triggers a cascade. code: 11000 (duplicate) is handled as an idempotent replay; every other PyMongoError re-raises so auth and connectivity faults page a human instead of demoting the document.
• Bounded, keyed retries. Retries apply only to transient network exceptions and are capped per tier. Because each write carries a unique idempotency_key, a retry that actually succeeded server-side surfaces as a DuplicateKeyError and is treated as acceptance — no double write.
• Deterministic routing. A validation failure is not retried within the same tier; the document moves straight to the next tier. This keeps the chain’s latency bounded by tier count, not by retry count.
• Explicit quarantine. Exhausting the chain never drops data. The document lands in quarantine with failure_reason and quarantine_timestamp, ready for the reconciliation described in graceful degradation for legacy document formats.

Fallback chains should not run inside a multi-document transaction: a tier-one validation failure would abort the transaction and roll back the very quarantine write you depend on. Route documents individually, or use bulk operations with ordered=False and parse BulkWriteError.details["writeErrors"] per document so one rejection does not abort the batch.

Diagnostic Fingerprints & Fast Resolution

When the chain misbehaves the failure is almost always in one of three places: the validator rejecting more than expected, a non-validation error being miscounted as a cascade, or quarantine growing unbounded. Each has a precise fingerprint.

Symptom	Fingerprint	Fast resolution
Everything cascades to quarantine	WriteError code: 121 on tier one for near-100% of writes	Upstream contract changed. Inspect errInfo.details.schemaRulesNotSatisfied to find the exact failing keyword; relax the transitional tier or fix the producer.
Writes fail but never quarantine	Uncaught pymongo.errors.OperationFailure code: 13 (auth) or code: 11000 reaching the caller	Not a validation problem. The chain correctly re-raises; fix credentials/index rather than adding tiers.
Quarantine collection grows without bound	db.events_v2_quarantine.stats() size climbing, no TTL deletes	TTL index missing or quarantine_timestamp stored as a float, not a Date. TTL requires a BSON date; convert the field.
Latency spikes on the write path	P95 write latency scales with tier count	Too many tiers. Cap at three and add a circuit breaker when P95 exceeds the SLO.

Copy-paste diagnostics to run during an incident:

# How many documents landed in each tier in the last hour? (application DB)
mongosh --quiet --eval 'db.events_v2_quarantine.aggregate([
  { $match: { quarantine_timestamp: { $gte: new Date(Date.now() - 3600*1000) } } },
  { $group: { _id: "$failure_reason", n: { $sum: 1 } } },
  { $sort: { n: -1 } }
])'

# Confirm the TTL index is actually present on quarantine
mongosh --quiet --eval 'db.events_v2_quarantine.getIndexes()' | jq '.[] | select(.name=="ttl_quarantine")'

# Pull the exact schema rules a rejected document violated (from a JSON worker log)
grep -F '"code": 121' chain.log | jq -r '.details.schemaRulesNotSatisfied'

The definitive indicator that a failure is a validator rejection — not a network or auth fault — is the presence of schemaRulesNotSatisfied in WriteError.details. If it is absent, the chain should never have cascaded, and the fix is upstream of validation.

Edge Cases, Gotchas & Known Limitations

Fallback chains trade write-path simplicity for a set of boundaries you must codify in runbooks:

• Quarantine is a Date, not a timestamp. The reference orchestrator stores time.time() (a float) for portability, but a MongoDB TTL index only expires BSON date fields. In production, write datetime.now(timezone.utc) so the TTL actually fires.
• Idempotency key must precede the chain. The chain enforces its presence, but the producer owns generating a stable key. A key derived from mutable payload fields breaks replay safety the moment the payload is edited.
• Tier count is a latency budget. Every tier a document fails adds a full round trip. Chains beyond three tiers routinely blow write SLOs; prefer widening a single transitional schema over adding a fourth tier.
• Bulk writes need ordered=False. With the default ordered=True, the first code: 121 halts the entire batch and the remaining documents never reach their fallback tiers.
• Tier deprecation is a scheduled process, not a delete. Once a legacy payload drops below ~1% of volume, mark its tier deprecated, shadow-route to a validation queue, and remove it only after an observation window — matching the strict-versus-relaxed cutover discipline in strict vs moderate validation levels.

Instrument every tier transition, rejection, and quarantine event as structured telemetry (OpenTelemetry counters and histograms) and route it to the panels in async validation monitoring dashboards. A sustained tier-one rejection rate above ~5% is the earliest signal of upstream drift; a quarantine document aging past 24 hours without reconciliation should page. How you bucket the underlying failures is exactly the taxonomy problem covered in categorizing schema validation errors, so the two designs should share error categories.

Verification & Rollback Procedures

Confirm the chain routes correctly before you trust it in production. Feed it three probes — one compliant, one transitional, one malformed — and assert each lands in the expected tier:

# Compliant -> tier "primary"; transitional -> tier "transitional"; junk -> "quarantine"
chain = FallbackValidationChain([
    ValidationTier("primary", db.events_v2),
    ValidationTier("transitional", db.events_v2_transitional),
])
from datetime import datetime, timezone
ok  = chain.execute({"idempotency_key": "p1", "event_id": "e1", "tenant_id": "t1", "created_at": datetime.now(timezone.utc)})
tr  = chain.execute({"idempotency_key": "p2", "event_id": "e2", "tenant_id": "t1"})       # no created_at
bad = chain.execute({"idempotency_key": "p3", "note": "no required fields at all"})
assert ok["validation_tier"] == "primary"
assert tr["validation_tier"] == "transitional"
assert bad["status"] == "rejected"

// mongosh — the quarantined probe should be present with a failure_reason
db.events_v2_quarantine.find({ idempotency_key: "p3" }, { failure_reason: 1 }).pretty();

To roll back to single-schema enforcement — for example because a regulated collection cannot tolerate transitional acceptance — the sequence is non-destructive and reversible. Stop routing to the transitional and quarantine tiers so the service inserts only into the strict collection; drain the transitional collection with a one-off reconciliation into the primary; then archive the fallback collections. Because every write was an idempotent upsert on idempotency_key, replaying the drain cannot double-insert. Time to recover is typically under five minutes to cut over routing; the only long pole is draining a large transitional backlog, which is proportional to its document count.

Frequently Asked Questions

Related

• Automated Schema Enforcement & Monitoring — the parent architecture this chain plugs into, spanning validators, middleware, and pre-flight checks.
• Graceful degradation for legacy document formats — the reconciliation half of the chain: transforming and upserting quarantined legacy payloads back into the strict collection.
• Implementing collection-level validators — the per-tier enforcement primitive the chain orchestrates.
• Fallback routing for invalid documents — the single-divert routing model this chain generalizes into an ordered sequence.
• Async validation monitoring dashboards — where tier-hit-rate and quarantine-aging telemetry become operational signal.