heapq Tie-Breaker Trap: The Priority Queue That Crashes on Equal Priorities2026-06-18
This scheduler pushes (priority, task) tuples onto a heap and pops them in priority order. It works perfectly in every unit test. Then production starts throwing TypeError at 3 a.m.
import heapq
class Task:
def __init__(self, name, payload):
self.name = name
self.payload = payload
def schedule(tasks):
queue = []
for priority, task in tasks:
heapq.heappush(queue, (priority, task))
order = []
while queue:
_, task = heapq.heappop(queue)
order.append(task.name)
return order
# Tests pass — all priorities are distinct
print(schedule([
(3, Task("backup", {"size": 100})),
(1, Task("alert", {"level": "high"})),
(2, Task("cleanup", {"files": 50})),
])) # ['alert', 'cleanup', 'backup'] ✓
# Production: two alerts fire in the same second
print(schedule([
(1, Task("alert_a", {"id": 1})),
(1, Task("alert_b", {"id": 2})), # same priority!
(2, Task("cleanup", {"files": 50})),
]))
# TypeError: '<' not supported between instances of 'Task' and 'Task'
Python's heapq orders tuples lexicographically. When two priorities tie, it falls through to compare the next element — the Task object itself. Task has no __lt__, so Python raises TypeError.
This is one of the nastiest bugs in the Python standard library because it is data-dependent and silent until the data collides:
Task with a dict or list would also fail — dicts aren't orderable, lists compare element-by-element and may recurse into the payload.dataclass(order=True) "fixes" the crash but introduces a worse bug: heap order now depends on attribute values, so identical-priority tasks pop in arbitrary semantic order, breaking FIFO guarantees callers may rely on.The correct fix is the same trick Python's own queue.PriorityQueue docs recommend: insert a monotonically increasing tiebreaker between the priority and the payload, so ties resolve by insertion order and the payload is never compared.
import heapq
import itertools
def schedule(tasks):
queue = []
counter = itertools.count() # unique, monotonic, cheap
for priority, task in tasks:
heapq.heappush(queue, (priority, next(counter), task))
order = []
while queue:
_, _, task = heapq.heappop(queue)
order.append(task.name)
return order
itertools.count() yields integers, integers are always orderable, and ties are now broken by insertion order — giving you a deterministic, stable priority queue. next(counter) is atomic at the bytecode level, so it's safe even under the GIL with multiple producers, though a true multi-threaded queue should use queue.PriorityQueue anyway.
If you want LIFO behavior for ties (e.g., newest-first within a priority class), negate the counter: (priority, -next(counter), task). Either way, the payload never participates in the comparison — which is the whole point.
