​

Chapter 5: Replication

​

Introduction

​

Why replicate data?

1. High availability: Keep the system running even if some parts fail
   • If one machine fails, the system can continue operating using other replicas
   • Critical for services that need 99.9% or higher uptime
   • Example: If Amazon’s database goes down in one region, replicas in other regions keep the site running
2. Reduced latency: Keep data geographically close to users
   • Users in Tokyo can read from a replica in Asia rather than waiting for data from a server in the US
   • This can reduce latency from 200ms to 20ms or less
   • Example: Netflix stores copies of popular movies in servers close to major population centers
3. Increased read throughput: Scale out the number of machines that can serve read queries
   • One machine might handle 1,000 queries/second, but 10 replicas can handle 10,000 queries/second
   • Read-heavy applications (like social media feeds) can distribute load across many replicas
   • Example: Twitter uses thousands of replicas to serve billions of timeline reads per day
4. Disaster recovery: Protect against catastrophic failures
   • If an entire datacenter is destroyed (fire, flood, power outage), data survives in other locations
   • Regulatory compliance may require data to be stored in multiple geographic locations

​

The fundamental challenge

​

Leaders and followers (single-leader replication)

​

How it works

1. One replica is designated as the leader (master/primary)
   • The leader is the authoritative source of truth
   • Only the leader accepts write operations
2. All writes go to the leader first
   • When a client wants to write to the database, it must send the request to the leader
   • The leader writes the new data to its local storage
3. The leader sends data changes to all followers (replicas/slaves/secondaries)
   • Changes are sent as part of a replication log or change stream
   • This happens either synchronously or asynchronously (see below)
4. Followers apply the changes in the same order as they were processed on the leader
   • This ensures that followers eventually have the same data as the leader
   • Order is critical - applying changes in wrong order could lead to incorrect state
5. Reads can be served from the leader or any follower
   • This allows read scaling - add more followers to handle more read queries
   • Reading from followers may return slightly stale data (eventual consistency)

​

Example: PostgreSQL streaming replication

-- On the leader (primary)
-- All write operations happen here
INSERT INTO users (name, email) VALUES ('Alice', '[email protected]');
UPDATE users SET email = '[email protected]' WHERE name = 'Alice';

-- The leader streams WAL (Write-Ahead Log) to followers
-- Followers apply these changes asynchronously

​

Synchronous vs. asynchronous replication

​

Synchronous replication

1. Client sends write to leader
2. Leader writes to its local storage
3. Leader sends change to followers
4. Leader waits for acknowledgment from follower(s)
5. Only after receiving acknowledgment, leader confirms success to client

• Durability guarantee: Follower guaranteed to have up-to-date copy consistent with leader
• If leader fails immediately after acknowledging write, the data is safe on follower
• Strong consistency for reads from synchronous followers

• Latency: Write is slower (network round trip to follower)
• Availability: Write is blocked if synchronous follower doesn’t respond
  • If follower crashes or network fails, writes cannot be processed
  • System becomes unavailable for writes until follower recovers
• Not practical to make all followers synchronous (one slow node blocks all writes)

​

Asynchronous replication

1. Client sends write to leader
2. Leader writes to its local storage
3. Leader immediately confirms success to client
4. Leader sends change to followers in the background (fire and forget)
5. Followers apply changes when they receive them

• Performance: Write latency is minimal (only limited by leader’s local write)
• Availability: Leader can continue processing writes even if all followers are down
• Can have many followers without impacting write performance
• Common choice for read-heavy workloads where eventual consistency is acceptable

• Durability risk: Not all followers guaranteed to have latest data if leader fails
  • If leader fails before followers receive recent writes, those writes are lost
  • Could lose seconds or minutes of data
• Replication lag: Followers may be seconds, minutes, or even hours behind
• Reads from followers return stale data

​

Semi-synchronous replication (best of both worlds)

• One follower is synchronous (often called the “synchronous standby”)
• All other followers are asynchronous
• If the synchronous follower becomes unavailable, one of the asynchronous followers is promoted to synchronous

• Guarantees at least two up-to-date copies (leader + one synchronous follower)
• Minimal data loss risk (at most one node’s worth of recent writes)
• Better performance than fully synchronous (not waiting for all followers)
• Better durability than fully asynchronous (one follower always up-to-date)

-- postgresql.conf
synchronous_standby_names = 'standby1'  -- One synchronous standby
-- Any other standbys are asynchronous

​

Setting up new followers

1. Take a consistent snapshot of the leader’s database at some point in time
   • Most databases have built-in snapshot features
   • The snapshot must be consistent (represent the database at a single point in time, not halfway through a write)
   • Example: PostgreSQL pg_basebackup, MySQL mysqldump, MongoDB mongodump
2. Copy the snapshot to the new follower node
   • Transfer the snapshot file(s) over the network
   • This may take hours or days for large databases
   • During this time, the leader continues accepting writes
3. Follower connects to leader and requests all changes since the snapshot
   • The snapshot is associated with an exact position in the leader’s replication log
   • Position identified by log sequence number (LSN), binlog coordinates, or replica set optime
   • Follower says: “Give me all changes since position X”
4. Follower processes the backlog of changes
   • Applies all writes that happened since the snapshot
   • “Catches up” to the leader’s current state
   • This is called catch-up recovery
5. Follower is now ready to serve reads
   • Once caught up, continues to receive and apply ongoing changes
   • Can now handle read queries

# PostgreSQL example
# On new follower machine:
pg_basebackup -h primary.db.example.com -D /var/lib/postgresql/data -U replication -Fp -Xs -P -R

# This command:
# 1. Takes a snapshot from the primary (-h primary.db.example.com)
# 2. Copies it to local storage (-D /var/lib/postgresql/data)
# 3. Includes WAL files needed for consistency (-Xs)
# 4. Shows progress (-P)
# 5. Sets up replication configuration (-R)
# 6. Follower automatically starts catching up

• The process must be entirely automatic (can’t afford hours of human intervention)
• Snapshot + catch-up can take a long time for multi-terabyte databases
• Some systems support “online” or “hot” backup (snapshot while database is running)
• Network bandwidth between leader and follower impacts catch-up time

​

Handling node outages

​

Follower failure: Catch-up recovery

1. Each follower keeps a log of data changes received from the leader on local disk
2. When a follower crashes and restarts (or network is restored after interruption), it knows the last transaction it processed
3. It connects to the leader and requests all changes since that point
4. It applies the backlog of changes (catch-up recovery)
5. Once caught up, it continues to receive ongoing changes

• It requests: “Give me all changes since transaction #1000”
• Leader sends transactions #1001 through #1500 (current position)
• Follower applies these 500 transactions
• Now caught up, continues normal operation

​

Leader failure: Failover

• Manually: Administrator decides which follower becomes new leader and reconfigures the system
• Automatically: System detects failure and performs failover automatically

1. Detecting that the leader has failed
   • No foolproof way to detect failure (cannot distinguish between slow response and crash)
   • Typically use timeout-based approach: nodes bounce heartbeat messages, if no response for X seconds (e.g., 30s), leader is considered dead
   • Must be careful: too short timeout causes unnecessary failovers, too long means longer downtime
2. Choosing a new leader
   • Could be done through election (remaining replicas elect a leader)
   • Or use a previously elected “standby” leader
   • Best candidate is usually the replica with the most up-to-date data changes from old leader
   • Getting consensus on new leader is a problem in itself (see Chapter 9 on consensus)
3. Reconfiguring the system to use the new leader
   • Clients need to send writes to the new leader
   • If old leader comes back, it must recognize that it’s no longer the leader (become a follower)
   • System needs to ensure old leader doesn’t believe it’s still the leader (split brain problem)

3. Choosing the right timeout
   • Too short: Unnecessary failovers from temporary network glitches or load spikes
     • Failover causes disruption (clients need to reconnect, system reconfiguration)
     • Can make problem worse if system is already overloaded
   • Too long: Longer recovery time, more downtime
     • Users see errors for duration of timeout
     • May violate SLA (Service Level Agreement)
   Typical values: 30-60 seconds, but depends on application
   • Financial systems: Very short (seconds)
   • Less critical systems: Longer (minutes)
4. Scenarios leading to false positives
   • Sudden load spike causes leader to respond slowly
   • Network congestion delays heartbeat messages
   • Garbage collection pause in Java/JVM-based systems (can be 10+ seconds)
   • These aren’t failures, but can trigger failover

• Allows human judgment (is this really a failure or just a slow response?)
• Prevents cascade failures from automatic actions
• Can verify data consistency before promoting follower
• Less risk of split brain scenarios

​

Replication logs implementation

​

Statement-based replication

-- Client executes on leader:
INSERT INTO users (name, email) VALUES ('Alice', '[email protected]');

-- Leader writes to replication log:
INSERT INTO users (name, email) VALUES ('Alice', '[email protected]');

-- Followers receive and execute the same statement:
INSERT INTO users (name, email) VALUES ('Alice', '[email protected]');

1. Non-deterministic functions produce different values on each replica
   -- This will insert different values on each replica!
   INSERT INTO logs (timestamp, message) VALUES (NOW(), 'User logged in');
   
   -- NOW() returns different time on leader vs follower
   -- RAND() returns different random values
   -- UUID() generates different IDs
   
   Solution: Leader can replace non-deterministic functions with fixed values before sending to followers
2. Statements with auto-incrementing columns
   INSERT INTO users (name) VALUES ('Bob');  -- Auto-generates ID
   
   -- What if there are concurrent writes on the leader?
   -- Followers must execute in exactly the same order
   -- Otherwise, auto-increment values will differ
   
   Solution: Strict ordering requirement - followers must execute in exact same order as leader
3. Side effects (triggers, stored procedures, user-defined functions)
   -- This trigger updates another table
   CREATE TRIGGER update_count AFTER INSERT ON users
     FOR EACH ROW UPDATE stats SET user_count = user_count + 1;
   
   -- If trigger uses non-deterministic function, replicas diverge
   -- If trigger behavior changed on follower (version mismatch), problems arise
   

​

Write-ahead log (WAL) shipping

• “In disk block 123, at offset 456, write these bytes: 0x4A6F686E…”
• Very detailed, byte-level modifications

• Exact replica: followers’ storage is byte-for-byte identical to leader
• No ambiguity about non-deterministic functions

• Tightly coupled to storage engine: WAL describes data at a very low level (disk blocks, offsets)
• If storage engine format changes between versions, can’t replicate between different software versions
• Makes zero-downtime upgrades difficult (can’t have leader on v1 and follower on v2)
• If you want to change database system (migrate from PostgreSQL 10 to 11), often need downtime

# Follower receives WAL files from leader
# Files named like: 000000010000000000000001
# Contain binary log of all database changes

​

Logical (row-based) log replication

• For an inserted row: log contains new values of all columns
• For a deleted row: log contains information to uniquely identify the row (typically primary key)
• For an updated row: log contains information to identify the row + new values of all (or changed) columns

-- Operation: INSERT INTO users (id, name, email) VALUES (1, 'Alice', '[email protected]')

-- Logical log entry:
{
  "operation": "INSERT",
  "table": "users",
  "values": {
    "id": 1,
    "name": "Alice",
    "email": "[email protected]"
  }
}

-- Operation: UPDATE users SET email = '[email protected]' WHERE id = 1

-- Logical log entry:
{
  "operation": "UPDATE",
  "table": "users",
  "key": {"id": 1},
  "new_values": {
    "email": "[email protected]"
  }
}

• Decoupled from storage engine: Leader and followers can run different storage engines or even different database systems
• Backward compatible: Easier to make replication compatible across different software versions
• External systems can parse it: Data warehouses, caches, search indexes can consume the log
  • Example: Debezium, Maxwell’s Daemon capture MySQL binlog and stream to Kafka
  • Example: LinkedIn’s Databus captures Oracle’s logical log

# Application reads MySQL binlog and updates Elasticsearch
for change in mysql_binlog_stream:
    if change.table == 'users':
        if change.operation == 'INSERT':
            elasticsearch.index('users', change.values)
        elif change.operation == 'UPDATE':
            elasticsearch.update('users', change.key, change.new_values)
        elif change.operation == 'DELETE':
            elasticsearch.delete('users', change.key)

​

Trigger-based replication

-- Set up trigger to log changes
CREATE TRIGGER users_replication_trigger
AFTER INSERT OR UPDATE OR DELETE ON users
FOR EACH ROW
BEGIN
    INSERT INTO replication_log (table_name, operation, row_data, timestamp)
    VALUES ('users', TG_OP, row_to_json(NEW), now());
END;

-- External process reads replication_log table and replicates changes

• Very flexible: Can replicate subset of data, transform data, replicate to different databases
• Application-level control: Can add custom logic (validation, filtering, transformation)

• Greater overhead: Triggers run for every write, slowing down writes
• More error-prone: Application code is more likely to have bugs than database built-in replication
• Complexity: Need to maintain trigger code and replication process

-- Only replicate certain columns for privacy
CREATE TRIGGER replicate_user_analytics
AFTER INSERT OR UPDATE ON users
FOR EACH ROW
    -- Replicate to analytics DB, but exclude sensitive fields
    INSERT INTO analytics_db.users (id, country, signup_date)
    VALUES (NEW.id, NEW.country, NEW.signup_date);
    -- Exclude: email, password, credit_card

​

Problems with replication lag

• The system is operating near capacity
• There’s a network problem
• A follower is recovering from a failure

​

Read-after-write consistency (read-your-writes consistency)

1. Always read user’s own writes from the leader
   def read_user_data(user_id, requesting_user_id):
       # If user is reading their own data, read from leader
       if user_id == requesting_user_id:
           return read_from_leader(user_id)
       else:
           return read_from_follower(user_id)
   
   • Simple and effective for user-generated content
   • Problem: If most things are editable by users, most reads go to leader (negates benefit of read replicas)
2. Track timestamp of last update; for some time after update, read from leader
   def read_user_profile(user_id, last_write_timestamp):
       # Read from leader if user recently wrote data
       if time.now() - last_write_timestamp < 60:  # 1 minute threshold
           return read_from_leader(user_id)
       else:
           return read_from_follower(user_id)
   
   • Good balance: reads from leader only when necessary
   • Requires tracking last write time (can store in client session or cookie)
3. Client remembers timestamp of most recent write
   • Most sophisticated approach
   • System ensures replica serving read has updates ≥ timestamp of user’s last write
   • Can use logical timestamps (like log sequence numbers) rather than wall-clock time
4. If replicas are distributed across datacenters, additional complexity:
   • User might be routed to different datacenters for different requests
   • Any replica that serves user reads must be in same datacenter as leader, or must reliably route to leader
   • Example: User in Europe writes (goes to Europe leader), then mobile app reads (might route to US datacenter)

• Approach 1 doesn’t work (one device doesn’t know about writes on other device)
• Metadata for last write timestamp must be centralized
• If replicas are in different datacenters, harder to guarantee routing to same datacenter from different devices

​

Monotonic reads

def get_replica_for_user(user_id, replicas):
    # Hash user ID to consistently choose same replica
    replica_index = hash(user_id) % len(replicas)
    return replicas[replica_index]

def read_data(user_id, key):
    # User always routed to same replica
    replica = get_replica_for_user(user_id, REPLICAS)
    return replica.read(key)

• Simple to implement (just hash user ID)
• Each user’s view of data is consistent (never goes backward)
• Load is distributed across replicas

• If a replica fails, users assigned to it must be rerouted (temporarily see older data)
• Not as strong as “read after write” consistency, but weaker guarantee is often sufficient

​

Consistent prefix reads

Mr. Poons: "What's the weather like?"
Mrs. Cake: "It's sunny outside!"

Mrs. Cake: "It's sunny outside!"
Mr. Poons: "What's the weather like?"

• Different partitions operate independently with no ordering guarantee across partitions
• Replication lag varies between partitions
• Some partitions might be slower (more load, slower disk, network issues)

# Use conversation thread ID for partitioning, not user ID
def write_message(thread_id, user_id, message):
    # All messages in same thread go to same partition
    partition = hash(thread_id) % NUM_PARTITIONS
    db.write(partition=partition, user=user_id, message=message)

# Now conversation order is preserved within partition

• Each message includes vector of “happened before” relationships
• System ensures messages aren’t shown until dependencies are met

• Some systems (like Riak) provide causal consistency by tracking causal relationships
• More complex but provides stronger guarantees

​

Multi-leader replication

• Leader becomes a bottleneck for writes
• If you want to have datacenters in multiple geographic regions, all writes must route to one datacenter
• If the leader fails, system is unavailable for writes until failover completes

​

Use cases