python.async_task_no_error_handling

Stability Medium

Detects asyncio.create_task() or asyncio.ensure_future() called without error handling.

Why It Matters

Background tasks swallow exceptions silently:

• Ghost failures — Operations appear to succeed but don’t
• Silent data loss — Writes fail without notification
• No alerts — Your monitoring never sees the error
• Debugging nightmare — Issues only surface through data inconsistency

Python’s event loop logs unhandled task exceptions, but by the time you see the log, the context is lost.

Example

# ❌ Before
async def main():
    asyncio.create_task(risky_operation())
    # If risky_operation fails, you'll never know

# ✅ After
async def main():
    task = asyncio.create_task(risky_operation())
    task.add_done_callback(handle_task_result)

def handle_task_result(task):
    try:
        task.result()
    except Exception as e:
        logger.error("Task failed", exc_info=e)

What Unfault Detects

• asyncio.create_task() without storing the result
• asyncio.create_task() without add_done_callback
• asyncio.ensure_future() without error handling
• Fire-and-forget patterns without try/except in the coroutine

Auto-Fix

Unfault adds a done callback that logs exceptions.

Best Practices

# Pattern 1: Callback-based handling
def handle_task_exception(task: asyncio.Task):
    try:
        task.result()
    except asyncio.CancelledError:
        pass  # Task was cancelled, usually intentional
    except Exception as e:
        logger.exception("Background task failed")

task = asyncio.create_task(work())
task.add_done_callback(handle_task_exception)

# Pattern 2: Await and handle
async def supervised_work():
    try:
        await work()
    except Exception as e:
        logger.exception("Work failed")

asyncio.create_task(supervised_work())

# Pattern 3: TaskGroup (Python 3.11+)
async with asyncio.TaskGroup() as tg:
    tg.create_task(work())  # Exceptions propagate to caller

Related Rules

• typescript.promise_no_catch