Pythonconsistency_patterns.py
Copy
1
import threading
2
import time
3
import random
4
import sqlite3
5
import contextlib
6
from dataclasses import dataclass, field
7
 
8
# ── Setup: in-memory SQLite with optimistic locking ─────────────────
9
conn = sqlite3.connect(':memory:', check_same_thread=False)
10
conn.row_factory = sqlite3.Row
11
lock = threading.Lock()
12
 
13
with lock:
14
    conn.execute("""
15
        CREATE TABLE accounts (
16
            id INTEGER PRIMARY KEY,
17
            owner TEXT NOT NULL,
18
            balance REAL NOT NULL,
19
            version INTEGER NOT NULL DEFAULT 0,
20
            updated_at REAL
21
        )
22
    """)
23
    conn.executemany(
24
        "INSERT INTO accounts (id, owner, balance) VALUES (?, ?, ?)",
25
        [(1, "Alice", 1000.0), (2, "Bob", 500.0)]
26
    )
27
    conn.commit()
28
 
29
# ── Optimistic locking: read-version, check-on-update ───────────────
30
class OptimisticLockError(Exception):
31
    pass
32
 
33
def get_account(account_id: int) -> dict:
34
    with lock:
35
        row = conn.execute("SELECT * FROM accounts WHERE id = ?", (account_id,)).fetchone()
36
        return dict(row)
37
 
38
def update_balance(account_id: int, new_balance: float, expected_version: int) -> bool:
39
    """Update only if version matches — fails if concurrent update occurred."""
40
    with lock:
41
        result = conn.execute(
42
            "UPDATE accounts SET balance = ?, version = version + 1, updated_at = ? "
43
            "WHERE id = ? AND version = ?",
44
            (new_balance, time.time(), account_id, expected_version)
45
        )
46
        conn.commit()
47
        if result.rowcount == 0:
48
            raise OptimisticLockError(f"Concurrent update detected on account {account_id}")
49
        return True
50
 
51
def transfer_with_retry(from_id: int, to_id: int, amount: float, max_retries: int = 3) -> str:
52
    """Transfer with optimistic locking and retry."""
53
    for attempt in range(max_retries):
54
        try:
55
            # Read both accounts
56
            source = get_account(from_id)
57
            dest = get_account(to_id)
58
            
59
            if source['balance'] < amount:
60
                return "insufficient_funds"
61
            
62
            # Simulate processing time (increases conflict chance)
63
            time.sleep(random.uniform(0.001, 0.005))
64
            
65
            # Attempt updates
66
            update_balance(from_id, source['balance'] - amount, source['version'])
67
            update_balance(to_id, dest['balance'] + amount, dest['version'])
68
            return f"success (attempt {attempt + 1})"
69
        
70
        except OptimisticLockError as e:
71
            if attempt == max_retries - 1:
72
                return f"failed after {max_retries} retries: {e}"
73
            time.sleep(0.01 * (2 ** attempt))  # Exponential backoff
74
    
75
    return "failed"
76
 
77
# ── Simulate concurrent transfers ────────────────────────────────────
78
results = []
79
success_count = 0
80
retry_count = 0
81
failures = 0
82
 
83
def do_transfer(i: int):
84
    # Randomly transfer between accounts
85
    result = transfer_with_retry(1, 2, 10.0)
86
    results.append(result)
87
 
88
threads = [threading.Thread(target=do_transfer, args=(i,)) for i in range(20)]
89
start = time.time()
90
for t in threads:
91
    t.start()
92
for t in threads:
93
    t.join()
94
elapsed = time.time() - start
95
 
96
# Count results
97
for r in results:
98
    if "success" in r:
99
        success_count += 1
100
        if "attempt 2" in r or "attempt 3" in r:
101
            retry_count += 1
102
    else:
103
        failures += 1
104
 
105
print("=== Optimistic Locking Results ===")
106
print(f"Transfers attempted: {len(results)}")
107
print(f"Successful: {success_count}")
108
print(f"With retries: {retry_count}")
109
print(f"Failed: {failures}")
110
print(f"Time: {elapsed:.3f}s")
111
 
112
alice = get_account(1)
113
bob = get_account(2)
114
print(f"\nAlice: ${alice['balance']:.2f} (v{alice['version']})")
115
print(f"Bob:   ${bob['balance']:.2f} (v{bob['version']})")
116
print(f"Total: ${alice['balance'] + bob['balance']:.2f} (should be $1500)")
117
 
118
# ── Eventual consistency: CRDT-like counter ──────────────────────────
119
@dataclass
120
class GCounter:
121
    """Grow-only counter (CRDT) — eventually consistent across nodes."""
122
    node_id: str
123
    counts: dict[str, int] = field(default_factory=dict)
124
    
125
    def __post_init__(self):
126
        self.counts[self.node_id] = 0
127
    
128
    def increment(self, amount: int = 1):
129
        self.counts[self.node_id] = self.counts.get(self.node_id, 0) + amount
130
    
131
    def merge(self, other: 'GCounter'):
132
        """Merge: take max of each node's count."""
133
        for node, count in other.counts.items():
134
            self.counts[node] = max(self.counts.get(node, 0), count)
135
    
136
    @property
137
    def value(self) -> int:
138
        return sum(self.counts.values())
139
 
140
# Simulate two nodes incrementing independently then merging
141
node_a = GCounter("A")
142
node_b = GCounter("B")
143
 
144
node_a.increment(5)  # 5 events on node A
145
node_b.increment(3)  # 3 events on node B (concurrent)
146
 
147
print("\n=== CRDT G-Counter ===")
148
print(f"Node A value (before merge): {node_a.value}")
149
print(f"Node B value (before merge): {node_b.value}")
150
 
151
# Sync: merge A→B and B→A
152
node_a.merge(node_b)
153
node_b.merge(node_a)
154
 
155
print(f"Node A value (after merge): {node_a.value} ✓")
156
print(f"Node B value (after merge): {node_b.value} ✓")
157
print("Both nodes converge to 8 (eventual consistency)")
158