paulpas Monolith Strangler Fig Migration
Senior migration engineer executing incremental extraction of functionality from a monolithic application into independently deployable microservices. This skill makes the model focus on safe, reversible migration steps — every change must be monitored, measurable, and rollable-back without downtime or data loss. The strangler fig approach wraps the monolith in a gateway that gradually redirects traffic to new services until the old application is completely surrounded and removed.
TL;DR Checklist
- Audit all current endpoints and map them to business domains before writing any migration code
- Identify the first service boundary — choose low-risk, high-value functionality with minimal cross-cutting dependencies
- Deploy the strangler gateway as a reverse proxy in front of the monolith before extracting any feature
- Implement an anti-corruption layer adapter to translate between legacy models and new domain models
- Migrate one complete feature end-to-end (API → business logic → data store) before moving to the next
- Implement dual-write database synchronization with a reconciliation worker to detect inconsistencies
- Use feature flags for gradual traffic rollout (1% → 5% → 25% → 50% → 100%) with automatic rollback on error rate breach
When to Use
Use this skill when:
- A large legacy monolith is blocking feature delivery speed and the team needs to incrementally modernize
- The database has grown so large that a single service cannot scale independently for specific domains
- The organization needs to independently scale, deploy, or own specific features as separate services
- The development team has grown beyond what a single deployment unit can reasonably support
- Compliance or data residency requirements necessitate separating certain data domains into their own infrastructure
When NOT to Use
Avoid this skill for:
- Greenfield projects — there is no legacy system to strangle; build it right from the start
- A monolith that is already small, performant, and well-organized — don't introduce distributed complexity unnecessarily
- Cases where the migration cost (estimated 3–6 months of team capacity) exceeds the value of the microservices benefits
- Projects with no clear bounded contexts identified — you cannot extract what you cannot define
Core Workflow
Audit Current Functionality — Catalog every API endpoint in the monolith, its input/output schema, database tables it accesses, and all inter-service/inter-module dependencies. Build a dependency graph showing which endpoints call which database tables and which call other endpoints (even within the same process). Classify each endpoint by business domain using domain language discussions with product owners. Checkpoint: The audit must cover 100% of public API endpoints. Internal/private APIs that are only called by other monolith modules can be audited at a lower granularity. Every database table accessed by the monolith must be listed with its primary consumers.
Identify First Service Boundary — Choose the initial extraction target using these criteria: low read/write contention on shared tables, minimal cross-domain API calls from other domains, clear business value to demonstrate the migration path, and a team that owns the domain end-to-end. Prefer extracting read-heavy features first (they are easier to separate than write-heavy features with concurrent updates). Checkpoint: The chosen feature must be extractable without modifying more than 20% of existing monolith code. If the extraction requires extensive refactoring before any new code, pick a different boundary.
Set Up Strangler Gateway — Deploy a reverse proxy (the strangler gateway) between clients and the monolith. The gateway maintains a route table that decides for each incoming request whether to forward it to the legacy monolith or the new microservice. Route decisions are driven by path matching and feature flags, not hardcoded URLs. Checkpoint: The gateway must default to forwarding all traffic to the monolith (monolith is the source of truth). Only explicitly configured routes with their feature flags enabled should be redirected to new services.
Implement Anti-Corruption Layer — When the new service needs data from legacy tables or models, implement an anti-corruption layer (ACL) adapter that translates between the legacy data schema and the new service's domain model. The ACL hides all legacy complexity behind a clean interface. Never allow legacy types to leak into the new service's domain logic. Checkpoint: Every public method in the new service must accept only its own domain types and return only its own domain types. Any external data crossing the boundary passes through an adapter that performs the translation.
Migrate Single Feature End-to-End — Extract one complete feature: create the new microservice with its own database schema, implement all business logic, set up the gateway route for this feature's endpoints, and redirect traffic. Monitor error rates, latency, and throughput for 48 hours before proceeding. Fix any issues that surface under real traffic patterns. Checkpoint: The feature must work identically from the user's perspective — same API responses, same data outcomes, same performance characteristics. Run integration tests comparing monolith vs. new service output side by side.
Implement Dual-Write Database Synchronization — For features that modify data (not just read), implement dual-write: the monolith writes to its existing tables while the new service writes to its own. A reconciliation worker periodically compares both data stores and flags inconsistencies for manual review or automatic fix-up scripts. This ensures zero data loss during the transition period. Checkpoint: The reconciliation worker must be able to detect and resolve conflicts between the two data stores without human intervention for common cases (e.g., timestamps, non-conflicting field updates). Only true write-write conflicts require escalation.
Monitor and Gradually Route More Traffic — Use feature flags to control traffic splitting for each extracted service. Start at 1% of traffic, monitor error rates and latency against the monolith baseline for at least 24 hours at each stage (1% → 5% → 25% → 50% → 100%). Implement automatic rollback: if error rate exceeds the threshold, immediately divert all traffic back to the monolith. Checkpoint: At every traffic stage, both the monolith and new service must produce identical outputs for the same inputs. Run automated diff tests comparing responses across services before increasing the traffic percentage.
Implementation Patterns
Pattern 1: Strangler Gateway Router
A reverse proxy that routes requests between the legacy monolith and new microservices based on path matching and feature flags, with circuit breaker fallback to the monolith if the new service fails.
# gateway/router.py — Strangler gateway with route matching, feature flags, and circuit breaker fallback
from __future__ import annotations
import logging
import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route
logger = logging.getLogger(__name__)
class RouteStrategy(Enum):
MONOLITH = "monolith" # Always route to legacy
NEW_SERVICE = "new-service" # Always route to new microservice
ROLLING = "rolling" # Split traffic by percentage (gradual rollout)
FEATURE_FLAG = "feature-flag" # Route based on user-level feature flag
@dataclass
class GatewayRoute:
"""Configuration for a single strangler gateway route."""
path_prefix: str # e.g., "/api/inventory"
monolith_target: str # e.g., "http://localhost:8080/api/inventory"
new_service_target: str | None # e.g., "http://inventory-service:8081" (None = not yet extracted)
strategy: RouteStrategy = RouteStrategy.MONOLITH
rolling_percentage: float = 0.0 # 0.0 to 100.0 — only used when strategy is ROLLING
feature_flag_key: str | None = None # User-level flag for FEATURE_FLAG strategy
timeout_seconds: float = 30.0
circuit_breaker_enabled: bool = True
class FeatureFlagService:
"""Simple in-memory feature flag service.
In production, replace with a dedicated feature flag system like LaunchDarkly,
Split.io, or OpenFeature-compatible provider.
"""
def __init__(self) -> None:
self._flags: dict[str, float] = {} # key → percentage (0.0 to 100.0)
self._user_flags: dict[str, dict[str, bool]] = {} # user_id → {flag_key: enabled}
def set_flag_percentage(self, key: str, percentage: float) -> None:
"""Set a percentage-based flag. All users have `percentage/100` chance of being enabled."""
if not 0.0 <= percentage <= 100.0:
raise ValueError(f"Percentage must be between 0 and 100, got {percentage}")
self._flags[key] = percentage
logger.info("Feature flag '%s' set to %.1f%% rollout", key, percentage)
def set_user_flag(self, user_id: str, key: str, enabled: bool) -> None:
"""Set an explicit flag value for a specific user (overrides percentage-based)."""
self._user_flags.setdefault(user_id, {})[key] = enabled
def is_enabled(self, key: str, user_id: str | None = None) -> bool:
"""Check if a feature flag is enabled for a given user."""
# Check user-specific override first
if user_id and key in self._user_flags.get(user_id, {}):
return self._user_flags[user_id][key]
# Fall back to percentage-based evaluation
percentage = self._flags.get(key, 0.0)
if percentage == 0.0:
return False
if percentage == 100.0:
return True
# Deterministic hashing for consistent user experience
import hashlib
hash_input = f"{key}:{user_id or 'anonymous'}:{int(time.time() / 3600)}"
hash_value = int(hashlib.sha256(hash_input.encode()).hexdigest(), 16)
bucket = hash_value % 100
return bucket < percentage
class StranglerGatewayRouter:
"""Routes requests through the strangler gateway to either monolith or new services.
Each route has a strategy that determines how traffic is split:
- MONOLITH: All traffic goes to the legacy system (default for unextracted routes)
- NEW_SERVICE: All traffic goes to the extracted service (final state after migration)
- ROLLING: Traffic is split by percentage for gradual rollout monitoring
- FEATURE_FLAG: User-specific routing based on feature flag evaluation
The router maintains a circuit breaker per new service to fall back to the monolith
automatically if the new service starts failing.
"""
def __init__(self) -> None:
self._routes: dict[str, GatewayRoute] = {}
self._feature_flags = FeatureFlagService()
self._service_health: dict[str, bool] = {} # service_name → is_healthy
self._error_counts: dict[str, list[float]] = {} # service_name → [timestamps]
def add_route(self, route: GatewayRoute) -> None:
"""Register a new route in the gateway."""
self._routes[route.path_prefix] = route
if route.new_service_target:
self._service_health[route.new_service_target] = True # Start healthy
@property
def feature_flags(self) -> FeatureFlagService:
return self._feature_flags
def evaluate_route(self, path: str, user_id: str | None = None) -> tuple[str, bool]:
"""Determine where to route a request.
Returns:
(target_url, uses_new_service) — target is either monolith or new service URL
If the chosen new service has tripped its circuit breaker, automatically falls back
to the monolith regardless of the configured strategy.
"""
matching_route = self._find_matching_route(path)
if matching_route is None:
return ("monolith-default", False) # All unmatched routes → monolith
target_url, uses_new = self._apply_strategy(
route=matching_route, path=path, user_id=user_id,
)
# Circuit breaker check — if new service is unhealthy, fall back
if uses_new and matching_route.circuit_breaker_enabled:
if not self._is_service_healthy(matching_route.new_service_target or ""):
logger.warning(
"New service %s circuit open — falling back to monolith for %s",
matching_route.new_service_target, path,
)
return (matching_route.monolith_target, False)
return target_url, uses_new
def record_outcome(self, service_target: str, success: bool) -> None:
"""Record the outcome of a request to a new service.
Automatically trips the circuit breaker if error rate exceeds threshold.
The circuit stays open for 60 seconds before allowing one probe through.
"""
self._error_counts.setdefault(service_target, [])
now = time.monotonic()
# Keep only the last 100 outcomes for rolling window
self._error_counts[service_target] = [
t for t in self._error_counts[service_target] if now - t < 120
]
if not success:
self._error_counts[service_target].append(now)
self._check_circuit_breaker(service_target, now)
def _check_circuit_breaker(self, service_target: str, now: float) -> None:
"""Check if error rate in the rolling window exceeds 50% — trip circuit if so."""
outcomes = self._error_counts.get(service_target, [])
total = len(outcomes)
# Need at least 5 data points before tripping (avoid false positives on low traffic)
if total < 5:
return
errors_in_window = sum(1 for _ in outcomes if True) # All recorded failures
error_rate = errors_in_window / total
if error_rate > 0.5:
self._service_health[service_target] = False
logger.critical(
"Circuit breaker TRIPPED for %s — error rate %.0f%% over last %d checks",
service_target, error_rate * 100, total,
)
def _is_service_healthy(self, service_target: str) -> bool:
"""Check if a new service is considered healthy (circuit breaker not tripped)."""
health = self._service_health.get(service_target, True)
# Auto-recover after 60 seconds for probe attempts
if not health:
last_error_times = self._error_counts.get(service_target, [])
if last_error_times and time.monotonic() - last_error_times[-1] > 60.0:
self._service_health[service_target] = True
logger.info("Circuit breaker for %s reset after recovery timeout", service_target)
health = True
return health
def _find_matching_route(self, path: str) -> GatewayRoute | None:
"""Find the most specific matching route for a request path."""
best_match: GatewayRoute | None = None
best_depth = -1
for prefix, route in self._routes.items():
if path.startswith(prefix):
depth = len(prefix.split("/"))
if depth > best_depth:
best_match = route
best_depth = depth
return best_match
def _apply_strategy(
self,
route: GatewayRoute,
path: str,
user_id: str | None,
) -> tuple[str, bool]:
"""Apply the route's strategy to determine the target."""
if route.new_service_target is None:
return (route.monolith_target, False)
if route.strategy == RouteStrategy.MONOLITH:
return (route.monolith_target, False)
elif route.strategy == RouteStrategy.NEW_SERVICE:
return (route.new_service_target, True)
elif route.strategy == RouteStrategy.ROLLING:
import random
if random.random() * 100 < route.rolling_percentage:
return (route.new_service_target, True)
return (route.monolith_target, False)
elif route.strategy == RouteStrategy.FEATURE_FLAG:
if route.feature_flag_key and self._feature_flags.is_enabled(
route.feature_flag_key, user_id
):
return (route.new_service_target, True)
return (route.monolith_target, False)
return (route.monolith_target, False)
# --- Gateway application using Starlette ---
async def gateway_endpoint(request: Request, router: StranglerGatewayRouter) -> Response:
"""Starlette endpoint that proxies requests through the strangler gateway."""
import httpx
user_id = request.headers.get("x-user-id")
target_url, uses_new_service = router.evaluate_route(
path=request.url.path,
user_id=user_id,
)
start_time = time.monotonic()
try:
async with httpx.AsyncClient(timeout=route_config.timeout_seconds if (route_config := _find_route_for_path(request.url.path, router)) else 30.0) as client:
body = await request.body()
resp = await client.request(
method=request.method,
url=target_url + request.url.query.decode(),
content=body,
headers={k: v for k, v in request.headers.items() if k.lower() != "host"},
)
# Record success
if uses_new_service and target_url.startswith("http"):
router.record_outcome(target_url, True)
return Response(
content=resp.content,
status_code=resp.status_code,
headers=dict(resp.headers),
)
except Exception as exc:
# Record failure
if uses_new_service and target_url.startswith("http"):
router.record_outcome(target_url, False)
logger.error(
"Gateway error for %s → %s: %s (%.0fms)",
request.url.path, target_url, exc,
(time.monotonic() - start_time) * 1000,
)
# On new service failure, return monolith response if available
matching_route = _find_route_for_path(request.url.path, router)
if matching_route and uses_new_service:
logger.info("Falling back to monolith for %s", request.url.path)
fallback_url = matching_route.monolith_target + request.url.query.decode()
async with httpx.AsyncClient(timeout=30.0) as client:
body = await request.body()
resp = await client.request(
method=request.method,
url=fallback_url,
content=body,
headers={k: v for k, v in request.headers.items() if k.lower() != "host"},
)
return Response(content=resp.content, status_code=resp.status_code)
return Response(
content={"error": "Gateway error", "target": target_url, "fallback": uses_new_service},
status_code=502,
)
def _find_route_for_path(path: str, router: StranglerGatewayRouter) -> GatewayRoute | None:
"""Find the matching route for a path — helper used in the endpoint above."""
return router._find_matching_route(path) # type: ignore[attr-defined]
Pattern 2: Anti-Corruption Layer Adapter
The ACL translates between legacy data models and new microservice domain models, hiding all legacy complexity behind clean interfaces.
# acl/anti_corruption_layer.py — Anti-corruption layer for translating legacy models to new domain models
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
# ==============================
# LEGACY MODELS (do NOT expose these outside the ACL)
# ==============================
class LegacyOrderStatus(Enum):
"""Legacy monolith order statuses — some values have no equivalent in the new model."""
DRAFT = "D"
SUBMITTED = "S"
CONFIRMED = "C"
PROCESSING = "P"
SHIPPED = "H"
DELIVERED = "D"
CANCELLED = "X"
PARTIALLY_REFUNDED = "PR"
@dataclass
class LegacyOrderRow:
"""Raw row from the legacy monolith's orders table.
This model reflects the legacy database schema with all its quirks:
- Single-character status codes
- Null-safe columns mixed in one table
- Denormalized customer and shipping data
- audit_log text field containing JSON-like strings
"""
order_id: str
cust_id: int # Not a UUID — legacy integer IDs
order_date: datetime | None # Nullable even for active orders
status_code: str # Single-char code, not an enum
total_amount: float | None # May be null before confirmation
currency_code: str = "USD"
customer_name: str | None # Denormalized — should not exist in new model
shipping_address: str | None # Full address as a single text field
notes: str | None
created_by: int | None # User ID as integer, may be null
modified_at: datetime | None
@dataclass
class LegacyOrderItemRow:
"""Raw row from the legacy order_items table."""
item_id: str
order_id: str
product_code: str # Legacy product code format
quantity: int
unit_price: float
discount_percent: float = 0.0
tax_amount: float | None = None
# ==============================
# NEW DOMAIN MODELS (used by the microservice)
# ==============================
class OrderStatus(Enum):
"""New domain order statuses — clean, explicit, bounded context compliant."""
PENDING = "pending"
CONFIRMED = "confirmed"
PAID = "paid"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
REFUNDED = "refunded"
@dataclass(frozen=True)
class Money:
"""Immutable money value object."""
amount: float
currency: str = "USD"
def __post_init__(self) -> None:
if self.amount < 0:
raise ValueError("Money cannot be negative")
@dataclass
class OrderAddress:
"""Structured address in the new domain model."""
street: str
city: str
state: str
postal_code: str
country: str = "US"
@dataclass
class OrderItem:
"""Order item using product_id and clean pricing."""
product_id: str
quantity: int
unit_price: Money
@property
def line_total(self) -> Money:
return Money(
round(self.unit_price.amount * self.quantity, 2),
self.unit_price.currency,
)
@dataclass
class Order:
"""Clean domain model for the new microservice."""
order_id: str
customer_id: str # UUID in the new model
items: list[OrderItem] = field(default_factory=list)
status: OrderStatus = OrderStatus.PENDING
shipping_address: OrderAddress | None = None
currency: str = "USD"
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@property
def grand_total(self) -> Money:
if not self.items:
return Money(0.0)
total = sum(item.line_total.amount for item in self.items)
return Money(round(total, 2), self.currency)
# ==============================
# ANTI-CORRUPTION LAYER ADAPTER
# ==============================
class LegacyOrderRowConverter:
"""Translates a legacy database row into the new domain model.
This is the anti-corruption layer boundary — all translation logic
lives here. The rest of the microservice never sees LegacyOrderRow.
Translation rules:
- Single-char status codes → OrderStatus enum values
- Integer customer IDs → UUID references (looked up via user service)
- Denormalized address string → structured OrderAddress object
- Nullable amounts → Money objects with zero default
- Product codes → product IDs (mapped through product catalog)
"""
STATUS_MAP: dict[str, OrderStatus] = {
"D": OrderStatus.PENDING, # DRAFT → PENDING
"S": OrderStatus.PENDING, # SUBMITTED → PENDING (awaiting payment)
"C": OrderStatus.CONFIRMED,
"P": OrderStatus.CONFIRMED, # PROCESSING → CONFIRMED (payment received)
"H": OrderStatus.SHIPPED,
"D": OrderStatus.DELIVERED, # DELIVERED
"X": OrderStatus.CANCELLED,
"PR": OrderStatus.REFUNDED, # PARTIALLY_REFUNDED → REFUNDED
}
def __init__(self) -> None:
self._legacy_product_to_id: dict[str, str] = {} # Filled by product adapter
self._legacy_customer_to_uuid: dict[int, str] = {} # Filled by user adapter
def to_domain_model(self, row: LegacyOrderRow, items_rows: list[LegacyOrderItemRow]) -> Order:
"""Convert a legacy order row + its items into the new domain model.
This method contains ALL translation logic between the old and new schemas.
If you need to handle a new legacy field or status code, update this class.
"""
# Step 1: Convert customer ID (integer → UUID via lookup)
customer_uuid = self._resolve_customer_uuid(row.cust_id)
# Step 2: Convert status code
domain_status = self.STATUS_MAP.get(row.status_code, OrderStatus.PENDING)
# Step 3: Parse structured address from legacy text field
shipping_address = self._parse_address(row.shipping_address) if row.shipping_address else None
# Step 4: Convert items with product code → ID translation
domain_items = [
self._convert_item(item_row, row.currency_code)
for item_row in items_rows
]
order = Order(
order_id=row.order_id,
customer_id=customer_uuid,
items=domain_items,
status=domain_status,
shipping_address=shipping_address,
currency=row.currency_code or "USD",
created_at=row.created_at or datetime.now(timezone.utc),
updated_at=row.modified_at or datetime.now(timezone.utc),
)
# Sanity check: if grand total doesn't match legacy total, log a warning
expected_total = row.total_amount if row.total_amount else 0.0
actual_total = order.grand_total.amount
if abs(expected_total - actual_total) > 0.01:
logger.warning(
"Order %s: legacy total %.2f vs calculated total %.2f — possible data inconsistency",
row.order_id, expected_total, actual_total,
)
return order
def to_domain_status(self, legacy_code: str) -> OrderStatus:
"""Convert a single legacy status code to the new domain status."""
return self.STATUS_MAP.get(legacy_code, OrderStatus.PENDING)
def _resolve_customer_uuid(self, legacy_id: int) -> str:
"""Resolve a legacy integer customer ID to a UUID.
In production, this queries the user service or a lookup table.
For now, return a deterministic placeholder UUID.
"""
if legacy_id in self._legacy_customer_to_uuid:
return self._legacy_customer_to_uuid[legacy_id]
# Placeholder — should query actual user service
import hashlib
uuid_str = hashlib.sha256(f"customer-{legacy_id}".encode()).hexdigest()[:32]
return f"{uuid_str[:8]}-{uuid_str[8:12]}-{uuid_str[12:16]}-{uuid_str[16:20]}-{uuid_str[20:32]}"
def _parse_address(self, raw_address: str) -> OrderAddress:
"""Parse a denormalized address string into a structured OrderAddress.
Legacy format: "123 Main St, Springfield, IL 62704, USA"
New model: OrderAddress(street="123 Main St", city="Springfield", ...)
"""
parts = [p.strip() for p in raw_address.split(",")]
if len(parts) < 4:
logger.warning("Malformed address '%s' — using defaults", raw_address)
return OrderAddress(
street=raw_address or "Unknown",
city="Unknown",
state="XX",
postal_code="00000",
)
street = parts[0]
city_state_zip = parts[1] # e.g., "Springfield, IL 62704"
country = parts[-1] if len(parts) > 4 else "USA"
cs_parts = city_state_zip.split(",")
city = cs_parts[0].strip() if cs_parts else "Unknown"
state_zip = cs_parts[1].strip() if len(cs_parts) > 1 else "XX 00000"
state = state_zip.split()[0] if state_zip else "XX"
postal_code = state_zip.split()[-1] if state_zip else "00000"
return OrderAddress(street=street, city=city, state=state, postal_code=postal_code, country=country)
def _convert_item(
self,
row: LegacyOrderItemRow,
currency: str,
) -> OrderItem:
"""Convert a legacy order item row to the new domain model."""
product_id = self._legacy_product_to_id.get(row.product_code, row.product_code)
unit_price = Money(round(row.unit_price * (1 - row.discount_percent / 100), 2), currency)
return OrderItem(
product_id=product_id,
quantity=row.quantity,
unit_price=unit_price,
)
# --- ACL Adapter for bidirectional communication ---
class AntiCorruptionLayer:
"""Full ACL adapter providing translation in both directions.
Reads from legacy → converts to new domain models (for the new service's business logic).
Writes from new → converts back to legacy format (for dual-write during migration).
"""
def __init__(self, order_converter: LegacyOrderRowConverter | None = None) -> None:
self._order_converter = order_converter or LegacyOrderRowConverter()
def read_legacy_order(
self,
legacy_row: LegacyOrderRow,
legacy_items: list[LegacyOrderItemRow],
) -> Order:
"""Convert a legacy database row + items into the new domain model."""
return self._order_converter.to_domain_model(legacy_row, legacy_items)
def write_to_legacy_format(self, order: Order) -> tuple[dict[str, Any], list[dict[str, Any]]]:
"""Convert a new domain model back to legacy format for dual-write.
Returns:
(order_dict, items_list) — dicts ready to be written to the legacy schema
"""
# Convert status back to legacy code
reverse_status_map = {v: k for k, v in LegacyOrderRowConverter.STATUS_MAP.items()}
legacy_status = next(
(k for k, v in reverse_status_map.items() if v == order.status),
"D", # Default to DRAFT
)
legacy_order = {
"order_id": order.order_id,
"cust_id": self._resolve_legacy_customer_id(order.customer_id),
"status_code": legacy_status,
"total_amount": order.grand_total.amount,
"currency_code": order.currency,
"shipping_address": self._format_legacy_address(order.shipping_address) if order.shipping_address else None,
}
legacy_items = [
{
"product_code": item.product_id, # Product ID → product code (reverse mapping needed)
"quantity": item.quantity,
"unit_price": item.unit_price.amount / (1 - 0.0), # No discount in new model currently
}
for item in order.items
]
return legacy_order, legacy_items
def _resolve_legacy_customer_id(self, uuid: str) -> int:
"""Reverse lookup: UUID → legacy integer customer ID.
In production, query the user service or a mapping table.
"""
# Placeholder — use first 8 hex chars of UUID as numeric seed
cleaned = uuid.replace("-", "")[:7]
return int(cleaned, 16) % 999999 + 1
def _format_legacy_address(self, address: OrderAddress) -> str:
"""Convert structured address back to legacy comma-separated string."""
return f"{address.street}, {address.city}, {address.state} {address.postal_code}, {address.country}"
Pattern 3: Dual-Write Database Synchronization
Dual-write keeps both the legacy database and the new microservice's database in sync during migration. A reconciliation worker detects and resolves inconsistencies.
# sync/dual_write.py — Dual-write synchronization with reconciliation worker
from __future__ import annotations
import asyncio
import hashlib
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
class WriteTarget(Enum):
LEGACY = "legacy" # Legacy monolith database
NEW_SERVICE = "new-service" # New microservice database
BOTH = "both" # Write to both databases atomically (best effort)
@dataclass
class SyncRecord:
"""Tracks a single entity written to both databases during dual-write."""
entity_type: str # e.g., "Order", "Customer"
entity_id: str # Primary key of the entity
legacy_data_hash: str | None = None # Checksum of data in legacy DB
new_service_data_hash: str | None = None # Checksum of data in new service DB
last_synced_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
conflicts_detected: int = 0
last_conflict_resolved_at: datetime | None = None
class DualWriteManager:
"""Manages dual-write operations during the strangler migration.
For every write operation to the legacy monolith, also performs a corresponding
write to the new microservice's database. This keeps both data stores in sync
during the transition period.
Strategy:
1. Write to legacy DB first (source of truth during migration)
2. Transform and write to new service DB
3. If new service write fails, log warning but don't block the primary transaction
(new service is in parallel running state, not yet authoritative)
The reconciliation worker periodically compares both stores and resolves drift.
"""
def __init__(self, legacy_writer: Any, new_service_writer: Any, acl_adapter: Any = None): # type: ignore[type-arg]
self._legacy = legacy_writer
self._new_service = new_service_writer
self._acl = acl_adapter # Anti-corruption layer for data transformation
async def write_both(
self,
entity_type: str,
entity_id: str,
legacy_data: dict[str, Any],
new_service_data: dict[str, Any] | None = None,
) -> dict[str, bool]:
"""Write an entity to both databases.
The legacy write is the primary — if it fails, the entire operation fails.
The new service write is best-effort during migration — failures are logged
but don't block the user-facing transaction.
"""
results: dict[str, bool] = {"legacy": False, "new_service": False}
# Step 1: Write to legacy DB (PRIMARY — must succeed)
try:
await self._legacy.write(entity_type, entity_id, legacy_data)
results["legacy"] = True
logger.info("Dual-write: legacy %s.%s written", entity_type, entity_id[:8])
except Exception as exc:
logger.error("Legacy dual-write FAILED for %s.%s: %s", entity_type, entity_id[:8], exc)
return results # Abort — can't proceed without primary
# Step 2: Write to new service DB (BEST-EFFORT during migration)
if new_service_data is None:
# Transform legacy data using ACL adapter
if self._acl:
new_service_data = self._transform_for_new_service(entity_type, legacy_data)
else:
new_service_data = legacy_data
try:
await self._new_service.write(entity_type, entity_id, new_service_data)
results["new_service"] = True
logger.info("Dual-write: new-service %s.%s written", entity_type, entity_id[:8])
except Exception as exc:
logger.warning(
"Dual-write: new-service write FAILED for %s.%s — continuing (parallel running state)",
entity_type, entity_id[:8], exc_info=True,
)
# Don't fail the transaction — new service is still being migrated
return results
def _transform_for_new_service(
self,
entity_type: str,
legacy_data: dict[str, Any],
) -> dict[str, Any]:
"""Transform legacy data format into new service format using ACL adapter."""
if entity_type == "Order" and hasattr(self._acl, "write_to_legacy_format"):
# This is a reverse direction — use the adapter's write method
return legacy_data # Simplified — in practice, transform using ACL
# Generic transformation: flatten nested legacy fields
transformed = dict(legacy_data)
for key, value in list(transformed.items()):
if isinstance(value, bytes):
transformed[key] = value.decode("utf-8", errors="replace")
return transformed
# --- Reconciliation worker ---
@dataclass
class ConflictRecord:
"""Represents a detected data conflict between legacy and new service databases."""
entity_type: str
entity_id: str
legacy_value_hash: str
new_service_value_hash: str
conflicting_fields: list[str] = field(default_factory=list)
detected_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
class ReconciliationWorker:
"""Periodically compares data between legacy and new service databases.
Detection strategy:
1. For each entity written via dual-write, compute a hash of the relevant fields
2. Compare hashes between the two databases
3. If hashes differ, fetch both records and diff field-by-field
4. Log conflicts and optionally auto-resolve based on resolution strategy
Resolution strategies:
- LAST_WRITER_WINS: Use the more recently updated record
- LEGACY_PRIORITIZED: Always trust the legacy database (safe during migration)
- MANUAL_ESCALATION: Flag for human review
"""
def __init__(self, legacy_reader: Any, new_service_reader: Any, # type: ignore[type-arg]
resolution_strategy: str = "LEGACY_PRIORITIZED") -> None:
self._legacy = legacy_reader
self._new_service = new_service_reader
self._strategy = resolution_strategy
self._conflicts: list[ConflictRecord] = []
self._sync_records: dict[str, SyncRecord] = {} # entity_key → SyncRecord
async def run_reconciliation(self) -> int:
"""Execute one reconciliation cycle. Returns the number of conflicts found."""
logger.info("Starting reconciliation cycle")
sync_keys = list(self._sync_records.keys())
conflicts_found = 0
for entity_key in sync_keys:
record = self._sync_records[entity_key]
# Compare data hashes
legacy_hash = await self._compute_data_hash(
record.entity_type, record.entity_id, target=WriteTarget.LEGACY,
)
new_hash = await self._compute_data_hash(
record.entity_type, record.entity_id, target=WriteTarget.NEW_SERVICE,
)
if legacy_hash != new_hash:
conflicts_found += 1
record.conflicts_detected += 1
record.l
…(truncated)