Database Replication Strategies: Primary-Replica, Multi-Master, and Leaderless

replication_types:
  - name: "Primary-Replica (Leader-Follower)"
    description: "One primary, multiple replicas"
    pros: "Simple, consistent writes"
    cons: "Single point of failure for writes"
    
  - name: "Multi-Master"
    description: "Multiple primaries accept writes"
    pros: "No write bottleneck"
    cons: "Conflict resolution complex"
    
  - name: "Leaderless"
    description: "No designated leader"
    pros: "High availability"
    cons: "Eventual consistency"

# PostgreSQL Primary-Replica setup

# PRIMARY (write to master)
import psycopg2

class PrimaryDatabase:
    def __init__(self, connection_string):
        self.conn = psycopg2.connect(connection_string)
    
    def write(self, query, params):
        with self.conn.cursor() as cur:
            cur.execute(query, params)
            self.conn.commit()
    
    def get_replication_lag(self):
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT now() - pg_last_xact_replay_timestamp() 
                AS lag
            """)
            return cur.fetchone()[0]

# Read from replica (load balancing)

class ReplicaPool:
    def __init__(self, replica_hosts):
        self.hosts = replica_hosts
        self.current = 0
    
    def read(self, query, params=None):
        host = self.hosts[self.current]
        self.current = (self.current + 1) % len(self.hosts)
        
        conn = psycopg2.connect(host)
        try:
            with conn.cursor() as cur:
                cur.execute(query, params or ())
                return cur.fetchall()
        finally:
            conn.close()
    
    def is_replica_synchronized(self, max_lag=1):
        """Check if replica is caught up"""
        for host in self.hosts:
            conn = psycopg2.connect(host)
            try:
                with conn.cursor() as cur:
                    cur.execute("SELECT pg_is_in_recovery()")
                    is_recovering = cur.fetchone()[0]
                    if is_recovering:
                        return False
            finally:
                conn.close()
        return True

# PostgreSQL streaming replication config

# postgresql.conf (primary)
wal_level = replica
max_wal_senders = 3
wal_keep_size = 1GB

# postgresql.conf (replica)
primary_conninfo = 'host=primary port=5432 user=replication'

# Create replication slot
# On replica:
# pg_create_physical_replication_slot('replica_slot')

class MultiMasterDB:
    """Simple last-writer-wins conflict resolution"""
    
    def __init__(self, nodes):
        self.nodes = nodes
    
    def write(self, key, value, timestamp=None):
        if timestamp is None:
            timestamp = time.time()
        
        # Write to all nodes
        for node in self.nodes:
            node.put(key, value, timestamp)
    
    def read(self, key):
        # Read from all and pick latest
        latest = None
        latest_value = None
        
        for node in self.nodes:
            value, ts = node.get(key)
            if ts and (latest is None or ts > latest):
                latest = ts
                latest_value = value
        
        return latest_value


# More sophisticated: CRDTs (Conflict-free Replicated Data Types)
from collections import Counter

class GCounter:
    """Grow-only counter CRDT"""
    
    def __init__(self):
        self.counts = {}  # node_id -> count
    
    def increment(self, node_id):
        self.counts[node_id] = self.counts.get(node_id, 0) + 1
    
    def value(self):
        return sum(self.counts.values())
    
    def merge(self, other):
        for node_id, count in other.counts.items():
            self.counts[node_id] = max(
                self.counts.get(node_id, 0), 
                count
            )

class LeaderlessDB:
    def __init__(self, nodes, r=2, w=2):
        self.nodes = nodes
        self.r = r  # Read quorum
        self.w = w  # Write quorum
    
    def write(self, key, value):
        # Write to N nodes
        success_count = 0
        for node in self.nodes:
            if node.put(key, value):
                success_count += 1
        
        return success_count >= self.w
    
    def read(self, key):
        # Read from R nodes
        values = []
        for node in self.nodes[:self.r]:
            value = node.get(key)
            if value:
                values.append(value)
        
        # Return most recent (simplified)
        return max(values, key=lambda v: v.timestamp) if values else None

# Monitor replication health

def check_replication_health(conn, replica_conns):
    health = {
        "primary": {"healthy": True},
        "replicas": []
    }
    
    # Check primary
    with conn.cursor() as cur:
        cur.execute("SELECT pg_is_in_recovery()")
        health["primary"]["healthy"] = not cur.fetchone()[0]
    
    # Check each replica
    for replica in replica_conns:
        try:
            with replica.cursor() as cur:
                # Check if receiving
                cur.execute("SELECT pg_is_in_recovery()")
                is_replica = cur.fetchone()[0]
                
                # Get lag
                if is_replica:
                    cur.execute("""
                        SELECT now() - pg_last_xact_replay_timestamp() 
                        AS lag
                    """)
                    lag = cur.fetchone()[0]
                    
                    health["replicas"].append({
                        "healthy": True,
                        "lag_seconds": lag.total_seconds()
                    })
        except Exception as e:
            health["replicas"].append({
                "healthy": False,
                "error": str(e)
            })
    
    return health

# PostgreSQL synchronous replication
synchronous_commit = on
synchronous_standby_names = 'replica1, replica2'

synchronous_commit = on
synchronous_standby_names = 'ANY 2 (replica1, replica2, replica3)'

         ┌── Replica A
         │
Primary ──┼── Replica B
         │
         └── Replica C

Primary → Replica A → Replica B
                ↓
             Replica C

Node A ↔ Node B
   ↕        ↕
Node D ↔ Node C

N = total nodes, R = read quorum, W = write quorum

Strong consistency:  R + W > N
Eventual consistency: R + W <= N

# PostgreSQL + Patroni
patroni:
  bootstrap:
    dcs:
      ttl: 30
      loop_wait: 10
      retry_timeout: 10
      maximum_lag_on_failover: 1048576
      postgresql:
        use_pg_rewind: true
        parameters:
          wal_level: replica
          hot_standby: "on"
          max_connections: 100

# Monitor replication health

def check_replication_health(conn, replica_conns):
    health = {
        "primary": {"healthy": True},
        "replicas": []
    }
    
    # Check primary
    with conn.cursor() as cur:
        cur.execute("SELECT pg_is_in_recovery()")
        health["primary"]["healthy"] = not cur.fetchone()[0]
    
    # Check each replica
    for replica in replica_conns:
        try:
            with replica.cursor() as cur:
                # Check if receiving
                cur.execute("SELECT pg_is_in_recovery()")
                is_replica = cur.fetchone()[0]
                
                # Get lag
                if is_replica:
                    cur.execute("""
                        SELECT now() - pg_last_xact_replay_timestamp() 
                        AS lag
                    """)
                    lag = cur.fetchone()[0]
                    
                    health["replicas"].append({
                        "healthy": True,
                        "lag_seconds": lag.total_seconds()
                    })
        except Exception as e:
            health["replicas"].append({
                "healthy": False,
                "error": str(e)
            })
    
    return health

# Streaming replication
wal_level = replica
max_wal_senders = 3
hot_standby = on
synchronous_commit = on

# MySQL source-replica replication
# my.cnf (source)
server_id = 1
log_bin = mysql-bin
binlog_format = ROW

# my.cnf (replica)
server_id = 2
read_only = ON
relay_log = mysql-relay-bin

# MongoDB replica set (primary + 2 secondaries)
replication:
  replSetName: rs0

# Redis master-replica
replicaof 127.0.0.1 6379

# Cassandra replication (network topology strategy)
replication:
  class: NetworkTopologyStrategy
  datacenter1: 3
  datacenter2: 3

class ReadWriteRouter:
    """Route reads to replicas, writes to primary."""

    def __init__(self, primary, replicas):
        self.primary = primary
        self.replicas = replicas
        self.replica_idx = 0

    def execute(self, query, params=None, is_write=False):
        if is_write or not self._is_read_only(query):
            return self.primary.execute(query, params)
        return self._read_from_replica(query, params)

    def _is_read_only(self, query):
        return query.strip().upper().startswith(
            ('SELECT', 'SHOW', 'EXPLAIN')
        )

    def _read_from_replica(self, query, params):
        replica = self.replicas[self.replica_idx % len(self.replicas)]
        self.replica_idx += 1
        return replica.execute(query, params)

Region A (primary) ←→ Region B (primary)
      ↓                    ↓
  Replica A1           Replica B1
  Replica A2           Replica B2

# Replication best practices

configuration:
  - "Use async for lower latency"
  - "Sync for critical data"
  - "3 replicas minimum"
  - "Spread across AZs"
  - "Enable semi-sync for durability"
  - "Configure proper quorum"

monitoring:
  - "Track replication lag"
  - "Alert on high lag"
  - "Monitor disk usage"
  - "Watch replica health"
  - "Track WAL sender usage"

failover:
  - "Automatic failover"
  - "Test failover regularly"
  - "Promote carefully"
  - "Use quorum to avoid split-brain"
  - "Verify lag before promotion"

Writes → Primary
Reads  → Replica 1, Replica 2 (round-robin)
Analytics → Dedicated Replica 3
Failover → Patroni manages promotion

US Region: Primary-US + 2 replicas
EU Region: Primary-EU + 2 replicas
APAC Region: Primary-APAC + 2 replicas
Cross-region: Async bidirectional replication
Conflict: LWW with application-level merge

Primary + 2 sync replicas (quorum)
synchronous_commit = on
synchronous_standby_names = 'ANY 2 (...)'
Async replicas for analytics/backup
Backups with PITR every 5 minutes