Detecting leaked tasks and forcing a race

A concurrency test that “usually passes” is not a test. It is a coin flip with a CI bill attached, and the first thing a team does with a flaky test is retry it — which converts a real bug into background noise.

Every concurrency bug in this course is teachable only because it can be made to fail reliably. The same techniques are what let you assert, in CI, that a shutdown path drains cleanly. Two of them here.

What to write

def assert_no_task_leaks(
    body: Callable[[], Awaitable[int]]
) -> tuple[int, list[str]]

Run body on its own event loop, and return its result together with the sorted names of every task it left pending. Then cancel and drain those tasks so the loop closes quietly.

A leaked task is the most common asyncio defect there is. asyncio.create_task(...) without holding the result gives you a task nobody awaits, whose exceptions nobody sees, and which the loop may cancel mid-write at shutdown. It never fails a test, because the test already returned.

Two details decide whether this works:

• Snapshot asyncio.all_tasks() before and after, inside the loop. It returns only pending tasks, so anything body awaited to completion has already dropped out and only genuine leaks remain. Exclude asyncio.current_task() — the runner is itself a task.
• Check immediately after body returns, before the loop shuts down. asyncio.run cancels stragglers on its way out; look afterwards and there is nothing left to find.

Use asyncio.Runner (3.11+) rather than asyncio.run. Same loop lifecycle, but as a context manager, so a test can run several coroutines on one loop and control exactly when it closes.

def force_race(n: int, fn: Callable[[], None]) -> None

Start n threads, hold every one of them at a threading.Barrier, and release them into fn simultaneously. Return when all n have finished.

threading.Barrier(n).wait() blocks until the n-th thread arrives, then releases all of them at once. That is the difference between “started n threads and hoped” — where thread 1 usually finishes before thread 2 is scheduled, and the race never happens — and actually putting n threads inside the same critical section. It is the standard way to make an unsynchronised counter fail on the first run instead of the four-hundredth.

n <= 0 does nothing; threading.Barrier(0) is an error.

solve is provided. It runs the body, then races race_threads threads through a locked counter, so the reported total is exactly the thread count — the point being that with the lock in place the barrier proves correctness rather than merely provoking a failure.

The rest of the toolkit

Neither function measures elapsed time, and neither uses sleep to “let things settle”. A concurrency test that depends on a duration is a test that fails on a loaded CI runner. Reach for a controllable clock, an asyncio.Event, or a threading.Barrier to pin the interleaving you want, and assert on an artefact: an ordering, a count, a recorded event log.

Two more settings worth putting in pyproject.toml on the day you write your first async test. -W error::RuntimeWarning turns “coroutine was never awaited” from a message nobody reads into a failure. And asyncio’s debug mode reports slow callbacks and tasks destroyed while pending — noise in production, exactly the signal you want in CI.