-- PostgreSQL streaming replication setup (leader)
-- postgresql.conf on leader
wal_level = replica
max_wal_senders = 10
wal_keep_size = '1GB'
hot_standby = on
 
-- Create replication user
CREATE ROLE replicator WITH REPLICATION LOGIN PASSWORD 'secure_password';
 
-- Check replication status on leader
SELECT
  client_addr,
  state,
  sent_lsn,
  write_lsn,
  flush_lsn,
  replay_lsn,
  pg_wal_lsn_diff(sent_lsn, replay_lsn) as replication_lag_bytes
FROM pg_stat_replication;

-- MySQL semi-synchronous replication setup
-- On source:
INSTALL PLUGIN rpl_semi_sync_source SONAME 'semisync_source.so';
SET GLOBAL rpl_semi_sync_source_enabled = ON;
SET GLOBAL rpl_semi_sync_source_timeout = 5000; -- 5 seconds
 
-- On replica:
INSTALL PLUGIN rpl_semi_sync_replica SONAME 'semisync_replica.so';
SET GLOBAL rpl_semi_sync_replica_enabled = ON;
STOP REPLICA; START REPLICA;
 
-- Monitor semi-sync status on source
SHOW STATUS LIKE 'Rpl_semi_sync%';
-- Key variables:
-- Rpl_semi_sync_source_status: ON/OFF
-- Rpl_semi_sync_source_clients: number of semi-sync replicas
-- Rpl_semi_sync_source_net_avg_wait_time: avg wait for replica ack
-- Rpl_semi_sync_source_no_tx: transactions not semi-sync confirmed

-- Setting up a follower (replica)
-- 1. Take a base backup from leader
pg_basebackup -h leader_host -D /var/lib/postgresql/data -U replicator -Fp -Xs -P -R
 
-- 2. Start the follower - it begins streaming automatically
 
-- Monitor lag on follower
SELECT
  pg_is_in_recovery() as is_replica,
  pg_last_wal_receive_lsn() as received,
  pg_last_wal_replay_lsn() as replayed,
  pg_wal_lsn_diff(
    pg_last_wal_receive_lsn(),
    pg_last_wal_replay_lsn()
  ) as lag_bytes,
  now() - pg_last_xact_replay_timestamp() as replay_lag;

-- Cascading replication setup: Follower C replicates from Follower A
-- On Follower C's postgresql.conf:
primary_conninfo = 'host=follower_a_host port=5432 user=replicator password=secure_password'
-- The rest of the configuration is identical to setting up any standby
 
-- Monitor cascading replication lag on Follower C
SELECT
  pg_is_in_recovery() as is_replica,
  pg_last_wal_receive_lsn() as received,
  pg_last_wal_replay_lsn() as replayed,
  now() - pg_last_xact_replay_timestamp() as time_since_last_replay;

-- Create a physical replication slot on the leader
SELECT pg_create_physical_replication_slot('standby_slot_1');
 
-- The standby connects using this slot
-- primary_conninfo includes: ...options='-c standby_slot_name=standby_slot_1'
 
-- Monitor slot lag - CRITICAL for disk space management
SELECT
  slot_name,
  slot_type,
  active,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as retained_wal,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) as lag
FROM pg_replication_slots;
 
-- Set maximum WAL retention for a slot to prevent disk bloat
ALTER SYSTEM SET max_slot_wal_keep_size = '50GB';
SELECT pg_reload_conf();

// Solution 1: Read-after-write consistency by routing to leader
async function updateProfile(userId: string, data: ProfileData) {
  await leaderPool.query('UPDATE users SET profile = $1 WHERE id = $2', [data, userId]);
  // Route the subsequent read to the leader for this user
  const result = await leaderPool.query('SELECT * FROM users WHERE id = $1', [userId]);
  return result.rows[0];
}
 
// Solution 2: Track write timestamp and wait for replica to catch up
async function readAfterWrite(userId: string, lastWriteTimestamp: Date) {
  const replicaLag = await followerPool.query(
    "SELECT now() - pg_last_xact_replay_timestamp() as lag"
  );
  const lagMs = replicaLag.rows[0].lag?.toMillis?.() || 0;
  const staleness = Date.now() - lastWriteTimestamp.getTime();
 
  if (lagMs > staleness) {
    // Replica hasn't caught up yet, read from leader
    return leaderPool.query('SELECT * FROM users WHERE id = $1', [userId]);
  }
  return followerPool.query('SELECT * FROM users WHERE id = $1', [userId]);
}

# patroni.yml - Patroni configuration for automatic failover
scope: postgres-cluster
name: node1
 
restapi:
  listen: 0.0.0.0:8008
  connect_address: node1:8008
 
etcd3:
  hosts: etcd1:2379,etcd2:2379,etcd3:2379
 
bootstrap:
  dcs:
    ttl: 30
    loop_wait: 10
    retry_timeout: 10
    maximum_lag_on_failover: 1048576  # 1MB max lag for failover
    synchronous_mode: true            # Require at least one sync replica
    synchronous_mode_strict: false    # Don't block writes if no sync replica
    postgresql:
      use_pg_rewind: true
      parameters:
        wal_level: replica
        hot_standby: "on"
        max_wal_senders: 10
        max_replication_slots: 10
        wal_log_hints: "on"
 
# Synchronous replication with Patroni
# When synchronous_mode is true, Patroni manages synchronous_standby_names
# automatically, ensuring at least one replica is synchronous before allowing writes

// Node.js: Read-write split with automatic failover handling
import { Pool } from 'pg';
 
const leaderPool = new Pool({
  host: process.env.PG_LEADER_HOST,
  database: 'myapp',
  max: 20,
  connectionTimeoutMillis: 3000,
});
 
const followerPool = new Pool({
  host: process.env.PG_FOLLOWER_HOST,
  database: 'myapp',
  max: 30,
  connectionTimeoutMillis: 3000,
});
 
async function query<T>(sql: string, params?: any[], opts?: { preferLeader?: boolean }): Promise<T[]> {
  const pool = opts?.preferLeader ? leaderPool : followerPool;
  try {
    const result = await pool.query(sql, params);
    return result.rows;
  } catch (err) {
    if (pool === followerPool) {
      console.warn('Follower unavailable, falling back to leader');
      const result = await leaderPool.query(sql, params);
      return result.rows;
    }
    throw err;
  }
}
 
async function createOrder(userId: string, items: any[]) {
  const client = await leaderPool.connect();
  try {
    await client.query('BEGIN');
    const order = await client.query(
      'INSERT INTO orders (user_id, items) VALUES ($1, $2) RETURNING *',
      [userId, JSON.stringify(items)]
    );
    await client.query('COMMIT');
    return order.rows[0];
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

-- PostgreSQL logical replication conflict handling
-- On subscriber, conflicts cause an error. You can configure behavior:
ALTER SUBSCRIPTION my_sub SET (disable_on_error = false);
 
-- For PostgreSQL 16+, you can use origin filtering:
CREATE SUBSCRIPTION my_sub
  CONNECTION 'host=publisher port=5432 dbname=mydb'
  PUBLICATION my_pub
  WITH (origin = 'none');  -- Don't replicate changes from other origins
 
-- Monitor conflicts on subscriber
SELECT * FROM pg_stat_subscription;

// G-Counter CRDT example for distributed page view counting
class GCounter {
  private counts: Map<string, number> = new Map();
 
  increment(nodeId: string, amount: number = 1): void {
    const current = this.counts.get(nodeId) || 0;
    this.counts.set(nodeId, current + amount);
  }
 
  value(): number {
    let total = 0;
    for (const count of this.counts.values()) {
      total += count;
    }
    return total;
  }
 
  merge(other: GCounter): void {
    for (const [nodeId, count] of other.counts) {
      const current = this.counts.get(nodeId) || 0;
      this.counts.set(nodeId, Math.max(current, count));
    }
  }
}
 
// Usage: Each datacenter increments its own counter
const nycCounter = new GCounter();
const londonCounter = new GCounter();
 
nycCounter.increment('nyc', 42);
londonCounter.increment('london', 37);
 
// Periodically merge counters across datacenters
nycCounter.merge(londonCounter);
console.log(nycCounter.value()); // 79

-- MySQL GTID replication setup
-- On source (my.cnf):
[mysqld]
server-id = 1
log-bin = mysql-bin
gtid_mode = ON
enforce_gtid_consistency = ON
binlog_format = ROW
 
-- On replica (my.cnf):
[mysqld]
server-id = 2
relay-log = relay-bin
gtid_mode = ON
enforce_gtid_consistency = ON
read_only = ON
super_read_only = ON
 
-- Configure replication on replica
CHANGE REPLICATION SOURCE TO
  SOURCE_HOST = 'source_host',
  SOURCE_USER = 'repl_user',
  SOURCE_PASSWORD = 'repl_password',
  SOURCE_AUTO_POSITION = 1;
 
START REPLICA;
SHOW REPLICA STATUS\G

-- MySQL Group Replication setup
[mysqld]
server_id = 1
gtid_mode = ON
enforce_gtid_consistency = ON
binlog_checksum = NONE
log_bin = mysql-bin
binlog_format = ROW
master_info_repository = TABLE
relay_log_info_repository = TABLE
transaction_write_set_extraction = XXHASH64
loose-group_replication_group_name = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
loose-group_replication_start_on_boot = OFF
loose-group_replication_local_address = "node1:33061"
loose-group_replication_group_seeds = "node1:33061,node2:33061,node3:33061"
loose-group_replication_single_primary_mode = ON

-- On the publisher (old server)
CREATE PUBLICATION migration_pub FOR ALL TABLES;
 
-- On the subscriber (new server)
CREATE SUBSCRIPTION migration_sub
CONNECTION 'host=old_server port=5432 dbname=myapp user=replicator'
PUBLICATION migration_pub;
 
-- Monitor subscription status
SELECT
  subname,
  worker_type,
  received_lsn,
  last_msg_send_time,
  last_msg_receipt_time,
  latest_end_lsn
FROM pg_stat_subscription;
 
-- After initial sync completes, switch application to new server
-- Then drop the subscription
ALTER SUBSCRIPTION migration_sub DISABLE;
ALTER SUBSCRIPTION migration_sub SET (slot_name = NONE);
DROP SUBSCRIPTION migration_sub;

-- Replicate only specific tables
CREATE PUBLICATION order_pub FOR TABLE orders, order_items, payments;
 
-- Row-level filtering (PostgreSQL 15+)
CREATE PUBLICATION active_users_pub FOR TABLE users
  WHERE (active = true);
 
-- Column lists (PostgreSQL 15+)
CREATE PUBLICATION user_emails_pub FOR TABLE users (id, email, created_at);

const regionalPools = {
  'us-east': new Pool({ host: 'db-us-east.internal', max: 30 }),
  'eu-west': new Pool({ host: 'db-eu-west.internal', max: 30 }),
  'ap-southeast': new Pool({ host: 'db-ap-southeast.internal', max: 30 }),
};
 
function getRegionalPool(region: string): Pool {
  return regionalPools[region] || regionalPools['us-east'];
}
 
async function getProductCatalog(region: string) {
  const pool = getRegionalPool(region);
  return pool.query(
    `SELECT p.*, i.quantity FROM products p
     JOIN inventory i ON i.product_id = p.id
     WHERE i.warehouse_region = $1 ORDER BY p.popularity DESC`,
    [region]
  );
}

-- Synchronous replication to primary standby
synchronous_standby_names = 'FIRST 1 (standby_sync)'
 
-- Monitor replication compliance
CREATE OR REPLACE FUNCTION check_replication_compliance()
RETURNS TABLE (replica_name TEXT, lag_bytes BIGINT, compliant BOOLEAN) AS $$
BEGIN
  RETURN QUERY
  SELECT
    application_name::TEXT,
    pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn),
    pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn) < 1048576
  FROM pg_stat_replication;
END;
$$ LANGUAGE plpgsql;

-- Tune WAL for replication throughput
wal_buffers = '64MB'
wal_writer_delay = '200ms'
wal_compression = on  -- Reduces network bandwidth by 50-70%
 
-- PostgreSQL 15+ allows lz4 compression
wal_compression = lz4
 
-- On followers: tune for read performance
shared_buffers = '8GB'
effective_cache_size = '24GB'
hot_standby_feedback = on
 
-- Prevent long-running queries from blocking WAL replay
max_standby_streaming_delay = '30s'
max_standby_archive_delay = '60s'
 
-- Monitor replication performance
SELECT
  application_name, state,
  pg_size_pretty(pg_wal_lsn_diff(sent_lsn, write_lsn)) as write_lag,
  pg_size_pretty(pg_wal_lsn_diff(sent_lsn, flush_lsn)) as flush_lag,
  pg_size_pretty(pg_wal_lsn_diff(sent_lsn, replay_lsn)) as replay_lag
FROM pg_stat_replication;

-- 1. Replication lag in bytes and time
SELECT
  application_name,
  client_addr,
  state,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn)) as replay_lag,
  now() - pg_last_xact_replay_timestamp() as time_lag
FROM pg_stat_replication;
 
-- 2. Replication slot health
SELECT
  slot_name,
  active,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as retained_wal,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) as lag
FROM pg_replication_slots
ORDER BY retained_wal DESC;
 
-- 3. WAL generation rate (bytes per second over last 5 minutes)
SELECT
  pg_size_pretty(
    pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0') /
    EXTRACT(EPOCH FROM (now() - pg_postmaster_start_time()))
  ) as avg_wal_rate_per_sec;
 
-- 4. Alert on missing synchronous replica
SELECT
  CASE WHEN count(*) = 0 THEN 'CRITICAL: No synchronous replica connected'
       ELSE 'OK: ' || count(*) || ' synchronous replica(s)'
  END as sync_status
FROM pg_stat_replication
WHERE state = 'streaming'
  AND pg_wal_lsn_diff(sent_lsn, replay_lsn) = 0;

// Prometheus metrics for replication monitoring
import { register, Gauge } from 'prom-client';
 
const replicationLagBytes = new Gauge({
  name: 'pg_replication_lag_bytes',
  help: 'Replication lag in bytes',
  labelNames: ['replica_name', 'replica_host'],
});
 
const replicationLagSeconds = new Gauge({
  name: 'pg_replication_lag_seconds',
  help: 'Replication lag in seconds',
  labelNames: ['replica_name', 'replica_host'],
});
 
const slotRetainedWal = new Gauge({
  name: 'pg_slot_retained_wal_bytes',
  help: 'WAL bytes retained by replication slot',
  labelNames: ['slot_name'],
});
 
async function collectReplicationMetrics() {
  const result = await leaderPool.query(`
    SELECT
      application_name,
      client_addr,
      pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn) as lag_bytes
    FROM pg_stat_replication
  `);
 
  for (const row of result.rows) {
    replicationLagBytes.set(
      { replica_name: row.application_name, replica_host: row.client_addr },
      row.lag_bytes
    );
  }
 
  // Alert threshold: > 100MB lag
  for (const row of result.rows) {
    if (row.lag_bytes > 100 * 1024 * 1024) {
      console.error(`ALERT: ${row.application_name} lag ${row.lag_bytes} bytes`);
    }
  }
}
 
setInterval(collectReplicationMetrics, 15000);

describe('Replication Tests', () => {
  test('data replicates within acceptable lag', async () => {
    const testId = `test-${Date.now()}`;
    await leader.query('INSERT INTO test_table (id, data) VALUES ($1, $2)', [testId, 'test']);
 
    const maxWait = 5000;
    const start = Date.now();
    let found = false;
 
    while (Date.now() - start < maxWait) {
      const result = await follower.query('SELECT * FROM test_table WHERE id = $1', [testId]);
      if (result.rows.length > 0) { found = true; break; }
      await new Promise(resolve => setTimeout(resolve, 100));
    }
    expect(found).toBe(true);
  });
 
  test('follower rejects writes', async () => {
    await expect(
      follower.query('INSERT INTO test_table (id) VALUES ($1)', ['should-fail'])
    ).rejects.toThrow(/read only/);
  });
 
  test('failover promotes replica and demotes old leader', async () => {
    // Trigger failover via Patroni REST API
    const res = await fetch('http://patroni-node2:8008/failover', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ leader: 'node1', candidate: 'node2' }),
    });
    expect(res.ok).toBe(true);
 
    // Wait for failover to complete
    await new Promise(resolve => setTimeout(resolve, 15000));
 
    // Verify node2 is now the leader
    const status = await fetch('http://patroni-node2:8008/cluster');
    const cluster = await status.json();
    expect(cluster.leader).toBe('node2');
  });
});