Database Read Replicas: Scaling Read Workloads

┌─────────┐     Writes      ┌─────────┐
│         │────────────────▶│ Primary │
│ Client  │                 │  DB     │────┐
│         │────────────────▶│         │    │
└─────────┘     Reads       └─────────┘    │
                  │                      │ Replication
                  ▼                      │
           ┌──────────┐                 │
           │ Replica 1│◀────────────────┘
           └──────────┘                  │
           ┌──────────┐                  │
           │ Replica 2│◀─────────────────┘
           └──────────┘

-- Primary: postgresql.conf
wal_level = replica
max_wal_senders = 5
wal_keep_size = 1GB

-- Primary: pg_hba.conf
host replication replicator 10.0.0.0/8 md5

-- Create replication user
CREATE USER replicator REPLICATION LOGIN PASSWORD 'secure_password';

-- Replica: recovery.conf
standby_mode = 'on'
primary_conninfo = 'host=primary-db.example.com port=5432 user=replicator password=secure_password'

-- Primary: my.cnf
[mysqld]
server-id = 1
log-bin = mysql-bin
binlog-format = ROW
expire_logs_days = 7

-- Create replication user
CREATE USER 'repl'@'%' IDENTIFIED BY 'password';
GRANT REPLICATION SLAVE ON *.* TO 'repl'@'%';

-- Show master status
SHOW MASTER STATUS;

class DatabaseRouter {
  private primary: Connection;
  private replicas: Connection[];
  
  constructor(primary: Connection, replicas: Connection[]) {
    this.primary = primary;
    this.replicas = replicas;
  }
  
  getReadConnection(): Connection {
    // Round-robin replica selection
    const index = Math.floor(Math.random() * this.replicas.length);
    return this.replicas[index];
  }
  
  getWriteConnection(): Connection {
    return this.primary;
  }
  
  async query(sql: string, params?: any[]): Promise<any> {
    // Route reads to replicas, writes to primary
    const isWrite = /^(INSERT|UPDATE|DELETE|CREATE|ALTER)/i.test(sql.trim());
    const conn = isWrite ? this.getWriteConnection() : this.getReadConnection();
    return conn.query(sql, params);
  }
}

class LagAwareRouter extends DatabaseRouter {
  async query(sql: string, params?: any[], options?: { maxLag?: number }): Promise<any> {
    const maxLag = options?.maxLag || 1000; // 1 second default
    
    const isWrite = /^(INSERT|UPDATE|DELETE)/i.test(sql.trim());
    
    if (isWrite) {
      return this.primary.query(sql, params);
    }
    
    // Check replica lag
    const replica = this.getReadConnection();
    const lag = await this.getReplicationLag(replica);
    
    if (lag > maxLag) {
      // Fallback to primary if replica is too far behind
      console.warn(`Replica lag ${lag}ms exceeds threshold, using primary`);
      return this.primary.query(sql, params);
    }
    
    return replica.query(sql, params);
  }
  
  private async getReplicationLag(replica: Connection): Promise<number> {
    const result = await replica.query('SHOW REPLICA STATUS');
    return result[0].replica_lag;
  }
}

// Route user to same replica during session
class SessionAwareRouter extends DatabaseRouter {
  private sessionMap = new Map<string, Connection>();
  
  getReadConnection(sessionId?: string): Connection {
    if (sessionId && this.sessionMap.has(sessionId)) {
      return this.sessionMap.get(sessionId)!;
    }
    
    const replica = this.selectReplica();
    if (sessionId) {
      this.sessionMap.set(sessionId, replica);
    }
    return replica;
  }
}

class ReadAfterWriteRouter extends DatabaseRouter {
  private writeTimestamp = new Map<string, number>();
  
  async write(sql: string, params?: any[]): Promise<any> {
    const result = await this.primary.query(sql, params);
    this.writeTimestamp.set('last_write', Date.now());
    return result;
  }
  
  async read(sql: string, params?: any[]): Promise<any> {
    const lastWrite = this.writeTimestamp.get('last_write') || 0;
    const timeSinceWrite = Date.now() - lastWrite;
    
    // Use primary if write was recent (within replication lag window)
    if (timeSinceWrite < 1000) {
      return this.primary.query(sql, params);
    }
    
    return this.getReadConnection().query(sql, params);
  }
}

import Prometheus from 'prom-client';

const replicaLag = new Prometheus.Gauge({
  name: 'database_replica_lag_seconds',
  help: 'Replication lag in seconds',
  labelNames: ['replica']
});

const connectionPool = new Prometheus.Gauge({
  name: 'database_connection_pool_size',
  help: 'Connection pool size',
  labelNames: ['database', 'type']
});

// Update metrics periodically
setInterval(async () => {
  for (const replica of replicas) {
    const lag = await getReplicationLag(replica);
    replicaLag.labels({ replica: replica.name }).set(lag);
  }
}, 5000);