Backend Coding Patterns and Practical Problems

Backend coding rounds look different from pure algorithm rounds. Instead of a puzzle you get a small practical system: an in-memory key-value store with expiry, a rate limiter, a log parser, a job scheduler, a URL shortener, a bank ledger. The test is how you model state, handle edge cases and concurrency, keep the code clean and extend it when the requirements change. This chapter gives a method and tested solutions to the problems that come up most often.

1. A method

1. Clarify the interface: what operations exist, their inputs and outputs, error behaviour, scale, and whether it must be thread-safe.
2. Choose the data structures and state their complexity (hash map for lookups, heap for expiry or scheduling, deque for windows, sorted structure for ranges).
3. Write a clean, small API first, with the simplest correct implementation.
4. Handle edge cases: empty input, duplicates, missing keys, time boundaries, overflow, invalid input.
5. Inject dependencies (the clock, the ID generator, the random source) so the code is testable.
6. Discuss extensions: concurrency, persistence, memory limits, distribution, observability.
7. Test as you go with a few assertions.

Interviewers often change a requirement mid-way ("now keys expire", "now it must be thread-safe", "now millions of keys"). Code that is small and well-structured adapts easily.

2. In-memory key-value store with TTL

Requirements: set(key, value, ttl=None), get(key), delete(key), expired keys must not be returned, memory must not grow without bound.

Design: a hash map for values plus a min-heap of expiry times for efficient cleanup (lazy deletion on read plus periodic sweep). The heap may hold stale entries when a key is overwritten, so each heap entry is validated against the map.

import heapq

class TTLStore:
    def __init__(self, clock):
        self.clock, self.data, self.expiries = clock, {}, []        # data: key -> (value, expires_at or None)

    def set(self, key, value, ttl=None):
        expires_at = self.clock() + ttl if ttl is not None else None
        self.data[key] = (value, expires_at)
        if expires_at is not None:
            heapq.heappush(self.expiries, (expires_at, key))

    def get(self, key, default=None):
        entry = self.data.get(key)
        if entry is None:
            return default
        value, expires_at = entry
        if expires_at is not None and expires_at <= self.clock():
            del self.data[key]                                       # lazy expiry on access
            return default
        return value

    def delete(self, key):
        return self.data.pop(key, None) is not None

    def sweep(self):
        """Remove expired keys using the heap; each heap entry is checked against the current value."""
        now, removed = self.clock(), 0
        while self.expiries and self.expiries[0][0] <= now:
            expires_at, key = heapq.heappop(self.expiries)
            entry = self.data.get(key)
            if entry and entry[1] == expires_at:                     # skip stale heap entries from overwritten keys
                del self.data[key]; removed += 1
        return removed

t = [0]
s = TTLStore(lambda: t[0])
s.set("a", 1, ttl=10); s.set("b", 2); s.set("c", 3, ttl=5)
t[0] = 6
assert s.get("c") is None and s.get("a") == 1 and s.get("b") == 2   # "c" expired, the others are alive
s.set("a", 99, ttl=100)                                              # overwriting extends the lifetime
t[0] = 11
assert s.get("a") == 99                                              # the old expiry of 10 must not delete the new value
assert s.sweep() == 0 and len(s.data) == 2
t[0] = 200
assert s.sweep() == 1 and s.get("a") is None and s.get("b") == 2

Follow-ups: thread safety (one lock, or a lock per shard), a background sweeper thread, a memory cap with LRU eviction, persistence (snapshots and an append-only log), and distribution (consistent hashing).

3. LRU cache with O(1) operations

A hash map plus a doubly linked list (Python's OrderedDict implements both).

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity):
        assert capacity > 0
        self.capacity, self.items = capacity, OrderedDict()

    def get(self, key):
        if key not in self.items:
            return -1
        self.items.move_to_end(key)
        return self.items[key]

    def put(self, key, value):
        if key in self.items:
            self.items.move_to_end(key)
        self.items[key] = value
        if len(self.items) > self.capacity:
            self.items.popitem(last=False)

c = LRUCache(2)
c.put(1, 1); c.put(2, 2)
assert c.get(1) == 1
c.put(3, 3)                                  # evicts key 2, the least recently used
assert c.get(2) == -1 and c.get(3) == 3
c.put(1, 10)                                 # updating an existing key refreshes it and does not evict
assert c.get(1) == 10 and c.get(3) == 3

4. Parsing and aggregating logs

A typical practical task: read access-log lines, compute top endpoints, error rates and percentiles.

from collections import Counter, defaultdict
import re

LOG = """\
2024-05-01T10:00:01Z GET /api/users 200 35
2024-05-01T10:00:02Z GET /api/users 500 120
2024-05-01T10:00:02Z POST /api/orders 201 80
2024-05-01T10:00:03Z GET /api/users 200 45
2024-05-01T10:00:04Z GET /api/orders 404 12
malformed line here
2024-05-01T10:00:05Z POST /api/orders 500 300
"""
LINE = re.compile(r"^(\S+) (GET|POST|PUT|DELETE) (\S+) (\d{3}) (\d+)$")

def parse(text):
    for raw in text.splitlines():
        m = LINE.match(raw)
        if not m:
            continue                                  # skip malformed lines instead of crashing; count them in real code
        ts, method, path, status, ms = m.groups()
        yield {"ts": ts, "method": method, "path": path, "status": int(status), "ms": int(ms)}

def summarise(records):
    count, errors, durations = Counter(), Counter(), defaultdict(list)
    for r in records:
        count[r["path"]] += 1
        durations[r["path"]].append(r["ms"])
        if r["status"] >= 500:
            errors[r["path"]] += 1
    return {p: {"requests": count[p], "error_rate": errors[p] / count[p], "max_ms": max(durations[p])} for p in count}

summary = summarise(parse(LOG))
assert summary["/api/users"] == {"requests": 3, "error_rate": 1 / 3, "max_ms": 120}
assert summary["/api/orders"]["requests"] == 3 and abs(summary["/api/orders"]["error_rate"] - 1 / 3) < 1e-12
assert len(list(parse(LOG))) == 6                                         # the malformed line was skipped
top = Counter({p: v["requests"] for p, v in summary.items()}).most_common(1)
assert top[0][1] == 3

Points to raise: stream large files line by line rather than loading them (a generator, as above), use a min-heap of size k for top-k on huge key sets, t-digest or histograms for approximate percentiles, and handle time windows with a sliding window or bucketing by minute.

5. Sliding-window rate limiter as a class

(The resilience chapter covers the algorithms. Here is the interview-style class with a per-user limit and an injectable clock.)

from collections import deque, defaultdict

class RateLimiter:
    def __init__(self, limit, window_seconds, clock):
        self.limit, self.window, self.clock = limit, window_seconds, clock
        self.hits = defaultdict(deque)

    def allow(self, user):
        now, q = self.clock(), self.hits[user]
        while q and q[0] <= now - self.window:
            q.popleft()
        if len(q) >= self.limit:
            return False
        q.append(now)
        return True

t = [0.0]
rl = RateLimiter(limit=3, window_seconds=10, clock=lambda: t[0])
assert [rl.allow("u1") for _ in range(4)] == [True, True, True, False]
assert rl.allow("u2") is True                       # limits are per user
t[0] = 10.0
assert rl.allow("u1") is True                       # the first hit left the window exactly at 10 s

Follow-ups: memory cleanup for idle users (periodic purge of empty deques), a token bucket for bursts, distribution via Redis, and fail-open or fail-closed behaviour.

6. Job scheduler and delayed queue

Run tasks at or after a given time, in time order, with priorities and retries. A min-heap keyed by run time gives $O(\log n)$ scheduling.

import heapq, itertools

class Scheduler:
    def __init__(self, clock):
        self.clock, self.heap, self.counter = clock, [], itertools.count()

    def schedule(self, run_at, name, fn, attempts_left=3):
        heapq.heappush(self.heap, (run_at, next(self.counter), name, fn, attempts_left))   # the counter breaks ties without comparing functions

    def run_due(self):
        ran, now = [], self.clock()
        while self.heap and self.heap[0][0] <= now:
            run_at, _, name, fn, attempts_left = heapq.heappop(self.heap)
            try:
                fn(); ran.append(name)
            except Exception:
                if attempts_left > 1:                                   # retry with exponential backoff
                    delay = 2 ** (4 - attempts_left)
                    self.schedule(now + delay, name, fn, attempts_left - 1)
        return ran

t = [0]
sched = Scheduler(lambda: t[0])
tries = {"n": 0}
def eventually_ok():
    tries["n"] += 1
    if tries["n"] < 3:
        raise RuntimeError("try again")

sched.schedule(5, "report", lambda: None)
sched.schedule(1, "job", eventually_ok)
done = []
for now in range(0, 12):                              # advance the fake clock one second at a time
    t[0] = now
    done += sched.run_due()
assert done == ["report", "job"]                       # "report" ran at 5; "job" failed at 1 and 3, then succeeded at 7
assert tries["n"] == 3

7. URL shortener core

Generate short codes. Options: counter encoded in base 62 (unique, sequential, guessable), random codes with collision checks, or hash prefixes with collision handling.

import string, secrets

ALPHABET = string.digits + string.ascii_lowercase + string.ascii_uppercase          # 62 characters

def encode_base62(n):
    if n == 0:
        return ALPHABET[0]
    out = []
    while n:
        n, r = divmod(n, 62)
        out.append(ALPHABET[r])
    return "".join(reversed(out))

def decode_base62(s):
    n = 0
    for ch in s:
        n = n * 62 + ALPHABET.index(ch)
    return n

assert encode_base62(0) == "0" and encode_base62(61) == "Z" and encode_base62(62) == "10"
assert all(decode_base62(encode_base62(n)) == n for n in (0, 1, 61, 62, 3843, 10**9, 2**40))
assert len(encode_base62(62 ** 7 - 1)) == 7                            # seven characters cover 62^7, about 3.5 trillion URLs

class Shortener:
    def __init__(self):
        self.by_code, self.by_url = {}, {}

    def shorten(self, url):
        if url in self.by_url:
            return self.by_url[url]                                    # the same URL gets the same code (idempotent)
        while True:
            code = "".join(secrets.choice(ALPHABET) for _ in range(7))  # unguessable random code
            if code not in self.by_code:                                # check for a collision before committing
                break
        self.by_code[code], self.by_url[url] = url, code
        return code

    def resolve(self, code):
        return self.by_code.get(code)

s = Shortener()
code = s.shorten("https://example.com/a/very/long/path")
assert len(code) == 7 and s.resolve(code) == "https://example.com/a/very/long/path"
assert s.shorten("https://example.com/a/very/long/path") == code
assert s.resolve("nope123") is None

Capacity maths: seven base-62 characters give $62^7\approx 3.5\times 10^{12}$ codes. In production the collision check is a unique constraint in the database, not an in-memory lookup, and codes can be pre-generated by a key service to avoid contention.

8. A bank ledger with invariants

Model money safely: integer units, double-entry style, idempotent operations, and invariants checked after every operation.

class Ledger:
    def __init__(self):
        self.balances, self.applied = {}, set()

    def open(self, account, initial=0):
        assert initial >= 0
        self.balances[account] = initial

    def transfer(self, op_id, src, dst, amount):
        if op_id in self.applied:
            return "duplicate"                                          # idempotent: replays do nothing
        if amount <= 0 or src == dst:
            return "invalid"
        if self.balances[src] < amount:
            return "insufficient funds"
        self.balances[src] -= amount
        self.balances[dst] += amount
        self.applied.add(op_id)
        return "ok"

    def total(self):
        return sum(self.balances.values())

l = Ledger(); l.open("a", 100); l.open("b", 50)
assert l.transfer("t1", "a", "b", 30) == "ok"
assert l.transfer("t1", "a", "b", 30) == "duplicate"                     # a retried request does not move money twice
assert l.transfer("t2", "a", "b", 500) == "insufficient funds"
assert l.transfer("t3", "a", "a", 1) == "invalid"
assert l.balances == {"a": 70, "b": 80} and l.total() == 150            # the invariant: money is conserved

In a real system this is a database transaction with row locks (lock accounts in a fixed order), a unique constraint on op_id, and an append-only ledger table.

9. Merging sorted streams and top-k (patterns behind many tasks)

import heapq

def merge_sorted(*streams):
    return list(heapq.merge(*streams))                                  # a k-way merge with a heap: O(n log k)

assert merge_sorted([1, 4, 9], [2, 3, 10], [0, 5]) == [0, 1, 2, 3, 4, 5, 9, 10]

def top_k_frequent(words, k):
    counts = {}
    for w in words:
        counts[w] = counts.get(w, 0) + 1
    return [w for w, _ in heapq.nsmallest(k, counts.items(), key=lambda kv: (-kv[1], kv[0]))]   # ties broken alphabetically

assert top_k_frequent(["b", "a", "b", "c", "a", "b", "d"], 2) == ["b", "a"]

10. Making code testable and extensible

• Inject the clock, random source and IDs (as above) so tests are deterministic.
• Separate pure logic from I/O, with small interfaces at the edges.
• Return explicit results or raise specific exceptions; do not return magic values for several different failures.
• Keep operations atomic with respect to your invariants (check and modify under one lock or transaction).
• Name things for the domain, keep functions short, and write the tests you would want to read.

11. Common mistakes

• Starting to code before pinning down the interface and the edge cases.
• Using wall-clock time directly, making tests flaky.
• Forgetting expiry or eviction, so memory grows without bound.
• Check-then-act races in "thread-safe" code.
• Floating-point money.
• Not making retried operations idempotent.
• Loading entire files into memory when streaming would do.
• Unstable tie-breaking in sorting and top-k, producing flaky results.
• Ignoring the follow-up question's hint: it usually points at the real concern (concurrency, scale, persistence).

12. Practice set

1. Design a thread-safe in-memory cache with TTL and a maximum size.
2. Implement a rate limiter (fixed window, sliding window, token bucket) and compare them.
3. Parse a large access log and report the 95th-percentile latency per endpoint.
4. Build a job scheduler with delays, priorities and retries.
5. Implement a URL shortener's data model and code generation.
6. Model a bank ledger with idempotent transfers and an audit trail.
7. Implement a publish-subscribe event bus with ordering guarantees per topic.
8. Design a leaderboard supporting add_score, top(k) and rank(user) efficiently.
9. Build a simple dependency resolver (topological sort) for build tasks.
10. Implement a bounded blocking queue with producers and consumers.