CAP Theorem: Fundamental Trade-offs in Distributed Systems

NON-PARTITION-TOLERANT SYSTEM (Invalid for distributed systems):
═══════════════════════════════════════════════════════════════════
 
Design: "All nodes must agree before proceeding"
 
Normal Operation:
  Client → Node A: Write x=1
  Node A → Node B: Propose x=1
  Node B → Node A: Accept
  Node A → Node C: Propose x=1
  Node C → Node A: Accept
  Node A → Client: Write successful ✓
 
During Partition (Node C unreachable):
  Client → Node A: Write x=2
  Node A → Node B: Propose x=2
  Node B → Node A: Accept
  Node A → Node C: Propose x=2
  [TIMEOUT - C unreachable]
  Node A: Cannot proceed without unanimous agreement
  Node A → Client: ??? 
  
  Options for non-P system:
    a) Wait forever for C → System hangs (not available)
    b) Give up entirely → System non-operational
    c) Proceed without C → But then we're partition tolerant!
    
  None of these are acceptable for a production system.
 
 
PARTITION-TOLERANT SYSTEM (CP design):
═══════════════════════════════════════════════════════════════════
 
Design: "Majority quorum (2 of 3) suffices for operations"
 
During Same Partition (Node C unreachable):
  Client → Node A: Write x=2
  Node A → Node B: Propose x=2
  Node B → Node A: Accept
  Node A → Node C: Propose x=2
  [TIMEOUT - C unreachable]
  Node A: I have 2 of 3 acks, quorum achieved!
  Node A → Client: Write successful ✓
  
  Later, when partition heals:
  Node A → Node C: Here's what you missed
  Node C: Updating my state...
  
  System remained operational throughout partition  ✓
  Consistency maintained via quorum             ✓
 
 
PARTITION-TOLERANT SYSTEM (AP design):
═══════════════════════════════════════════════════════════════════
 
Design: "Each node serves local reads/writes independently"
 
During Same Partition:
  Client 1 → Node A: Write x=2
  Node A: Written locally!
  Node A → Client 1: Write successful ✓
  
  Meanwhile in the other partition...
  
  Client 2 → Node C: Write x=5
  Node C: Written locally!
  Node C → Client 2: Write successful ✓
  
  Now we have x=2 on A,B and x=5 on C
  
  When partition heals:
  Conflict detected: x=2 vs x=5
  Resolution needed (LWW, vector clocks, merge, etc.)
  
  System remained operational throughout partition ✓
  Consistency was sacrificed                       ✗ (temporarily)
  Eventually consistent after resolution           ✓

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

import time
from enum import Enum
from dataclasses import dataclass
 
class NodeStatus(Enum):
    HEALTHY = "healthy"
    SUSPECT = "suspect"  
    UNREACHABLE = "unreachable"
 
@dataclass
class FailureDetector:
    """
    Phi Accrual Failure Detector - adaptive partition/failure detection.
    
    Instead of a binary alive/dead decision, maintains a 'suspicion level'
    that increases as heartbeats are missed.
    
    Used by: Cassandra, Akka, and other distributed systems.
    """
    
    # History of heartbeat arrival times
    heartbeat_history: list[float]
    
    # Configuration
    window_size: int = 100
    phi_threshold: float = 8.0  # Higher = more tolerant of slow responses
    
    def record_heartbeat(self, arrival_time: float):
        """Record a received heartbeat."""
        self.heartbeat_history.append(arrival_time)
        if len(self.heartbeat_history) > self.window_size:
            self.heartbeat_history.pop(0)
    
    def calculate_phi(self, current_time: float) -> float:
        """
        Calculate suspicion level (phi) based on time since last heartbeat.
        
        Phi represents the likelihood that the node has failed, expressed
        as a log probability. Higher phi = more likely to have failed.
        
        phi = -log10(probability that node is still alive)
        
        phi = 1  → 90% chance node is alive
        phi = 2  → 99% chance it's NOT alive
        phi = 8  → extremely likely node is unreachable
        """
        if len(self.heartbeat_history) < 2:
            return 0.0  # Not enough data
        
        # Calculate mean and std of inter-arrival times
        intervals = [
            self.heartbeat_history[i] - self.heartbeat_history[i-1]
            for i in range(1, len(self.heartbeat_history))
        ]
        mean_interval = sum(intervals) / len(intervals)
        variance = sum((x - mean_interval) ** 2 for x in intervals) / len(intervals)
        std = variance ** 0.5
        
        # Time since last heartbeat
        last_heartbeat = self.heartbeat_history[-1]
        elapsed = current_time - last_heartbeat
        
        # Calculate phi using exponential distribution
        # This accounts for normal variance in heartbeat timing
        if std == 0:
            std = 0.1  # Avoid division by zero
        
        phi = elapsed / (mean_interval + 3 * std)
        return max(0, phi)
    
    def get_status(self, current_time: float) -> NodeStatus:
        """Determine node status based on phi value."""
        phi = self.calculate_phi(current_time)
        
        if phi < self.phi_threshold * 0.3:
            return NodeStatus.HEALTHY
        elif phi < self.phi_threshold:
            return NodeStatus.SUSPECT
        else:
            return NodeStatus.UNREACHABLE
 
 
class QuorumPartitionDetector:
    """
    Partition detection using quorum-based gossip.
    
    If a node can reach a majority of peers, it's probably not partitioned.
    If it can only reach a minority, it's likely in a minority partition.
    """
    
    def __init__(self, all_nodes: list[str], self_node: str):
        self.all_nodes = all_nodes
        self.self_node = self_node
        self.total_nodes = len(all_nodes)
        self.quorum_size = (self.total_nodes // 2) + 1
    
    def detect_partition_status(self, reachable_nodes: set[str]) -> dict:
        """
        Determine if we're in the majority or minority partition.
        
        Returns diagnosis with recommendations.
        """
        reachable_count = len(reachable_nodes) + 1  # Include self
        
        if reachable_count >= self.quorum_size:
            return {
                "status": "majority_partition",
                "reachable": reachable_count,
                "quorum": self.quorum_size,
                "action": "Safe to accept writes (CP behavior)",
                "confidence": "high"
            }
        else:
            return {
                "status": "minority_partition",
                "reachable": reachable_count,
                "quorum": self.quorum_size,
                "action": "Reject writes to maintain consistency",
                "confidence": "high",
                "recommendation": "Wait for partition to heal or failover"
            }
 
 
# Example usage:
#
# Scenario: 5-node cluster, nodes see different network views
#
# Node A sees: [A, B, C]      → 3/5, has quorum, can proceed
# Node D sees: [D, E]         → 2/5, no quorum, should not proceed
# Node B sees: [A, B, C, D]   → Asymmetric! B thinks D is reachable
#
# The asymmetric case is why partition detection is hard:
# - B thinks 4 nodes are healthy
# - D thinks only 2 are reachable
# - Who is correct? Both, from their perspective!

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

class CPPartitionHandler:
    """
    Consistency-Preferring partition handling.
    Sacrifices availability to maintain data consistency.
    """
    
    def __init__(self, cluster_size: int, my_node_id: str):
        self.cluster_size = cluster_size
        self.my_node_id = my_node_id
        self.quorum_size = (cluster_size // 2) + 1
    
    def can_accept_write(self, reachable_nodes: set[str]) -> bool:
        """
        Determine if we can safely accept a write.
        
        Only accept if we can reach a quorum - this ensures
        any two operations (one here, one elsewhere) will have
        overlapping participants, maintaining consistency.
        """
        reachable_count = len(reachable_nodes) + 1  # Include self
        
        if reachable_count >= self.quorum_size:
            return True
        else:
            # Cannot guarantee consistency - refuse operation
            return False
    
    def handle_write_request(self, data, reachable_nodes: set[str]):
        """
        Handle an incoming write request during potential partition.
        """
        if not self.can_accept_write(reachable_nodes):
            # Return error - we cannot serve this request consistently
            raise PartitionError(
                f"Cannot accept write: only {len(reachable_nodes) + 1} "
                f"of {self.cluster_size} nodes reachable, need {self.quorum_size}. "
                "Retry when partition heals."
            )
        
        # We have quorum, proceed with distributed write
        return self.replicate_to_quorum(data, reachable_nodes)
    
    def replicate_to_quorum(self, data, reachable_nodes: set[str]):
        """
        Replicate write to quorum of nodes before acknowledging.
        """
        acks = 1  # Self
        for node_id in reachable_nodes:
            try:
                self.send_write(node_id, data)
                acks += 1
                if acks >= self.quorum_size:
                    return WriteResult(success=True, acks=acks)
            except NodeUnreachableError:
                continue
        
        # Couldn't get quorum - should not happen if can_accept_write passed
        # Roll back local write
        self.rollback(data)
        raise PartitionError("Lost quorum during write")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

class APPartitionHandler:
    """
    Availability-Preferring partition handling.
    Accepts operations during partitions, resolves conflicts on merge.
    """
    
    def __init__(self, my_node_id: str):
        self.my_node_id = my_node_id
        self.local_data = {}  # key → (value, vector_clock)
    
    def handle_write_request(self, key: str, value: Any, client_clock: VectorClock):
        """
        Always accept writes - we prioritize availability.
        Track causality using vector clocks for later resolution.
        """
        # Increment our position in the vector clock
        new_clock = client_clock.increment(self.my_node_id)
        
        # Store locally with vector clock
        self.local_data[key] = (value, new_clock)
        
        # Try to replicate asynchronously (best effort)
        asyncio.create_task(self.async_replicate(key, value, new_clock))
        
        # Immediately acknowledge to client - we don't wait for replication
        return WriteResult(
            success=True,
            consistency="eventual",
            vector_clock=new_clock
        )
    
    async def async_replicate(self, key: str, value: Any, clock: VectorClock):
        """
        Attempt to replicate to other nodes asynchronously.
        Failures are logged but don't affect the write result.
        """
        for peer in self.known_peers:
            try:
                await peer.async_write(key, value, clock)
            except NodeUnreachableError:
                # Expected during partition - anti-entropy will handle later
                pass
    
    def on_partition_heal(self, other_partition_data: dict):
        """
        Called when partition heals and we receive data from other side.
        Must resolve any conflicts.
        """
        conflicts = []
        
        for key, (other_value, other_clock) in other_partition_data.items():
            if key in self.local_data:
                local_value, local_clock = self.local_data[key]
                
                # Compare vector clocks
                comparison = local_clock.compare(other_clock)
                
                if comparison == Ordering.BEFORE:
                    # Other value is causally after ours - accept it
                    self.local_data[key] = (other_value, other_clock)
                    
                elif comparison == Ordering.AFTER:
                    # Our value is causally after - keep ours
                    pass
                    
                elif comparison == Ordering.CONCURRENT:
                    # TRUE CONFLICT - values diverged during partition
                    # Requires resolution strategy
                    conflicts.append({
                        "key": key,
                        "local": (local_value, local_clock),
                        "remote": (other_value, other_clock)
                    })
            else:
                # We don't have this key - accept from other partition
                self.local_data[key] = (other_value, other_clock)
        
        # Handle true conflicts
        for conflict in conflicts:
            self.resolve_conflict(conflict)
    
    def resolve_conflict(self, conflict: dict):
        """
        Resolve concurrent writes that happened during partition.
        Multiple strategies possible.
        """
        key = conflict["key"]
        local_value, local_clock = conflict["local"]
        remote_value, remote_clock = conflict["remote"]
        
        # Strategy 1: Last-Write-Wins (using timestamps if available)
        if hasattr(local_clock, 'timestamp') and hasattr(remote_clock, 'timestamp'):
            if remote_clock.timestamp > local_clock.timestamp:
                self.local_data[key] = conflict["remote"]
            # else keep local
        
        # Strategy 2: Merge (for mergeable data types like counters)
        elif isinstance(local_value, Counter) and isinstance(remote_value, Counter):
            merged_value = local_value.merge(remote_value)
            merged_clock = local_clock.merge(remote_clock)
            self.local_data[key] = (merged_value, merged_clock)
        
        # Strategy 3: Store both, let application resolve
        else:
            self.local_data[key] = Conflict(local_value, remote_value)
            # Application must handle Conflict type on read