Consistency Models in Distributed Systems

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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

# Sticky Session Implementation
import hashlib
from dataclasses import dataclass
from typing import Dict, List, Optional
import random
 
@dataclass
class Backend:
    """A backend server in the cluster."""
    id: str
    host: str
    port: int
    healthy: bool = True
    current_load: int = 0
 
class StickySessionLoadBalancer:
    """
    Load balancer implementing sticky sessions for RYW consistency.
    """
    
    def __init__(self, backends: List[Backend]):
        self.backends = backends
        self.session_map: Dict[str, str] = {}  # session_id -> backend_id
        
        # For consistent hashing (optional)
        self.ring: Dict[int, Backend] = {}
        self._build_hash_ring()
    
    def _build_hash_ring(self, replicas: int = 100):
        """Build consistent hash ring for IP-hash routing."""
        for backend in self.backends:
            for i in range(replicas):
                key = f"{backend.id}:{i}"
                hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
                self.ring[hash_val] = backend
    
    # ════════════════════════════════════════════════════════════════
    # Strategy 1: Cookie-Based Affinity
    # ════════════════════════════════════════════════════════════════
    
    def route_by_cookie(self, request) -> Backend:
        """
        Route based on session cookie.
        
        Pros:
        - Explicit control over session identity
        - Works across IP changes (mobile networks)
        
        Cons:
        - Requires cookie support in client
        - Cookie can be cleared/lost
        """
        session_id = request.cookies.get('SESSION_ID')
        
        if session_id and session_id in self.session_map:
            backend_id = self.session_map[session_id]
            backend = self._get_backend(backend_id)
            
            if backend and backend.healthy:
                return backend
            else:
                # Backend unhealthy - reassign (breaks RYW!)
                return self._assign_new_session(session_id)
        else:
            # New session
            session_id = self._generate_session_id()
            backend = self._assign_new_session(session_id)
            # Response will include Set-Cookie header
            return backend
    
    def _assign_new_session(self, session_id: str) -> Backend:
        """Assign a new session to a healthy backend."""
        healthy = [b for b in self.backends if b.healthy]
        if not healthy:
            raise Exception("No healthy backends available")
        
        # Choose least-loaded backend
        backend = min(healthy, key=lambda b: b.current_load)
        self.session_map[session_id] = backend.id
        return backend
    
    # ════════════════════════════════════════════════════════════════
    # Strategy 2: IP-Hash Routing
    # ════════════════════════════════════════════════════════════════
    
    def route_by_ip_hash(self, client_ip: str) -> Backend:
        """
        Route based on hash of client IP.
        
        Pros:
        - No session state needed at load balancer
        - Automatic affinity without cookies
        
        Cons:
        - IP can change (mobile networks, VPNs)
        - Uneven distribution if IPs are clustered
        - Backend failure requires rehashing many clients
        """
        hash_val = int(hashlib.md5(client_ip.encode()).hexdigest(), 16)
        
        # Find nearest backend in hash ring
        ring_keys = sorted(self.ring.keys())
        for key in ring_keys:
            if hash_val <= key:
                backend = self.ring[key]
                if backend.healthy:
                    return backend
        
        # Wrap around
        return self.ring[ring_keys[0]]
    
    # ════════════════════════════════════════════════════════════════
    # Strategy 3: Consistent Hashing with Session ID
    # ════════════════════════════════════════════════════════════════
    
    def route_by_consistent_hash(self, session_id: str) -> Backend:
        """
        Route based on consistent hash of session ID.
        
        Pros:
        - Minimal disruption when backends added/removed
        - Session ID can be client-controlled
        
        Cons:
        - Still breaks when mapped backend fails
        - Requires session ID in every request
        """
        hash_val = int(hashlib.md5(session_id.encode()).hexdigest(), 16)
        
        # Find clockwise nearest healthy backend
        ring_keys = sorted(self.ring.keys())
        for key in ring_keys:
            if hash_val <= key:
                backend = self.ring[key]
                if backend.healthy:
                    return backend
                # Skip unhealthy, try next
        
        # Wrap around and find first healthy
        for key in ring_keys:
            if self.ring[key].healthy:
                return self.ring[key]
        
        raise Exception("No healthy backends")
    
    def _get_backend(self, backend_id: str) -> Optional[Backend]:
        for b in self.backends:
            if b.id == backend_id:
                return b
        return None
    
    def _generate_session_id(self) -> str:
        return hashlib.sha256(str(random.random()).encode()).hexdigest()[:32]

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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

# Version Tracking for Read-Your-Writes
from dataclasses import dataclass, field
from typing import Dict, Optional, Tuple, List
import time
 
@dataclass
class VersionedValue:
    """Value with version metadata."""
    value: any
    version: int
    timestamp: float
    writer_id: str
 
@dataclass
class Replica:
    """Database replica with versioned data."""
    id: str
    data: Dict[str, VersionedValue] = field(default_factory=dict)
    current_version: int = 0  # Global version counter for this replica
    
    def write(self, key: str, value: any, writer_id: str) -> int:
        """Write and return the new version."""
        self.current_version += 1
        self.data[key] = VersionedValue(
            value=value,
            version=self.current_version,
            timestamp=time.time(),
            writer_id=writer_id
        )
        return self.current_version
    
    def read(self, key: str, min_version: int = 0) -> Optional[VersionedValue]:
        """
        Read value if it meets minimum version requirement.
        Returns None if key doesn't exist or version too old.
        """
        if key not in self.data:
            return None
        
        entry = self.data[key]
        if entry.version >= min_version:
            return entry
        return None  # Version requirement not met
    
    def read_blocking(self, key: str, min_version: int, 
                      timeout: float = 5.0) -> Optional[VersionedValue]:
        """
        Read value, blocking until minimum version is available.
        Used when replica is catching up via replication.
        """
        deadline = time.time() + timeout
        
        while time.time() < deadline:
            entry = self.read(key, min_version)
            if entry:
                return entry
            time.sleep(0.01)  # Poll interval
        
        return None  # Timeout
 
 
class VersionTrackingClient:
    """
    Client that tracks write versions for RYW consistency.
    """
    
    def __init__(self, client_id: str, replicas: List[Replica]):
        self.client_id = client_id
        self.replicas = replicas
        
        # Track minimum acceptable version for each key
        self.write_versions: Dict[str, int] = {}
        
        # Track global minimum version (for cross-key consistency)
        self.global_min_version: int = 0
    
    def write(self, key: str, value: any) -> int:
        """
        Write to primary replica and record version.
        """
        primary = self._get_primary()
        version = primary.write(key, value, self.client_id)
        
        # Record that we need at least this version for future reads
        self.write_versions[key] = version
        self.global_min_version = max(self.global_min_version, version)
        
        return version
    
    def read(self, key: str) -> Optional[any]:
        """
        Read ensuring we see at least our last write.
        """
        min_version = self.write_versions.get(key, 0)
        
        # Try replicas in order of likely freshness
        for replica in self._order_by_freshness():
            result = replica.read(key, min_version)
            if result:
                # Update our tracking (we've now seen this version)
                self.write_versions[key] = max(
                    self.write_versions.get(key, 0),
                    result.version
                )
                return result.value
        
        # No replica met version requirement - fall back to primary
        primary = self._get_primary()
        result = primary.read_blocking(key, min_version)
        if result:
            return result.value
        
        return None
    
    def read_consistent(self, keys: List[str]) -> Dict[str, any]:
        """
        Read multiple keys with consistent view.
        All reads will reflect at least our last writes.
        """
        results = {}
        min_version = self.global_min_version
        
        # Read all keys from same replica for snapshot consistency
        for replica in self._order_by_freshness():
            if replica.current_version >= min_version:
                for key in keys:
                    result = replica.read(key, self.write_versions.get(key, 0))
                    if result:
                        results[key] = result.value
                return results
        
        # Fall back to primary
        primary = self._get_primary()
        for key in keys:
            result = primary.read(key)
            if result:
                results[key] = result.value
        return results
    
    def _get_primary(self) -> Replica:
        """Get primary replica (first in list by convention)."""
        return self.replicas[0]
    
    def _order_by_freshness(self) -> List[Replica]:
        """Order replicas by current version (freshest first)."""
        return sorted(self.replicas, key=lambda r: -r.current_version)
 
 
# Demonstration
def version_tracking_demo():
    """
    Show how version tracking provides RYW even with
    multiple replicas at different replication points.
    """
    # Create replicas with different lag
    primary = Replica("primary")
    secondary1 = Replica("secondary1")
    secondary2 = Replica("secondary2")
    replicas = [primary, secondary1, secondary2]
    
    client = VersionTrackingClient("user123", replicas)
    
    # Write to primary
    version = client.write("profile", {"name": "Jonathan"})
    print(f"Wrote profile with version {version}")
    
    # Simulate replication lag
    # secondary1 has replicated, secondary2 has not
    secondary1.data["profile"] = primary.data["profile"]
    secondary1.current_version = primary.current_version
    # secondary2 is still behind
    
    # Client read will check version requirement
    # - secondary2 will be skipped (version too old)
    # - secondary1 will succeed
    result = client.read("profile")
    print(f"Read result: {result}")  # {"name": "Jonathan"}

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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193

# Session Token Implementation (Cosmos DB-style)
from dataclasses import dataclass, field
from typing import Dict, Optional, List
import json
import base64
import time
 
@dataclass
class PartitionState:
    """State for a single partition within the session."""
    partition_id: str
    lsn: int  # Logical Sequence Number
    timestamp: float
 
@dataclass  
class SessionToken:
    """
    Session token encoding observed state across partitions.
    Inspired by Azure Cosmos DB's session token design.
    """
    partitions: Dict[str, PartitionState] = field(default_factory=dict)
    global_lsn: int = 0
    
    def update_partition(self, partition_id: str, lsn: int, timestamp: float):
        """Update the observed state for a partition."""
        if partition_id not in self.partitions:
            self.partitions[partition_id] = PartitionState(partition_id, 0, 0)
        
        current = self.partitions[partition_id]
        if lsn > current.lsn:
            current.lsn = lsn
            current.timestamp = timestamp
            self.global_lsn = max(self.global_lsn, lsn)
    
    def get_partition_lsn(self, partition_id: str) -> int:
        """Get the minimum LSN required for a partition."""
        if partition_id in self.partitions:
            return self.partitions[partition_id].lsn
        return 0
    
    def merge(self, other: 'SessionToken') -> 'SessionToken':
        """Merge two session tokens (point-wise maximum)."""
        merged = SessionToken()
        
        all_partitions = set(self.partitions.keys()) | set(other.partitions.keys())
        for pid in all_partitions:
            self_state = self.partitions.get(pid)
            other_state = other.partitions.get(pid)
            
            if self_state and other_state:
                if self_state.lsn >= other_state.lsn:
                    merged.partitions[pid] = self_state
                else:
                    merged.partitions[pid] = other_state
            elif self_state:
                merged.partitions[pid] = self_state
            else:
                merged.partitions[pid] = other_state
        
        merged.global_lsn = max(self.global_lsn, other.global_lsn)
        return merged
    
    def serialize(self) -> str:
        """Serialize to string for transmission in HTTP header."""
        data = {
            "partitions": {
                pid: {"lsn": ps.lsn, "ts": ps.timestamp}
                for pid, ps in self.partitions.items()
            },
            "global": self.global_lsn
        }
        return base64.b64encode(json.dumps(data).encode()).decode()
    
    @classmethod
    def deserialize(cls, token_str: str) -> 'SessionToken':
        """Deserialize from string."""
        data = json.loads(base64.b64decode(token_str))
        token = cls()
        token.global_lsn = data.get("global", 0)
        
        for pid, state in data.get("partitions", {}).items():
            token.partitions[pid] = PartitionState(
                partition_id=pid,
                lsn=state["lsn"],
                timestamp=state.get("ts", 0)
            )
        return token
 
 
class SessionConsistentServer:
    """
    Server implementing session token-based consistency.
    """
    
    def __init__(self, partition_id: str):
        self.partition_id = partition_id
        self.data: Dict[str, any] = {}
        self.lsn: int = 0
    
    def write(self, key: str, value: any, 
              session_token: Optional[SessionToken] = None) -> SessionToken:
        """
        Write operation returning updated session token.
        """
        self.lsn += 1
        self.data[key] = {"value": value, "lsn": self.lsn}
        
        # Create/update session token
        token = session_token or SessionToken()
        token.update_partition(self.partition_id, self.lsn, time.time())
        
        return token
    
    def read(self, key: str, 
             session_token: Optional[SessionToken] = None) -> Tuple[any, SessionToken]:
        """
        Read operation respecting session token requirements.
        """
        required_lsn = 0
        if session_token:
            required_lsn = session_token.get_partition_lsn(self.partition_id)
        
        # Check if we can satisfy the read
        if self.lsn < required_lsn:
            # We're behind - either wait or redirect
            self._wait_for_lsn(required_lsn)
        
        entry = self.data.get(key)
        
        # Create response token
        token = session_token or SessionToken()
        if entry:
            token.update_partition(self.partition_id, entry["lsn"], time.time())
            return entry["value"], token
        
        token.update_partition(self.partition_id, self.lsn, time.time())
        return None, token
    
    def _wait_for_lsn(self, target_lsn: int, timeout: float = 5.0):
        """Wait for replication to catch up to target LSN."""
        deadline = time.time() + timeout
        while self.lsn < target_lsn and time.time() < deadline:
            time.sleep(0.01)
        
        if self.lsn < target_lsn:
            raise Exception(f"Timeout waiting for LSN {target_lsn}")
 
 
class SessionConsistentClient:
    """
    Client maintaining session token for RYW consistency.
    Modeled after Azure Cosmos DB SDK behavior.
    """
    
    def __init__(self, servers: Dict[str, SessionConsistentServer]):
        self.servers = servers  # partition_id -> server
        self.session_token = SessionToken()
    
    def write(self, key: str, value: any):
        """Write with automatic session token management."""
        partition = self._get_partition(key)
        server = self.servers[partition]
        
        new_token = server.write(key, value, self.session_token)
        self.session_token = self.session_token.merge(new_token)
        
        return new_token.serialize()
    
    def read(self, key: str) -> any:
        """Read with session token for RYW guarantee."""
        partition = self._get_partition(key)
        server = self.servers[partition]
        
        value, new_token = server.read(key, self.session_token)
        self.session_token = self.session_token.merge(new_token)
        
        return value
    
    def get_session_token(self) -> str:
        """Get current session token for persistence/transfer."""
        return self.session_token.serialize()
    
    def set_session_token(self, token_str: str):
        """Restore session token (e.g., from cookie)."""
        restored = SessionToken.deserialize(token_str)
        self.session_token = self.session_token.merge(restored)
    
    def _get_partition(self, key: str) -> str:
        """Determine which partition owns this key."""
        # Simple hash-based partitioning
        partition_list = list(self.servers.keys())
        idx = hash(key) % len(partition_list)
        return partition_list[idx]

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

# Multi-Device Read-Your-Writes Strategies
from dataclasses import dataclass, field
from typing import Dict, Optional
import time
 
# ════════════════════════════════════════════════════════════════
# Strategy 1: User-Based Session Storage
# ════════════════════════════════════════════════════════════════
 
class UserSessionStore:
    """
    Store session tokens server-side, keyed by user ID.
    All devices of a user share the same session.
    """
    
    def __init__(self):
        self.user_sessions: Dict[str, SessionToken] = {}
    
    def get_session(self, user_id: str) -> SessionToken:
        """Get or create session for user."""
        if user_id not in self.user_sessions:
            self.user_sessions[user_id] = SessionToken()
        return self.user_sessions[user_id]
    
    def update_session(self, user_id: str, new_token: SessionToken):
        """Update user's session with new observations."""
        current = self.get_session(user_id)
        self.user_sessions[user_id] = current.merge(new_token)
 
 
class MultiDeviceClient:
    """
    Client that fetches session token from server on each request.
    Ensures cross-device RYW at cost of extra round-trip.
    """
    
    def __init__(self, user_id: str, session_store: UserSessionStore,
                 db_client):
        self.user_id = user_id
        self.session_store = session_store
        self.db = db_client
    
    def write(self, key: str, value: any):
        """Write with cross-device session consistency."""
        # Get current user session
        session = self.session_store.get_session(self.user_id)
        
        # Perform write
        new_token = self.db.write(key, value, session)
        
        # Update server-side session
        self.session_store.update_session(self.user_id, new_token)
    
    def read(self, key: str) -> any:
        """Read with cross-device RYW guarantee."""
        # Fetch latest session from server
        session = self.session_store.get_session(self.user_id)
        
        # Read with session token
        value, new_token = self.db.read(key, session)
        
        # Update server-side session (we've now seen more)
        self.session_store.update_session(self.user_id, new_token)
        
        return value
 
 
# ════════════════════════════════════════════════════════════════
# Strategy 2: Hybrid Approach
# ════════════════════════════════════════════════════════════════
 
class HybridConsistencyClient:
    """
    Use different consistency levels for different operations.
    - Strong consistency for critical paths
    - Session consistency for normal operations
    - Eventual consistency for bulk reads
    """
    
    def __init__(self, db_client, user_id: str):
        self.db = db_client
        self.user_id = user_id
        self.local_session = SessionToken()
    
    def write_critical(self, key: str, value: any):
        """
        Critical write with strong consistency.
        Wait for quorum acknowledgment.
        """
        return self.db.write_strong(key, value)
    
    def read_critical(self, key: str) -> any:
        """
        Critical read with strong consistency.
        Read from quorum, guaranteed fresh.
        """
        return self.db.read_strong(key)
    
    def write_normal(self, key: str, value: any):
        """
        Normal write with session consistency.
        Local session guarantees RYW for this device.
        """
        new_token = self.db.write(key, value, self.local_session)
        self.local_session = self.local_session.merge(new_token)
    
    def read_normal(self, key: str) -> any:
        """
        Normal read with session consistency.
        Guaranteed to see own writes.
        """
        value, new_token = self.db.read(key, self.local_session)
        self.local_session = self.local_session.merge(new_token)
        return value
    
    def read_bulk(self, keys: List[str]) -> Dict[str, any]:
        """
        Bulk read with eventual consistency.
        May see stale data, but high throughput.
        """
        return self.db.read_eventual(keys)