1"""Chapter 16 practice (Consistency and CAP):
2- two replicas and a network partition: CP mode refuses, AP mode accepts and diverges (Level 3)
3- reconciling after the partition: last-write-wins loses data, a counter CRDT doesn't
4- the gift-card double-spend in AP mode (Level 4)
5Standard library only. Run: python3 ch16_consistency.py"""
6
7
8class Unavailable(Exception):
9 pass
10
11
12class Replica:
13 def __init__(self, name):
14 self.name, self.data, self.clock = name, {}, {}
15
16
17class Pair:
18 """Two replicas (Mumbai, Frankfurt) that normally copy every write to each other."""
19
20 def __init__(self, mode):
21 self.mode = mode # "CP" or "AP"
22 self.a, self.b = Replica("mumbai"), Replica("frankfurt")
23 self.partitioned = False
24 self.time = 0
25
26 def other(self, r):
27 return self.b if r is self.a else self.a
28
29 def write(self, replica, key, value):
30 self.time += 1
31 if self.partitioned and self.mode == "CP":
32 raise Unavailable(f"{replica.name}: can't reach the other replica, refusing the write")
33 replica.data[key], replica.clock[key] = value, self.time
34 if not self.partitioned: # replicate synchronously while connected
35 o = self.other(replica)
36 o.data[key], o.clock[key] = value, self.time
37
38 def read(self, replica, key):
39 if self.partitioned and self.mode == "CP":
40 raise Unavailable(f"{replica.name}: might be stale, refusing the read")
41 return replica.data.get(key)
42
43 def heal_last_write_wins(self):
44 self.partitioned = False
45 for key in set(self.a.data) | set(self.b.data):
46 winner = max((self.a, self.b), key=lambda r: r.clock.get(key, 0))
47 for r in (self.a, self.b):
48 r.data[key], r.clock[key] = winner.data[key], winner.clock[key]
49
50
51class GCounter:
52 """A grow-only counter CRDT: each replica counts its own increments; merge takes the max
53 per replica, so concurrent increments are never lost."""
54
55 def __init__(self, me):
56 self.me, self.counts = me, {}
57
58 def incr(self, n=1):
59 self.counts[self.me] = self.counts.get(self.me, 0) + n
60
61 def merge(self, other):
62 for k, v in other.counts.items():
63 self.counts[k] = max(self.counts.get(k, 0), v)
64
65 def value(self):
66 return sum(self.counts.values())
67
68
69# ---- Level 4: spending a gift card on both sides of a partition ---------------------------------
70
71def try_spend(pair, replica, card, amount):
72 balance = pair.read(replica, card)
73 if balance is None or balance < amount:
74 return False
75 pair.write(replica, card, balance - amount)
76 return True
77
78
79def run_checks():
80 # CP: during a partition, both sides refuse rather than risk a wrong answer.
81 cp = Pair("CP")
82 cp.write(cp.a, "stock:42", 10)
83 cp.partitioned = True
84 for op in (lambda: cp.write(cp.a, "stock:42", 9), lambda: cp.read(cp.b, "stock:42")):
85 try:
86 op()
87 raise AssertionError("CP should refuse during a partition")
88 except Unavailable:
89 pass
90
91 # AP: both sides keep answering, and diverge.
92 ap = Pair("AP")
93 ap.write(ap.a, "bio:7", "Loves sci-fi")
94 ap.partitioned = True
95 ap.write(ap.a, "bio:7", "Loves sci-fi and history") # edit in Mumbai
96 ap.write(ap.b, "bio:7", "Now reads poetry") # edit in Frankfurt
97 assert ap.read(ap.a, "bio:7") != ap.read(ap.b, "bio:7")
98 ap.heal_last_write_wins()
99 assert ap.a.data["bio:7"] == ap.b.data["bio:7"] == "Now reads poetry" # Mumbai's edit is lost
100
101 # A counter CRDT merges concurrent increments without losing any.
102 m, f = GCounter("mumbai"), GCounter("frankfurt")
103 m.incr(3); f.incr(5) # likes counted on both sides
104 m.merge(f); f.merge(m)
105 assert m.value() == f.value() == 8
106 lww_result = 5 # last-write-wins would keep one side's 3 or 5
107 assert m.value() > lww_result
108
109 # Level 4: gift card with ₹500, AP mode, partition, two ₹400 purchases.
110 shop = Pair("AP")
111 shop.write(shop.a, "card:G1", 500)
112 shop.partitioned = True
113 ok1 = try_spend(shop, shop.a, "card:G1", 400)
114 ok2 = try_spend(shop, shop.b, "card:G1", 400)
115 assert ok1 and ok2 # ₹800 spent from a ₹500 card
116
117 # The fix: balances in CP mode. One side is refused while partitioned.
118 safe = Pair("CP")
119 safe.write(safe.a, "card:G1", 500)
120 assert try_spend(safe, safe.a, "card:G1", 400)
121 safe.partitioned = True
122 try:
123 try_spend(safe, safe.b, "card:G1", 400)
124 raise AssertionError("should refuse")
125 except Unavailable:
126 pass
127
128 print("All checks pass. AP double spend: 2 x ₹400 from a ₹500 card; "
129 "CP refused the second. LWW kept:", repr(ap.a.data["bio:7"]))
130
131
132if __name__ == "__main__":
133 run_checks()