# MongoDB CP behaviortry:    db.accounts.update_one(        {"user_id": "alice"},        {"$inc": {"balance": -100}},        write_concern=WriteConcern(w="majority")  # Wait for majority    )except NetworkTimeout:    # Cannot reach majority → Reject write    return "Service temporarily unavailable"

# Cassandra AP behaviorsession.execute(    "UPDATE users SET last_login = ? WHERE user_id = ?",    [datetime.now(), "alice"],    consistency_level=ConsistencyLevel.ONE  # Respond immediately)# Returns success even if some replicas are unreachable

# PostgreSQL Snapshot Isolation (REPEATABLE READ)BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;# T1: Read account balance at snapshot time t=100SELECT balance FROM accounts WHERE id = 'alice';  # Returns $500# T2: (Another transaction updates balance to $600 at t=101)# T3: Read again in same transactionSELECT balance FROM accounts WHERE id = 'alice';  # Still returns $500# Transaction sees consistent snapshot!COMMIT;

# Doctor on-call scheduling (write skew example)# Constraint: At least 1 doctor must be on-call# Currently: Alice and Bob both on-call# Transaction 1 (Alice):BEGIN;SELECT COUNT(*) FROM on_call WHERE on_call = true;  # Returns 2# Alice sees Bob is on-call, so she can leaveUPDATE on_call SET on_call = false WHERE doctor = 'Alice';COMMIT;# Transaction 2 (Bob) - concurrent:BEGIN;SELECT COUNT(*) FROM on_call WHERE on_call = true;  # Returns 2# Bob sees Alice is on-call, so he can leaveUPDATE on_call SET on_call = false WHERE doctor = 'Bob';COMMIT;# RESULT: No doctors on-call! Constraint violated!# Snapshot isolation doesn't prevent this

# PostgreSQL Serializable IsolationBEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE;# Transaction automatically aborts if serialization anomaly detectedtry:    # Perform operations    cur.execute("SELECT COUNT(*) FROM on_call WHERE on_call = true")    count = cur.fetchone()[0]    if count > 1:        cur.execute("UPDATE on_call SET on_call = false WHERE doctor = 'Alice'")    conn.commit()except SerializationFailure:    # Database detected write skew, retry transaction    conn.rollback()    retry_transaction()

# PostgreSQL synchronous replication# postgresql.confsynchronous_commit = onsynchronous_standby_names = 'replica1,replica2'# Write waits for replicasconn.execute("INSERT INTO users VALUES (...)")# Returns only after replicas confirm write

# MySQL async replication# Primary returns immediatelyconn.execute("INSERT INTO users VALUES (...)")# Replica receives update later (lag: milliseconds to seconds)

# Write to 3 out of 5 replicassession.execute(    "INSERT INTO users (id, name) VALUES (?, ?)",    [user_id, "Alice"],    consistency_level=ConsistencyLevel.QUORUM  # 3/5 nodes)# Read from 3 out of 5 replicasresult = session.execute(    "SELECT name FROM users WHERE id = ?",    [user_id],    consistency_level=ConsistencyLevel.QUORUM  # 3/5 nodes)# Returns most recent value (by timestamp)

# Leader receives client requestdef handle_write(data):    # 1. Append to local log    log_entry = LogEntry(        term=current_term,        index=last_log_index + 1,        data=data    )    append_to_log(log_entry)    # 2. Replicate to followers    for follower in followers:        send_append_entries(follower, [log_entry])    # 3. Wait for majority acknowledgment    acks = wait_for_acks(majority_count)    # 4. Commit entry (apply to state machine)    if acks >= majority_count:        commit_index = log_entry.index        apply_to_state_machine(log_entry)        return SUCCESS    return FAILURE  # Couldn't reach majority

class PaxosAcceptor:    def __init__(self):        self.promised_id = None  # Highest proposal ID promised        self.accepted_id = None  # ID of accepted proposal        self.accepted_value = None  # Accepted value    def prepare(self, proposal_id):        ""Phase 1b: Promise""        if self.promised_id is None or proposal_id > self.promised_id:            self.promised_id = proposal_id            return {                'promised': True,                'accepted_id': self.accepted_id,                'accepted_value': self.accepted_value            }        return {'promised': False}    def accept(self, proposal_id, value):        ""Phase 2b: Accept""        if self.promised_id is None or proposal_id >= self.promised_id:            self.promised_id = proposal_id            self.accepted_id = proposal_id            self.accepted_value = value            return {'accepted': True}        return {'accepted': False}class PaxosProposer:    def propose(self, value, acceptors):        proposal_id = generate_unique_id()        # Phase 1: Prepare        promises = []        for acceptor in acceptors:            response = acceptor.prepare(proposal_id)            if response['promised']:                promises.append(response)        # Need majority to proceed        if len(promises) < len(acceptors) // 2 + 1:            return None  # Failed to get quorum        # Check if any acceptor already accepted a value        # If so, must propose that value (Paxos invariant)        max_accepted_id = -1        proposed_value = value        for promise in promises:            if promise['accepted_id'] and promise['accepted_id'] > max_accepted_id:                max_accepted_id = promise['accepted_id']                proposed_value = promise['accepted_value']        # Phase 2: Accept        acceptances = []        for acceptor in acceptors:            response = acceptor.accept(proposal_id, proposed_value)            if response['accepted']:                acceptances.append(response)        # Value chosen if majority accepted        if len(acceptances) >= len(acceptors) // 2 + 1:            return proposed_value        return None  # Failed to get quorum

import consul# Consul uses Raft for consistent K/V storec = consul.Consul()# Writes go through Raft consensussuccess = c.kv.put('config/database/url', 'postgresql://...')# Guaranteed consistent readsindex, data = c.kv.get('config/database/url')# CAS (Compare-and-Set) operations atomic via Raftc.kv.put('counter', '0')index, current = c.kv.get('counter')# Only succeeds if modify_index matches (no concurrent updates)success = c.kv.put(    'counter',    str(int(current['Value']) + 1),    cas=current['ModifyIndex']  # Compare-and-swap)

# Cassandra with sloppy quorumclass SloppyQuorum:    def __init__(self, replication_factor=3):        self.N = replication_factor        self.W = 2  # QUORUM        self.R = 2  # QUORUM    def write(self, key, value):        # Determine preference list (N nodes)        nodes = self.get_preference_list(key, self.N)        # Try to write to N nodes        written = []        for node in nodes:            try:                node.put(key, value)                written.append(node)                # Stop once W nodes acknowledged                if len(written) >= self.W:                    return SUCCESS            except NodeDown:                continue        # If < W primary nodes available, use "hinted handoff"        # Write to extra nodes temporarily        if len(written) < self.W:            extra_nodes = self.get_extra_nodes(key)            for node in extra_nodes:                # Store hint: "This write belongs to node X"                node.put_with_hint(key, value, intended_node=nodes[0])                written.append(node)                if len(written) >= self.W:                    return SUCCESS        return FAILURE    def read(self, key):        nodes = self.get_preference_list(key, self.N)        # Read from R nodes        responses = []        for node in nodes:            try:                value = node.get(key)                responses.append(value)                if len(responses) >= self.R:                    break            except NodeDown:                continue        # Return most recent value (by timestamp)        return max(responses, key=lambda v: v.timestamp)

# WAL record structureclass WALRecord:    def __init__(self):        self.lsn = None          # Log Sequence Number (position in WAL)        self.xid = None          # Transaction ID        self.rmid = None         # Resource Manager ID        self.info = None         # Record type info        self.data = None         # Actual change data    def __repr__(self):        return f"WAL@{self.lsn}: XID={self.xid} {self.data}"# Example WAL records for a transaction:# WAL@0/1000: XID=100 BEGIN# WAL@0/1020: XID=100 INSERT INTO users (id, name) VALUES (1, 'Alice')# WAL@0/1040: XID=100 COMMIT

# Without fsync (DANGEROUS!)with open('wal_file', 'ab') as f:    f.write(wal_record)    # Data might still be in OS page cache    # Power loss = data loss!# With fsync (SAFE)import oswith open('wal_file', 'ab') as f:    f.write(wal_record)    f.flush()       # Flush Python buffer to OS    os.fsync(f.fileno())  # Force OS to write to physical disk    # Now guaranteed on disk!

# Primary configuration# postgresql.confwal_level = replica              # Generate enough WAL for replicationmax_wal_senders = 10             # Max concurrent replica connectionswal_keep_size = '1GB'            # Keep at least 1GB of WALwal_sender_timeout = 60s         # Disconnect dead replicas# Replica configuration# postgresql.confhot_standby = on                 # Allow read queries on replicamax_standby_streaming_delay = 30s# recovery.conf (or postgresql.auto.conf)primary_conninfo = 'host=primary.example.com port=5432 user=replicator password=...'primary_slot_name = 'replica_1'  # Replication slot (prevents WAL deletion)# Monitoring replication lagSELECT    client_addr,    state,    sent_lsn,    write_lsn,    flush_lsn,    replay_lsn,    sync_state,    pg_wal_lsn_diff(sent_lsn, replay_lsn) AS lag_bytesFROM pg_stat_replication;-- Example output:-- client_addr | state     | lag_bytes-- 10.0.1.5    | streaming | 2048      (2KB lag - excellent!)-- 10.0.1.6    | streaming | 5242880   (5MB lag - investigate)

# Problem: fsync is expensive (5-10ms per call)# Solution: Batch multiple transactions into one fsyncclass GroupCommit:    def __init__(self):        self.pending_txns = []        self.commit_delay = 0.001  # 1ms    async def commit_transaction(self, txn):        # Add to batch        self.pending_txns.append(txn)        # Wait for batch to fill or timeout        await asyncio.sleep(self.commit_delay)        # Flush all pending        if self.pending_txns:            batch = self.pending_txns            self.pending_txns = []            # Single fsync for entire batch!            write_and_fsync_wal(batch)            # All transactions durable            for t in batch:                t.notify_committed()# Result: 1000 commits/sec → 100,000 commits/sec!# Latency: +1ms

import boto3# Create Global Tabledynamodb = boto3.client('dynamodb')# Define replication groupresponse = dynamodb.create_global_table(    GlobalTableName='users',    ReplicationGroup=[        {'RegionName': 'us-east-1'},        {'RegionName': 'eu-west-1'},        {'RegionName': 'ap-southeast-1'}    ])# Writes to any region automatically replicate# Typical replication lag: < 1 seconddynamodb_us = boto3.resource('dynamodb', region_name='us-east-1')table = dynamodb_us.Table('users')# User in US writestable.put_item(Item={'user_id': 'alice', 'name': 'Alice'})# Available in EU within ~1 seconddynamodb_eu = boto3.resource('dynamodb', region_name='eu-west-1')table_eu = dynamodb_eu.Table('users')response = table_eu.get_item(Key={'user_id': 'alice'})# Returns {'user_id': 'alice', 'name': 'Alice'}

# T1: User updates name in UStable_us.update_item(    Key={'user_id': 'alice'},    UpdateExpression='SET #name = :name',    ExpressionAttributeNames={'#name': 'name'},    ExpressionAttributeValues={':name': 'Alice Smith'})# T2: Same user updates name in EU (concurrent!)table_eu.update_item(    Key={'user_id': 'alice'},    UpdateExpression='SET #name = :name',    ExpressionAttributeNames={'#name': 'name'},    ExpressionAttributeValues={':name': 'Alice Johnson'})# DynamoDB resolves conflict using timestamps# Later write wins (based on internal timestamp)# Final value: "Alice Johnson" (if EU write was newer)

import boto3rds = boto3.client('rds', region_name='us-east-1')# Create read replica in EUrds.create_db_instance_read_replica(    DBInstanceIdentifier='mydb-eu-replica',    SourceDBInstanceIdentifier='mydb-primary',    DBInstanceClass='db.r5.large',    AvailabilityZone='eu-west-1a',    PubliclyAccessible=False)# Application routing logicclass DatabaseRouter:    def __init__(self):        self.primary = "mydb-primary.us-east-1.rds.amazonaws.com"        self.replicas = {            "us": "mydb-primary.us-east-1.rds.amazonaws.com",            "eu": "mydb-eu-replica.eu-west-1.rds.amazonaws.com",            "ap": "mydb-ap-replica.ap-south-1.rds.amazonaws.com"        }    def get_write_connection(self):        # All writes to primary        return connect(self.primary)    def get_read_connection(self, user_region):        # Reads from nearest replica        replica = self.replicas.get(user_region, self.primary)        return connect(replica)# Usagerouter = DatabaseRouter()# Write (always primary)write_conn = router.get_write_connection()write_conn.execute("INSERT INTO users (name) VALUES ('Alice')")# Read (region-specific)read_conn = router.get_read_connection(user_region="eu")users = read_conn.execute("SELECT * FROM users WHERE active = true")

class LWWRegister:    def __init__(self):        self.value = None        self.timestamp = 0    def write(self, value, timestamp):        if timestamp > self.timestamp:            self.value = value            self.timestamp = timestamp    def merge(self, remote_value, remote_timestamp):        # Keep value with higher timestamp        if remote_timestamp > self.timestamp:            self.value = remote_value            self.timestamp = remote_timestamp# Usageregister_us = LWWRegister()register_eu = LWWRegister()# Concurrent writesregister_us.write("Alice Smith", timestamp=1000)register_eu.write("Alice Johnson", timestamp=1001)# Replication and mergeregister_us.merge("Alice Johnson", 1001)  # EU value winsregister_eu.merge("Alice Smith", 1000)    # US value discarded# Both converge to "Alice Johnson"

class VectorClock:    def __init__(self, node_id):        self.node_id = node_id        self.clock = {}    def increment(self):        self.clock[self.node_id] = self.clock.get(self.node_id, 0) + 1        return self.clock.copy()    def update(self, other_clock):        for node, count in other_clock.items():            self.clock[node] = max(self.clock.get(node, 0), count)    def compare(self, other_clock):        # Returns: "before", "after", "concurrent", or "equal"        less = False        greater = False        all_nodes = set(self.clock.keys()) | set(other_clock.keys())        for node in all_nodes:            self_count = self.clock.get(node, 0)            other_count = other_clock.get(node, 0)            if self_count < other_count:                less = True            elif self_count > other_count:                greater = True        if less and greater:            return "concurrent"  # Conflict!        elif less:            return "before"        elif greater:            return "after"        else:            return "equal"# Example: Detecting conflictsvc_us = VectorClock("US")vc_eu = VectorClock("EU")# Sequential writes (no conflict)clock1 = vc_us.increment()  # {US: 1}vc_eu.update(clock1)        # EU knows about US:1clock2 = vc_eu.increment()  # {US: 1, EU: 1}print(vc_us.compare(clock2))  # "before" (causal order)# Concurrent writes (conflict!)vc_us2 = VectorClock("US")vc_eu2 = VectorClock("EU")clock_us = vc_us2.increment()  # {US: 1}clock_eu = vc_eu2.increment()  # {EU: 1}print(vc_us2.compare(clock_eu))  # "concurrent" (CONFLICT!)

class HybridLogicalClock:    ""    HLC combines:    - Physical time (wall clock)    - Logical counter (for events within same physical time)    ""    def __init__(self):        self.latest_time = 0  # Physical time        self.logical = 0      # Logical counter    def send_event(self):        ""Generate timestamp for local event""        physical_time = int(time.time() * 1000000)  # Microseconds        if physical_time > self.latest_time:            # Physical time advanced            self.latest_time = physical_time            self.logical = 0        else:            # Multiple events in same physical time            self.logical += 1        return (self.latest_time, self.logical)    def receive_event(self, remote_time, remote_logical):        ""Update clock on receiving remote event""        physical_time = int(time.time() * 1000000)        # Take maximum of local and remote physical time        self.latest_time = max(physical_time, remote_time, self.latest_time)        if self.latest_time == physical_time and self.latest_time == remote_time:            # All three equal: increment logical            self.logical = max(self.logical, remote_logical) + 1        elif self.latest_time == physical_time:            # Local physical time wins            self.logical = max(self.logical, remote_logical) + 1        elif self.latest_time == remote_time:            # Remote time wins            self.logical = remote_logical + 1        else:            # latest_time from old local time            self.logical += 1        return (self.latest_time, self.logical)# Usage Examplenode1 = HybridLogicalClock()node2 = HybridLogicalClock()# Node 1 performs local writets1 = node1.send_event()  # (1638360000000000, 0)print(f"Node 1 writes at HLC: ${ts1}")# Node 2 receives message with ts1ts2 = node2.receive_event(ts1[0], ts1[1])  # (1638360000000000, 1)print(f"Node 2 receives, HLC: ${ts2}")# Node 2 writes locallyts3 = node2.send_event()  # (1638360000000000, 2)print(f"Node 2 writes at HLC: ${ts3}")# Timestamps maintain causality AND approximate wall-clock time!

# G-Counter (Grow-only Counter) - CRDTclass GCounter:    def __init__(self, node_id):        self.node_id = node_id        self.counts = {}    def increment(self, amount=1):        self.counts[self.node_id] = self.counts.get(self.node_id, 0) + amount    def value(self):        return sum(self.counts.values())    def merge(self, other):        # Merge is commutative and idempotent        for node, count in other.counts.items():            self.counts[node] = max(self.counts.get(node, 0), count)# Usage: Distributed page view countercounter_us = GCounter("US")counter_eu = GCounter("EU")counter_asia = GCounter("ASIA")# Concurrent incrementscounter_us.increment(100)    # 100 views in UScounter_eu.increment(50)     # 50 views in EUcounter_asia.increment(75)   # 75 views in Asia# Merge all counterscounter_us.merge(counter_eu)counter_us.merge(counter_asia)print(counter_us.value())  # 225 (no conflicts!)

class ShoppingCart:    def __init__(self):        self.items = {}  # product_id -> quantity    def add_item(self, product_id, quantity):        self.items[product_id] = self.items.get(product_id, 0) + quantity    def remove_item(self, product_id):        if product_id in self.items:            del self.items[product_id]    def merge(self, other_cart):        # Domain-specific merge logic        for product_id, quantity in other_cart.items.items():            if product_id in self.items:                # Take maximum quantity (user benefit)                self.items[product_id] = max(                    self.items[product_id],                    quantity                )            else:                # Add new item                self.items[product_id] = quantity# User adds items from multiple devices concurrentlycart_mobile = ShoppingCart()cart_mobile.add_item("shoe-123", 1)cart_desktop = ShoppingCart()cart_desktop.add_item("shirt-456", 2)cart_desktop.add_item("shoe-123", 2)  # Same product, different quantity!# Merge cartscart_mobile.merge(cart_desktop)# Result: shoe-123: 2, shirt-456: 2# Took maximum quantity for conflicting item

# Application codefrom cassandra.cluster import Clusterfrom cassandra import ConsistencyLevelcluster = Cluster(['cassandra.us-east.netflix.com'])session = cluster.connect('netflix')# Write with LOCAL_QUORUM (fast, regional consistency)session.execute(    ""    INSERT INTO user_viewing_history    (user_id, content_id, watched_at, position, device_id)    VALUES (?, ?, ?, ?, ?)    "",    [user_id, content_id, datetime.now(), position, device_id],    consistency_level=ConsistencyLevel.LOCAL_QUORUM  # 2/3 local nodes)# Async replication to other regions happens automatically# Viewing history available globally within seconds

# Schemaless architecture# Each row has: DocID, Timestamp, Payload (JSON)# Region-aware routingclass UberDocstore:    def __init__(self):        self.regions = {            'us': MySQLConnection('us-mysql-cluster'),            'eu': MySQLConnection('eu-mysql-cluster'),            'apac': MySQLConnection('apac-mysql-cluster')        }    def write(self, doc_id, payload, region='us'):        # Write to regional cluster        conn = self.regions[region]        conn.execute(            "INSERT INTO documents (doc_id, timestamp, payload) VALUES (?, ?, ?)",            [doc_id, time.time(), json.dumps(payload)]        )        # Async replication to other regions        self._async_replicate(doc_id, payload, source_region=region)    def read(self, doc_id, region='us'):        # Read from regional cluster        conn = self.regions[region]        result = conn.execute(            "SELECT payload FROM documents WHERE doc_id = ? ORDER BY timestamp DESC LIMIT 1",            [doc_id]        )        return json.loads(result[0]['payload'])# Usage: Driver location updatesdocstore = UberDocstore()# Driver in San Franciscodocstore.write(    doc_id="driver:12345",    payload={        "location": {"lat": 37.7749, "lon": -122.4194},        "status": "available",        "updated_at": "2025-11-29T10:30:00Z"    },    region="us")# Available in EU within ~100ms for cross-region matching

# Cassandra-style read repairclass ReadRepair:    def __init__(self, consistency_level='QUORUM'):        self.consistency_level = consistency_level    def read(self, key, replicas):        # Read from multiple replicas        responses = []        for replica in replicas[:3]:  # Read from 3 replicas            response = replica.get(key)            responses.append((replica, response))        # Find most recent value (by timestamp)        latest_value = max(responses, key=lambda r: r[1].timestamp)        # Check for inconsistencies        stale_replicas = []        for replica, response in responses:            if response.timestamp < latest_value[1].timestamp:                stale_replicas.append(replica)        # Repair stale replicas in background        if stale_replicas:            self.async_repair(key, latest_value[1], stale_replicas)        # Return latest value to client        return latest_value[1].data    async def async_repair(self, key, latest_value, stale_replicas):        ""Asynchronously update stale replicas""        for replica in stale_replicas:            try:                replica.put(key, latest_value)                print(f"Read repair: Updated ${replica} with latest value")            except Exception as e:                print(f"Read repair failed for ${replica}: ${e}")# Usagerepair = ReadRepair()# Client reads datavalue = repair.read(key='user:123', replicas=cluster.get_replicas('user:123'))# → Returns latest value# → Triggers background repair of stale replicas

# DON'T DO THIS - transfers entire datasetdef naive_sync(local_db, remote_db):    # Get all data from both databases    local_data = local_db.get_all()   # 1TB of data!    remote_data = remote_db.get_all() # 1TB of data!    # Compare and sync    for key, value in local_data.items():        if key not in remote_data or remote_data[key] != value:            remote_db.put(key, value)# Problem: Transfers gigabytes even if only 1 key differs!

import hashlibclass MerkleTree:    ""    Hierarchical hash tree for efficient comparison    ""    def __init__(self, key_ranges):        self.tree = {}        self.key_ranges = key_ranges    def build(self, data):        ""Build Merkle tree from data""        # Leaf nodes: hash each key range        for range_id, (start, end) in enumerate(self.key_ranges):            range_data = {k: v for k, v in data.items() if start <= k < end}            range_hash = self.hash_dict(range_data)            self.tree[f'leaf_{range_id}'] = range_hash        # Internal nodes: hash of children        self.tree['root'] = self.hash_children(            [self.tree[f'leaf_{i}'] for i in range(len(self.key_ranges))]        )    def hash_dict(self, data):        ""Hash dictionary of key-value pairs""        items = sorted(data.items())        return hashlib.md5(str(items).encode()).hexdigest()    def hash_children(self, child_hashes):        ""Hash of child node hashes""        combined = ''.join(child_hashes)        return hashlib.md5(combined.encode()).hexdigest()# Anti-entropy with Merkle treesdef merkle_sync(local_db, remote_db):    # Define key ranges (e.g., 256 ranges)    key_ranges = [(i * 1000, (i + 1) * 1000) for i in range(256)]    # Build Merkle trees    local_tree = MerkleTree(key_ranges)    remote_tree = MerkleTree(key_ranges)    local_tree.build(local_db.get_all())    remote_tree.build(remote_db.get_all())    # Compare roots    if local_tree.tree['root'] == remote_tree.tree['root']:        print("Databases in sync! No transfer needed.")        return    # Find differing ranges    differing_ranges = []    for range_id in range(len(key_ranges)):        local_hash = local_tree.tree[f'leaf_{range_id}']        remote_hash = remote_tree.tree[f'leaf_{range_id}']        if local_hash != remote_hash:            differing_ranges.append(range_id)    # Sync only differing ranges    for range_id in differing_ranges:        start, end = key_ranges[range_id]        range_data = {k: v for k, v in local_db.get_all().items() if start <= k < end}        # Transfer only this range        for key, value in range_data.items():            remote_db.put(key, value)    print(f"Synced ${len(differing_ranges)}/${len(key_ranges)} ranges")# Example: 1TB database, only 10MB different# Without Merkle: Transfer 1TB# With Merkle: Transfer ~10MB (100x savings!)

# Cassandra nodetool repair# Runs Merkle tree sync across replicas# Repair specific keyspacenodetool repair myapp# Repair workflow:# 1. Build Merkle trees on all replicas (in parallel)# 2. Exchange root hashes# 3. If roots differ, recurse into differing ranges# 4. Stream only differing SSTables# Monitoring repairnodetool compactionstats# Output:# pending tasks: 0# Active compaction:# - compaction_type: Validation (Merkle tree build)# - keyspace: myapp# - progress: 45%# Schedule periodic repairs (cron)# Recommended: Every 7-10 days0 2 * * 0 nodetool repair myapp  # Sunday 2AM

# DynamoDB uses continuous background anti-entropy# Merkle tree per partition# Trees rebuilt every 10 minutes# Automatic sync of differing ranges# No user action needed, but can monitor:import boto3cloudwatch = boto3.client('cloudwatch')# Check replication lag (indicator of anti-entropy load)response = cloudwatch.get_metric_statistics(    Namespace='AWS/DynamoDB',    MetricName='ReplicationLatency',    Dimensions=[        {'Name': 'TableName', 'Value': 'users'},        {'Name': 'ReceivingRegion', 'Value': 'us-west-2'}    ],    StartTime=datetime.now() - timedelta(hours=1),    EndTime=datetime.now(),    Period=300,    Statistics=['Average'])# High latency might indicate anti-entropy catching up

# PostgreSQL replication lag monitoringimport psycopg2import timedef check_replication_lag(primary_conn, replica_conn):    # Get WAL position from primary    primary_cur = primary_conn.cursor()    primary_cur.execute("SELECT pg_current_wal_lsn()")    primary_lsn = primary_cur.fetchone()[0]    # Get replay position from replica    replica_cur = replica_conn.cursor()    replica_cur.execute("SELECT pg_last_wal_replay_lsn()")    replica_lsn = replica_cur.fetchone()[0]    # Calculate lag in bytes    replica_cur.execute(        "SELECT pg_wal_lsn_diff(%s, %s)",        [primary_lsn, replica_lsn]    )    lag_bytes = replica_cur.fetchone()[0]    # Alert if lag > 100MB    if lag_bytes > 100 * 1024 * 1024:        alert(f"Replication lag: ${lag_bytes / 1024 / 1024:.2f} MB")    return lag_bytes# Monitor continuouslywhile True:    lag = check_replication_lag(primary_conn, replica_conn)    print(f"Replication lag: ${lag} bytes")    time.sleep(10)

# Automatic failover with retry logicclass ReplicationManager:    def __init__(self, primary, replicas):        self.primary = primary        self.replicas = replicas        self.max_lag_seconds = 10    def write_with_retry(self, query, params, max_retries=3):        for attempt in range(max_retries):            try:                # Attempt write to primary                self.primary.execute(query, params)                # Wait for replication (optional)                self.wait_for_replication()                return True            except PrimaryUnavailable:                # Promote replica to primary                new_primary = self.promote_replica()                self.primary = new_primary            except ReplicationTimeout:                if attempt == max_retries - 1:                    raise                time.sleep(2 ** attempt)  # Exponential backoff        return False    def promote_replica(self):        # Find replica with least lag        best_replica = min(            self.replicas,            key=lambda r: r.get_replication_lag()        )        # Promote to primary        best_replica.promote()        # Reconfigure other replicas        for replica in self.replicas:            if replica != best_replica:                replica.follow(best_replica)        return best_replica

# Periodic consistency checksimport hashlibdef verify_cross_region_consistency(regions):    ""    Check if all regions have same data    ""    checksums = {}    for region_name, conn in regions.items():        # Compute checksum of all data        cursor = conn.execute(            ""            SELECT MD5(STRING_AGG(CAST(id AS TEXT) || data, '' ORDER BY id))            FROM critical_table            ""        )        checksums[region_name] = cursor.fetchone()[0]    # Compare checksums    reference = list(checksums.values())[0]    inconsistent_regions = [        region for region, checksum in checksums.items()        if checksum != reference    ]    if inconsistent_regions:        alert(f"Data inconsistency detected in: {inconsistent_regions}")        # Trigger reconciliation        reconcile_regions(inconsistent_regions)    return len(inconsistent_regions) == 0

# Distributed consensus with etcdimport etcd3class PrimaryElection:    def __init__(self, etcd_client, node_id):        self.etcd = etcd_client        self.node_id = node_id        self.lease = None    def try_become_primary(self, ttl=10):        # Create lease        self.lease = self.etcd.lease(ttl)        # Try to acquire primary lock        success = self.etcd.put_if_not_exists(            '/cluster/primary',            self.node_id,            lease=self.lease        )        if success:            # Successfully became primary            self.heartbeat()  # Keep lease alive            return True        else:            # Another node is primary            return False    def heartbeat(self):        # Refresh lease to stay primary        while True:            self.lease.refresh()            time.sleep(5)  # Refresh every 5 seconds# Usageetcd = etcd3.client()election = PrimaryElection(etcd, node_id="us-east-1")if election.try_become_primary():    # This node is primary - accept writes    start_primary_mode()else:    # This node is replica - reject writes    start_replica_mode()

import randomimport threadingimport timeclass ChaosTest:    def __init__(self, cluster):        self.cluster = cluster        self.failures = []    def partition_network(self, duration=30):        ""Simulate network partition""        # Split cluster into two groups        group_a = self.cluster.nodes[:len(self.cluster.nodes)//2]        group_b = self.cluster.nodes[len(self.cluster.nodes)//2:]        print(f"CHAOS: Network partition for ${duration}s")        print(f"  Group A: ${[n.id for n in group_a]}")        print(f"  Group B: ${[n.id for n in group_b]}")        # Block traffic between groups        for node_a in group_a:            for node_b in group_b:                node_a.block_traffic_to(node_b)                node_b.block_traffic_to(node_a)        # Wait        time.sleep(duration)        # Heal partition        for node_a in group_a:            for node_b in group_b:                node_a.allow_traffic_to(node_b)                node_b.allow_traffic_to(node_a)        print("CHAOS: Network healed")    def kill_random_node(self):        ""Kill random node""        node = random.choice(self.cluster.nodes)        print(f"CHAOS: Killing node ${node.id}")        node.kill()        time.sleep(5)  # Wait for cluster to detect failure        return node    def induce_clock_skew(self, max_skew_ms=1000):        ""Introduce clock skew""        for node in self.cluster.nodes:            skew = random.randint(-max_skew_ms, max_skew_ms)            node.adjust_clock(skew)            print(f"CHAOS: Node ${node.id} clock skew: ${skew}ms")    def corrupt_wal_file(self, node):        ""Simulate disk corruption""        print(f"CHAOS: Corrupting WAL on node ${node.id}")        wal_file = node.get_latest_wal_file()        # Corrupt random bytes        with open(wal_file, 'r+b') as f:            f.seek(random.randint(0, os.path.getsize(wal_file)))            f.write(bytes([random.randint(0, 255)]))    def run_chaos_scenario(self):        ""Run comprehensive chaos test""        print("=== Starting Chaos Engineering Test ===")        # Baseline: verify system works        assert self.verify_consistency(), "System inconsistent before chaos!"        # Chaos 1: Network partition during writes        write_thread = threading.Thread(target=self.continuous_writes, args=(30,))        write_thread.start()        time.sleep(5)  # Let some writes succeed        self.partition_network(duration=15)        write_thread.join()        # Verify consistency after partition        assert self.verify_consistency(), "Consistency violated after partition!"        # Chaos 2: Kill node during heavy load        write_thread = threading.Thread(target=self.continuous_writes, args=(20,))        write_thread.start()        time.sleep(5)        killed_node = self.kill_random_node()        write_thread.join()        # Verify system still works        assert self.verify_consistency(), "Consistency violated after node kill!"        # Chaos 3: Restore killed node, verify catch-up        killed_node.start()        time.sleep(30)  # Wait for replication to catch up        assert self.verify_consistency(), "Node didn't catch up correctly!"        print("=== Chaos Test PASSED ===")    def continuous_writes(self, duration):        ""Write data continuously""        start = time.time()        count = 0        while time.time() - start < duration:            try:                self.cluster.write(f"key_${count}", f"value_${count}")                count += 1            except Exception as e:                print(f"Write failed: ${e}")            time.sleep(0.1)        print(f"Completed ${count} writes in ${duration}s")    def verify_consistency(self):        ""Check all replicas have same data""        print("Verifying consistency across replicas...")        # Get data from all nodes        node_data = {}        for node in self.cluster.nodes:            if node.is_running():                node_data[node.id] = node.get_all_data()        # Compare checksums        checksums = {            node_id: hash(frozenset(data.items()))            for node_id, data in node_data.items()        }        if len(set(checksums.values())) == 1:            print("✅ All replicas consistent")            return True        else:            print("❌ Replicas diverged!")            for node_id, checksum in checksums.items():                print(f"  Node ${node_id}: ${checksum}")            return False# Run chaos testcluster = create_test_cluster(nodes=5, replication_factor=3)chaos = ChaosTest(cluster)chaos.run_chaos_scenario()

# Chaos Monkey randomly kills instances in production# Ensures system resilient to failuresclass ChaosMonkey:    def __init__(self, cluster, probability=0.01):        self.cluster = cluster        self.kill_probability = probability    def run(self):        ""Periodically kill random instances""        while True:            time.sleep(3600)  # Every hour            if random.random() < self.kill_probability:                # Kill random instance                instance = random.choice(self.cluster.instances)                print(f"🐵 Chaos Monkey: Terminating ${instance.id}")                instance.terminate()                # Monitor recovery                self.monitor_recovery(instance)    def monitor_recovery(self, killed_instance):        ""Verify system recovered from failure""        # Check that:        # 1. Traffic rerouted to healthy instances        # 2. New instance launched automatically        # 3. No user-facing errors        pass# Run in production (yes, really!)monkey = ChaosMonkey(production_cluster, probability=0.01)monkey.run()  # 1% chance per hour of killing instance

from toxiproxy import Toxiproxy# Toxiproxy: proxy for simulating network problemstoxiproxy = Toxiproxy()# Create proxy for database connectiondb_proxy = toxiproxy.create(    name="db",    listen="127.0.0.1:5433",    upstream="postgres:5432")# Test 1: High latencydb_proxy.add_toxic(    name="latency",    type="latency",    attributes={"latency": 1000}  # Add 1s latency)# Application now experiences slow queries# Verify timeout handling works# Test 2: Packet lossdb_proxy.add_toxic(    name="packet_loss",    type="timeout",    attributes={"timeout": 0}  # 0 = drop packets)# Verify application retries correctly# Test 3: Connection limitsdb_proxy.add_toxic(    name="limit_data",    type="limit_data",    attributes={        "bytes": 1000  # Close after 1KB    })# Verify application handles partial responses# Remove toxics to restore normal operationdb_proxy.remove_toxic("latency")

# Calculate bandwidth needed for replication# Parameterswrite_rate = 10_000  # writes per secondavg_write_size = 1024  # bytes per writereplication_factor = 3  # 3 replicas# Bandwidth per secondbytes_per_second = write_rate * avg_write_sizeprint(f"Primary generates: ${bytes_per_second / 1024 / 1024:.2f} MB/s")# Replication traffic# Each write goes to (RF - 1) replicasreplication_bandwidth = bytes_per_second * (replication_factor - 1)print(f"Replication bandwidth: ${replication_bandwidth / 1024 / 1024:.2f} MB/s")# Cross-region bandwidth (more expensive)# Assume 2 regions, each with 3 replicascross_region_bandwidth = bytes_per_second * 3  # All writes to other regionprint(f"Cross-region bandwidth: ${cross_region_bandwidth / 1024 / 1024:.2f} MB/s")# Monthly cost estimate (AWS)# Inter-region: $0.02 per GBmonthly_seconds = 30 * 24 * 60 * 60monthly_gb = (cross_region_bandwidth * monthly_seconds) / 1024 / 1024 / 1024monthly_cost = monthly_gb * 0.02print(f"Monthly cross-region cost: $${monthly_cost:.2f}")

# Define SLOs for replication lagSLO_TARGETS = {    "p50": 10,    # ms - median lag    "p95": 100,   # ms - 95th percentile    "p99": 500,   # ms - 99th percentile    "p99.9": 2000 # ms - 99.9th percentile}# Monitor and alertdef check_replication_lag_slo(metrics):    for percentile, target_ms in SLO_TARGETS.items():        actual = metrics.get_percentile(percentile)        if actual > target_ms:            alert(                f"Replication lag SLO violated: "                f"{percentile} = {actual}ms (target: {target_ms}ms)"            )# Dashboard queries (Prometheus)# rate(replication_lag_seconds_sum[5m]) / rate(replication_lag_seconds_count[5m])# histogram_quantile(0.95, rate(replication_lag_seconds_bucket[5m]))