A checksum that proves identity

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""A checksum that proves identity, not just integrity.

A toy block store, two designs: (a) each block stores crc(content) next to
its content; (b) a separate index stores, per block number, a checksum over
(block number, content). Then a misdirected write (block 7's bytes and
checksum land in block 3's place) and a lost write (a write that never
happened): which design notices?
"""
import argparse, json, zlib

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--blocks", type=int, default=16, help="Blocks in the store.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
crc = lambda b: zlib.crc32(b)

class InlineStore:                       # (a) checksum next to the data
    def __init__(self, n): self.disk = {i: (f"block {i} v1".encode(), crc(f"block {i} v1".encode())) for i in range(n)}
    def write(self, i, data, lose=False):
        if not lose: self.disk[i] = (data, crc(data))
    def read(self, i):
        data, c = self.disk[i]
        return data, crc(data) == c

class IndexedStore:                      # (b) checksum of (block number, content) kept elsewhere
    def __init__(self, n):
        self.disk = {i: f"block {i} v1".encode() for i in range(n)}
        self.index = {i: crc(i.to_bytes(8, "big") + self.disk[i]) for i in range(n)}
    def write(self, i, data, lose=False):
        if not lose: self.disk[i] = data
        self.index[i] = crc(i.to_bytes(8, "big") + data)        # the index is updated even if the data write is lost
    def read(self, i):
        data = self.disk[i]
        return data, crc(i.to_bytes(8, "big") + data) == self.index[i]

rows = []
for name, cls in (("(a) checksum stored with the block", InlineStore), ("(b) checksum of (block no., content) in an index", IndexedStore)):
    s = cls(args.blocks)
    s.disk[3] = s.disk[7]                                        # misdirected write: block 7's bytes (and inline checksum) land at 3
    data, ok = s.read(3)
    mis = "returned block 7's data as valid" if ok else "reported a mismatch"
    s = cls(args.blocks)
    s.write(5, b"block 5 v2", lose=True)                         # lost write: the new version never reaches the disk
    data, ok = s.read(5)
    lost = "returned the stale version as valid" if ok else "reported a mismatch"
    rows.append({"design": name, "misdirected_write": mis, "lost_write": lost})
    print(f"{name:<50} misdirected write: {mis:<34} lost write: {lost}")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"designs": rows}}, f)