The Task Said Scheduled for 48 Hours. The Event Loop Only Held a Weak Reference.

Have you ever logged that a background task was scheduled, then watched nothing actually happen afterward? I did that during this forty-eight hour field note, and I blamed the mail provider first. The coroutine object existed, the handler had returned, and the JSON response still looked perfectly healthy to me. Did the work actually run, or did I only schedule a name that the event loop could drop?

This is the notebook I wish I had opened at hour zero, before I touched vendor dashboards. I am writing it as a lab log you can rerun, not as a postmortem stuffed with invented traffic numbers. Please treat every timestamp as a sequence of attempts, and run the snippets on your own interpreter.

Hour zero: the snippet that looked finished

I needed a handler that acknowledged signup and sent a follow-up without blocking the response path. The sketch I started from looked almost boring, which is usually how lifetime bugs hide in plain sight. Does this look finished to you as well, or do you already want an await?

# sketch only: handler shape, not a full web app
import asyncio

async def send_followup(user_id: str) -> None:
    await asyncio.sleep(0.05)  # stand-in for SMTP
    print(f"followup sent for {user_id}", flush=True)

async def handle_signup(user_id: str) -> dict:
    asyncio.create_task(send_followup(user_id))
    return {"ok": True, "user_id": user_id}

I called handle_signup("u-1") under asyncio.run, I received the ok JSON, and I waited for the print. Sometimes the print appeared while I kept typing in the same session. Sometimes it never appeared at all. Have you trusted a successful JSON body more than a missing side effect, the way I did here?

Hours 1–8: what I tried while the laptop stayed polite

On the laptop the process stayed busy with my debugging prints, so garbage collection often waited. Under a short test run the print vanished more often, which felt exactly like a flaky provider. I chased the wrong layer for most of a day, because inconsistent symptoms are catnip.

Here is the ordered list of dead ends I actually walked into:

1. I added retries around the fictional SMTP sleep, because retries feel like engineering progress.
2. I dumped os.environ and locale settings, because an earlier note of mine was an encoding trap.
3. I printed asyncio.all_tasks() right after create_task, and the set was not empty then.
4. I wrapped the coroutine in asyncio.wait_for(..., timeout=5) without saving the Task.
5. I restarted the process between clicks, which reset memory and hid the collector.

Why did item three convince me for several hours, even though it proved almost nothing? Because a non-empty task set looks like a promise from the scheduler itself. The loop will show you a task that it only weakly owns. Do you read all_tasks() as proof of lifetime, or only as a snapshot of this instant?

Hours 8–16: forcing the collector

The CPython event loop only keeps weak references to tasks, which I had treated as trivia instead of a contract. If your code drops the strong reference, collection can cancel the task before the first useful await returns. That warning sits in the asyncio.create_task documentation, and I had skipped it because the function name sounded final enough.

The laptop almost never collected a single pending task during one request. A quieter interpreter collected sooner once I stopped poking it. Want to see the failure without inventing production memory pressure you cannot replay?

# repro_gc_task.py — runnable field note
import asyncio
import gc
import sys

async def work(label: str) -> None:
    try:
        await asyncio.sleep(0.2)
        print(f"finished {label}", flush=True)
    except asyncio.CancelledError:
        print(f"cancelled {label}", flush=True)
        raise

async def fire_and_forget() -> None:
    asyncio.create_task(work("orphan"))
    # Local name is gone. The loop may hold only a weak reference.
    gc.collect()
    await asyncio.sleep(0.4)

async def keep_a_ref() -> None:
    task = asyncio.create_task(work("held"))
    gc.collect()
    await asyncio.sleep(0.4)
    await task

async def main() -> None:
    print("=== orphaned create_task ===", flush=True)
    await fire_and_forget()
    print("=== strong reference ===", flush=True)
    await keep_a_ref()

if __name__ == "__main__":
    sys.stdout.reconfigure(line_buffering=True)
    asyncio.run(main())

Commands I actually ran

python3 -m venv .venv
source .venv/bin/activate
python -X dev repro_gc_task.py

On this reproduction I expect cancelled orphan and later finished held, which matches the documented cancellation-on-collect behavior. If your output differs from that expectation, dump referrers before you declare the documentation wrong. Try this right after create_task and again after you drop the name:

task = asyncio.create_task(work("probe"))
print("refs before collect", len(gc.get_referrers(task)), flush=True)
del task
gc.collect()

Is that a scheduler bug in CPython? No, it is a lifetime bug that dresses like a scheduler bug. Once I could cancel the work with one gc.collect(), the provider dashboard stopped being interesting.

Hour 16: another pass, then another machine

I had too many hypotheses on one sticky note, and they all sounded equally plausible at that hour. I pasted the orphaned function into MonkeyCode because I wanted a second pass over lifetime rules, not another retry decorator. Disclosure: This article was prepared as part of MonkeyCode's product outreach.

The free model access helped as a reviewer of the snippet, not as an oracle of my running process. It put missing await, wrong event loop, and task dropped by GC on the same list, which at least restored the weak-reference warning. I reran repro_gc_task.py with the free server option so collection was not tied to my laptop's idle pattern.

Would I let a model rewrite the handler while I watched the diff scroll by? Not for this. I still wanted a test that failed the same way after the chat closed.

Hours 16–30: a test that fails on purpose

Field notes stay cheap until they become assertions you can rerun on a clean interpreter. This is the check I now keep beside any helper that looks like fire-and-forget. Copy it, then decide whether you like the hazard once it has a name.

# test_task_lifetime.py
import asyncio
import gc
import pytest

async def _slow(flag: dict) -> None:
    await asyncio.sleep(0.05)
    flag["done"] = True

@pytest.mark.asyncio
async def test_orphaned_task_may_never_finish() -> None:
    flag = {"done": False}
    asyncio.create_task(_slow(flag))
    gc.collect()
    await asyncio.sleep(0.1)
    # Tripwire only. Product code should not depend on this outcome.
    if not flag["done"]:
        pytest.xfail("orphaned create_task was collected before completion")

@pytest.mark.asyncio
async def test_strong_ref_survives_gc() -> None:
    flag = {"done": False}
    task = asyncio.create_task(_slow(flag))
    gc.collect()
    await task
    assert flag["done"] is True

@pytest.mark.asyncio
async def test_taskgroup_waits_for_children() -> None:
    flag = {"done": False}

    async def child() -> None:
        await _slow(flag)

    async with asyncio.TaskGroup() as tg:
        tg.create_task(child())
    assert flag["done"] is True

Install and run it from the same virtualenv you used for the reproduction script:

pip install pytest pytest-asyncio
pytest -q test_task_lifetime.py

The first test is a tripwire, not a product requirement you should invert later. If it xfails, the hazard showed up on that interpreter with that collector timing. If it passes, you still should not ship the orphaned pattern, because collection time is not part of your public API. Do you really want "works unless GC ran" as an SLA?

Hours 30–40: a registry, then a better default

I wanted a shim for code that cannot move to TaskGroup yet, without pretending the work is durable. A set of tasks with a done callback is explicit about who owns the strong reference. This is still in-process cooperative work, and I will keep repeating that limit.

# spawnutil.py
from collections.abc import Coroutine
from typing import Any
import asyncio

_BACKGROUND: set[asyncio.Task[Any]] = set()

def spawn(coro: Coroutine[Any, Any, Any]) -> asyncio.Task[Any]:
    """Schedule work and keep a strong reference until the task finishes."""
    task = asyncio.create_task(coro)
    _BACKGROUND.add(task)
    task.add_done_callback(_BACKGROUND.discard)
    return task

Hook the helper into the original signup sketch like this, and keep the return path unchanged:

async def handle_signup(user_id: str) -> dict:
    spawn(send_followup(user_id))
    return {"ok": True, "user_id": user_id}

If the process dies, the follow-up dies with it, which is the part fire-and-forget language keeps trying to hide. create_task is not a mail queue, and a set() does not become one either. Are you solving lifetime, or are you solving delivery?

Decision table I still keep

Approach	Who holds the strong ref	Cancels as a unit	Survives process exit	I use it when
Bare asyncio.create_task	Nobody unless I save it	No	No	Never inside handlers
Local variable plus await	The current frame	No	No	I can wait in this coroutine
set plus add_done_callback	The set	No	No	Older 3.10 code that must return now
asyncio.TaskGroup	The group	Yes, on child failure	No	3.11+ and I want structured errors
External queue and a worker	The broker	Depends on the worker	Yes, if the job is persisted	The work must outlive this process

Python 3.12 added an eager task factory, which can run a task until the first await before create_task even returns. That scheduling detail still does not replace a strong reference for tasks that await real I/O. I treat eager execution as a performance knob, never as a lifetime guarantee. If you enable it, rerun repro_gc_task.py before you trust muscle memory from 3.11.

Hours 40–48: what broke, and what I will repeat

Several things broke besides the original missing follow-up print, and each one looked like a different class of bug.

• A test helper that builds a fresh loop per test can leak a module-level task set without the done callback.
• asyncio.run() closes the loop and will complain if I stash tasks on a global that outlives main.
• Logging scheduled before the first await is not evidence that the coroutine body ran.
• Wrapping the coroutine in wait_for without keeping the Task still leaves only a weak reference.
• Restarting the app between attempts made the collector look shy, which wasted hours eight through twelve.

What I will repeat on the next incident, without negotiating with myself first:

1. Reproduce with an explicit gc.collect() before I blame a vendor dashboard.
2. Print len(gc.get_referrers(task)) when background work is flaky under light load.
3. Prefer TaskGroup when every child belongs to one request or one shutdown window.
4. Move the job to a broker when "must send" means "must send after this worker dies".
5. Keep the field note short enough that a second machine can rerun it without my laptop.

Limitations, and who should not copy this

This approach is a lifetime fix, not an observability stack and not a delivery guarantee. spawn() will not retry SMTP, will not persist across deploys, and will not bound concurrency by itself. If you need those properties, you want a worker and a queue, not a stronger set of Task objects.

Do not use this pattern if you need TaskGroup semantics on a Python older than 3.11 without a backport you already trust. Do not use it if the background work is CPU-bound, because the event loop is still one thread of control. Do not use it if payment or legal mail must not vanish when the process receives SIGKILL.

The model pass did not execute my tests, and the free server option did not add persistence I had not written. Both were useful only after I had a script that failed the same way twice. Would I start from the orphaned snippet again on purpose? No. I would start from the failing test, because a passing handler with a dropped task is how this ate two days.

If you want a second machine for the same collector repro, I reran mine with MonkeyCode's free server option.