The Mystery of the Vanishing Asyncio Worker

In Python 3.11 and 3.12, asyncio has become incredibly powerful for managing concurrent, long-running background tasks. However, many developers encounter a frustrating issue: a background worker task silently stops executing, leaving no traceback, no crash, and no error logs—even though exceptions seemed to be handled correctly.

If you have a worker loop that occasionally encounters a network glitch, recovers, and then suddenly stops updating your shared state, you are likely running into one of two common Python asyncio pitfalls: garbage collection of unreferenced tasks or unhandled background exceptions. Let's dive deep into why this happens and how to implement a bulletproof pattern for your background workers.

Why Do Asyncio Tasks Become Inactive?

1. The Garbage Collection Trap (Weak References)

Starting in recent Python versions, the asyncio event loop only keeps weak references to tasks created via asyncio.create_task(). If you do not keep a strong reference to the returned Task object elsewhere in your code, Python's garbage collector (GC) may destroy the task mid-execution, even if it is currently running.

In your original code:

tasks = [
    asyncio.create_task(worker(i))
    for i in range(3)
]

While asyncio.gather(*tasks) keeps a reference while it awaits, if the task layout changes or if tasks are spawned dynamically without maintaining a persistent reference, they can vanish silently. The Python documentation explicitly warns about this behavior.

2. Unhandled Exceptions Outside the Try-Except Block

In your worker loop, you catch ConnectionError:

try:
    if random.random() < 0.2:
        raise ConnectionError("Temporary API failure")
    cache[id] = "updated"
except ConnectionError as e:
    print(id, "recovered:", e)

But what happens if a different exception is raised? For example, a KeyError, a TypeError, or a CancelledError? If any exception other than ConnectionError occurs, it will bypass your catch block, propagate to the top of the task, and terminate the task. Because asyncio tasks fail silently unless they are explicitly awaited or monitored, your application will keep running, but the worker will be dead.

How to Detect If a Task is Still Alive

To prevent tasks from dying silently, you need mechanisms to monitor their state. Here are the three best ways to detect dead tasks:

Method A: Use task.add_done_callback()

You can register a callback that triggers immediately when a task finishes. This is the most reliable way to log failures or trigger restarts.

def handle_result(task):
    try:
        task.result()
    except asyncio.CancelledError:
        print("Task was cancelled")
    except Exception as e:
        print(f"Task failed with exception: {e}")

task = asyncio.create_task(worker(1))
task.add_done_callback(handle_result)

Method B: Query task.done()

You can periodically check the status of your tasks in a supervisor loop:

if task.done():
    if task.exception():
        print(f"Task died with exception: {task.exception()}")
    else:
        print("Task completed normally.")

The Recommended Pattern for Reliable Long-Running Workers

To build a resilient service in Python 3.12, you should combine strong references, broad exception handling, and a supervisor pattern (or asyncio.TaskGroup).

Here is the modernized, production-ready version of your code:

import asyncio
import random
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("WorkerSystem")

cache = {}
# Keep a strong reference to running tasks to prevent Garbage Collection
running_tasks = set()

async def worker(worker_id):
    while True:
        try:
            # Your actual business logic
            if random.random() < 0.2:
                raise ConnectionError("Temporary API failure")
            
            cache[worker_id] = "updated"
            logger.info(f"Worker {worker_id} updated state.")

        except ConnectionError as e:
            logger.warning(f"Worker {worker_id} handled transient error: {e}")
        except Exception as e:
            # Catch-all to prevent the worker loop from dying due to unexpected bugs
            logger.error(f"Worker {worker_id} encountered an unexpected crash: {e}", exc_info=True)
        
        # Sleep to yield control back to the event loop
        await asyncio.sleep(1)

async def main():
    # Spawn workers and maintain strong references
    for i in range(3):
        task = asyncio.create_task(worker(i), name=f"worker-{i}")
        running_tasks.add(task)
        
        # Automatically discard the task from our set when it finishes
        task.add_done_callback(running_tasks.discard)

    # Keep the main loop alive and monitor background tasks
    while True:
        await asyncio.sleep(5)
        logger.info(f"Active background workers: {len(running_tasks)}")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("Application shutting down.")

Key Takeaways for Production

  • Always keep a strong reference: Store your tasks in a global or class-level set so the garbage collector doesn't reap them.
  • Catch broad exceptions: Inside your infinite while True loop, wrap your code in a generic except Exception block to prevent unexpected errors from killing the thread of execution.
  • Use proper logging: Avoid raw print() statements. Use Python's logging library with exc_info=True to get full stack traces of why your workers are failing.