Problem 71 · Object & System Design
In-Memory Key-Value Store with TTL (Lazy + Active Expiry) Medium Classic DSA & systems
01 Problem
Build a KV store with set(key, value, ttl=None), get(key), delete(key), and sweep() that actively reclaims expired keys. Expired keys must never be returned.
02 Key insight
Lazy expiry on read guarantees correctness; active expiry via a min-heap of (expire_at, key) bounds memory for keys never read again. Heap entries can be stale after overwrites, so validate against the live record before deleting.
03 Solution
solution.py python 46 lines Copy
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import heapq
import time
from typing import Any, Callable
class TTLStore :
def __init__ (self , now: Callable[[], float ] = time.monotonic) -> None :
self .now = now
self .data: dict [str , tuple [Any, float | None ]] = {} # key -> (value, expire_at)
self .heap: list [tuple [float , str ]] = []
def set (self , key: str , value: Any, ttl: float | None = None ) -> None :
expire_at = None if ttl is None else self .now() + ttl
self .data[key] = (value, expire_at)
if expire_at is not None :
heapq.heappush(self .heap, (expire_at, key))
def _expired (self , expire_at: float | None ) -> bool :
return expire_at is not None and expire_at <= self .now()
def get (self , key: str , default: Any = None ) -> Any:
rec = self .data.get(key)
if rec is None :
return default
if self ._expired(rec[1 ]):
del self .data[key] # lazy expiry
return default
return rec[0 ]
def delete (self , key: str ) -> bool :
return self .data.pop(key, None ) is not None
def sweep (self , max_items: int = 1000 ) -> int :
"""Active expiry; bounded work per call. Returns keys removed."""
removed, t = 0 , self .now()
while self .heap and self .heap[0 ][0 ] <= t and max_items > 0 :
expire_at, key = heapq.heappop(self .heap)
max_items -= 1
rec = self .data.get(key)
if rec is not None and rec[1 ] == expire_at: # skip stale heap entries
del self .data[key]
removed += 1
return removed
def __len__ (self ) -> int :
return len (self .data)Try it first.
Reveal solution Time O(1) get; O(log n) set; O(k log n) sweep
Space O(n + heap entries)
tests.py · 3 passing tests.py python 10 lines Copy
1
2
3
4
5
6
7
8
9
10
clock = [0.0 ]
s = TTLStore(now=lambda : clock[0 ])
s.set ("a" , 1 , ttl=5 ); s.set ("b" , 2 , ttl=10 ); s.set ("c" , 3 )
clock[0 ] = 5
assert s.get("a" ) is None and s.get("b" ) == 2
s.set ("b" , 22 , ttl=100 ) # overwrite makes old heap entry stale
clock[0 ] = 50
assert s.sweep() == 0 and s.get("b" ) == 22 and len (s) == 2
clock[0 ] = 200
assert s.sweep() == 1 and s.get("c" ) == 3 and len (s) == 1 04 Staff-level follow-up
Heap grows with overwrites; at scale Redis instead samples 20 random TTL keys per cycle and repeats if >25% expired, trading precision for O(1) memory overhead. A timing wheel is another option.