Fallback Routing for Invalid Documents

Within the broader MongoDB JSON Schema Validation Architecture, fallback routing is the resilience layer that stops a strict validator from ever meaning silent data loss. This guide is a complete implementation workflow for platform engineers: it shows how to intercept a WriteError (code 121) at the driver boundary, persist the rejected payload to an isolated quarantine namespace, and hand it to an asynchronous worker for transformation and retry — all without letting the validation failure propagate as a hard error to the upstream caller. Deployed correctly, this pattern decouples schema enforcement from write availability, which is exactly what zero-downtime migrations, gradual schema adoption, and legacy-data compatibility demand in production. Deployed carelessly, the same interception layer becomes a black hole that swallows corruption, recurses on its own quarantine collection, or grows unbounded until it exhausts disk.

The deliverable by the end of this page is a runnable pymongo routing handler you can drop into a service’s write path, plus the diagnostic, verification, and rollback commands to operate the quarantine namespace safely.

Architectural Context & Enforcement Boundaries

MongoDB evaluates $jsonSchema constraints synchronously on the write path. When a document violates the declared contract, the server rejects the operation with a WriteError carrying code: 121 and — on MongoDB 5.0 and later — an errInfo.details.schemaRulesNotSatisfied array that pinpoints the exact failing rule and path. Fallback routing sits around that rejection: it is a driver-side catch that converts the exception into a durable, observable state instead of a 500 to the caller. It does not replace the validator, and it does not weaken it — the contract stays strict while the routing layer absorbs the fallout.

This layer is deliberately downstream of the schema itself. The precision of your rules, defined against the MongoDB $jsonSchema syntax, determines what actually lands in quarantine; a loose schema routes too little and lets drift through, while an over-strict one routes legitimate traffic and floods the dead-letter namespace. The interaction with strict vs moderate validation levels is subtle and load-bearing: under moderate, updates to already-non-compliant documents bypass validation entirely, so the routing layer never sees them and drift accumulates invisibly unless you track validation context per operation. Fallback routing is the application-tier sibling of the asynchronous fallback validation chains built on the enforcement-and-monitoring side of the platform, and it feeds the same telemetry that categorizing schema validation errors depends on to tell a transient bug apart from a migration gap.

The quarantine collection is a dead-letter queue, not a permanent archive. Each routed document needs a stable deduplication key, the raw payload, the validation error string, the originating namespace, a routing status, and temporal markers. A compound unique index on the payload hash and status makes routing idempotent under retries; a second index on the timestamp powers both retention sweeps and reconciliation queries.

Prerequisites & Operational Requirements

The workflow below targets a supported production topology. Confirm the following before wiring fallback routing into a live service.

• MongoDB version: 5.0 or later. The rich errInfo.details.schemaRulesNotSatisfied structure — which lets the worker decide how to transform a payload — was introduced in 5.0. On 4.x you receive only code: 121 and a generic errmsg, so routing still works but the worker is blind to the specific failing rule.
• Driver: PyMongo 4.x (pip install "pymongo>=4.6,<5"). Pin the driver in your service image so WriteError / BulkWriteError classes and the details payload shape stay stable across builds.
• Permissions: the service principal needs insert on both the primary and the quarantine collection, plus createIndex on first run. The remediation worker needs find and update on the quarantine namespace and insert on the primary. Isolate these into a dedicated remediation role — the quarantine collection should never be writable by general application traffic.
• Environment assumptions: a replica set with w: "majority" write concern on the quarantine upsert, so a routed document is durable before you acknowledge the caller. Provision the quarantine namespace with its own disk-usage alert; a routing storm can grow it faster than any single collection you monitor.
• Schema source of truth: the validator that triggers routing must come from a version-controlled registry. Aligning the routing layer with your schema versioning strategies is what lets a worker know which schema version a quarantined document failed against, and therefore how to migrate it forward.

One boundary deserves special attention up front: the quarantine collection must be excluded from any cross-collection validation patterns and from any wildcard validator deployment. If a validator is ever applied to *_quarantine, a rejected raw payload can itself fail validation on the way into quarantine, and the router recurses. The quarantine namespace is intentionally schema-free.

Idempotent Implementation Workflow

A production routing path must be deterministic: the same rejected payload arriving twice must produce exactly one quarantine record, and a transient network blip must not drop the document on the floor. Follow this sequence.

1. Ensure the quarantine indexes once. On service startup, create the compound unique index on (payload_hash, status) and the retention index on routed_at. Index creation is idempotent, so it is safe to call on every boot:

   db.orders_quarantine.createIndex(
     { payload_hash: 1, status: 1 },
     { unique: true, name: "idx_quarantine_hash_status" }
   )
   db.orders_quarantine.createIndex(
     { routed_at: 1 },
     { name: "idx_quarantine_retention", expireAfterSeconds: 2592000 }
   )

   The expireAfterSeconds: 2592000 turns the retention index into a TTL index that purges routed documents after 30 days unless a worker has promoted their status out of the sweep window.

2. Catch narrowly at the write boundary. Wrap the primary insert_one / bulk_write in a handler that inspects exc.code == 121 and re-raises everything else. A WriteError from a duplicate key or a NotWritablePrimary is not a validation failure and must never be routed.

3. Hash the payload deterministically. Serialize the rejected document with sorted keys and a stable default encoder, then SHA-256 it. This hash is both the deduplication key and a cheap fingerprint for grouping identical failures in dashboards.

4. Upsert with $setOnInsert. Route with an upsert keyed on the hash so a retried write collapses into the existing record instead of creating a duplicate. Using $setOnInsert preserves the original routed_at and retry_count on repeat arrivals.

5. Acknowledge the caller with a routed status. Return a structured {"status": "quarantined"} result rather than raising, so the upstream service records a recoverable outcome instead of an outage.

6. Reconcile asynchronously. A separate worker polls quarantine by status: "pending_remediation", applies transformation logic keyed on the failing rule, and re-inserts into the primary collection — promoting the record’s status on success so the TTL sweep can reclaim it.

Production-Ready Automation Implementation

The following pymongo implementation performs the full routing path: it ensures the quarantine indexes, hashes the payload for deduplication, and upserts to the quarantine namespace with bounded exponential backoff on transient failures. It is safe to construct once per collection and reuse across requests. The safe_insert_with_fallback wrapper shows the narrow catch that separates a schema rejection from every other write error.

import hashlib
import json
import logging
import time
from datetime import datetime, timezone
from typing import Dict, Any, Optional

from pymongo import MongoClient, errors
from pymongo.results import UpdateResult

logger = logging.getLogger(__name__)


class FallbackRouter:
    def __init__(
        self,
        client: MongoClient,
        db_name: str,
        target_collection: str,
        quarantine_suffix: str = "_quarantine",
    ):
        self.client = client
        self.db = client[db_name]
        self.target_collection = target_collection
        self.quarantine_collection = self.db[f"{target_collection}{quarantine_suffix}"]
        self._ensure_indexes()

    def _ensure_indexes(self) -> None:
        """Create compound indexes for idempotent routing and TTL retention sweeps."""
        self.quarantine_collection.create_index(
            [("payload_hash", 1), ("status", 1)],
            unique=True,
            name="idx_quarantine_hash_status",
        )
        # 30-day TTL: purge routed documents unless a worker promotes their status.
        self.quarantine_collection.create_index(
            [("routed_at", 1)],
            name="idx_quarantine_retention",
            expireAfterSeconds=2_592_000,
        )

    @staticmethod
    def _generate_hash(payload: Dict[str, Any]) -> str:
        """Deterministic SHA-256 hash for payload deduplication."""
        canonical = json.dumps(payload, sort_keys=True, default=str).encode("utf-8")
        return hashlib.sha256(canonical).hexdigest()

    def route(
        self,
        original_payload: Dict[str, Any],
        error_details: str,
        operation_type: str = "insert",
    ) -> Optional[UpdateResult]:
        """Intercept a validation failure and persist to the quarantine namespace."""
        payload_hash = self._generate_hash(original_payload)
        quarantine_doc = {
            "payload_hash": payload_hash,
            "original_payload": original_payload,
            "validation_error": error_details,
            "source_collection": self.target_collection,
            "operation_type": operation_type,
            "status": "pending_remediation",
            "routed_at": datetime.now(timezone.utc),
            "retry_count": 0,
        }

        max_retries = 3
        for attempt in range(max_retries):
            try:
                result = self.quarantine_collection.update_one(
                    {"payload_hash": payload_hash},
                    {"$setOnInsert": quarantine_doc},
                    upsert=True,
                )
                logger.info(
                    "Quarantined document %s (upserted: %s)",
                    payload_hash,
                    result.upserted_id is not None,
                )
                return result
            except errors.DuplicateKeyError:
                # Concurrent router already inserted this payload — routing is idempotent.
                logger.info("Document %s already quarantined by a concurrent writer.", payload_hash)
                return None
            except errors.PyMongoError as exc:
                logger.warning("Quarantine write attempt %d failed: %s", attempt + 1, exc)
                if attempt < max_retries - 1:
                    time.sleep(0.5 * (2 ** attempt))
                else:
                    logger.error("Failed to quarantine document after %d attempts. Raising.", max_retries)
                    raise


def safe_insert_with_fallback(
    router: FallbackRouter,
    document: Dict[str, Any],
) -> Dict[str, Any]:
    """Insert into the primary collection, routing schema rejections to quarantine."""
    coll = router.db[router.target_collection]
    try:
        result = coll.insert_one(document)
        return {"status": "success", "inserted_id": str(result.inserted_id)}
    except errors.WriteError as exc:
        if exc.code == 121:
            # Reach the rich rule detail on 5.0+; fall back to errmsg otherwise.
            details = exc.details or {}
            error_msg = json.dumps(details.get("errInfo", details.get("errmsg", "validation failed")))
            router.route(document, error_msg, operation_type="insert")
            return {"status": "quarantined", "error": error_msg}
        raise  # Non-validation write errors (duplicate key, etc.) propagate immediately.
    except errors.PyMongoError as exc:
        logger.critical("Database operation failed outside validation scope: %s", exc)
        raise

Key Operational Safeguards

• Deterministic hashing makes routing idempotent: a payload that fails and is retried collapses into one quarantine record via the unique index, and a concurrent race surfaces as a caught DuplicateKeyError rather than a duplicate row.
• Narrow exception handling routes only code: 121. A duplicate-key WriteError or a ServerSelectionTimeoutError re-raises untouched, so infrastructure faults never masquerade as schema violations.
• $setOnInsert semantics preserve the first-seen routed_at and retry_count, keeping the retention TTL and remediation counters honest across repeat arrivals.
• Bounded backoff limits transient-failure retries to three attempts before raising, so the caller fails fast on a genuinely unreachable quarantine namespace instead of hanging.

Diagnostic Fingerprints & Fast Resolution

When the validator rejects a write, PyMongo raises pymongo.errors.WriteError for a single insert or pymongo.errors.BulkWriteError for a batch. The distinguishing signature is code: 121 / codeName: "DocumentValidationFailure". On MongoDB 5.0+, exc.details["errInfo"]["details"]["schemaRulesNotSatisfied"] carries the exact rule the worker needs:

{
  "code": 121,
  "codeName": "DocumentValidationFailure",
  "errmsg": "Document failed validation",
  "errInfo": {
    "details": {
      "operatorName": "$jsonSchema",
      "schemaRulesNotSatisfied": [
        {
          "operatorName": "bsonType",
          "specifiedAs": { "bsonType": "double" },
          "reason": "type did not match",
          "consideredType": "string"
        }
      ]
    }
  }
}

For a BulkWriteError, the per-document failures live under exc.details["writeErrors"], each with its own index, code, and errInfo — route each entry whose code is 121 and re-raise the batch only if a non-validation error is present.

Copy-paste diagnostics for the quarantine namespace from mongosh:

// How much is currently pending, grouped by the failing rule fingerprint.
db.orders_quarantine.aggregate([
  { $match: { status: "pending_remediation" } },
  { $group: { _id: "$validation_error", count: { $sum: 1 } } },
  { $sort: { count: -1 } }
])

// Confirm the recursion guard: the quarantine namespace must have NO validator.
db.getCollectionInfos({ name: "orders_quarantine" })[0].options.validator

Common fingerprints and their fix:

• Quarantine volume spikes with a single dominant validation_error. This is upstream schema drift or a bad deploy, not scattered bad data. Correlate the spike against your async validation monitoring dashboards and roll the offending producer back before draining quarantine.
• DuplicateKeyError on the quarantine upsert with no $setOnInsert. You are routing the same payload under two different statuses. Keep the unique index on (payload_hash, status) and only ever insert with status: "pending_remediation"; let the worker transition status with an update, not a fresh insert.
• Documents route into quarantine and are never re-inserted. The worker is transforming against the wrong schema version. Verify the quarantined validation_error still matches the live validator — if the schema moved on, the transformation target moved with it.

Edge Cases, Gotchas & Known Limitations

• Recursive routing. If any validator is ever applied to the quarantine collection, a raw rejected payload can fail validation on the way in and re-enter the router. Keep *_quarantine schema-free and exclude it explicitly from wildcard validator automation.
• moderate validation hides drift from the router. Under validationLevel: "moderate", updates to already-non-compliant documents bypass validation, so those writes never raise code: 121 and never reach fallback routing. If your goal is to catch legacy drift, you need strict, not moderate.
• Unbounded growth under a routing storm. A misconfigured producer can route millions of documents faster than any worker drains them. The TTL index is a floor, not a throttle — pair it with a disk-usage alert and a circuit breaker that trips the producer when quarantine depth crosses a threshold.
• BSON size on the raw payload. Storing original_payload verbatim means a 16 MB document consumes 16 MB of quarantine plus overhead. For large payloads, store a reference to object storage instead of the inline body.
• PII in quarantine. Rejected payloads are still real production data. The quarantine namespace inherits every data-handling obligation of the primary collection — apply the same security boundaries in schema design and access restrictions, and never widen read access to debug it.

Verification & Rollback Procedures

Confirm routing is wired correctly before you rely on it in production:

// 1. Attempt a known-bad insert against the primary and confirm it lands in quarantine.
db.orders.insertOne({ intentionallyMissing: true })   // expect WriteError 121 in the app
db.orders_quarantine.find({ status: "pending_remediation" }).sort({ routed_at: -1 }).limit(1)

// 2. Prove idempotency: the same bad payload twice yields one quarantine record.
db.orders_quarantine.countDocuments({ payload_hash: "<hash-from-step-1>" })   // expect 1

If routing misbehaves — for example a bug in the handler that quarantines valid documents — the safest rollback is to disable the fallback path and let the validator surface rejections directly, which restores the original fail-loud behavior while you fix the handler. That is a deploy of the service, not a database change, so time-to-recover is one rollout cycle.

To drain and reset the quarantine namespace itself after remediation:

// Soft rollback: promote drained records so the TTL sweep reclaims them.
db.orders_quarantine.updateMany(
  { status: "pending_remediation", payload_hash: { $in: [ /* remediated hashes */ ] } },
  { $set: { status: "remediated" } }
)

// Hard reset: drop the namespace entirely (indexes are re-created on next startup).
db.orders_quarantine.drop()

Dropping the collection is immediate and metadata-only; the router recreates its indexes on the next service boot via _ensure_indexes. Never drop quarantine before confirming every pending document has been either re-inserted or intentionally discarded, because a drop is unrecoverable.

Frequently Asked Questions

Related

• MongoDB JSON Schema Validation Architecture — the parent architecture this resilience layer plugs into, spanning syntax, enforcement levels, and versioning.
• Strict vs moderate validation levels — why the level you choose decides which rejections ever reach the router.
• Understanding MongoDB $jsonSchema syntax — the rule semantics that determine what actually lands in quarantine.
• Cross-collection validation patterns — the multi-collection rules the quarantine namespace must be excluded from to avoid recursion.
• Building fallback validation chains — the asynchronous, monitoring-side counterpart to this application-tier routing handler.
• Categorizing schema validation errors — how to classify a routed code-121 rejection into a bug, a migration gap, or an intentional schema change.