Running Data Migrations with Priority Lanes

The Problem

Every growing system eventually needs to backfill data: adding a new field to existing records, enriching records from an external source, or recalculating derived values after a business rule change. In a CQRS system with asynchronous event processing, these migrations are not simple database updates. Each record change produces domain events that flow through the same pipeline as production traffic.

Consider this scenario: your e-commerce platform has been running for months. You need to backfill a loyalty_tier field onto 500,000 existing customer records. The migration script loads each customer, sets the tier, and saves, producing 500,000 LoyaltyTierAssigned events.

Those events flood the same Redis Stream that handles real-time customer registrations, order placements, and payment confirmations. The Engine processes them in FIFO order, so production events are now stuck behind half a million migration events. A customer who just placed an order waits minutes (or longer) for their confirmation because the projector is busy processing backfill events.

Without priority lanes:

  Production event ──┐
  Migration event  ──┤
  Migration event  ──┤──► Single Redis Stream ──► [Processed in FIFO order]
  Production event ──┤      "identity::customer"   Migration events block
  Migration event  ──┘                              production traffic

You could pause the migration, wait for production to catch up, and resume in small batches, but this is manual, error-prone, and slow. You could run the migration at 3 AM, but that only works if your traffic has a quiet window.

The root issue is that migration events and production events are treated identically, even though they have fundamentally different urgency profiles.

The Pattern

Route migration events to a separate backfill lane that the Engine processes only when production work is idle. This way, migrations run continuously at full speed without ever delaying a production event.

Tag migration commands with a low priority, and let the infrastructure route them accordingly.

With priority lanes:

  Production events ──► Outbox(priority=0)  ──► "identity::customer"           ──► [Drained first]
                                                                                     Non-blocking read
                                                                                          │
  Migration events  ──► Outbox(priority=-50) ──► "identity::customer:backfill" ──► [Drained when idle]
                                                                                     Only when primary empty

The migration script does not need to know about streams, consumer groups, or the Engine's scheduling strategy. It only needs to mark its work as low priority. The framework handles the rest.

How Protean Supports It

Protean provides this pattern through three mechanisms:

1. Processing Priority Context

The processing_priority() context manager tags all commands processed within a block with a specific priority. Events produced by those commands inherit the priority in their outbox records:

# fragment
from protean.utils.processing import processing_priority, Priority

with processing_priority(Priority.LOW):
    domain.process(SomeCommand(...))  # Events get priority=-50

2. OutboxProcessor Routing

The OutboxProcessor checks each message's priority against a configurable threshold (default: 0). Messages below the threshold are published to the backfill stream (here identity::customer:backfill) instead of the primary stream (identity::customer).

3. StreamSubscription Draining

The Engine's StreamSubscription reads from the primary stream first (non-blocking). Only when the primary stream is empty does it make one blocking read that waits on both streams (capped at 1 second). Production events are never blocked by backfill events.

No handler changes are needed. The same projector processes events identically regardless of which lane they arrived on.

For the full configuration and priority level reference, see Priority Lanes.

Applying the Pattern

A Complete Migration Script

The script works on this Customer aggregate. Priority lanes are switched on in the domain's configuration:

from protean import Domain, handle
from protean.fields import Float, Identifier, String
from protean.utils.globals import current_domain

domain = Domain(
    name="Identity",
    config={
        "command_processing": "sync",
        "server": {
            "default_subscription_type": "stream",
            "priority_lanes": {"enabled": True, "threshold": 0},
        },
    },
)


@domain.aggregate
class Customer:
    name: String(required=True)
    lifetime_spend: Float(default=0.0)
    loyalty_tier: String()

    def assign_tier(self, tier: str) -> None:
        self.loyalty_tier = tier
        self.raise_(LoyaltyTierAssigned(customer_id=self.id, loyalty_tier=tier))


@domain.event(part_of=Customer)
class LoyaltyTierAssigned:
    customer_id: Identifier(required=True)
    loyalty_tier: String(required=True)


@domain.command(part_of=Customer)
class UpdateCustomerTier:
    customer_id: Identifier(required=True)
    loyalty_tier: String(required=True)


@domain.command_handler(part_of=Customer)
class CustomerCommandHandler:
    @handle(UpdateCustomerTier)
    def update_tier(self, command: UpdateCustomerTier) -> None:
        repo = current_domain.repository_for(Customer)
        customer = repo.get(command.customer_id)
        customer.assign_tier(command.loyalty_tier)
        repo.add(customer)


domain.init(traverse=False)

Here is a migration script that backfills a loyalty_tier field on existing customer records. The key is wrapping all domain.process() calls inside a processing_priority(Priority.LOW) context manager. The script reads customers one page at a time, because a query returns at most 100 records unless you set a limit.

"""migrate_loyalty_tiers.py

Backfill loyalty_tier on all existing customers based on their
total lifetime spend. Uses Priority.LOW so that events produced
by this script are routed to the backfill lane.
"""

import logging
import sys
import time

from protean.utils.processing import Priority, processing_priority

logger = logging.getLogger(__name__)

TIER_THRESHOLDS = [
    (10000, "platinum"),
    (5000, "gold"),
    (1000, "silver"),
    (0, "bronze"),
]


def determine_tier(lifetime_spend: float) -> str:
    for threshold, tier in TIER_THRESHOLDS:
        if lifetime_spend >= threshold:
            return tier
    return "bronze"


def customers_in_batches(batch_size: int):
    # A query returns 100 records unless told otherwise, so page through
    # the table instead of reading it with a single `all()`. Production
    # keeps writing while the script runs, so each page starts after the
    # last ID seen. With an offset, deleting a row from an earlier page
    # would shift a later row back past the offset, and it would be skipped.
    query = domain.repository_for(Customer).query.order_by("id").limit(batch_size)
    page = query.all()
    while page.items:
        yield page
        if not page.has_next:
            return
        page = query.filter(id__gt=page.items[-1].id).all()


def run_migration(batch_size: int = 100, dry_run: bool = False):
    with domain.domain_context():
        total = domain.repository_for(Customer).query.all().total
        logger.info(f"Found {total} customers to migrate")

        if dry_run:
            logger.info("Dry run mode -- no changes will be made")
            return

        migrated = 0
        errors = 0
        start_time = time.time()

        # Wrap the entire migration in a LOW priority context.
        # Every command processed inside this block will produce
        # events tagged with priority=-50, routed to backfill.
        with processing_priority(Priority.LOW):
            for page in customers_in_batches(batch_size):
                for customer in page.items:
                    try:
                        tier = determine_tier(customer.lifetime_spend)
                        domain.process(
                            UpdateCustomerTier(
                                customer_id=customer.id,
                                loyalty_tier=tier,
                            )
                        )
                        migrated += 1
                    except Exception as exc:
                        logger.error(f"Failed to migrate customer {customer.id}: {exc}")
                        errors += 1

                elapsed = time.time() - start_time
                rate = migrated / elapsed if elapsed > 0 else 0
                logger.info(
                    f"Progress: {migrated + errors}/{total} "
                    f"({rate:.0f} records/sec, "
                    f"{migrated} migrated, {errors} errors)"
                )

        elapsed = time.time() - start_time
        logger.info(
            f"Migration complete: {migrated}/{total} migrated, "
            f"{errors} errors, {elapsed:.1f}s elapsed"
        )


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
    )
    dry_run = "--dry-run" in sys.argv
    run_migration(dry_run=dry_run)

Running the Script

Run a dry run first to verify the migration logic:

python migrate_loyalty_tiers.py --dry-run

Then run the migration while production is running:

python migrate_loyalty_tiers.py

Key Points

processing_priority(Priority.LOW) wraps all domain.process() calls. Every event produced inside the block is tagged with priority=-50 in the outbox record.
The OutboxProcessor reads these outbox records and checks message.priority < threshold (default threshold is 0). Since -50 < 0, the events are published to identity::customer:backfill instead of identity::customer.
The StreamSubscription first does a non-blocking read on identity::customer (primary). Only when the primary stream is empty does it read identity::customer:backfill, in one blocking read that waits on both streams. Production events are always processed first.
No handler changes needed: A projector processes events identically regardless of which lane they arrived on.

Monitoring During the Migration

Watch the consumer groups on both streams to track progress:

# Primary -- lag should stay near 0 during migration
redis-cli XINFO GROUPS identity::customer

# Backfill -- lag grows during migration, then drains
redis-cli XINFO GROUPS identity::customer:backfill

lag (Redis 7 and later) is the number of entries the group has not read yet. pending is the number it has read but not acknowledged. When both are 0 for every group on the backfill stream, every event published to it has been handled. XLEN does not tell you this: it counts every entry still stored in the stream, read or not.

Zero lag does not mean the migration succeeded. Outbox records may still be waiting to be published, and a message whose handler failed is acknowledged once it moves to the dead-letter stream (identity::customer:backfill:dlq). Treat the migration as done only when the outbox has no unpublished records and the dead-letter stream holds no migration events.

If you are running the Observatory, outbox.published traces show each migration event as it is published. Their stream field is the stream category (identity::customer), not the backfill stream, so use XINFO GROUPS above to confirm which lane the events went to.

Anti-Patterns

Running migrations directly against the database

# fragment
# BAD: Bypasses the domain model entirely
db.execute("UPDATE customers SET loyalty_tier = 'gold' WHERE ...")

This skips domain events, projections, and any downstream side effects. The read model will not be updated. Event-sourced aggregates will have no record of the change. Direct database writes work for truly schema-only changes, but any migration that changes domain state should go through domain.process().

Running migrations during a maintenance window "because it's simpler"

Taking the system offline to run a migration avoids the priority lane problem but creates a different one: downtime. For systems that need to stay available, priority lanes let you run migrations during normal operation. The added complexity is minimal, a single with processing_priority(Priority.LOW): wrapper.

Parallelizing migration scripts without coordination

Running multiple migration script processes simultaneously can cause contention on the same aggregates. If two processes try to update the same customer concurrently, one will fail with an optimistic concurrency error. Run migration scripts sequentially, or partition the data so each process handles a disjoint subset.

When Not to Use

Small datasets: If the migration processes fewer events than the Engine can handle in a few seconds, priority lanes add no value. run the script and let the events flow through the normal pipeline.
Schema-only migrations: If you are adding a column, changing an index, or altering table structure without changing domain state, use database migration tools directly. Priority lanes are for domain-level data changes that produce events.
Migrations requiring strict ordering with production: If migration events must be interleaved with production events in a specific order, priority lanes will break that ordering. The two-lane system intentionally allows production events to jump ahead of migration events.

Summary

Aspect	Without Priority Lanes	With Priority Lanes
Migration events	Share the production stream	Routed to a separate backfill stream
Production latency	Degrades proportionally to migration volume	Unaffected
Handler changes	None	None
Script changes	None	Wrap in `processing_priority(Priority.LOW)`
Monitoring	One stream to watch	Two streams to watch
Ordering	Global FIFO	Per-lane FIFO; production events processed first