1"""Chapter 14 practice (Replication):
2- a leader with asynchronous followers that lag (Level 3)
3- read-your-writes using the write's log position
4- quorums: N=3 with W+R>N always sees the latest write
5- async failover losing the writes a follower hadn't received
6- the 'write to primary, read from replica' bug (Level 4)
7Standard library only. Run: python3 ch14_replication.py"""
8
9
10class Node:
11 def __init__(self, name):
12 self.name, self.data, self.applied = name, {}, 0 # applied = log position reached
13
14
15class Leader(Node):
16 def __init__(self, name, followers):
17 super().__init__(name)
18 self.log, self.followers = [], followers
19
20 def write(self, key, value):
21 self.log.append((key, value))
22 self.data[key] = value
23 self.applied = len(self.log)
24 return self.applied # the write's log position (LSN)
25
26 def replicate(self, follower, max_entries):
27 """Asynchronous: ship up to max_entries log entries the follower hasn't applied."""
28 for key, value in self.log[follower.applied:follower.applied + max_entries]:
29 follower.data[key] = value
30 follower.applied += 1
31
32
33def read_your_writes(user_last_lsn, leader, follower, key):
34 """Serve from the follower only if it has caught up to this user's last write."""
35 if follower.applied >= user_last_lsn:
36 return follower.data.get(key), follower.name
37 return leader.data.get(key), leader.name
38
39
40# ---- Quorums (leaderless) ---------------------------------------------------------------------
41
42class Replica:
43 def __init__(self):
44 self.store = {} # key -> (version, value)
45 self.up = True
46
47
48def quorum_write(replicas, key, value, version, w):
49 acks = 0
50 for r in replicas:
51 if r.up:
52 r.store[key] = (version, value)
53 acks += 1
54 if acks == w:
55 return True # the rest may get it later (or never)
56 return False
57
58
59def quorum_read(replicas, key, r):
60 answers = [rep.store.get(key, (0, None)) for rep in replicas if rep.up][:r]
61 if len(answers) < r:
62 raise RuntimeError("not enough replicas")
63 return max(answers)[1] # the highest version wins
64
65
66# ---- Level 4: read from a replica right after writing ------------------------------------------
67
68def buggy_update_address(leader, follower, user, address):
69 leader.write(f"address:{user}", address)
70 return follower.data.get(f"address:{user}") # BUG: the replica may be behind
71
72
73def fixed_update_address(leader, follower, user, address):
74 lsn = leader.write(f"address:{user}", address)
75 value, _ = read_your_writes(lsn, leader, follower, f"address:{user}")
76 return value
77
78
79def run_checks():
80 f1 = Node("replica-1")
81 leader = Leader("primary", [f1])
82 leader.write("address:7", "Old Street 1")
83 leader.replicate(f1, max_entries=10)
84 assert f1.data["address:7"] == "Old Street 1"
85
86 # Lag: the next write hasn't been shipped yet.
87 lsn = leader.write("address:7", "New Road 9")
88 assert f1.data["address:7"] == "Old Street 1" # stale read on the replica
89 value, served_by = read_your_writes(lsn, leader, f1, "address:7")
90 assert value == "New Road 9" and served_by == "primary"
91 leader.replicate(f1, max_entries=10)
92 value, served_by = read_your_writes(lsn, leader, f1, "address:7")
93 assert value == "New Road 9" and served_by == "replica-1" # caught up: the replica can serve
94
95 # Async failover: the leader dies with 3 writes the replica never got.
96 f2 = Node("replica-2")
97 l2 = Leader("primary", [f2])
98 for i in range(10):
99 l2.write(f"order:{i}", "paid")
100 l2.replicate(f2, max_entries=7)
101 lost = [k for k in l2.data if k not in f2.data]
102 assert len(lost) == 3 # promote f2 and these 3 are gone
103
104 # Quorums with N = 3.
105 reps = [Replica(), Replica(), Replica()]
106 assert quorum_write(reps, "stock:42", 10, version=1, w=3)
107 assert quorum_write(reps, "stock:42", 9, version=2, w=2) # third replica still has version 1
108 assert reps[2].store["stock:42"] == (1, 10)
109 assert quorum_read(reps, "stock:42", r=2) == 9 # W + R = 4 > 3: overlap guaranteed
110 reps_order = [reps[2], reps[0], reps[1]] # even if the stale replica answers first
111 assert quorum_read(reps_order, "stock:42", r=2) == 9
112 assert quorum_read(reps_order, "stock:42", r=1) == 10 # W + R = 3, not > 3: can be stale
113 reps[0].up = False
114 assert quorum_read([reps[2], reps[1]], "stock:42", r=2) == 9 # still fine with one down
115
116 # Level 4
117 f3 = Node("replica-3")
118 l3 = Leader("primary", [f3])
119 l3.write("address:7", "Old Street 1")
120 l3.replicate(f3, 10)
121 assert buggy_update_address(l3, f3, 7, "New Road 9") == "Old Street 1"
122 assert fixed_update_address(l3, f3, 7, "Lake View 3") == "Lake View 3"
123
124 print("All checks pass. Async failover lost:", lost)
125
126
127if __name__ == "__main__":
128 run_checks()