1"""Chapter 39 (case study: file storage and sync, Google Drive / Dropbox) deep-dive code.
21. Chunking: fixed-size vs content-defined (a rolling hash), and what an edit costs.
32. De-duplication: a content-addressed block store.
43. Sync and conflicts: versions with compare-and-set; conflicted copies.
54. Metadata: file -> list of block hashes; a change journal read by cursor.
65. Presigned uploads: a signed, expiring URL.
7Run: python3 ch39_file_sync.py"""
8import hashlib
9import hmac
10import random
11
12
13# ---- 1. chunking ---------------------------------------------------------------------
14
15def fixed_chunks(data, size):
16 return [data[i:i + size] for i in range(0, len(data), size)]
17
18
19def cdc_chunks(data, avg=4096, min_size=1024, max_size=16384, window=48):
20 """Content-defined chunking: cut where a rolling hash of the last `window` bytes hits a pattern.
21 Cuts depend on content, so an insertion only changes chunks near it."""
22 mask = avg - 1
23 base, mod = 257, (1 << 61) - 1
24 power = pow(base, window, mod)
25 chunks, start, h = [], 0, 0
26 for i, byte in enumerate(data):
27 h = (h * base + byte) % mod
28 if i - start >= window:
29 h = (h - data[i - window] * power) % mod
30 length = i - start + 1
31 if (length >= min_size and (h & mask) == mask) or length >= max_size:
32 chunks.append(data[start:i + 1])
33 start, h = i + 1, 0
34 if start < len(data):
35 chunks.append(data[start:])
36 return chunks
37
38
39def h(b):
40 return hashlib.sha256(b).hexdigest()
41
42
43def new_chunks(old, new):
44 old_set = {h(c) for c in old}
45 return [c for c in new if h(c) not in old_set]
46
47
48# ---- 2. block store ----------------------------------------------------------------
49
50class BlockStore:
51 def __init__(self):
52 self.blocks = {}
53
54 def missing(self, hashes):
55 return [x for x in hashes if x not in self.blocks]
56
57 def put(self, data):
58 self.blocks[h(data)] = data
59
60 def size(self):
61 return sum(len(b) for b in self.blocks.values())
62
63
64# ---- 3/4. metadata, versions, journal -----------------------------------------------
65
66class Metadata:
67 def __init__(self):
68 self.files = {} # path -> (version, [block hashes])
69 self.journal = [] # (seq, path, version)
70
71 def commit(self, path, base_version, blocks):
72 current = self.files.get(path, (0, None))[0]
73 if base_version != current: # someone else changed it since you synced
74 return False, current
75 v = current + 1
76 self.files[path] = (v, blocks)
77 self.journal.append((len(self.journal) + 1, path, v))
78 return True, v
79
80 def changes_since(self, cursor):
81 return [e for e in self.journal if e[0] > cursor]
82
83
84def save_with_conflict_copy(meta, path, base_version, blocks, device):
85 ok, v = meta.commit(path, base_version, blocks)
86 if ok:
87 return path
88 copy = f"{path.rsplit('.', 1)[0]} ({device}'s conflicted copy).{path.rsplit('.', 1)[1]}"
89 meta.commit(copy, 0, blocks)
90 return copy
91
92
93# ---- 5. presigned upload ------------------------------------------------------------
94
95SECRET = b"storage-signing-key"
96
97
98def presign(block_hash, expires_at):
99 sig = hmac.new(SECRET, f"PUT {block_hash} {expires_at}".encode(), hashlib.sha256).hexdigest()
100 return f"https://blocks.example/{block_hash}?expires={expires_at}&sig={sig}"
101
102
103def verify_presigned(url, now):
104 path, query = url.split("?")
105 block_hash = path.rsplit("/", 1)[1]
106 params = dict(p.split("=") for p in query.split("&"))
107 expected = hmac.new(SECRET, f"PUT {block_hash} {params['expires']}".encode(), hashlib.sha256).hexdigest()
108 return hmac.compare_digest(expected, params["sig"]) and now <= int(params["expires"])
109
110
111# ---- checks -------------------------------------------------------------------------
112
113def run_checks():
114 rnd = random.Random(39)
115 doc = bytes(rnd.getrandbits(8) for _ in range(200_000)) # a 200 KB file
116 edited = b"NEW TITLE PAGE\n" + doc # insert 15 bytes at the start
117
118 # 1. fixed chunks: everything shifts, so every chunk changes
119 f_old, f_new = fixed_chunks(doc, 4096), fixed_chunks(edited, 4096)
120 fixed_reupload = len(new_chunks(f_old, f_new))
121 assert fixed_reupload == len(f_new) # 49 of 49 chunks re-uploaded
122
123 # content-defined chunks: only the first chunk changes
124 c_old, c_new = cdc_chunks(doc), cdc_chunks(edited)
125 cdc_reupload = len(new_chunks(c_old, c_new))
126 assert cdc_reupload <= 2 and len(c_new) > 20
127
128 # 2. de-duplication: two users upload the same 200 KB file
129 store = BlockStore()
130 hashes = [h(c) for c in c_old]
131 for c in c_old:
132 if h(c) in store.missing([h(c)]):
133 store.put(c)
134 before = store.size()
135 assert store.missing(hashes) == [] # second user uploads nothing
136 for c in c_new: # the edit adds only new blocks
137 store.put(c)
138 assert store.size() - before < 20_000 # ~1 chunk, not 200 KB
139
140 # 3/4. sync with versions: laptop and phone both edit v1 offline
141 meta = Metadata()
142 ok, v1 = meta.commit("/Reading list.txt", 0, ["h1"])
143 assert ok and v1 == 1
144 laptop = save_with_conflict_copy(meta, "/Reading list.txt", 1, ["h2"], "laptop")
145 phone = save_with_conflict_copy(meta, "/Reading list.txt", 1, ["h3"], "phone")
146 assert laptop == "/Reading list.txt"
147 assert phone == "/Reading list (phone's conflicted copy).txt"
148 assert meta.files["/Reading list.txt"] == (2, ["h2"])
149 # a third device catches up from its cursor
150 assert [(p, v) for _, p, v in meta.changes_since(1)] == [("/Reading list.txt", 2),
151 ("/Reading list (phone's conflicted copy).txt", 1)]
152
153 # 5. presigned URLs
154 url = presign(hashes[0], expires_at=1_000)
155 assert verify_presigned(url, now=900)
156 assert not verify_presigned(url, now=1_001) # expired
157 assert not verify_presigned(url.replace(hashes[0], hashes[1]), now=900) # tampered
158
159 # Step 2 arithmetic: 100M users x 20 GB used (assumed average) = 2 EB before dedupe
160 assert 100_000_000 * 20e9 / 1e18 == 2.0
161 assert 1e9 / 4e6 == 250
162
163 return {"fixed_reupload": fixed_reupload, "cdc_reupload": cdc_reupload, "cdc_chunks": len(c_new),
164 "conflict_copy": phone}
165
166
167if __name__ == "__main__":
168 print("All checks pass.", run_checks())