1"""Chapter 29: Failures, retries, circuit breakers. Tested with Python 3, standard library only.
2
3Running example: the bookstore's checkout, which calls the payment gateway, the
4inventory service and (for the "you may also like" box) the recommendations service.
5"""
6import random
7
8
9# ---------- 29.3 nines ----------
10
11def downtime_minutes(availability, days):
12 return (1 - availability) * days * 24 * 60
13
14
15def serial(*parts):
16 out = 1.0
17 for a in parts:
18 out *= a
19 return out
20
21
22def parallel(a, copies):
23 return 1 - (1 - a) ** copies
24
25
26# ---------- 29.4 timeouts from measured latency ----------
27
28def percentile(samples, p):
29 s = sorted(samples)
30 return s[min(len(s) - 1, int(p / 100 * len(s)))]
31
32
33# ---------- 29.5 retries ----------
34
35def success_with_retries(p_fail, retries):
36 """Independent failures only. Real outages are not independent."""
37 return 1 - p_fail ** (retries + 1)
38
39
40def amplification(attempts_per_layer, layers):
41 return attempts_per_layer ** layers
42
43
44def backoff_full_jitter(attempt, base=0.1, cap=10.0, rnd=random):
45 return rnd.uniform(0, min(cap, base * 2 ** attempt))
46
47
48# ---------- 29.6 retry storms ----------
49
50def simulate(capacity, base_load, blip_at, blip_len, retry_factor, budget=None, seconds=60, max_attempts=4):
51 """Each second: offered = new requests + retries of last second's failures.
52 The server serves up to `capacity`; above that, everything it can't serve fails.
53 `retry_factor`: retries generated per failure (1.5 = some failures retried at two layers).
54 `budget` caps retries at a fraction of new requests (a retry budget)."""
55 retries_next, history = 0.0, []
56 for t in range(seconds):
57 cap = 0 if blip_at <= t < blip_at + blip_len else capacity
58 offered = base_load + retries_next
59 served = min(offered, cap)
60 failed = offered - served
61 retries_next = min(failed * retry_factor, base_load * (max_attempts - 1)) # clients give up after 4 tries
62 if budget is not None:
63 retries_next = min(retries_next, budget * base_load)
64 history.append((t, round(offered), round(served)))
65 return history
66
67
68# ---------- 29.7 circuit breaker ----------
69
70class CircuitBreaker:
71 def __init__(self, threshold=5, cooldown=30, counts=lambda status: status >= 500):
72 self.threshold, self.cooldown, self.counts = threshold, cooldown, counts
73 self.state, self.failures, self.opened_at = "closed", 0, None
74
75 def allow(self, now):
76 if self.state == "open" and now - self.opened_at >= self.cooldown:
77 self.state = "half-open" # let one trial call through
78 return self.state != "open"
79
80 def record(self, status, now):
81 if self.counts(status):
82 self.failures += 1
83 if self.state == "half-open" or self.failures >= self.threshold:
84 self.state, self.opened_at = "open", now
85 else:
86 if self.state == "half-open":
87 self.state = "closed"
88 self.failures = 0
89
90
91# ---------- 29.8 bulkheads ----------
92
93def threads_needed(rps, seconds):
94 return rps * seconds
95
96
97def shared_pool(pool, deps):
98 """deps: {name: (rps, latency_s)}. One pool for everything."""
99 need = sum(threads_needed(r, l) for r, l in deps.values())
100 return need <= pool
101
102
103def bulkheads(pools, deps):
104 return {name: threads_needed(*deps[name]) <= pools[name] for name in deps}
105
106
107# ---------- 29.9 graceful degradation ----------
108
109def recommendations_box(reco_up, cache):
110 if reco_up:
111 return "personal picks"
112 if cache:
113 return "yesterday's picks (cached)"
114 return "bestsellers (static)"
115
116
117# ---------- 29.10 health checks ----------
118
119def detection_seconds(interval, unhealthy_threshold):
120 return interval * unhealthy_threshold
121
122
123# ---------- 29.11 RPO ----------
124
125def data_lost_orders(replication_lag_s, orders_per_s):
126 return replication_lag_s * orders_per_s
127
128
129# ---------- checks ----------
130
131def run_checks():
132 # nines
133 assert round(downtime_minutes(0.999, 30)) == 43
134 assert round(downtime_minutes(0.9999, 365)) == 53
135 assert round(downtime_minutes(0.99999, 365), 1) == 5.3
136 assert round(serial(0.999, 0.999, 0.9995), 4) == 0.9975 # checkout needs all three
137 assert round(parallel(0.99, 2), 4) == 0.9999 # two independent copies
138
139 # timeouts from measured latency
140 rnd = random.Random(29)
141 lat = [rnd.lognormvariate(4.5, 0.4) for _ in range(10_000)] # ms, median ~90
142 p50, p99, p999 = (round(percentile(lat, p)) for p in (50, 99, 99.9))
143 assert 80 < p50 < 100 and 200 < p99 < 260 and p999 < 400
144
145 # retries
146 assert round(success_with_retries(0.01, 2), 6) == 0.999999
147 assert amplification(4, 3) == 64 # 3 retries at 3 layers
148 delays = [backoff_full_jitter(n, rnd=rnd) for n in range(6)]
149 assert all(0 <= d <= min(10, 0.1 * 2 ** n) for n, d in enumerate(delays))
150
151 # retry storm: capacity 1,000/s, normal load 800/s, a 3-second outage
152 storm = simulate(1_000, 800, blip_at=10, blip_len=3, retry_factor=1.5)
153 assert storm[-1][1] == 3_200 and storm[-1][2] == 1_000 # 47 s later: 4x the load, still stuck
154 assert all(o == 3_200 for t, o, _ in storm if t >= 14) # stable overload ("metastable")
155 calm = simulate(1_000, 800, blip_at=10, blip_len=3, retry_factor=1.5, budget=0.1)
156 assert calm[20][1] == 800 # recovered with a 10% budget
157
158 # circuit breaker
159 cb = CircuitBreaker(threshold=5, cooldown=30)
160 for t in range(5):
161 assert cb.allow(t)
162 cb.record(503, t)
163 assert cb.state == "open" and not cb.allow(10) # fail fast
164 assert cb.allow(35) and cb.state == "half-open" # one trial after cooldown
165 cb.record(200, 35)
166 assert cb.state == "closed"
167
168 # bulkheads: recommendations get slow (5 s); checkout and inventory are fine
169 deps = {"payment": (50, 0.4), "inventory": (100, 0.05), "reco": (100, 5.0)}
170 assert not shared_pool(200, deps) # 20 + 5 + 500 > 200: everything stalls
171 iso = bulkheads({"payment": 60, "inventory": 40, "reco": 100}, deps)
172 assert iso == {"payment": True, "inventory": True, "reco": False} # only reco suffers
173
174 # degradation
175 assert recommendations_box(False, ["dune"]) == "yesterday's picks (cached)"
176 assert recommendations_box(False, []) == "bestsellers (static)"
177
178 # health checks and DR
179 assert detection_seconds(10, 3) == 30
180 assert data_lost_orders(2, 50) == 100 # async replica 2 s behind
181
182 # Level 4: a breaker that counts 404s
183 buggy = CircuitBreaker(threshold=5, cooldown=30, counts=lambda s: s >= 400)
184 for t in range(5):
185 buggy.record(404, t) # a crawler asks for missing books
186 assert not buggy.allow(6) # real checkouts now blocked
187 fixed = CircuitBreaker(threshold=5, cooldown=30)
188 for t in range(5):
189 fixed.record(404, t)
190 assert fixed.allow(6)
191
192 return {"latency_ms": (p50, p99, p999), "storm_last": storm[-1], "calm_t20": calm[20]}
193
194
195if __name__ == "__main__":
196 print("All checks pass.", run_checks())