Implement a crash-resilient LRU cache
Company: Anthropic
Role: Software Engineer
Category: Coding & Algorithms
Difficulty: medium
Interview Round: Onsite
Implement an LRU-based memoization helper with behavior similar to a standard Python LRU cache.
You are given an interface like this:
```python
class LRU:
def __init__(self, capacity: int, persistence_path: str):
...
def generate_key(self, func, *args, **kwargs):
# return a deterministic, hashable cache key
pass
def call(self, func, *args, **kwargs):
# if the result for this function call is cached, return it
# otherwise compute it, cache it, and return it
pass
```
Requirements:
1. Cache results of pure function calls.
2. The cache key must include the function identity and its arguments.
3. `generate_key` must handle both positional and keyword arguments.
4. Different keyword argument orders must produce the same key.
5. When the cache exceeds `capacity`, evict the least recently used entry.
6. Assume arguments and return values are serializable.
Follow-up: if the process crashes and the in-memory cache is lost, how would you persist enough information to restore the cache after restart while keeping the cache correct? Describe the data you would write, when you would write it, and how recovery would work.
Overview: This question evaluates knowledge of cache design, least-recently-used (LRU) eviction policies, deterministic memoization key construction for function identity and arguments, and persistence mechanisms for crash resilience in the Coding & Algorithms domain.
Read the full Anthropic Software Engineer interview experience this question came from
Part 1: Implement an in-memory LRU-based memoization helper
Simulate an **in-memory LRU (Least Recently Used) memoization cache** over a sequence of pure-function calls, and report each call's result together with how many real computations were actually performed.
## What to implement
```python
def solution(capacity, calls):
...
```
Process the `calls` in order against a cache that can hold at most `capacity` distinct entries. For each call, either return its cached result (a **hit**) or compute it fresh and cache it (a **miss**), evicting the least recently used entry whenever the cache would exceed `capacity`.
## Input
- **`capacity`** — a non-negative integer, the maximum number of entries the cache may hold.
- **`calls`** — a list of calls, each a 3-tuple `(func_name, args, kwargs)`:
- `func_name` — a string naming the function to invoke (see below).
- `args` — a list of positional arguments.
- `kwargs` — a dict of keyword arguments.
## Supported functions
`func_name` is always one of:
| Name | Signature | Result |
|------|-----------|--------|
| `'add'` | `add(a, b)` | `a + b` |
| `'mul'` | `mul(a, b)` | `a * b` |
| `'pow'` | `pow(base, exp)` | `base ** exp` |
| `'affine'` | `affine(x, scale=1, bias=0)` | `scale * x + bias` |
Every call's `args`/`kwargs` are valid for the named function (so `scale` and `bias` may be passed positionally or by keyword, or omitted to use their defaults).
## Cache key
A cache entry is identified by the combination of:
1. the **function name**,
2. the **positional arguments** (`args`, in order), and
3. the **keyword arguments** (`kwargs`).
**Keyword order must not matter.** Two calls with the same keyword pairs in a different order refer to the *same* cache entry — e.g. `affine(5, scale=2, bias=1)` and `affine(5, bias=1, scale=2)` share one entry.
## Per-call behavior
For each call, in order:
- **Hit** — if its key is already in the cache: return the cached value and mark that entry as **most recently used** (do **not** recompute).
- **Miss** — otherwise: invoke the function to compute the value, count this as one real computation, and return the value. Then, **only if `capacity > 0`**, store the value under its key as the most recently used entry; if this makes the cache size exceed `capacity`, **evict the least recently used entry**.
Note that when `capacity == 0`, nothing is ever stored, so every call is a miss and is recomputed.
## Output
Return a tuple `(results, computed_count)` where:
- **`results`** — the list of returned values, one per call, in call order.
- **`computed_count`** — the total number of misses, i.e. the number of times a value was actually computed (cache hits are not counted).
## Constraints
- `0 <= capacity <= 100000`
- `0 <= len(calls) <= 100000`
- `func_name` is one of `'add'`, `'mul'`, `'pow'`, `'affine'`
- All provided calls are valid for the named function
- All argument values are integers and keyword arguments are hashable
Constraints
- 0 <= capacity <= 100000
- 0 <= len(calls) <= 100000
- func_name is one of 'add', 'mul', 'pow', 'affine'
- All provided calls are valid for the named function
- All argument values are integers and keyword arguments are hashable
Examples
Input: (2, [('add', [1, 2], {}), ('add', [1, 2], {})])
Expected Output: ([3, 3], 1)
Explanation: The second call is a cache hit.
Input: (2, [('affine', [5], {'scale': 2, 'bias': 1}), ('affine', [5], {'bias': 1, 'scale': 2})])
Expected Output: ([11, 11], 1)
Explanation: Keyword argument order should not change the cache key.
Hints
- Use a canonical key like (func_name, tuple(args), tuple(sorted(kwargs.items()))).
- A hash map plus an OrderedDict-style structure makes LRU updates efficient.
Part 2: Recover a crash-resilient LRU cache from a write-ahead journal
Reconstruct the exact state of an **LRU cache** after a crash by replaying its **write-ahead journal**.
## Background
The cache persists durably: before any operation is considered successful, it appends a journal record describing that operation. After a crash, only a **prefix** of the journal survived (passed to you as `journal`). Replaying that surviving prefix in order, with LRU eviction, reproduces the exact recovered cache state.
Implement:
```python
def solution(capacity, journal):
...
```
## Input
- **`capacity`** — the maximum number of entries the cache may hold.
- **`journal`** — a list of records, already in durable (chronological) order. Each record is a tuple/list of one of two forms:
- **PUT:** `('PUT', func_name, args, kwargs, value)` — the computed `value` for this key is now stored, and the key becomes the **most-recently-used (MRU)** entry.
- **HIT:** `('HIT', func_name, args, kwargs)` — an existing key was read, so it becomes the **MRU** entry.
In every record, `func_name` is a string, `args` is a **list**, and `kwargs` is a **dict**.
## Canonical key
Each record identifies a cache entry by a **canonical key** built as:
```python
(func_name, tuple(args), tuple(sorted(kwargs.items())))
```
Sorting the `kwargs` items by name means different keyword-argument orders map to the **same** key (e.g. `{'scale': 2, 'bias': 1}` and `{'bias': 1, 'scale': 2}` are identical).
## Replay rules
Process records in order, maintaining LRU recency:
- **PUT:** Store the value under the canonical key and mark that key as **MRU**. If this makes the number of stored entries exceed `capacity`, evict the **least-recently-used (LRU)** entry.
- **HIT:** If the canonical key is **currently present**, mark it as **MRU**. If the key is **not present** during replay, **ignore** the record (no insertion, no error).
## Output
Return the recovered cache contents ordered from **most-recently-used to least-recently-used**, as a list of `(canonical_key, value)` pairs, where `canonical_key` is the tuple described above.
## Edge cases
- If `capacity` is `0`, no entry can ever be stored — return an empty list.
- If `journal` is empty, return an empty list.
## Constraints
- `0 <= capacity <= 100000`
- `0 <= len(journal) <= 100000`
- Each record's operation is either `'PUT'` or `'HIT'`.
- All argument values and stored values are integers.
- Records are already in durable journal order.
Constraints
- 0 <= capacity <= 100000
- 0 <= len(journal) <= 100000
- Each record type is either 'PUT' or 'HIT'
- All argument values and stored values are integers
- Records are already in durable journal order
Examples
Input: (2, [('PUT', 'add', [1, 2], {}, 3), ('PUT', 'mul', [2, 3], {}, 6), ('HIT', 'add', [1, 2], {})])
Expected Output: [(('add', (1, 2), ()), 3), (('mul', (2, 3), ()), 6)]
Explanation: The hit on 'add' makes it the most recently used entry.
Input: (2, [('PUT', 'add', [1, 2], {}, 3), ('PUT', 'mul', [2, 3], {}, 6), ('PUT', 'pow', [2, 5], {}, 32), ('HIT', 'add', [1, 2], {})])
Expected Output: [(('pow', (2, 5), ()), 32), (('mul', (2, 3), ()), 6)]
Explanation: When 'pow' is inserted, 'add' is evicted as the least recently used entry. The later hit on 'add' is ignored because that key is no longer present.
Approach
Approach. The recovered cache is just the result of replaying the surviving journal prefix in order, so the code simulates an LRU cache record by record. An OrderedDict is the natural data structure: insertion order is preserved, and we treat the rightmost end as most-recently-used (MRU) and the leftmost as least-recently-used (LRU).
Canonical key. Every record's key is normalized via make_key to (func_name, tuple(args), tuple(sorted(kwargs.items()))). Sorting the kwargs items makes {scale:2, bias:1} and {bias:1, scale:2} collapse to the same key, which is why test 3's HIT with reordered kwargs correctly refreshes the existing entry.
Replay rules.
- HIT: only move_to_end(key) if the key is present; a HIT on a missing key is silently ignored, exactly as required.
Edge handling. capacity <= 0 short-circuits to [], so the capacity-0 case (test 4) never stores anything.
Output. OrderedDict holds entries LRU→MRU; the code copies items() and reverse()s them so the result is MRU→LRU as (canonical_key, value) pairs.
Correctness. Because the journal records the durable order of operations and each replay step reproduces the cache's own LRU bookkeeping, replaying yields the exact crash-recovered state.
Time complexity: O(r * k log k)
Space complexity: O(capacity)
Hints
- Replay the journal exactly in order; a durable log is only as good as its replay rules.
- Use the same normalized key for both 'PUT' and 'HIT', and keep LRU order in an OrderedDict-style structure.
Community answers
Answer by sujeet.banerjee.apal01
from collections import OrderedDict
def mul(args, kwargs): prod = 1 # print(args) # print(kwargs) for arg in args: # print(arg) prod *= arg return prod
def eval(fn): (first, second, third) = fn #print(f"{first} ==> {second} and {third}") if first == 'add': return sum(second, dict(third)) elif first == 'mul': # why second sometimes, but not other times!! return mul(second, dict(third)) elif first == 'pow': return second[0] second[1] elif first == 'affine': third = dict(third) return second[0] third['scale'] + third['bias'] return None
def immute_key(fn: tuple): (first, second, third) = fn return first, tuple(second), frozenset(third.items())
def solution(capacity, calls): print(f"Capacity {capacity}") print(f"Calls {calls}") lru_cache = OrderedDict() exec_count = 0 res_list = [] calls_seq = [immute_key(call) for call in calls] for call in calls_seq: if call not in lru_cache: # drop elements from 0 to (len(lru_cache)-capacity) while len(lru_cache) >= capacity: lru_cache.popitem(last=False) # Evaluate and update cache lru_cache[call] = eval(call) # count execution exec_count += 1 else: ret = lru_cache[call] lru_cache.move_to_end(call) res_list.append(lru_cache[call]) return res_list, exec_count
print(solution(2, [('add', [1, 2], {}), ('a
Answer by QL 2026
Hi, I can't select other languages like Java to code, but only R/SQL/Python/PostgreSQL. Do I miss anything for coding language?
Answer by vimmer
import hashlibimport picklefrom collections import OrderedDict
def solution(capacity, calls): lru_cache = LRUCache(capacity) results = [] for call in calls: result = lru_cache.call(call) results.append(result) return (results, lru_cache.count)
class LRUCache: def init(self, capacity: int): self.capacity = capacity self.cache = OrderedDict() self.count = 0 self.func = { "add": lambda x, y: x + y, "mul": lambda x, y: x y, "pow": lambda base, exp: baseexp, "affine": lambda x, scale=1, bias=0: scale * x + bias, }
def hash_key(self, call): payload = {"f": call[0], "args": call[1], "kwargs": sorted(call[2].items())} blob = pickle.dumps(payload) return hashlib.sha256(blob).hexdigest()
def execute(self, call): self.count += 1 return self.func[call[0]](call[1], call[2])
def call(self, call): key = self.hash_key(call) if key in self.cache: self.cache.move_to_end(key) return self.cache[key] else: result = self.execute(call) self.cache[key] = result if len(self.cache) > self.capacity: self.cache.popitem(last=False) return result
Answer by vimmer
import hashlib
import pickle
from collections import OrderedDict
def solution(capacity, calls):
lru_cache = LRUCache(capacity)
results = []
for call in calls:
result = lru_cache.call(call)
results.append(result)
return (results, lru_cache.count)
class LRUCache:
def init(self, capacity: int):
self.capacity = capacity
self.cache = OrderedDict()
self.count = 0
self.func = {
"add": lambda x, y: x + y,
"mul": lambda x, y: x * y,
"pow": lambda base, exp: base**exp,
"affine": lambda x, scale=1, bias=0: scale * x + bias,
}
def hash_key(self, call):
payload = {"f": call[0], "args": call[1], "kwargs": sorted(call[2].items())}
blob = pickle.dumps(payload)
return hashlib.sha256(blob).hexdigest()
def execute(self, call):
self.count += 1
return self.func[call[0]](call[1], call[2])
def call(self, call):
key = self.hash_key(call)
if key in self.cache:
self.cache.move_to_end(key)
return self.cache[key]
else:
result = self.execute(call)
self.cache[key] = result
if len(self.cache) > self.capacity:
self.cache.popitem(last=False)
return result
if name == "main":
# res = solution(2, [("add", [1, 2], {}), ("add", [1, 2], {})])
res = solution(
2,
[
("affine", [5], {"scale": 2, "bias": 1}),
("affine", [5], {"bias": 1, "scale": 2}),
],
)
print(res)
Answer by trevor_reznik
Is it possible to have c++ supported for this question?