Track A — Warmup: The Ten Patterns, From Zero

Self-contained. You should be able to read this file and nothing else, and afterwards be able to implement every pattern the loop reportedly asks, explain why each data structure was chosen, state its complexity, and answer the follow-ups.

Every implementation here is complete and runnable — no ... placeholders, no "left as an exercise". Read the code; it is the material, not an illustration of it.

Table of Contents

• Chapter 0: What Makes These Problems Different
  • 0.1 The family they belong to
  • 0.2 Representation-first thinking
  • 0.3 The three questions that pick the structure
• Chapter 1: Predecessor Queries and Versioned State
  • 1.1 What a predecessor query is
  • 1.2 Why a hash map cannot answer it
  • 1.3 Binary search, derived and implemented
  • 1.4 Tombstones: delete is a write
  • 1.5 Global versus per-key versions
  • 1.6 Snapshots and reachability GC
  • 1.7 MVCC, snapshot isolation, and write skew
  • 1.8 Complete implementation
  • 1.9 Interview Q&A
• Chapter 2: Delta Logs — Undo, Redo, Checkpoint
  • 2.1 Snapshot versus delta, from first principles
  • 2.2 The three-integer checkpoint
  • 2.3 Redo, and why a new edit destroys it
  • 2.4 Streaming diff: the online constraint
  • 2.5 Complete implementation
  • 2.6 Interview Q&A
• Chapter 3: The Intrusive List — LRU and Friends
  • 3.1 Why O(1) eviction needs two structures
  • 3.2 Sentinels, and why they delete every edge case
  • 3.3 TTL: lazy, sampled, and active expiry
  • 3.4 Size-aware eviction
  • 3.5 Stampede control: single-flight
  • 3.6 Complete implementation
  • 3.7 Interview Q&A
• Chapter 4: Rate Limiting — Four Algorithms and Their Lies
  • 4.1 Fixed window, and the boundary burst
  • 4.2 Sliding window log
  • 4.3 Sliding window counter
  • 4.4 Token bucket, derived
  • 4.5 Leaky bucket, and how it differs
  • 4.6 Distributed limiting
  • 4.7 Complete implementation
  • 4.8 Interview Q&A
• Chapter 5: Heaps and Deterministic Scheduling
  • 5.1 What a binary heap actually is
  • 5.2 The tie-breaking bug that flakes your tests
  • 5.3 Cancellation: the lazy-deletion trick
  • 5.4 Fixed-rate versus fixed-delay
  • 5.5 Backoff with jitter, and why it matters
  • 5.6 Complete implementation
  • 5.7 Interview Q&A
• Chapter 6: Streaming State Machines and Chunk Boundaries
  • 6.1 The only hard part: the boundary
  • 6.2 Why a regex cannot do this
  • 6.3 Bounded buffers and the pathological input
  • 6.4 Complete implementation
  • 6.5 Interview Q&A
• Chapter 7: Dependency Graphs, Topological Order, Cycles
  • 7.1 Two algorithms and when each wins
  • 7.2 The three-colour DFS
  • 7.3 Incremental recomputation
  • 7.4 Complete implementation
  • 7.5 Interview Q&A
• Chapter 8: Write-Ahead Logs and Crash Recovery
  • 8.1 What durability actually means
  • 8.2 Record framing and torn writes
  • 8.3 Checkpointing and compaction
  • 8.4 Complete implementation
  • 8.5 Interview Q&A
• Chapter 9: Deduplication and Probabilistic Structures
  • 9.1 The exactly-once illusion
  • 9.2 Bloom filters, derived
  • 9.3 Which direction the error points
  • 9.4 Complete implementation
  • 9.5 Interview Q&A
• Chapter 10: Backpressure and Bounded Concurrency
  • 10.1 The unbounded queue is a memory leak
  • 10.2 Four responses to too much work
  • 10.3 Graceful shutdown
  • 10.4 Complete implementation
  • 10.5 Interview Q&A
• The Complexity Table
• The Thirty Questions
• References

Chapter 0: What Makes These Problems Different

0.1 The family they belong to

A conventional algorithm interview asks you to compute a function: given this input, produce this output. Two-sum, reverse a linked list, longest palindromic substring. The difficulty is finding the trick, and once you have it the code is short.

These loops reportedly ask something else: build a stateful component. A store, a cache, a limiter, a scheduler, a differ, a parser. The difficulty is not finding a trick — there usually isn't one — it is choosing a representation for the state such that all the operations you are asked for are cheap, and such that the operations you will be asked for in eight minutes are also cheap.

Three practical consequences:

1. You write much more code. Reported sources agree on this: substantially more than a typical FAANG interview. A four-gate stateful problem is 120–200 lines. If you type at 30 lines of correct code per 10 minutes, you have a throughput problem before you have an algorithms problem.
2. Edge cases are the content, not the garnish. "What does delete on a key that never existed do?" is not a nitpick; it is the question that reveals whether you understand what a version number is.
3. The follow-up is guaranteed. Every one of these has an obvious next requirement (make it concurrent, make it durable, bound the memory) and you will be asked for it.

0.2 Representation-first thinking

Here is the single habit that separates candidates who clear four gates from candidates who clear two.

Before writing any code, ask: what is the shape of the queries? Not "what are the operations named" — what shape are they.

The mistake that costs gates is hearing "key-value store" and reaching for a dict, because the words said key-value. The words say what it is called. The query shape says what it must be built from. "The value as of version V" is a predecessor query wearing a key-value store's clothes.

0.3 The three questions that pick the structure

Ask these out loud in the first ninety seconds of any of these problems. Each one has, historically, been the question that determined whether gate 3 was additive or a rewrite.

Q1 — "Is the input replayable / immutable?" If yes, you can store positions into it rather than copies of it. That is the difference between an O(1) checkpoint and an O(n) one.

Q2 — "Is this identifier global or per-entity?" Global identifiers make a snapshot a single integer. Per-entity identifiers force a vector, and every downstream operation gets harder. (Chapter 1.5.)

Q3 — "Will I ever need to undo this?" If there is any chance, store deltas rather than states. Deltas compose; snapshots do not. (Chapter 2.1.)

Chapter 1: Predecessor Queries and Versioned State

This is the reported technical-screen coding question — a versioned key-value store — and it is the single most corroborated problem across independent sources. If you master one chapter here, make it this one.

1.1 What a predecessor query is

Start from nothing. Suppose you have written values at various moments:

version 1 : key "a" = 10
version 4 : key "a" = 20
version 9 : key "a" = 30

Now someone asks: "what was a at version 6?"

There is no write at version 6. The answer is 20 — the value written by the largest version that is less than or equal to 6. That operation has a name: a predecessor query (also called floor, or the last entry at or before X).

This is not a lookup. A lookup asks "what is stored at exactly this key?" and the answer is either a value or nothing. A predecessor query asks "what is stored at the closest key at or below this one?" and it needs the keys to be ordered to answer.

Say the words "that's a predecessor query" out loud in the interview. It is the sentence that selects the data structure, and interviewers notice when a candidate names the operation rather than describing it.

1.2 Why a hash map cannot answer it

The tempting first design is:

data = {"a": {1: 10, 4: 20, 9: 30}}    # key -> {version: value}

Read at version 9? data["a"][9] → 30. Works. Read at version 4? Works. Read at version 6? data["a"][6] → KeyError.

To answer it you would have to scan every key of the inner dict and take the maximum that is ≤ 6. That is O(number of writes to this key), on every read, forever. And it is not a sub-optimal implementation of the right idea — it is the wrong idea, because a hash map destroys order by construction. Hashing maps 4 and 6 to unrelated buckets; there is no "next lower key" to walk to.

The general rule, worth internalizing far beyond this problem:

A hash map answers "exactly", never "nearest". The moment a requirement contains the words before, after, as of, range, nearest, or at most, you need an ordered structure.

So: per key, keep an append-only list of (version, value) pairs, and because versions only ever increase, that list is already sorted with no sorting work. Appends are O(1) and the ordering is free — a very pleasant property to point out.

1.3 Binary search, derived and implemented

You now need "the last entry with version ≤ target" in a sorted list. Linear scan is O(n). Binary search is O(log n). Python has bisect, and you should use it — but you must be able to write it, because "implement bisect" is a plausible follow-up and because getting the boundary condition right requires understanding the invariant.

The derivation. Maintain two indices, lo and hi, with the invariant:

• every entry at index < lo has version ≤ target
• every entry at index ≥ hi has version > target
• the answer is somewhere in [lo, hi)

Start with lo = 0, hi = len(entries) — vacuously true, since there are no indices below 0 and none at or above len. Each step halves the range while preserving the invariant. When lo == hi the range is empty, and by the invariant everything below lo is ≤ target and everything at or above is > target. So lo is the count of entries ≤ target, and lo - 1 is the index of the last one — or -1, meaning none exist.

def bisect_right_on_version(entries, target):
    """Index one past the last entry whose version is <= target.

    entries: list of (version, value), strictly increasing in version.
    """
    lo, hi = 0, len(entries)
    while lo < hi:
        mid = (lo + hi) // 2          # floor division: mid is always < hi
        if entries[mid][0] <= target:
            lo = mid + 1              # entries[mid] is <= target, so it belongs left of lo
        else:
            hi = mid                  # entries[mid] is > target, so hi can come down to it
    return lo


def value_at(entries, target):
    index = bisect_right_on_version(entries, target)
    return entries[index - 1] if index else None

Why lo = mid + 1 and not lo = mid. Because entries[mid] <= target means mid itself satisfies "≤ target", so it belongs in the already decided left region. If you wrote lo = mid the loop would not shrink when hi == lo + 1 and you would spin forever. This is the classic infinite-loop bug in hand-written binary search, and it is worth being able to explain rather than just avoid.

Why hi = mid and not hi = mid - 1. Because hi is exclusive. entries[mid] > target means mid is the first index we know is too big, so the answer range ends just before it — which, with an exclusive bound, is exactly hi = mid.

In production, use the standard library — bisect.bisect_right(entries, target, key=lambda e: e[0]) (the key parameter arrived in Python 3.10). Saying "I'd use bisect here, and here's what it does under the hood" is strictly better than either using it silently or reimplementing it unprompted.

1.4 Tombstones: delete is a write

Now: how do you delete a key from a store whose entire purpose is remembering the past?

Not by removing it. If delete("a") erased the entry list, then get("a", version=1) would return None — and it should return 10, because at version 1 the key genuinely had the value 10. Deleting the history destroys the product.

So a delete is a write of a special value — a tombstone:

version 1 : "a" = 10
version 4 : "a" = 20
version 7 : "a" = TOMBSTONE      <- delete happened here

• get("a") → predecessor of "now" is the tombstone → return None
• get("a", version=5) → predecessor of 5 is (4, 20) → return 20
• get("a", version=9) → still the tombstone → None
• put("a", 99) at version 12 → the key is alive again, and all of the above still holds

Use a sentinel singleton for the tombstone, not None:

class _Deleted:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    def __repr__(self):
        return "DELETED"

DELETED = _Deleted()

Why a singleton rather than None? Because None is a legitimate value a user might store. put("a", None) and delete("a") must be distinguishable, or your store silently corrupts data for any caller who stores nulls. Using a private sentinel and comparing with is makes that distinction airtight. This is a real production concern (it is why Cassandra, HBase, and every LSM engine have explicit tombstone records) and mentioning it unprompted reads well.

The subtle one: deleting a key that never existed. Does that consume a version?

Yes. And the reason is worth having ready, because it is a favourite follow-up:

Versions describe the log, not the data. Version N means "the state after the Nth operation." If some operations silently don't get a version, then two clients performing the same sequence of operations end up with different version numbers for the same logical state, and "read as of version N" stops being a well-defined question. The counter is a property of the operation stream, so every operation advances it.

1.5 Global versus per-key versions

A design decision you must make at gate 1 that pays off — or costs you — at gate 3.

Per-key versions: each key has its own counter. a is at version 3, b is at version 1.

Global versions: one counter for the whole store. Every write anywhere takes the next number.

Global wins, decisively, and here is why:

The moment gate 3 asks for a snapshot, per-key versioning turns a one-integer object into a map that grows with the keyspace, and gate 4's transactions become vector-clock comparisons. Choosing global at gate 1 is the decision that makes gates 3 and 4 additive rather than a rewrite.

The cost, which you should name because it is real: a global counter is a serialization point. Every write, on every key, must agree on the next number. On one machine that is a lock or an atomic increment and it is fine to tens of millions of ops/sec. Distributed, it becomes a consensus problem — a Raft-replicated counter, or a timestamp oracle like Percolator's, or hybrid logical clocks if you will accept bounded staleness. Saying that sentence is the bridge to the distributed-design round.

1.6 Snapshots and reachability GC

A snapshot is a stable read view: it pins a version, and every read through it sees the store exactly as it was at that moment, no matter what else commits meanwhile.

With global versions this is almost embarrassingly simple:

class Snapshot:
    def __init__(self, store, version):
        self._store, self.version = store, version
    def get(self, key):
        return self._store.get(key, version=self.version)

That is the whole thing. The snapshot is an integer. Readers never block writers and writers never block readers, because a write only ever appends — it never mutates an entry a reader might be looking at. This is the central benefit of multi-version storage and it is worth saying explicitly.

Compaction is the other side of the bargain. Append-only means unbounded growth, so you need to reclaim versions that nobody can ever see again. Which ones are those?

Think of it as reachability. Define the set of pins — every version some reader could still land on:

pins = {current_version} ∪ {version of each live snapshot}

For each key, for each pin, exactly one entry is visible: the predecessor of that pin. Every entry that is not the predecessor of any pin is unreachable — no query can ever return it — and can be dropped.

def compact(self):
    pins = {self._version} | {s.version for s in self._live_snapshots}
    dropped = 0
    for key, entries in list(self._data.items()):
        if len(entries) <= 1:
            continue
        keep = set()
        for pin in pins:
            index = bisect_right_on_version(entries, pin)
            if index:
                keep.add(index - 1)
        if len(keep) < len(entries):
            dropped += len(entries) - len(keep)
            self._data[key] = [entries[i] for i in sorted(keep)]
    return dropped

This is precisely the same reasoning a garbage collector uses — reachability from a root set — and precisely the same reasoning Postgres's VACUUM uses to decide which dead row versions can go (its root set is the oldest running transaction's xmin). Making that connection out loud is a strong signal.

The failure mode to name: a long-running snapshot pins old versions and blocks all reclamation behind it. In Postgres this is the notorious "long transaction prevents vacuum" problem that leads to table bloat. Your API should therefore have a way to expire abandoned snapshots, and you should say so.

1.7 MVCC, snapshot isolation, and write skew

You have now, without naming it, built MVCC — Multi-Version Concurrency Control. Keeping multiple versions of each row, giving each reader a consistent view, and never letting readers block writers. It is how Postgres, MySQL/InnoDB, Oracle, and essentially every serious transactional database works.

Gate 4 adds optimistic concurrency control on top:

1. A transaction records its start version.
2. All its reads are at that version — so it sees a consistent snapshot, and repeats of the same read return the same answer no matter what commits meanwhile.
3. It buffers its writes locally; nobody else can see them.
4. At commit, it validates: was any key that I read written by somebody else since my start version? If yes → abort with a conflict. If no → apply all my writes atomically at one new version.

"Atomically at one new version" matters: if a transaction's writes got separate versions, a reader could land between them and observe half a transaction. One version per commit makes partial observation impossible by construction.

The isolation level you get from this is snapshot isolation. It prevents dirty reads, non-repeatable reads, and lost updates. It does not prevent one thing, and this is the follow-up you must be ready for:

Write skew. Two transactions read overlapping data, write disjoint keys, and both commit — jointly violating an invariant that neither violated alone.

The canonical example. A hospital requires at least one doctor on call. Alice and Bob are both on call. Both simultaneously request to go off call.

T1: read on_call_count -> 2. "2 > 1, safe." write alice.on_call = false
T2: read on_call_count -> 2. "2 > 1, safe." write bob.on_call   = false

T1 read Bob's row and wrote Alice's. T2 read Alice's row and wrote Bob's. Neither one's read set was written by the other — T1 wrote alice, which T2 only read... wait, T2 did read alice. Under strict read-set validation as implemented here, one of them would abort. But real snapshot isolation as shipped in Postgres's REPEATABLE READ and Oracle's SERIALIZABLE validates only write-write conflicts, not read-write ones — and under that rule both commit, and the hospital has zero doctors on call.

So there are two honest things to say:

1. What snapshot isolation permits in general is write skew, and the mechanism is that a transaction's reads are not protected against concurrent writes to those rows.
2. What your specific implementation does — if you validate the full read set (as the reference implementation here does), you are stricter than textbook SI and closer to serializable, at the cost of more aborts on read-heavy transactions.

The fixes, in ascending order of cost: promote the read to a write (SELECT ... FOR UPDATE), materialize the conflict (write to a shared row so the write-write check catches it), or use Serializable Snapshot Isolation (SSI), which tracks read-write dependencies and aborts transactions that form a dangerous structure. Postgres's SERIALIZABLE is SSI.

Naming write skew unprompted is one of the highest-value single moves available in this problem.

1.8 Complete implementation

from bisect import bisect_right


class _Deleted:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    def __repr__(self):
        return "DELETED"


DELETED = _Deleted()


class ConflictError(Exception):
    """A transaction's read set was written under it."""


class Snapshot:
    __slots__ = ("_store", "version", "_released")

    def __init__(self, store, version):
        self._store, self.version, self._released = store, version, False

    def get(self, key):
        if self._released:
            raise RuntimeError("snapshot released")
        return self._store.get(key, version=self.version)

    def release(self):
        if not self._released:
            self._released = True
            self._store._drop_snapshot(self)

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.release()


class Transaction:
    __slots__ = ("_store", "_read_version", "_reads", "_writes", "_done")

    def __init__(self, store):
        self._store = store
        self._read_version = store.version
        self._reads = set()
        self._writes = {}
        self._done = False

    def get(self, key):
        self._reads.add(key)
        if key in self._writes:                      # read-your-own-writes
            value = self._writes[key]
            return None if value is DELETED else value
        return self._store.get(key, version=self._read_version)

    def put(self, key, value):
        self._writes[key] = value

    def delete(self, key):
        self._writes[key] = DELETED

    def commit(self):
        if self._done:
            raise RuntimeError("transaction finished")
        for key in self._reads:                      # validate BEFORE applying anything
            if self._store._last_write_version(key) > self._read_version:
                self._done = True
                raise ConflictError(f"{key!r} was written after this txn started")
        self._done = True
        if not self._writes:
            return self._store.version
        return self._store._apply_batch(self._writes)


class VersionedKV:
    """MVCC key-value store with point-in-time reads."""

    __slots__ = ("_data", "_version", "_snapshots")

    def __init__(self):
        self._data = {}          # key -> [(version, value), ...] ascending
        self._version = 0
        self._snapshots = []

    @property
    def version(self):
        return self._version

    # ---- writes ----------------------------------------------------------
    def put(self, key, value):
        self._version += 1
        self._data.setdefault(key, []).append((self._version, value))
        return self._version

    def delete(self, key):
        # A tombstone, not a removal — and it consumes a version even for an
        # absent key, because versions describe the log, not the data.
        self._version += 1
        self._data.setdefault(key, []).append((self._version, DELETED))
        return self._version

    # ---- reads -----------------------------------------------------------
    def _entry_at(self, key, version):
        entries = self._data.get(key)
        if not entries:
            return None
        index = bisect_right(entries, version, key=lambda e: e[0])
        return entries[index - 1] if index else None

    def get(self, key, version=None):
        entry = self._entry_at(key, self._version if version is None else version)
        if entry is None or entry[1] is DELETED:
            return None
        return entry[1]

    def history(self, key):
        return list(self._data.get(key, ()))

    def keys(self, version=None):
        at = self._version if version is None else version
        return sorted(k for k in self._data
                      if (e := self._entry_at(k, at)) and e[1] is not DELETED)

    # ---- snapshots -------------------------------------------------------
    def snapshot(self):
        snap = Snapshot(self, self._version)
        self._snapshots.append(snap)
        return snap

    def _drop_snapshot(self, snap):
        try:
            self._snapshots.remove(snap)
        except ValueError:
            pass

    def compact(self):
        pins = {self._version} | {s.version for s in self._snapshots}
        dropped = 0
        for key, entries in list(self._data.items()):
            if len(entries) <= 1:
                continue
            keep = set()
            for pin in pins:
                index = bisect_right(entries, pin, key=lambda e: e[0])
                if index:
                    keep.add(index - 1)
            if len(keep) < len(entries):
                dropped += len(entries) - len(keep)
                self._data[key] = [entries[i] for i in sorted(keep)]
        return dropped

    # ---- transactions ----------------------------------------------------
    def begin(self):
        return Transaction(self)

    def _last_write_version(self, key):
        entries = self._data.get(key)
        return entries[-1][0] if entries else 0

    def _apply_batch(self, writes):
        self._version += 1                       # ONE version for the whole txn
        for key, value in writes.items():
            self._data.setdefault(key, []).append((self._version, value))
        return self._version

Complexities, which you should state unprompted:

1.9 Interview Q&A

Q: What's the complexity of a read at a version? O(log w) in the number of writes to that key — not to the store. Binary search over that key's version list. If I had scanned linearly it would be O(w), which on a hot key with a million writes is 20× worse per read and gets worse over time rather than better.

Q: Why not a dict of dicts? Because "as of version V" is a predecessor query, and a hash map destroys order. data["a"][6] raises KeyError when no write happened at exactly 6, and recovering from that requires scanning every version of that key. The requirement contains the words "as of", which is the tell that you need an ordered structure.

Q: Why does deleting a non-existent key consume a version? Versions describe the log, not the data. Version N means "the state after N operations". If some operations don't get a number, two clients performing the same operation sequence disagree about what version N means, and "read as of N" stops being well-defined.

Q: Memory grows forever. What do you do? Compaction, using reachability: the pins are the current version plus every live snapshot's version, and any entry that isn't the predecessor of some pin is unreachable and can be dropped. Beyond that you need a retention policy — time-based, count-based, or pinned-by-reader — because compaction alone can't help if readers keep old snapshots alive. That's the same failure mode as a long transaction blocking VACUUM in Postgres and causing table bloat.

Q: Two transactions both read A and both write B. Both commit. Is that OK? That's the shape of write skew — the anomaly snapshot isolation permits. Textbook SI validates write-write conflicts only, so two transactions that read overlapping data and write disjoint keys both commit, and can jointly break an invariant neither broke alone. The classic example is the on-call doctors constraint. My implementation validates the full read set, which is stricter than textbook SI — it costs extra aborts on read-heavy transactions but it catches this case. If I wanted textbook behaviour plus safety I'd use SSI, which is what Postgres's SERIALIZABLE does.

Q: How would you make keys() fast? As written it's O(K log w) — a predecessor query per key — which is the weakest operation in the design. Two options. Maintain a per-version delta of key additions and removals, so keys(v) replays deltas from a nearby checkpoint; that trades write cost for read cost. Or keep a separate ordered structure — a skip list or B-tree keyed by key name, with version chains hanging off each entry — which makes it a range scan with a predecessor query per live key, and gives you prefix queries for free.

Q: Make it durable. Put a write-ahead log in front: append (version, key, value) to a log with a length prefix and a checksum, fsync according to your durability policy, then apply to the in-memory structure. On restart, replay the log to rebuild. The in-memory structure becomes a reconstructible cache rather than the system of record. Then you need checkpointing so replay doesn't take longer every restart — Chapter 8.

Q: Make it concurrent. The good news is inherent to MVCC: readers never block, because writes only append and never mutate an entry a reader might be reading. The version counter is the contention point — an atomic increment, or a lock held only for the increment. The bad news is that the counter is now your throughput ceiling, and sharding it costs you the global ordering the entire design rests on. Under free-threaded Python you'd also need the append itself to be atomic with respect to readers, which means either a lock per key or an immutable-tuple swap.

Q: What if two writes land on the same version? They can't, by construction — the counter increments before each write. But a transaction's writes deliberately share one version, and that's what makes a transaction atomic to readers: there's no version at which half of it is visible.

Q: How would you support "give me all changes between version A and B"? A per-key scan is O(K log w). Better: keep a secondary append-only log of (version, key) and binary-search it for the range — the versions are increasing, so it's sorted for free. That's a change-data-capture feed, and it's how you'd drive replication or invalidate a downstream cache.

Q: What breaks first at 100× the data? keys() and compact(), both of which are O(total keys). Compaction becomes a stop-the-world pass, so you'd make it incremental — compact a bounded number of keys per call, remembering a cursor. That's exactly what an LSM background compactor does.

Chapter 2: Delta Logs — Undo, Redo, Checkpoint

This is the reported onsite coding question: a token-level streaming differ that tracks state changes with rollback.

2.1 Snapshot versus delta, from first principles

Any system that must "go back" has two options.

Snapshot: periodically save a full copy of the state. Going back means restoring a copy.

• Restore is O(size of state) and trivially correct.
• Storage is O(number of save points × size of state).
• The granularity is fixed at save time. If you saved every 100 operations and someone asks to undo 1 operation, you cannot.

Delta (a log of changes): record what each operation changed. Going back means applying the inverse of each change in reverse.

• Undo is O(size of the change), not O(size of the state).
• Storage is O(total changes), which for small changes is enormously less.
• Granularity is per operation — you can undo exactly one.
• It requires that each change be invertible, which is a real design constraint.

Here is the property that decides it for interview problems and for real editors, databases, and version-control systems alike:

Deltas compose; snapshots do not.

Any position in the delta log is a valid restore point — for free, without having planned for it. So checkpoints become "remember the log length", nesting is free, and undo granularity is whatever the operation granularity is. With snapshots you must decide the granularity in advance, and any requirement that arrives later at a finer granularity forces a rewrite.

That is exactly what a gated problem does to you. Gate 3 asks for named checkpoints (coarse). Gate 4 asks for per-operation undo (fine). A snapshot design passes gate 3 and dies at gate 4; a delta design passes both without change.

How to see it coming without seeing gate 4: ask Q3 from §0.3 — "will I ever need to undo this?" — while writing gate 1. For anything that accumulates state incrementally, the answer is almost always yes.

2.2 The three-integer checkpoint

Concretely, for a differ that consumes tokens one at a time and appends edit events:

history[i] = (cursor_before, n_events_emitted, token)

Three machine words per input token. From that:

A checkpoint is three integers because the log is the history — you do not copy anything, you just remember where you were in it. That is the whole trick, and it is why gate 3 collapses from "how do I snapshot this efficiently" to "remember three numbers".

Checkpoint invalidation. Rolling back to label L must drop every label created after L, because those labels point at log positions that no longer exist. If you leave them, a later rollback to one of them truncates to a length longer than the current log — a silent no-op that corrupts state. Keep labels in an ordered list and truncate it at the same time.

2.3 Redo, and why a new edit destroys it

Undo pushes the undone operations onto a redo stack. Redo pops them and re-applies.

But if you undo three operations and then perform a new operation, the redo stack must be cleared. Why? Because the redo entries describe operations that were applied to a state that no longer exists. Re-applying them would produce nonsense — you would be replaying a branch of history that was abandoned.

This is exactly the linear undo model in every text editor: undo, undo, type a character, and your redo is gone. It is not a limitation, it is the only coherent semantics without a full history tree (which is what Vim's :undolist and Emacs's undo-tree implement instead).

The implementation detail that makes it correct: feed() clears the redo stack, and redo() must not — it applies through a shared internal _apply() that does not clear. Getting this backwards is a common bug and there is a test for it.

2.4 Streaming diff: the online constraint

Now the algorithmic part. You are diffing a stream of tokens against a known baseline, but you receive the stream one token at a time and must emit edits as you go. You never see the whole input.

That rules out real diff algorithms. Myers' algorithm — the one git diff uses — is O(ND) where N is the input size and D the edit distance, and it needs the entire input because it searches for the shortest edit script through a full edit graph. You cannot search a graph whose right-hand side has not arrived yet.

So you use a greedy online algorithm with bounded lookahead:

• Keep a cursor into the baseline.
• Token matches baseline[cursor] → emit keep, advance cursor.
• Otherwise, look ahead up to w positions for a match at baseline[j], taking the smallest such j. If found, the baseline tokens in [cursor, j) were skipped → emit delete for each, then keep, and set cursor to j+1.
• No match within the window → emit insert, cursor unchanged.

The window is what makes this O(w) per token — O(1) amortized for constant w — instead of requiring random access to the whole baseline. It buys you the ability to be online.

The cost, which you must state: a skip longer than the window is misreported as an insert plus, eventually, trailing deletes. That is a stated, bounded inaccuracy, not a bug. Saying "here's the inaccuracy I'm accepting and here's the parameter that controls it" is a much stronger answer than pretending the algorithm is exact.

Why the smallest j. Suppose baseline is ["a", "x", "b", "y", "b"] and the stream sends "b" while the cursor is at index 1. Both index 2 and index 4 hold "b". Taking index 4 would emit three deletions and set the cursor past almost everything — a locally plausible but globally terrible alignment. Greedy nearest-match keeps the edit script minimal under the online constraint. Scan forward and return the first hit; do not scan the whole window.

2.5 Complete implementation

from collections import deque


class StreamDiffer:
    """Incremental diff of a token stream against a known baseline."""

    def __init__(self, baseline, lookahead=8):
        self.baseline = tuple(baseline)
        if lookahead < 0:
            raise ValueError("lookahead must be non-negative")
        self.lookahead = lookahead
        self._cursor = 0
        self._events = []                 # ("keep"|"insert"|"delete", token)
        self._history = []                # (cursor_before, n_events, token) per feed
        self._redo = []
        self._checkpoints = {}            # label -> (len(events), cursor, len(history))
        self._ckpt_order = []
        self._closed = False

    @property
    def events(self):
        return list(self._events)

    @property
    def cursor(self):
        return self._cursor

    # ---- the diff itself -------------------------------------------------
    def _match_within_window(self, token):
        """Smallest j in (cursor, cursor+lookahead] with baseline[j] == token."""
        if self.lookahead <= 0:
            return None
        end = min(self._cursor + self.lookahead, len(self.baseline) - 1)
        for j in range(self._cursor + 1, end + 1):
            if self.baseline[j] == token:
                return j
        return None

    def _apply(self, token):
        """Apply one token and record the DELTA. Shared by feed() and redo()."""
        cursor_before = self._cursor
        emitted = []

        if self._cursor < len(self.baseline) and self.baseline[self._cursor] == token:
            emitted.append(("keep", token))
            self._cursor += 1
        else:
            j = self._match_within_window(token)
            if j is None:
                emitted.append(("insert", token))
            else:
                for i in range(self._cursor, j):
                    emitted.append(("delete", self.baseline[i]))
                emitted.append(("keep", token))
                self._cursor = j + 1

        self._events.extend(emitted)
        self._history.append((cursor_before, len(emitted), token))
        return emitted

    def feed(self, token):
        if self._closed:
            raise RuntimeError("stream is closed")
        self._redo.clear()                # a new edit abandons the redo branch
        return self._apply(token)

    def close(self):
        if self._closed:
            return []
        trailing = [("delete", self.baseline[i])
                    for i in range(self._cursor, len(self.baseline))]
        self._events.extend(trailing)
        self._cursor = len(self.baseline)
        self._closed = True
        return trailing

    # ---- checkpoints -----------------------------------------------------
    def checkpoint(self, label):
        if self._closed:
            raise RuntimeError("stream is closed")
        if label in self._ckpt_order:
            self._ckpt_order.remove(label)
        self._checkpoints[label] = (len(self._events), self._cursor, len(self._history))
        self._ckpt_order.append(label)

    def labels(self):
        return list(self._ckpt_order)

    def rollback(self, label):
        if label not in self._checkpoints:
            raise KeyError(label)
        n_events, cursor, n_history = self._checkpoints[label]
        del self._events[n_events:]
        del self._history[n_history:]
        self._cursor = cursor
        self._redo.clear()
        pos = self._ckpt_order.index(label)          # later labels point at a
        for later in self._ckpt_order[pos + 1:]:     # history that no longer exists
            self._checkpoints.pop(later, None)
        del self._ckpt_order[pos + 1:]

    # ---- undo / redo -----------------------------------------------------
    def undo(self, n=1):
        if self._closed:
            raise RuntimeError("stream is closed")
        if n > len(self._history):                   # validate BEFORE mutating
            raise IndexError("cannot undo past the start")
        undone = []
        for _ in range(n):
            cursor_before, n_events, token = self._history.pop()
            if n_events:
                del self._events[len(self._events) - n_events:]
            self._cursor = cursor_before
            undone.append(token)
        self._redo.extend(undone)
        return undone

    def redo(self, n=1):
        if self._closed:
            raise RuntimeError("stream is closed")
        if n > len(self._redo):
            raise IndexError("nothing to redo")
        redone = []
        for _ in range(n):
            token = self._redo.pop()
            self._apply(token)                       # note: does NOT clear redo
            redone.append(token)
        return redone

2.6 Interview Q&A

Q: Why not just snapshot the state at each checkpoint? It works for coarse checkpoints and fails the moment you need per-operation undo, because the granularity is fixed when you save. Deltas give you every log position as a restore point for free, so checkpoints become "remember the log length" and undo becomes "pop one entry". The memory difference is the decisive part: snapshotting per input token is O(inputs × state), which for 20,000 tokens against a 3,000-token baseline is gigabytes.

Q: Why bounded lookahead instead of a real diff? Myers diff is O(ND) and needs the entire input, because it searches an edit graph whose right-hand side hasn't arrived. This is online — I must emit before I've seen the end. The window is the price of being online. The cost is that a skip longer than the window is misreported as an insert plus trailing deletes; that's bounded and parameterized, and I'd document it rather than hide it.

Q: What if the baseline is 10 GB? The cursor becomes a file offset and the lookahead window becomes a bounded read-ahead buffer — which is only possible because the window is bounded. An exact diff needs random access to the whole baseline; this needs a sliding view of w tokens. That's the property that makes this design work at scale and the exact one that makes Myers not.

Q: What's the memory ceiling at 100M tokens? Events dominate — one to a few tuples per input token. Two fixes: cap the event log and spill to disk, or, better, push events to a consumer instead of accumulating them. The second changes the API from "ask me for the log" to "I'll call you with each event," which is the right shape for a genuinely streaming system, and it makes the memory O(1) in stream length.

Q: How would you parallelize it? You wouldn't, along the stream — the cursor is inherently sequential state and each token's handling depends on the previous cursor. You parallelize across streams: one differ per document, sharded by document ID. If you truly had to split one stream you'd need synchronized anchor points in the baseline that both halves agree on, which is essentially what rsync's rolling checksum does.

Q: What breaks if the baseline changes mid-stream? Everything. Every history entry stores cursor_before, which is an index into the baseline, so every recorded delta becomes meaningless. You'd version the baseline and invalidate the differ on change — or, if you needed to support it, store the baseline content in the delta rather than the index, which costs memory but makes the log self-describing.

Q: Undo across a multi-event feed — how do you know how much to remove? That's exactly what n_events in the history tuple is for. One feed can emit several events (a skip emits multiple deletes plus a keep), so undo has to remove exactly that many. If you only stored the cursor, you'd have no way to know — which is the specific reason a cursor-only design has to be rewritten at gate 3.

Q: A failed undo(5) when only 3 operations exist — what happens? It raises IndexError and changes nothing. Validate n before mutating anything. A partially-applied failed operation is worse than a rejected one, because the caller now has no idea what state they're in.

Chapter 3: The Intrusive List — LRU and Friends

3.1 Why O(1) eviction needs two structures

An LRU cache needs two things at once:

1. get(key) in O(1) — "is this key here, and what's its value?"
2. "which key was used longest ago?" in O(1), and update on every access

No single structure does both. A hash map gives you (1) and knows nothing about ordering. A list gives you (2) and needs O(n) to find a key. So you use both, and the trick is wiring them together.

The wiring is what "intrusive" means. Instead of a list of values, you make the list nodes be the cache entries, and the hash map points directly at the nodes:

map:  "a" ──────────────┐   "b" ─────────┐    "c" ───┐
                        ▼                ▼           ▼
list: HEAD ⇄ [c: 3] ⇄ [a: 1] ⇄ [b: 2] ⇄ TAIL
              ^most recent          ^least recent

Now get("a") is: hash lookup → node (O(1)) → unlink node from wherever it is (O(1), because a doubly-linked node knows its own neighbours) → relink at the front (O(1)).

This is why the list must be doubly linked. In a singly-linked list, unlinking a node requires knowing its predecessor, which requires a scan — O(n) — and the whole design collapses. That is the sentence to say when asked "why doubly?".

Eviction is: take the node before TAIL, remove it from the list, delete its key from the map. O(1).

3.2 Sentinels, and why they delete every edge case

A naive linked list is a swamp of special cases: inserting into an empty list, removing the only node, removing the head, removing the tail. Each is a branch, and each branch is a bug.

Sentinel nodes eliminate all of them. Allocate two permanent nodes, head and tail, that hold no data and are never removed. The invariant becomes: every real node always has a non-None prev and next. Now:

def _unlink(node):
    node.prev.next = node.next        # never None-checks, because sentinels
    node.next.prev = node.prev

def _push_front(self, node):
    first = self._head.next
    node.prev, node.next = self._head, first
    self._head.next = node
    first.prev = node

No branches. An empty list is just head ⇄ tail, and pushing into it works by the same code path as pushing into a full one. This is a general technique — the same idea makes red-black tree code tractable via a NIL sentinel — and demonstrating it is a small but real signal of someone who has written data structures rather than only used them.

3.3 TTL: lazy, sampled, and active expiry

Add per-entry expiry. Three strategies, and the interview question is knowing the tradeoffs rather than picking "the right one".

Lazy expiry — check on read. If the entry is past its deadline, treat it as a miss and remove it.

• Zero background cost.
• An entry never read again is never freed. Memory leaks in proportion to your cold keyspace. This is the tradeoff to name; it is why lazy-only is not shippable.

Active expiry — a background sweeper walks all entries.

• Bounded memory.
• O(n) per sweep, and it competes with request traffic for the lock. At millions of keys, the sweep itself becomes the latency problem.

Sampled expiry — on each write (or on a timer), check a small random sample; if a high fraction were expired, sample again. This is what Redis does.

• O(1) amortized, no full scan.
• Probabilistic: it converges on keeping the expired fraction below a bound rather than guaranteeing zero.

The production answer is lazy + sampled. Lazy gives correctness on the read path — you never return a stale value. Sampled gives you a memory bound without a stop-the-world sweep. Saying "Redis does lazy plus sampled, and here's why neither alone is enough" is exactly the level of specificity these rounds reward.

One more subtlety worth raising: use a monotonic clock (time.monotonic()), not wall clock. Wall clock can jump backwards on NTP correction, which makes entries un-expire.

3.4 Size-aware eviction

Bounding by entry count is a lie when entries differ in size — 1,000 entries could be 1 MB or 1 GB. Real caches bound by bytes.

Two problems appear immediately:

How big is an entry? sys.getsizeof measures only the object's own footprint, not what it points to (see Track B). For a cache you generally want a caller-supplied cost function, or the serialized length if you're storing bytes anyway. Guessing produces a cache that thinks it's 100 MB and is actually 2 GB.

Evict until under the bound, not once. Inserting a 10 MB entry into a 100 MB cache that is 99 MB full must evict repeatedly. A single eviction leaves you over budget.

while self._bytes + cost > self._max_bytes and self._map:
    self._evict_lru()

And the guard: if a single item exceeds the whole budget, you must decide — reject it, or admit it and evict everything. Say which and why. (Reject is usually right; admitting it means one request destroys the cache for everyone else.)

3.5 Stampede control: single-flight

The failure that takes down real systems: a popular key expires. Five hundred concurrent requests miss simultaneously. All five hundred call the backing store. The backing store falls over — at the exact moment the cache was supposed to be protecting it.

This is a cache stampede (also: thundering herd, dogpile).

Single-flight fixes it: the first miss on a key starts the load and installs a promise; subsequent misses on the same key wait on that promise instead of starting their own load. One backend call, N waiters.

async def get_or_load(self, key, loader):
    hit = self.get(key)
    if hit is not None:
        return hit
    if key in self._inflight:              # someone is already loading it
        return await self._inflight[key]
    future = asyncio.get_running_loop().create_future()
    self._inflight[key] = future
    try:
        value = await loader(key)
        self.put(key, value)
        future.set_result(value)
        return value
    except Exception as exc:
        future.set_exception(exc)
        raise
    finally:
        self._inflight.pop(key, None)

Two related techniques worth naming:

• Negative caching — cache the absence of a key for a short TTL, so a flood of requests for nonexistent keys doesn't hit the backend repeatedly. This is a DoS mitigation, not a performance optimization, and framing it that way is the better answer.
• Probabilistic early expiration — refresh an entry slightly before it expires, with a probability that rises as the deadline approaches, so expirations desynchronize instead of all firing at once. The published version is "XFetch".

3.6 Complete implementation

import time


class _Node:
    __slots__ = ("key", "value", "expires_at", "cost", "prev", "next")

    def __init__(self, key=None, value=None, expires_at=None, cost=1):
        self.key, self.value = key, value
        self.expires_at, self.cost = expires_at, cost
        self.prev = self.next = None


class LRUCache:
    """LRU with per-entry TTL and a byte budget. O(1) get/put."""

    def __init__(self, max_bytes=1 << 20, clock=time.monotonic, sample=8):
        self._map = {}
        self._head, self._tail = _Node(), _Node()      # sentinels: no edge cases
        self._head.next, self._tail.prev = self._tail, self._head
        self._max_bytes, self._bytes = max_bytes, 0
        self._clock, self._sample = clock, sample
        self.hits = self.misses = self.evictions = self.expirations = 0

    # ---- list primitives (no branches, thanks to sentinels) --------------
    @staticmethod
    def _unlink(node):
        node.prev.next, node.next.prev = node.next, node.prev

    def _push_front(self, node):
        first = self._head.next
        node.prev, node.next = self._head, first
        self._head.next, first.prev = node, node

    def _touch(self, node):
        self._unlink(node)
        self._push_front(node)

    # ---- expiry ----------------------------------------------------------
    def _expired(self, node):
        return node.expires_at is not None and self._clock() >= node.expires_at

    def _drop(self, node):
        self._unlink(node)
        del self._map[node.key]
        self._bytes -= node.cost

    def _sample_expired(self):
        """Redis-style: check a small random sample instead of scanning."""
        checked = 0
        node = self._tail.prev                       # coldest end first
        while node is not self._head and checked < self._sample:
            nxt, checked = node.prev, checked + 1
            if self._expired(node):
                self._drop(node)
                self.expirations += 1
            node = nxt

    # ---- public API ------------------------------------------------------
    def get(self, key, default=None):
        node = self._map.get(key)
        if node is None:
            self.misses += 1
            return default
        if self._expired(node):                      # lazy expiry on the read path
            self._drop(node)
            self.expirations += 1
            self.misses += 1
            return default
        self._touch(node)
        self.hits += 1
        return node.value

    def put(self, key, value, ttl=None, cost=1):
        if cost > self._max_bytes:
            raise ValueError("entry exceeds the whole budget")
        existing = self._map.get(key)
        if existing is not None:
            self._bytes -= existing.cost
            self._unlink(existing)
            del self._map[key]

        expires_at = None if ttl is None else self._clock() + ttl
        node = _Node(key, value, expires_at, cost)
        self._map[key] = node
        self._push_front(node)
        self._bytes += cost

        self._sample_expired()                       # cheap, bounded
        while self._bytes > self._max_bytes and len(self._map) > 1:
            victim = self._tail.prev                 # evict UNTIL under budget
            self._drop(victim)
            self.evictions += 1
        return node

    def __len__(self):
        return len(self._map)

    def stats(self):
        total = self.hits + self.misses
        return {"size": len(self._map), "bytes": self._bytes,
                "hits": self.hits, "misses": self.misses,
                "hit_rate": self.hits / total if total else 0.0,
                "evictions": self.evictions, "expirations": self.expirations}

3.7 Interview Q&A

Q: Why a doubly-linked list? Because eviction and promotion both need to unlink a node you already have a pointer to, in O(1). A singly-linked node doesn't know its predecessor, so unlinking requires an O(n) scan and the whole design degenerates. The hash map gives me the node pointer; the double links make removing it constant time.

Q: Why not OrderedDict? For production I would — OrderedDict.move_to_end is exactly this, implemented in C. In an interview the question is whether I know the mechanism, so I build it. Worth noting that functools.lru_cache is also this structure, and that its key is the argument tuple — which means an lru_cache on a method keys on self and holds a strong reference to every instance it has ever seen. That's an unbounded leak on a long-lived class.

Q: Lazy expiry leaks. How do you bound it? Lazy alone means an entry never read again is never freed, so memory grows with the cold keyspace. Add sampled expiry: on each write, check a small random sample from the cold end and drop what's expired. That's what Redis does, and it's O(1) amortized with no stop-the-world sweep. Full active sweeping is the third option but it's O(n) and competes with request traffic for the lock.

Q: How do you know how many bytes an entry is? Honestly: you don't, reliably. sys.getsizeof only measures the object's own footprint, not what it references — a list of 10,000 strings reports ~80 KB while costing megabytes. So I take a caller-supplied cost function, or use the serialized length if I'm storing bytes anyway. Guessing gives you a cache that believes it's 100 MB and is actually 2 GB, and you find out during an incident.

Q: Popular key expires, 500 requests miss at once. What happens? A cache stampede — all 500 hit the backing store simultaneously, at the moment the cache was supposed to be protecting it. Single-flight: the first miss installs a promise, the rest await it, so one backend call serves N waiters. I'd add negative caching for nonexistent keys — which is a DoS mitigation rather than a perf win — and probabilistic early expiration so a whole cohort of keys doesn't expire in lockstep.

Q: Make it thread-safe. One lock around every mutation is correct and becomes the bottleneck, because every get mutates the list. Sharding by hash(key) % N into N independently-locked caches removes the contention, at the cost of a global LRU order — each shard evicts its own coldest, which is slightly worse than true global LRU but almost always an acceptable trade. Shard count comes from measured contention, not from a round number.

Q: When is LRU the wrong policy? When a scan touches every key once — a batch job walking the whole keyspace evicts your entire hot set and gets nothing in return. That's cache pollution. LFU or a segmented LRU (a small probation segment that entries must be hit in twice to be promoted) resists it. Modern practice is admission control: TinyLFU keeps a compact frequency sketch and refuses to admit an item unless it's likely more valuable than the victim.

Q: What's your hit rate telling you? That it's the only number that matters for a cache, and it must be measured, not assumed. A 90% hit rate against a 50 ms backend gives a 5.4 ms average; 80% gives 10.4 ms. Halving the miss rate nearly halves the latency, which is why capacity decisions should be driven by a hit-rate-versus-size curve rather than by picking a memory number.

Chapter 4: Rate Limiting — Four Algorithms and Their Lies

4.1 Fixed window, and the boundary burst

The simplest limiter: count requests per fixed clock interval.

window = int(now // 60)
counts[key, window] += 1
allow = counts[key, window] <= limit

100 requests per minute. Simple, O(1) memory per key.

The flaw, and you must be able to state it precisely: a client can send 2× the limit in a window as short as one instant. Send 100 requests at 11:00:59.9 (all in the 11:00 window), then 100 more at 11:01:00.1 (all in the 11:01 window). Both windows are within limit. 200 requests in 200 milliseconds.

That is not a rounding error. It is a 2× breach of your stated contract, and it is exactly what a client optimizing for throughput will discover and exploit.

4.2 Sliding window log

Store the timestamp of every request; on each call, drop timestamps older than the window and count what remains.

log = deque()
def allow(now):
    cutoff = now - window
    while log and log[0] <= cutoff:
        log.popleft()
    if len(log) < limit:
        log.append(now)
        return True
    return False

Exactly correct. Never allows more than limit in any window of length window, no boundary effect at all.

The cost: O(limit) memory per key. At 10,000 requests/minute per key across a million keys, that is ten billion timestamps. Not shippable at scale, which is why it exists mainly as the correctness baseline the approximations are measured against.

4.3 Sliding window counter

The engineering compromise, and the one most production systems actually run.

Keep counters for the current and previous fixed windows, and interpolate based on how far into the current window you are:

elapsed  = now % window_size
weight   = 1 - elapsed / window_size          # how much of the previous window still counts
estimate = previous_count * weight + current_count
allow    = estimate < limit

Worked example. Limit 100/minute. It is 11:00:30 — half way through the current window. The 10:59 window saw 80 requests; the 11:00 window has seen 40 so far.

weight   = 1 - 30/60 = 0.5
estimate = 80 * 0.5 + 40 = 80
80 < 100  →  allow

O(1) memory, two counters per key, and the boundary burst is gone.

The lie you must disclose: it assumes requests were spread uniformly across the previous window. If all 80 arrived in the last second of 10:59, the true count in the trailing 60 seconds is 120 and you allowed it. Cloudflare published measurements putting the error under 1% on real traffic — good enough for almost everything, and "good enough with a measured error bound" is a much better interview answer than "correct" or "approximate".

4.4 Token bucket, derived

Different model, and the one to reach for when bursts are legitimate.

Imagine a bucket holding at most capacity tokens, refilled continuously at rate tokens per second. Each request removes one token. No token, no request.

The insight that makes it O(1): you do not need a background thread refilling it. Compute the refill lazily from elapsed time:

elapsed = now - self.last
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last = now
if self.tokens >= 1:
    self.tokens -= 1
    return True
return False

Two knobs with distinct meanings, and being able to separate them is the point:

• rate is the sustained throughput you permit — the long-run average.
• capacity is the burst you tolerate — how much unused allowance can accumulate.

A client that has been idle for a minute accumulates a full bucket and may burst immediately. That is usually desirable: it rewards well-behaved clients and matches how real traffic arrives. Fixed windows cannot express this at all.

Make the clock injectable. A limiter you cannot test deterministically is a limiter you cannot ship. Passing a clock callable turns "sleep 60 seconds in a test" into "advance a fake clock", and it is the single most useful design decision in the whole component.

4.5 Leaky bucket, and how it differs

Same bucket picture, opposite plumbing. Requests enter a queue; the queue drains at a constant rate. If the queue is full, the request is dropped.

The difference that matters:

Token bucket limits the average with allowed bursts. Leaky bucket limits the instantaneous rate. Choosing the wrong one is how you build a limiter that lets 100 requests hit a downstream that can handle 10 at a time.

4.6 Distributed limiting

Now N processes must share one limit. Four problems appear, and naming all four is the staff-level answer.

1. Atomicity. Read-then-write over the network is a race: two processes both read 99, both write 100, and 101 requests get through. You need a single atomic operation — a Lua script in Redis, INCR with expiry, or a compare-and-swap loop.

2. Round-trip cost. A network hop per request may cost more than the work you are protecting. Mitigate by having each process lease a batch of permits — take 10, spend them locally, then go back. Trades precision for latency, and you should say which you chose.

3. Clock skew. Timestamps generated on different machines disagree. Use the store's clock (Redis TIME) rather than each caller's, or use logical counters so no clock is involved.

4. The store is down. Fail open (allow everything) or fail closed (deny everything)? There is no universally right answer, and interviewers ask precisely because they want to hear you reason:

• Fail open if the limiter protects against accidental overload — a limiter outage should not become a total outage.
• Fail closed if the limiter enforces billing or abuse limits — failing open there means free unlimited usage during your incident.
• The sophisticated answer: fail open with a degraded local limit, so each process enforces global_limit / process_count on its own. You lose global precision and keep a bound.

4.7 Complete implementation

import time
from collections import deque


class TokenBucket:
    """Sustained `rate` per second, burst up to `capacity`. O(1), lazy refill."""

    __slots__ = ("capacity", "rate", "_tokens", "_last", "_clock")

    def __init__(self, capacity, rate, clock=time.monotonic):
        if capacity <= 0 or rate <= 0:
            raise ValueError("capacity and rate must be positive")
        self.capacity, self.rate = float(capacity), float(rate)
        self._tokens = float(capacity)
        self._clock = clock
        self._last = clock()

    def _refill(self):
        now = self._clock()
        elapsed = now - self._last
        if elapsed > 0:
            self._tokens = min(self.capacity, self._tokens + elapsed * self.rate)
            self._last = now

    def allow(self, cost=1):
        self._refill()
        if self._tokens >= cost:
            self._tokens -= cost
            return True
        return False

    def retry_after(self, cost=1):
        """Seconds until `cost` tokens exist — send this in the 429 response."""
        self._refill()
        if self._tokens >= cost:
            return 0.0
        return (cost - self._tokens) / self.rate


class SlidingWindowCounter:
    """O(1) memory, no boundary burst, ~1% error on real traffic."""

    __slots__ = ("limit", "window", "_curr_start", "_curr", "_prev", "_clock")

    def __init__(self, limit, window, clock=time.monotonic):
        self.limit, self.window = limit, float(window)
        self._clock = clock
        self._curr_start = self._floor(clock())
        self._curr = self._prev = 0

    def _floor(self, now):
        return now - (now % self.window)

    def _roll(self, now):
        start = self._floor(now)
        if start == self._curr_start:
            return
        if start - self._curr_start == self.window:
            self._prev, self._curr = self._curr, 0      # slid by exactly one
        else:
            self._prev, self._curr = 0, 0               # gap: both windows stale
        self._curr_start = start

    def allow(self):
        now = self._clock()
        self._roll(now)
        weight = 1.0 - (now - self._curr_start) / self.window
        if self._prev * weight + self._curr < self.limit:
            self._curr += 1
            return True
        return False


class SlidingWindowLog:
    """Exactly correct. O(limit) memory per key — the correctness baseline."""

    __slots__ = ("limit", "window", "_log", "_clock")

    def __init__(self, limit, window, clock=time.monotonic):
        self.limit, self.window = limit, float(window)
        self._log = deque()
        self._clock = clock

    def allow(self):
        now = self._clock()
        cutoff = now - self.window
        while self._log and self._log[0] <= cutoff:
            self._log.popleft()
        if len(self._log) < self.limit:
            self._log.append(now)
            return True
        return False


class ShardedLimiter:
    """Per-key limiters with sharded locks — one global lock is the bottleneck,
    because every call mutates state."""

    def __init__(self, factory, shards=16):
        import threading
        self._factory = factory
        self._shards = [({}, threading.Lock()) for _ in range(shards)]

    def _shard(self, key):
        return self._shards[hash(key) % len(self._shards)]

    def allow(self, key, cost=1):
        buckets, lock = self._shard(key)
        with lock:
            bucket = buckets.get(key)
            if bucket is None:
                bucket = buckets[key] = self._factory()
            return bucket.allow(cost)

4.8 Interview Q&A

Q: Fixed window — what's wrong with it? It permits 2× the limit across a window boundary. 100 requests at 11:00:59.9 and 100 more at 11:01:00.1 are both within their windows and are 200 requests in 200 milliseconds. That's not a rounding error, it's a full breach of the stated contract, and it's exactly what a client tuning for throughput will find.

Q: So use the sliding window log? It's exactly correct, and it's O(limit) memory per key. At 10k/min across a million keys that's ten billion timestamps. I keep it as the correctness baseline and ship the sliding window counter — two counters per key, O(1), interpolating the previous window by how far into the current one we are. It assumes uniform arrival in the previous window; Cloudflare measured the error under 1% on real traffic. I'd rather quote a measured error bound than claim exactness.

Q: When token bucket over sliding window? When bursts are legitimate. Token bucket separates two things a window conflates: rate is the sustained throughput, capacity is the burst tolerated. A client idle for a minute accumulates a full bucket and may spend it at once — usually what you want, because it rewards well-behaved clients. A sliding window can't express "average 10/s but 100 at once is fine."

Q: Token bucket vs leaky bucket? Token bucket allows a burst; the output is bursty. Leaky bucket queues and drains at a fixed rate; the output is perfectly smooth. If I'm protecting a downstream with a hard concurrency limit that genuinely cannot absorb a burst, leaky is right and token bucket will hurt it. If I'm protecting against sustained abuse and the service can absorb spikes, token bucket is right.

Q: Why an injectable clock? Because otherwise the tests have to sleep, which makes them slow and flaky, and slow flaky tests get deleted. With an injected clock I advance time to any point instantly and test the boundary conditions exactly. It's also how I test the distributed version's skew handling.

Q: Make it distributed. Four problems. Atomicity — read-then-write over the network races, so I need a Lua script or INCR with expiry, one round trip, one atomic op. Round-trip cost — I'd lease permits in batches so the common path is local, trading precision for latency. Clock skew — use the store's clock, not each caller's. And availability — decide fail-open vs fail-closed explicitly: fail open for overload protection, fail closed for billing and abuse, or fail open with a degraded local limit of global / process_count, which is the answer I'd actually ship because it bounds the damage in both directions.

Q: One global lock for per-key limiters? It's correct and it's the bottleneck, because every call mutates state — there's no read-only fast path. Shard by hash(key) % N into independently-locked maps. Shard count from measured contention, not a round number. Under free-threaded Python this matters much more, because that's exactly the workload that stops being serialized by the GIL.

Q: What should the 429 response contain? Retry-After, computed as (cost - tokens) / rate — the actual time until the request would succeed. Without it clients retry blindly and you get a retry storm on top of the overload you were already limiting. That one header is the difference between a limiter that sheds load and one that amplifies it.

Chapter 5: Heaps and Deterministic Scheduling

5.1 What a binary heap actually is

A binary heap is a complete binary tree stored in a flat array, satisfying the heap property: every node is ≤ both of its children (a min-heap).

"Complete" means every level is full except possibly the last, which fills left to right. That lets you drop the pointers entirely and use arithmetic:

node at index i:  parent = (i - 1) // 2
                  left   = 2i + 1
                  right  = 2i + 2

Array [1, 3, 2, 7, 4, 5] is the tree:

        1
      /   \
     3     2
    / \   /
   7   4 5

Push appends at the end and sifts up: while the new element is smaller than its parent, swap. The tree has depth ⌊log₂ n⌋, so this is O(log n).

Pop takes the root (the minimum — O(1) to read), moves the last element to the root, and sifts down: swap with the smaller child until the property holds. Also O(log n).

Why a heap rather than a sorted list? A sorted list gives O(1) min but O(n) insert. A heap gives O(log n) for both, and — crucially — it does not maintain a total order it does not need. For a scheduler you only ever ask "what fires next?", so paying to keep everything sorted is waste.

heapq in Python is a min-heap over the array you give it. For a max-heap, negate the keys.

5.2 The tie-breaking bug that flakes your tests

Here is a bug that will cost you a gate if you have not seen it.

heapq.heappush(self._heap, (fire_at, job))

Two jobs scheduled for the same instant. heapq compares tuples element-wise: it compares fire_at, they are equal, so it moves on to compare job against job — and if Job does not define __lt__, you get:

TypeError: '<' not supported between instances of 'Job' and 'Job'

If Job is comparable (say it's a dict, or a dataclass with order=True), it is worse: no exception, but the ordering depends on job contents, which means your scheduler fires same-instant jobs in an order determined by their payloads. Tests pass locally and flake in CI, and you will spend a day on it.

The fix is a monotonically increasing sequence number as a tie-breaker:

self._seq += 1
heapq.heappush(self._heap, (fire_at, self._seq, job))

Now ties are broken by insertion order — FIFO among simultaneous jobs, which is both deterministic and the semantics people expect. The third element is never reached, so job never needs to be comparable.

This is a small thing that signals real experience. Mention it before it bites you.

5.3 Cancellation: the lazy-deletion trick

heapq has no "remove this element". Finding an arbitrary element is O(n), and removing it means re-heapifying. So how do you cancel a scheduled job?

Lazy deletion. Keep a set of cancelled IDs. Leave the entry in the heap; skip it when it surfaces.

def cancel(self, job_id):
    if job_id in self._live:
        self._cancelled.add(job_id)
        del self._live[job_id]
        return True
    return False

def _pop_next_valid(self):
    while self._heap:
        fire_at, seq, job_id = heapq.heappop(self._heap)
        if job_id in self._cancelled:
            self._cancelled.discard(job_id)     # tombstone consumed
            continue
        return fire_at, job_id
    return None

cancel is O(1); the cost is deferred to the pop that eventually discards it.

The failure mode you must name unprompted: if a workload schedules far into the future and cancels most of it, the heap fills with tombstones that will not surface for hours, and memory grows without bound. The fix is a rebuild threshold — when len(cancelled) > len(heap) // 2, filter and heapify in O(n). That is amortized O(1) per cancellation and bounds the waste at 2×.

This exact pattern — lazy deletion plus a rebuild threshold — is how asyncio's event loop handles cancelled timers. Saying so is a good, specific reference.

5.4 Fixed-rate versus fixed-delay

For recurring jobs there are two schedules, and confusing them causes real incidents.

Fixed rate — fire at t₀, t₀+p, t₀+2p, … regardless of how long a run takes. Fixed delay — fire p after the previous run finished.

Suppose period = 60s and one run takes 150s.

Fixed rate with an overrunning job is how you get the same job running four times concurrently, all fighting over the same rows. Fixed delay guarantees no overlap and lets the schedule drift.

Neither is "correct" — they answer different needs. What is not acceptable is choosing silently. The design must surface a catch-up policy for fixed rate: run_all (fire every missed occurrence), run_latest_only (collapse missed occurrences into one), or skip. A scheduler that comes back after a two-hour outage and fires 40,000 overdue jobs at once has turned an outage into a second, worse outage.

5.5 Backoff with jitter, and why it matters

When a job fails, retry — but not immediately, and not on a fixed schedule.

Exponential backoff: wait base × 2^attempt, capped. Gives a struggling dependency increasing room to recover.

Why jitter is not optional. Without it, every client that failed at the same moment retries at the same moment. A downstream that dropped 1,000 requests gets all 1,000 back simultaneously, fails again, and the cohort stays synchronized forever. This is a retry storm, and it converts a partial outage into a total one.

Three variants (the naming follows the AWS Architecture Blog's analysis):

# "Full jitter" — the AWS default recommendation
sleep = random.uniform(0, min(cap, base * 2 ** attempt))

# "Equal jitter" — half fixed, half random; less variance, keeps a floor
temp  = min(cap, base * 2 ** attempt)
sleep = temp / 2 + random.uniform(0, temp / 2)

# "Decorrelated jitter" — walks from the previous sleep; smoother, harder to bound
sleep = min(cap, random.uniform(base, previous * 3))

AWS's published simulation found full jitter minimized both total work and completion time under contention. Equal jitter keeps a minimum wait, which matters if you need a floor. Decorrelated is smoothest but its worst case is harder to reason about.

Jitter alone is not enough, and this is the part people miss. Even perfectly jittered retries multiply offered load: with 3 attempts and a 95% failure rate you send ~2.85× your base traffic to a dependency that is already failing. The primary fix is a retry budget — cap retries at a fraction (say 10%) of base traffic, so amplification is bounded no matter how bad things get. Then a circuit breaker to stop trying entirely. Then jitter. In that order.

5.6 Complete implementation

import heapq
import itertools
import random
import time


class Scheduler:
    """Delayed and recurring execution with O(1) cancellation, retries with
    jitter, and an explicit catch-up policy."""

    def __init__(self, clock=time.monotonic, rng=None):
        self._heap = []                     # (fire_at, seq, job_id)
        self._live = {}                     # job_id -> job record
        self._cancelled = set()
        self._seq = itertools.count()
        self._ids = itertools.count(1)
        self._clock = clock
        self._rng = rng or random.Random(0)   # seeded: deterministic tests

    # ---- scheduling ------------------------------------------------------
    def schedule(self, fn, delay=0.0, *, period=None, mode="fixed_delay",
                 catch_up="run_latest_only", max_attempts=5,
                 base_backoff=0.2, cap_backoff=30.0):
        if period is not None and mode not in ("fixed_rate", "fixed_delay"):
            raise ValueError("mode must be fixed_rate or fixed_delay")
        job_id = next(self._ids)
        self._live[job_id] = {
            "fn": fn, "period": period, "mode": mode, "catch_up": catch_up,
            "max_attempts": max_attempts, "attempt": 0,
            "base": base_backoff, "cap": cap_backoff,
        }
        self._push(self._clock() + delay, job_id)
        return job_id

    def _push(self, fire_at, job_id):
        # The sequence number breaks ties deterministically. Without it heapq
        # falls through to comparing the payload, which either raises or makes
        # ordering depend on job contents — a classic CI-only flake.
        heapq.heappush(self._heap, (fire_at, next(self._seq), job_id))

    def cancel(self, job_id):
        if job_id not in self._live:
            return False
        del self._live[job_id]
        self._cancelled.add(job_id)          # lazy deletion: O(1)
        self._maybe_rebuild()
        return True

    def _maybe_rebuild(self):
        # Bound tombstone waste at 2x. Without this, a workload that schedules
        # far ahead and cancels most of it grows the heap without bound.
        if len(self._cancelled) > max(32, len(self._heap) // 2):
            self._heap = [e for e in self._heap if e[2] not in self._cancelled]
            heapq.heapify(self._heap)
            self._cancelled.clear()

    # ---- running ---------------------------------------------------------
    def _backoff(self, job):
        ceiling = min(job["cap"], job["base"] * (2 ** job["attempt"]))
        return self._rng.uniform(0, ceiling)      # full jitter

    def run_due(self, limit=None):
        """Run everything due now. Returns (ran, errors)."""
        now, ran, errors = self._clock(), 0, []
        while self._heap and (limit is None or ran < limit):
            fire_at, _, job_id = self._heap[0]
            if fire_at > now:
                break
            heapq.heappop(self._heap)
            if job_id in self._cancelled:
                self._cancelled.discard(job_id)
                continue
            job = self._live.get(job_id)
            if job is None:
                continue

            started = self._clock()
            try:
                job["fn"]()
                job["attempt"] = 0
                ran += 1
            except Exception as exc:
                errors.append((job_id, exc))
                job["attempt"] += 1
                if job["attempt"] < job["max_attempts"]:
                    self._push(self._clock() + self._backoff(job), job_id)
                    continue
                del self._live[job_id]        # dead-lettered
                continue

            period = job["period"]
            if period is None:
                del self._live[job_id]
                continue

            if job["mode"] == "fixed_delay":
                self._push(self._clock() + period, job_id)
            else:                              # fixed_rate
                nxt = fire_at + period
                if nxt <= now:                 # we overran; apply catch-up policy
                    if job["catch_up"] == "run_latest_only":
                        missed = int((now - nxt) // period) + 1
                        nxt += missed * period
                    elif job["catch_up"] == "skip":
                        nxt = now + period
                    # "run_all" leaves nxt in the past — it fires immediately,
                    # once per missed occurrence, until it catches up.
                self._push(nxt, job_id)
        return ran, errors

    def next_fire_time(self):
        while self._heap and self._heap[0][2] in self._cancelled:
            self._cancelled.discard(heapq.heappop(self._heap)[2])
        return self._heap[0][0] if self._heap else None

5.7 Interview Q&A

Q: Why a heap and not a sorted list? A sorted list gives O(1) min and O(n) insert. A heap gives O(log n) for both. A scheduler only ever asks "what's next?", so paying to maintain a total order over everything is waste — the heap maintains exactly the partial order the query needs.

Q: Two jobs at the same timestamp — what happens? Without a tie-breaker, heapq falls through to comparing the payloads: either TypeError, or, worse, silent ordering by job contents that flakes only in CI. I push (fire_at, sequence, job_id). Ties break by insertion order — FIFO among simultaneous jobs — and the payload is never compared.

Q: How do you cancel? Lazy deletion. heapq has no remove, and finding an element is O(n), so I mark the ID cancelled and skip it when it surfaces — O(1) cancel. The failure mode is tombstone accumulation when you schedule far ahead and cancel most of it, so I rebuild when cancellations exceed half the heap: filter and heapify, O(n), amortized O(1) per cancel, waste bounded at 2×. That's the same pattern asyncio's loop uses for cancelled timers.

Q: Fixed rate vs fixed delay? Fixed rate fires on a fixed schedule regardless of run duration; if a run overruns its period you get concurrent executions of the same job. Fixed delay waits p after the previous run finished; it never overlaps and the schedule drifts. Neither is correct in general — but the design has to expose the choice, plus a catch-up policy for fixed rate, because a scheduler that comes back from a two-hour outage and fires 40,000 overdue jobs has turned one outage into a worse one.

Q: Why jitter? Without it every client that failed at the same instant retries at the same instant. A downstream that dropped 1,000 requests gets all 1,000 back at once, fails again, and the cohort stays permanently synchronized. Full jitter — uniform over [0, cap] — minimized both total work and completion time in AWS's published simulation. But jitter alone isn't enough: 3 attempts at a 95% failure rate is still ~2.85× offered load. The primary fix is a retry budget capping retries at ~10% of base traffic; then a circuit breaker; then jitter, in that order.

Q: Make it multi-process. The heap moves to shared storage — a table with an index on next_run_at, claimed with SELECT ... FOR UPDATE SKIP LOCKED, which is genuinely the right answer up to a few thousand dispatches/second. Then you need a lease so a dead worker's claim is released, and a fencing token so a worker that pauses past its lease and wakes up can't write after its replacement already ran. That last part is the one people miss, and it's the difference between at-least-once being safe and being silently corrupting.

Q: What's the throughput ceiling here? The single heap and single-threaded run_due. Each run_due is O(k log n) for k due jobs. To scale I'd shard by job ID, one heap per shard, which trades global ordering for parallelism — and global ordering across independent jobs usually isn't a requirement, so it's a cheap trade. Worth stating explicitly rather than assuming.

Chapter 6: Streaming State Machines and Chunk Boundaries

6.1 The only hard part: the boundary

You are handed data in arbitrary chunks — network reads, file blocks — and must emit tokens. Everything about this problem is the case where a token straddles a chunk boundary.

chunk 1: b'{"na'
chunk 2: b'me": "alice"}'

The token "name" exists in neither chunk. A naive chunk.split() per chunk emits {"na and me": "alice"} — two wrong tokens, and the error is silent.

The universal solution is a carry buffer. Keep the unconsumed tail of the previous chunk; prepend it to the next one; consume only complete tokens; carry the remainder.

def feed(self, chunk):
    self._buffer += chunk
    tokens = []
    while True:
        token, consumed = self._try_consume(self._buffer)
        if token is None:
            break                            # incomplete: wait for more data
        tokens.append(token)
        self._buffer = self._buffer[consumed:]
    return tokens

The contract that makes it correct: _try_consume returns a token only if it is certainly complete. If the buffer ends mid-token, it returns None and the bytes stay in the buffer. No token is ever emitted twice, and none is lost.

close() then handles the end-of-stream case: whatever is left in the buffer is either a final token (if the grammar allows an unterminated one) or an error.

6.2 Why a regex cannot do this

The instinct is re.findall. It fails, for a specific reason.

A regex match is computed against a complete string. Given '{"na' the pattern for a quoted string does not match — correct. But given '{"name": "alice"' a greedy pattern may match a prefix that happens to be well-formed, and you cannot tell whether more input would have extended the match. The regex engine has no concept of "this input might continue."

Python's re does expose this: re.match sets a partial notion only in the third-party regex module, not the stdlib. So in the stdlib you have no way to distinguish "no match" from "no match yet".

The correct tool is a character-level state machine: an explicit state variable, a loop over characters, and a transition per character.

DEFAULT --- '"' ---> IN_STRING --- '\' ---> ESCAPED
                          |  ^                 |
                          |  +-----------------+  (any char)
                          |
                        '"' ---> DEFAULT (emit token)

State machines handle boundaries naturally because the state survives between chunks. If you end a chunk in ESCAPED, you resume the next chunk in ESCAPED. A regex has no state to carry.

This is why every real streaming parser — HTTP, JSON, protobuf framing — is a hand-written state machine, and it is a good thing to say out loud.

6.3 Bounded buffers and the pathological input

A carry buffer that grows without bound is a denial-of-service vector: a client sends 4 GB with no delimiter, your buffer holds all of it, your process dies.

So cap it, and decide explicitly what happens on overflow:

• Error the connection. Right for most protocols — a token longer than the cap is malformed by definition.
• Truncate and emit what you have with a flag. Right for logs, where you would rather keep going.
• Spill to disk. Right when huge tokens are legitimate.

Say which and why. "I'd cap the buffer at 1 MB and error, because in this protocol a million-character token is malformed by definition" is a complete answer; silently unbounded is a security bug.

Related and worth mentioning: track a byte offset for every token. When a parse fails at byte 4,821,993 of a stream, "invalid character" without a position is unusable, and offsets cost you one integer.

6.4 Complete implementation

class StreamingTokenizer:
    """Chunk-boundary-safe tokenizer with quoting, escapes, nesting and a
    bounded buffer. A character-level state machine, not a regex."""

    DEFAULT, IN_STRING, ESCAPED = 0, 1, 2

    def __init__(self, max_token_bytes=1 << 20, delimiters=" \t\n\r,"):
        self._buffer = ""
        self._state = self.DEFAULT
        self._token = []                 # chars of the token under construction
        self._depth = 0
        self._offset = 0                 # absolute byte offset of the stream
        self._token_start = 0
        self._max = max_token_bytes
        self._delims = set(delimiters)
        self._closed = False

    def feed(self, chunk):
        if self._closed:
            raise RuntimeError("tokenizer is closed")
        out = []
        for ch in chunk:
            self._offset += 1
            if self._state == self.DEFAULT:
                if ch == '"':
                    self._flush(out)
                    self._state = self.IN_STRING
                    self._token_start = self._offset - 1
                elif ch in "[{":
                    self._flush(out)
                    self._depth += 1
                    out.append(("open", ch, self._depth, self._offset - 1))
                elif ch in "]}":
                    self._flush(out)
                    out.append(("close", ch, self._depth, self._offset - 1))
                    self._depth = max(0, self._depth - 1)
                elif ch in self._delims:
                    self._flush(out)
                else:
                    if not self._token:
                        self._token_start = self._offset - 1
                    self._token.append(ch)
            elif self._state == self.IN_STRING:
                if ch == "\\":
                    self._state = self.ESCAPED       # state SURVIVES the chunk
                elif ch == '"':
                    out.append(("string", "".join(self._token),
                                self._depth, self._token_start))
                    self._token.clear()
                    self._state = self.DEFAULT
                else:
                    self._token.append(ch)
            else:                                     # ESCAPED
                self._token.append({"n": "\n", "t": "\t", "r": "\r"}.get(ch, ch))
                self._state = self.IN_STRING

            if len(self._token) > self._max:
                raise ValueError(
                    f"token exceeded {self._max} bytes at offset {self._offset}; "
                    "a token this long is malformed in this grammar")
        return out

    def _flush(self, out):
        if self._token:
            out.append(("bare", "".join(self._token), self._depth, self._token_start))
            self._token.clear()

    def close(self):
        self._closed = True
        if self._state != self.DEFAULT:
            raise ValueError(f"stream ended mid-string at offset {self._offset}")
        out = []
        self._flush(out)
        if self._depth:
            raise ValueError(f"stream ended with {self._depth} unclosed container(s)")
        return out

Feed this '{"na' then 'me": "ali' then 'ce"}' and it emits the name and alice strings whole, exactly once each, with correct offsets — because the state and the partial token both survive between calls.

6.5 Interview Q&A

Q: Why not a regex? Because a regex matches against a complete string and has no notion of "this input might continue." Given a buffer that ends mid-token it can't distinguish "no match" from "no match yet", and a greedy pattern may match a well-formed prefix that more input would have extended. A character-level state machine handles it naturally because the state survives between chunks — end a chunk in ESCAPED, resume the next in ESCAPED. That's why every real streaming parser is a hand-written state machine.

Q: What if a chunk splits an escape sequence? That's exactly the case the ESCAPED state exists for. The backslash sets the state; the chunk ends; the next feed starts in ESCAPED and consumes the next character as the escapee. No special-casing at all — which is the point of modelling it as a state machine rather than as lookahead.

Q: A client sends 4 GB with no delimiter. Without a bound that's a denial-of-service — my buffer holds all of it. So I cap it and decide the overflow behaviour explicitly. Here I error, because in this grammar a megabyte-long token is malformed by definition. In a log pipeline I'd truncate and flag instead; if huge tokens were legitimate I'd spill to disk. What isn't acceptable is unbounded, and the choice should be stated rather than defaulted.

Q: How do you report errors usefully? Absolute byte offsets on every token and every error. "Invalid character" at byte 4,821,993 of a stream is actionable; "invalid character" is not. It costs one integer that I increment per character.

Q: How would you recover and keep going after a malformed region? Add a RESYNC state: on error, discard input until a character that can only appear at a structural boundary — a top-level delimiter — then resume in DEFAULT and emit a gap marker so the consumer knows something was dropped. The important part is being honest downstream that data was skipped, rather than silently producing a shorter stream.

Q: Make it faster. Profile before answering, but the usual finding is that per-character Python is the cost. Batch with str.find to jump to the next interesting character rather than looping over every one, which turns the common case (long runs of ordinary characters) into a C-level scan. If it's still hot, that's the case for a C extension or re used only on complete buffered regions — but I'd want the measurement first.

Chapter 7: Dependency Graphs, Topological Order, Cycles

7.1 Two algorithms and when each wins

A spreadsheet, a build system, and a task runner are the same problem: things depend on other things, and you must evaluate in an order where dependencies come first. That order is a topological sort of a directed acyclic graph.

Two standard algorithms:

Kahn's algorithm (BFS-flavoured). Compute each node's in-degree. Repeatedly take a node with in-degree 0, output it, and decrement its successors' in-degrees.

• Naturally detects cycles: if you finish with nodes remaining, those nodes are in cycles.
• Naturally parallel: everything at in-degree 0 at a given moment can run concurrently. This is exactly why build systems use it.
• Needs the in-degree map up front.

DFS with three colours. Depth-first; output each node after all its descendants; reverse.

• Detects cycles precisely, and can report the actual cycle path.
• No pre-pass needed.
• Recursion depth equals graph depth, so deep graphs need an explicit stack.

Use Kahn when you want parallelism or a simple "is there a cycle" answer. Use DFS when you want to report the cycle — and a spreadsheet does, because #CIRCULAR is much less useful than "A1 → B1 → C1 → A1".

7.2 The three-colour DFS

The colours encode exactly what you need to distinguish a cycle from a diamond:

• WHITE — not visited.
• GREY — visiting: on the current recursion stack, descendants still being explored.
• BLACK — done: fully explored.

Encountering a GREY node means you have reached a node that is an ancestor of yourself on the current path. That is a back edge, and a back edge is exactly a cycle.

Encountering a BLACK node is fine — you have reached something already fully processed by a different path. That is a diamond (A → B → D, A → C → D), which is perfectly legal in a DAG.

This is the distinction a two-state "visited" set cannot make, and it is the specific reason naive cycle detection either misses cycles or falsely reports diamonds as cycles. Being able to say "you need three states, because 'visited' conflates 'on my current path' with 'already done'" is the whole answer to this question.

WHITE, GREY, BLACK = 0, 1, 2

def toposort(nodes, deps):
    color = {n: WHITE for n in nodes}
    order, path = [], []

    def visit(n):
        if color[n] == BLACK:
            return
        if color[n] == GREY:
            cycle = path[path.index(n):] + [n]
            raise CycleError(cycle)          # report the actual path
        color[n] = GREY
        path.append(n)
        for m in deps.get(n, ()):
            visit(m)
        path.pop()
        color[n] = BLACK
        order.append(n)                      # after all descendants

    for n in nodes:
        visit(n)
    return order                             # dependencies first

Note that order is already dependency-first because a node is appended after its descendants. No reversal needed with this formulation, which is a nice thing to get right.

7.3 Incremental recomputation

Editing one cell should not recompute the sheet. It should recompute exactly the cells that transitively depend on the edited one.

That needs the reverse graph — the dependents of each node, not its dependencies:

self._deps      = {}   # cell -> cells it reads     (forward)
self._dependents = {}  # cell -> cells that read it (reverse)

Then a change to A1:

1. BFS/DFS over _dependents from A1 to collect the dirty set.
2. Topologically sort only the dirty set.
3. Recompute in that order.

The cost is O(dirty subgraph), not O(sheet). For a sheet with 100,000 cells where one edit affects 12, that is a four-order-of-magnitude difference — and it is the difference between a spreadsheet that feels instant and one that freezes.

Assert the recompute count in your tests, not just the values. A test that only checks values passes even if you recomputed everything, which means the optimization can silently regress. Counting evaluations is what actually pins the behaviour, and mentioning that you'd test it that way is a small but real signal.

Both edges must be maintained on every edit: when a formula changes, remove the cell from the _dependents of its old dependencies before adding it to the new ones. Forgetting the removal is the classic bug — stale reverse edges cause phantom recomputation that grows over time.

7.4 Complete implementation

import re
from collections import deque


class CycleError(Exception):
    def __init__(self, cycle):
        self.cycle = cycle
        super().__init__(" -> ".join(cycle))


CELL = re.compile(r"\b([A-Z]+[0-9]+)\b")
RANGE = re.compile(r"\b([A-Z]+[0-9]+):([A-Z]+[0-9]+)\b")


class Sheet:
    """Formula evaluation with cycle detection and incremental recompute."""

    WHITE, GREY, BLACK = 0, 1, 2

    def __init__(self):
        self._raw = {}          # cell -> literal or "=formula"
        self._value = {}        # cell -> computed value
        self._deps = {}         # cell -> set of cells it reads
        self._dependents = {}   # cell -> set of cells that read it
        self.evaluations = 0    # tests assert on this, not just on values

    # ---- editing ---------------------------------------------------------
    def set_cell(self, cell, raw):
        self._raw[cell] = raw

        # Detach the old forward edges from the reverse index, or stale
        # dependents accumulate and phantom recomputation grows over time.
        for old in self._deps.get(cell, ()):
            self._dependents.get(old, set()).discard(cell)

        deps = set(self._extract_refs(raw)) if str(raw).startswith("=") else set()
        self._deps[cell] = deps
        for dep in deps:
            self._dependents.setdefault(dep, set()).add(cell)

        self._recompute_from(cell)

    def _extract_refs(self, formula):
        body = formula[1:]
        refs = set()
        for start, end in RANGE.findall(body):
            refs.update(self._expand_range(start, end))
        refs.update(CELL.findall(RANGE.sub("", body)))
        return refs

    @staticmethod
    def _expand_range(start, end):
        c0, r0 = re.match(r"([A-Z]+)([0-9]+)", start).groups()
        c1, r1 = re.match(r"([A-Z]+)([0-9]+)", end).groups()
        for col in range(ord(c0), ord(c1) + 1):
            for row in range(int(r0), int(r1) + 1):
                yield f"{chr(col)}{row}"

    # ---- incremental recompute ------------------------------------------
    def _dirty_set(self, changed):
        """Everything transitively reading `changed`, via the REVERSE graph."""
        seen, queue = {changed}, deque([changed])
        while queue:
            node = queue.popleft()
            for dependent in self._dependents.get(node, ()):
                if dependent not in seen:
                    seen.add(dependent)
                    queue.append(dependent)
        return seen

    def _recompute_from(self, changed):
        dirty = self._dirty_set(changed)
        try:
            order = self._toposort(dirty)
        except CycleError as exc:
            for cell in exc.cycle:
                self._value[cell] = f"#CIRCULAR({' -> '.join(exc.cycle)})"
            return
        for cell in order:
            self._value[cell] = self._evaluate(cell)

    def _toposort(self, subset):
        color = {c: self.WHITE for c in subset}
        order, path = [], []

        def visit(node):
            state = color.get(node, self.BLACK)
            if state == self.BLACK:
                return
            if state == self.GREY:                    # back edge == cycle
                raise CycleError(path[path.index(node):] + [node])
            color[node] = self.GREY
            path.append(node)
            for dep in self._deps.get(node, ()):
                if dep in color:                      # stay inside the dirty set
                    visit(dep)
            path.pop()
            color[node] = self.BLACK
            order.append(node)

        for cell in subset:
            visit(cell)
        return order

    # ---- evaluation ------------------------------------------------------
    def _evaluate(self, cell):
        self.evaluations += 1
        raw = self._raw.get(cell, 0)
        if not str(raw).startswith("="):
            return raw
        body = raw[1:]

        def sum_range(match):
            cells = self._expand_range(match.group(1), match.group(2))
            return str(sum(self._numeric(c) for c in cells))

        body = re.sub(r"SUM\(([A-Z]+[0-9]+):([A-Z]+[0-9]+)\)", sum_range, body)
        body = CELL.sub(lambda m: str(self._numeric(m.group(1))), body)
        try:
            return eval(body, {"__builtins__": {}}, {})   # demo only — see Q&A
        except Exception:
            return "#ERROR"

    def _numeric(self, cell):
        value = self._value.get(cell, 0)
        return value if isinstance(value, (int, float)) else 0

    def get_value(self, cell):
        return self._value.get(cell, 0)

7.5 Interview Q&A

Q: How do you detect a cycle? Three-colour DFS. WHITE unvisited, GREY on the current recursion stack, BLACK fully explored. Reaching a GREY node means a back edge, which is exactly a cycle, and the path stack gives me the actual cycle to report. A two-state "visited" set can't do this: it conflates "on my current path" with "already finished", so it either misses cycles or flags legal diamonds as cycles.

Q: Kahn's algorithm or DFS? Kahn when I want parallelism — everything at in-degree zero can run concurrently, which is why build systems use it — or when I only need a yes/no on cycles. DFS when I need to report the cycle path, which a spreadsheet does, because A1 → B1 → C1 → A1 is far more useful to a user than #CIRCULAR.

Q: Why do you need the reverse graph? Because the forward graph answers "what does this cell read" and incremental recompute needs "what reads this cell". Editing A1 means walking dependents to find the dirty set, then topologically sorting only that set. Cost is O(dirty subgraph) instead of O(sheet) — for 100,000 cells where an edit affects 12, four orders of magnitude.

Q: How do you know the incremental path actually works? I count evaluations and assert on the count, not just the values. A test that only checks values passes even when you recomputed the whole sheet, so the optimization can regress silently. Counting is what pins the behaviour.

Q: You used eval. Isn't that a security hole? Yes, and in production it's disqualifying — even with __builtins__ stripped, there are known escapes through attribute traversal on literals. The correct implementation is a Pratt parser producing an AST, evaluated by an interpreter that only knows the operations I chose to implement. I used eval here to keep the chapter focused on the dependency graph, which is the part being tested, and I'd say exactly that in the interview — naming the shortcut is much better than being caught taking it.

Q: What are volatile functions and what do they break? Functions like NOW() or RAND() whose value changes without any input changing. They break the caching story: the dependency graph says nothing is dirty, but the value is stale. The standard answer is to mark them volatile and recompute them on every evaluation pass, propagating dirtiness to their dependents — which is why a sheet full of NOW() recalculates constantly and feels slow. Worth being explicit that this is a correctness/performance trade the user can observe.

Q: The graph is 10 million nodes deep. What breaks? Recursion — Python's default limit is 1000 frames, so I'd blow the stack. Convert the DFS to an explicit stack with an "enter/exit" marker per node so I can still do post-order. Kahn's algorithm is iterative by construction and sidesteps this entirely, which is another point in its favour at scale.

Chapter 8: Write-Ahead Logs and Crash Recovery

8.1 What durability actually means

"The write succeeded" is ambiguous, and the ambiguity is where data loss lives. There are four distinct places a write can be:

1. In your process's buffer — write() not yet called. A process crash loses it.
2. In the kernel's page cache — write() returned. A process crash is survivable; a machine crash (power loss, kernel panic) loses it.
3. In the device's volatile cache — fsync() returned, but the drive lied. Consumer SSDs do this. Power loss loses it.
4. On stable media — fsync() returned and the device honoured it.

write() returning tells you almost nothing about durability. It means the kernel accepted the bytes. Only fsync() (or O_DSYNC, or fdatasync) pushes toward stable media, and even then you are trusting the device.

A write-ahead log turns this into a usable guarantee with one rule:

Append the intent to the log and make it durable before mutating the main structure.

Then after a crash you replay the log. Every operation is either fully in the log (replay it) or not there at all (it never happened). There is no partial state.

The cost is unavoidable: an fsync per record is one disk round trip, roughly 0.1–1 ms on NVMe and far worse on network storage. That caps you at a few thousand durable writes per second per log.

Group commit is the standard escape. Batch the fsyncs: many writers append to the buffer, one fsync covers all of them, all of them return. Throughput goes up by the batch factor; latency goes up by at most the batch window. This is what Postgres's commit_delay and every serious database's group-commit path do, and naming it is the expected answer to "that's slow, what now?"

8.2 Record framing and torn writes

A log file is a byte stream. To read records back you must know where each one ends.

Framing: length prefix + payload + checksum.

[ 4-byte length ][ payload ][ 4-byte CRC32 ]

Now the crash case. A crash mid-write leaves a partial record at the tail: maybe 4 bytes of a length prefix, maybe half a payload. This is not an error condition to be surprised by — it is the expected state after a crash, and handling it is the whole point.

The recovery rule:

A truncated or corrupt tail is discarded, not fatal. Read records until one fails to parse; everything before it is valid; truncate the file there and continue.

Three ways the tail can be bad, all handled by the same rule:

1. Fewer than 4 bytes remain → no length prefix → stop.
2. The length prefix says N but fewer than N bytes remain → incomplete → stop.
3. The bytes are all there but the CRC doesn't match → torn or corrupted → stop.

Case 3 is the one people forget. Without a checksum, a torn write where the length happened to be complete but the payload was partially written is read back as a valid-looking record containing garbage — silent corruption, which is far worse than a crash. The CRC is what turns silent corruption into a clean truncation.

Test this by actually truncating the file at every byte offset and asserting that recovery succeeds and returns a prefix of the writes. That test finds real bugs, and describing it is a strong answer to "how do you know it works?"

8.3 Checkpointing and compaction

Replaying from the beginning of time means startup gets slower forever. So periodically write a checkpoint: a snapshot of the full state plus the log offset it corresponds to. Recovery becomes "load the newest checkpoint, replay only the log after its offset."

The dangerous part is that checkpointing must itself be crash-safe. If you crash halfway through writing a checkpoint and then trust it, you load corrupt state.

The standard technique is atomic rename:

tmp = path + ".tmp"
with open(tmp, "wb") as fh:
    fh.write(serialized)
    fh.flush()
    os.fsync(fh.fileno())      # the DATA is durable
os.replace(tmp, path)          # atomic on POSIX; either old or new, never half
dir_fd = os.open(os.path.dirname(path) or ".", os.O_DIRECTORY)
try:
    os.fsync(dir_fd)           # the DIRECTORY ENTRY is durable too
finally:
    os.close(dir_fd)

Three details, each of which is a real bug if omitted:

• fsync the file before renaming, or the rename can be durable while the contents are not.
• os.replace is atomic on POSIX — a reader sees either the old file or the new one, never a mix.
• fsync the directory afterwards, because the rename is a directory-entry change and that entry also needs to reach stable storage. This is the one almost everyone forgets, and it is a genuinely good thing to mention.

Only after the checkpoint is durable may you truncate the log prefix it covers.

8.4 Complete implementation

import os
import struct
import zlib

HEADER = struct.Struct("<I")     # 4-byte little-endian length
CRC = struct.Struct("<I")


class WriteAheadLog:
    """Append-only log with length+CRC framing and torn-tail recovery."""

    def __init__(self, path, fsync_policy="always", group_size=64):
        self.path = path
        self._fsync_policy = fsync_policy      # always | group | never
        self._group_size = group_size
        self._pending = 0
        self._file = open(path, "a+b")
        self._file.seek(0, os.SEEK_END)
        self.offset = self._file.tell()

    def append(self, payload: bytes) -> int:
        record = HEADER.pack(len(payload)) + payload + CRC.pack(zlib.crc32(payload))
        self._file.write(record)
        self.offset += len(record)
        self._pending += 1
        if self._fsync_policy == "always":
            self._durable()
        elif self._fsync_policy == "group" and self._pending >= self._group_size:
            self._durable()
        return self.offset

    def _durable(self):
        self._file.flush()
        os.fsync(self._file.fileno())          # the only call that means "durable"
        self._pending = 0

    def flush(self):
        self._durable()

    def replay(self, from_offset=0):
        """Yield every intact record. A truncated or corrupt tail ENDS the
        iteration — it is the expected post-crash state, not an error."""
        with open(self.path, "rb") as fh:
            fh.seek(from_offset)
            position = from_offset
            while True:
                head = fh.read(HEADER.size)
                if len(head) < HEADER.size:
                    break                                  # (1) no length prefix
                (length,) = HEADER.unpack(head)
                body = fh.read(length)
                if len(body) < length:
                    break                                  # (2) incomplete payload
                tail = fh.read(CRC.size)
                if len(tail) < CRC.size:
                    break
                (expected,) = CRC.unpack(tail)
                if zlib.crc32(body) != expected:
                    break                                  # (3) torn write
                position += HEADER.size + length + CRC.size
                yield position, body
            self.valid_end = position

    def truncate_to_valid(self):
        """Drop a partial tail so the next append starts clean."""
        list(self.replay())
        with open(self.path, "r+b") as fh:
            fh.truncate(self.valid_end)
        self._file.close()
        self._file = open(self.path, "a+b")
        self._file.seek(0, os.SEEK_END)
        self.offset = self._file.tell()
        return self.valid_end

    def close(self):
        self._durable()
        self._file.close()


def write_checkpoint_atomically(path: str, blob: bytes) -> None:
    """Crash-safe: either the old checkpoint or the new one, never a mix."""
    tmp = path + ".tmp"
    with open(tmp, "wb") as fh:
        fh.write(blob)
        fh.flush()
        os.fsync(fh.fileno())            # 1. the DATA is durable
    os.replace(tmp, path)                # 2. atomic on POSIX
    dir_path = os.path.dirname(os.path.abspath(path))
    dir_fd = os.open(dir_path, os.O_DIRECTORY)
    try:
        os.fsync(dir_fd)                 # 3. the DIRECTORY ENTRY is durable
    finally:
        os.close(dir_fd)

8.5 Interview Q&A

Q: What does write() actually guarantee? That the kernel accepted the bytes into its page cache. Nothing about the disk. A process crash is survivable at that point; a machine crash is not. Only fsync pushes toward stable media — and even then you're trusting the device not to lie about its volatile cache, which consumer SSDs have historically done.

Q: fsync per record is too slow. Now what? Group commit. Many writers append to the buffer, one fsync covers the batch, and they all return together. Throughput multiplies by the batch factor while latency rises by at most the batch window. That's what Postgres's commit_delay does. The alternative — fsync never — is a legitimate choice for a cache, but then say plainly that you've traded durability for throughput rather than pretending you have both.

Q: You crash halfway through a write. What's in the file? A partial record. That's expected, not exceptional. Recovery reads records until one fails to parse — no length prefix, incomplete payload, or CRC mismatch — and truncates there. Everything before it is valid. The CRC is the one people skip and it's the important one: without it a torn write whose length happened to be complete reads back as a valid-looking record full of garbage, which is silent corruption. The checksum turns that into a clean truncation.

Q: How do you test it? Write N records, then truncate the file at every byte offset from 0 to its length, and assert recovery succeeds and returns a prefix of what was written. It's a loop over offsets and it finds real bugs — particularly off-by-ones in the framing arithmetic.

Q: Replay gets slower forever. Fix it. Checkpoints: periodically serialize the full state plus the log offset it corresponds to, then truncate the log prefix that the checkpoint covers. Recovery becomes "load the newest checkpoint, replay what's after it."

Q: What if you crash while writing the checkpoint? That's why it's write-to-temp then atomic rename. Fsync the temp file so the data is durable, os.replace so a reader sees either the old checkpoint or the new one and never a mix, then fsync the directory — because the rename is a directory-entry change and that entry needs to reach stable storage too. That last fsync is the one almost everyone forgets, and without it you can lose the rename on power loss and come back to the old checkpoint.

Q: Log versus LSM tree — what's the relationship? An LSM tree is this idea taken all the way: the log is the database. Writes append to a memtable backed by a WAL; the memtable is flushed to an immutable sorted file; background compaction merges files. That's why LSMs are write-optimized — every write is sequential — and why they pay for it on reads, which may have to check several levels, mitigated by Bloom filters per file. B-trees make the opposite trade: in-place updates give good reads and random writes.

Chapter 9: Deduplication and Probabilistic Structures

9.1 The exactly-once illusion

Start with the honest claim, because interviewers ask this specifically to see whether you will overclaim:

Exactly-once delivery over a network is impossible. Exactly-once processing is achievable, and the mechanism is at-least-once delivery plus idempotent consumers.

The impossibility is not a limitation of any protocol. Sender sends, receiver processes, ack is lost. The sender cannot distinguish "the receiver never got it" from "the receiver got it and the ack was lost". Its only options are resend (risking a duplicate) or not resend (risking a loss). No amount of extra round trips removes this — the same argument applies to the ack of the ack. This is the Two Generals Problem.

So: deliver at least once, and make the consumer's effect idempotent. The consumer keeps a record of processed IDs and skips repeats.

The requirement that follows is a stable idempotency key: an identifier generated by the producer, attached to the message, and unchanged across retries. If the key is generated at send time, every retry has a different key and dedupe cannot work — a real and common bug, and exactly why Stripe's API requires the client to supply the Idempotency-Key header.

9.2 Bloom filters, derived

The exact dedupe set is unbounded: to remember every ID forever you need memory proportional to every ID forever. Two bounded options.

Windowed exact dedupe. Keep IDs seen in the last T. Bounded by rate × T. Correct within the window; a duplicate arriving later than T gets through. This is the honest trade and usually the right one, because retries happen in seconds, not days.

Probabilistic dedupe: a Bloom filter. Constant memory, no matter how many items.

The mechanism, from zero: a bit array of m bits, all zero, and k independent hash functions.

• Insert x: set the bits at positions h₁(x) % m, …, h_k(x) % m.
• Query x: if all those bits are 1, report "possibly present". If any is 0, report "definitely absent".

"Definitely absent" is exact: if x had been inserted, all its bits would be set. "Possibly present" can be wrong, because other insertions may have set exactly those bits by coincidence.

The false-positive rate, derived. After inserting n items with k hashes into m bits, the probability that one specific bit is still 0 is:

P(bit still 0) = (1 - 1/m)^(kn) ≈ e^(-kn/m)

A false positive requires all k of a query's bits to be 1:

FPR ≈ (1 - e^(-kn/m))^k

Differentiate with respect to k and the optimum is:

k* = (m/n) · ln 2 ≈ 0.693 · m/n

and at that k, FPR ≈ 0.6185^(m/n). So to size one: pick your target FPR, solve for bits per item.

10 bits per item — about 1.25 bytes — for under 1% error. Compare with a Python set of UUID strings at roughly 100+ bytes per entry. Two orders of magnitude, and that is why Bloom filters are in every LSM engine, CDN, and crawler.

The limitation to state: a standard Bloom filter cannot delete. Clearing bits would break other items that share them. Counting Bloom filters (counters instead of bits) support deletion at 4× the space; cuckoo filters support it more efficiently and also give better locality, at the cost of a more complex insert path that can fail.

9.3 Which direction the error points

This is the question that separates people who have read about Bloom filters from people who have used them.

A Bloom filter has false positives, never false negatives. It can say "probably seen" about something new; it can never say "not seen" about something it has seen.

Now apply that to dedupe:

A false positive means "I think I've seen this" about a message you have not seen. A dedupe filter would therefore drop a real, unprocessed message. Silently.

That is data loss. Whether it is acceptable depends entirely on the workload, and you must say so rather than treating the FPR as a generic quality knob:

• Analytics counting? Fine. Losing 0.1% of events barely moves an aggregate.
• Payment processing? Absolutely not. Losing one payment in a thousand is a company-ending bug.

The design that fixes it: use the Bloom filter as a negative cache in front of an exact store. "Definitely absent" — which is exact — means process immediately, no lookup needed. "Possibly present" means go check the authoritative store. Now the filter eliminates the vast majority of expensive lookups while the exact store guarantees correctness, and the FPR costs you extra lookups rather than lost data.

That is exactly how an LSM engine uses per-file Bloom filters to avoid reading files that cannot contain a key, and describing it that way lands well.

9.4 Complete implementation

import hashlib
import math
import time
from collections import deque


class BloomFilter:
    """Constant memory. False positives, never false negatives."""

    def __init__(self, capacity, error_rate=0.01):
        if not 0 < error_rate < 1:
            raise ValueError("error_rate must be in (0, 1)")
        # m = -n ln(p) / (ln 2)^2      k = (m/n) ln 2
        self.capacity = capacity
        self.error_rate = error_rate
        self.m = max(8, int(math.ceil(-capacity * math.log(error_rate) / (math.log(2) ** 2))))
        self.k = max(1, int(round(self.m / capacity * math.log(2))))
        self._bits = bytearray((self.m + 7) // 8)
        self.count = 0

    def _positions(self, item):
        # Kirsch-Mitzenmacher: two independent hashes simulate k of them, so
        # you pay for two digests instead of k.
        data = item.encode() if isinstance(item, str) else bytes(item)
        digest = hashlib.blake2b(data, digest_size=16).digest()
        h1 = int.from_bytes(digest[:8], "little")
        h2 = int.from_bytes(digest[8:], "little") | 1     # odd -> full period
        for i in range(self.k):
            yield (h1 + i * h2) % self.m

    def add(self, item):
        for pos in self._positions(item):
            self._bits[pos >> 3] |= 1 << (pos & 7)
        self.count += 1

    def __contains__(self, item):
        return all(self._bits[p >> 3] & (1 << (p & 7)) for p in self._positions(item))

    def current_fpr(self):
        """Actual FPR at the current fill — it degrades as you overfill."""
        return (1 - math.exp(-self.k * self.count / self.m)) ** self.k

    def stats(self):
        set_bits = sum(bin(b).count("1") for b in self._bits)
        return {"m_bits": self.m, "k": self.k, "items": self.count,
                "bytes": len(self._bits), "fill": set_bits / self.m,
                "fpr_now": self.current_fpr(),
                "bytes_per_item": len(self._bits) / max(self.count, 1)}


class WindowedDedupe:
    """Exact within the window. Memory bounded by rate x window."""

    def __init__(self, window_seconds, clock=time.monotonic):
        self.window = window_seconds
        self._clock = clock
        self._seen = {}                    # key -> timestamp
        self._order = deque()              # (timestamp, key) in arrival order
        self.duplicates = 0

    def _evict(self, now):
        cutoff = now - self.window
        while self._order and self._order[0][0] <= cutoff:
            _, key = self._order.popleft()
            self._seen.pop(key, None)

    def is_duplicate(self, key):
        now = self._clock()
        self._evict(now)
        if key in self._seen:
            self.duplicates += 1
            return True
        self._seen[key] = now
        self._order.append((now, key))
        return False


class SafeDedupe:
    """Bloom as a NEGATIVE CACHE in front of an exact store.

    'Definitely absent' is exact -> process with no lookup.
    'Possibly present'  -> consult the authoritative store.
    So the filter saves lookups; it never loses a message.
    """

    def __init__(self, capacity, exact_store, error_rate=0.01):
        self._bloom = BloomFilter(capacity, error_rate)
        self._exact = exact_store          # set-like: __contains__ and add
        self.lookups_avoided = 0
        self.lookups_performed = 0

    def is_duplicate(self, key):
        if key not in self._bloom:         # exact answer: definitely new
            self.lookups_avoided += 1
            self._bloom.add(key)
            self._exact.add(key)
            return False
        self.lookups_performed += 1        # maybe: must check authoritatively
        if key in self._exact:
            return True
        self._bloom.add(key)
        self._exact.add(key)
        return False

9.5 Interview Q&A

Q: Can you guarantee exactly-once delivery? No, and nobody can. Sender sends, receiver processes, ack is lost — the sender cannot distinguish "never arrived" from "arrived and the ack was lost", and adding round trips just moves the problem to the ack of the ack. That's the Two Generals Problem. What's achievable is exactly-once processing: at-least-once delivery plus an idempotent consumer. The key detail is that the idempotency key must be generated by the producer and stay stable across retries — if it's generated at send time, every retry has a new key and dedupe silently does nothing.

Q: Unbounded dedupe set. Bound it. Windowed exact dedupe: keep IDs from the last T, memory bounded by rate × T. It's exact within the window and a duplicate arriving later gets through — which is fine, because retries happen in seconds, not days. State the window and the assumption behind it.

Q: Bloom filter — how do you size it? m = -n ln(p) / (ln 2)² bits and k = (m/n) ln 2 hashes. About 10 bits per item — 1.25 bytes — gives under 1% false positives; 16 bits gives 0.05%. Compare to a Python set of UUID strings at 100+ bytes per entry.

Q: Which direction does the error go, and does it matter? False positives, never false negatives — and for dedupe that's the dangerous direction. A false positive means "I think I've seen this" about a message you haven't, so the filter drops a real message, silently. For analytics counting that's fine; for payments it's company-ending. So I use the Bloom filter as a negative cache in front of an exact store: "definitely absent" is exact and processes with no lookup, "possibly present" consults the authoritative store. The filter then saves lookups instead of losing data — which is exactly how an LSM engine uses per-file Blooms to skip files that can't contain a key.

Q: What about deletion? A standard Bloom can't delete — clearing bits would break other items sharing them. Counting Bloom filters use small counters instead of bits and support deletion at about 4× the space. Cuckoo filters do it more space-efficiently and have better cache locality, at the cost of an insert path that can fail and require a rebuild. For a sliding window I'd use rotating Bloom filters instead: two or three generations, retire the oldest, which approximates deletion without any of that machinery.

Q: What happens if you overfill it? The FPR degrades continuously and silently — nothing raises. That's why I expose current_fpr() and would alarm on it. An overfilled Bloom filter eventually returns "possibly present" for everything, at which point it's doing no work and you've lost the optimization without noticing.

Q: Out-of-order messages? Dedupe by key handles duplicates but not ordering. For ordering I'd use per-key sequence numbers and a bounded reorder buffer: hold out-of-order arrivals up to a window, emit in sequence, and after the window give up and emit with a gap marker. Unbounded reordering means unbounded memory, so the window is not optional — and the gap marker matters because a silent gap is a correctness bug the consumer can't see.

Chapter 10: Backpressure and Bounded Concurrency

10.1 The unbounded queue is a memory leak

Producer feeds a queue, consumer drains it. Consumer is slower. What happens?

With an unbounded queue: the queue grows. Latency grows with it — by Little's law, wait time is queue length divided by service rate, so a queue of 10,000 items served at 100/s means every new item waits 100 seconds. Then memory runs out and the process dies, losing everything queued.

The important reframing:

An unbounded queue does not absorb overload. It converts a throughput problem into a latency problem, and then into an out-of-memory crash.

By the time you notice, every item in the queue is already too old to be useful. Worse, the crash loses in-flight work that a rejection would have let the client retry.

A bounded queue makes the producer feel the consumer's slowness. When it is full, put blocks (or fails). That is backpressure: the signal propagates upstream to whoever can actually do something about it — slow down, shed load, or scale out.

10.2 Four responses to too much work

When work arrives faster than you can serve it, there are exactly four things you can do, and a good design says which and why.

Shedding is a feature, not a failure. Rejecting 10% of requests in 1 ms so the other 90% meet their SLO is strictly better than accepting 100% and having all of them time out at 30 seconds. Everyone loses in the second case, and you burn 30 seconds of capacity per doomed request.

The reason is the utilization/latency curve. For an M/M/1 queue, response time scales as 1/(1-ρ):

Latency is hyperbolic in utilization, not linear. That is why a system at 85% looks comfortable on a dashboard and falls over at 92%, and it is the quantitative argument for admission control. Real traffic is burstier than Poisson, so the true knee arrives earlier than this table suggests, not later.

Which to shed matters too. Shedding the oldest queued item is usually right — it is the one most likely to have already timed out on the client side, so serving it is pure waste. This is sometimes called LIFO-under-load, and it is counter-intuitive until you notice that FIFO under overload serves nothing but requests nobody is waiting for any more.

10.3 Graceful shutdown

Shutdown is where concurrency bugs live, because it is the least-tested path.

Requirements for a correct shutdown:

1. Stop accepting new work — immediately.
2. Finish in-flight work — up to a deadline.
3. Do not lose queued work — either drain it or persist it.
4. Be idempotent — shutdown may be called twice, or while already shutting down.
5. Have a hard deadline — after which you cancel, because a hang is worse than a loss.

Two mechanisms, with different properties:

Sentinel — push a None (one per consumer) into the queue. Consumers exit on seeing it.

• Naturally drains: everything queued before the sentinel is processed.
• Requires knowing the consumer count, and does not interrupt a consumer blocked on I/O.

Cancellation — cancel the consumer tasks.

• Immediate, and interrupts blocked I/O.
• Loses queued work unless you drain first.

The production answer is usually both: sentinel to drain, then a timeout, then cancellation as the hard stop.

And the async-specific rule that must be respected: asyncio.CancelledError inherits from BaseException, not Exception. A blanket except Exception: will not swallow it — which is deliberate. If you catch it explicitly for cleanup you must re-raise, or you have made the task uncancellable and your shutdown deadline becomes a hang.

10.4 Complete implementation

import asyncio
import time


class BoundedPipeline:
    """Bounded queue + worker pool + real backpressure, load shedding, and a
    shutdown that drains before it cancels."""

    def __init__(self, handler, *, workers=4, max_queue=100,
                 item_timeout=5.0, shed_after=0.25):
        self._handler = handler
        self._queue = asyncio.Queue(maxsize=max_queue)   # the bound IS the backpressure
        self._n_workers = workers
        self._item_timeout = item_timeout
        self._shed_after = shed_after
        self._workers = []
        self._running = False
        self.stats = {"accepted": 0, "shed": 0, "done": 0,
                      "failed": 0, "timeout": 0}

    async def start(self):
        if self._running:
            return
        self._running = True
        self._workers = [asyncio.create_task(self._worker(i))
                         for i in range(self._n_workers)]

    async def submit(self, item, *, block=True):
        """Backpressure when block=True; load shedding when False."""
        if not self._running:
            raise RuntimeError("pipeline is not running")
        if block:
            try:
                # Wait a bounded time, then shed. Waiting forever converts a
                # throughput problem into an unbounded latency problem.
                await asyncio.wait_for(self._queue.put(item), self._shed_after)
            except (asyncio.TimeoutError, TimeoutError):
                self.stats["shed"] += 1
                raise OverflowError("queue full — shedding") from None
        else:
            try:
                self._queue.put_nowait(item)
            except asyncio.QueueFull:
                self.stats["shed"] += 1
                raise OverflowError("queue full — shedding") from None
        self.stats["accepted"] += 1

    async def _worker(self, index):
        while True:
            item = await self._queue.get()
            try:
                if item is None:                    # sentinel: drain complete
                    return
                try:
                    await asyncio.wait_for(self._handler(item), self._item_timeout)
                    self.stats["done"] += 1
                except (asyncio.TimeoutError, TimeoutError):
                    self.stats["timeout"] += 1
                except asyncio.CancelledError:
                    # BaseException, not Exception. Observe it for cleanup and
                    # RE-RAISE — swallowing it makes the task uncancellable.
                    raise
                except Exception:
                    self.stats["failed"] += 1
            finally:
                self._queue.task_done()

    async def shutdown(self, drain_timeout=10.0):
        """Idempotent. Drains, then cancels at a hard deadline."""
        if not self._running:
            return self.stats
        self._running = False                        # 1. stop accepting

        for _ in self._workers:                      # 2. one sentinel per worker
            await self._queue.put(None)

        done, pending = await asyncio.wait(self._workers, timeout=drain_timeout)
        for task in pending:                         # 3. hard deadline
            task.cancel()
        if pending:
            await asyncio.gather(*pending, return_exceptions=True)
        self._workers.clear()
        return self.stats


class AdaptiveConcurrency:
    """AIMD concurrency limit — additive increase, multiplicative decrease.
    The same control law as TCP congestion control, and for the same reason:
    the right limit is discovered from feedback, not configured."""

    def __init__(self, initial=10, minimum=1, maximum=200, target_latency=0.1):
        self.limit = float(initial)
        self.minimum, self.maximum = minimum, maximum
        self.target = target_latency
        self._inflight = 0

    def try_acquire(self):
        if self._inflight >= int(self.limit):
            return False
        self._inflight += 1
        return True

    def release(self, latency, failed=False):
        self._inflight -= 1
        if failed or latency > self.target * 2:
            self.limit = max(self.minimum, self.limit * 0.8)      # back off hard
        elif latency < self.target:
            self.limit = min(self.maximum, self.limit + 1.0)      # probe gently

10.5 Interview Q&A

Q: Why bound the queue? Because an unbounded queue doesn't absorb overload — it converts a throughput problem into a latency problem and then into an OOM crash. By Little's law, 10,000 queued items served at 100/s means every new item waits 100 seconds, so by the time you notice, everything in the queue is already useless. And the crash loses in-flight work that a rejection would have let the client retry. The bound is what makes the producer feel the consumer's slowness.

Q: Queue is full. Block or reject? Depends on who the producer is. Internal pipeline where the producer can slow down: block — that's backpressure and it propagates the signal to someone who can act on it. External client that can retry: reject fast with a 429 and a Retry-After. What I wouldn't do is block indefinitely on an external request, because that turns a bounded queue back into an unbounded one — the queue is now the clients' connection pool.

Q: Isn't shedding a failure? It's a feature. Rejecting 10% in a millisecond so the other 90% meet SLO beats accepting 100% and timing all of them out at 30 seconds — in the second case everyone loses and you burned 30 seconds of capacity per doomed request. The quantitative argument is the M/M/1 response curve: 1/(1-ρ), so 80% utilization is 5× service time and 95% is 20×. Latency is hyperbolic in utilization, which is why a system at 85% looks fine on a dashboard and falls over at 92%.

Q: Which item do you shed? Usually the oldest queued one, because it's the most likely to have already timed out on the client side, so serving it is pure waste. That's counter-intuitive — it looks unfair — until you notice that FIFO under sustained overload serves nothing but requests nobody is waiting for any more.

Q: How do you shut down without losing work? Stop accepting, push one sentinel per worker so everything already queued still drains, wait with a deadline, then cancel whatever is left. Make it idempotent because shutdown gets called twice. And the async-specific rule: CancelledError is a BaseException, not an Exception, so a blanket except Exception won't swallow it — and if I catch it explicitly for cleanup I must re-raise, or the task becomes uncancellable and my hard deadline turns into a hang.

Q: How do you pick the concurrency limit? I'd rather not pick it. A static limit is either too low (wasted capacity) or too high (overload), and the right value changes with downstream health. AIMD — additive increase, multiplicative decrease — discovers it from feedback: raise the limit by one when latency is good, cut it 20% on a failure or a latency spike. That's TCP congestion control's law, applied to application concurrency, and it's what Netflix's concurrency-limits library does. Failing that, Little's law gives a starting point: concurrency = target_throughput × latency.

Q: gather vs TaskGroup for the workers? TaskGroup, and not as a style preference. gather propagates the first exception to the awaiter but leaves the sibling tasks running — orphaned, holding connections, writing to stores you thought you'd rolled back. TaskGroup cancels the siblings and raises an ExceptionGroup you handle with except*. That's structured concurrency: no task outlives its scope, so gather's behaviour is a resource leak rather than a different flavour.

The Complexity Table

Know these cold. Being asked "what's the complexity?" and pausing is a bad look; the answer should be immediate.

The four sentences worth memorizing:

• Hash maps answer "exactly", never "nearest".
• Heaps maintain only the partial order you need — that's why they beat sorted lists for "next".
• O(1) removal from a list requires knowing the node and double links.
• Append-only data is sorted for free if the key is monotonic.

The Thirty Questions

Ask yourself these before any Track A mock. If any answer takes more than fifteen seconds, that is your next study item.

Structures

1. Why can't a hash map answer "the value as of version V"?
2. Write bisect_right and state its loop invariant.
3. Why must an LRU's list be doubly linked?
4. What do sentinel nodes buy you?
5. Why does a heap beat a sorted list for a scheduler?
6. What breaks if you push (fire_at, job) into a heap?
7. Why do you need three colours for cycle detection, not two?
8. When is a trie better than a hash map?
9. Why is an append-only version list sorted for free?
10. What's the difference between a B-tree and an LSM tree, in one sentence?

Semantics 11. Why is delete a tombstone rather than a removal? 12. Why does deleting an absent key consume a version? 13. Global vs per-key versions — which, and what does the loser cost? 14. What anomaly does snapshot isolation permit, and what's the example? 15. Why must a transaction's writes share one version? 16. Why can't you guarantee exactly-once delivery? 17. What must be true of an idempotency key? 18. Fixed rate vs fixed delay — what goes wrong with each? 19. What does write() guarantee? What does fsync guarantee? 20. Why fsync the directory after a rename?

Trade-offs 21. Sliding window log vs counter — what do you trade? 22. Token bucket vs leaky bucket — when does each lose? 23. Lazy vs sampled vs active expiry — why is lazy alone unshippable? 24. Which direction does a Bloom filter's error go, and why does that matter for dedupe? 25. Backpressure vs buffering vs shedding vs degrading — pick one and defend it. 26. Why is shedding the oldest queued item usually right? 27. Why is jitter not sufficient on its own? 28. Why is gather orphaning siblings a bug rather than a style choice? 29. When is LRU the wrong eviction policy? 30. What's the response-time multiplier at 90% utilization, and why does that matter?

References

Books

• Kleppmann, M. Designing Data-Intensive Applications, 2nd ed. O'Reilly. — Ch. 3 (LSM vs B-tree, WAL), Ch. 7 (snapshot isolation, write skew, SSI), Ch. 11 (exactly-once, idempotence)
• Cormen, Leiserson, Rivest, Stein. Introduction to Algorithms, 4th ed. — heaps (Ch. 6), topological sort and DFS colours (Ch. 20)
• Sedgewick, R. and Wayne, K. Algorithms, 4th ed. — the structures, with clean implementations
• Beyer et al. Site Reliability Engineering. O'Reilly, 2016. — Ch. 21 (handling overload), Ch. 22 (cascading failures)
• Ramalho, L. Fluent Python, 2nd ed. — iterators, generators, and the data model behind several of these

Papers and primary sources

• Bloom, B. Space/Time Trade-offs in Hash Coding with Allowable Errors. CACM, 1970.
• Kirsch, A. and Mitzenmacher, M. Less Hashing, Same Performance: Building a Better Bloom Filter. ESA 2006 — the two-hash trick used above.
• Fan, B. et al. Cuckoo Filter: Practically Better Than Bloom. CoNEXT 2014.
• Ports, D. and Grittner, K. Serializable Snapshot Isolation in PostgreSQL. VLDB 2012.
• Berenson et al. A Critique of ANSI SQL Isolation Levels. SIGMOD 1995 — where write skew is named.
• O'Neil et al. The Log-Structured Merge-Tree (LSM-Tree). Acta Informatica, 1996.
• Myers, E. An O(ND) Difference Algorithm and Its Variations. Algorithmica, 1986 — the diff you cannot use online.
• Vandevoorde & Roberts / Chandra et al. — "Two Generals" formalizations; see also Gray, J. Notes on Data Base Operating Systems (1978) for the origin of the impossibility argument.

Engineering writing

• Brooker, M. Exponential Backoff and Jitter. AWS Architecture Blog. https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/
• Amazon Builders' Library. Timeouts, retries, and backoff with jitter. https://aws.amazon.com/builders-library/timeouts-retries-and-backoff-with-jitter/
• Cloudflare. How we built rate limiting capable of scaling to millions of domains. — the sliding-window-counter error measurement
• Netflix. Performance Under Load (concurrency-limits / AIMD). https://netflixtechblog.medium.com/performance-under-load-3e6fa9a60581
• Vattani, Chierichetti, Lowenstein. Optimal Probabilistic Cache Stampede Prevention. VLDB 2015 — XFetch.
• Redis documentation, Key eviction — the lazy + sampled expiry design.
• PostgreSQL documentation, Routine Vacuuming — reachability GC and the long-transaction bloat failure.

In this repo

• README.md — Track A drills, failure modes, and rubric
• harness/progressive.py — the gated practice harness
• harness/problems/versioned_kv/ — Chapter 1 as a timed problem
• harness/problems/token_stream_differ/ — Chapter 2 as a timed problem
• ../python-internals/WARMUP.md — the runtime behaviour behind these implementations
• ../systems-design/WARMUP.md — what each of these becomes when distributed