This handler processes an API request and emits a latency metric on the side. The metric call is deliberately fire-and-forget — the caller shouldn't wait on a network round-trip to StatsD just to return a response. It passes code review, ships, and works fine in tests. In production, roughly one metric in twenty silently disappears — never sent, never logged, never raised.

import asyncio, logging, random

async def send_metric(name: str, value: float) -> None:
    await asyncio.sleep(0.05)               # network call to metrics backend
    if random.random() < 0.01:
        raise RuntimeError("metrics backend down")
    logging.info("sent %s=%s", name, value)

async def handle(req):
    await asyncio.sleep(0.01)
    return {"ms": 12.3}

async def process_request(req):
    result = await handle(req)
    # Fire-and-forget: don't block the response on metrics.
    asyncio.create_task(send_metric("request.latency", result["ms"]))
    return result

async def main():
    await asyncio.gather(*(process_request({}) for _ in range(1000)))

asyncio.run(main())

The Bug

The event loop keeps only a weak reference to tasks created by asyncio.create_task. The only strong reference in this code is the return value of create_task(...) — and it's discarded on the next line. As soon as the garbage collector runs (which it will, under load, at unpredictable moments), any pending task with no other strong owner becomes eligible for collection. Python destroys the task mid-await, prints a terse Task was destroyed but it is pending! warning to stderr if you're lucky enough to have logging wired to it, and moves on.

Even worse: exceptions raised inside a garbage-collected task are simply dropped. That RuntimeError("metrics backend down") above never surfaces anywhere. Your dashboards report 100% success while metrics silently vanish. In staging you can't reproduce it because the GC threshold isn't hit; the bug shows up only under real traffic and long-lived processes.

This is documented — but easy to miss — in the CPython source and the asyncio docs: "Save a reference to the result of this function, to avoid a task disappearing mid-execution." The event loop's task registry uses a WeakSet so that completed tasks can be collected promptly; the price is that uncompleted tasks can also be collected if you don't hold them.

The Fix

Hold a strong reference until the task finishes. The canonical pattern is a module-level set plus a done-callback that removes the task after completion:

_background_tasks: set[asyncio.Task] = set()

async def process_request(req):
    result = await handle(req)
    task = asyncio.create_task(send_metric("request.latency", result["ms"]))
    _background_tasks.add(task)
    task.add_done_callback(_background_tasks.discard)
    return result

Now every in-flight task has a strong referent (the set), the GC leaves it alone, and the done_callback tidies up so the set doesn't grow unbounded. As a bonus, add_done_callback gives you a natural hook to log exceptions (task.exception()) that would otherwise vanish into the void.

For structured concurrency in Python 3.11+, prefer asyncio.TaskGroup, which owns its child tasks and propagates their exceptions deterministically — no weak-reference footgun.