Data Replication & Consistency

# Simulating quorum-based replication
import time
import random
from dataclasses import dataclass, field
from typing import Optional, Dict, List

@dataclass
class VersionedValue:
    value: str
    version: int
    timestamp: float = field(default_factory=time.time)

class Replica:
    def __init__(self, replica_id: str, is_healthy=True):
        self.replica_id = replica_id
        self.is_healthy = is_healthy
        self.store: Dict[str, VersionedValue] = {}

    def write(self, key: str, value: str,
              version: int) -> bool:
        if not self.is_healthy:
            raise ConnectionError(f"{self.replica_id} is down")
        # Only accept newer versions
        existing = self.store.get(key)
        if existing and existing.version >= version:
            return False
        self.store[key] = VersionedValue(value, version)
        return True

    def read(self, key: str) -> Optional[VersionedValue]:
        if not self.is_healthy:
            raise ConnectionError(f"{self.replica_id} is down")
        return self.store.get(key)

class QuorumStore:
    def __init__(self, n=3, w=2, r=2):
        self.n = n
        self.w = w
        self.r = r
        self.replicas = [Replica(f"R{i}") for i in range(n)]
        self.version_counter = 0

    def write(self, key: str, value: str) -> bool:
        """Write to W replicas for quorum."""
        self.version_counter += 1
        acks = 0
        for replica in self.replicas:
            try:
                if replica.write(key, value, self.version_counter):
                    acks += 1
            except ConnectionError:
                continue
        success = acks >= self.w
        print(f"Write '{key}'='{value}' v{self.version_counter}: "
              f"{acks}/{self.n} acks ({'OK' if success else 'FAIL'})")
        return success

    def read(self, key: str) -> Optional[str]:
        """Read from R replicas, return newest."""
        responses: List[VersionedValue] = []
        for replica in self.replicas:
            try:
                val = replica.read(key)
                if val:
                    responses.append(val)
            except ConnectionError:
                continue

        if len(responses) < self.r:
            print(f"Read '{key}': quorum not met")
            return None

        # Return the value with highest version
        newest = max(responses, key=lambda v: v.version)
        print(f"Read '{key}': '{newest.value}' (v{newest.version})")
        return newest.value

# Demo
store = QuorumStore(n=3, w=2, r=2)
store.write("user:1", "Alice")
store.read("user:1")

# Simulate a node failure
store.replicas[2].is_healthy = False
store.write("user:1", "Alice Updated")  # Still succeeds (2 acks)
store.read("user:1")   # Still succeeds (2 responses)