BackgroundTasks: strong references, and a drain that raises

Own your fire-and-forget tasks.

class BackgroundTasks:
    def __init__(self) -> None: ...
    def spawn(self, coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]: ...
    async def drain(self) -> None: ...
    def pending(self) -> int: ...

• spawn schedules coro, retains a strong reference for its lifetime, drops that reference when it finishes, and returns the Task.
• drain waits for every spawned task and re-raises the failure of the earliest-spawned failing task.
• pending is the number of spawned tasks that have not finished.

Keep the strong references in self._tasks — the driver reads it directly to check that completed tasks are not retained.

Why this class exists

The stdlib warns, in bold, that the event loop keeps only weak references to tasks: “a task that isn’t referenced elsewhere may get garbage-collected at any time, even before it’s done.” So asyncio.create_task(refresh()) on a line by itself discards the only strong reference the task had, and whether it survives depends on whether a collection happens to run at the wrong moment — which depends on allocation pressure, which depends on load. It works in tests and fails in production.

The other half is the exception. A Task holds its exception until someone asks; if nobody ever does, it surfaces at garbage-collection time as “Task exception was never retrieved”, timestamped whenever the collector ran rather than when the failure happened.

What the report proves

• pending_start — every spawned task is accounted for before the loop has run a single one.
• survived_gc / pending_partial — the driver calls gc.collect() several times mid-flight. Tasks in progress must still be there afterwards.
• log — completion order, driven by each task’s hop count.
• outcome / message — drain() surfaces the earliest-spawned failure. “Earliest-spawned” rather than “first to fail” is what makes it reproducible.
• pending_end — 0 after drain.
• held_end — len(self._tasks) after drain, also 0. Retaining completed tasks turns the leak around: a long-lived service accumulates one Task, its frames and everything they close over, per spawn.

Note that drain must surface the earliest-spawned failure even when a later task failed first, and must still wait for every task before raising.

Where the type system earns its keep

set[asyncio.Task[None]] is the annotation that does the teaching. The None forces you to notice that a fire-and-forget task’s return value goes nowhere — if the coroutine returns something, either somebody must await it or you are discarding a result. And add_done_callback expects a Callable[[Task[None]], object], which set.discard satisfies exactly; that is not a coincidence, it is the idiom the documentation prescribes.