Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
A function can pass its first test and still fail when the next requirement arrives. Imagine you're asked to collect pages from a documentation site without visiting a page twice. Before you finish, a reviewer adds parallel workers, a per-host request limit, and a stop button. The round tests whether you can extend correct behavior without losing track of who owns each piece of state.
A CI-docs crawler makes that problem concrete: it has a queue of pages to fetch and a set of pages already claimed. Start there, then reuse the distinction between stored state and permitted transitions in a TTL store, scheduler, rate limiter, ledger, and parser. These examples use Python 3.10+ and the standard library. They run locally without network requests.
As of October 10, 2026, OpenAI's interview guide emphasizes solution design, code quality, performance, tests, and communication. It also says formats and permission to use AI tools vary by assessment.[1] Anthropic's careers page describes live coding with Colab and CodeSignal and permits reference lookups.[2] These are preparation exercises, not reported questions or a universal lab format. Check your own instructions before using documentation, autocomplete, or AI assistance.
Keep one invariant visible
Start by stating the contract in one breath: inputs, outputs, malformed input, failure behavior, and ordering. Then build the smallest state model that can satisfy that contract. A first version with one clear transition gives you something concrete to test and extend.
When a follow-up arrives, pause long enough to name the new state and the old invariant it must not break. Add a focused edge test before changing the implementation. If the follow-up introduces workers, identify the claim point and protect only the shared mutation; slow fetches and task execution should stay outside that critical section.
Close by naming the complexity, race risks, and one hardening step you would take with more time. The code that survives a staged round is the version whose visited set, TTL record, or ledger key you can still defend after the next requirement lands.
For a timed practice session, implement the first contract before adding a follow-up. Leave time to run a failing case and explain complexity. A working baseline is easier to extend than a half-written general framework.
Python tools to know cold
You don't need every API memorized. You need to recognize the state shape quickly: a deque owns FIFO work, a heap owns priority, an injected clock owns time decisions, and a lock owns a small shared transition. These are the standard-library choices that keep the conversation precise:
| Need | Python building block |
|---|---|
| FIFO work queue | collections.deque, queue.Queue, asyncio.Queue |
| Counts and top errors | collections.Counter |
| LRU cache | collections.OrderedDict |
| Deadlines and TTL | time.monotonic, injected now function |
| Priority scheduling | heapq, queue.PriorityQueue |
| Thread safety | threading.Lock, threading.RLock, threading.Condition, threading.Event |
| Worker fanout | concurrent.futures.ThreadPoolExecutor, as_completed |
| Parsing | splitlines, re, explicit state machines |
Use a small run_tests() with plain assert when a test framework would slow the round down. Keep each assertion tied to one contract boundary, so a failure tells you which state transition to inspect.
Why inject now instead of calling time.sleep() in a TTL or rate-limit interview test?
Answer
Injected time makes edge cases deterministic. You can test expiry, refill, and retry-after math immediately instead of slowing the interview down or creating flaky tests.
A crawler with an explicit URL policy
Given https://repo.example/start and get_links(url) -> list[str], return URLs in breadth-first order, preserving the callback's link order. Fetch each normalized URL at most once. An invalid start returns an empty list; invalid or out-of-scope links are skipped; a callback exception propagates. Determinism depends on a deterministic callback.
“Same host” leaves scheme and port ambiguous. This version chooses same origin: scheme, lowercase hostname, and effective port must match. It removes fragments and default ports, maps an empty path to /, and preserves query strings and nonempty paths. Thus /ci#install and /ci share a fetch, but /ci?version=2 is a different URL. HTTP and HTTPS origins differ. These are policies to clarify before coding, not properties a visited set can decide.
Predict the first fixture: /tests should become https://repo.example/tests, /ci should follow it, and the off-host URL should disappear. The invariant that makes that prediction true is simple: after host validation, claim a URL in visited before putting it in the queue or fetching it. Relative links need urljoin; the claim point must happen before slow work.
from collections import deque
from urllib.parse import urlsplit, urljoin
PAGES = {
"https://repo.example/start": ["/tests", "/ci#install", "https://other.example/x"],
"https://repo.example/tests": ["/ci"],
"https://repo.example/ci": [],
}
def normalize(raw: str, base: str = "") -> str | None:
if not isinstance(raw, str) or any(c.isspace() or ord(c) < 32 for c in raw) or "\\" in raw:
return None
try:
parts = urlsplit(urljoin(base, raw))
if parts.scheme not in {"http", "https"} or not parts.hostname:
return None
if parts.username is not None or parts.password is not None:
return None
port = parts.port # Invalid or out-of-range ports raise ValueError.
host = parts.hostname
host = f"[{host}]" if ":" in host else host
default = 443 if parts.scheme == "https" else 80
authority = host if port is None or port == default else f"{host}:{port}"
return parts._replace(netloc=authority, path=parts.path or "/", fragment="").geturl()
except ValueError:
return None
def crawl(start_url: str, get_links) -> list[str]:
start_url = normalize(start_url)
if start_url is None:
return []
start = urlsplit(start_url)
visited = {start_url}
queue = deque([start_url])
ordered: list[str] = []
while queue:
url = queue.popleft()
ordered.append(url)
for raw_link in get_links(url):
candidate = normalize(raw_link, url)
if candidate is None:
continue
parsed = urlsplit(candidate)
if (parsed.scheme, parsed.netloc) != (start.scheme, start.netloc):
continue
if candidate in visited:
continue
visited.add(candidate)
queue.append(candidate)
return ordered
visited_order = crawl("https://repo.example/start", lambda url: PAGES.get(url, []))
print(visited_order)
assert visited_order == [
"https://repo.example/start",
"https://repo.example/tests",
"https://repo.example/ci",
]
assert normalize("https://REPO.example:443/ci#x") == "https://repo.example/ci"
assert normalize("https://[broken") is None
assert normalize("https://repo.example:99999/") is None
assert crawl("ftp://repo.example/", lambda _: []) == []['https://repo.example/start', 'https://repo.example/tests', 'https://repo.example/ci']/tests is enqueued before /ci, so BFS returns start, tests, then ci. When /tests points at /ci, that normalized URL is already claimed. The queue determines order; visited prevents duplicate admission. For a finite graph, traversal takes expected O(V + E) set/queue operations and O(V) retained state, excluding URL-string processing and callback costs.
The normalizer implements this exercise's rules, not complete URL validation. It doesn't merge percent-encoded aliases or resolve DNS. Python explicitly warns that URL parsing isn't validation. A real fetcher also needs redirect checks, network-address restrictions, timeouts, and response-size limits; a same-origin string comparison isn't an SSRF defense.[3]
Now add two workers and ask what happens if both see /ci before either fetch finishes. A fetch-then-mark implementation lets both workers call get_links; the single-threaded invariant no longer protects you. Move only the claim into a short critical section.
Protect the claim step, not the whole fetch. Concurrent completion order won't generally equal BFS order. If exact BFS output is required, process one frontier at a time and merge child lists in parent order, or keep separate ordering metadata. A global fetch lock removes concurrency; sorting final URLs produces lexical order, not BFS.
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier, Lock
visited = {"https://repo.example/start"}
visited_lock = Lock()
def claim(candidate: str) -> bool:
with visited_lock:
if candidate in visited:
return False
visited.add(candidate)
return True
print(claim("https://repo.example/ci"))
print(claim("https://repo.example/ci"))
print(sorted(visited))
# A fresh URL lets two actual workers compete for the same transition.
start = Barrier(2)
def competing_claim():
start.wait()
return claim("https://repo.example/docs")
with ThreadPoolExecutor(max_workers=2) as pool:
futures = [pool.submit(competing_claim) for _ in range(2)]
results = sorted(future.result() for future in futures)
assert results == [False, True]
print("concurrent claims:", results)True
False
['https://repo.example/ci', 'https://repo.example/start']
concurrent claims: [False, True]The barrier starts two actual workers at the same claim boundary. Exactly one claims /docs; the test deliberately avoids requiring worker A or B to win. Removing the lock may still happen to pass on one schedule, so one successful run is evidence of this fixture, not proof that an unlocked check-and-add is safe.
Once claims are safe, add a per-host token bucket or next-allowed timestamp at the boundary before fetch. That state answers a different question from visited: may this host be contacted now? Cancellation is another boundary, not a new lock. Check a threading.Event before scheduling work, before fetching, and after fetching before enqueueing children. Say the three responsibilities separately: duplicate prevention, bounded fetch concurrency, and a cooperative stop.
For an at-most-one-attempt contract, a failed URL stays claimed intentionally. Retries require additional state:
| State | Meaning | Retry? |
|---|---|---|
in_flight | Claimed, fetch not finished | No second worker |
visited / success | Content processed | No |
retry_wait | Retryable failure with attempts remaining | When backoff expires |
failed | Permanent failure or exhausted attempts | No |
If the interviewer asks about failed fetches, keep the claim point and split its outcomes. A claimed URL can be in flight, succeed, or fail under a retry policy. A production sketch keeps visited for terminal success, in_flight for active work, and a retry queue with attempt counts, so a timeout doesn't become permanent silence.
What invariant should you say before adding threads to a crawler, scheduler, or cache?
Answer
Name the shared state and the exact claim point. For a crawler, a URL is inserted into visited while holding the lock before any worker can fetch or enqueue it.
The crawler claims work before starting it. A store asks a related question at a time boundary: which records still exist when a read begins?
TTL key/value store
Implement an in-memory string store with set(key, value, ttl=None), get(key), delete(key), and scan(prefix). Time-to-live (TTL) is the number of seconds a value remains readable. Expired keys behave as missing, zero TTL expires immediately, and negative or non-finite TTLs are invalid. None means no expiration and is reserved as the missing-value result, not a stored value.
Clarify the contract before choosing fields. Should scan(prefix) return keys, values, or pairs? Is equality at the deadline expired? Should compare-and-set treat an expired key as missing? Those answers decide whether expiry cleanup belongs only in get or in every read path.
Predict the boundary before code: a key written at time 10.0 with ttl=5.0 should exist at 14.999 and be missing at 15.0. Inject now so the test can jump there. Store expires_at beside each value, and keep this invariant visible: every public read either returns a non-expired value or removes the expired key. This store treats equality as expired (expires_at <= now).
from dataclasses import dataclass
from typing import Callable
import math
@dataclass
class Entry:
value: str
expires_at: float | None
class Store:
def __init__(self, now: Callable[[], float]) -> None:
self.now = now
self.items: dict[str, Entry] = {}
def set(self, key: str, value: str, ttl: float | None = None) -> None:
if not isinstance(key, str) or not isinstance(value, str):
raise TypeError("keys and values must be strings")
if ttl is not None and (not math.isfinite(ttl) or ttl < 0):
raise ValueError("ttl must be nonnegative and finite")
expires_at = None if ttl is None else self.now() + ttl
self.items[key] = Entry(value, expires_at)
def get(self, key: str) -> str | None:
return self._get_at(key, self.now())
def _get_at(self, key: str, now: float) -> str | None:
entry = self.items.get(key)
if entry is None:
return None
if entry.expires_at is not None and entry.expires_at <= now:
self.items.pop(key, None)
return None
return entry.value
def delete(self, key: str) -> None:
self.items.pop(key, None)
def scan(self, prefix: str) -> list[str]:
now = self.now()
return sorted(
key
for key in list(self.items)
if key.startswith(prefix) and self._get_at(key, now) is not None
)
clock = {"now": 10.0}
store = Store(lambda: clock["now"])
store.set("task:1", "running", ttl=5.0)
store.set("task:2", "ready")
print(store.get("task:1"), store.scan("task:"))
clock["now"] = 14.999
assert store.get("task:1") == "running"
clock["now"] = 15.0
print(store.get("task:1"), store.scan("task:"))
store.set("task:0", "gone", ttl=0)
assert store.get("task:0") is None
store.set("task:3", "old", ttl=1)
store.set("task:3", "replacement")
clock["now"] = 20
assert store.get("task:3") == "replacement"
try:
store.set("task:none", None)
except TypeError as error:
print(error)
assert "task:none" not in store.itemsrunning ['task:1', 'task:2']
None ['task:2']
keys and values must be stringsBoth reads use _get_at, so expiry isn't a get-only side path. scan samples time once: every result belongs to the same logical instant. It copies the keys because cleanup can delete entries while iterating. For n stored keys and k matches, scanning costs O(n + k log k) plus prefix comparisons; ordinary key operations are expected O(1) excluding string costs.
Why reject None on write? Without that check, a stored record could return None and disappear from scan while still occupying items. The reserved missing-value result would no longer distinguish absence from a stored value. Type annotations alone don't enforce this contract.[4]
The injected clock must return finite, nondecreasing seconds, such as time.monotonic. These process-relative deadlines aren't portable timestamps for disk storage. Lazy cleanup also leaves unread expired entries in memory. A bounded service needs an eviction or cleanup policy; this single-threaded exercise doesn't claim one.
Compare-and-set and locking
Compare-and-set adds a second boundary: it must clean any expired value and compare while holding the same lock. The compact example below omits TTL so the atomic boundary is easy to see:
from threading import RLock
class AtomicStore:
def __init__(self) -> None:
self.items: dict[str, str] = {}
self.lock = RLock()
def compare_and_set(self, key: str, expected: str | None, value: str) -> bool:
if not isinstance(key, str) or not isinstance(value, str):
raise TypeError("keys and values must be strings")
with self.lock:
if self.items.get(key) != expected:
return False
self.items[key] = value
return True
store = AtomicStore()
print(store.compare_and_set("task:1", None, "running"))
print(store.compare_and_set("task:1", None, "ready"))
print(store.compare_and_set("task:1", "running", "ready"))
try:
store.compare_and_set("task:none", None, None)
except TypeError:
pass
assert "task:none" not in store.itemsTrue
False
TrueFor the TTL store, wrap every public method with the same lock. If compare_and_set() calls get() and set() while holding that lock, use an RLock or split internal helpers so the lock is acquired once. The property to defend is atomicity: no other thread can change the key between compare and set.
Expiry is a time boundary on one record. A scheduler adds a graph, so some tasks are ready, some retry, and some never run. The state transition now depends on other tasks completing.
Task scheduler with retries
Schedule CI tasks with dependencies. A task becomes runnable only after all dependencies complete. Detect cycles before running anything, run ready tasks in priority order with lower numbers first, retry failures up to max_attempts, and return completed, permanently failed, and blocked tasks separately.
Clarify the policies that change observable behavior. Does lower or higher priority win? If a dependency exhausts retries, is its dependent failed, skipped, or blocked? Are retries immediate or delayed with backoff? Say these decisions before you pick the queue, because a heap can order work but can't decide what failure means.
Predict the small graph in the example: docs and test can start, deploy must wait for test, and a permanent test failure leaves deploy blocked. Represent that graph explicitly: dependents[task] lists tasks released by a completed task, while remaining[task] counts unmet dependencies. The invariant is: a task enters the heap only when remaining[task] == 0. Three-state depth-first search (DFS) (0 unseen, 1 active, 2 done) rejects a back edge to an active node in O(V + E) before run is ever called.
import heapq
from collections import defaultdict
from dataclasses import dataclass
@dataclass(frozen=True)
class Task:
name: str
priority: int
deps: tuple[str, ...] = ()
def schedule(tasks: list[Task], run, max_attempts: int = 2) -> tuple[list[str], list[str], list[str]]:
if type(max_attempts) is not int or max_attempts <= 0:
raise ValueError("max_attempts must be positive")
by_name = {task.name: task for task in tasks}
if len(by_name) != len(tasks):
raise ValueError("duplicate task name")
dependents: dict[str, list[str]] = defaultdict(list)
remaining = {task.name: len(task.deps) for task in tasks}
for task in tasks:
if len(set(task.deps)) != len(task.deps):
raise ValueError("duplicate dependency")
for dep in task.deps:
if dep not in by_name:
raise ValueError(f"unknown dependency: {dep}")
dependents[dep].append(task.name)
state: dict[str, int] = {}
def visit(name: str) -> None:
marker = state.get(name, 0)
if marker == 1:
raise ValueError("cycle")
if marker == 2:
return
state[name] = 1
for dep in by_name[name].deps:
visit(dep)
state[name] = 2
for name in by_name:
visit(name)
ready = [(task.priority, task.name) for task in tasks if remaining[task.name] == 0]
heapq.heapify(ready)
attempts = defaultdict(int)
completed: list[str] = []
failed: list[str] = []
while ready:
_, name = heapq.heappop(ready)
attempts[name] += 1
outcome = run(name)
if type(outcome) is not bool:
raise TypeError("run must return a boolean")
if not outcome:
if attempts[name] < max_attempts:
heapq.heappush(ready, (by_name[name].priority, name))
else:
failed.append(name)
continue
completed.append(name)
for child in dependents[name]:
remaining[child] -= 1
if remaining[child] == 0:
heapq.heappush(ready, (by_name[child].priority, child))
blocked = sorted(set(by_name) - set(completed) - set(failed))
return completed, failed, blocked
tasks = [
Task("docs", priority=1),
Task("test", priority=2),
Task("deploy", priority=3, deps=("test",)),
]
fail_once = {"test"}
def run_with_one_retry(name: str) -> bool:
if name in fail_once:
fail_once.remove(name)
return False
return True
print(schedule(tasks, run_with_one_retry))
print(schedule([Task("test", 1), Task("deploy", 2, ("test",))], lambda _: False, max_attempts=1))
try:
schedule(
[Task("lint", 1, ("test",)), Task("test", 1, ("lint",))],
lambda _: True,
)
except ValueError as error:
print(error)
called: list[str] = []
def malformed_result(name: str):
called.append(name)
return "False"
try:
schedule([Task("test", 1), Task("deploy", 2, ("test",))], malformed_result)
except TypeError as error:
print(error)
assert called == ["test"] # The dependent must never run.(['docs', 'test', 'deploy'], [], [])
([], ['test'], ['deploy'])
cycle
run must return a booleandocs and test start ready. Priority 1 runs first, so docs completes. test fails once, retries, then succeeds and releases deploy. In the second call, test is permanently failed after one attempt, so deploy is blocked: it never ran. The cycle lint ↔ test is rejected before any run call, including unrelated ready work. That last result follows from validating the graph before execution, not from the heap.
Keep permanently failed tasks separate from blocked dependents. A failed task ran and exhausted its policy; a blocked task never became eligible. Callers can skip a blocked task, surface its failed prerequisite, or retry after repair without pretending the task itself ran.
Equal priorities break ties by task name because heap entries are (priority, name). Retries are immediate and keep that priority. run must return an actual boolean: the string "False" is truthy and would otherwise mark a failed task completed and release its dependent. The malformed-result fixture now aborts before deploy. An exception also aborts instead of silently becoming a retry. Don't retry arbitrary side effects unless their contract supports it.
For at most A attempts per task, heap work is O(AV log V) and graph work is O(V + E); retained state is O(V + E). Recursive DFS is convenient here, but a long dependency chain can exceed Python's recursion limit. An iterative traversal or Kahn's algorithm removes that depth limit.
Worker fanout
Keep graph construction and cycle detection single-threaded. Then protect only shared scheduler state: ready heap, attempts, completed, failed, and remaining dependency counts. Worker threads can run tasks outside the lock, then reacquire it to publish success or failure and release dependents. The lock protects state transitions, not the task body.
If asked about retries, state whether retries preserve priority or use backoff. A complete answer connects policy to handoff: workers claim ready tasks under a condition variable, execute outside the lock, and publish completion so waiting workers learn when new tasks become ready.
The crawler still needs a quota. Keep the same boundary-first habit: decide whether a request may start, then do the slow work. The next drill makes that decision testable with refill math and an injected clock, no sleep.
Token bucket with deterministic time
Rate limiters combine state, boundary conditions, and overload behavior. Retry storms get worse when every client wakes at the same backoff; jitter spreads those retries.[5] The bucket below starts full and refills from elapsed time. allow either consumes cost or returns how long the caller should wait, while an impossible cost fails immediately instead of producing retry advice that can never succeed.
Predict the fixture before reading the implementation: three cost-1 requests at time 0 use all three tokens. At time 0.5 only half a token has returned, so another cost-1 request should wait 0.5 seconds. At time 1.0 it should pass. Injected now makes those boundaries immediate and repeatable.
import math
from dataclasses import dataclass
@dataclass
class Bucket:
capacity: float
refill_per_second: float
tokens: float
updated_at: float
class TokenBucketLimiter:
def __init__(self, capacity: int, refill_per_second: float) -> None:
if type(capacity) is not int or capacity <= 0:
raise ValueError("capacity must be a positive integer")
if not math.isfinite(refill_per_second) or refill_per_second <= 0:
raise ValueError("refill_per_second must be positive and finite")
self.capacity = float(capacity)
self.refill_per_second = float(refill_per_second)
self._buckets: dict[str, Bucket] = {}
def allow(self, key: str, now: float, cost: float = 1.0) -> tuple[bool, float]:
if not math.isfinite(now):
raise ValueError("now must be finite")
if not math.isfinite(cost) or not 0 < cost <= self.capacity:
raise ValueError("cost must be greater than zero and no larger than capacity")
bucket = self._buckets.get(key)
if bucket is None:
bucket = Bucket(self.capacity, self.refill_per_second, self.capacity, now)
self._buckets[key] = bucket
if now < bucket.updated_at:
raise ValueError("now must not move backwards")
elapsed = max(0.0, now - bucket.updated_at)
bucket.tokens = min(bucket.capacity, bucket.tokens + elapsed * bucket.refill_per_second)
bucket.updated_at = now
if bucket.tokens >= cost:
bucket.tokens -= cost
return True, 0.0
missing = cost - bucket.tokens
retry_after = missing / bucket.refill_per_second
return False, retry_after
limiter = TokenBucketLimiter(capacity=3, refill_per_second=1.0)
print([limiter.allow("org-a", now=0.0)[0] for _ in range(4)])
print(limiter.allow("org-a", now=0.5))
print(limiter.allow("org-a", now=1.0))
try:
limiter.allow("org-a", now=2.0, cost=4.0)
except ValueError as error:
print(error)[True, True, True, False]
(False, 0.5)
(True, 0.0)
cost must be greater than zero and no larger than capacity
Each key owns one bucket. Refill is min(capacity, old_tokens + elapsed * rate); an allowed call then subtracts its cost. A denial consumes nothing. retry_after is advice assuming no competing consumption, not a reservation or a guarantee. The limiter bounds admission rate and burst size, not the number of in-flight requests.
Per-call dictionary work is expected O(1) and memory is O(number of keys). A thread-safe version must lock lookup and mutation together. Idle cleanup must preserve the refill contract: deleting an empty bucket and immediately recreating it full would grant an extra burst. Nondecreasing timestamps are enforced per key, and floating-point boundary behavior still deserves tests for your chosen units.
Denied quota calls get retried. If the retry also debits an organization's remaining credits, "run it again" is the wrong model. The next state boundary is an idempotency key that names one effect across repeated delivery.
Credit ledger with idempotency
Keep credit-like state consistent: org token budgets, GPU-second reservations, or eval-run spend. Use append-only events when you can. If you also keep a balance, update the balance and event together. A retry should return the original result, not apply the same debit or deposit again.
Predict the example: after deposit-1 adds 100 and withdraw-1 removes 30, replaying deposit-1 must leave the balance at 70 and the event count at two. A request key has to identify both the original effect and its parameters, so reusing it for a different delta fails loudly.
from dataclasses import dataclass
@dataclass(frozen=True)
class Event:
idempotency_key: str
account: str
delta: int
balance_after: int
class Ledger:
def __init__(self) -> None:
self.balance: dict[str, int] = {}
self.events: list[Event] = []
self.results: dict[str, Event] = {}
def apply(self, key: str, account: str, delta: int) -> int:
if (not isinstance(key, str) or not key or not isinstance(account, str)
or not account or type(delta) is not int):
raise ValueError("nonempty string identifiers and an integer delta are required")
if key in self.results:
prior = self.results[key]
if (prior.account, prior.delta) != (account, delta):
raise ValueError("idempotency key reused with different operation")
return prior.balance_after
new_balance = self.balance.get(account, 0) + delta
if new_balance < 0:
raise ValueError("insufficient credits")
self.balance[account] = new_balance
event = Event(key, account, delta, new_balance)
self.events.append(event)
self.results[key] = event
return new_balance
ledger = Ledger()
print(ledger.apply("deposit-1", "org-a", 100))
print(ledger.apply("withdraw-1", "org-a", -30))
print(ledger.apply("deposit-1", "org-a", 100))
print(ledger.balance["org-a"], len(ledger.events))
try:
ledger.apply("deposit-1", "org-a", 200)
except ValueError as error:
print(error)
try:
ledger.apply(True, "org-a", 5)
except ValueError as error:
print(error)
assert ledger.balance["org-a"] == 70 and len(ledger.events) == 2100
70
100
70 2
idempotency key reused with different operation
nonempty string identifiers and an integer delta are requiredA replay with the same key and operation returns the stored result, even if the organization's balance changed later. Reusing a key for a different operation fails loudly instead of silently returning an unrelated balance. That's idempotency: the request key names the original effect, not "try the debit again."
The identifier check matters too. Python dictionary keys True and 1 collide because they compare equal. Accepting arbitrary truthy identifiers could merge distinct requests; string-only keys and accounts keep this fixture's identity contract explicit.
This ledger caches successful operations only. An insufficient-credit rejection doesn't reserve the key, so retrying it after a deposit may succeed. The example is single-threaded and volatile: balance, event, and result updates aren't a durable transaction. A crash or concurrent caller can break that larger contract. A durable service needs an atomic write for all three, a uniqueness rule for request keys, and a stated retention period. Each call is expected O(1); retained events and keys grow with successful operations.
A transfer follow-up spans two accounts. A shared ledger lock is the simplest in-process solution. Finer-grained account locks need a global acquisition order, and the request-key registry needs synchronization too: two disjoint account pairs can still reuse the same key. Durable debit, credit, and deduplication records belong in one transaction. An external side effect outside that transaction needs its own recovery and idempotency protocol.
A reusable lock-order helper makes deadlock prevention concrete:
from contextlib import ExitStack, contextmanager
from threading import Lock
locks = {"org-a": Lock(), "org-b": Lock()}
@contextmanager
def lock_accounts(*account_ids: str):
with ExitStack() as stack:
for account_id in sorted(set(account_ids)):
stack.enter_context(locks[account_id])
yield
with lock_accounts("org-b", "org-a"):
print("locked:", sorted({"org-b", "org-a"}))
try:
with lock_accounts("org-a", "unknown"):
raise AssertionError("unknown account must fail before entering the body")
except KeyError:
assert not locks["org-a"].locked()locked: ['org-a', 'org-b']Enter ExitStack before acquiring the first lock. If a later account lookup or acquisition fails, its context releases earlier locks. Returning an unentered stack after acquisition would leak those locks on the exception path. Deduplicating IDs avoids acquiring the same non-reentrant lock twice.[6]
This helper assumes a fixed lock registry shared by every account operation. It prevents lock-order cycles among cooperating callers; it doesn't promise fairness, handle process crashes, or make unprotected readers atomic. A parser needs a different kind of state: a current record rather than a lock owner.
Multiline log parser
Group space-prefixed continuation lines under the previous accepted record. The grammar here is deliberately small: a header contains two | separators and nonempty timestamp and level fields. The parser retains those fields as strings; it doesn't validate ISO dates or a vocabulary of log levels. Extra separators belong to the message. Orphan continuations and rejected headers are ignored, with rejected headers also clearing the current record. Predict two events in the fixture.
import json
def parse_events(text: str) -> list[dict[str, str]]:
events: list[dict[str, str]] = []
current: dict[str, str] | None = None
for line in text.splitlines():
if line.startswith(" "):
if current is not None:
current["message"] += f" {line.strip()}"
continue
parts = line.split("|", maxsplit=2)
if len(parts) != 3:
current = None
continue
timestamp, level, message = parts
if not timestamp or not level:
current = None
continue
current = {"timestamp": timestamp, "level": level, "message": message}
events.append(current)
return events
LOGS = """ orphan continuation
2026-06-02T12:00:00Z|ERROR|request failed
timeout while calling model
malformed line
2026-06-02T12:01:00Z|INFO|retry queued"""
print(json.dumps(parse_events(LOGS), indent=2))[
{
"timestamp": "2026-06-02T12:00:00Z",
"level": "ERROR",
"message": "request failed timeout while calling model"
},
{
"timestamp": "2026-06-02T12:01:00Z",
"level": "INFO",
"message": "retry queued"
}
]The leading continuation has no record to extend. The malformed header clears current, so later continuation text can't silently attach to the ERROR. Only leading spaces count; a tab isn't a continuation under this contract. splitlines() also handles a final line without a newline, and records are appended when headers arrive, so no final flush is needed.
The whole input and output live in memory. Repeated string concatenation can take quadratic time for a record with many continuations. A streaming follow-up should accumulate message fragments in a list, join once when emitting a record, retain incomplete lines between chunks, and bound record size. Test every split position in a short input before claiming chunk boundaries are handled correctly.
Once the state model is right, an interviewer may say "make it faster with threads." Treat that as a concurrency-model question. First identify whether time is going into waiting or into Python computation; only then choose ThreadPoolExecutor, asyncio, or processes.
When threads help
On a standard GIL-enabled CPython build, only one thread executes Python bytecode at a time. Free-threaded builds arrived in 3.13 and became officially supported, still optional, in 3.14.[7] Build support doesn't prove the running process has disabled the GIL: an incompatible extension can re-enable it. On CPython 3.13+, sys._is_gil_enabled() checks the runtime state.[8]
Free-threading isn't data-race freedom: shared mutable state still needs synchronization. The official howto recommends threading.Lock instead of relying on internal locks of built-in dicts and sets.[8] The crawler's membership test and insertion still form one atomic claim.
| Work type | Example | GIL effect | Right tool |
|---|---|---|---|
| I/O-bound | fetching URLs, reading disk, calling a model API | blocking I/O generally releases the GIL, so threads can overlap waits | threads, ThreadPoolExecutor, or asyncio |
| CPU-bound Python | heavy parsing or numeric loops in Python code | threads take turns within one GIL-enabled interpreter | processes, or InterpreterPoolExecutor on Python 3.14+ |
A crawler can overlap network waits with threads. For pure-Python CPU work, Python 3.14 also offers InterpreterPoolExecutor: each worker thread has an isolated interpreter with its own GIL, permitting parallel execution across cores.[9] The workers don't share the parent's mutable dictionaries or lists. Return results and merge them in the parent; replacing a thread pool without changing that ownership model can silently lose updates.
Both interpreter and process pools serialize submitted callables, arguments, and results. A lambda that works in a thread pool isn't a portable worker function. Predict the two checks below: the lambda can't be pickled, and changing the reconstructed list leaves its source unchanged.[10]
import pickle
try:
pickle.dumps(lambda records: len(records))
except (pickle.PicklingError, AttributeError):
print("lambda cannot cross this pickle boundary")
else:
raise AssertionError("expected the lambda to fail serialization")
source = ["ERROR", "INFO"]
worker_copy = pickle.loads(pickle.dumps(source))
worker_copy.append("ERROR")
assert source == ["ERROR", "INFO"] and worker_copy != source
print("parent:", source)
print("worker copy:", worker_copy)lambda cannot cross this pickle boundary
parent: ['ERROR', 'INFO']
worker copy: ['ERROR', 'INFO', 'ERROR']This local round trip demonstrates serialization and copy ownership, not an actual executor or speedup; its bytes come from the fixture itself. Use module-level def functions in an importable module for portable workers. Process-pool startup belongs under if __name__ == "__main__":; Python 3.14's default start method moved away from fork, so don't assume parent globals are inherited.[9] Measure startup, transfer, merge, and computation costs before choosing a pool.
asyncio schedules cooperating tasks on an event-loop thread. A blocking requests.get(...) or time.sleep(...) blocks other tasks on that loop. Use an async client or offload suitable blocking I/O with asyncio.to_thread. An await isn't necessarily a yield: an already-complete awaitable can return immediately. asyncio.sleep(0) specifically suspends the task so other tasks can run.[11]
Each worker yields at await asyncio.sleep(0), so the event loop interleaves them:
import asyncio
async def worker(name: str, log: list[str]) -> None:
log.append(f"{name} start")
await asyncio.sleep(0) # yields control back to the event loop
log.append(f"{name} resume")
async def main() -> list[str]:
log: list[str] = []
await asyncio.gather(worker("a", log), worker("b", log))
return log
print(asyncio.run(main()))['a start', 'b start', 'a resume', 'b resume']Swap await asyncio.sleep(0) for a blocking time.sleep(0) and the output becomes ['a start', 'a resume', 'b start', 'b resume']: worker a runs start-to-finish before b ever begins because a blocking call never returns control to the loop.
An interviewer asks you to make a log parser "faster with threads." The parser is pure-Python string processing over one large file. What do you say?
Answer
First measure whether parsing or I/O dominates. GIL-enabled CPython won't run pure-Python parsing on multiple cores through threads. Consider an algorithm change, native code that releases the GIL, or processes if partitioning and transfer costs are worthwhile. With multiline records, splitting a file arbitrarily can break correctness. Free-threaded builds change the parallelism question, not the need to measure or protect shared state.
Before adding workers, narrate ownership in concrete terms: name shared state X, say exactly what the lock protects, point to the claim point for one work item, explain whether shutdown uses a sentinel, event, or executor lifecycle, and state where a failed item records its error. That short explanation ties synchronization, handoff, shutdown, and failure together before any worker starts.
Debugging when a test fails
When an assertion fails, start with expected versus actual and shrink the fixture until one transition is visible. Inspect the state that owns the invariant, then fix that transition instead of patching the symptom. Add one regression assertion before moving to the next follow-up. Saying those moves as you make them gives the interviewer a trace of your reasoning, not a stream of guesses.
| Symptom | First thing to inspect |
|---|---|
| Duplicate output | claim point or visited insertion timing |
| Missing output | enqueue condition, prefix filter, or failure policy |
| Wrong order | queue/heap tie-break and where sorting happens |
| Expired key returned | read path skipped cleanup |
| Retry ran too many times | attempt counter increment point |
| Blocked task marked failed | dependency failure policy mixed with task execution failure |
| Threaded result flakes | shared state changed outside the lock |
You don't have to be bug-free. Recover by naming which invariant broke: visited was claimed too late, a TTL read skipped cleanup, or a lock dropped between compare and set. Then make the smallest state change that restores it.
When the interviewer adds a follow-up, classify the new boundary before changing code. The first move below tells you which state owns it; the trap reminds you what not to mix into that state.
| Follow-up type | First move | Common trap |
|---|---|---|
| Time | inject now, store deadline, clean on read | sleeping in tests |
| Ordering | choose deque, heap, sort, or insertion order | mixing policy into unrelated state |
| Capacity | evict by explicit rule | hidden off-by-one at capacity 0 or 1 |
| Retry | store attempt count and final state | retrying external writes without idempotency |
| Snapshot | version records or copy-on-write | mutating data a snapshot should freeze |
| Compare-and-set | lock compare and write together | calling get then set as two operations |
| Thread safety | protect claim point and shared maps | holding lock during slow fetch or execution |
| Cancellation | shared event plus cooperative checks | stopping new work while workers keep publishing children |
Extend one example without breaking it
Choose one exercise and change its contract. Write the expected result before implementation, then keep the original tests passing. The feedback below is enough to check your design without prescribing every line of code.
| Exercise | Change | Check your result |
|---|---|---|
| Crawler | Two threads attempt the same claim | Exactly one receives True; don't assert which thread wins. Use a barrier, not sleeps, to start competing attempts. |
| TTL store | Add atomic compare-and-set with expiry | At the exact deadline the old value is missing. Comparing against None can create a replacement while holding one lock. |
| Scheduler | A task fails once, then succeeds | Its dependent runs once, after success. A permanently failed prerequisite leaves the dependent blocked. |
| Token bucket | Evict idle keys | Recreating a key must not grant more tokens than elapsed refill would have supplied. Delete only when that state is equivalent to a full bucket. |
| Ledger | Retry an insufficient-credit request after depositing | Under the stated success-only cache policy, the retry may succeed. Change the tests if you choose to cache failures instead. |
| Parser | Accept arbitrary input chunks | Splitting inside a header or newline must not change emitted records. Flush the final unterminated line once at end-of-input. |
For a fresh prompt, try TTL Key/Value Store, Dependency Scheduler With Retries, or SSE Event Parser. Read each prompt's policy rather than assuming it matches these examples. Then use Bounded Worker Queue to practice termination and backpressure, which a correct claim function alone doesn't solve.
The examples are small enough to inspect, but their limits matter. A passing in-process assertion doesn't establish crash recovery, thread safety of every method, or performance under load. Explain what the test rules out and what additional experiment or implementation the next requirement would need.