21. Implement a TTL cache.
Implement an in-memory cache where every key expires after a time-to-live duration, and define how expired entries are detected and removed.
I would use a dictionary for the current record of each key and a min-heap for expiration records. Each heap item stores the expiration time, version, and key. Before every put or get, I pop records whose expiration time is less than or equal to now. A version check prevents an old heap record from deleting a newer value. Dictionary operations are O(1) on average, heap operations are O(log h), and auxiliary space is O(k + h).
See the Code while reading this explanation.
The cache stores values in memory and makes each value unavailable after a fixed TTL. Scanning every key on each operation would be wasteful. Instead, the solution uses a dictionary for direct key lookup and a min-heap for expiration order. A version number makes overwrites safe when an older expiration record is still inside the heap.
- What input sizes, value ranges, and edge cases should the solution handle?
- What output should be returned for empty, invalid, or duplicate input?
- Should I prioritize execution time or memory use, and may I use the standard library?
The constructor receives ttl_seconds.
put(key, value, now) stores the value under the key. Its expiration time is now + ttl_seconds.
get(key, now) returns the stored value when the current record has not expired. It returns None when the key is missing or expired.
A record is expired when expires_at <= now.
The dictionary maps each key to:
(value, expires_at, version)
It stores the newest record for that key.
The min-heap stores:
(expires_at, version, key)
Python heapq is a min-heap. Therefore, the record with the smallest expiration time is always at the top.
The version number identifies one specific write. It prevents an old heap record from deleting a newer value written under the same key.
The dictionary starts empty. The heap starts empty. next_version starts at zero.
Before every put and get, the cache runs _evict_expired(now).
After this cleanup, every record remaining in the dictionary has expires_at greater than now.
The heap may still contain stale records. A popped record may delete a dictionary entry only when both its expiration time and version match the current dictionary record.
The TTL is 5 seconds.
At t=0, put A=100.
The cache creates version 1 and calculates expires_at = 0 + 5 = 5. The dictionary becomes A -> (100, 5, 1). The heap becomes [(5, 1, A)].
At t=1, put B=200.
The cache creates version 2 and calculates expires_at = 1 + 5 = 6. The dictionary contains A -> (100, 5, 1) and B -> (200, 6, 2). The heap contains (5, 1, A) and (6, 2, B).
At t=3, put A=150.
No record has expired. The cache creates version 3 and calculates expires_at = 3 + 5 = 8. The dictionary replaces A with A -> (150, 8, 3). The heap receives (8, 3, A). The older record (5, 1, A) remains in the heap, but it is stale.
At t=5, get A.
Cleanup pops (5, 1, A). The current dictionary record for A is (150, 8, 3). The expiration time and version do not match, so the popped record is stale and cannot delete A. The cache returns 150. The dictionary remains unchanged. The heap contains (6, 2, B) and (8, 3, A).
At t=6, get B.
Cleanup pops (6, 2, B). It matches the current dictionary record for B, so B is deleted. The following lookup does not find B, so get returns None. The dictionary contains only A -> (150, 8, 3). The heap contains (8, 3, A).
At t=9, get A.
Cleanup pops (8, 3, A). It matches the current dictionary record for A, so A is deleted. The following lookup does not find A, so get returns None.
The returned values are [150, None, None]. The final dictionary is empty, and the final heap is empty.
The heap always exposes the next possible expiration in time order.
Cleanup removes every heap record whose expiration time is less than or equal to now.
A matching expiration time and version proves that the popped record still represents the current dictionary record. It is therefore safe to delete that key.
A mismatch proves that the heap record belongs to an older write. Ignoring it prevents a stale record from deleting a newer value.
After cleanup, every dictionary record is valid at the supplied time. Therefore, get returns a value only when that value has not expired.
The constructor validates the TTL and creates the dictionary, heap, and version counter.
_evict_expired repeatedly examines the heap top. It pops every record with expires_at <= now. It deletes a key only when the popped expiration time and version match the current dictionary record.
put runs cleanup first. It creates a new version, calculates the expiration time, updates the dictionary, and pushes one heap record.
get also runs cleanup first. It then returns the value from the current dictionary record or None when the key is absent.
Let h be the number of heap records before cleanup. Let r be the number of expired or stale records popped during one operation.
A put takes O(r log h + log h) time for cleanup and the new heap push, plus O(1) average dictionary work.
A get takes O(r log h) cleanup time plus O(1) average dictionary lookup.
Each heap record is pushed once and popped at most once. Cleanup work is therefore amortized across the sequence of operations.
Auxiliary space is O(k + h), where k is the number of current dictionary records. Repeated overwrites can make h larger than k because stale heap records remain until they reach the top.
A negative TTL is rejected. A TTL of zero makes an entry expire at the same timestamp. The next operation at that time or later removes it.
The key insight is to separate direct lookup from expiration order. The dictionary provides O(1) average access to the newest record for a key. The min-heap exposes the next possible expiration without scanning all dictionary entries. The central invariant is that after _evict_expired(now), every dictionary record has expires_at greater than now. Version numbers make lazy cleanup safe. A heap record can delete a key only when its expires_at and version still match the current dictionary record. Any mismatch means the heap record belongs to an older overwrite and must be ignored.
from __future__ import annotations
import heapq
from typing import Any, Optional
class TTLCache:
def __init__(self, ttl_seconds: int) -> None:
# Reject a TTL that would expire entries before they are written.
if ttl_seconds < 0:
raise ValueError("ttl_seconds must be non-negative")
# Every write expires after this fixed number of seconds.
self.ttl_seconds = ttl_seconds
# key -> (value, expires_at, version)
# The dictionary stores the newest record for each key.
self.store: dict[str, tuple[Any, int, int]] = {}
# Each heap record is (expires_at, version, key).
# heapq keeps the smallest expires_at at index 0.
self.expiry_heap: list[tuple[int, int, str]] = []
# Each write receives a unique increasing version.
self.next_version = 0
def _evict_expired(self, now: int) -> None:
# Pop every record whose expiration time has arrived.
while self.expiry_heap and self.expiry_heap[0][0] <= now:
expires_at, version, key = heapq.heappop(self.expiry_heap)
# The key may already have been removed.
current = self.store.get(key)
if current is None:
continue
_, current_expires_at, current_version = current
# Delete only when the popped heap record still represents
# the current dictionary record for this key.
if current_expires_at == expires_at and current_version == version:
del self.store[key]
def put(self, key: str, value: Any, now: int) -> None:
# Remove expired current records and stale heap records
# that have reached the top.
self._evict_expired(now)
# Give this write a new version.
self.next_version += 1
version = self.next_version
# Calculate the exact expiration time.
expires_at = now + self.ttl_seconds
# Store the newest record for the key.
self.store[key] = (value, expires_at, version)
# Add its expiration record to the min-heap.
heapq.heappush(
self.expiry_heap,
(expires_at, version, key),
)
def get(self, key: str, now: int) -> Optional[Any]:
# Remove records that are expired at this time.
self._evict_expired(now)
# Read the newest remaining record for the key.
current = self.store.get(key)
if current is None:
return None
# The first tuple item is the cached value.
return current[0]
if __name__ == "__main__":
cache = TTLCache(ttl_seconds=5)
# Exact example from the diagram.
cache.put("A", 100, now=0)
cache.put("B", 200, now=1)
cache.put("A", 150, now=3)
results = [
cache.get("A", now=5),
cache.get("B", now=6),
cache.get("A", now=9),
]
print(results) # [150, None, None]
print(cache.store) # {}
print(cache.expiry_heap) # []Let h be the number of heap records before cleanup, and let r be the number of expired or stale records popped during the operation. A put takes O(r log h + log h) time for cleanup and the new heap push, plus O(1) average dictionary work. A get takes O(r log h) time for cleanup, plus O(1) average dictionary lookup. Each heap record is pushed once and popped at most once, so cleanup work is amortized across many operations. Auxiliary space is O(k + h), where k is the number of current dictionary records. Repeated overwrites can make h larger than k.
This pattern is useful for in-memory caches, temporary authentication tokens, sessions, rate-limit state, deduplication records, and other data that becomes invalid after a fixed time. The dictionary supports fast access by key. The min-heap supports ordered lazy expiration without scanning every stored key during each operation.
This question tests whether you can combine data structures with different strengths. The interviewer is checking whether you choose fast dictionary lookup, use a min-heap for expiration order, handle overwrites without deleting newer values, define the expiration boundary correctly, maintain a clear invariant, and explain amortized heap cleanup accurately. It also tests whether your Python code and complexity claims match the behavior you describe.
One mistake is deleting a key whenever any old heap record expires. That can remove a newer value written under the same key. Compare both expires_at and version before deleting. Another mistake is running cleanup only during get. put must also clean expired records before writing. A third mistake is using expires_at < now instead of expires_at <= now. In this design, an entry is expired exactly at its expiration timestamp. Candidates may also claim every operation is O(1), even though heap pushes and pops cost O(log h). Finally, repeated overwrites can leave stale records in the heap, so space can grow beyond the number of current keys.
Say the invariant before coding: after cleanup at time now, every dictionary record expires after now, and a popped heap record may delete a key only when both its expiration time and version match the current record.
