S076 · Multi-Region — Active-Passive, Active-Active, Failover

#!/usr/bin/env python3
"""multi_region_sim.py — a toy simulation of an active-passive setup
with async replication, health checks, and failover.
 
Run:  python multi_region_sim.py
No dependencies.
 
Simulates:
  - Region A (primary): accepts writes, replicates to B
  - Region B (secondary): reads only, tracks replication lag
  - A user client that fails over to B based on health checks
  - We kill A partway through and measure lost writes (RPO) and
    time-to-first-successful-write on B (RTO).
"""
from __future__ import annotations
 
import random
import threading
import time
from dataclasses import dataclass, field
 
 
@dataclass
class Region:
    name: str
    is_primary: bool = False
    healthy: bool = True
    accepts_writes: bool = False
    store: dict[int, str] = field(default_factory=dict)
    # Async replication queue: (msg_id, value, ts_written)
    replication_queue: list[tuple[int, str, float]] = field(default_factory=list)
    lock: threading.Lock = field(default_factory=threading.Lock)
 
    def write(self, key: int, value: str) -> bool:
        if not (self.accepts_writes and self.healthy):
            return False
        with self.lock:
            self.store[key] = value
            self.replication_queue.append((key, value, time.time()))
        return True
 
    def read(self, key: int) -> str | None:
        if not self.healthy:
            return None
        with self.lock:
            return self.store.get(key)
 
    def apply_replicated(self, key: int, value: str) -> None:
        with self.lock:
            self.store[key] = value
 
 
class Replicator:
    """Streams from a source region's queue to a target region's apply()."""
    def __init__(self, source: Region, target: Region, lag_ms: int = 100):
        self.source, self.target = source, target
        self.lag_ms = lag_ms
        self.stopped = False
        self.applied_count = 0
        threading.Thread(target=self._loop, daemon=True).start()
 
    def _loop(self) -> None:
        while not self.stopped:
            time.sleep(0.02)
            with self.source.lock:
                if not self.source.healthy:
                    # Source is down — replication halts. This is the whole point.
                    continue
                batch = list(self.source.replication_queue)
                self.source.replication_queue.clear()
            for key, value, ts in batch:
                # Simulate network latency between regions.
                time.sleep(self.lag_ms / 1000.0 / max(len(batch), 1))
                self.target.apply_replicated(key, value)
                self.applied_count += 1
 
    def stop(self) -> None:
        self.stopped = True
 
 
class HealthChecker:
    """Checks a region every check_interval; declares unhealthy after
    fail_threshold consecutive failures."""
    def __init__(self, region: Region, check_interval: float = 0.5,
                 fail_threshold: int = 3):
        self.region = region
        self.interval = check_interval
        self.threshold = fail_threshold
        self.consecutive_failures = 0
        self.declared_unhealthy = False
        self.stopped = False
        threading.Thread(target=self._loop, daemon=True).start()
 
    def _loop(self) -> None:
        while not self.stopped:
            time.sleep(self.interval)
            if self.region.healthy:
                self.consecutive_failures = 0
                self.declared_unhealthy = False
            else:
                self.consecutive_failures += 1
                if self.consecutive_failures >= self.threshold:
                    self.declared_unhealthy = True
 
    def stop(self) -> None:
        self.stopped = True
 
 
class Client:
    """Writes to whichever region the health checker declares healthy."""
    def __init__(self, primary: Region, secondary: Region,
                 hc_primary: HealthChecker, hc_secondary: HealthChecker):
        self.primary, self.secondary = primary, secondary
        self.hc_p, self.hc_s = hc_primary, hc_secondary
        self.writes_ok = 0
        self.writes_failed = 0
 
    def write(self, key: int, value: str) -> None:
        if not self.hc_p.declared_unhealthy and self.primary.write(key, value):
            self.writes_ok += 1
            return
        # Failover — try secondary
        if not self.hc_s.declared_unhealthy and self.secondary.write(key, value):
            self.writes_ok += 1
            return
        self.writes_failed += 1
 
 
def main() -> None:
    region_a = Region("region-a", is_primary=True, accepts_writes=True)
    region_b = Region("region-b", is_primary=False, accepts_writes=False)
 
    repl = Replicator(region_a, region_b, lag_ms=200)
    hc_a = HealthChecker(region_a, check_interval=0.3, fail_threshold=3)
    hc_b = HealthChecker(region_b, check_interval=0.3, fail_threshold=3)
    client = Client(region_a, region_b, hc_a, hc_b)
 
    print("Phase 1: normal operations for 2s")
    start = time.time()
    key = 0
    while time.time() - start < 2.0:
        client.write(key, f"v{key}")
        key += 1
        time.sleep(0.005)
    print(f"  writes_ok={client.writes_ok}, "
          f"region_a store size={len(region_a.store)}, "
          f"replicated to B={repl.applied_count}")
 
    print("\nPhase 2: KILL region A")
    region_a.healthy = False
    kill_time = time.time()
    last_replicated_at_kill = repl.applied_count
    a_size_at_kill = len(region_a.store)
 
    # Wait for health check to declare A unhealthy, then failover B
    while not hc_a.declared_unhealthy:
        time.sleep(0.05)
    detect_time = time.time()
    print(f"  A declared unhealthy after {(detect_time - kill_time)*1000:.0f}ms")
 
    # Enable writes on B (manual failover step)
    region_b.accepts_writes = True
    failover_time = time.time()
    print(f"  B accepting writes at t={(failover_time - kill_time)*1000:.0f}ms after A died")
 
    # RPO measurement: how many writes were on A but not yet replicated to B?
    rpo_lost_writes = a_size_at_kill - repl.applied_count
    print(f"  RPO: {rpo_lost_writes} writes were on A but never made it to B")
 
    print("\nPhase 3: continue writing (now to B)")
    first_success_after_failover = None
    for _ in range(200):
        before = client.writes_ok
        client.write(key, f"v{key}")
        key += 1
        if first_success_after_failover is None and client.writes_ok > before:
            first_success_after_failover = time.time()
        time.sleep(0.005)
 
    if first_success_after_failover:
        rto = first_success_after_failover - kill_time
        print(f"  RTO: {rto*1000:.0f}ms from A-dies to first successful write on B")
 
    print(f"\nFinal: writes_ok={client.writes_ok}, writes_failed={client.writes_failed}")
 
    repl.stop(); hc_a.stop(); hc_b.stop()
 
 
if __name__ == "__main__":
    main()