Zero-Downtime Database Migration: A Practical Guide

class ZeroDowntimeMigrationPlanner:
    def __init__(self):
        self.migration_patterns = {
            'dual_write': {
                'complexity': 'medium',
                'consistency_model': 'eventual',
                'rollback_speed': 'fast',
                'application_changes': 'required'
            },
            'change_data_capture': {
                'complexity': 'low',
                'consistency_model': 'strong',
                'rollback_speed': 'medium',
                'application_changes': 'minimal'
            },
            'logical_replication': {
                'complexity': 'low',
                'consistency_model': 'strong',
                'rollback_speed': 'fast',
                'application_changes': 'none'
            },
            'blue_green_deployment': {
                'complexity': 'high',
                'consistency_model': 'strong',
                'rollback_speed': 'instant',
                'application_changes': 'configuration'
            }
        }

    def assess_migration_feasibility(self, system_requirements):
        """Assess feasibility for zero-downtime migration"""
        feasibility_score = 100
        requirements_analysis = {}
        
        # Database technology compatibility
        source_db = system_requirements['source_database']
        target_db = system_requirements['target_database']
        compatibility = self.assess_database_compatibility(source_db, target_db)
        requirements_analysis['compatibility'] = compatibility
        
        if compatibility['compatibility_score'] < 70:
            feasibility_score -= 30
            
        return {
            'feasibility_score': max(0, feasibility_score),
            'feasibility_level': self.get_feasibility_level(feasibility_score),
            'requirements_analysis': requirements_analysis,
            'recommended_pattern': recommended_pattern
        }

class CDCMigrationOrchestrator:
    def __init__(self, source_config, target_config):
        self.source_config = source_config
        self.target_config = target_config
        self.cdc_tools = {
            'debezium': {
                'supported_sources': ['postgresql', 'mysql', 'mongodb', 'sqlserver'],
                'latency': 'low',
                'complexity': 'medium',
                'exactly_once': True
            },
            'aws_dms': {
                'supported_sources': ['oracle', 'postgresql', 'mysql', 'sqlserver'],
                'latency': 'medium',
                'complexity': 'low',
                'exactly_once': False
            },
            'kafka_connect': {
                'supported_sources': ['postgresql', 'mysql'],
                'latency': 'low',
                'complexity': 'high',
                'exactly_once': True
            }
        }

    def design_cdc_pipeline(self, migration_requirements):
        """Design CDC pipeline for zero-downtime migration"""
        # Select optimal CDC tool
        optimal_tool = self.select_cdc_tool(migration_requirements)
        
        # Design pipeline architecture
        pipeline_config = {
            'source_connector': self.configure_source_connector(optimal_tool),
            'streaming_platform': self.configure_streaming_platform(optimal_tool),
            'transformation_layer': self.configure_transformations(migration_requirements),
            'target_connector': self.configure_target_connector(optimal_tool),
            'monitoring': self.configure_monitoring(),
            'error_handling': self.configure_error_handling()
        }
        
        return {
            'selected_tool': optimal_tool,
            'pipeline_configuration': pipeline_config,
            'deployment_steps': self.generate_deployment_steps(pipeline_config),
            'validation_procedures': self.generate_validation_procedures()
        }

// Java implementation of dual-write pattern with consistency guarantees
@Component
public class DualWriteMigrationService {
    private final DataSource primaryDataSource;
    private final DataSource migrationDataSource;
    private final TransactionTemplate transactionTemplate;
    private final MigrationStatusService migrationStatusService;
    
    @Autowired
    public DualWriteMigrationService(
            @Qualifier("primary") DataSource primaryDataSource,
            @Qualifier("migration") DataSource migrationDataSource,
            PlatformTransactionManager transactionManager,
            MigrationStatusService migrationStatusService) {
        this.primaryDataSource = primaryDataSource;
        this.migrationDataSource = migrationDataSource;
        this.transactionTemplate = new TransactionTemplate(transactionManager);
        this.migrationStatusService = migrationStatusService;
    }
    
    public void executeWithDualWrite(DualWriteOperation operation) {
        MigrationPhase currentPhase = migrationStatusService.getCurrentPhase();
        switch (currentPhase) {
            case PREPARATION:
                // Write only to primary during preparation
                executePrimaryOnly(operation);
                break;
            case DUAL_WRITE:
                // Write to both primary and migration target
                executeDualWrite(operation);
                break;
            case VALIDATION:
                // Continue dual write with enhanced validation
                executeDualWriteWithValidation(operation);
                break;
            case CUTOVER:
                // Write only to migration target
                executeMigrationOnly(operation);
                break;
            default:
                throw new IllegalStateException("Unknown migration phase: " + currentPhase);
        }
    }
    
    private void executeDualWrite(DualWriteOperation operation) {
        transactionTemplate.execute(status -> {
            try {
                // Execute primary write first (source of truth)
                Object primaryResult = operation.executePrimary(primaryDataSource);
                try {
                    // Execute migration write
                    Object migrationResult = operation.executeMigration(migrationDataSource);
                    // Validate consistency if possible
                    if (operation.supportsConsistencyCheck()) {
                        boolean consistent = operation.validateConsistency(primaryResult, migrationResult);
                        if (!consistent) {
                            // Log inconsistency but don't fail the transaction
                            logConsistencyIssue(operation, primaryResult, migrationResult);
                        }
                    }
                } catch (Exception migrationException) {
                    // Migration write failed - log but don't fail primary transaction
                    logMigrationFailure(operation, migrationException);
                    migrationStatusService.recordFailure(operation.getOperationType(), migrationException);
                }
                return primaryResult;
            } catch (Exception primaryException) {
                // Primary write failed - fail the entire transaction
                status.setRollbackOnly();
                throw new DualWriteException("Primary write failed", primaryException);
            }
        });
    }
    
    private void executeDualWriteWithValidation(DualWriteOperation operation) {
        transactionTemplate.execute(status -> {
            // Execute dual write
            Object primaryResult = operation.executePrimary(primaryDataSource);
            Object migrationResult = operation.executeMigration(migrationDataSource);
            
            // Enhanced validation during validation phase
            ValidationResult validation = performEnhancedValidation(
                operation, primaryResult, migrationResult
            );
            
            if (!validation.isValid()) {
                migrationStatusService.recordValidationFailure(validation);
                // Depending on validation failure severity, might rollback
                if (validation.getSeverity() == ValidationSeverity.CRITICAL) {
                    status.setRollbackOnly();
                    throw new ValidationException("Critical validation failure", validation);
                }
            }
            return primaryResult;
        });
    }
    
    private ValidationResult performEnhancedValidation(
            DualWriteOperation operation, Object primaryResult, Object migrationResult) {
        List<ValidationIssue> issues = new ArrayList<>();
        
        // Data consistency validation
        if (!operation.validateDataConsistency(primaryResult, migrationResult)) {
            issues.add(new ValidationIssue(
                ValidationSeverity.HIGH,
                "Data consistency check failed",
                operation.getOperationType()
            ));
        }
        
        // Performance validation
        long primaryExecutionTime = operation.getPrimaryExecutionTime();
        long migrationExecutionTime = operation.getMigrationExecutionTime();
        if (migrationExecutionTime > primaryExecutionTime * 2) {
            issues.add(new ValidationIssue(
                ValidationSeverity.MEDIUM,
                "Migration performance significantly slower than primary",
                String.format("Primary: %dms, Migration: %dms", primaryExecutionTime, migrationExecutionTime)
            ));
        }
        
        return new ValidationResult(issues);
    }
}

-- PostgreSQL logical replication setup for zero-downtime migration
-- Source database configuration
-- Enable logical replication
ALTER SYSTEM SET wal_level = 'logical';
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;

-- Create publication for all tables
CREATE PUBLICATION migration_publication FOR ALL TABLES;

-- Create replication slot
SELECT pg_create_logical_replication_slot('migration_slot', 'pgoutput');

-- Target database subscription setup
CREATE SUBSCRIPTION migration_subscription
CONNECTION 'host=source-db port=5432 dbname=production user=replicator password=secret'
PUBLICATION migration_publication
WITH (copy_data = true, create_slot = false, slot_name = 'migration_slot');

-- Monitor replication lag
SELECT 
    slot_name,
    plugin,
    slot_type,
    database,
    active,
    active_pid,
    xmin,
    catalog_xmin,
    restart_lsn,
    confirmed_flush_lsn,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as replication_lag
FROM pg_replication_slots
WHERE slot_name = 'migration_slot';

class ConflictResolutionEngine:
    def __init__(self):
        self.resolution_strategies = {
            'timestamp_based': self.resolve_by_timestamp,
            'source_wins': self.resolve_source_wins,
            'target_wins': self.resolve_target_wins,
            'manual_resolution': self.flag_for_manual_resolution,
            'business_rule_based': self.resolve_by_business_rules
        }
        
    def resolve_conflict(self, conflict_type, source_record, target_record, resolution_strategy):
        """Resolve data conflicts during synchronization"""
        if resolution_strategy not in self.resolution_strategies:
            raise ValueError(f"Unknown resolution strategy: {resolution_strategy}")
            
        resolver = self.resolution_strategies[resolution_strategy]
        resolution = resolver(conflict_type, source_record, target_record)
        
        # Log conflict resolution for audit trail
        self.log_conflict_resolution(conflict_type, source_record, target_record, resolution)
        return resolution
        
    def resolve_by_timestamp(self, conflict_type, source_record, target_record):
        """Resolve conflicts using timestamp-based last-writer-wins"""
        source_timestamp = source_record.get('updated_at') or source_record.get('created_at')
        target_timestamp = target_record.get('updated_at') or target_record.get('created_at')
        
        if source_timestamp >= target_timestamp:
            return {
                'action': 'use_source',
                'record': source_record,
                'reason': f'Source timestamp {source_timestamp} >= target timestamp {target_timestamp}'
            }
        else:
            return {
                'action': 'use_target',
                'record': target_record,
                'reason': f'Target timestamp {target_timestamp} > source timestamp {source_timestamp}'
            }
            
    def resolve_by_business_rules(self, conflict_type, source_record, target_record):
        """Resolve conflicts using business-specific rules"""
        # Example business rules for customer data
        if conflict_type == 'customer_update':
            # Priority: email changes from source, address from most recent
            resolved_record = target_record.copy()
            
            # Email updates from source take priority
            if source_record.get('email') != target_record.get('email'):
                resolved_record['email'] = source_record['email']
                resolved_record['email_verified'] = False  # Require re-verification
                
            # Use most recent address information
            source_addr_updated = source_record.get('address_updated_at')
            target_addr_updated = target_record.get('address_updated_at')
            
            if source_addr_updated and target_addr_updated:
                if source_addr_updated > target_addr_updated:
                    resolved_record.update({
                        'address': source_record['address'],
                        'city': source_record['city'],
                        'postal_code': source_record['postal_code'],
                        'address_updated_at': source_record['address_updated_at']
                    })
                    
            return {
                'action': 'use_merged',
                'record': resolved_record,
                'reason': 'Applied business rules for customer data merging'
            }
            
        # Default to timestamp-based resolution for unknown types
        return self.resolve_by_timestamp(conflict_type, source_record, target_record)
        
    def detect_conflicts(self, source_batch, target_batch):
        """Detect conflicts between source and target data batches"""
        conflicts = []
        
        # Create lookup maps
        source_map = {record['id']: record for record in source_batch}
        target_map = {record['id']: record for record in target_batch}
        
        # Find overlapping records
        common_ids = set(source_map.keys()) & set(target_map.keys())
        
        for record_id in common_ids:
            source_record = source_map[record_id]
            target_record = target_map[record_id]
            
            # Compare record content
            if not self.records_equal(source_record, target_record):
                conflict = {
                    'id': record_id,
                    'type': 'data_mismatch',
                    'source_record': source_record,
                    'target_record': target_record,
                    'detected_at': datetime.utcnow().isoformat(),
                    'differences': self.find_differences(source_record, target_record)
                }
                conflicts.append(conflict)
                
        return conflicts
        
    def find_differences(self, record1, record2):
        """Find specific differences between two records"""
        differences = []
        all_keys = set(record1.keys()) | set(record2.keys())
        
        for key in all_keys:
            value1 = record1.get(key)
            value2 = record2.get(key)
            if value1 != value2:
                differences.append({
                    'field': key,
                    'source_value': value1,
                    'target_value': value2
                })
                
        return differences