Zero-Downtime Architecture for Enterprise Systems: A Practical Guide to Always-On Services

-- DON'T DO THIS
ALTER TABLE users RENAME COLUMN user_email TO email_address;

-- Migration file: 20260218_add_email_address_column.sql
ALTER TABLE users ADD COLUMN email_address VARCHAR(255);
CREATE INDEX CONCURRENTLY idx_users_email_address ON users(email_address);

-- Backfill in batches (run separately, not in migration)
-- Update 10k rows every 500ms to avoid replication lag
UPDATE users 
SET email_address = user_email 
WHERE id >= ? AND id < ? AND email_address IS NULL;

// Write to BOTH columns during transition
async function updateUserEmail(userId, newEmail) {
  await db.query(
    'UPDATE users SET user_email = $1, email_address = $1 WHERE id = $2',
    [newEmail, userId]
  );
}

// Read from OLD column (maintains backward compatibility)
async function getUserEmail(userId) {
  const result = await db.query(
    'SELECT user_email FROM users WHERE id = $1',
    [userId]
  );
  return result.rows[0].user_email;
}

// Now reading from NEW column, still writing to both
async function getUserEmail(userId) {
  const result = await db.query(
    'SELECT email_address FROM users WHERE id = $1',
    [userId]
  );
  return result.rows[0].email_address;
}

-- Migration file: 20260225_remove_user_email_column.sql
ALTER TABLE users DROP COLUMN user_email;

// Stop writing to old column
async function updateUserEmail(userId, newEmail) {
  await db.query(
    'UPDATE users SET email_address = $1 WHERE id = $2',
    [newEmail, userId]
  );
}

POST /api/v1/payments
{
  "amount": 5000,
  "currency": "USD",
  "customer_id": "cust_abc123"
}

POST /api/v2/payments
{
  "amount": {
    "value": 5000,
    "currency": "USD"
  },
  "customer": {
    "id": "cust_abc123",
    "email": "user@example.com"
  }
}

// Internal domain model (canonical representation)
class PaymentRequest {
  constructor({ amount, currency, customerId, customerEmail }) {
    this.amount = amount;
    this.currency = currency;
    this.customerId = customerId;
    this.customerEmail = customerEmail;
  }
}

// v1 endpoint (legacy)
app.post('/api/v1/payments', async (req, res) => {
  const payment = new PaymentRequest({
    amount: req.body.amount,
    currency: req.body.currency,
    customerId: req.body.customer_id,
    customerEmail: null // v1 didn't capture this
  });
  
  const result = await processPayment(payment);
  
  // Return v1 response format
  res.json({
    transaction_id: result.transactionId,
    status: result.status
  });
});

// v2 endpoint (modern)
app.post('/api/v2/payments', async (req, res) => {
  const payment = new PaymentRequest({
    amount: req.body.amount.value,
    currency: req.body.amount.currency,
    customerId: req.body.customer.id,
    customerEmail: req.body.customer.email
  });
  
  const result = await processPayment(payment);
  
  // Return v2 response format with additional metadata
  res.json({
    id: result.transactionId,
    status: result.status,
    created_at: result.timestamp,
    customer: {
      id: payment.customerId,
      email: payment.customerEmail
    }
  });
});

// Shared business logic (single source of truth)
async function processPayment(payment) {
  // Validation, fraud checks, payment gateway interaction
  // This code doesn't care which API version called it
}

// Server-side protocol negotiation
const WebSocket = require('ws');

const wss = new WebSocket.Server({ port: 8080 });

wss.on('connection', (ws, req) => {
  // Client sends protocol version in initial connection
  const clientVersion = req.headers['sec-websocket-protocol'] || 'v1';
  
  // Server supports v1 and v2 simultaneously
  const protocol = negotiateProtocol(clientVersion);
  
  ws.protocol = protocol; // Store for this connection's lifetime
  
  ws.on('message', (data) => {
    const message = JSON.parse(data);
    
    // Route to version-specific handler
    if (ws.protocol === 'v1') {
      handleV1Message(ws, message);
    } else if (ws.protocol === 'v2') {
      handleV2Message(ws, message);
    }
  });
});

function negotiateProtocol(requested) {
  const supported = ['v1', 'v2'];
  return supported.includes(requested) ? requested : 'v1'; // Default to v1
}

// v1 handler (legacy format)
function handleV1Message(ws, message) {
  if (message.type === 'subscribe') {
    // Old format: { type: 'subscribe', symbol: 'AAPL' }
    subscribeToSymbol(ws, message.symbol, 'v1');
  }
}

// v2 handler (new format with additional options)
function handleV2Message(ws, message) {
  if (message.type === 'subscribe') {
    // New format: { type: 'subscribe', symbols: ['AAPL', 'GOOGL'], depth: 10 }
    message.symbols.forEach(symbol => {
      subscribeToSymbol(ws, symbol, 'v2', message.depth);
    });
  }
}

// Shared subscription logic
function subscribeToSymbol(ws, symbol, version, depth = 1) {
  // Subscribe to market data feed
  // Send updates in version-specific format
  marketDataFeed.on(`update:${symbol}`, (data) => {
    if (version === 'v1') {
      ws.send(JSON.stringify({
        symbol: symbol,
        price: data.lastPrice
      }));
    } else if (version === 'v2') {
      ws.send(JSON.stringify({
        symbol: symbol,
        last: data.lastPrice,
        bid: data.bidPrice,
        ask: data.askPrice,
        depth: data.orderBook.slice(0, depth)
      }));
    }
  });
}

# Don't do this (state in application instances)
class FileUploadHandler:
    def __init__(self):
        self.local_storage = {}  # BAD: State lives in memory
    
    def upload(self, user_id, file_data):
        file_id = generate_id()
        self.local_storage[file_id] = file_data
        return file_id

# Do this (state in external persistence layer)
import boto3

class FileUploadHandler:
    def __init__(self):
        self.s3_client = boto3.client('s3')
        self.bucket = 'user-uploads-production'
    
    def upload(self, user_id, file_data):
        file_id = generate_id()
        
        # State persists outside application instances
        self.s3_client.put_object(
            Bucket=self.bucket,
            Key=f'{user_id}/{file_id}',
            Body=file_data,
            Metadata={'uploaded_at': datetime.utcnow().isoformat()}
        )
        
        # Record in database for queryability
        db.execute(
            'INSERT INTO uploads (id, user_id, s3_key, created_at) VALUES (?, ?, ?, ?)',
            (file_id, user_id, f'{user_id}/{file_id}', datetime.utcnow())
        )
        
        return file_id
    
    def download(self, user_id, file_id):
        # Works regardless of which environment handled upload
        response = self.s3_client.get_object(
            Bucket=self.bucket,
            Key=f'{user_id}/{file_id}'
        )
        return response['Body'].read()

-- Migration deployed BEFORE application code
ALTER TABLE subscriptions ADD COLUMN renewal_algorithm VARCHAR(50) DEFAULT 'legacy';
CREATE INDEX CONCURRENTLY idx_subscriptions_renewal_algorithm ON subscriptions(renewal_algorithm);

# Blue code (deployed currently, compatible with new column)
def calculate_renewal(subscription):
    # Ignores new column entirely, continues using old logic
    return subscription.price * 12

# Green code (new deployment, uses new column if present)
def calculate_renewal(subscription):
    algorithm = getattr(subscription, 'renewal_algorithm', 'legacy')
    
    if algorithm == 'legacy':
        return subscription.price * 12
    elif algorithm == 'usage_based':
        return calculate_usage_renewal(subscription)
    elif algorithm == 'tiered':
        return calculate_tiered_renewal(subscription)

const crypto = require('crypto');

class CanaryRouter {
  constructor(canaryPercentage = 10) {
    this.canaryPercentage = canaryPercentage;
  }
  
  // Deterministic routing: same user always routes to same version
  routeRequest(userId) {
    const hash = crypto.createHash('md5').update(userId).digest('hex');
    const hashValue = parseInt(hash.substring(0, 8), 16);
    const bucket = hashValue % 100;
    
    return bucket < this.canaryPercentage ? 'canary' : 'stable';
  }
  
  // Middleware integration
  middleware() {
    return (req, res, next) => {
      const userId = req.user?.id || req.sessionId;
      const targetVersion = this.routeRequest(userId);
      
      // Add routing header for upstream services
      req.headers['x-deployment-version'] = targetVersion;
      
      // Log for analysis
      console.log({
        userId: userId,
        version: targetVersion,
        endpoint: req.path,
        timestamp: Date.now()
      });
      
      next();
    };
  }
}

// Application setup
const canaryRouter = new CanaryRouter(10); // 10% canary traffic

app.use(canaryRouter.middleware());

// Service selection based on routing decision
app.get('/api/data', async (req, res) => {
  const version = req.headers['x-deployment-version'];
  
  const serviceUrl = version === 'canary' 
    ? 'http://backend-canary:8080'
    : 'http://backend-stable:8080';
  
  const response = await fetch(`${serviceUrl}/data`);
  const data = await response.json();
  
  res.json(data);
});

import time
from dataclasses import dataclass
from typing import Dict

@dataclass
class DeploymentMetrics:
    error_rate: float
    p95_latency: float
    p99_latency: float
    request_count: int

class CanaryMonitor:
    def __init__(self, metrics_client, rollback_callback):
        self.metrics_client = metrics_client
        self.rollback_callback = rollback_callback
        
        # Thresholds (configure based on baseline)
        self.max_error_rate_delta = 0.02  # 2% increase triggers rollback
        self.max_p95_latency_delta = 200  # 200ms increase triggers rollback
        self.min_sample_size = 1000  # Need statistically significant data
    
    def monitor_canary(self, duration_minutes=60):
        """Monitor canary deployment and auto-rollback if unhealthy"""
        start_time = time.time()
        check_interval = 60  # Check every minute
        
        while time.time() - start_time < duration_minutes * 60:
            stable_metrics = self.get_metrics('stable')
            canary_metrics = self.get_metrics('canary')
            
            if canary_metrics.request_count < self.min_sample_size:
                print(f"Insufficient canary traffic: {canary_metrics.request_count}")
                time.sleep(check_interval)
                continue
            
            # Compare error rates
            error_rate_delta = canary_metrics.error_rate - stable_metrics.error_rate
            if error_rate_delta > self.max_error_rate_delta:
                print(f"ERROR RATE SPIKE: Canary {canary_metrics.error_rate:.2%} vs Stable {stable_metrics.error_rate:.2%}")
                self.rollback_callback("Error rate threshold exceeded")
                return False
            
            # Compare latency
            latency_delta = canary_metrics.p95_latency - stable_metrics.p95_latency
            if latency_delta > self.max_p95_latency_delta:
                print(f"LATENCY SPIKE: Canary p95={canary_metrics.p95_latency}ms vs Stable p95={stable_metrics.p95_latency}ms")
                self.rollback_callback("Latency threshold exceeded")
                return False
            
            print(f"Canary healthy: errors={canary_metrics.error_rate:.2%}, p95={canary_metrics.p95_latency}ms")
            time.sleep(check_interval)
        
        print("Canary monitoring complete. No issues detected.")
        return True
    
    def get_metrics(self, version: str) -> DeploymentMetrics:
        """Fetch metrics from monitoring system (Prometheus, Datadog, etc.)"""
        query = f'error_rate{{version="{version}"}}'
        error_rate = self.metrics_client.query(query)
        
        query = f'latency_p95{{version="{version}"}}'
        p95_latency = self.metrics_client.query(query)
        
        query = f'request_count{{version="{version}"}}'
        request_count = self.metrics_client.query(query)
        
        return DeploymentMetrics(
            error_rate=error_rate,
            p95_latency=p95_latency,
            p99_latency=0,  # Simplified
            request_count=request_count
        )

# Usage in deployment pipeline
def deploy_canary():
    # Deploy canary version
    deploy_to_cluster('canary', new_version='v2.4.0')
    
    # Configure routing (10% traffic to canary)
    configure_traffic_split(canary_percentage=10)
    
    # Monitor with automatic rollback
    monitor = CanaryMonitor(
        metrics_client=prometheus_client,
        rollback_callback=lambda reason: rollback_deployment(reason)
    )
    
    success = monitor.monitor_canary(duration_minutes=30)
    
    if success:
        # Promote canary to stable
        promote_canary_to_stable()
    else:
        print("Canary failed health checks. Already rolled back.")

class CacheManager {
  constructor(redisClient, appVersion) {
    this.redis = redisClient;
    this.version = appVersion; // e.g., 'v2.4.0'
  }
  
  // Cache keys include version
  buildKey(namespace, identifier) {
    return `${namespace}:${this.version}:${identifier}`;
  }
  
  async get(namespace, identifier) {
    const key = this.buildKey(namespace, identifier);
    const cached = await this.redis.get(key);
    
    if (cached) {
      return JSON.parse(cached);
    }
    
    return null;
  }
  
  async set(namespace, identifier, data, ttl = 3600) {
    const key = this.buildKey(namespace, identifier);
    await this.redis.setex(key, ttl, JSON.stringify(data));
  }
}

// Usage
const cache = new CacheManager(redisClient, process.env.APP_VERSION);

async function getUserProfile(userId) {
  // Check cache (version-specific)
  let profile = await cache.get('user_profile', userId);
  
  if (profile) {
    return profile;
  }
  
  // Cache miss, fetch from database
  profile = await db.query('SELECT * FROM users WHERE id = ?', [userId]);
  
  // Store in version-specific cache
  await cache.set('user_profile', userId, profile, 7200);
  
  return profile;
}

class CacheManagerWithPER extends CacheManager {
  async get(namespace, identifier) {
    const key = this.buildKey(namespace, identifier);
    const cached = await this.redis.get(key);
    
    if (!cached) {
      return null;
    }
    
    const { data, cachedAt, ttl } = JSON.parse(cached);
    const age = Date.now() - cachedAt;
    
    // Probabilistic early expiration formula
    // As cache entry ages, probability of treating it as expired increases
    // This distributes cache refresh load over time instead of at exact TTL
    const delta = ttl * 1000; // Convert to milliseconds
    const beta = 1.0; // Tuning parameter
    
    const xfetch = delta * beta * Math.log(Math.random());
    
    if (age - xfetch >= 0) {
      // Treat as expired (even if technically still valid)
      return null;
    }
    
    return data;
  }
  
  async set(namespace, identifier, data, ttl = 3600) {
    const key = this.buildKey(namespace, identifier);
    const cached = {
      data: data,
      cachedAt: Date.now(),
      ttl: ttl
    };
    await this.redis.setex(key, ttl, JSON.stringify(cached));
  }
}

class FeatureFlagManager {
  constructor(flagConfig) {
    this.flags = flagConfig;
  }
  
  isEnabled(flagName, userId) {
    const flag = this.flags[flagName];
    
    if (!flag) {
      return false; // Unknown flags default to disabled
    }
    
    // Global on/off
    if (flag.enabled === false) {
      return false;
    }
    
    // User whitelist (for testing/VIP users)
    if (flag.whitelist && flag.whitelist.includes(userId)) {
      return true;
    }
    
    // Percentage rollout
    if (flag.percentage !== undefined) {
      const hash = crypto.createHash('md5')
        .update(`${flagName}:${userId}`)
        .digest('hex');
      const bucket = parseInt(hash.substring(0, 8), 16) % 100;
      return bucket < flag.percentage;
    }
    
    return flag.enabled;
  }
}

// Configuration (stored in database, config service, or file)
const flagConfig = {
  'new_checkout_flow': {
    enabled: true,
    percentage: 25, // 25% of users see new checkout
    whitelist: ['internal_user_1', 'internal_user_2']
  },
  'ai_recommendations': {
    enabled: true,
    percentage: 5, // Canary rollout at 5%
    whitelist: []
  },
  'legacy_dashboard': {
    enabled: false // Fully disabled
  }
};

const flags = new FeatureFlagManager(flagConfig);

// Application code
app.get('/checkout', async (req, res) => {
  const userId = req.user.id;
  
  if (flags.isEnabled('new_checkout_flow', userId)) {
    return res.render('checkout_v2', { cart: req.cart });
  } else {
    return res.render('checkout_v1', { cart: req.cart });
  }
});

CREATE TABLE feature_flags (
  name VARCHAR(100) PRIMARY KEY,
  enabled BOOLEAN NOT NULL DEFAULT false,
  percentage INTEGER CHECK (percentage >= 0 AND percentage <= 100),
  whitelist TEXT[], -- Array of user IDs
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_feature_flags_enabled ON feature_flags(enabled);

import psycopg2
import json
from typing import List, Optional

class DatabaseFeatureFlagManager:
    def __init__(self, db_connection):
        self.db = db_connection
        self.cache = {}
        self.cache_ttl = 60  # Refresh from DB every 60 seconds
        self.last_refresh = 0
    
    def refresh_cache(self):
        """Load all flags from database into memory cache"""
        cursor = self.db.cursor()
        cursor.execute('SELECT name, enabled, percentage, whitelist FROM feature_flags')
        
        self.cache = {}
        for row in cursor.fetchall():
            self.cache[row[0]] = {
                'enabled': row[1],
                'percentage': row[2],
                'whitelist': row[3] or []
            }
        
        self.last_refresh = time.time()
    
    def is_enabled(self, flag_name: str, user_id: str) -> bool:
        # Refresh cache if stale
        if time.time() - self.last_refresh > self.cache_ttl:
            self.refresh_cache()
        
        flag = self.cache.get(flag_name)
        if not flag or not flag['enabled']:
            return False
        
        # Whitelist check
        if user_id in flag['whitelist']:
            return True
        
        # Percentage-based rollout
        if flag['percentage'] is not None:
            hash_value = int(hashlib.md5(f"{flag_name}:{user_id}".encode()).hexdigest()[:8], 16)
            bucket = hash_value % 100
            return bucket < flag['percentage']
        
        return True

# Admin API to control flags without deployment
@app.post('/admin/feature-flags/{flag_name}/percentage')
async def update_flag_percentage(flag_name: str, percentage: int):
    db.execute(
        'UPDATE feature_flags SET percentage = $1, updated_at = NOW() WHERE name = $2',
        [percentage, flag_name]
    )
    return {"status": "updated", "flag": flag_name, "percentage": percentage}

app.get('/health', (req, res) => {
  res.status(200).send('OK');
});

const healthChecks = {
  database: async () => {
    try {
      const result = await db.query('SELECT 1');
      return { healthy: true, latency: result.duration };
    } catch (error) {
      return { healthy: false, error: error.message };
    }
  },
  
  redis: async () => {
    try {
      const start = Date.now();
      await redis.ping();
      return { healthy: true, latency: Date.now() - start };
    } catch (error) {
      return { healthy: false, error: error.message };
    }
  },
  
  externalAPI: async () => {
    try {
      const start = Date.now();
      const response = await fetch('https://api.partner.com/health', {
        timeout: 2000 // Don't let external dependency slow health checks
      });
      return { 
        healthy: response.ok, 
        latency: Date.now() - start,
        status: response.status
      };
    } catch (error) {
      return { healthy: false, error: error.message };
    }
  }
};

app.get('/health', async (req, res) => {
  const results = {};
  let overallHealthy = true;
  
  for (const [name, check] of Object.entries(healthChecks)) {
    results[name] = await check();
    if (!results[name].healthy) {
      overallHealthy = false;
    }
  }
  
  // Return 503 if any critical dependency is unhealthy
  const statusCode = overallHealthy ? 200 : 503;
  
  res.status(statusCode).json({
    status: overallHealthy ? 'healthy' : 'unhealthy',
    timestamp: new Date().toISOString(),
    checks: results
  });
});

// Separate liveness check (for container orchestrators)
app.get('/health/live', (req, res) => {
  // Just confirm process is alive
  res.status(200).send('OK');
});

// Separate readiness check (for deployment coordination)
app.get('/health/ready', async (req, res) => {
  // Check if app is ready to receive traffic
  const dbReady = await healthChecks.database();
  const cacheReady = await healthChecks.redis();
  
  if (dbReady.healthy && cacheReady.healthy) {
    res.status(200).json({ ready: true });
  } else {
    res.status(503).json({ 
      ready: false, 
      database: dbReady.healthy,
      cache: cacheReady.healthy
    });
  }
});

# nginx.conf
upstream backend {
  server backend-1:8080;
  server backend-2:8080;
  server backend-3:8080;
  
  # Health check every 5 seconds
  check interval=5000 rise=2 fall=3 timeout=2000 type=http;
  check_http_send "GET /health/ready HTTP/1.0\r\n\r\n";
  check_http_expect_alive http_2xx;
}

const express = require('express');
const app = express();

let isShuttingDown = false;
const activeRequests = new Set();

// Track active requests
app.use((req, res, next) => {
  if (isShuttingDown) {
    // Reject new requests during shutdown
    res.status(503).send('Server is shutting down');
    return;
  }
  
  activeRequests.add(req);
  
  res.on('finish', () => {
    activeRequests.delete(req);
  });
  
  next();
});

// Application routes
app.get('/api/data', async (req, res) => {
  // Simulate slow endpoint
  await new Promise(resolve => setTimeout(resolve, 2000));
  res.json({ data: 'example' });
});

const server = app.listen(8080, () => {
  console.log('Server started on port 8080');
});

// Graceful shutdown handler
process.on('SIGTERM', () => {
  console.log('SIGTERM received, starting graceful shutdown');
  isShuttingDown = true;
  
  // Stop accepting new connections
  server.close(() => {
    console.log('Server closed, no longer accepting connections');
  });
  
  // Wait for active requests to complete (with timeout)
  const shutdownTimeout = setTimeout(() => {
    console.error(`Forcefully terminating ${activeRequests.size} active requests`);
    process.exit(1);
  }, 30000); // 30 second grace period
  
  const checkInterval = setInterval(() => {
    console.log(`Waiting for ${activeRequests.size} active requests to complete`);
    
    if (activeRequests.size === 0) {
      clearInterval(checkInterval);
      clearTimeout(shutdownTimeout);
      
      console.log('All requests completed, shutting down gracefully');
      process.exit(0);
    }
  }, 1000);
});

apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-server
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    spec:
      containers:
      - name: api
        image: api-server:v2.4.0
        ports:
        - containerPort: 8080
        livenessProbe:
          httpGet:
            path: /health/live
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health/ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5
        lifecycle:
          preStop:
            exec:
              # Give load balancer time to deregister before sending SIGTERM
              command: ["/bin/sh", "-c", "sleep 10"]
        terminationGracePeriodSeconds: 40

class SessionStore:
    def __init__(self, postgres_client, mongo_client, migration_phase):
        self.postgres = postgres_client
        self.mongo = mongo_client
        self.phase = migration_phase
    
    async def save_session(self, user_id, session_data):
        if self.phase == 'PHASE_1_DUAL_WRITE':
            # Write to PostgreSQL (source of truth)
            await self.postgres.execute(
                'INSERT INTO sessions (user_id, data, created_at) VALUES ($1, $2, NOW())',
                [user_id, json.dumps(session_data)]
            )
            
            # Write to MongoDB (new system, building up data)
            try:
                await self.mongo.sessions.insert_one({
                    'user_id': user_id,
                    'data': session_data,
                    'created_at': datetime.utcnow()
                })
            except Exception as e:
                # Log but don't fail (MongoDB is not yet authoritative)
                logger.error(f'MongoDB write failed: {e}')
        
        # Other phases handled below
    
    async def get_session(self, user_id):
        if self.phase == 'PHASE_1_DUAL_WRITE':
            # Read from PostgreSQL only (still source of truth)
            result = await self.postgres.fetch_one(
                'SELECT data FROM sessions WHERE user_id = $1',
                [user_id]
            )
            return json.loads(result['data']) if result else None

async def backfill_historical_sessions():
    """Copy existing PostgreSQL sessions to MongoDB in batches"""
    batch_size = 10000
    offset = 0
    total_migrated = 0
    
    while True:
        # Fetch batch from PostgreSQL
        sessions = await postgres.fetch_all(
            'SELECT user_id, data, created_at FROM sessions ORDER BY id LIMIT $1 OFFSET $2',
            [batch_size, offset]
        )
        
        if not sessions:
            break
        
        # Bulk insert into MongoDB
        documents = [
            {
                'user_id': s['user_id'],
                'data': json.loads(s['data']),
                'created_at': s['created_at'],
                'backfilled': True  # Mark for verification
            }
            for s in sessions
        ]
        
        try:
            await mongo.sessions.insert_many(documents, ordered=False)
            total_migrated += len(documents)
            print(f'Migrated {total_migrated} sessions')
        except Exception as e:
            logger.error(f'Backfill batch failed: {e}')
        
        offset += batch_size
        
        # Rate limiting to avoid overwhelming MongoDB
        await asyncio.sleep(0.5)

async def get_session(self, user_id):
    if self.phase == 'PHASE_3_DUAL_READ':
        # Read from BOTH systems and compare
        pg_result = await self.postgres.fetch_one(
            'SELECT data FROM sessions WHERE user_id = $1',
            [user_id]
        )
        pg_data = json.loads(pg_result['data']) if pg_result else None
        
        mongo_result = await self.mongo.sessions.find_one({'user_id': user_id})
        mongo_data = mongo_result['data'] if mongo_result else None
        
        # Compare and log discrepancies
        if pg_data != mongo_data:
            logger.warning(f'Data mismatch for user {user_id}')
            logger.warning(f'PostgreSQL: {pg_data}')
            logger.warning(f'MongoDB: {mongo_data}')
        
        # Still return PostgreSQL data (source of truth)
        return pg_data

async def get_session(self, user_id):
    if self.phase == 'PHASE_4_MONGO_PRIMARY':
        # Read from MongoDB (new source of truth)
        result = await self.mongo.sessions.find_one({'user_id': user_id})
        
        if result:
            return result['data']
        
        # Fallback to PostgreSQL (should rarely happen)
        pg_result = await self.postgres.fetch_one(
            'SELECT data FROM sessions WHERE user_id = $1',
            [user_id]
        )
        
        if pg_result:
            logger.info(f'Fallback to PostgreSQL for user {user_id}')
            return json.loads(pg_result['data'])
        
        return None

async def save_session(self, user_id, session_data):
    if self.phase == 'PHASE_5_MONGO_ONLY':
        # Write to MongoDB only
        await self.mongo.sessions.insert_one({
            'user_id': user_id,
            'data': session_data,
            'created_at': datetime.utcnow()
        })

async def get_session(self, user_id):
    if self.phase == 'PHASE_5_MONGO_ONLY':
        result = await self.mongo.sessions.find_one({'user_id': user_id})
        return result['data'] if result else None

// Cache versioning with backward compatibility
class MultiRegionCacheManager {
  constructor(redisClient, appVersion, region) {
    this.redis = redisClient;
    this.version = appVersion;
    this.region = region;
  }
  
  async get(key) {
    // Try current version first
    const currentKey = `${this.version}:${this.region}:${key}`;
    let data = await this.redis.get(currentKey);
    
    if (data) {
      return this.deserialize(data, this.version);
    }
    
    // Fallback to previous version (for cross-region compatibility)
    const previousVersion = this.getPreviousVersion(this.version);
    const fallbackKey = `${previousVersion}:${this.region}:${key}`;
    data = await this.redis.get(fallbackKey);
    
    if (data) {
      return this.deserialize(data, previousVersion);
    }
    
    return null;
  }
  
  deserialize(data, version) {
    const parsed = JSON.parse(data);
    
    // Handle version-specific formats
    if (version === 'v2.3.0') {
      // Old format: { email: 'user@example.com' }
      return parsed;
    } else if (version === 'v2.4.0') {
      // New format: { contact: { email: 'user@example.com', phone: '...' } }
      return parsed;
    }
  }
  
  getPreviousVersion(currentVersion) {
    const versionMap = {
      'v2.4.0': 'v2.3.0',
      'v2.3.0': 'v2.2.0'
    };
    return versionMap[currentVersion];
  }
}

import datetime
import hashlib

class DeploymentAuditLogger:
    def __init__(self, db_connection):
        self.db = db_connection
    
    def log_deployment_start(self, version, deployed_by, environment, artifacts):
        """Log deployment initiation with artifact checksums"""
        artifact_checksums = {
            name: self.calculate_checksum(path)
            for name, path in artifacts.items()
        }
        
        deployment_id = self.generate_deployment_id()
        
        self.db.execute('''
            INSERT INTO deployment_audit_log (
                deployment_id, version, environment, deployed_by,
                started_at, artifacts, status
            ) VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', [
            deployment_id,
            version,
            environment,
            deployed_by,
            datetime.datetime.utcnow(),
            json.dumps(artifact_checksums),
            'IN_PROGRESS'
        ])
        
        return deployment_id
    
    def log_deployment_complete(self, deployment_id, success, metrics):
        """Log deployment completion with success metrics"""
        self.db.execute('''
            UPDATE deployment_audit_log
            SET status = ?, completed_at = ?, metrics = ?
            WHERE deployment_id = ?
        ''', [
            'SUCCESS' if success else 'FAILED',
            datetime.datetime.utcnow(),
            json.dumps(metrics),
            deployment_id
        ])
    
    def calculate_checksum(self, file_path):
        """Generate SHA-256 checksum for artifact verification"""
        sha256 = hashlib.sha256()
        with open(file_path, 'rb') as f:
            for chunk in iter(lambda: f.read(4096), b''):
                sha256.update(chunk)
        return sha256.hexdigest()
    
    def generate_deployment_id(self):
        """Create unique deployment identifier"""
        timestamp = datetime.datetime.utcnow().isoformat()
        random_suffix = secrets.token_hex(8)
        return f'deploy-{timestamp}-{random_suffix}'

# Usage in deployment pipeline
auditor = DeploymentAuditLogger(db_connection)

deployment_id = auditor.log_deployment_start(
    version='v2.4.0',
    deployed_by='john.doe@company.com',
    environment='production',
    artifacts={
        'api_server': '/artifacts/api-server-v2.4.0.tar.gz',
        'worker': '/artifacts/worker-v2.4.0.tar.gz',
        'migrations': '/artifacts/migrations-v2.4.0.sql'
    }
)

try:
    # Perform deployment
    deploy_result = perform_deployment(version='v2.4.0')
    
    # Log success metrics
    auditor.log_deployment_complete(
        deployment_id=deployment_id,
        success=True,
        metrics={
            'duration_seconds': deploy_result.duration,
            'instances_updated': deploy_result.instance_count,
            'rollback_performed': False
        }
    )
except Exception as e:
    # Log failure
    auditor.log_deployment_complete(
        deployment_id=deployment_id,
        success=False,
        metrics={
            'error': str(e),
            'rollback_performed': True
        }
    )

class ReplicationLagAwareDatabase:
    def __init__(self, primary_conn, replica_conn, max_acceptable_lag_seconds=10):
        self.primary = primary_conn
        self.replica = replica_conn
        self.max_lag = max_acceptable_lag_seconds
    
    async def get_replication_lag(self):
        """Query replica for replication lag in seconds"""
        result = await self.replica.fetch_one(
            "SELECT EXTRACT(EPOCH FROM (NOW() - pg_last_xact_replay_timestamp())) AS lag"
        )
        return result['lag']
    
    async def query(self, sql, params, consistency='eventual'):
        """Execute query with consistency requirement"""
        if consistency == 'strong':
            # Always use primary for strong consistency
            return await self.primary.fetch_all(sql, params)
        
        # Check replication lag
        lag = await self.get_replication_lag()
        
        if lag > self.max_lag:
            # Replica too far behind, use primary
            logger.warning(f'Replica lag {lag}s exceeds threshold, using primary')
            return await self.primary.fetch_all(sql, params)
        
        # Replica is current enough
        return await self.replica.fetch_all(sql, params)

# Usage
db = ReplicationLagAwareDatabase(primary_db, replica_db)

# Eventual consistency acceptable (analytics, listings)
users = await db.query('SELECT * FROM users', [], consistency='eventual')

# Strong consistency required (payment processing, account updates)
account_balance = await db.query(
    'SELECT balance FROM accounts WHERE id = $1',
    [account_id],
    consistency='strong'
)