1"""Chapter 17 practice (Transactions and locking):
2- the lost update, and three fixes: atomic update, optimistic version check, pessimistic lock
3- deadlock detection with a wait-for graph
4- seat holds that expire
5- a saga with compensation
6- Level 4: 'check then insert' double booking, fixed with a UNIQUE constraint (real sqlite3)
7Standard library only. Run: python3 ch17_transactions.py"""
8import sqlite3
9import threading
10
11
12# ---- The lost update (deterministic interleaving) ---------------------------------------------
13
14def lost_update():
15 stock = {"signing-seats": 100}
16 a_read = stock["signing-seats"] # Asha's request reads 100
17 b_read = stock["signing-seats"] # Ravi's request reads 100
18 stock["signing-seats"] = a_read - 1 # writes 99
19 stock["signing-seats"] = b_read - 1 # writes 99: Asha's booking vanished
20 return stock["signing-seats"]
21
22
23def atomic_decrement(db, n_requests):
24 """UPDATE ... SET left = left - 1 WHERE left > 0: the database does read+write as one step."""
25 sold = 0
26 for _ in range(n_requests):
27 cur = db.execute("UPDATE events SET seats_left = seats_left - 1 "
28 "WHERE id = 1 AND seats_left > 0")
29 sold += cur.rowcount
30 db.commit()
31 return sold
32
33
34class VersionedRow:
35 def __init__(self, value):
36 self.value, self.version = value, 1
37
38 def compare_and_set(self, expected_version, new_value):
39 if self.version != expected_version:
40 return False # someone changed it since you read it
41 self.value, self.version = new_value, self.version + 1
42 return True
43
44
45def optimistic_book(row, retries=5):
46 for _ in range(retries):
47 seen_value, seen_version = row.value, row.version
48 if seen_value <= 0:
49 return False
50 if row.compare_and_set(seen_version, seen_value - 1):
51 return True
52 return False
53
54
55def pessimistic_book(lock, state, n_threads=50):
56 def book():
57 with lock: # like SELECT ... FOR UPDATE: one at a time
58 if state["left"] > 0:
59 state["left"] -= 1
60 state["sold"] += 1
61 threads = [threading.Thread(target=book) for _ in range(n_threads)]
62 for t in threads:
63 t.start()
64 for t in threads:
65 t.join()
66 return state
67
68
69# ---- Deadlocks ------------------------------------------------------------------------------------
70
71def find_deadlock(waits_for):
72 """waits_for: {transaction: transaction it waits on}. Returns a cycle, or None."""
73 for start in waits_for:
74 seen, node = [], start
75 while node in waits_for:
76 if node in seen:
77 return seen[seen.index(node):]
78 seen.append(node)
79 node = waits_for[node]
80 return None
81
82
83# ---- Seat holds with expiry -------------------------------------------------------------------------
84
85class Seats:
86 def __init__(self, n, hold_seconds=600):
87 self.state = {s: ("free", None, None) for s in range(1, n + 1)} # status, user, expires
88 self.hold = hold_seconds
89
90 def _expire(self, now):
91 for s, (status, user, exp) in self.state.items():
92 if status == "held" and exp <= now:
93 self.state[s] = ("free", None, None)
94
95 def hold_seat(self, seat, user, now):
96 self._expire(now)
97 if self.state[seat][0] != "free":
98 return False
99 self.state[seat] = ("held", user, now + self.hold)
100 return True
101
102 def confirm(self, seat, user, now):
103 self._expire(now)
104 status, holder, _ = self.state[seat]
105 if status == "held" and holder == user:
106 self.state[seat] = ("sold", user, None)
107 return True
108 return False
109
110
111# ---- A saga --------------------------------------------------------------------------------------------
112
113def run_saga(steps):
114 """steps: [(do, undo)]. If a step fails, undo the completed ones in reverse order."""
115 done, log = [], []
116 for do, undo in steps:
117 try:
118 log.append(do())
119 done.append(undo)
120 except Exception as e:
121 log.append(f"failed: {e}")
122 for u in reversed(done):
123 log.append(u())
124 return False, log
125 return True, log
126
127
128# ---- Level 4: check, then insert -----------------------------------------------------------------------
129
130def make_bookings_db(unique):
131 db = sqlite3.connect(":memory:")
132 constraint = ", UNIQUE (event_id, seat)" if unique else ""
133 db.execute(f"CREATE TABLE bookings (id INTEGER PRIMARY KEY, event_id INT, seat INT, user TEXT{constraint})")
134 return db
135
136
137def buggy_book(db, event, seat, user, other_request_runs_now=None):
138 taken = db.execute("SELECT 1 FROM bookings WHERE event_id=? AND seat=?", (event, seat)).fetchone()
139 if taken:
140 return False
141 if other_request_runs_now:
142 other_request_runs_now() # another request checks and inserts in this gap
143 db.execute("INSERT INTO bookings (event_id, seat, user) VALUES (?, ?, ?)", (event, seat, user))
144 db.commit()
145 return True
146
147
148def fixed_book(db, event, seat, user):
149 try:
150 db.execute("INSERT INTO bookings (event_id, seat, user) VALUES (?, ?, ?)", (event, seat, user))
151 db.commit()
152 return True
153 except sqlite3.IntegrityError: # the database itself refuses the second booking
154 db.rollback()
155 return False
156
157
158def run_checks():
159 assert lost_update() == 99 # two bookings, but only one seat counted
160
161 db = sqlite3.connect(":memory:")
162 db.execute("CREATE TABLE events (id INTEGER PRIMARY KEY, seats_left INT)")
163 db.execute("INSERT INTO events VALUES (1, 100)")
164 assert atomic_decrement(db, 130) == 100 # never oversells
165 assert db.execute("SELECT seats_left FROM events").fetchone()[0] == 0
166
167 row = VersionedRow(1)
168 stale_version = row.version
169 assert optimistic_book(row) # Asha books the last seat
170 assert not row.compare_and_set(stale_version, 0) # Ravi's write with an old version is rejected
171 assert not optimistic_book(row) # and on retry he sees 0 left
172
173 state = pessimistic_book(threading.Lock(), {"left": 30, "sold": 0})
174 assert state == {"left": 0, "sold": 30}
175
176 assert find_deadlock({"T1": "T2", "T2": "T1"}) in (["T1", "T2"], ["T2", "T1"])
177 assert find_deadlock({"T1": "T2", "T2": "T3"}) is None
178
179 seats = Seats(2, hold_seconds=600)
180 assert seats.hold_seat(1, "asha", now=0)
181 assert not seats.hold_seat(1, "ravi", now=100) # held by Asha
182 assert seats.hold_seat(1, "ravi", now=601) # Asha's hold expired
183 assert not seats.confirm(1, "asha", now=602) # too late for Asha
184 assert seats.confirm(1, "ravi", now=650)
185
186 ledger = []
187 ok, log = run_saga([
188 (lambda: ledger.append("seat held") or "hold seat", lambda: ledger.remove("seat held") or "release seat"),
189 (lambda: ledger.append("₹499 charged") or "charge card", lambda: ledger.remove("₹499 charged") or "refund ₹499"),
190 (lambda: (_ for _ in ()).throw(RuntimeError("email service down")), lambda: "unsend email"),
191 ])
192 assert not ok and ledger == [] and log[-2:] == ["refund ₹499", "release seat"]
193
194 # Level 4
195 bad = make_bookings_db(unique=False)
196 assert buggy_book(bad, 1, 14, "asha",
197 other_request_runs_now=lambda: buggy_book(bad, 1, 14, "ravi"))
198 count = bad.execute("SELECT COUNT(*) FROM bookings WHERE event_id=1 AND seat=14").fetchone()[0]
199 assert count == 2 # seat 14 sold twice
200
201 good = make_bookings_db(unique=True)
202 assert fixed_book(good, 1, 14, "ravi")
203 assert not fixed_book(good, 1, 14, "asha")
204 assert good.execute("SELECT COUNT(*) FROM bookings").fetchone()[0] == 1
205
206 print("All checks pass. Lost update left 99 (should be 98); atomic update sold exactly 100 of 130 "
207 "requests; check-then-insert sold seat 14 twice; UNIQUE stopped it.")
208
209
210if __name__ == "__main__":
211 run_checks()