Apple Design Task Manager：优先级堆、懒删除与版本号

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

• 训练版按标准操作 add(userId,taskId,priority)、edit(taskId,newPriority)、rmv(taskId)、execTop()；taskId 当前唯一，执行最高 priority，平局取最大 taskId，返回 userId；空任务返回 −1。
• 为稳健处理删除后重用相同 taskId/priority，代码增设单调版本号；原标准题是否允许重用，按实际接口核对。
• 非法重复 add 或缺失 edit/rmv 明确报错，题目未明确错误协议。

解题路线

1. 哈希表记录当前 taskId 对应的 userId、priority、version；堆条目为 (−priority,−taskId,version)，让 Python 最小堆弹出目标任务。
2. add 和 edit 写当前状态并压入新堆条目；rmv 只删当前表。旧堆项等到堆顶时检查 priority 和 version，不匹配即丢弃。
3. execTop 反复弹出直到找到仍是当前版本的任务，删除映射并返回 owner；堆空返回 −1。版本号防止 ABA：旧任务删除后同 ID、同优先级再次加入，旧条目不能冒充新任务。
4. 懒删除会积累旧条目；长期高频 edit 可在旧项远多于活项时重建堆，把内存压回 O(活任务数)。

手动走一遍

add(10,7,5)、add(20,9,5) 后，execTop 返回 20，因为 task 9 在相同优先级下更大。若 edit(7,8)，下一次 execTop 返回 10。删除后重加 task 7 仍由版本检查排除旧堆项。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

import heapq


class TaskManager:
    """LeetCode 3408-style training contract; task_id is globally unique."""

    def __init__(self, tasks: list[tuple[int, int, int]] = ()) -> None:
        self.current: dict[int, tuple[int, int, int]] = {}
        self.heap: list[tuple[int, int, int]] = []
        self.version = 0
        for user_id, task_id, priority in tasks:
            self.add(user_id, task_id, priority)

    def add(self, user_id: int, task_id: int, priority: int) -> None:
        if task_id in self.current:
            raise ValueError("duplicate task_id")
        self.version += 1
        self.current[task_id] = (user_id, priority, self.version)
        heapq.heappush(self.heap, (-priority, -task_id, self.version))

    def edit(self, task_id: int, priority: int) -> None:
        if task_id not in self.current:
            raise KeyError(task_id)
        user_id, _, _ = self.current[task_id]
        self.version += 1
        self.current[task_id] = (user_id, priority, self.version)
        heapq.heappush(self.heap, (-priority, -task_id, self.version))

    def rmv(self, task_id: int) -> None:
        if task_id not in self.current:
            raise KeyError(task_id)
        del self.current[task_id]

    def exec_top(self) -> int:
        while self.heap:
            neg_priority, neg_task_id, version = heapq.heappop(self.heap)
            task_id = -neg_task_id
            item = self.current.get(task_id)
            if item is not None and item[1] == -neg_priority and item[2] == version:
                del self.current[task_id]
                return item[0]
        return -1

复杂度

add/edit 均 O(log H)，rmv 平均 O(1)，execTop 每个弹出 O(log H)，其中 H 为历次压堆条目数；跨整段操作每条旧记录最多弹一次。空间 O(H+活任务数)，周期重建可控制陈旧项。

易错点与追问

容易失分的边界

• 只按优先级排序，忘记同分取更大 taskId。
• 编辑堆中原条目而破坏堆序。
• 删映射后不清陈旧堆顶。
• 仅比较 taskId/priority，重用相同 ID 和优先级时发生 ABA。
• 声称懒删除永久 O(当前任务数) 内存。

继续深挖

• 如果优先级可能下降，旧高优先级条目何时被过滤？
• 如何支持按用户查询和取消全部任务？
• 什么条件下重建堆？
• 并发 add 与 execTop 的线性化点放在哪里？