Persistent key-value store with binary snapshots

The store itself is a dictionary. The real question is the byte format: how do
you turn a dict[str, str] into one blob and get exactly the same dictionary
back, when keys and values may contain any character?

The tempting first try is a delimiter format, such as key=value\n per
entry. It breaks as soon as a key holds = or a value holds \n. Escaping
fixes that but adds a second set of edge cases (escaping the escape
character), and many interviewers also ban json and pickle for this
question, so say out loud why you are not using them.

The robust method is length-prefixed encoding. Write each string as its
UTF-8 byte length in a fixed-width integer, then the bytes themselves. The
reader never searches for a separator: it reads 4 bytes, learns n, takes
exactly n bytes, and moves on. Any byte sequence round-trips, including the
empty string and multi-byte characters.

Blob layout:

magic "KVS1" | entry count (u32) | [key_len u32 | key | val_len u32 | val] * count | crc32 (u32)

• The magic and version let you reject a blob that is not ours and change the
  format later.
• The entry count lets the reader detect truncation instead of silently
  returning half a store.
• The CRC32 over everything before it catches corruption.

Invariants:

• Lengths are byte lengths after UTF-8 encoding, never len(str).
  "é" is one character but two bytes.
• load either replaces the whole state or raises and leaves the current
  state alone. It parses into a fresh dict and swaps it in at the end.
• An empty medium (nothing saved yet) loads as an empty store.

import struct
import zlib
from typing import Optional

MAGIC = b"KVS1"
U32 = struct.Struct(">I")


class Medium:
    """Stores one binary blob. Given by the interviewer."""

    def save(self, blob: bytes) -> None:
        raise NotImplementedError

    def load(self) -> Optional[bytes]:
        raise NotImplementedError


class InMemoryMedium(Medium):
    def __init__(self) -> None:
        self._blob: Optional[bytes] = None

    def save(self, blob: bytes) -> None:
        self._blob = bytes(blob)

    def load(self) -> Optional[bytes]:
        return self._blob


class CorruptBlobError(ValueError):
    pass


class PersistentKVStore:
    def __init__(self, medium: Medium) -> None:
        self._medium = medium
        self._data: dict[str, str] = {}

    def get(self, key: str) -> Optional[str]:
        return self._data.get(key)

    def put(self, key: str, value: str) -> None:
        self._data[key] = value

    def delete(self, key: str) -> bool:
        return self._data.pop(key, None) is not None

    def save(self) -> None:
        self._medium.save(self.serialize(self._data))

    def restore(self) -> None:
        blob = self._medium.load()
        self._data = {} if not blob else self.deserialize(blob)

    @staticmethod
    def serialize(data: dict[str, str]) -> bytes:
        parts = [MAGIC, U32.pack(len(data))]
        for key, value in data.items():
            for s in (key, value):
                raw = s.encode("utf-8")
                parts.append(U32.pack(len(raw)))
                parts.append(raw)
        body = b"".join(parts)
        return body + U32.pack(zlib.crc32(body))

    @staticmethod
    def deserialize(blob: bytes) -> dict[str, str]:
        if len(blob) < len(MAGIC) + 2 * U32.size:
            raise CorruptBlobError("blob too short")
        body, (crc,) = blob[:-U32.size], U32.unpack(blob[-U32.size:])
        if zlib.crc32(body) != crc:
            raise CorruptBlobError("checksum mismatch")
        if body[: len(MAGIC)] != MAGIC:
            raise CorruptBlobError("unknown format")
        pos = len(MAGIC)

        def read_u32() -> int:
            nonlocal pos
            if pos + U32.size > len(body):
                raise CorruptBlobError("truncated length")
            (n,) = U32.unpack_from(body, pos)
            pos += U32.size
            return n

        def read_str() -> str:
            nonlocal pos
            n = read_u32()
            if pos + n > len(body):
                raise CorruptBlobError("truncated string")
            s = body[pos : pos + n].decode("utf-8")
            pos += n
            return s

        count = read_u32()
        out: dict[str, str] = {}
        for _ in range(count):
            key = read_str()
            out[key] = read_str()
        if pos != len(body):
            raise CorruptBlobError("trailing bytes")
        return out

Let n be the number of keys and B the total UTF-8 bytes of all keys and
values.

• get, put, delete: O(1) average time (hash map), plus O(len(key)) to
  hash the key.
• save / serialize: O(n + B) time; O(B) extra space for the blob.
• restore / deserialize: O(n + B) time; O(B) space for the new dict. The
  old dict is replaced only after parsing succeeds, so peak memory is about
  twice the store.

• Empty store: save() then restore() gives {}; the blob is 12 bytes
  (magic, count 0, CRC).
• Nothing ever saved: restore() on a fresh InMemoryMedium gives an empty
  store, not an error.
• Basic: put("a", "1"), put("b", "2"), save(), then a new store on the
  same medium calls restore(); get("a") == "1", get("b") == "2".
• Delimiters inside data: key "x=y\n|z", value "line1\nline2:\x00" round-trip
  exactly.
• Empty strings: key "" with value "" round-trips and is distinct from a
  missing key.
• Multi-byte text: key "café", value "日本語🙂" round-trip; the stored key
  length is 5 bytes, not 4.
• Overwrite and delete before saving: put("k","v1"), put("k","v2"),
  delete("gone") returns False; after round-trip only {"k": "v2"}.
• Save is a snapshot: save(), then put("late","x") without saving, then
  restore(); get("late") is None.
• Corruption: flip one byte in the blob, restore() raises
  CorruptBlobError and the store keeps its previous contents.
• Truncation: drop the last 10 bytes, restore() raises.
• Large: 100,000 keys of random printable and non-ASCII text round-trip.

1. The medium has a maximum blob size (for example 1 KB), so one store
   needs many blobs. The medium becomes save(name, blob) /
   load(name). Serialize to one byte stream as above, then cut it into
   fixed-size chunks named chunk_0..chunk_k, and write a small manifest
   blob (chunk count, total length, CRC) last. The reader trusts only
   the manifest, so a crash halfway through a save leaves the old manifest
   pointing at a consistent set. To make that true, write new chunks under a
   new generation prefix (gen7/chunk_0), then flip the manifest, then
   delete the old generation.
2. Saving the whole store after every write is too slow. Add an
   append-only log: each put/delete appends a record
   (op | key_len | key | val_len | val | crc) to the medium. restore
   loads the last snapshot and replays the log. When the log grows past a
   threshold (for example, larger than the snapshot), compact: write a new
   snapshot and truncate the log. A torn last record is detected by its CRC
   and dropped.
3. Thread safety. Guard the dict with a lock. For save, copy the dict
   under the lock (or use a read-write lock) and serialize outside it so
   writers are not blocked for the whole encode. With the log variant,
   append under the same lock that mutates the dict so log order matches
   apply order.
4. Values that are not strings (ints, bytes, nested lists). Add a one-byte
   type tag before each value (0 = str, 1 = bytes, 2 = int64,
   3 = list followed by an element count) and recurse. Bump the magic to
   KVS2 and keep a reader for KVS1.
5. Memory larger than RAM / faster restore. Move toward an SSTable: write
   entries sorted by key with an index block of (key, offset) at the end, so
   a reader can binary-search the index and fetch one value without loading
   the whole store.

Mid-level (L4)

• Working get/put/delete/save/restore with a correct round trip.
• Spots that a delimiter format breaks on keys containing the delimiter and
  moves to length prefixes, ideally without a hint.
• Encodes lengths in bytes, handles the empty-store and never-saved cases,
  and tests them.

Senior (L5)

• Designs the format deliberately: magic/version, count, checksum, and
  explains what each one protects against.
• restore is all-or-nothing and leaves the store intact on a bad blob.
• Handles the chunked-medium follow-up cleanly, including what happens if
  the process dies between chunk writes.
• Discusses snapshot cost vs. a write-ahead log and when to compact.

Staff (L6+)

• Frames durability guarantees precisely (what is lost on crash for snapshot
  vs. log, fsync and atomic-rename semantics on a real filesystem).
• Plans format evolution and backward compatibility from the start.
• Covers concurrency (snapshotting without stalling writers) and scaling past
  memory (SSTable/LSM layout, compaction policy), and knows when that
  complexity is not worth it.