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

# Session Guarantees: Formal Model and Violation Detection
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set, Tuple
from enum import Enum
 
class OperationType(Enum):
    READ = "read"
    WRITE = "write"
 
@dataclass
class Operation:
    """A single operation in a session."""
    op_type: OperationType
    key: str
    value: Optional[any]  # For writes
    observed_version: Optional[int]  # For reads: version that was returned
    written_version: Optional[int]  # For writes: version that was created
    timestamp: float
 
@dataclass
class SessionHistory:
    """Record of operations in a session for guarantee checking."""
    operations: List[Operation] = field(default_factory=list)
    
    # Tracking state
    last_read_versions: Dict[str, int] = field(default_factory=dict)  # key -> version
    last_write_versions: Dict[str, int] = field(default_factory=dict)  # key -> version
    
    def add_read(self, key: str, observed_version: int, timestamp: float):
        """Record a read operation."""
        op = Operation(
            op_type=OperationType.READ,
            key=key,
            value=None,
            observed_version=observed_version,
            written_version=None,
            timestamp=timestamp
        )
        self.operations.append(op)
        
        # Check monotonic reads
        if key in self.last_read_versions:
            if observed_version < self.last_read_versions[key]:
                raise MonotonicReadsViolation(
                    f"Read version {observed_version} < prior {self.last_read_versions[key]}"
                )
        
        # Check read-your-writes
        if key in self.last_write_versions:
            if observed_version < self.last_write_versions[key]:
                raise ReadYourWritesViolation(
                    f"Read version {observed_version} < written {self.last_write_versions[key]}"
                )
        
        self.last_read_versions[key] = max(
            self.last_read_versions.get(key, 0),
            observed_version
        )
    
    def add_write(self, key: str, value: any, written_version: int, 
                  timestamp: float, observed_context: Dict[str, int]):
        """
        Record a write operation.
        
        Args:
            observed_context: Versions that were visible when write was issued
                             (for writes-follow-reads checking)
        """
        op = Operation(
            op_type=OperationType.WRITE,
            key=key,
            value=value,
            observed_version=None,
            written_version=written_version,
            timestamp=timestamp
        )
        self.operations.append(op)
        
        # Check monotonic writes
        if key in self.last_write_versions:
            if written_version <= self.last_write_versions[key]:
                raise MonotonicWritesViolation(
                    f"Write version {written_version} not > prior {self.last_write_versions[key]}"
                )
        
        # Check writes-follow-reads (system must have applied based on observed context)
        for read_key, read_version in observed_context.items():
            if read_key in self.last_read_versions:
                if self.last_read_versions[read_key] > read_version:
                    raise WritesFollowReadsViolation(
                        f"Write context has {read_key}@{read_version}, "
                        f"but we read {self.last_read_versions[read_key]}"
                    )
        
        self.last_write_versions[key] = written_version
 
 
class MonotonicReadsViolation(Exception):
    """Raised when monotonic reads guarantee is violated."""
    pass
 
class ReadYourWritesViolation(Exception):
    """Raised when read-your-writes guarantee is violated."""
    pass
 
class MonotonicWritesViolation(Exception):
    """Raised when monotonic writes guarantee is violated."""
    pass
 
class WritesFollowReadsViolation(Exception):
    """Raised when writes-follow-reads guarantee is violated."""
    pass
 
 
# ════════════════════════════════════════════════════════════════════
# Demonstration: Violation Scenarios
# ════════════════════════════════════════════════════════════════════
 
def demonstrate_violations():
    """
    Show examples of each guarantee violation.
    These are the scenarios session consistency prevents.
    """
    
    # Scenario 1: Monotonic Reads Violation
    print("=== Monotonic Reads Violation ===")
    history = SessionHistory()
    history.add_read("profile", observed_version=5, timestamp=1.0)
    try:
        history.add_read("profile", observed_version=3, timestamp=2.0)  # Went backward!
    except MonotonicReadsViolation as e:
        print(f"Caught: {e}")
    
    # Scenario 2: Read-Your-Writes Violation
    print("
=== Read-Your-Writes Violation ===")
    history2 = SessionHistory()
    history2.add_write("profile", {"name": "Alice"}, written_version=10, 
                       timestamp=1.0, observed_context={})
    try:
        history2.add_read("profile", observed_version=8, timestamp=2.0)  # Didn't see write!
    except ReadYourWritesViolation as e:
        print(f"Caught: {e}")
    
    # Scenario 3: Monotonic Writes Violation
    print("
=== Monotonic Writes Violation ===")
    history3 = SessionHistory()
    history3.add_write("counter", 1, written_version=5, timestamp=1.0, observed_context={})
    try:
        history3.add_write("counter", 2, written_version=3, timestamp=2.0)  # Out of order!
    except MonotonicWritesViolation as e:
        print(f"Caught: {e}")

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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235

# Complete Session Consistency Implementation
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
import time
import json
import base64
 
@dataclass
class SessionState:
    """
    Complete state for a client session.
    Combines all four session guarantee requirements.
    """
    session_id: str
    
    # Read tracking (for MR + RYW)
    observed_reads: Dict[str, int] = field(default_factory=dict)  # key -> max version seen
    
    # Write tracking (for RYW + MW)
    written_versions: Dict[str, int] = field(default_factory=dict)  # key -> version written
    
    # Global position (for cross-key consistency)
    global_position: int = 0
    
    # Write order tracking (for MW)
    pending_writes: List[str] = field(default_factory=list)  # write IDs in order
    
    def minimum_acceptable_version(self, key: str) -> int:
        """
        Minimum version acceptable for a read on this key.
        Satisfies both RYW (must see own writes) and MR (must not go backward).
        """
        ryw_requirement = self.written_versions.get(key, 0)
        mr_requirement = self.observed_reads.get(key, 0)
        return max(ryw_requirement, mr_requirement)
    
    def record_read(self, key: str, version: int):
        """Update state after a read operation."""
        current_max = self.observed_reads.get(key, 0)
        self.observed_reads[key] = max(current_max, version)
        self.global_position = max(self.global_position, version)
    
    def record_write(self, key: str, version: int, write_id: str):
        """Update state after a write operation."""
        self.written_versions[key] = version
        self.observed_reads[key] = max(
            self.observed_reads.get(key, 0),
            version
        )
        self.global_position = max(self.global_position, version)
        self.pending_writes.append(write_id)
    
    def get_write_context(self) -> Dict[str, int]:
        """
        Get context for writes-follow-reads.
        This context must be respected by the server when applying the write.
        """
        return {
            **self.observed_reads,
            **{k: v for k, v in self.written_versions.items()}
        }
    
    def serialize(self) -> str:
        """Serialize for transmission/storage."""
        data = {
            "sid": self.session_id,
            "reads": self.observed_reads,
            "writes": self.written_versions,
            "pos": self.global_position
        }
        return base64.b64encode(json.dumps(data).encode()).decode()
    
    @classmethod
    def deserialize(cls, token: str) -> 'SessionState':
        """Restore from serialized form."""
        data = json.loads(base64.b64decode(token))
        state = cls(session_id=data["sid"])
        state.observed_reads = data.get("reads", {})
        state.written_versions = data.get("writes", {})
        state.global_position = data.get("pos", 0)
        return state
 
 
class SessionConsistentStore:
    """
    Store providing full session consistency guarantees.
    """
    
    def __init__(self, replicas: List['Replica']):
        self.replicas = replicas
        self.primary = replicas[0]  # Convention: first replica is primary
    
    def read(self, session: SessionState, key: str) -> Tuple[any, int]:
        """
        Read with session consistency guarantees.
        
        Returns: (value, version)
        Raises: ConsistencyError if no replica can satisfy requirements
        """
        min_version = session.minimum_acceptable_version(key)
        
        # Try replicas in order of freshness
        for replica in self._order_by_freshness():
            value, version = replica.read_versioned(key)
            
            if version >= min_version:
                session.record_read(key, version)
                return value, version
        
        # No replica fresh enough - try blocking read from primary
        value, version = self.primary.read_blocking(key, min_version, timeout=5.0)
        
        if version >= min_version:
            session.record_read(key, version)
            return value, version
        
        raise ConsistencyError(
            f"Cannot satisfy session requirement: need version >= {min_version}"
        )
    
    def write(self, session: SessionState, key: str, value: any) -> int:
        """
        Write with session consistency guarantees.
        
        Ensures:
        - Monotonic writes (new version > session's last write to this key)
        - Writes follow reads (write context includes observed reads)
        
        Returns: version of the write
        """
        # Build context for writes-follow-reads
        context = session.get_write_context()
        
        # Generate write ID for ordering
        write_id = f"{session.session_id}:{time.time_ns()}"
        
        # Execute write on primary with context
        version = self.primary.write_with_context(key, value, context, write_id)
        
        # Verify monotonic writes
        if key in session.written_versions:
            if version <= session.written_versions[key]:
                raise ConsistencyError(
                    f"Monotonic write violation: {version} <= {session.written_versions[key]}"
                )
        
        session.record_write(key, version, write_id)
        return version
    
    def read_multiple(self, session: SessionState, 
                      keys: List[str]) -> Dict[str, Tuple[any, int]]:
        """
        Read multiple keys with consistent session view.
        
        All reads will be from the same snapshot that satisfies
        session requirements for all keys.
        """
        # Find minimum global position needed
        min_global = session.global_position
        for key in keys:
            min_global = max(min_global, session.minimum_acceptable_version(key))
        
        # Find a replica that can satisfy all requirements
        for replica in self._order_by_freshness():
            if replica.current_position >= min_global:
                results = {}
                for key in keys:
                    value, version = replica.read_versioned(key)
                    results[key] = (value, version)
                    session.record_read(key, version)
                return results
        
        raise ConsistencyError(f"No replica at position >= {min_global}")
    
    def _order_by_freshness(self) -> List['Replica']:
        """Order replicas by current position (freshest first)."""
        return sorted(self.replicas, key=lambda r: -r.current_position)
 
 
class Replica:
    """A database replica with versioned data."""
    
    def __init__(self, replica_id: str, is_primary: bool = False):
        self.id = replica_id
        self.is_primary = is_primary
        self.data: Dict[str, Tuple[any, int]] = {}  # key -> (value, version)
        self.current_position: int = 0
        self.write_order: List[str] = []  # Ordered write IDs
    
    def read_versioned(self, key: str) -> Tuple[any, int]:
        """Read value and its version."""
        if key in self.data:
            return self.data[key]
        return None, 0
    
    def read_blocking(self, key: str, min_version: int, 
                      timeout: float) -> Tuple[any, int]:
        """Read, blocking until min_version is available."""
        deadline = time.time() + timeout
        while time.time() < deadline:
            value, version = self.read_versioned(key)
            if version >= min_version:
                return value, version
            time.sleep(0.01)
        return self.read_versioned(key)
    
    def write_with_context(self, key: str, value: any, 
                           context: Dict[str, int], write_id: str) -> int:
        """
        Execute write respecting session context.
        
        The context ensures writes-follow-reads by requiring that
        the current state includes all versions the client observed.
        """
        # Verify context is satisfied
        for ctx_key, ctx_version in context.items():
            current = self.data.get(ctx_key, (None, 0))[1]
            if current < ctx_version:
                # Context not satisfied - would violate writes-follow-reads
                # In practice, this might block or fail
                raise ConsistencyError(
                    f"Context violation: {ctx_key} at {current} < required {ctx_version}"
                )
        
        # Apply write
        self.current_position += 1
        self.data[key] = (value, self.current_position)
        self.write_order.append(write_id)
        
        return self.current_position
 
 
class ConsistencyError(Exception):
    """Raised when session consistency cannot be satisfied."""
    pass

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

# Session Boundary Management Strategies
from dataclasses import dataclass, field
from typing import Optional, Dict
from datetime import datetime, timedelta
import threading
 
# ════════════════════════════════════════════════════════════════════
# Strategy 1: Connection-Based Sessions
# ════════════════════════════════════════════════════════════════════
 
class ConnectionSession:
    """
    Session tied to database connection.
    Used by traditional database drivers.
    """
    
    # Thread-local storage for session per connection
    _local = threading.local()
    
    @classmethod
    def get_or_create(cls, connection) -> SessionState:
        """Get session for current connection."""
        if not hasattr(cls._local, 'sessions'):
            cls._local.sessions = {}
        
        conn_id = id(connection)
        if conn_id not in cls._local.sessions:
            cls._local.sessions[conn_id] = SessionState(
                session_id=f"conn-{conn_id}"
            )
        return cls._local.sessions[conn_id]
    
    @classmethod
    def clear(cls, connection):
        """Clear session when connection returns to pool."""
        if hasattr(cls._local, 'sessions'):
            cls._local.sessions.pop(id(connection), None)
 
 
# ════════════════════════════════════════════════════════════════════
# Strategy 2: Token-Based Sessions with Lifecycle Management
# ════════════════════════════════════════════════════════════════════
 
@dataclass
class ManagedSession:
    """Session with lifecycle metadata."""
    state: SessionState
    created_at: datetime
    last_access: datetime
    access_count: int = 0
    
    def is_expired(self, max_age: timedelta, max_idle: timedelta) -> bool:
        """Check if session should be expired."""
        now = datetime.now()
        if now - self.created_at > max_age:
            return True  # Absolute expiration
        if now - self.last_access > max_idle:
            return True  # Idle expiration
        return False
 
class TokenSessionManager:
    """
    Manages token-based sessions with lifecycle policies.
    """
    
    def __init__(self, 
                 max_age: timedelta = timedelta(hours=24),
                 max_idle: timedelta = timedelta(hours=1),
                 max_state_size: int = 10000):  # Max keys tracked
        self.max_age = max_age
        self.max_idle = max_idle
        self.max_state_size = max_state_size
        self.sessions: Dict[str, ManagedSession] = {}
    
    def create_session(self) -> str:
        """Create new session, return token."""
        import uuid
        session_id = str(uuid.uuid4())
        state = SessionState(session_id=session_id)
        
        self.sessions[session_id] = ManagedSession(
            state=state,
            created_at=datetime.now(),
            last_access=datetime.now()
        )
        
        return state.serialize()
    
    def get_session(self, token: str) -> Optional[SessionState]:
        """Restore session from token."""
        try:
            state = SessionState.deserialize(token)
        except:
            return None
        
        session_id = state.session_id
        
        if session_id in self.sessions:
            managed = self.sessions[session_id]
            
            # Check expiration
            if managed.is_expired(self.max_age, self.max_idle):
                del self.sessions[session_id]
                return None
            
            # Update access time
            managed.last_access = datetime.now()
            managed.access_count += 1
            
            # Merge token state with server state (client might have fresher info)
            managed.state = self._merge_states(managed.state, state)
            
            return managed.state
        else:
            # Session not on server - validate and accept token state
            if self._validate_token(state):
                self.sessions[session_id] = ManagedSession(
                    state=state,
                    created_at=datetime.now(),
                    last_access=datetime.now()
                )
                return state
            return None
    
    def update_session(self, state: SessionState) -> str:
        """Update session and return new token."""
        session_id = state.session_id
        
        if session_id in self.sessions:
            self.sessions[session_id].state = state
            self.sessions[session_id].last_access = datetime.now()
        
        # Compact state if too large
        if self._state_size(state) > self.max_state_size:
            state = self._compact_state(state)
        
        return state.serialize()
    
    def _merge_states(self, server: SessionState, 
                      client: SessionState) -> SessionState:
        """Merge server and client states (point-wise maximum)."""
        merged = SessionState(session_id=server.session_id)
        
        all_keys = set(server.observed_reads.keys()) | set(client.observed_reads.keys())
        for key in all_keys:
            merged.observed_reads[key] = max(
                server.observed_reads.get(key, 0),
                client.observed_reads.get(key, 0)
            )
        
        all_keys = set(server.written_versions.keys()) | set(client.written_versions.keys())
        for key in all_keys:
            merged.written_versions[key] = max(
                server.written_versions.get(key, 0),
                client.written_versions.get(key, 0)
            )
        
        merged.global_position = max(server.global_position, client.global_position)
        
        return merged
    
    def _state_size(self, state: SessionState) -> int:
        """Estimate state size (number of tracked keys)."""
        return len(state.observed_reads) + len(state.written_versions)
    
    def _compact_state(self, state: SessionState) -> SessionState:
        """
        Reduce state size by removing old entries.
        Trades precision for size. May require fallback to stronger reads.
        """
        # Keep only global position, drop per-key tracking
        compacted = SessionState(session_id=state.session_id)
        compacted.global_position = state.global_position
        return compacted
    
    def _validate_token(self, state: SessionState) -> bool:
        """Validate token hasn't been tampered with."""
        # In production: verify HMAC or signature
        return True

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

# Patterns for Cross-Session Consistency
 
# ════════════════════════════════════════════════════════════════════
# Pattern 1: Session Token Propagation for Microservices
# ════════════════════════════════════════════════════════════════════
 
class SessionPropagatingClient:
    """
    HTTP client that propagates session tokens between services.
    Ensures cross-service session consistency.
    """
    
    SESSION_TOKEN_HEADER = "X-Session-Token"
    
    def __init__(self, base_url: str):
        self.base_url = base_url
        self.session_token: Optional[str] = None
    
    def set_session_token(self, token: str):
        """Set token received from upstream service."""
        self.session_token = token
    
    def get_session_token(self) -> Optional[str]:
        """Get token to pass to downstream services."""
        return self.session_token
    
    def request(self, method: str, path: str, **kwargs) -> dict:
        """Make request with session token propagation."""
        headers = kwargs.get('headers', {})
        
        # Include session token if we have one
        if self.session_token:
            headers[self.SESSION_TOKEN_HEADER] = self.session_token
        
        kwargs['headers'] = headers
        response = self._do_request(method, path, **kwargs)
        
        # Capture session token from response
        if self.SESSION_TOKEN_HEADER in response.headers:
            self.session_token = response.headers[self.SESSION_TOKEN_HEADER]
        
        return response
 
 
# ════════════════════════════════════════════════════════════════════
# Pattern 2: User-Scoped Sessions for Multi-Device
# ════════════════════════════════════════════════════════════════════
 
class UserScopedSessionStore:
    """
    Store sessions keyed by user ID rather than device/connection.
    Provides cross-device session consistency.
    """
    
    def __init__(self, cache_client):
        self.cache = cache_client  # e.g., Redis
    
    def get_session(self, user_id: str) -> SessionState:
        """Get or create session for user (not device)."""
        cached = self.cache.get(f"session:{user_id}")
        
        if cached:
            return SessionState.deserialize(cached)
        
        # New session for user
        state = SessionState(session_id=f"user-{user_id}")
        self.cache.set(f"session:{user_id}", state.serialize())
        return state
    
    def update_session(self, user_id: str, state: SessionState):
        """Update user's session (atomic CAS to handle concurrent updates)."""
        key = f"session:{user_id}"
        
        for _ in range(3):  # Retry on conflict
            current = self.cache.get(key)
            if current:
                current_state = SessionState.deserialize(current)
                # Merge to handle concurrent device updates
                merged = self._merge(current_state, state)
                if self.cache.cas(key, current, merged.serialize()):
                    return
            else:
                if self.cache.setnx(key, state.serialize()):
                    return
        
        raise Exception("Failed to update session after retries")
 
 
# ════════════════════════════════════════════════════════════════════
# Pattern 3: Hybrid Consistency for Critical Paths
# ════════════════════════════════════════════════════════════════════
 
class HybridConsistencyStore:
    """
    Uses session consistency by default, strong consistency for critical paths.
    Balances performance with correctness requirements.
    """
    
    def __init__(self, db_client):
        self.db = db_client
        self.session = SessionState(session_id="hybrid")
    
    def read_session(self, key: str) -> any:
        """Read with session consistency (default)."""
        return self.db.read(key, session_token=self.session.serialize())
    
    def read_strong(self, key: str) -> any:
        """Read with strong consistency (critical paths)."""
        value = self.db.read_strong(key)  # Quorum read
        # Update session to reflect this strong read
        self.session.record_read(key, self.db.get_current_version(key))
        return value
    
    def write_session(self, key: str, value: any):
        """Write with session consistency (default)."""
        version = self.db.write(key, value, session_token=self.session.serialize())
        self.session.record_write(key, version, f"w-{time.time_ns()}")
    
    def write_strong(self, key: str, value: any):
        """Write with strong consistency (critical paths)."""
        version = self.db.write_strong(key, value)  # Quorum write
        self.session.record_write(key, version, f"w-{time.time_ns()}")