Anthropic OA: leveled in-memory record store

What the assessment is. A timed online assessment (about 90 minutes) where one class grows over four levels and a hidden test suite grades each level. Later levels are only unlocked once earlier tests pass, and the earlier tests keep running. The skill being graded is less "clever algorithm" and more "pick a data model on Level 1 that survives Level 4", plus careful reading of edge cases.

The spec used here (the real one differs in names and small details; the shape is the same):

Clarify first (each one is a hidden test somewhere): Is set on an existing field an overwrite (yes)? Does deleting an expired field count as a delete (no, it was already gone)? Is the TTL window half-open (yes: at timestamp + ttl the field is dead)? Does a non-timestamped set mean "never expires" (yes)? What does restore do when no backup qualifies (assume the tests guarantee one; return None and change nothing)?

Data model, chosen for Level 4 on Level 1. One dict: records: key -> {field: (value, expires_at)} with expires_at = None for no TTL. Every read goes through one helper, _live_fields(key, now), that filters out expired fields. The untimed Level 1 and 2 methods are just the timed ones called with now = None, meaning "no expiry check". So Level 3 adds a parameter, not a second code path.

Expiry: lazy, not eager. Never run a timer. A field with expires_at <= now is treated as absent by every read. Timestamps only increase, so a field that is dead now stays dead. Optionally purge it when touched so memory does not grow forever.

Backup and restore. A backup stores, for each live field, its value and its remaining TTL (expires_at - backup_ts, or None). Restore rebuilds with expires_at = restore_ts + remaining. Backups go in a list in timestamp order (they are taken in time order), so bisect finds "latest backup at or before t" in O(log B). The snapshot must be a deep copy: if it shares the inner dicts with live state, the next set silently edits the backup. This is the most common Level 4 bug.

Worked example.

• set_at_with_ttl("u1", "name", "ana", 10, 50) makes the field live on [10, 60).
• backup(30) records name with 30 left and returns 1.
• set_at("u1", "name", "bo", 40) overwrites it with no TTL.
• restore(100, 35) picks the backup from time 30. name is "ana" again with 30 left, so it expires at 130.
• get_at("u1", "name", 129) gives "ana", and at 130 it gives None.

from bisect import bisect_right


class RecordStore:
    """In-memory records of string fields, with TTL and point-in-time restore.

    records: key -> {field: (value, expires_at)}; expires_at None = forever.
    now=None in the helpers means "untimed call": no expiry check.
    """

    def __init__(self):
        self.records = {}
        self.backups = []  # [(timestamp, {key: {field: (value, remaining)}})]

    # -- helpers -----------------------------------------------------------
    @staticmethod
    def _alive(entry, now):
        _, expires_at = entry
        return now is None or expires_at is None or now < expires_at

    def _live_fields(self, key, now):
        fields = self.records.get(key, {})
        return {f: e for f, e in fields.items() if self._alive(e, now)}

    def _put(self, key, field, value, expires_at):
        self.records.setdefault(key, {})[field] = (value, expires_at)

    def _format(self, fields, prefix=""):
        return [f"{f}({v})" for f, (v, _) in sorted(fields.items()) if f.startswith(prefix)]

    # -- level 1 -----------------------------------------------------------
    def set(self, key, field, value):
        self._put(key, field, value, None)

    def get(self, key, field):
        return self.get_at(key, field, None)

    def delete(self, key, field):
        return self.delete_at(key, field, None)

    # -- level 2 -----------------------------------------------------------
    def scan(self, key):
        return self.scan_at(key, None)

    def scan_by_prefix(self, key, prefix):
        return self.scan_by_prefix_at(key, prefix, None)

    # -- level 3 -----------------------------------------------------------
    def set_at(self, key, field, value, timestamp):
        self._put(key, field, value, None)

    def set_at_with_ttl(self, key, field, value, timestamp, ttl):
        self._put(key, field, value, timestamp + ttl)

    def get_at(self, key, field, timestamp):
        entry = self.records.get(key, {}).get(field)
        if entry is None or not self._alive(entry, timestamp):
            return None
        return entry[0]

    def delete_at(self, key, field, timestamp):
        fields = self.records.get(key)
        if not fields or field not in fields:
            return False
        alive = self._alive(fields[field], timestamp)
        del fields[field]  # an expired field is purged either way
        if not fields:
            del self.records[key]
        return alive

    def scan_at(self, key, timestamp):
        return self._format(self._live_fields(key, timestamp))

    def scan_by_prefix_at(self, key, prefix, timestamp):
        return self._format(self._live_fields(key, timestamp), prefix)

    # -- level 4 -----------------------------------------------------------
    def backup(self, timestamp):
        snapshot = {}
        for key in self.records:
            live = self._live_fields(key, timestamp)
            if live:
                snapshot[key] = {
                    f: (v, None if exp is None else exp - timestamp)
                    for f, (v, exp) in live.items()
                }  # a new dict: later writes must not reach the backup
        self.backups.append((timestamp, snapshot))
        return len(snapshot)

    def restore(self, timestamp, timestamp_to_restore):
        i = bisect_right([t for t, _ in self.backups], timestamp_to_restore)
        if i == 0:
            return None
        _, snapshot = self.backups[i - 1]
        self.records = {
            key: {
                f: (v, None if left is None else timestamp + left)
                for f, (v, left) in fields.items()
            }
            for key, fields in snapshot.items()
        }
        return None

The list comprehension in restore makes it O(B) to find the backup. Keep a parallel backup_times list if restores are frequent; the tests here call it a handful of times, so clarity wins.

Let F be the number of fields in one record, R the number of records, N the total number of fields, and B the number of backups.

Space is O(N) for live state plus O(N) per backup. If backups are frequent and most data is unchanged between them, a persistent (copy-on-write) map makes each backup O(changes since the last one) instead.

1. Make backup cheap. Store a version log per field (field -> [(ts, value, expires_at)]) instead of copying everything. Restoring to time t is then a binary search per field. Better still is a persistent hash map (HAMT), where a backup is just keeping a pointer to the current root.
2. Expiry at scale. Lazy expiry leaks memory for keys nobody reads. Add a min-heap of (expires_at, key, field) and pop due entries on each call, skipping stale heap entries whose expiry no longer matches. Redis does a mix of the two: lazy on access plus a sampled sweep.
3. Concurrency. A single lock around the store is correct and simple. For more throughput, shard by key with one lock per shard. backup then needs either a global pause or per-shard snapshots taken at the same logical time.
4. Durability. Append every mutation to a write-ahead log, fsync in batches, and treat backup as a checkpoint so recovery replays only the log after it.
5. Out-of-order timestamps. If calls can arrive with older timestamps, lazy expiry stays correct for reads, but the "dead stays dead" argument breaks for purging. Purge only fields that expired before the oldest timestamp still allowed.

Mid-level (L4)

• Passes Levels 1 and 2 quickly, including empty records, missing keys and sorted output.
• Gets the TTL boundary right (dead at exactly timestamp + ttl) and treats expired fields as absent in every read.
• Reaches Level 4 within the time limit, even if restore is not fully right.

Senior (L5)

• Routes untimed methods through the timed ones so Level 3 adds no second code path.
• Snapshots by copy and stores remaining TTL, so restored fields expire relative to the restore time.
• Writes a few tests per level before submitting rather than relying on the hidden suite.

Staff (L6+)

• Finishes all four levels with time left and code another engineer could extend.
• Names the trade-off between eager and lazy expiry and between full-copy and versioned backups, with their costs.
• Explains how the same model becomes a durable store: write-ahead log, checkpoints, sharded locking.