Migrating Work Order Payloads Across Schema Versions Without Dead-Lettering v1 Messages

A schema bump that adds mandatory SLA fields or renames a column is often necessary and always disruptive to whatever v1 payloads are still sitting in queues, replay topics, and cold backups. This page follows one such bump end to end: an ingestion consumer that validates every message — live, replayed, or restored — against the newest work order schema standards contract, with no notion that some of those messages predate the bump by months. The fix is not a looser schema; it is a small, ordered migration layer that upcasts each payload to the current version before the validator ever sees it.

Incident Profile

A scheduled backup restore replays 1,318 archived work orders through the ingestion consumer after a maintenance window. The consumer code has not changed — it has always validated inbound payloads against whatever Python model represents “the schema” — but the schema underneath it moved twice since those messages were written: priority, requested_completion, and escalation_tier became mandatory fields, and site_id was renamed to location_id. None of the archived messages carry a schema_version tag, because that field did not exist when they were queued.

The replay stalls at message 4,402:

2026-07-14 03:22:18,114 ERROR [wo_ingest.consumer] Validation failed for message backup-2024-11-02T14:03:00Z-WO-55210 on queue "workorders.replay"
Traceback (most recent call last):
  File "/opt/cmms-integration/consumers/work_order_consumer.py", line 58, in handle_message
    payload = WorkOrderPayloadV3.model_validate(raw)
  File "/usr/local/lib/python3.11/site-packages/pydantic/main.py", line 503, in model_validate
    raise validation_error
pydantic_core._pydantic_core.ValidationError: 3 validation errors for WorkOrderPayloadV3
location_id
  Field required [type=missing, input_value={'work_order_id': 'WO-5521...uired_quantities': {...}}, input_type=dict]
priority
  Field required [type=missing, input_value={'work_order_id': 'WO-5521...uired_quantities': {...}}, input_type=dict]
escalation_tier
  Field required [type=missing, input_value={'work_order_id': 'WO-5521...uired_quantities': {...}}, input_type=dict]
2026-07-14 03:22:18,115 WARNING [wo_ingest.consumer] Message WO-55210 dead-lettered to "workorders.dlq" after 0 retries (no schema_version tag, no migration path)
2026-07-14 03:22:18,115 ERROR [wo_ingest.consumer] Replay batch aborted at offset 4402 — 1,318 messages remaining in backup unprocessed

Three details separate this from an ordinary validation bug:

  • “3 validation errors” with no malformed content. The message round-tripped through storage untouched. Every field it has is exactly what it always was — the model simply grew fields the message never had a chance to carry.
  • “0 retries.” The dead-letter handler is working as designed; retrying a payload against a schema it can never satisfy without transformation wastes attempts and hides the real defect, which is upstream of the retry loop entirely.
  • A batch abort, not a single rejected message. Because the consumer treats a ValidationError as fatal for the whole replay run, one version-less message blocks 1,318 others behind it — turning a per-message quirk into an operational incident.
Validate-only versus migrate-then-validate for a v1 work order payload A two-lane pipeline diagram. In the before lane, an untagged v1 payload with no schema_version field is validated directly against the WorkOrderPayloadV3 model, which raises a validation error for missing location_id, priority, and escalation_tier, and the message is dead-lettered. In the after lane, the same v1 payload first passes through migrate_v1_to_v2, which renames site_id to location_id and backfills priority and requested_completion, then through migrate_v2_to_v3, which backfills escalation_tier as zero; the resulting v3 payload passes validation, and its priority and escalation_tier fields are then used to start SLA timers. Why a v1 payload dies at the validator — and how upcasting fixes it BEFORE — validate-only (dead-letters v1) v1 payload no schema_version tag site_id, no priority, no escalation_tier validate() against WorkOrderPayloadV3 no upcasting step run first, in every case ValidationError location_id: missing priority: missing escalation_tier: missing Dead-lettered workorders.dlq 0 retries attempted replay batch aborted AFTER — migrate-then-validate (upcasts v1 → v2 → v3) v1 payload no schema_version inferred as v1 (has site_id) migrate_v1_to_v2() site_id → location_id + priority (default) + requested_completion v2 payload location_id, priority set schema_version = 2 escalation_tier missing migrate_v2_to_v3() + escalation_tier = 0 schema_version → 3 safe, explicit default v3 payload validate(WorkOrderPayloadV3) passes — dispatched priority + escalation_tier now safe to start SLA timers

Root Cause Analysis

Nothing about the archived work order is wrong; it was a perfectly valid message the day it was written. Three gaps converge so that a version-less message can never satisfy the newest contract without help.

  1. No schema_version tag on the message. The producer that wrote this backup predates the practice of stamping a version onto every payload, so every legacy message in that store is version-less by construction, not by error. A consumer that assumes every message it reads carries a version tag has already excluded every message written before that assumption existed.
  2. The consumer validates directly against the latest model. The batch-replay path inside the async batch processing consumer calls WorkOrderPayloadV3.model_validate(raw) on every message it pulls off a live queue, a replay topic, or a cold backup, with no branch for “this message might predate the current contract.”
  3. No migration step exists between ingestion and validation. The schema owns the shape a work order must have; nothing owns the transformation from an old shape to a new one. Bumping the schema without shipping a migration alongside it is, from the point of view of every message already in flight, equivalent to deleting the old schema outright.

The three missing fields in the traceback are not random — they are exactly the SLA fields priority, requested_completion, and escalation_tier that made the schema bump necessary in the first place, plus location_id, which only reads as “missing” because the validator is looking for a name the v1 message never used (site_id).

This is a different failure mode than the type and required-field mismatches covered in debugging JSON schema validation failures on CMMS work order payloads: that incident is a single schema rejecting current data that is legitimately shaped differently than the schema assumes. This incident is a single schema being asked to describe every version of the data that has ever existed, with no bridge between them. The generalized lesson belongs at the taxonomy level: any versioned contract inside CMMS architecture and maintenance taxonomy needs a version stamped at write time and a migration path shipped with every bump, not a validator that quietly assumes the newest shape is the only shape in the wild.

Resolution

The canonical work order object used across this site carries three mandatory SLA fields — priority, requested_completion, escalation_tier — as a plain dataclass. This page models the current version as a pydantic BaseModel instead, because validation (and the exact error list from the incident log above) is the point, and pydantic’s ValidationError is what the consumer actually raises in production. The field names and SLA fields stay identical to the canonical form; only the construction style differs.

Before — validate-only, no version awareness:

from pydantic import BaseModel, ValidationError
from typing import Dict, List, Optional
from datetime import datetime
from enum import Enum


class Priority(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"


class WorkOrderPayloadV3(BaseModel):
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    location_id: str
    priority: Priority
    requested_completion: Optional[datetime] = None
    escalation_tier: int
    status: str = "open"


def handle_message(raw: dict) -> None:
    # Every message, regardless of age, validated against the latest schema.
    try:
        payload = WorkOrderPayloadV3.model_validate(raw)
    except ValidationError as exc:
        # A v1 message missing fields introduced by later bumps lands here,
        # including untouched backups replayed months after the fact.
        dead_letter(raw, reason=str(exc))
        return
    dispatch(payload)

After — tag, migrate in order, then validate:

from typing import Callable, Dict as TDict

MigrationFn = Callable[[dict], dict]


def migrate_v1_to_v2(payload: dict) -> dict:
    """v1 -> v2: rename site_id, backfill the first SLA fields."""
    payload = dict(payload)
    payload["location_id"] = payload.pop("site_id")
    payload.setdefault("priority", Priority.STANDARD.value)
    payload.setdefault("requested_completion", None)
    payload["schema_version"] = 2
    return payload


def migrate_v2_to_v3(payload: dict) -> dict:
    """v2 -> v3: escalation_tier becomes required; default to 0 (no escalation)."""
    payload = dict(payload)
    payload.setdefault("escalation_tier", 0)
    payload["schema_version"] = 3
    return payload


MIGRATIONS: TDict[int, MigrationFn] = {
    1: migrate_v1_to_v2,
    2: migrate_v2_to_v3,
}
LATEST_SCHEMA_VERSION = 3


def infer_schema_version(payload: dict) -> int:
    """Untagged legacy payloads predate schema_version; detect them by shape."""
    if "schema_version" in payload:
        return payload["schema_version"]
    if "site_id" in payload:
        return 1
    return 2  # has location_id already, but no escalation_tier yet


def migrate_to_latest(payload: dict) -> dict:
    version = infer_schema_version(payload)
    while version < LATEST_SCHEMA_VERSION:
        payload = MIGRATIONS[version](payload)
        version = payload["schema_version"]
    return payload


def handle_message(raw: dict) -> None:
    # Upcast first, at every version boundary in order, then validate once.
    upcasted = migrate_to_latest(raw)
    try:
        payload = WorkOrderPayloadV3.model_validate(upcasted)
    except ValidationError as exc:
        dead_letter(raw, reason=str(exc))
        return
    dispatch(payload)

Three changes carry the fix. infer_schema_version gives every untagged legacy message a starting point instead of assuming it is already current. MIGRATIONS is an ordered registry keyed by the version a function upgrades from, so migrate_to_latest can walk a v1 payload through v2 on its way to v3 without a special case for “how far behind is this message.” And each migration backfills new required fields with an explicit, documented default (STANDARD priority, escalation_tier = 0) rather than leaving the validator to reject silently on their absence.

Minimal Reproducible Pipeline

This script runs as-is. It defines the versioned canonical WorkOrderPayloadV3 model, the ordered migration registry, and a migrate-then-validate consumer, then replays one legacy v1 dict both without and with the migration step so you can see the incident and the fix side by side.

import logging
from datetime import datetime, timezone
from enum import Enum
from typing import Callable, Dict, List, Optional

from pydantic import BaseModel, Field, ValidationError

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s [%(name)s] %(message)s")
logger = logging.getLogger("wo_schema_migration")


class Priority(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"


class WorkOrderPayloadV3(BaseModel):
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    schema_version: int = 3
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    location_id: str
    priority: Priority
    requested_completion: Optional[datetime] = None
    escalation_tier: int
    status: str = "open"
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


MigrationFn = Callable[[dict], dict]


def migrate_v1_to_v2(payload: dict) -> dict:
    """v1 -> v2: rename site_id to location_id, backfill the earliest SLA fields."""
    payload = dict(payload)
    payload["location_id"] = payload.pop("site_id")
    payload.setdefault("priority", Priority.STANDARD.value)
    payload.setdefault("requested_completion", None)
    payload["schema_version"] = 2
    return payload


def migrate_v2_to_v3(payload: dict) -> dict:
    """v2 -> v3: escalation_tier becomes required; default to 0 (no escalation)."""
    payload = dict(payload)
    payload.setdefault("escalation_tier", 0)
    payload["schema_version"] = 3
    return payload


MIGRATIONS: Dict[int, MigrationFn] = {
    1: migrate_v1_to_v2,
    2: migrate_v2_to_v3,
}
LATEST_SCHEMA_VERSION = 3


def infer_schema_version(payload: dict) -> int:
    """Untagged legacy payloads predate schema_version; detect them by shape."""
    if "schema_version" in payload:
        return payload["schema_version"]
    if "site_id" in payload:
        return 1
    return 2


def migrate_to_latest(payload: dict) -> dict:
    version = infer_schema_version(payload)
    while version < LATEST_SCHEMA_VERSION:
        migration = MIGRATIONS[version]
        payload = migration(payload)
        version = payload["schema_version"]
    return payload


def dead_letter(raw: dict, reason: str) -> None:
    logger.warning("Dead-lettering %s: %s", raw.get("work_order_id", "UNKNOWN"), reason)


def dispatch(payload: WorkOrderPayloadV3) -> None:
    logger.info(
        "Dispatch-ready %s (schema v%d, priority=%s, escalation_tier=%d)",
        payload.work_order_id,
        payload.schema_version,
        payload.priority.value,
        payload.escalation_tier,
    )


def handle_message(raw: dict, skip_migration: bool = False) -> None:
    """Migrate-then-validate consumer. skip_migration=True reproduces the incident."""
    candidate = raw if skip_migration else migrate_to_latest(raw)
    try:
        payload = WorkOrderPayloadV3.model_validate(candidate)
    except ValidationError as exc:
        dead_letter(raw, reason=str(exc))
        return
    dispatch(payload)


if __name__ == "__main__":
    legacy_v1_message = {
        "work_order_id": "WO-55210",
        "asset_id": "CONV-08",
        "part_skus": ["BLT-2201"],
        "required_quantities": {"BLT-2201": 2},
        "site_id": "WH-02",
        "status": "open",
        "created_at": "2024-11-02T14:03:00Z",
    }

    logger.info("--- replaying without migration (reproduces the incident) ---")
    handle_message(legacy_v1_message, skip_migration=True)

    logger.info("--- replaying with migrate-then-validate ---")
    handle_message(legacy_v1_message, skip_migration=False)

Run it and the first pass logs a dead-letter warning naming location_id, priority, and escalation_tier as missing — the same three fields from the incident traceback. The second pass, on the identical dict, logs a dispatch-ready message at schema v3 with priority=standard and escalation_tier=0 backfilled by the migration chain. Nothing about the input changed between the two runs; only whether it passed through migrate_to_latest first.

Prevention Checklist

FAQ

Why not just make the new SLA fields optional instead of migrating old payloads?

Because priority, requested_completion, and escalation_tier drive real downstream behavior — routing, timers, and escalation — making them optional just moves the missing-field problem from the validator to every consumer that reads them, at a point further from the source and harder to diagnose. A migration with an explicit default keeps the fields mandatory everywhere they are used while giving old data a documented, deliberate value instead of None scattered through the codebase.

What happens to a payload with a schema_version newer than anything my migration registry knows about?

migrate_to_latest should raise rather than guess. If infer_schema_version (or an explicit tag) returns a version higher than LATEST_SCHEMA_VERSION, that means the consumer is older than the data it received — a deploy-ordering problem, not a migration problem — and the message belongs in a hold queue for replay once the consumer catches up, not in the dead-letter queue for a shape defect it doesn’t actually have.

Should schema_version live in the payload or in queue/broker metadata?

Either works as long as it is written once and read consistently; the failure mode in this incident came from having it in neither place. Putting it in the payload body keeps it intact across export/backup/replay boundaries that may not preserve broker headers, which is why the migrations here treat it as a payload field. Broader storage-format decisions like this belong with the rest of the CMMS architecture and maintenance taxonomy documentation so every producer agrees on where to look.

Do dead-lettered messages need to be replayed after the migration ships, or are they gone for good?

They should be replayed, not discarded — a dead letter caused by a missing migration is a transient defect in the consumer, not a defect in the message. Once migrate_to_latest covers the version those messages predate, replay the dead-letter queue through the same handle_message path used for live traffic so the fix is verified against the exact failures it was built to resolve.

This incident sits inside work order schema standards; pair it with the structural rules in debugging JSON schema validation failures on CMMS work order payloads, keep replay-safe consumers aligned with async batch processing, and confirm the backfilled SLA fields feed correctly into SLA timer escalation once messages are dispatched. For the wider set of taxonomy decisions this schema depends on, see CMMS architecture and maintenance taxonomy.

Part of: Work Order Schema Standards.