Design a key-value store with TTL

01The design, in the six-step framework

1. Requirements and use cases

Reported as a machine-coding round and as a phased phone screen: first a map whose keys die after a lifetime, then a capacity cap, then the awkward questions about what "expired" means at the exact boundary.

• A caller puts a key, a value and a lifetime in seconds; a lifetime of none means it never expires.
• A caller gets a key and never receives a value whose lifetime has passed.
• A caller deletes a key explicitly.
• The store enforces a capacity; when full it first discards expired entries, then the entry closest to expiry.
• A caller can ask for the size, counting only live entries.

Out of scope: persistence, replication and pluggable eviction policies (that is the cache manager). Non-functional constraints: get in O(1); put in O(log n) is acceptable because of the expiry ordering; expiry must be decided by an injected clock so tests do not sleep; and the boundary rule must be stated once and kept: an entry is alive while now < expires_at, so at exactly the expiry instant it is dead.

2. Core entities and relationships

• TTLStore: the public object; owns the map of entries, the expiry heap and the clock; enforces the boundary rule and the capacity.
• Entry: value, expiry instant and a version number used to recognise stale heap records.
• ExpiryHeap: a min-heap of (expires_at, version, key) records; stale records are skipped lazily.
• Clock: now(); real or fake.

The store composes the map and the heap; the heap is private bookkeeping, so no caller can push into it. The clock is aggregated (injected). The version number is the small idea that makes the heap correct without a delete operation: when a key is overwritten, its version rises, and any heap record carrying an older version is ignored when it surfaces. That is the lazy deletion idea from heaps in object design.

3. Interfaces and key methods

from typing import Generic, Hashable, Optional, Protocol, TypeVar

K = TypeVar("K", bound=Hashable)
V = TypeVar("V")


class Clock(Protocol):
    def now(self) -> float: ...


class TTLStore(Generic[K, V]):
    def __init__(self, capacity: int, clock: Clock) -> None:
        """capacity >= 1; the clock decides every expiry question."""

    def put(self, key: K, value: V, ttl: Optional[float] = None) -> None:
        """Insert or replace. ttl in seconds; None never expires; ttl <= 0 raises ValueError."""

    def get(self, key: K) -> Optional[V]:
        """Value if alive at clock.now(); None if missing or expired (expired entries are dropped)."""

    def delete(self, key: K) -> bool:
        """True if a live entry was removed."""

    def __len__(self) -> int:
        """Live entries only; purges whatever is expired first."""

Invariants: a key in entries has exactly one current version; every live key with a finite expiry has at least one heap record with its current version; a heap record whose version differs from the entry's is stale and ignored; and the public methods never observe or return an entry that is dead at clock.now().

4. Design patterns and the choices behind them

Lazy expiry on read plus a heap for eviction

The force is two different questions about time: "is this key alive" (asked on every get) and "which key dies soonest" (asked only when full). The first is a comparison against the entry's own timestamp, O(1), no structure needed. The second needs ordering by expiry, so a min-heap. A background sweeper thread is rejected for a single-process design: it adds locking and shutdown concerns and does not change the big-O. A sorted structure keyed by expiry with true deletes is rejected because Python's heap has no delete, and the version trick makes deletes unnecessary. This is the lazy expiry approach with the clock injected.

Versioned heap records instead of a position index

The force is overwriting a key with a new ttl. The old heap record still exists; if the heap surfaces it later we must not evict the key using the stale time. A version counter on the entry, copied into each record, lets the eviction loop pop and discard stale records. The alternative, a dictionary from key to heap index with manual sift operations, is more code and more bugs for the same complexity.

Size means live size

The force is a boundary question interviewers ask on purpose: do expired keys count towards capacity? The honest answer is "not after we notice them, and len notices on purpose". __len__ purges from the heap top until the top is alive, so the number it returns is exact. That costs amortised O(log n) per expired entry, paid once per entry. Stating the rule in the method's docstring is what prevents a later argument about it.

5. Core flow and code

Flow for put("a", 1, ttl=5) on a full store:

1. Validate the ttl and compute expires_at = now + 5.
2. If a exists, overwrite its value, raise its version and push a new heap record; done (no eviction for a replace).
3. Otherwise purge expired records from the heap top; if still full, pop the heap until a record matches a live entry and remove that key.
4. Insert a with version 1 and push its record.

import heapq
import time
from dataclasses import dataclass
from typing import Dict, Generic, Hashable, List, Optional, Protocol, Tuple, TypeVar

K = TypeVar("K", bound=Hashable)
V = TypeVar("V")


class Clock(Protocol):
    def now(self) -> float: ...


class MonotonicClock:
    def now(self) -> float:
        return time.monotonic()


class FakeClock:
    def __init__(self) -> None:
        self.t = 0.0

    def now(self) -> float:
        return self.t

    def advance(self, dt: float) -> None:
        self.t += dt


@dataclass
class Entry(Generic[V]):
    value: V
    expires_at: Optional[float]
    version: int

    def alive(self, now: float) -> bool:
        return self.expires_at is None or now < self.expires_at


class TTLStore(Generic[K, V]):
    INFINITE = float("inf")

    def __init__(self, capacity: int, clock: Optional[Clock] = None) -> None:
        if capacity < 1:
            raise ValueError("capacity must be at least 1")
        self.capacity = capacity
        self.clock: Clock = clock or MonotonicClock()
        self.entries: Dict[K, Entry[V]] = {}
        self.heap: List[Tuple[float, int, K]] = []   # (expires_at, version, key)
        self._versions = 0

    def put(self, key: K, value: V, ttl: Optional[float] = None) -> None:
        if ttl is not None and ttl <= 0:
            raise ValueError("ttl must be positive")
        now = self.clock.now()
        expires_at = None if ttl is None else now + ttl
        self._versions += 1
        existing = self.entries.get(key)
        if existing is not None and existing.alive(now):
            self.entries[key] = Entry(value, expires_at, self._versions)
            self._push(key, expires_at, self._versions)
            return
        if existing is not None:          # dead key being reused
            del self.entries[key]
        self._purge(now)
        if len(self.entries) >= self.capacity:
            self._evict_soonest(now)
        self.entries[key] = Entry(value, expires_at, self._versions)
        self._push(key, expires_at, self._versions)

    def get(self, key: K) -> Optional[V]:
        entry = self.entries.get(key)
        if entry is None:
            return None
        if not entry.alive(self.clock.now()):
            del self.entries[key]
            return None
        return entry.value

    def delete(self, key: K) -> bool:
        entry = self.entries.pop(key, None)
        return entry is not None and entry.alive(self.clock.now())

    def __len__(self) -> int:
        self._purge(self.clock.now())
        return len(self.entries)

    # ---- internals -------------------------------------------------
    def _push(self, key: K, expires_at: Optional[float], version: int) -> None:
        heapq.heappush(self.heap, (self.INFINITE if expires_at is None else expires_at, version, key))

    def _current(self, record: Tuple[float, int, K]) -> bool:
        entry = self.entries.get(record[2])
        return entry is not None and entry.version == record[1]

    def _purge(self, now: float) -> None:
        """Drop stale heap records and expired entries from the top of the heap."""
        while self.heap:
            expires_at, version, key = self.heap[0]
            if not self._current(self.heap[0]):
                heapq.heappop(self.heap)
            elif expires_at <= now:
                heapq.heappop(self.heap)
                del self.entries[key]
            else:
                break

    def _evict_soonest(self, now: float) -> None:
        while self.heap:
            record = heapq.heappop(self.heap)
            if self._current(record):
                del self.entries[record[2]]
                return


if __name__ == "__main__":
    clock = FakeClock()
    s: TTLStore[str, int] = TTLStore(2, clock)
    s.put("a", 1, ttl=5)
    s.put("b", 2)
    clock.advance(5)
    assert s.get("a") is None            # dead at exactly expires_at
    assert len(s) == 1
    s.put("c", 3, ttl=1)
    s.put("d", 4)                        # full: c expires soonest, so c goes
    assert s.get("c") is None and s.get("b") == 2 and s.get("d") == 4

6. Extensions, concurrency and what interviewers probe

• The boundary. Alive while now < expires_at. Pick it, write it in the docstring, test it with the fake clock at exactly the instant. Candidates who use <= in one place and < in another lose the round on a follow-up.
• Do expired keys count towards capacity? Only until noticed; len and put notice. If the interviewer wants a hard guarantee, add a sweeper that calls _purge on a timer; the method already exists.
• Reusing a dead key. The put path treats a dead existing key as absent; forgetting this makes a dead key immortal because the replace branch skips eviction.
• Thread safety. One lock around the four public methods. The heap and the map must change together, so there is no finer-grained option worth its complexity in an hour.
• Time-based reads. "Give me the value as of time t" is a different problem with a different structure; see the time-based key-value store.
• What to test first. The fake-clock scenario in the code, then overwrite-with-shorter-ttl followed by eviction, which exercises the stale-record path.

02Where this question was reported

• DesignGurus · DesignGurus case study 'In-Memory Cache'
• LLD Mastery · Google-tagged (Hard); LRU/LFU eviction, TTL expiry
• LLD Mastery · Google-tagged with Amazon/Microsoft/Meta/Salesforce
• DesignGurus · DesignGurus blog common LLD question
• PracHub · PracHub OOD practice prompt 'Cache'
• LeetCode Discuss · Thread-safe LFU; follow-up on concurrency; interviewer pushed back on relying only on `synchronized`
• LeetCode Discuss · Round 'felt like implementing an LRU/LFU-style system'; edge cases were the hard part
• Taro · Taro tags Google; edge case: get immediately followed by put must evict the now-least-recent key
• Prepfully · Listed among recently reported Google coding questions: O(1) insert/delete/retrieve LRU
• ReturnZero · Phased: get/put -> eviction -> O(1) -> recency queries + TTL expiry (Google tag hidden behind '+7')
• PracHub · Data-structure design problem 'similar to an LRU variant' in coding round
• GitHub · Build a KV store with TTL expiry (same round: cursor-based pagination API)