recovery

name	recovery
description	Implement disaster recovery and point-in-time recovery strategies
shortcut	reco

Database Recovery Manager

Implement comprehensive disaster recovery, point-in-time recovery (PITR), and automated failover strategies for production database systems with automated backup verification and recovery testing.

When to Use This Command

Use /recovery when you need to:

• Set up disaster recovery infrastructure for production databases
• Implement point-in-time recovery (PITR) capabilities
• Automate backup validation and recovery testing
• Design multi-region failover strategies
• Recover from data corruption or accidental deletions
• Meet compliance requirements for backup retention and recovery time objectives (RTO)

DON'T use this when:

• Only need basic database backups (use backup automator instead)
• Working with development databases without recovery requirements
• Database system doesn't support WAL/binary log replication
• Compliance doesn't require tested recovery procedures

Design Decisions

This command implements comprehensive disaster recovery with PITR because:

• Point-in-time recovery prevents data loss from user errors or corruption
• Automated failover ensures minimal downtime (RTO < 5 minutes)
• Regular recovery testing validates backup integrity before disasters
• Multi-region replication provides geographic redundancy
• WAL archiving enables recovery to any point in last 30 days

Alternative considered: Snapshot-only backups

• Simpler to implement and restore
• No point-in-time recovery capability
• Recovery point objective (RPO) limited to snapshot frequency
• Recommended only for non-critical databases

Alternative considered: Manual recovery procedures

• No automation or testing
• Prone to human error during incidents
• Longer recovery times (RTO hours vs minutes)
• Recommended only for development environments

Prerequisites

Before running this command:

1. Database with WAL/binary logging enabled
2. Object storage for backup retention (S3, GCS, Azure Blob)
3. Monitoring infrastructure for backup validation
4. Understanding of RTO (Recovery Time Objective) and RPO (Recovery Point Objective) requirements
5. Separate recovery environment for testing

Implementation Process

Step 1: Configure WAL Archiving and Continuous Backup

Enable write-ahead logging (WAL) archiving for point-in-time recovery capabilities.

Step 2: Implement Automated Base Backup System

Set up scheduled base backups with compression and encryption to object storage.

Step 3: Design Failover and High Availability Architecture

Configure streaming replication with automated failover for zero-downtime recovery.

Step 4: Build Recovery Testing Framework

Automate recovery validation by restoring backups to test environments regularly.

Step 5: Document and Drill Recovery Procedures

Create runbooks and conduct disaster recovery drills quarterly.

Output Format

The command generates:

• config/recovery.conf - PostgreSQL recovery configuration
• scripts/pitr-restore.sh - Point-in-time recovery automation script
• monitoring/backup-validator.py - Automated backup verification
• failover/replication-monitor.py - Streaming replication health monitoring
• docs/recovery-runbook.md - Step-by-step recovery procedures

Code Examples

Example 1: PostgreSQL PITR with WAL Archiving

# postgresql.conf - Enable WAL archiving
wal_level = replica
archive_mode = on
archive_command = 'aws s3 cp %p s3://my-db-backups/wal-archive/%f --region us-east-1'
archive_timeout = 300  # Force segment switch every 5 minutes
max_wal_senders = 10
wal_keep_size = 1GB

# Continuous archiving with monitoring
restore_command = 'aws s3 cp s3://my-db-backups/wal-archive/%f %p'
archive_cleanup_command = 'pg_archivecleanup /path/to/archive %r'

#!/bin/bash
# scripts/pitr-restore.sh - Point-in-Time Recovery Script

set -euo pipefail

# Configuration
BACKUP_BUCKET="s3://my-db-backups"
PGDATA="/var/lib/postgresql/14/main"
TARGET_TIME="${1:-latest}"  # Format: '2024-10-15 14:30:00 UTC'
RECOVERY_TARGET="${2:-immediate}"  # immediate, time, xid, name

# Colors for output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'

log() {
    echo -e "${GREEN}[$(date +'%Y-%m-%d %H:%M:%S')]${NC} $1"
}

error() {
    echo -e "${RED}[ERROR]${NC} $1" >&2
    exit 1
}

warn() {
    echo -e "${YELLOW}[WARNING]${NC} $1"
}

# Step 1: Verify prerequisites
check_prerequisites() {
    log "Checking prerequisites..."

    # Check if PostgreSQL is stopped
    if systemctl is-active --quiet postgresql; then
        warn "PostgreSQL is running. Stopping service..."
        systemctl stop postgresql
    

    
     ! aws sts get-caller-identity &>/dev/null; 
        error 
    

    
    REQUIRED_SPACE=$(( *  * ))  
    AVAILABLE_SPACE=$( -k  |  -1 | awk )

     [  -lt  ]; 
        error 
    

     
}


() {
     

    aws s3   --recursive | \
        grep  | \
        awk  | \
         -r | \
         -10

     -p  SELECTED_BACKUP

     [ -z  ]; 
        SELECTED_BACKUP=$(aws s3   --recursive | \
            grep  | \
            awk  | \
             -r | \
             -1)
    

     
}


() {
     

    
     [ -d  ]; 
        BACKUP_DIR=
        warn 
          
    

     -p 

    
     
    aws s3   - | \
        tar -xzf - -C 

    
     -R postgres:postgres 
     700 

     
}


() {
     

     >  << 

    
     >  << 

    
      
        )
              >> 
             
            ;;
        xid)
              >> 
             
            ;;
        name)
              >> 
             
            ;;
        immediate)
              >> 
             
            ;;
    

     
}


() {
     

    systemctl start postgresql

     

    
     ; 
          -u postgres psql -c  2>/dev/null | grep -q ; 
             
            
        

        
        RECOVERY_INFO=$( -u postgres psql -c  2>/dev/null |  -3 |  -1 ||  )

         -ne 
         5
    

     
}


() {
     

    
     -u postgres psql -c 
     -u postgres psql -c 

    
    SLOT_COUNT=$( -u postgres psql -t -c )
     [  -gt 0 ]; 
        warn 
    

     
}


() {
     
     
     

    check_prerequisites
    list_backups
    restore_base_backup
    configure_recovery
    start_recovery
    verify_recovery

     
     

     << 
}


main 

# monitoring/backup-validator.py - Automated Backup Verification
import subprocess
import boto3
import psycopg2
import logging
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class BackupValidationResult:
    """Results from backup validation."""
    backup_name: str
    backup_date: datetime
    validation_date: datetime
    size_mb: float
    restore_time_seconds: float
    integrity_check_passed: bool
    table_count: int
    row_sample_count: int
    errors: List[str]
    warnings: List[str]

    def to_dict(self) -> dict:
        result = asdict(self)
        result['backup_date'] = self.backup_date.isoformat()
        result['validation_date'] = self.validation_date.isoformat()
        return result

class PostgreSQLBackupValidator:
    """Validates PostgreSQL backups by restoring to test environment."""

    def __init__():
        .s3_bucket = s3_bucket
        .test_db_config = test_db_config
        .retention_days = retention_days
        .s3_client = boto3.client()

     () -> [[, ]]:
        
        prefix = 
        response = .s3_client.list_objects_v2(
            Bucket=.s3_bucket,
            Prefix=prefix
        )

        cutoff_date = datetime.now() - timedelta(days=days)
        backups = []

         obj  response.get(, []):
             obj[].replace(tzinfo=) > cutoff_date:
                backups.append({
                    : obj[],
                    : obj[],
                    : obj[]
                })

         (backups, key= x: x[], reverse=)

     () -> :
        
        :
            logger.info()
            .s3_client.download_file(
                .s3_bucket,
                backup_key,
                local_path
            )
            logger.info()
             
         Exception  e:
            logger.error()
             

     () -> []:
        
        start_time = datetime.now()

        :
            
            conn = psycopg2.connect(
                host=.test_db_config[],
                user=.test_db_config[],
                password=.test_db_config[],
                database=
            )
            conn.autocommit = 

             conn.cursor()  cur:
                cur.execute()
                cur.execute()

            conn.close()

            
            restore_cmd = [
                ,
                , .test_db_config[],
                , .test_db_config[],
                , .test_db_config[],
                ,
                ,
                ,
                backup_path
            ]

            result = subprocess.run(
                restore_cmd,
                capture_output=,
                text=,
                env={: .test_db_config[]}
            )

             result.returncode != :
                logger.error()
                 

            restore_time = (datetime.now() - start_time).total_seconds()
            logger.info()

             restore_time

         Exception  e:
            logger.error()
             

     () -> [, ]:
        
        checks = {
            : ,
            : ,
            : ,
            : [],
            : [],
            : []
        }

        :
            conn = psycopg2.connect(
                host=.test_db_config[],
                user=.test_db_config[],
                password=.test_db_config[],
                database=.test_db_config[]
            )

             conn.cursor()  cur:
                
                cur.execute()
                checks[] = cur.fetchone()[]

                
                cur.execute()

                 schema, table  cur.fetchall():
                    :
                        cur.execute()
                        row_count = cur.fetchone()[]
                        checks[] += row_count
                     Exception  e:
                        checks[].append()

                
                cur.execute()

                 schema, table, index  cur.fetchall():
                    :
                        cur.execute()
                     Exception  e:
                        checks[].append()
                        checks[] = 

            conn.close()

         Exception  e:
            checks[].append()

         checks

     () -> BackupValidationResult:
        
        logger.info()

        errors = []
        warnings = []

        
        local_backup = 
          .download_backup(backup_info[], local_backup):
            errors.append()

        
        restore_time = .restore_to_test_db(local_backup)
         restore_time  :
            errors.append()

        
        integrity_checks = .verify_database_integrity()
        errors.extend(integrity_checks.get(, []))
        warnings.extend(integrity_checks.get(, []))

        result = BackupValidationResult(
            backup_name=backup_info[],
            backup_date=backup_info[].replace(tzinfo=),
            validation_date=datetime.now(),
            size_mb=backup_info[] / ( * ),
            restore_time_seconds=restore_time  ,
            integrity_check_passed=(errors) == ,
            table_count=integrity_checks.get(, ),
            row_sample_count=integrity_checks.get(, ),
            errors=errors,
            warnings=warnings
        )

        
         result.integrity_check_passed:
            logger.info()
        :
            logger.error()
             error  errors:
                logger.error()

         result

     ():
        
        logger.info()

        backups = .list_recent_backups(days=)
          backups:
            logger.warning()
            

        latest_backup = backups[]
        result = .validate_backup(latest_backup)

        
        report_file = 
         (report_file, )  f:
            json.dump(result.to_dict(), f, indent=)

        logger.info()

        
          result.integrity_check_passed:
            .send_alert(result)

     ():
        
        logger.critical()
        


 __name__ == :
    validator = PostgreSQLBackupValidator(
        s3_bucket=,
        test_db_config={
            : ,
            : ,
            : ,
            : 
        },
        retention_days=
    )

    validator.run_daily_validation()

Example 2: MySQL PITR with Binary Log Replication

# my.cnf - Enable binary logging for PITR
[mysqld]
server-id = 1
log-bin = /var/log/mysql/mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7
sync_binlog = 1
innodb_flush_log_at_trx_commit = 1

# Replication for high availability
gtid_mode = ON
enforce_gtid_consistency = ON
log_slave_updates = ON

#!/bin/bash
# scripts/mysql-pitr-restore.sh

set -euo pipefail

BACKUP_DIR="/backups/mysql"
BINLOG_DIR="/var/log/mysql"
TARGET_DATETIME="$1"  # Format: '2024-10-15 14:30:00'
RESTORE_DIR="/var/lib/mysql-restore"

log() { echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"; }
error() { echo "[ERROR] $1" >&2; exit 1; }

# Find base backup before target time
find_base_backup() {
    log "Finding appropriate base backup..."

    TARGET_EPOCH=$(date -d "$TARGET_DATETIME" +%s)

    for backup in $(ls -t "$BACKUP_DIR"/*.tar.gz); do
        BACKUP_TIME=$(basename "$backup" .tar.gz | cut -d'-' -f2)
        BACKUP_EPOCH=$(date -d "$BACKUP_TIME" +%s 2>/dev/null || continue)

        if [ "$BACKUP_EPOCH" -lt "$TARGET_EPOCH" ]; then
            echo ""
             0
        
    

    error 
}


BASE_BACKUP=$(find_base_backup)
 

systemctl stop mysql
 -rf 
 -p 
tar -xzf  -C 


 

mysqlbinlog \
    --stop-datetime= \
    /mysql-bin.* | \
    mysql --defaults-file=/my.cnf

 

Error Handling

Error	Cause	Solution
"WAL segment not found"	Missing archived WAL files	Check archive_command and S3 bucket permissions
"Invalid checkpoint"	Corrupted base backup	Restore from previous base backup
"Recovery target not reached"	Target time beyond available WAL	Verify WAL archiving is functioning
"Insufficient disk space"	Large database or WAL files	Provision additional storage or compress archives
"Connection refused during recovery"	PostgreSQL still in recovery mode	Wait for recovery to complete before connecting

Configuration Options

WAL Archiving

• wal_level = replica: Enable WAL archiving (PostgreSQL)
• archive_mode = on: Activate WAL archiving
• archive_timeout = 300: Force WAL segment switch every 5 minutes
• log-bin: Enable binary logging (MySQL)

Recovery Targets

• recovery_target_time: Restore to specific timestamp
• recovery_target_xid: Restore to transaction ID
• recovery_target_name: Restore to named restore point
• recovery_target = 'immediate': Stop at end of base backup

Replication

• max_wal_senders = 10: Maximum replication connections
• wal_keep_size = 1GB: Minimum WAL retention on primary

Best Practices

DO:

• Test recovery procedures monthly in isolated environment
• Monitor WAL archiving lag and alert if > 5 minutes
• Encrypt backups at rest and in transit
• Store backups in multiple regions for geographic redundancy
• Validate backup integrity automatically after creation
• Document RTO/RPO requirements and measure against them

DON'T:

• Skip recovery testing (untested backups are useless)
• Store backups on same infrastructure as production database
• Ignore WAL archiving failures (creates recovery gaps)
• Use same credentials for production and backup storage
• Assume backups work without validation

Performance Considerations

• WAL archiving adds ~1-5% overhead depending on write workload
• Use parallel backup tools (pgBackRest, Barman) for large databases
• Compress WAL archives to reduce storage costs (50-70% reduction typical)
• Use incremental backups to minimize backup window
• Consider backup network bandwidth (1TB database = ~30 minutes over 10 Gbps)

Related Commands

• /database-backup-automator - Automated backup scheduling
• /database-replication-manager - Configure streaming replication
• /database-health-monitor - Monitor backup and replication health
• /database-migration-manager - Schema change management with recovery points

Version History

• v1.0.0 (2024-10): Initial implementation with PostgreSQL and MySQL PITR support
• Planned v1.1.0: Add automated failover orchestration and multi-region replication