mission-peace/interview · error · KeyError

Task not found

Error message

Task not found

What it means

change_task_priority removes and re-inserts the task to update its priority, but only if the task exists in entry_finder; otherwise it raises KeyError("Task not found"). Priority changes are only meaningful for tracked tasks.

Solutions

1. Verify the task exists (task in pq.entry_finder) before changing priority
2. Add the task first if it may not exist, then change priority
3. Catch KeyError and treat as add-or-update: pq.add_task(priority, task) on miss

Example fix

// before
pq.change_task_priority(new_priority, task)
// after
try:
    pq.change_task_priority(new_priority, task)
except KeyError:
    pq.add_task(new_priority, task)

if task in pq.entry_finder:
    pq.change_task_priority(priority, task)

def is_tracked(pq, task):
    return task in pq.entry_finder

try:
    pq.change_task_priority(priority, task)
except KeyError:
    pq.add_task(priority, task)

When it happens

Trigger: Calling change_task_priority(priority, task) for a task never added, or one already popped/removed from the queue.

Common situations: Reprioritizing a task after a worker already popped it, updating tasks from a different queue instance, or a task key type mismatch (string vs tuple id).

Understand the failure class

Background: Record Not Found Errors: "not found", RecordNotFound, and "was not found" — what they mean and how to fix them — this error's family across 28 libraries.

Related errors

• task does not exist
• Key already exists
• pop from an empty priority queue

AI-assisted analysis of mission-peace/interview@94be5deb0c (2026-09-08). Data as JSON: /api/errors/a92894ee98b54e5a. Report an issue: GitHub.

Appendix: source

Thrown at python/graph/priorityqueue.py:37

            return True
        else:
            return False

    def get_task_priority(self, task):
        if task in self.entry_finder:
            return (self.entry_finder[task])[0]
        raise ValueError("task does not exist")
        
    def add_task(self, priority, task):
        if task in self.entry_finder:
            raise KeyError("Key already exists")
        entry = [self.mul*priority, False, task]
        self.entry_finder[task] = entry
        heappush(self.pq, entry)

    def change_task_priority(self, priority, task):
        if task not in self.entry_finder:
            raise KeyError("Task not found")
        self.remove_task(task)
        entry = [self.mul*priority, False, task]
        self.entry_finder[task] = entry
        heappush(self.pq, entry)
 
    def remove_task(self, task):
        entry = self.entry_finder.pop(task)
        entry[1] = True

    def pop_task(self):
        while self.pq:
            priority, removed, task = heappop(self.pq)
            if removed is False:
                del self.entry_finder[task]
                return task
        raise KeyError("pop from an empty priority queue")

    def peek_task(self):

View on GitHub (pinned to 94be5deb0c)