Build an Owner for Background Tasks

Build BackgroundTasks, a small owner for background work. It keeps every task alive until it finishes, and drain() waits for a batch of tasks and raises the first failure.

background = BackgroundTasks()
background.spawn(send_email("a@example.com"))
background.spawn(send_email("b@example.com"))
background.pending()      # 2
await background.drain()  # waits for both, and raises if either failed
background.pending()      # 0

What to write

• spawn(coro) -> asyncio.Task[None] starts coro with asyncio.create_task, adds the task to the set self._tasks, and returns it. Register a done callback that removes the task from self._tasks when it finishes. Also append it to a list that records spawn order.
• pending() -> int counts spawned tasks where task.done() is False. len(self._tasks) gives the wrong answer, because the done callback runs a loop turn after the task finishes. For that moment the set still holds a finished task.
• async def drain() -> None works on one batch, the tasks spawned since the last drain:
  1. If the batch is empty, return at once.
  2. Wait until every task in the batch has finished, even after one fails.
  3. Skip cancelled tasks, and call exception() on every other task.
  4. Remove the batch from self._tasks and from your list.
  5. If any task failed, raise the exception of the failing task that was spawned first, even if a later one failed sooner.

The driver reads self._tasks, so keep that name.

Edge cases

• A second drain() with an empty batch returns at once, without raising the old failure again.
• Tasks spawned while drain() waits belong to the next batch.
• If drain() itself is cancelled while it waits, let CancelledError propagate. Keep the batch, with its tasks still running, for a later drain.
• You only need to handle one drain at a time.

What the driver reports

solve spawns one worker for each entry in hops. Worker i waits hops[i] loop turns, adds i to a log, then raises ValueError(f"task {i} failed") if i is in failing. The driver then runs settle loop turns, calling gc.collect() before each one, and calls drain().

• pending_start is pending() right after spawning, and pending_partial is pending() after the settle turns.
• survived_gc is how many workers had added themselves to the log by then. Despite its name, it counts finished workers rather than live task objects.
• log lists the workers in the order they finished.
• outcome is "ok", or the name of the exception drain() raised, and message is its text.
• pending_end and held_end are pending() and len(self._tasks) after the drain. Both must be 0. The tests check only the set, but clear your list too, or a long-running service keeps every task it ever started.