Task manager with TTL expiry and time-travel snapshots

The spec used here. Timestamps are integers and grow across calls.

Clarify first. Can an expired task be updated back to life (no: it is gone)? Does update_task change the TTL (no)? What does tasks_at return for at before the user existed ([])? Is at always at most ts (yes)?

Design for Level 4 from Level 1: store versions, not just the current state. Each task keeps a history list of (ts, state) where state is (name, priority, expires_at), or None after a delete. Current state is the last entry. Any past state is a binary search on the history. Expiry needs no history entry: "alive at t" is state is not None and t < expires_at, which works for the present and the past alike.

So every write goes through one _record(task, ts, state) helper, and every read through one _state_at(task, t) helper. Levels 1–3 call _state_at(task, ts); Level 4 calls _state_at(task, at). No separate snapshot code.

Why not copy a snapshot per write? A full copy per write is O(tasks) per write and O(writes × tasks) memory. Per-task version lists cost O(1) per write and O(writes) memory, and a time-travel read is O(k log v) for a user with k tasks and v versions each.

Same-timestamp writes. If two writes to one task happen at the same ts, replace the last history entry instead of appending, so the search sees the final state at that time.

from bisect import bisect_right
from itertools import count


class Task:
    def __init__(self, task_id, user, seq):
        self.id, self.user, self.seq = task_id, user, seq
        self.times, self.states = [], []  # parallel lists, times ascending


class TaskManager:
    def __init__(self):
        self.tasks = {}       # task id -> Task
        self.by_user = {}     # user -> [Task] in creation order
        self._seq = count(1)

    # -- helpers -----------------------------------------------------------
    @staticmethod
    def _record(task, ts, state):
        if task.times and task.times[-1] == ts:
            task.states[-1] = state
        else:
            task.times.append(ts)
            task.states.append(state)

    @staticmethod
    def _state_at(task, t):
        i = bisect_right(task.times, t)
        if i == 0:
            return None
        state = task.states[i - 1]
        if state is None or (state[2] is not None and t >= state[2]):
            return None  # deleted or expired
        return state

    def _add(self, ts, user, name, priority, expires_at):
        seq = next(self._seq)
        task = Task(f"task{seq}", user, seq)
        self.tasks[task.id] = task
        self.by_user.setdefault(user, []).append(task)
        self._record(task, ts, (name, priority, expires_at))
        return task.id

    # -- level 1 -----------------------------------------------------------
    def add_task(self, ts, user, name, priority):
        return self._add(ts, user, name, priority, None)

    def get_task(self, ts, task_id):
        task = self.tasks.get(task_id)
        state = task and self._state_at(task, ts)
        if not state:
            return None
        return {"name": state[0], "priority": state[1], "user": task.user}

    def update_task(self, ts, task_id, name, priority):
        task = self.tasks.get(task_id)
        state = task and self._state_at(task, ts)
        if not state:
            return False
        self._record(task, ts, (name, priority, state[2]))
        return True

    def delete_task(self, ts, task_id):
        task = self.tasks.get(task_id)
        if not task or not self._state_at(task, ts):
            return False
        self._record(task, ts, None)
        return True

    # -- level 2 -----------------------------------------------------------
    def list_tasks(self, ts, user):
        return self.tasks_at(ts, user, ts)

    # -- level 3 -----------------------------------------------------------
    def add_task_with_ttl(self, ts, user, name, priority, ttl):
        return self._add(ts, user, name, priority, ts + ttl)

    # -- level 4 -----------------------------------------------------------
    def tasks_at(self, ts, user, at):
        live = []
        for task in self.by_user.get(user, []):
            state = self._state_at(task, at)
            if state:
                live.append((-state[1], task.seq, state[0]))
        return [name for _, _, name in sorted(live)]

Let k be a user's task count (ever created) and v the versions of one task.

Space is O(total writes). Long-lived systems compact: versions older than the oldest time anyone may query can be dropped, keeping the last one before that point.

1. Quotas. "A user may have at most N live tasks." Count live tasks at ts before adding. Lazy expiry makes the count need a scan; keep an expiry heap per user, or a live counter decremented when the heap pops due tasks.
2. Snapshots of the whole system at time t. The same per-task search over every task gives it in O(T log v). If such queries are frequent, a persistent balanced tree keyed by (priority, seq) per user makes each version share structure with the last.
3. Undo. Undo of the last write is popping the last history entry, which the version list gives for free. Make it a new write (record(previous state)) if history must stay append-only.
4. Out-of-order writes. If writes can arrive late, insert into the history with bisect.insort and decide whether a late update may change the past answer (usually yes for audit logs, no for user-facing reads).
5. Storage. In a database this is a versioned table: rows of (task_id, valid_from, valid_to, fields), with an index on (user_id, valid_from). A time-travel query is valid_from <= at AND (valid_to IS NULL OR valid_to > at).

Mid-level (L4)

• Gets CRUD and sorting right, including tie order and failure returns.
• Applies the TTL boundary consistently in every read, update and delete.
• Reaches Level 4 with a working, if slower, approach (for example, replaying a write log up to at).

Senior (L5)

• Stores per-task version history from Level 1 so time travel is a binary search, and routes all reads through one "state at t" helper.
• Treats expiry as part of the state, so it works for past reads with no extra history.
• Handles same-timestamp writes and explains the memory and time trade-off against snapshot copies.

Staff (L6+)

• Explains compaction and how this maps to a temporal table in a real database.
• Discusses quotas, undo and out-of-order writes as additions to the same model.
• Finishes with time left and clear tests per level.