Heaps and priority queues, from first idea to production

Every time Python code runs await asyncio.sleep(1) on the standard event loop, one line in CPython’s asyncio/base_events.py executes: heapq.heappush(self._scheduled, timer). The loop keeps every pending timer in a heap, and before each wait for network events it reads the top of that heap to decide how long it may sleep. A web server with ten thousand open connections, each with a timeout, is a heap of ten thousand timers that answers “which one fires first?” in one step.

That question, “what’s the most urgent thing right now?”, is what a heap answers, and it turns up far beyond timers: the next job to run, the closest unvisited city on a map, the ten slowest requests in a log, the next event in a simulation. A heap is the structure behind all of them, and it’s simpler than its reputation. It’s a list with one rule.

By the end you’ll be able to use Python’s heapq without its usual traps, build a heap from scratch and explain why each operation costs what it does, solve the classic heap interview problems, change priorities without a decrease-key operation, and reason about timers, schedulers and job queues at scale. The last section is honest about ezychat.io, the multi-tenant AI messaging platform I’m the primary engineer of: its backend never imports heapq, yet priority queues run under every request. Start at Beginner if “priority queue” is new. If you’ve used heapq.heappush, start at Intermediate.

Beginner: the most urgent thing next

Why first come, first served is wrong in an emergency room

Five people walk into an emergency room in this order: a sprained ankle, chest pain, a cut finger, a broken arm, and someone with stroke symptoms. A queue at a bakery would serve them in that order. An emergency room doesn’t. A triage nurse gives each patient a level, 1 for “can’t wait” up to 5 for “can wait hours”, and the next doctor sees the most urgent patient, however recently they arrived.

Here’s the difference in code. A deque is Python’s first-in, first-out queue; heapq is the priority version:

from collections import deque
import heapq

# (patient, triage level): level 1 is the most urgent
arrivals = [("sprained ankle", 4), ("chest pain", 1), ("cut finger", 5),
            ("broken arm", 3), ("stroke symptoms", 1)]

waiting_room = deque(arrivals)
print("first come:", [name for name, _ in waiting_room])

triage = []
for order, (name, level) in enumerate(arrivals):
    heapq.heappush(triage, (level, order, name))
print("most urgent:", [heapq.heappop(triage)[2] for _ in range(len(arrivals))])
# first come: ['sprained ankle', 'chest pain', 'cut finger', 'broken arm', 'stroke symptoms']
# most urgent: ['chest pain', 'stroke symptoms', 'broken arm', 'sprained ankle', 'cut finger']

Two patients share level 1, and chest pain arrived first, so chest pain goes first. That’s what the order number in the middle of each tuple is for. Hold on to it; it fixes a real bug later.

What a priority queue promises

A priority queue is any container that makes one promise: whenever you take something out, you get the item with the best priority. It doesn’t promise anything about the order of the rest. It has four operations:

“Most urgent” means smallest or largest, depending on the convention. A min-priority queue hands back the smallest value first, which suits triage levels, deadlines and distances, where smaller means sooner. A max-priority queue hands back the largest, which suits scores and bids. Python’s heapq is a min-priority queue. You’ll see how to get the other kind in a moment.

A priority queue is a contract, not a data structure. You could keep the items in a plain list and still keep the promise. The question is what the promise costs.

Two obvious ways to build one, and what they cost

An unsorted list. Push appends to the end: one step. Pop has to find the smallest item, and in an unsorted list it could be anywhere, so it looks at every item, then walks the list again to delete it. With n items waiting, every pop costs about n steps. Steps that grow in a straight line with n are written O(n); steps that don’t grow at all are O(1).

A sorted list. Keep the items in order, most urgent at the end, and pop is one step: take the last item. Now push is the slow one. Finding the right spot is quick with binary search, but inserting into the middle of a Python list shifts every item after it one place along. That’s O(n) per push.

Either way, one of the two operations walks the whole collection. In a system that does both all day, a busy queue pays for that walk on every request. Here are all three, a heap included, doing the same work: two arrivals and one pop per round, so the queue grows as it runs.

import heapq, random, time
from bisect import insort

class UnsortedQueue:
    def __init__(self): self.items = []
    def push(self, p): self.items.append(p)              # just add it at the end
    def pop(self):
        best = min(self.items)                           # look at every item
        self.items.remove(best)                          # ...and walk again to delete it
        return best

class SortedQueue:
    def __init__(self): self.items = []                  # kept largest first
    def push(self, p): insort(self.items, -p)            # find the spot, shift the rest
    def pop(self): return -self.items.pop()              # most urgent is at the end

class HeapQueue:
    def __init__(self): self.items = []
    def push(self, p): heapq.heappush(self.items, p)
    def pop(self): return heapq.heappop(self.items)

def run(cls, n, seed=1):
    rng = random.Random(seed)
    q, out = cls(), []
    start = time.perf_counter()
    for _ in range(n):              # a busy queue: two arrivals for every patient seen
        q.push(rng.random()); q.push(rng.random())
        out.append(q.pop())
    return time.perf_counter() - start, out

for n in (10_000, 40_000):
    results = {cls.__name__: run(cls, n) for cls in (UnsortedQueue, SortedQueue, HeapQueue)}
    same = len({tuple(out) for _, out in results.values()}) == 1
    times = "   ".join(f"{name}: {t * 1000:8.1f} ms" for name, (t, _) in results.items())
    print(f"n={n:>6,}  {times}   same answers: {same}")
# n=10,000  UnsortedQueue:    673.5 ms   SortedQueue:     16.1 ms   HeapQueue:      4.5 ms   same answers: True
# n=40,000  UnsortedQueue:  11185.2 ms   SortedQueue:    230.1 ms   HeapQueue:     19.9 ms   same answers: True

Measured on my laptop (an M1 Pro, Python 3.13). All three give the same answers. Four times the work makes the unsorted list 17 times slower and the sorted list 14 times slower: both are doing a walk that grows with the queue, inside a loop that also grows. The heap gets 4.4 times slower, close to the 4 you’d hope for. At 40,000 rounds it’s 560 times faster than the unsorted list.

The sorted list looks respectable at 10,000 because shifting memory is something CPUs do very fast. Don’t let that fool you. It’s the same shape of cost as the unsorted list, with a smaller constant in front.

The heap property, in pictures

A heap gets both operations down to a few steps by being only partly sorted. Picture the items arranged as a family tree: one item at the top, called the root, each item with up to two children below it, and every item except the root having exactly one parent above it. The rule, the heap property, is:

Every parent is smaller than or equal to each of its children.

That’s the whole rule. It says nothing about left versus right, and nothing about cousins. A 5 can sit on one side of the tree and a 4 lower down on the other.

Follow the rule up from any item and the values can only get smaller, so the smallest item in the whole heap must be the root. Peek is free: look at the top.

The second rule is about shape. A heap is a complete binary tree: every level is full except possibly the last, and the last level fills from the left. That shape keeps the tree as short as possible. A tree with 1,000,000 items in this shape has just 20 levels, because each level doubles the room: 1, 2, 4, 8, … and 2²⁰ is just over a million. Pushing or popping only ever moves an item along one path from top to bottom, so it costs at most about 20 steps, not a million. That count, “how many times can you double before you reach n”, is called log₂ n, and costs that grow like it are written O(log n).

Using heapq

Python’s heap lives in the heapq module, and it’s not a class. It’s a set of functions that treat an ordinary list as a heap:

import heapq

heap = []
for x in [5, 3, 8, 1, 9, 2]:
    heapq.heappush(heap, x)
print(heap)                                 # a heap, not a sorted list
print(heap[0])                              # the smallest is always at index 0
print(heapq.heappop(heap), heapq.heappop(heap))
print(heap)                                 # still a valid heap

nums = [5, 3, 8, 1, 9, 2]
heapq.heapify(nums)                         # rearrange an existing list in place
print(nums)
# [1, 3, 2, 5, 9, 8]
# 1
# 1 2
# [3, 5, 8, 9]
# [1, 3, 2, 5, 9, 8]

[1, 3, 2, 5, 9, 8] isn’t sorted: 3 comes before 2. It’s a heap, which is weaker and cheaper. Only heap[0] has a guaranteed meaning. Printing a heap and expecting sorted output is the most common beginner surprise with heapq.

Two habits from the start. First, heapq only knows “smallest first”. When you want the largest first, push the negated number (-score) and negate again when you pop. Second, when priorities can tie, push a tuple with a counter in the middle, like the triage example did. Intermediate explains why both work.

import heapq

class SupportQueue:
    def __init__(self):
        self.heap = []
        self.arrived = 0                       # arrival counter: the tie-breaker

    def add(self, ticket, urgency):            # urgency 1 is the most urgent
        heapq.heappush(self.heap, (urgency, self.arrived, ticket))
        self.arrived += 1

    def next(self):
        return heapq.heappop(self.heap)[2] if self.heap else None

q = SupportQueue()
q.add("can't log in", 2)
q.add("typo on pricing page", 4)
q.add("payments failing", 1)
q.add("password reset email missing", 2)
print([q.next() for _ in range(5)])
# ['payments failing', "can't log in", 'password reset email missing', 'typo on pricing page', None]

import heapq

def last_stone(stones):
    heap = [-s for s in stones]                # negate: heapq only knows "smallest first"
    heapq.heapify(heap)
    while len(heap) > 1:
        a = -heapq.heappop(heap)               # heaviest
        b = -heapq.heappop(heap)               # second heaviest
        if a != b:
            heapq.heappush(heap, -(a - b))
    return -heap[0] if heap else 0

print(last_stone([2, 7, 4, 1, 8, 1]), last_stone([3, 3]), last_stone([10]))
# 1 0 10

Intermediate: how a binary heap works

A tree stored in a list

The heap above prints as a list, not a tree, because it is a list. The complete shape is what makes that possible. Number the tree’s positions level by level, left to right, starting at 0: the root is 0, its children are 1 and 2, their children are 3, 4, 5, 6, and so on. Because every level fills from the left with no gaps, those numbers are exactly the list indices, and the family relationships become arithmetic:

• the children of index i are at 2*i + 1 and 2*i + 2;
• the parent of index i is at (i - 1) // 2.

No node objects, no pointers, no wasted memory: a heap of a million floats is one list of a million references. That’s a large part of why heaps are fast in practice, not only in theory.

Push: add at the end, then sift up

To push, put the new item in the next free spot, which is the end of the list. The shape stays complete. The heap property might now be broken between the new item and its parent, so repair it: compare the item with its parent, and if it’s smaller, swap them. Repeat from its new position until the parent is smaller or equal, or the item reaches the root. This is sift-up.

Push 1 into [3, 5, 8, 9]. It lands at index 4. Its parent is index (4−1)//2 = 1, holding 5. 1 is smaller: swap, and 1 is at index 1. Its parent is index 0, holding 3. Swap again. It’s at the root, so it stops: [1, 3, 8, 9, 5]. Two comparisons, two swaps.

Pop: move the last item to the top, then sift down

To pop, take the root: that’s the answer. Now there’s a hole at the top. Fill it with the last item in the list, which keeps the shape complete, and repair downward: compare the item with its two children, and if the smaller child is smaller than it, swap with that child. Repeat until both children are bigger or there are no children. This is sift-down. It must swap with the smaller child; swapping with the larger one would put a bigger value above a smaller one.

Here’s the whole structure in about forty lines, with a comparison counter so you can see the cost:

class MinHeap:
    """A binary min-heap in a plain list, counting every comparison."""

    def __init__(self):
        self.a = []
        self.comparisons = 0

    def _less(self, i, j):
        self.comparisons += 1
        return self.a[i] < self.a[j]

    def push(self, x):
        self.a.append(x)                        # 1. add at the end: the tree stays complete
        i = len(self.a) - 1
        while i > 0:                            # 2. sift up
            parent = (i - 1) // 2
            if not self._less(i, parent):
                break                           # parent is smaller or equal: done
            self.a[i], self.a[parent] = self.a[parent], self.a[i]
            i = parent

    def pop(self):
        a = self.a
        top = a[0]
        last = a.pop()                          # 1. take the last leaf off
        if a:
            a[0] = last                         # 2. put it at the root
            i = 0
            while True:                         # 3. sift down
                left, right = 2 * i + 1, 2 * i + 2
                smallest = i
                if left < len(a) and self._less(left, smallest):
                    smallest = left
                if right < len(a) and self._less(right, smallest):
                    smallest = right
                if smallest == i:
                    break                       # both children are bigger: done
                a[i], a[smallest] = a[smallest], a[i]
                i = smallest
        return top

h = MinHeap()
for x in [7, 4, 9, 1, 6, 3]:
    h.push(x)
print(h.a, "comparisons:", h.comparisons)
print([h.pop() for _ in range(6)], "comparisons:", h.comparisons)
# a large one, to check it against sorted()
import random
big = MinHeap()
nums = [random.random() for _ in range(100_000)]
for x in nums:
    big.push(x)
print([big.pop() for _ in range(len(nums))] == sorted(nums))
# [1, 4, 3, 7, 6, 9] comparisons: 7
# [1, 3, 4, 6, 7, 9] comparisons: 15
# True

Popping everything returns the items in sorted order, which is a good test of any heap you write: compare it against sorted() on a hundred thousand random numbers and you’ll catch almost every off-by-one.

Now watch it work. Push a few small numbers to see them climb, then pop and watch the last leaf sink from the root:

Things to try: push 1, then 99, and compare the comparison counts. The 1 climbs all the way to the root; the 99 stops after one comparison. Pop a few times and notice that sift-down usually goes all the way to the bottom: the item moved to the root came from the bottom row, so it tends to be large. Then press “Heapify 15 random” and count how few comparisons building a whole heap takes.

Why each operation is O(log n)

Sift-up moves one item along one path toward the root, with at most one comparison per level. Sift-down moves one item along one path toward the leaves, with at most two comparisons per level (left child against right child, then the smaller one against the item). A complete tree of n items has ⌊log₂ n⌋ + 1 levels. So:

The last two rows matter. A heap is bad at “is X in here?” and “remove X”: the heap property tells you nothing about where a non-minimum item is. Keep that in mind for Advanced.

Building a heap in O(n): heapify

Given a list of n items, you could push them one at a time: n pushes at O(log n) each, O(n log n) in total. heapify does better by working bottom-up. Every leaf is already a one-item heap, and half the items in a complete tree are leaves. So start at the last parent, index (n−2)//2, sift it down, and walk backwards to the root, sifting each one down. When you reach an item, both of its subtrees are already heaps, so one sift-down makes its own subtree a heap.

Why is that linear when each sift-down can cost O(log n)? Because the expensive sifts are rare. An item can only sink as far as the levels below it. Half the items are leaves and sink zero levels. A quarter are one level up and sink at most one. An eighth sink at most two. Adding it up for n items, the total is at most n/4·1 + n/8·2 + n/16·3 + …, and that sum is less than n.

Pushing one at a time is the opposite: the many items near the bottom are exactly the ones that might climb a long way. Count the comparisons in heapq itself:

import heapq, random

class Counted:
    """A number that counts how often it is compared."""
    comparisons = 0
    __slots__ = ("v",)
    def __init__(self, v): self.v = v
    def __lt__(self, other):
        Counted.comparisons += 1
        return self.v < other.v

def push_all(values):
    Counted.comparisons = 0
    heap = []
    for v in values:
        heapq.heappush(heap, v)
    return Counted.comparisons

def heapify_all(values):
    Counted.comparisons = 0
    heapq.heapify(list(values))
    return Counted.comparisons

n = 1_000_000
shuffled = [Counted(random.random()) for _ in range(n)]
descending = sorted(shuffled, key=lambda c: -c.v)     # every new item is the new minimum

for name, values in (("random order", shuffled), ("descending", descending)):
    print(f"{name:>12}:  n pushes {push_all(values):>10,}   heapify {heapify_all(values):>9,}")
# random order:  n pushes  2,283,479   heapify 1,649,092
#   descending:  n pushes 17,951,445   heapify 1,499,986

Two things worth seeing. heapify costs about 1.6 comparisons per item whatever the input order. And pushes are cheap on average with random input, about 2.3 per item, because a random newcomer usually belongs near the bottom. The O(n log n) is the worst case, and descending input hits it: every newcomer is the new minimum and climbs to the root, nearly 18 comparisons each. When you have all the items up front, call heapify.

Python’s heapq in practice

The module’s own docstring states the property it maintains: “Heaps are arrays for which a[k] <= a[2*k+1] and a[k] <= a[2*k+2] for all k, counting elements from 0.” Everything else follows from treating your list that way. Three things to know.

It’s a min-heap. For numbers, negate on the way in and out. For things you can’t negate, such as strings, wrap them in a small class whose __lt__ is reversed. Python 3.14 added public max-heap functions (heapq.heappush_max, heappop_max, heapify_max and friends); if you’re on 3.14 or later, use those instead of negating.

Ties fall through to the payload. heapq compares whole items with <. A tuple compares its first elements, and only if they’re equal moves on to the second. So when two jobs have the same priority, Python compares the jobs themselves. If they’re dicts, that’s an error; if they’re strings, it’s alphabetical order, which is quietly wrong. The fix is a counter that’s never equal:

import heapq, itertools

jobs = []
heapq.heappush(jobs, (2, {"task": "send invoice"}))
try:
    heapq.heappush(jobs, (2, {"task": "resize image"}))   # same priority: Python compares the dicts
except TypeError as e:
    print("TypeError:", e)

jobs = []
order = itertools.count()                 # 0, 1, 2, ...: never equal, so payloads are never compared
for priority, task in [(2, "send invoice"), (1, "reset password"), (2, "resize image"), (1, "refund")]:
    heapq.heappush(jobs, (priority, next(order), {"task": task}))
print([heapq.heappop(jobs)[2]["task"] for _ in range(len(jobs))])

# heapq is a min-heap. For "largest first", push the negated priority.
scores = [40, 95, 12, 77]
biggest_first = [-s for s in scores]
heapq.heapify(biggest_first)
print([-heapq.heappop(biggest_first) for _ in range(len(scores))])
# TypeError: '<' not supported between instances of 'dict' and 'dict'
# ['reset password', 'refund', 'send invoice', 'resize image']
# [95, 77, 40, 12]

The counter does two jobs at once: it stops the payload comparison, and it makes equal priorities come out first-in, first-out. A heap on its own is not stable: equal items can come out in any order. The counter is how you buy stability.

Combined operations save a sift. heapq.heapreplace(heap, x) pops the root and pushes x in one sift-down. heapq.heappushpop(heap, x) pushes then pops, and returns x straight back if it’s smaller than the root, without touching the heap at all. Both appear in the patterns below.

nlargest, nsmallest, and when to just sort

“The ten slowest requests in this log” is a top-K question, and there are three ways to answer it. Sort everything and take the first ten: O(n log n). Or keep a heap of only the best K seen so far: O(n log K), and only K items of memory. Or, for K = 1, just call max. heapq.nlargest and heapq.nsmallest implement the heap version. Here they are against sorting, on a million random floats:

import heapq, random, timeit

rng = random.Random(42)
data = [rng.random() for _ in range(1_000_000)]

for k in (10, 1_000, 100_000):
    heap_s = min(timeit.repeat(lambda: heapq.nlargest(k, data), number=1, repeat=3))
    sort_s = min(timeit.repeat(lambda: sorted(data, reverse=True)[:k], number=1, repeat=3))
    assert heapq.nlargest(k, data) == sorted(data, reverse=True)[:k]
    print(f"k={k:>7,}   nlargest {heap_s * 1000:6.1f} ms   sorted()[:k] {sort_s * 1000:6.1f} ms")
print(f"max() for k=1: {min(timeit.repeat(lambda: max(data), number=1, repeat=3)) * 1000:.1f} ms")
# k=     10   nlargest   17.7 ms   sorted()[:k]  155.8 ms
# k=  1,000   nlargest   20.4 ms   sorted()[:k]  153.3 ms
# k=100,000   nlargest  301.6 ms   sorted()[:k]  153.3 ms
# max() for k=1: 7.4 ms

For small K the heap is nine times faster, because nearly every item is rejected with one comparison against the root. At K = 100,000 sorting is twice as fast, because the heap is now big and sorted is very well-optimized C. The standard library knows this: nlargest calls max when n is 1 and sorted when K is at least the input’s size. My rule: max for one, nlargest when K is a small fraction of n, sorted otherwise. And if the input fits in a few hundred items, use whichever reads best, because you’re arguing about microseconds.

The heap version has a second advantage that matters more than speed: it works on a stream. A size-K heap never holds more than K items, so it can watch a feed of a billion values with constant memory. The demo below runs that version on thirty response times:

Watch the root. Once the heap is full, the root is the smallest of the K kept, the bar every newcomer must clear. Most arrivals don’t clear it and cost one comparison. Switch K and see the trade: a larger K keeps more and rejects less.

The classic problems

Most heap problems are one of five moves.

Top-K with a size-K heap. To keep the K largest, use a min-heap of size K. That’s backwards-sounding, and it’s the point: you want quick access to the weakest member of the current top K, the one a newcomer has to beat.

import heapq

def top_k_stream(stream, k):
    """Keep the k largest values seen so far. Memory: k items, whatever the stream's length."""
    heap = []                               # a min-heap of the k largest: the root is the weakest of them
    for x in stream:
        if len(heap) < k:
            heapq.heappush(heap, x)
        elif x > heap[0]:                   # beats the weakest of the current top k
            heapq.heapreplace(heap, x)      # pop the root and push x in one sift
    return sorted(heap, reverse=True)

response_ms = [120, 85, 940, 60, 310, 2200, 75, 450, 95, 1800, 130]
print(top_k_stream(response_ms, 3))

from collections import Counter
errors = ["timeout", "429", "timeout", "500", "429", "timeout", "bad json", "500", "timeout"]
print(Counter(errors).most_common(2))    # uses heapq.nlargest under the hood
# [2200, 1800, 940]
# [('timeout', 4), ('429', 2)]

Counter.most_common(n) is exactly this pattern: CPython implements it as heapq.nlargest(n, self.items(), key=_itemgetter(1)). Counting with a hash map, then a heap for the top K, is the standard answer to “most frequent K” questions.

Merge K sorted lists. Keep one candidate from each list in a heap. Pop the smallest, output it, push the next item from the same list. The heap never holds more than K items, so merging N items in total costs O(N log K).

def merge_sorted(*lists):
    """Merge k sorted lists: the heap holds one candidate per list."""
    heap = [(lst[0], i, 0) for i, lst in enumerate(lists) if lst]
    heapq.heapify(heap)
    out = []
    while heap:
        value, which, pos = heap[0]
        out.append(value)
        if pos + 1 < len(lists[which]):
            heapq.heapreplace(heap, (lists[which][pos + 1], which, pos + 1))   # next from the same list
        else:
            heapq.heappop(heap)                                                # that list is used up
    return out

print(merge_sorted([1, 4, 9], [2, 3, 10, 11], [], [5]))
print(list(heapq.merge([1, 4, 9], [2, 3, 10, 11], [], [5])))   # the standard library version, lazy
# [1, 2, 3, 4, 5, 9, 10, 11]
# [1, 2, 3, 4, 5, 9, 10, 11]

heapq.merge is the same algorithm as a lazy generator, which makes it the tool for merging files too big for memory: each input is read one line at a time. It’s also how external sorting finishes: sort chunks that fit in memory, write them out, then merge the sorted chunks with a heap.

The running median. Keep the smaller half of the values in a max-heap and the larger half in a min-heap, balanced so the smaller half has the same number or one more. The median is then the top of the smaller half, or the average of both tops.

import heapq

class RunningMedian:
    """Two heaps: `low` is a max-heap (negated) of the smaller half, `high` a min-heap of the larger half."""

    def __init__(self):
        self.low, self.high = [], []

    def add(self, x):
        if self.low and x > -self.low[0]:
            heapq.heappush(self.high, x)
        else:
            heapq.heappush(self.low, -x)
        # rebalance so len(low) == len(high) or len(high) + 1
        if len(self.low) > len(self.high) + 1:
            heapq.heappush(self.high, -heapq.heappop(self.low))
        elif len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))

    def median(self):
        if len(self.low) > len(self.high):
            return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2

rm = RunningMedian()
out = []
for latency in [120, 85, 940, 60, 310, 2200, 75]:
    rm.add(latency)
    out.append(rm.median())
print(out)

import random, statistics
rm, seen = RunningMedian(), []
for _ in range(5_000):
    x = random.randint(-50, 50)
    rm.add(x); seen.append(x)
    assert rm.median() == statistics.median(seen)
print("matches statistics.median on 5,000 random values")
# [120, 102.5, 120, 102.5, 120, 215.0, 120]
# matches statistics.median on 5,000 random values

Each add is O(log n) and each median is O(1). Note how little the 2,200 ms outlier moved it: that’s why latency dashboards report medians and percentiles rather than means.

Scheduling with a heap of finish times. “How many meeting rooms?”, “how many servers?”, “when is the next free worker?” all reduce to a min-heap of the times things end. It’s one of the exercises below.

import heapq

class KthLargest:
    def __init__(self, k, nums):
        self.k = k
        self.heap = nums[:]
        heapq.heapify(self.heap)               # O(n)
        while len(self.heap) > k:
            heapq.heappop(self.heap)           # keep only the k largest

    def add(self, x):
        if len(self.heap) < self.k:
            heapq.heappush(self.heap, x)
        elif x > self.heap[0]:
            heapq.heapreplace(self.heap, x)
        return self.heap[0]                    # the weakest of the top k is the k-th largest

scores = KthLargest(3, [4, 5, 8, 2])
print([scores.add(x) for x in [3, 5, 10, 9, 4]])
# [4, 5, 5, 8, 8]

import heapq

def min_rooms(meetings):
    """Fewest rooms so no two overlapping meetings share one. meetings: (start, end), end exclusive."""
    ends = []                                  # min-heap of end times of meetings in progress
    for start, end in sorted(meetings):
        if ends and ends[0] <= start:          # the room that frees up first is free by now
            heapq.heapreplace(ends, end)       # reuse it
        else:
            heapq.heappush(ends, end)          # every room is busy: open a new one
    return len(ends)

print(min_rooms([(9, 10), (9, 12), (10, 11), (11, 13), (12, 13)]))
print(min_rooms([(0, 30), (5, 10), (15, 20)]))
print(min_rooms([(7, 10), (2, 4)]))
# 2
# 2
# 1

import heapq
from collections import Counter

def top_k_words(words, k):
    counts = Counter(words)
    # most frequent first; equal counts in alphabetical order
    return [w for w, _ in heapq.nsmallest(k, counts.items(), key=lambda wc: (-wc[1], wc[0]))]

print(top_k_words(["refund", "hours", "refund", "price", "hours", "address", "price"], 2))
print(top_k_words(["b", "a", "c", "a", "b", "c", "d"], 3))
# ['hours', 'price']
# ['a', 'b', 'c']

Advanced: changing priorities, heapsort, d-ary heaps and Dijkstra

Decrease-key, and why heapq doesn’t have it

Real priority queues need two operations that heapq doesn’t offer: change an item’s priority (usually making it more urgent, called decrease-key in a min-heap) and remove an arbitrary item. A support ticket gets escalated. A timer gets cancelled. A shorter route to a city turns up.

The sift itself is easy: if an item’s priority drops, sift it up; if it rises, sift it down. The hard part is finding the item. Its index changes on every swap, and the heap property says nothing about where a non-root item is, so finding it means scanning: O(n). The textbook fix is an indexed heap that keeps a map from each item to its current index and updates it on every swap:

class IndexedMinHeap:
    """A binary heap that knows where every key is, so decrease-key is a real sift."""

    def __init__(self):
        self.a = []                            # [(priority, key)]
        self.pos = {}                          # key -> index in self.a, updated on every swap

    def _swap(self, i, j):
        a = self.a
        a[i], a[j] = a[j], a[i]
        self.pos[a[i][1]] = i
        self.pos[a[j][1]] = j

    def _up(self, i):
        while i > 0 and self.a[i] < self.a[(i - 1) // 2]:
            self._swap(i, (i - 1) // 2)
            i = (i - 1) // 2

    def _down(self, i):
        n = len(self.a)
        while True:
            small = i
            for c in (2 * i + 1, 2 * i + 2):
                if c < n and self.a[c] < self.a[small]:
                    small = c
            if small == i:
                return
            self._swap(i, small)
            i = small

    def push(self, key, priority):
        self.a.append((priority, key))
        self.pos[key] = len(self.a) - 1
        self._up(len(self.a) - 1)

    def decrease_key(self, key, priority):
        i = self.pos[key]                      # O(1) thanks to the position map
        assert priority <= self.a[i][0]
        self.a[i] = (priority, key)
        self._up(i)                            # a smaller priority can only move up: O(log n)

    def pop(self):
        self._swap(0, len(self.a) - 1)
        priority, key = self.a.pop()
        del self.pos[key]
        if self.a:
            self._down(0)
        return key, priority

h = IndexedMinHeap()
for key, p in [("a", 5), ("b", 3), ("c", 8), ("d", 6)]:
    h.push(key, p)
h.decrease_key("c", 1)
print(h.a, h.pos)
print([h.pop() for _ in range(4)])
# [(1, 'c'), (5, 'a'), (3, 'b'), (6, 'd')] {'a': 1, 'b': 2, 'c': 0, 'd': 3}
# [('c', 1), ('b', 3), ('a', 5), ('d', 6)]

It works, and every swap now does two extra dictionary writes. heapq doesn’t do this because it isn’t a class: it’s functions over a list you own, so it has nowhere to keep a position map, and every user would pay for the bookkeeping whether they needed it or not.

Lazy deletion: the trick heapq recommends

The heapq documentation suggests something simpler, and in Python it’s what I’d use. Don’t find the old entry. Mark it dead where it lies, push a fresh entry with the new priority, and when a pop surfaces a dead entry, throw it away and pop again. A dictionary from each task to its live entry is what makes “mark it dead” O(1):

import heapq, itertools

class LazyPQ:
    """A priority queue with update and remove, built on heapq plus a dict."""

    def __init__(self):
        self.heap = []                         # [priority, seq, task, alive] lists
        self.entry = {}                        # task -> its live entry in the heap
        self.seq = itertools.count()

    def push(self, task, priority):
        if task in self.entry:
            self.remove(task)                  # an update is "kill the old entry, add a new one"
        e = [priority, next(self.seq), task, True]
        self.entry[task] = e
        heapq.heappush(self.heap, e)

    def remove(self, task):
        e = self.entry.pop(task)               # O(1): find it through the dict...
        e[3] = False                           # ...and mark it dead where it lies

    def pop(self):
        while self.heap:
            priority, _, task, alive = heapq.heappop(self.heap)
            if alive:                          # dead entries are skipped when they reach the top
                del self.entry[task]
                return task, priority
        raise KeyError("pop from an empty priority queue")

    def __len__(self):
        return len(self.entry)                 # live tasks only, not len(self.heap)

pq = LazyPQ()
for task, p in [("reply to lead", 3), ("approve refund", 2), ("call back", 3), ("send quote", 4)]:
    pq.push(task, p)
pq.push("send quote", 1)                       # the customer asked again: more urgent now
pq.remove("call back")                         # resolved elsewhere
print(len(pq), "live,", len(pq.heap), "entries in the heap")
print([pq.pop() for _ in range(len(pq))])
# 3 live, 5 entries in the heap
# [('send quote', 1), ('approve refund', 2), ('reply to lead', 3)]

The entries are lists, not tuples, so a dead flag can be flipped in place. The sequence number sits before the task, so a dead entry and a live one with the same priority never compare tasks. Every operation stays O(log n), with one cost: dead entries take memory and make the heap deeper until pops clear them. If cancellations are common, compact now and then, rebuilding the list from live entries and calling heapify, which is O(n). You’ll see CPython’s event loop do exactly that in Expert.

Heapsort: guaranteed O(n log n), and rarely the sort you use

Heapify a list, then pop everything: the items come out sorted. That’s heapsort, and on paper it’s excellent. It’s O(n log n) in the worst case, unlike quicksort’s O(n²) worst case, and the classic version needs no extra memory: it builds a max-heap in place and repeatedly swaps the root to the end of the shrinking heap.

In practice almost nothing uses it as its main sort, for three reasons:

import heapq, random, timeit

def heapsort(items):
    heap = list(items)
    heapq.heapify(heap)                                  # O(n)
    return [heapq.heappop(heap) for _ in range(len(heap))]   # n pops, O(log n) each

rng = random.Random(7)
shuffled = [rng.random() for _ in range(1_000_000)]
already_sorted = sorted(shuffled)

for name, data in (("random", shuffled), ("already sorted", already_sorted)):
    assert heapsort(data) == sorted(data)
    h = min(timeit.repeat(lambda: heapsort(data), number=1, repeat=3))
    s = min(timeit.repeat(lambda: sorted(data), number=1, repeat=3))
    print(f"{name:>14}: heapsort {h * 1000:6.0f} ms   sorted() {s * 1000:5.0f} ms")

# Stability: orders with equal amounts should stay in arrival order.
orders = [(30, "first"), (10, "second"), (30, "third"), (10, "fourth")]
print(sorted(orders, key=lambda o: o[0]))

class ByAmount:
    def __init__(self, order): self.order = order
    def __lt__(self, other): return self.order[0] < other.order[0]   # compares the amount only

print([b.order for b in heapsort(ByAmount(o) for o in orders)])
#         random: heapsort    699 ms   sorted()   149 ms
# already sorted: heapsort    418 ms   sorted()    18 ms
# [(10, 'second'), (10, 'fourth'), (30, 'first'), (30, 'third')]
# [(10, 'fourth'), (10, 'second'), (30, 'first'), (30, 'third')]

1. It does more comparisons. CPython’s heapq.py has a comment measuring this on 1,000 random items: heapify plus 1,000 pops took about 1,660 + 8,680 comparisons, while list.sort() on the same lists took about 8,630. Its verdict: “list.sort() is (unsurprisingly!) more efficient for sorting.”
2. It’s unfriendly to memory. Sift-down jumps from index i to 2i+1. Near the top of a big heap those jumps are short, but lower down they leap across memory, and each leap can miss the CPU cache. Merge sort and Timsort read memory in straight runs, which caches love. That’s a constant factor Big-O doesn’t show, and it’s large.
3. It isn’t stable, and it ignores existing order. Equal items can swap places: above, “fourth” overtook “second”. And on already-sorted input, sorted() (Timsort) notices the run and finishes in 18 ms, 23 times faster than heapsort, which does almost the same work whatever the input.

Where heapsort earns its place is as a safety net. C++’s std::sort is usually an introsort: quicksort that switches to heapsort if the recursion gets too deep, which caps the worst case at O(n log n). Go’s sort package does the same with pattern-defeating quicksort. Heapsort is the guarantee, not the workhorse.

d-ary heaps: shorter trees, wider nodes

Nothing forces two children per node. A d-ary heap gives each node d children, at indices d·i + 1 through d·i + d, with the parent at (i − 1) // d. The tree gets shorter: log_d n levels instead of log₂ n. Push gets cheaper, since sift-up does one comparison per level and there are fewer levels. Pop gets more expensive per level, since sift-down has to find the smallest of d children (d − 1 comparisons) before comparing it with the item.

import math, random

class DaryHeap:
    def __init__(self, d):
        self.d, self.a, self.comparisons = d, [], 0

    def _less(self, x, y):
        self.comparisons += 1
        return x < y

    def push(self, x):
        a, d = self.a, self.d
        a.append(x)
        i = len(a) - 1
        while i > 0 and self._less(a[i], a[(i - 1) // d]):
            p = (i - 1) // d
            a[i], a[p] = a[p], a[i]
            i = p

    def pop(self):
        a, d = self.a, self.d
        top, last = a[0], a.pop()
        if a:
            a[0] = last
            i = 0
            while True:
                first = d * i + 1                     # children are d*i+1 ... d*i+d
                if first >= len(a):
                    break
                best = first
                for c in range(first + 1, min(first + d, len(a))):
                    if self._less(a[c], a[best]):
                        best = c
                if not self._less(a[best], a[i]):
                    break
                a[i], a[best] = a[best], a[i]
                i = best
        return top

n = 100_000
rng = random.Random(3)
values = [rng.random() for _ in range(n)]
for d in (2, 4, 8):
    h = DaryHeap(d)
    for v in values:
        h.push(v)
    pushes = h.comparisons
    h.comparisons = 0
    out = [h.pop() for _ in range(n)]
    assert out == sorted(values)
    print(f"d={d}: height {math.ceil(math.log(n * (d - 1) + 1, d)) - 1:>2}   "
          f"comparisons per push {pushes / n:4.2f}   per pop {h.comparisons / n:5.2f}")
# d=2: height 16   comparisons per push 2.29   per pop 28.32
# d=4: height  9   comparisons per push 1.55   per pop 29.45
# d=8: height  6   comparisons per push 1.30   per pop 40.85

The 4-ary heap is the sweet spot here: about the same comparisons per pop as the binary heap, cheaper pushes, and half the levels. Half the levels means half the jumps through memory, and the four children of a node sit next to each other, often in one cache line. That’s why real systems pick 4. Go’s runtime keeps its timers in a 4-ary heap (const timerHeapN = 4 in runtime/time.go), and .NET’s PriorityQueue<TElement, TPriority> is documented as “an array-backed, quaternary min-heap”. Java’s PriorityQueue is a plain binary heap. When decrease-key dominates, as it does in Dijkstra on dense graphs, a wider heap helps more, because decrease-key is a sift-up.

Dijkstra’s shortest paths with a heap

Dijkstra’s algorithm finds the shortest route from one place to every other place in a network whose edges have non-negative lengths. It’s a priority queue problem through and through: always expand the closest place you haven’t finished yet. With a heap, and lazy deletion instead of decrease-key:

import heapq

def dijkstra(graph, source):
    """graph: {node: [(neighbour, weight), ...]} with weights >= 0."""
    dist = {source: 0}
    heap = [(0, source)]
    pops = stale = 0
    while heap:
        d, node = heapq.heappop(heap)
        pops += 1
        if d > dist[node]:
            stale += 1                         # an old, longer entry for a node we already settled
            continue
        for nxt, w in graph[node]:
            nd = d + w
            if nd < dist.get(nxt, float("inf")):
                dist[nxt] = nd
                heapq.heappush(heap, (nd, nxt))   # no decrease-key: push a duplicate instead
    return dist, pops, stale

# travel minutes between delivery stops
roads = {
    "depot": [("A", 7), ("B", 2)],
    "B": [("A", 3), ("C", 8)],
    "A": [("C", 1), ("D", 6)],
    "C": [("D", 2)],
    "D": [],
}
dist, pops, stale = dijkstra(roads, "depot")
print(dist)
print(f"{pops} pops, {stale} of them stale")
# {'depot': 0, 'A': 5, 'B': 2, 'C': 6, 'D': 8}
# 8 pops, 3 of them stale

The depot reaches A directly in 7 minutes, but through B in 5, so A gets pushed twice; the 7 is stale by the time it surfaces and is skipped by the d > dist[node] check. That check is lazy deletion with no dictionary needed, because dist already records the best known distance. The cost is O((V + E) log V) for V places and E roads: every road can cause one push, and each push or pop is O(log V).

Event simulation

A discrete-event simulation models a system as a list of things that will happen at future times: a customer arrives, a chat ends, a server finishes a job. The engine is a min-heap ordered by time: pop the earliest event, update the state, push whatever events it causes. Nothing happens between events, so the clock jumps straight from one to the next. Here’s a support desk with two or three agents:

import heapq, itertools, random

def simulate(agents, customers, seed=11):
    """A support desk: customers arrive at random, each chat takes a random time."""
    rng = random.Random(seed)
    seq = itertools.count()                    # tie-breaker for events at the same instant
    events = []                                # (time, seq, kind, customer)
    t = 0.0
    for c in range(customers):
        t += rng.expovariate(1 / 2.0)          # a new customer every 2 minutes on average
        heapq.heappush(events, (t, next(seq), "arrive", c))

    free, line, waits = agents, [], []
    while events:
        now, _, kind, c = heapq.heappop(events)        # always the earliest thing that happens next
        if kind == "arrive":
            line.append((now, c))
        else:
            free += 1                                   # a chat ended
        while free and line:
            arrived, nxt = line.pop(0)
            waits.append(now - arrived)
            free -= 1
            heapq.heappush(events, (now + rng.expovariate(1 / 3.5), next(seq), "done", nxt))
    waits.sort()
    return sum(waits) / len(waits), waits[int(len(waits) * 0.95)]

for agents in (2, 3):
    mean, p95 = simulate(agents, 10_000)
    print(f"{agents} agents: mean wait {mean:5.1f} min, 95th percentile {p95:5.1f} min")
# 2 agents: mean wait  10.9 min, 95th percentile  37.5 min
# 3 agents: mean wait   0.9 min, 95th percentile   5.4 min

Chats average 3.5 minutes and customers arrive every 2, so two agents are busy 87.5% of the time, and three are busy 58%. Adding one agent cuts the 95th-percentile wait from 37.5 minutes to 5.4. Queues near full capacity are brutal, and a twenty-line heap simulation shows it faster than any formula. Note the seq again: two events at the exact same time must not compare their kind strings, or “arrive” would always beat “done” alphabetically, which is a modelling decision you’d want to make on purpose.

import heapq
from collections import defaultdict

def network_delay(times, n, source):
    """times: (u, v, ms) edges. How long until every node 1..n has the signal? -1 if some never do."""
    graph = defaultdict(list)
    for u, v, w in times:
        graph[u].append((v, w))
    dist = {}
    heap = [(0, source)]
    while heap:
        d, node = heapq.heappop(heap)
        if node in dist:
            continue                           # already settled with a shorter time
        dist[node] = d
        for nxt, w in graph[node]:
            if nxt not in dist:
                heapq.heappush(heap, (d + w, nxt))
    return max(dist.values()) if len(dist) == n else -1

print(network_delay([(2, 1, 1), (2, 3, 1), (3, 4, 1)], 4, 2))
print(network_delay([(1, 2, 1)], 2, 2))
print(network_delay([(1, 2, 4), (1, 3, 1), (3, 2, 1), (2, 4, 1)], 4, 1))
# 2
# -1
# 3

import heapq, itertools

class CompactingPQ:
    """Lazy deletion that cleans up after itself once dead entries are the majority."""

    def __init__(self):
        self.heap, self.entry, self.dead = [], {}, 0
        self.seq = itertools.count()

    def push(self, task, priority):
        if task in self.entry:
            self.remove(task)
        e = [priority, next(self.seq), task, True]
        self.entry[task] = e
        heapq.heappush(self.heap, e)

    def remove(self, task):
        self.entry.pop(task)[3] = False
        self.dead += 1
        if len(self.heap) > 100 and self.dead > len(self.heap) // 2:
            self.heap = [e for e in self.heap if e[3]]   # O(n) rebuild...
            heapq.heapify(self.heap)                      # ...paid for by the removals before it
            self.dead = 0

    def pop(self):
        while self.heap:
            e = heapq.heappop(self.heap)
            if e[3]:
                del self.entry[e[2]]
                return e[2], e[0]
            self.dead -= 1
        raise KeyError("empty")

pq = CompactingPQ()
for i in range(1_000):
    pq.push(f"t{i}", i)
for i in range(900):
    pq.remove(f"t{i}")
print(len(pq.entry), "live,", len(pq.heap), "entries in the heap,", pq.dead, "dead")
print(pq.pop())
# 100 live, 124 entries in the heap, 24 dead
# ('t900', 900)

Expert: timers, schedulers and the heaps you don’t write

Millions of timers: heaps against timing wheels

A server holding 100,000 connections has at least 100,000 timers: idle timeouts, read deadlines, keep-alive pings. Two facts about those timers change the design. Most of them never fire, because the client sends data in time and the timer is cancelled or reset. And most of them don’t need millisecond precision: an idle timeout that fires at 30.1 seconds instead of 30.0 hurts nobody.

A heap handles timers well, but every insert is O(log n), and cancellation is either an O(log n) indexed removal or a lazy dead entry. When timers are created and cancelled on every request, that’s a lot of sifting for work that’s mostly thrown away.

The alternative is a timing wheel, described by George Varghese and Tony Lauck in their 1987 paper “Hashed and Hierarchical Timing Wheels”. Picture a clock face with a fixed number of slots, each one tick wide, say 100 ms, and a hand that advances one slot per tick. To schedule a timer, compute which slot its deadline falls in and add it to that slot’s list: O(1). To cancel, unlink it from that list: O(1). On each tick, fire everything in the slot under the hand. A deadline more than one revolution away goes into the slot it hashes to with a lap count, and the hand decrements the count each time it passes. That’s the hashed wheel. A hierarchical wheel instead stacks wheels like a clock’s second, minute and hour hands, and moves timers down to finer wheels as their time approaches.

Here it is in Python, with an O(1) cancel through a dict from timer to slot:

class HashedWheel:
    """A hashed timing wheel: `size` buckets, each one tick wide. Insert and cancel are O(1)."""

    def __init__(self, size=8, tick_ms=100):
        self.size, self.tick_ms = size, tick_ms
        self.buckets = [dict() for _ in range(size)]   # timer id -> [rounds left, callback]
        self.where = {}                                # timer id -> bucket index, for O(1) cancel
        self.now_tick = 0
        self.next_id = 0

    def schedule(self, delay_ms, callback):
        ticks = max(1, -(-delay_ms // self.tick_ms))   # round up: never fire early
        slot = (self.now_tick + ticks) % self.size
        rounds = (ticks - 1) // self.size              # full laps of the wheel to wait first
        tid, self.next_id = self.next_id, self.next_id + 1
        self.buckets[slot][tid] = [rounds, callback]
        self.where[tid] = slot
        return tid

    def cancel(self, tid):
        slot = self.where.pop(tid, None)
        if slot is not None:
            del self.buckets[slot][tid]

    def tick(self):
        self.now_tick += 1
        bucket = self.buckets[self.now_tick % self.size]
        for tid, timer in list(bucket.items()):
            if timer[0] == 0:
                del bucket[tid]; del self.where[tid]
                timer[1](self.now_tick * self.tick_ms)
            else:
                timer[0] -= 1                          # not this lap: wait for the next one

fired = []
wheel = HashedWheel(size=8, tick_ms=100)
wheel.schedule(250, lambda t: fired.append(("typing indicator off", t)))
idle = wheel.schedule(300, lambda t: fired.append(("idle timeout", t)))
wheel.schedule(1_000, lambda t: fired.append(("retry webhook", t)))   # more than one lap away
wheel.cancel(idle)                                     # the client sent a message in time
for _ in range(12):
    wheel.tick()
print(fired)
# [('typing indicator off', 300), ('retry webhook', 1000)]

The 250 ms timer fires at 300 ms: it rounds up to the next tick, because firing a timeout early is a bug and firing it a little late usually isn’t. That rounding is the price of the wheel. You trade exact deadlines for O(1) everything.

Two well-documented systems made exactly that trade:

• Netty, the Java networking library, ships HashedWheelTimer, whose documentation describes it as “optimized for approximated I/O timeout scheduling”. Its defaults are a 100 ms tick and 512 ticks per wheel, and it’s explicitly based on the Varghese and Lauck paper.
• Apache Kafka rebuilt its “purgatory”, where requests wait for a condition or a timeout, around hierarchical timing wheels. Yasuhiro Matsuda’s 2015 write-up on the Confluent blog explains why: the old version kept requests in Java’s DelayQueue, a heap with O(log n) insert and delete, and completed requests weren’t removed promptly, so a reaper thread had to scan for them. The new design keeps timers in wheel buckets with O(1) insert and delete, and puts only the buckets into a DelayQueue to wake the thread. In the post’s benchmark the old implementation saturated at around 25,000 requests per second in the high-timeout scenario, and the new one at around 105,000.

The Kafka design is my favourite detail here: it doesn’t throw the heap away. It uses the heap for the few things that need exact ordering (when should the thread next wake up?) and the wheel for the many things that don’t. In Python application code you’ll almost never build either, because the event loop owns your timers. Which brings us to the heap that’s running under your code right now.

Inside asyncio: a heap under every await

CPython’s asyncio event loop keeps every scheduled callback in a list called _scheduled, maintained with heapq. call_later(delay, cb) computes an absolute time and calls call_at, which ends like this (asyncio/base_events.py, Python 3.13):

        timer = events.TimerHandle(when, callback, args, self, context)
        if timer._source_traceback:
            del timer._source_traceback[-1]
        heapq.heappush(self._scheduled, timer)
        timer._scheduled = True
        return timer

asyncio.sleep, asyncio.wait_for and asyncio.timeout all end up here. Each turn of the loop, _run_once uses the heap three ways (trimmed):

        sched_count = len(self._scheduled)
        if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
            self._timer_cancelled_count / sched_count >
                _MIN_CANCELLED_TIMER_HANDLES_FRACTION):
            # Remove delayed calls that were cancelled if their number
            # is too high
            new_scheduled = []
            for handle in self._scheduled:
                if handle._cancelled:
                    handle._scheduled = False
                else:
                    new_scheduled.append(handle)

            heapq.heapify(new_scheduled)
            self._scheduled = new_scheduled
            self._timer_cancelled_count = 0
        else:
            # Remove delayed calls that were cancelled from head of queue.
            while self._scheduled and self._scheduled[0]._cancelled:
                self._timer_cancelled_count -= 1
                handle = heapq.heappop(self._scheduled)
                handle._scheduled = False

        timeout = None
        if self._ready or self._stopping:
            timeout = 0
        elif self._scheduled:
            # Compute the desired timeout.
            timeout = self._scheduled[0]._when - self.time()
            ...
        event_list = self._selector.select(timeout)
        ...
        # Handle 'later' callbacks that are ready.
        end_time = self.time() + self._clock_resolution
        while self._scheduled:
            handle = self._scheduled[0]
            if handle._when >= end_time:
                break
            handle = heapq.heappop(self._scheduled)
            handle._scheduled = False
            self._ready.append(handle)

Read it as a list of everything in this article:

1. Lazy deletion. TimerHandle.cancel() doesn’t touch the heap. It sets a flag and bumps a counter. Cancelled handles at the top are popped and dropped.
2. Compaction. When there are more than _MIN_SCHEDULED_TIMER_HANDLES (100) timers and more than _MIN_CANCELLED_TIMER_HANDLES_FRACTION (0.5) of them are cancelled, the loop rebuilds the list from live handles and calls heapify. That’s the exercise above, line for line.
3. Peek decides the sleep. self._scheduled[0]._when - self.time() is how long the loop may block waiting for I/O. If the heap is empty and nothing is ready, the timeout is None: sleep until a socket wakes you.
4. Pop everything due. All timers due within one clock-resolution step move to the ready queue.

You can watch the first two happen. _scheduled is private, so never touch it in real code, but it’s fine for a look:

import asyncio

async def main():
    loop = asyncio.get_running_loop()
    handles = [loop.call_later(delay, print, f"after {delay} s") for delay in (30, 10, 20)]
    heap = loop._scheduled                     # private attribute: the loop's timer heap
    print(type(loop).__name__, "holds", len(heap), "timers")
    print("heap[0] fires in", round(heap[0].when() - loop.time()), "s")

    handles[1].cancel()                        # cancel the 10 s timer
    print("after cancel:", len(heap), "in the heap; heap[0] cancelled?", heap[0].cancelled())
    await asyncio.sleep(0)                     # one turn of the loop pops a cancelled top
    print("after one loop turn:", len(loop._scheduled))

    many = [loop.call_later(60 + i, lambda: None) for i in range(1_000)]
    for h in many[:600]:
        h.cancel()                             # 600 of 1,002 are now dead
    print("before compaction:", len(loop._scheduled))
    await asyncio.sleep(0)
    print("after one loop turn:", len(loop._scheduled))
    for h in handles + many:
        h.cancel()

asyncio.run(main())
# _UnixSelectorEventLoop holds 3 timers
# heap[0] fires in 10 s
# after cancel: 3 in the heap; heap[0] cancelled? True
# after one loop turn: 2
# before compaction: 1002
# after one loop turn: 402

The cancelled 10-second timer stayed in the heap until a loop turn found it on top. The 600 cancelled 60-second timers would have sat there for a minute, but crossing the “more than half” line triggered the rebuild on the next turn.

One subtlety. TimerHandle.__lt__ compares only _when, and the call_later docstring says it plainly: “If two callbacks are scheduled for exactly the same time, it is undefined which will be called first.” No counter, no stability. libuv, the C event loop under Node.js and under uvloop, made the other choice. Its timer heap breaks ties with a start counter (src/timer.c):

  if (a->timeout < b->timeout)
    return 1;
  if (b->timeout < a->timeout)
    return 0;
  return a->start_id < b->start_id;

That’s the (priority, next(counter), item) tuple from Intermediate, written in C. Whether your event loop needs it depends on whether any code relies on same-instant timers running in scheduling order. Don’t write code that does.

Starvation, aging and priority inversion

A strict priority queue has a failure mode built in: if urgent work keeps arriving, routine work never runs. That’s starvation. The standard fix is aging: a waiting item’s effective priority improves the longer it waits, so eventually even the lowest-priority item outranks fresh urgent ones.

Here’s one worker that can finish one job a minute, urgent jobs arriving about 0.75 times a minute and routine ones about 0.2, so the worker is busy 95% of the time:

import heapq, itertools, random

def run(aging_per_min, minutes=2000, seed=5):
    """One worker, one job per minute. Urgent jobs (priority 1) arrive at ~0.75 a minute,
    routine ones (priority 3) at 0.2 a minute. Return the worst wait for each kind."""
    rng = random.Random(seed)
    seq = itertools.count()
    waiting = []                               # (arrival minute, seq, priority)
    worst = {1: 0, 3: 0}
    for now in range(minutes):
        if rng.random() < 0.75:
            waiting.append((now, next(seq), 1))
        if rng.random() < 0.2:
            waiting.append((now, next(seq), 3))
        if not waiting:
            continue
        # effective priority improves by `aging_per_min` for every minute spent waiting
        heap = [(p - aging_per_min * (now - t), s, t, p) for t, s, p in waiting]
        heapq.heapify(heap)
        _, s, t, p = heapq.heappop(heap)
        waiting = [(t2, s2, p2) for _, s2, t2, p2 in heap]
        worst[p] = max(worst[p], now - t)
    left = sum(1 for *_, p in waiting if p == 3)
    return worst, left

for aging in (0, 0.1):
    worst, left = run(aging)
    print(f"aging {aging}/min: worst wait urgent {worst[1]:>3} min, routine {worst[3]:>3} min, routine still queued at the end: {left}")
# aging 0/min: worst wait urgent   0 min, routine  54 min, routine still queued at the end: 2
# aging 0.1/min: worst wait urgent   9 min, routine  28 min, routine still queued at the end: 2

(Aging changes priorities over time, so the simulation re-heapifies every minute. A real scheduler would use a formula that doesn’t need that, such as ordering by arrival_time + priority_offset, which is fixed per item.)

Strict priority keeps urgent jobs at zero wait and lets a routine job wait 54 minutes. Aging at 0.1 per minute halves the routine worst case to 28 minutes and costs urgent jobs up to 9. Neither is “right”; it’s a policy decision with numbers attached, and it belongs in the design doc. When I pushed the same simulation past capacity (0.9 urgent plus 0.3 routine a minute, over 600 minutes), strict priority left 132 routine jobs unserved, and aging only spread the pain: 44 unserved, with urgent jobs waiting up to 98 minutes. Aging redistributes waiting; it doesn’t create capacity. If the queue grows without bound, you need more workers or less work.

Priority inversion is starvation’s nastier cousin. A low-priority task holds a lock that a high-priority task needs. Medium-priority tasks, which don’t need the lock, keep preempting the low-priority one, so it never finishes and never releases the lock, and the high-priority task waits behind work less important than itself. The famous case is NASA’s Mars Pathfinder lander in 1997, which kept resetting until engineers enabled priority inheritance on the offending mutex: a task holding a lock temporarily inherits the priority of the most urgent task waiting for it. If you build a scheduler where jobs can wait on each other, priorities have to flow along those waits.

Priority queues in job systems

Message brokers offer priorities, and it’s worth knowing how they behave before you rely on them.

RabbitMQ supports priority queues through the x-max-priority argument. Its documentation recommends few levels (“Using a single digit number of priorities is highly recommended” for classic queues) because each level has memory, disk and CPU cost. It also warns about the trap that catches most people: if consumers are idle and have spare prefetch capacity, messages are handed out the moment they arrive and never wait in the queue, so there’s nothing to reorder.

Celery on Redis doesn’t get real priorities, because Redis lists have no such notion. Celery’s routing guide explains that it emulates them “by creating n lists for each queue”, consolidating its 10 levels into 4 by default, and is frank about the result: “this will never be as good as priorities implemented at the broker server level, and may be approximate at best.” It adds that because of worker prefetching, tasks submitted together “may be out of priority order at first”.

My opinion, and the one the production section backs up: for job systems, prefer separate queues with their own workers over a priority field. A priority only reorders work that’s waiting. A separate queue gives the urgent work its own capacity, so a flood of routine jobs can’t occupy every worker in the first place. Priorities answer “which next?”; separate queues answer “who’s free?”, and the second question is usually the one that hurts.

Pairing heaps and Fibonacci heaps, honestly

Binary heaps can’t meld (merge two heaps) cheaply, and their decrease-key is O(log n). Theory has better answers. The Fibonacci heap (Michael Fredman and Robert Tarjan, 1984) does insert, meld and decrease-key in O(1) amortized time and delete-min in O(log n) amortized. That improves Dijkstra’s bound from O((V + E) log V) to O(E + V log V), a real difference on dense graphs in a textbook.

In practice Fibonacci heaps are rarely the fastest choice. Each node carries a parent, a child, two sibling pointers, a degree and a mark bit, every operation chases pointers through memory, and the constant factors are large. A 2007 University of Texas technical report, “Priority Queues and Dijkstra’s Algorithm” (Mo Chen, Rezaul Alam Chowdhury, Vijaya Ramachandran, David Lan Roche and Lingling Tong), ran Dijkstra both ways and found that a priority queue without decrease-key, inserting duplicates and skipping stale pops as above, was faster in most cases.

The pairing heap (Fredman, Sedgewick, Sleator and Tarjan, 1986) is the practical one of the fancy heaps: a tree where each node has a list of children, meld is “make the larger root a child of the smaller”, and delete-min merges the root’s children in pairs. It’s short to write and fast in practice. Its exact decrease-key cost is subtle enough that Fredman proved in 1999 it can’t be O(1) amortized, unlike the Fibonacci heap.

It also runs in software you use. PostgreSQL ships both a binary heap (src/include/lib/binaryheap.h) and a pairing heap (src/include/lib/pairingheap.h). Its MergeAppend executor node, which merges several already-sorted inputs such as partitions, keeps one slot per input in a binaryheap: the merge-K-sorted-lists exercise, inside a database. And pgvector’s HNSW index search allocates two pairing heaps per layer, one of candidates ordered nearest-first and one of results ordered furthest-first:

pairingheap *C = pairingheap_allocate(CompareNearestCandidates, NULL);
pairingheap *W = pairingheap_allocate(CompareFurthestCandidates, NULL);

That’s the top-K pattern again: W is the size-K “best so far” heap whose worst member is the bar to beat. The lesson I take from all of this: pick the binary or 4-ary heap by default, and treat the others as tools for specific operations (cheap meld, frequent decrease-key) that you’ve measured you need.

Reviewing a priority queue in a design

When a design has “and then we put it in a priority queue”, these are the questions I ask:

• What exactly is the priority, and what breaks a tie? If the answer to the second is “nothing”, equal items come out in an arbitrary order. Usually you want arrival order.
• Can priorities change after insertion? Then you need an indexed heap or lazy deletion, and someone has to own the dead entries.
• What’s the cancellation rate? If most entries are cancelled, as with I/O timeouts, a timing wheel or a compacting lazy heap beats a plain one.
• Is there starvation? Under sustained load, does the lowest priority ever run? If it must, how long may it wait, and is that written down?
• Does it have to survive a restart, or span machines? An in-memory heap is per process. Shared priority queues live in the database (ORDER BY priority, created_at with FOR UPDATE SKIP LOCKED), in a Redis sorted set (ZADD and ZPOPMIN), or in the broker. Each moves the heap’s job somewhere with different failure modes.
• Is it bounded? A priority queue that only grows is a memory leak with excellent ordering.
• Would separate queues be simpler? Two or three fixed levels are often better as two or three queues with their own workers.

import heapq
from collections import deque

class FairScheduler:
    """Each tenant gets its own FIFO line. A heap of (virtual time, tenant) picks the tenant
    that has used the least so far, so one tenant's backlog can't push another's work back."""

    def __init__(self):
        self.lines = {}                        # tenant -> deque of (job, cost)
        self.vtime = {}                        # tenant -> work charged so far
        self.ready = []                        # heap of (vtime, tenant) for tenants with work
        self.clock = 0.0                       # the vtime of the last job started

    def submit(self, tenant, job, cost=1.0):
        line = self.lines.setdefault(tenant, deque())
        if not line:                           # tenant becomes active: no credit for idle time
            self.vtime[tenant] = max(self.vtime.get(tenant, 0.0), self.clock)
            heapq.heappush(self.ready, (self.vtime[tenant], tenant))
        line.append((job, cost))

    def next(self):
        if not self.ready:
            return None
        vt, tenant = heapq.heappop(self.ready)
        job, cost = self.lines[tenant].popleft()
        self.clock = vt
        self.vtime[tenant] = vt + cost
        if self.lines[tenant]:                 # still has work: back in the heap at its new vtime
            heapq.heappush(self.ready, (self.vtime[tenant], tenant))
        return tenant, job

s = FairScheduler()
for i in range(6):
    s.submit("big-campaign", f"comment-{i}")
s.submit("small-shop", "customer message")
s.submit("clinic", "booking request")
print([s.next() for _ in range(5)])
# [('big-campaign', 'comment-0'), ('clinic', 'booking request'), ('small-shop', 'customer message'), ('big-campaign', 'comment-1'), ('big-campaign', 'comment-2')]

In production: priority queues in ezychat.io

ezychat.io is a multi-tenant platform where AI answers customers on WhatsApp, Instagram, Messenger and web chat, with Python (FastAPI and Celery), PostgreSQL and Redis underneath. I’m its primary engineer. Here’s the honest starting point: I searched the backend for heapq, nlargest, nsmallest and PriorityQueue, and there are no matches. Not one heap is written by hand.

Priority queues run all through it anyway. They’re expressed as ORDER BY clauses, run by the database, the event loop and the broker. That’s the normal case in backend work, and it’s why the most useful thing to know about heaps is often where someone else has already put one. These are trimmed excerpts from the codebase.

The work queue is ORDER BY priority, created_at

When the AI (or a workflow, or a person) needs a human to act, whether that’s an escalation, an approval of a drafted reply or a task, it creates a work item. Items have a priority, set by one small function:

def derive_priority(kind: str, category: str | None, context: dict | None) -> int:
    """Seats spec §4: human_requested / negative-sentiment escalations = 1,
    other escalations and approvals = 2, tasks = 3."""
    if kind == "escalation":
        negative = bool(context and context.get("sentiment") == "negative")
        return 1 if category == "human_requested" or negative else 2
    return 2 if kind == "approval" else 3

Three levels, 1 the most urgent, stored as a SmallInteger with the model comment # 1 highest. That’s the triage scale from Beginner, and the same convention as heapq: smaller comes first.

The queue itself is a query. When a team member comes online, the routing engine re-evaluates their teams’ pending items in this order:

                    select(WorkItem)
                    .where(
                        WorkItem.org_id == org_id,
                        WorkItem.status == "pending",
                        WorkItem.assignee_membership_id.is_(None),
                        WorkItem.routed_team_id.in_(team_ids),
                    )
                    .order_by(WorkItem.priority, WorkItem.created_at)

ORDER BY priority, created_at is the (priority, next(counter), item) tuple from Intermediate, written in SQL. created_at is the tie-breaker, and it does both jobs the counter did: equal priorities never fall through to comparing anything else, and they come out first-in, first-out. The same ordering is used when a departing member’s open items are rerouted and when the SLA sweeper picks due items.

One deliberate difference: the list endpoint people browse sorts differently within a band. Its comment reads “Team queues sort by (priority, created_at) — spec §4; the cursor keys on created_at+id which stays stable within a priority band”, and the query is order_by(WorkItem.priority, WorkItem.created_at.desc(), WorkItem.id.desc()). A person scanning the list sees the newest item in each band first; the engine hands out the oldest. Same priority, two tie-break rules, each chosen for its reader.

Decrease-key, as a business rule

What happens when a customer who’s already waiting for a person asks again? Creating a second work item would be a duplicate. The code treats it as decrease-key:

    A duplicate open ask bumps priority instead of duplicating; the bump is
    itself recorded so the queue explains why an item jumped. New items run
    through the routing engine unless the caller opts out (`auto_route`).
    """
    if conversation_id is not None:
        existing = await find_open_duplicate(session, org_id, conversation_id, kind, category)
        if existing is not None:
            if existing.priority > 1:
                existing.priority -= 1
                record_item_event(
                    session,
                    existing,
                    event_type="reprioritized",
                    actor_kind=created_by_kind,
                    actor_id=created_by_id,
                    reason="duplicate open ask bumped priority",
                )
            return existing, False

find_open_duplicate is the lookup that a heap can’t do and a hash map or index can: its docstring calls conversation, kind, category and open status the “Idempotency key”. In a heap, a priority change needs an indexed heap or lazy deletion. In the database it’s an UPDATE, because the order is computed when the queue is read, not maintained as items change. That’s the database’s version of the trade: no sifting on write, a sort on read.

Two details I like. The priority is clamped at 1, so repeated asks can’t push it past the top band. And created_at doesn’t change, so a bumped item jumps to the front of its new band, as the diagram shows: a customer who has waited longest and asked twice really is the most urgent. And the bump writes an event, “so the queue explains why an item jumped”. A priority queue that people work from needs to explain its order, not only keep it.

Where the database does the heap work

The SLA sweeper runs every minute and moves overdue items one hop along an escalation chain. It picks its work like this:

async def _due_items(session: AsyncSession, org_id: uuid.UUID, now: datetime) -> list[WorkItem]:
    """Items whose next hop is due, highest priority first. `skip_locked`
    keeps a second sweeper (or an overlapping run of this one) from acting on
    the same item twice — the loser simply sees fewer rows."""
    result = await session.execute(
        select(WorkItem)
        .where(
            WorkItem.org_id == org_id,
            WorkItem.status.in_(("routed", "pending")),
            WorkItem.response_target_at.isnot(None),
            WorkItem.response_target_at < now,
        )
        .order_by(WorkItem.priority, WorkItem.created_at)
        .limit(MAX_ITEMS_PER_ORG)
        .with_for_update(skip_locked=True)
    )
    return list(result.scalars())

ORDER BY … LIMIT 500 is a top-K query. PostgreSQL has a special plan for it: when the planner knows a sort only needs its first K rows, it runs a top-N heapsort, which is the size-K heap from Intermediate inside the database, and EXPLAIN ANALYZE reports it by name. But this query also says FOR UPDATE SKIP LOCKED, and I wanted to know what that does to the plan. I can’t run experiments on production data, so I built a synthetic work_items table on my laptop (PostgreSQL 14, default work_mem of 4 MB) with the same columns and the (org_id, response_target_at) index, and ran the query with and without the lock:

The reason is in the plan’s shape. Without the lock it’s Limit → Sort, so the sort knows it needs 500 rows and keeps a 500-row heap. With the lock it’s Limit → LockRows → Sort. Locking sits between the limit and the sort. Locking can discard rows (SKIP LOCKED skips rows another transaction holds, and a row that changed meanwhile may no longer match), so the sort can’t be told how many rows will be enough. It sorts everything that matched.

Is that a problem here? No, and I’m not changing anything. The sort runs over the due items of one organization, not over the table, and the 2,000-due rows above are already a large backlog for one team’s escalations: at that size the difference is 0.7 ms. The cap comes with its own comment: “One item is at most one hop per pass; a pathological org cannot monopolise the minute’s budget (its remaining items are simply swept next minute).” That cap is fairness across tenants, the same idea as the Expert exercise. If due backlogs ever reached tens of thousands, the fix would be an index that matches the ORDER BY, so the database reads rows already in order and stops early. On the all-due synthetic table, an (org_id, priority, created_at) index took the locked query to 0.3 ms; with few due rows the same index would make it step over every not-yet-due item, so it’s a trade to measure, not a free win. Knowing where the heap went is what tells you which index to add.

Top-K in retrieval: sorted() is the right call

Hybrid retrieval for AI replies runs a vector search and a full-text search, fuses the two rankings, and keeps the best 12 for the reranker:

VECTOR_TOP_K = 20
FTS_TOP_K = 20
RRF_K = 60
RERANK_POOL = 12
FINAL_TOP_K = 5

def _rrf(*ranked_lists: list[uuid.UUID]) -> list[uuid.UUID]:
    """Reciprocal-rank fusion; deterministic tie-break on the id."""
    scores: dict[uuid.UUID, float] = {}
    for ranked in ranked_lists:
        for rank, chunk_id in enumerate(ranked):
            scores[chunk_id] = scores.get(chunk_id, 0.0) + 1.0 / (RRF_K + rank + 1)
    return sorted(scores, key=lambda cid: (-scores[cid], str(cid)))

The caller does _rrf(vector_ids, fts_ids)[:RERANK_POOL]. That’s “sort everything, take the first K”, the approach Intermediate said a heap beats. Should it be heapq.nsmallest(12, …)? Two searches of up to 20 give at most 40 candidates. I measured both on 30 (half the hits overlapping), then at a hundred and ten thousand times that, using the same key function:

#      30 candidates: sorted()[:12]      17.8 us   nsmallest(12)      20.6 us
#   3,000 candidates: sorted()[:12]    1749.5 us   nsmallest(12)    1676.3 us
# 300,000 candidates: sorted()[:12]  214982.9 us   nsmallest(12)  205465.3 us
# just building the 300,000 sort keys:  205012.0 us

At the real size, sorted is faster, by 3 microseconds, inside a request that also makes network calls to an embedding model and a reranker. Even at 300,000 candidates the heap saves under 5%, and the last line shows why: building the key, str(cid) for every UUID, costs as much as the entire nsmallest call. The selection algorithm isn’t where the time goes. This is the Big-O article’s lesson in miniature: at small n the constant wins, and the expensive part is often not the part with the interesting complexity. I’d leave sorted exactly as it is.

The heaps in this pipeline are elsewhere. Both searches are ORDER BY … LIMIT 20 in PostgreSQL. The vector search runs on a pgvector HNSW index (USING hnsw (embedding vector_cosine_ops) in the migration), and pgvector’s HNSW search is the pair of pairing heaps quoted in Expert. Every retrieval that misses the result cache walks them.

The heap under every await

Which event loop, and so which timer heap, depends on the process. The API containers run uvicorn (the Dockerfile’s CMD ["uvicorn", "ezychat.main:create_app", "--factory", ...]), and the backend depends on uvicorn[standard], which installs uvloop. uvicorn’s default loop setting is auto, and its auto_loop_factory picks uvloop whenever it can be imported. So the API’s timers live in libuv’s heap, the one with the start_id tie-break. Celery task bodies are different: they bridge into async code with asyncio.run on a throwaway loop, so they run CPython’s own loop and its heapq-managed _scheduled list.

Either way, every timeout in the code is a heap entry. The AI reply call is wrapped like this:

        response = await asyncio.wait_for(
            generate_with_failover(
                current, turn.route.primary, turn.route.fallback, clients=turn.clients
            ),
            timeout=settings.ai_reply_timeout_s,
        )

and each tool call the model makes gets asyncio.wait_for(provider.execute(...), timeout=definition.timeout_ms / 1000). Each wait_for pushes a timer; most are cancelled when the call finishes first, which is exactly the “most timers never fire” workload that pushed Netty and Kafka to wheels. For a Python process with a few hundred concurrent requests, a heap is the right structure and I’ve never had a reason to think about it.

The one place ezychat’s code explicitly reasons about timer cost is the admission controller, which queues requests for a limited number of slots when the API is saturated:

            # Bounded queue: a waiter costs a future and a timer on the very
            # loop that is short of time, so unbounded queueing made even the
            # rejections slow (0.5 s timeouts firing seconds late). Refusing
            # outright caps total work at `limit + max_waiters`.
            if len(self._waiters) >= max_waiters:
                return False

“0.5 s timeouts firing seconds late” is the event loop’s heap telling the truth about an overloaded process. A timer is popped only when the loop gets a turn, and a loop that’s drowning in callbacks gets turns late. The heap ordered the deadlines perfectly and the deadlines were still missed, because a priority queue decides which thing runs next, not when the CPU is free to run it. The fix wasn’t a better heap. It was fewer waiters.

Lanes instead of priorities

Celery’s broker here is Redis, and the configuration doesn’t set a single task priority. It uses separate queues. The comments say why, queue by queue:

# AI pipeline (ai/tasks.py): LLM calls block for seconds — isolated so slow
# model I/O never delays the fast messages queue (parent spec Celery-queues
# table); runs on its own ECS task definition in production.
AI_QUEUE = "ai"
# Knowledge ingestion (ai/knowledge/tasks.py): parse/embed jobs are bursty
# and long-lived (a 50-page crawl or a 10MB PDF) — kept off `ai` so a large
# sync never starves live reply generation.
INGEST_QUEUE = "ingest"

# Comment and reaction webhooks (plan Stage C, C7). Bounded by reach rather
# than by MAC, so one campaign can put thousands of ingest tasks in front of a
# paying customer's message — measured at ~900 backed-up tasks, a probe behind
# them waited 1.45s against 0.06s on a quiet queue. Celery's Redis transport
# rotates between queues rather than draining one before the next, so this
# alone restores fairness: no second worker, and the existing one consumes it.
ENGAGEMENT_QUEUE = "engagement"

The engagement comment is the strongest argument in this article for lanes over priorities. A priority field on Redis would have been, in Celery’s own words, “approximate at best”. A separate queue that the same worker consumes in rotation turned a 1.45-second wait behind about 900 comment tasks back into the 0.06 seconds of a quiet queue, with no new worker. And “starves” appears in the ingest comment for a reason: starvation is the failure mode these lanes exist to prevent.

The webhook receiver picks the lane per payload, and its docstring states the rule for anything new: “Anything unrecognised goes to the conversation queue: wrong for volume, right for correctness”. When in doubt, the customer’s message lane, not the bulk lane.

What production adds to the textbook

• The heap is usually someone else’s. In this codebase, priority queues are ORDER BY clauses, event loop timers, broker queues and an HNSW index’s search. Knowing where the heap is tells you what an EXPLAIN plan or a slow timeout means.
• The tie-breaker is a design decision. created_at ascending for the engine, descending for the list people read, start_id in libuv, nothing at all in asyncio. Decide it on purpose.
• Changing a priority should explain itself. The work-item bump writes an event so the queue can say why an item jumped.
• Lanes beat priorities for capacity. A priority reorders waiting work. A lane gives work its own share of the workers.
• At small n, sort. A size-K heap is the right answer to an interview question and often the wrong change to a 30-item list.

Cheat sheet