Design a cache manager with pluggable eviction and TTL

01The design, in the six-step framework

1. Requirements and use cases

The prompt is deliberately broader than "write an LRU": build a cache whose eviction policy can be swapped and whose entries can expire. It is a test of whether you can separate what changes from what stays.

• A caller gets and puts entries by key with an optional time-to-live per entry.
• A caller configures the eviction policy at construction: LRU, LFU or FIFO today, others tomorrow.
• An expired entry is never returned, and it stops counting towards capacity once noticed.
• When full, the cache asks the policy which key to evict and removes exactly that one.
• A caller can remove a key and read size.

Out of scope: distribution, persistence, and a per-key policy. Non-functional constraints: adding a new policy must not touch the cache class (the named future change); the cache must be testable with a fake clock so expiry tests do not sleep; and get and put stay O(1) for the policies shipped.

2. Core entities and relationships

• Cache: the public object; owns the store of entries, the clock and the policy; decides when to evict and when to expire.
• Entry: a value plus its expiry instant (or none).
• EvictionPolicy: an interface with on_access, on_insert, on_remove and victim; it ranks keys and nothing else.
• LRUPolicy, LFUPolicy, FIFOPolicy: implementations, each with its own bookkeeping structure.
• Clock: an interface with now(); the real one reads monotonic time, the test one is set by hand.

The cache composes its store and aggregates its policy and clock: they are passed in, so a test can hand it a fake clock and a spy policy. The policy never holds values, only keys, so the two cannot disagree about what a value is; they can only disagree about which keys exist, and the cache keeps them in step by calling the policy on every mutation. This is the dependency inversion shape: the cache depends on two small interfaces, not on any concrete policy or on the system clock.

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:
        """Seconds on a monotonic scale."""


class EvictionPolicy(Protocol[K]):
    def on_insert(self, key: K) -> None: """A new key entered the cache."""
    def on_access(self, key: K) -> None: """An existing key was read or overwritten."""
    def on_remove(self, key: K) -> None: """A key left the cache (eviction, expiry or explicit remove)."""
    def victim(self) -> K: """The key to evict next; the cache is non-empty when this is called."""


class Cache(Generic[K, V]):
    def __init__(self, capacity: int, policy: EvictionPolicy[K], clock: Clock) -> None: ...
    def get(self, key: K) -> Optional[V]:
        """Value if present and not expired; expired entries are removed on the spot."""
    def put(self, key: K, value: V, ttl: Optional[float] = None) -> None:
        """Insert or replace; evicts policy.victim() when a new key would exceed capacity."""
    def remove(self, key: K) -> bool:
        """True if the key was present."""
    def __len__(self) -> int: ...

Invariants: the set of keys the policy knows equals the set of keys in the store; victim() is only called when the store is non-empty; an entry with expires_at in the past is treated as absent by every public method; and the policy is told about every removal, including expiry, so it never nominates a key that is already gone.

4. Design patterns and the choices behind them

Strategy for the policy

The force is that the ranking rule varies while the storage, expiry and capacity logic do not. The Strategy pattern puts the rule behind an interface the cache calls at four moments. Adding a policy is a new class, which is the open-closed principle in its most literal form. The rejected alternative is an enum field with if policy == LRU branches inside the cache; it works for three policies and rots at the fourth, and it makes each policy's bookkeeping leak into the cache's fields.

Lazy expiry with an injected clock

The force is time. A background sweeper thread adds a thread, a lock and a shutdown story to a one-hour design. Lazy expiry, checking expires_at on read and on the eviction path, costs nothing extra per operation and keeps the class single-threaded. Its weakness is that expired entries that are never read keep occupying slots until the policy happens to evict them; mention that a periodic sweep can be added later as a separate collaborator. The injected clock is what makes any of this testable: a test advances a fake clock by 61 seconds and asserts the miss, without sleeping.

Policies store keys, never values

The force is duplication of state. If the policy held entries, two objects would own the same value and could disagree. Keeping the policy as a pure ordering over keys, and the store as the single owner of values, is single responsibility applied to data rather than behaviour. Both LRU and FIFO become an OrderedDict; the only difference is whether on_access moves the key to the end.

5. Core flow and code

Flow for get("a") on an entry that has expired:

1. The cache finds the entry for a in the store and asks the clock for now().
2. entry.expired(now) is true, so the cache deletes the entry and calls policy.on_remove("a").
3. The cache returns None; the caller cannot tell an expired key from a missing one, which is the contract.

Flow for put("d", 4) on a full cache: the cache first clears any expired entries it can see cheaply (the store is not scanned; only the policy's victim is checked), then asks policy.victim(), removes that key from both the store and the policy, inserts d and calls policy.on_insert("d").

import time
from collections import OrderedDict
from dataclasses import dataclass
from typing import Dict, Generic, Hashable, Optional, Protocol, 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, t: float = 0.0) -> None:
        self.t = t

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

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


class EvictionPolicy(Protocol[K]):
    def on_insert(self, key: K) -> None: ...
    def on_access(self, key: K) -> None: ...
    def on_remove(self, key: K) -> None: ...
    def victim(self) -> K: ...


class LRUPolicy(Generic[K]):
    def __init__(self) -> None:
        self.order: "OrderedDict[K, None]" = OrderedDict()

    def on_insert(self, key: K) -> None:
        self.order[key] = None

    def on_access(self, key: K) -> None:
        self.order.move_to_end(key)

    def on_remove(self, key: K) -> None:
        self.order.pop(key, None)

    def victim(self) -> K:
        return next(iter(self.order))


class FIFOPolicy(LRUPolicy[K]):
    def on_access(self, key: K) -> None:
        pass  # insertion order only


class LFUPolicy(Generic[K]):
    def __init__(self) -> None:
        self.freq: Dict[K, int] = {}
        self.buckets: Dict[int, "OrderedDict[K, None]"] = {}
        self.min_freq = 0

    def on_insert(self, key: K) -> None:
        self.freq[key] = 1
        self.buckets.setdefault(1, OrderedDict())[key] = None
        self.min_freq = 1

    def on_access(self, key: K) -> None:
        f = self.freq[key]
        del self.buckets[f][key]
        if not self.buckets[f]:
            del self.buckets[f]
            if self.min_freq == f:
                self.min_freq = f + 1
        self.freq[key] = f + 1
        self.buckets.setdefault(f + 1, OrderedDict())[key] = None

    def on_remove(self, key: K) -> None:
        f = self.freq.pop(key, None)
        if f is not None:
            self.buckets[f].pop(key, None)
            if not self.buckets[f]:
                del self.buckets[f]
                if self.min_freq == f:
                    self.min_freq = min(self.buckets) if self.buckets else 0

    def victim(self) -> K:
        return next(iter(self.buckets[self.min_freq]))


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

    def expired(self, now: float) -> bool:
        return self.expires_at is not None and now >= self.expires_at


class Cache(Generic[K, V]):
    def __init__(self, capacity: int, policy: EvictionPolicy[K], clock: Optional[Clock] = None) -> None:
        if capacity < 1:
            raise ValueError("capacity must be at least 1")
        self.capacity, self.policy = capacity, policy
        self.clock: Clock = clock or MonotonicClock()
        self.store: Dict[K, Entry[V]] = {}

    def get(self, key: K) -> Optional[V]:
        entry = self.store.get(key)
        if entry is None:
            return None
        if entry.expired(self.clock.now()):
            self._remove(key)
            return None
        self.policy.on_access(key)
        return entry.value

    def put(self, key: K, value: V, ttl: Optional[float] = None) -> None:
        expires_at = None if ttl is None else self.clock.now() + ttl
        if key in self.store:
            self.store[key] = Entry(value, expires_at)
            self.policy.on_access(key)
            return
        if len(self.store) >= self.capacity:
            self._remove(self.policy.victim())
        self.store[key] = Entry(value, expires_at)
        self.policy.on_insert(key)

    def remove(self, key: K) -> bool:
        if key not in self.store:
            return False
        self._remove(key)
        return True

    def _remove(self, key: K) -> None:
        del self.store[key]
        self.policy.on_remove(key)

    def __len__(self) -> int:
        return len(self.store)


if __name__ == "__main__":
    clock = FakeClock()
    c: Cache[str, int] = Cache(2, LRUPolicy(), clock)
    c.put("a", 1, ttl=10)
    c.put("b", 2)
    clock.advance(11)
    assert c.get("a") is None and len(c) == 1   # expired and dropped on read
    c.put("c", 3)
    c.put("d", 4)                                # evicts b (LRU among live keys)
    assert c.get("b") is None and c.get("c") == 3

6. Extensions, concurrency and what interviewers probe

• Adding a policy. A random-eviction or size-aware policy is one new class implementing four methods; the cache is untouched. Interviewers ask for this live; have the four method names ready, since they are the extensibility point you chose in advance.
• The victim may be expired. Evicting an already-expired key is fine and cheap, but a strictly better version checks whether the victim is expired and loops once more so a live key survives; say it, do not necessarily code it.
• Thread safety. Wrap the public methods in one lock, inside the cache, not inside each policy; policies are then free of locking and stay trivial to test. Striping by key, as in the thread-safe LFU, is the throughput step.
• A read-through proxy. Wrapping the cache in an object that loads on miss from a slower source is the Proxy shape; keep it outside this class so the cache stays a cache.
• What to test first. The fake clock scenario in the code; then "policy and store agree" after every operation, checked by a spy policy that records calls.
• Common mistake. Letting the policy nominate a key that expiry already removed, because expiry deleted from the store but forgot on_remove. Route every deletion through one private method so that cannot happen.

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'
• YouTube · 'Cache' is Episode 2 of 'Google SWE Teaches Low Level Design' (interface-driven eviction policies); watch URL not resolved