LRU Cache with Expiry
Published 6 October 2026
Build an LRU cache where every entry also has a TTL. The eviction rule when the cache is full:
- First, clear out entries that have already expired.
- If that freed nothing, evict the least recently used entry as usual.
The rule itself is simple. The work is in step 1: how do you find the expired entries
without scanning the whole cache? Scanning the map every time the cache fills up costs
O(n) per put. The fix is a second index, ordered by expiry time, so the entry that
expires soonest is always at the front.
The data structures¶
Three structures, each answering one question:
| Structure | Answers | Cost |
|---|---|---|
dict key → node |
"Is this key cached, and where is it?" | O(1) |
| Doubly linked list | "Who is least recently used?" (front = MRU, back = LRU) | O(1) |
Min-heap of (expires_at, seq, key) |
"Who expires next?" | O(log n) |
map: a ─► Node(a) b ─► Node(b) c ─► Node(c)
recency: head <-> c <-> a <-> b <-> tail
MRU LRU
by_expiry: (t=105, a) (t=110, c) (t=300, b)
▲ heap top: the next entry to expire
The recency list and the expiry heap order the same entries by two unrelated things.
b is the least recently used, but a expires first. That's why one list can't do both
jobs: reading a key changes its recency and leaves its expiry alone.
In C++ or Java the expiry index would be a sorted set (std::set, TreeSet). Python's
standard library has no sorted set, so this uses heapq. The catch is that a heap can't
delete an arbitrary element, which comes up whenever a key is overwritten or removed
early. The standard workaround is lazy deletion: leave the old heap entry where it
is, and when it reaches the top, check whether it still matches a live node. If not,
throw it away.
The operations¶
get(key)¶
- Key not in the map: miss.
- Key present but
expires_at <= now: remove it and return a miss (lazy expiry). - Otherwise move the node to the front of the recency list and return the value.
put(key, value, ttl)¶
- If the key already exists, remove the old entry.
- If the cache is full, pop the heap while its top has
expires_at <= now, removing each entry that's still live. - If it is still full, evict the node at the back of the recency list.
- Insert the new node at the front of the list, in the map, and in the heap.
Python implementation¶
import heapq
import itertools
import time
class Node:
__slots__ = ("key", "value", "expires_at", "seq", "prev", "next")
def __init__(self, key, value, expires_at, seq):
self.key = key
self.value = value
self.expires_at = expires_at
self.seq = seq
self.prev = None
self.next = None
class LRUTTLCache:
def __init__(self, capacity, clock=time.monotonic):
if capacity <= 0:
raise ValueError("capacity must be positive")
self.capacity = capacity
self.clock = clock
self.map = {} # key -> Node
self.by_expiry = [] # heap of (expires_at, seq, key)
self.seq = itertools.count() # tells a live heap entry from a stale one
# recency list with sentinels: head.next = MRU, tail.prev = LRU
self.head = Node(None, None, None, None)
self.tail = Node(None, None, None, None)
self.head.next = self.tail
self.tail.prev = self.head
# ---- linked list ----
def _unlink(self, node):
node.prev.next = node.next
node.next.prev = node.prev
def _push_front(self, node):
node.prev = self.head
node.next = self.head.next
self.head.next.prev = node
self.head.next = node
# ---- bookkeeping ----
def _remove(self, node):
# The heap entry is left behind; it goes stale because the key no
# longer maps to a node with this seq.
self._unlink(node)
del self.map[node.key]
def _purge_expired(self, now):
heap = self.by_expiry
while heap and heap[0][0] <= now:
_, seq, key = heapq.heappop(heap)
node = self.map.get(key)
if node is not None and node.seq == seq: # skip stale entries
self._remove(node)
# ---- public API ----
def get(self, key):
node = self.map.get(key)
if node is None:
return None
if node.expires_at <= self.clock(): # lazy expiry
self._remove(node)
return None
self._unlink(node) # mark as most recently used
self._push_front(node)
return node.value
def put(self, key, value, ttl):
now = self.clock()
old = self.map.get(key)
if old is not None:
self._remove(old)
if len(self.map) >= self.capacity:
self._purge_expired(now) # 1. drop expired entries
if len(self.map) >= self.capacity:
self._remove(self.tail.prev) # 2. still full: evict the LRU
node = Node(key, value, now + ttl, next(self.seq))
self.map[key] = node
self._push_front(node)
heapq.heappush(self.by_expiry, (node.expires_at, node.seq, key))
# Overwrites and lazy expiry leave stale heap entries behind. Rebuild
# once they outnumber the live ones so the heap stays O(capacity).
if len(self.by_expiry) > 2 * self.capacity:
self.by_expiry = [(n.expires_at, n.seq, n.key) for n in self.map.values()]
heapq.heapify(self.by_expiry)
Usage, with ttl in seconds:
cache = LRUTTLCache(capacity=2)
cache.put("a", 1, ttl=60)
cache.put("b", 2, ttl=0.5)
cache.get("b") # 2, so "b" is MRU and "a" is LRU
time.sleep(1)
cache.put("c", 3, ttl=60) # full: "b" has expired, so "b" goes, not LRU "a"
cache.get("a") # 1, still there
Without the expiry pass, the last put would have evicted a, the least recently used
key, and kept the dead b around. With it, the dead entry goes first and live data
survives.
I checked this against a brute-force reference (a dict of (value, expires_at,
last_used) that scans for expired keys and the LRU victim) on 3,000 random operation
sequences with a fake clock, and the two agree on every get and on the set of cached
keys after every step.
Details that are easy to get wrong¶
- Why
seqis in the heap entry. It does two jobs. It's how_purge_expiredtells a live entry from a stale one: ifawas written at t=0 with a 10s TTL and rewritten at t=5, the heap holds twoaentries, and only the one whoseseqmatches the current node may remove it. Without the check, the old entry would pop at t=10 and delete the new value early. It also breaks ties between equalexpires_atvalues, soheapqnever falls through to comparing keys (which may not be comparable). - The stale entries need a ceiling. Without the rebuild at the end of
put, a cache that never fills up never purges, so rewriting one hot key 100,000 times with a long TTL leaves 100,000 entries in the heap. The rebuild is O(n), but it only happens after n stale pushes, so it's O(1) amortized. With it, the heap stays at most2 * capacity. - Use a monotonic clock.
time.time()can jump backwards (NTP, manual changes), which would resurrect expired entries or expire fresh ones.time.monotonic()can't. Takingclockas a parameter also makes the cache testable withoutsleep. - Purge only when full. Expired entries that nobody reads stay in memory until the
cache fills up or someone reads them. That's fine for a bounded cache, since they
only occupy slots that would otherwise be empty, but it means
len(cache.map)counts dead entries.
Complexity¶
| Operation | Cost |
|---|---|
get |
O(1) |
put |
O(log n) for the heap push |
purge inside put |
O(k log n) for k expired entries, O(log n) amortized |
The purge loop looks like it could be expensive, but each entry is pushed once and
popped at most once, so its cost is paid for by the put that inserted it.
Follow-up: can it be O(1)?¶
Yes, if every entry has the same TTL. Then expiry order is just write order: the entry written first expires first. A second linked list in write order replaces the heap, and its front is always the next entry to expire.
The catch, and the point of the question, is that you still need two lists. A
get refreshes recency but must not refresh expiry, so the write-order list can't be
the recency list.
Python's OrderedDict is a hash map over a doubly linked list, so two of them are
enough:
from collections import OrderedDict
import time
class FixedTTLLRUCache:
def __init__(self, capacity, ttl, clock=time.monotonic):
if capacity <= 0:
raise ValueError("capacity must be positive")
self.capacity = capacity
self.ttl = ttl
self.clock = clock
self.recency = OrderedDict() # key -> value, LRU first
self.written = OrderedDict() # key -> expires_at, oldest write first
def _remove(self, key):
del self.recency[key]
del self.written[key]
def get(self, key):
expires_at = self.written.get(key)
if expires_at is None:
return None
if expires_at <= self.clock():
self._remove(key)
return None
self.recency.move_to_end(key) # refresh recency, NOT the write order
return self.recency[key]
def put(self, key, value):
now = self.clock()
if key in self.written:
self._remove(key)
if len(self.written) >= self.capacity:
# Same TTL for everyone, so the oldest write expires first.
while self.written and next(iter(self.written.values())) <= now:
self._remove(next(iter(self.written)))
if len(self.written) >= self.capacity:
self._remove(next(iter(self.recency)))
self.recency[key] = value
self.written[key] = now + self.ttl
Every step is O(1) (amortized for the purge loop). This one also matches the brute-force reference on 3,000 random sequences.
With a small set of distinct TTLs (say 1 minute, 1 hour, 1 day) the same idea extends to one write-order list per TTL, and the purge checks the front of each. For arbitrary TTLs you're back to an ordered index, or to a timing wheel if you can accept expiry at bucket granularity.
How real systems do it¶
Real caches mostly don't keep a strict expiry order at all.
- Redis expires lazily, checking a key's TTL when it's accessed, and also runs an
active cycle that samples a handful of keys with TTLs, deletes the expired ones, and
repeats if a large fraction of the sample was expired. Eviction under
maxmemoryis approximate LRU: sample a few keys and evict the oldest among them, with no global recency list. - Caffeine (Java) uses a hierarchical timing wheel for variable per-entry expiry, which makes scheduling and expiring O(1) at the cost of coarse time buckets.
Both trade exactness for constant-time operations and less memory per entry. That's worth saying in an interview after presenting the exact version.
Critique¶
The implementation is correct against the reference, and it's what an interviewer expects. Here's where it falls short of something I'd ship.
Not thread-safe¶
get mutates the recency list, so even reads are writes. Each operation is several
separate pointer and dict updates, and the GIL only makes individual bytecodes atomic,
not the sequence. Two threads calling get concurrently can corrupt the list. The
minimum fix is one threading.Lock around every public method:
That serializes everything, including reads. At higher load you'd shard the cache by key hash (one lock and one LRU per shard, so LRU becomes per-shard rather than global), or buffer read events and apply them to the recency list in batches, which is what Caffeine does.
Interface rough edges¶
Nonemeans "miss", andNoneis also a value. Afterput("a", None, 60), a hit and a miss look the same. Use a sentinel default or raiseKeyError.ttl <= 0is accepted. The entry is written and immediately expired. Either reject it, or treat it as "don't cache".- No "no expiry" option. Callers must pass a TTL.
ttl=Nonemeaning "never expires" is a common need; it would keep the entry out of the heap entirely. - No
delete. Real caches need invalidation. It's easy to add here:_removealready does it, and lazy deletion handles the heap.
The policy question¶
"Expired first, then LRU" means a full cache with no expired entries behaves exactly
like plain LRU, so a short-TTL entry can outlive a long-TTL one just by being read
recently. Some designs prefer the opposite trade-off and evict whichever entry is
closest to expiring (Redis calls this volatile-ttl). Which one is right depends on
whether TTL means "this data goes stale" (then expired-first is correct and LRU is the
right tie-break) or "this data is less important" (then soonest-to-expire is a
reasonable proxy for value).
Summary¶
| Arbitrary TTLs | One fixed TTL | |
|---|---|---|
| Lookup | dict |
OrderedDict |
| Recency | doubly linked list | OrderedDict order |
| Expiry order | min-heap with lazy deletion | second OrderedDict (write order) |
put |
O(log n) | O(1) |
| Catch | stale heap entries need seq checks and periodic rebuilds |
get must not touch write order |