1"""Chapter 10 practice: a Snowflake-style ID generator and a UUIDv7 builder (Level 3 Build),
2and the 'clock went backwards' bug (Level 4 Debug). Standard library only.
3Run: python3 ch10_ids.py"""
4import os
5import string
6import time
7import uuid
8
9ALPHABET = string.digits + string.ascii_lowercase + string.ascii_uppercase
10
11
12def to_base62(n):
13 out = []
14 while n:
15 n, r = divmod(n, 62)
16 out.append(ALPHABET[r])
17 return "".join(reversed(out)) or "0"
18
19
20# ---- Snowflake-style IDs: 41 bits time | 10 bits machine | 12 bits sequence ---------------
21
22EPOCH_MS = 1_767_225_600_000 # 2026-01-01 00:00:00 UTC, our own starting point
23MACHINE_BITS, SEQ_BITS = 10, 12
24MAX_MACHINE, MAX_SEQ = (1 << MACHINE_BITS) - 1, (1 << SEQ_BITS) - 1
25
26
27class ClockMovedBackwards(Exception):
28 pass
29
30
31class Snowflake:
32 def __init__(self, machine_id, clock_ms=None):
33 if not 0 <= machine_id <= MAX_MACHINE:
34 raise ValueError("machine_id must fit in 10 bits")
35 self.machine_id = machine_id
36 self.clock_ms = clock_ms or (lambda: int(time.time() * 1000))
37 self.last_ms = -1
38 self.seq = 0
39
40 def next_id(self):
41 now = self.clock_ms()
42 if now < self.last_ms:
43 raise ClockMovedBackwards(f"clock went back {self.last_ms - now} ms")
44 if now == self.last_ms:
45 self.seq = (self.seq + 1) & MAX_SEQ
46 if self.seq == 0: # 4,096 IDs used this millisecond: wait
47 while now <= self.last_ms:
48 now = self.clock_ms()
49 else:
50 self.seq = 0
51 self.last_ms = now
52 return ((now - EPOCH_MS) << (MACHINE_BITS + SEQ_BITS)) | (self.machine_id << SEQ_BITS) | self.seq
53
54
55def decode(snowflake_id):
56 return {
57 "ms": (snowflake_id >> (MACHINE_BITS + SEQ_BITS)) + EPOCH_MS,
58 "machine": (snowflake_id >> SEQ_BITS) & MAX_MACHINE,
59 "seq": snowflake_id & MAX_SEQ,
60 }
61
62
63# ---- UUIDv7: 48 bits of Unix milliseconds, then version, randomness, variant -------------
64
65def uuid7(ms=None):
66 ms = int(time.time() * 1000) if ms is None else ms
67 rand = int.from_bytes(os.urandom(10), "big")
68 value = (ms & ((1 << 48) - 1)) << 80 # 48-bit timestamp
69 value |= 0x7 << 76 # version 7
70 value |= ((rand >> 62) & 0xFFF) << 64 # 12 random bits
71 value |= 0b10 << 62 # variant
72 value |= rand & ((1 << 62) - 1) # 62 random bits
73 return uuid.UUID(int=value)
74
75
76# ---- Level 4: the generator that trusts the clock ------------------------------------------
77
78class BuggySnowflake(Snowflake):
79 def next_id(self):
80 now = self.clock_ms()
81 if now == self.last_ms:
82 self.seq = (self.seq + 1) & MAX_SEQ
83 else:
84 self.seq = 0 # BUG: a step BACK in time also lands here
85 self.last_ms = now
86 return ((now - EPOCH_MS) << (MACHINE_BITS + SEQ_BITS)) | (self.machine_id << SEQ_BITS) | self.seq
87
88
89def run_checks():
90 # Snowflake: unique and time-ordered on one machine, even many per millisecond.
91 t = [EPOCH_MS + 1_000]
92 gen = Snowflake(machine_id=5, clock_ms=lambda: t[0])
93 ids = []
94 for i in range(10_000):
95 if i % 3000 == 0:
96 t[0] += 1 # the clock ticks every 3,000 IDs
97 ids.append(gen.next_id())
98 assert len(set(ids)) == len(ids)
99 assert ids == sorted(ids)
100 d = decode(ids[0])
101 assert d["machine"] == 5 and d["ms"] == EPOCH_MS + 1_001 and d["seq"] == 0
102
103 # Sequence overflow: the 4,097th ID in one millisecond waits for the next millisecond.
104 ticks = iter([EPOCH_MS + 5] * 4097 + [EPOCH_MS + 6] * 10)
105 gen2 = Snowflake(machine_id=1, clock_ms=lambda: next(ticks))
106 burst = [gen2.next_id() for _ in range(4097)]
107 assert len(set(burst)) == 4097
108 assert decode(burst[-1])["ms"] == EPOCH_MS + 6 and decode(burst[-1])["seq"] == 0
109
110 # Two machines never collide: the machine bits differ.
111 a = Snowflake(1, clock_ms=lambda: EPOCH_MS + 9).next_id()
112 b = Snowflake(2, clock_ms=lambda: EPOCH_MS + 9).next_id()
113 assert a != b
114
115 # The clock going backwards is refused.
116 t3 = [EPOCH_MS + 100]
117 g3 = Snowflake(3, clock_ms=lambda: t3[0])
118 g3.next_id()
119 t3[0] -= 5
120 try:
121 g3.next_id()
122 raise AssertionError("should refuse")
123 except ClockMovedBackwards:
124 pass
125
126 # 41 bits of milliseconds last about 69.7 years; a 64-bit ID fits in 11 base62 characters.
127 assert abs((1 << 41) / 1000 / 86_400 / 365.25 - 69.68) < 0.01
128 assert len(to_base62((1 << 63) - 1)) == 11
129
130 # UUIDv7: version 7, variant RFC, and sorted by time.
131 u1, u2 = uuid7(ms=1_767_225_600_000), uuid7(ms=1_767_225_600_001)
132 assert u1.version == 7 and u1.variant == uuid.RFC_4122
133 assert u1 < u2 and (u1.int >> 80) == 1_767_225_600_000
134 if hasattr(uuid, "uuid7"): # Python 3.14+ has one built in
135 assert uuid.uuid7().version == 7
136 v4 = uuid.uuid4()
137 assert v4.version == 4
138
139 # Level 4: the buggy generator hands out a duplicate after the clock steps back.
140 t4 = [EPOCH_MS + 500]
141 bad = BuggySnowflake(7, clock_ms=lambda: t4[0])
142 first = bad.next_id() # at 500 ms
143 t4[0] -= 1
144 bad.next_id() # at 499 ms (clock corrected by NTP)
145 t4[0] += 1
146 again = bad.next_id() # at 500 ms again, sequence reset to 0
147 assert again == first, "expected a duplicate"
148
149 print("All checks pass. Sample Snowflake:", ids[0], "=", decode(ids[0]),
150 "| base62:", to_base62(ids[0]), "| UUIDv7:", u1)
151
152
153if __name__ == "__main__":
154 run_checks()