gh-157299: Fix missing awaited-by edge in eager tasks (#157300)

Author: deadlovelll

Doc/library/asyncio-task.rst

@@ -666,6 +666,10 @@ Eager task factory
         changes to existing applications. For example,
         the application's task execution order is likely to change.
 
+        During the eager step, call graph introspection reports the creating
+        task as an awaiter, even for a plain :func:`create_task` call.
+        This relationship is removed when the eager step finishes.
+
     .. versionadded:: 3.12
 
 .. function:: create_eager_task_factory(custom_task_constructor)

Lib/asyncio/tasks.py

@@ -257,9 +257,12 @@ def __eager_start(self):
         prev_task = _py_swap_current_task(self._loop, self)
         try:
             _py_register_eager_task(self)
+            # gh-157299: the eager step must record the awaited_by edge
+            futures.future_add_to_awaited_by(self, prev_task)
             try:
                 self._context.run(self.__step_run_and_handle_result, None)
             finally:
+                futures.future_discard_from_awaited_by(self, prev_task)
                 _py_unregister_eager_task(self)
         finally:
             try:

Lib/test/test_asyncio/test_eager_task_factory.py

@@ -69,6 +69,20 @@ async def run():
 
         self.assertEqual(self.run_coro(run()), 'my message')
 
+    def test_awaited_by_during_eager_step(self):
+        async def run():
+            parent = asyncio.current_task()
+
+            async def coro():
+                self.assertEqual(asyncio.current_task()._asyncio_awaited_by,
+                                 frozenset({parent}))
+
+            t = self.loop.create_task(coro())
+            self.assertFalse(t._asyncio_awaited_by)
+            await t
+
+        self.run_coro(run())
+
     def test_eager_completion(self):
 
         async def coro():
@@ -292,11 +306,29 @@ def setUp(self):
         self._current_task = asyncio.current_task
         asyncio.current_task = asyncio.tasks.current_task = asyncio.tasks._py_current_task
         asyncio.all_tasks = asyncio.tasks.all_tasks = asyncio.tasks._py_all_tasks
+
+        futures = asyncio.futures
+
+        self._future_add_to_awaited_by = asyncio.future_add_to_awaited_by
+        futures.future_add_to_awaited_by = futures._py_future_add_to_awaited_by
+        asyncio.future_add_to_awaited_by = futures.future_add_to_awaited_by
+
+        self._future_discard_from_awaited_by = asyncio.future_discard_from_awaited_by
+        futures.future_discard_from_awaited_by = futures._py_future_discard_from_awaited_by
+        asyncio.future_discard_from_awaited_by = futures.future_discard_from_awaited_by
         return super().setUp()
 
     def tearDown(self):
         asyncio.current_task = asyncio.tasks.current_task = self._current_task
         asyncio.all_tasks = asyncio.tasks.all_tasks = self._all_tasks
+
+        futures = asyncio.futures
+
+        futures.future_discard_from_awaited_by = self._future_discard_from_awaited_by
+        asyncio.future_discard_from_awaited_by = self._future_discard_from_awaited_by
+
+        futures.future_add_to_awaited_by = self._future_add_to_awaited_by
+        asyncio.future_add_to_awaited_by = self._future_add_to_awaited_by
         return super().tearDown()
 
 
@@ -318,6 +350,49 @@ def tearDown(self):
         asyncio.all_tasks = asyncio.tasks.all_tasks = self._all_tasks
         return super().tearDown()
 
+    def test_eager_step_with_non_native_parent(self):
+        async def child():
+            self.assertFalse(asyncio.current_task()._asyncio_awaited_by)
+
+        async def run():
+            parent = tasks._c_swap_current_task(self.loop, object())
+            try:
+                task = self.loop.create_task(child())
+            finally:
+                tasks._c_swap_current_task(self.loop, parent)
+            await task
+
+        self.run_coro(run())
+
+    def test_eager_step_preserves_exception(self):
+        # gh-157299: check that the exception is saved and cleared before
+        # discarding the awaited-by edge, and restored afterwards.
+        # A default task can still pass with the wrong order since its hash
+        # is in C. Overriding __hash__ makes cleanup call back into Python.
+        # If the exception is still pending, this raises SystemError.
+        class ParentTask(self.Task):
+            def __hash__(self):
+                return super().__hash__()
+
+        child_tasks = []
+        exc = SystemExit("eager step")
+
+        async def child():
+            task = asyncio.current_task()
+            child_tasks.append(task)
+            asyncio.future_add_to_awaited_by(task, self.loop.create_future())
+            raise exc
+
+        async def parent():
+            with self.assertRaises(SystemExit) as cm:
+                self.loop.create_task(child())
+            task, = child_tasks
+            self.assertIs(cm.exception, exc)
+            self.assertIs(task.exception(), exc)
+            self.assertNotIn(asyncio.current_task(), task._asyncio_awaited_by)
+
+        self.loop.run_until_complete(ParentTask(parent(), loop=self.loop))
+
     def test_issue105987(self):
         code = """if 1:
         from _asyncio import _swap_current_task, _set_running_loop

Lib/test/test_asyncio/test_graph.py

@@ -969,6 +969,23 @@ async def main():
         await main()
         self.assertEqual(stack[:3], ['gen', 'middle', 'main'])
 
+    def set_eager_task_factory(self):
+        loop = asyncio.get_running_loop()
+        loop.set_task_factory(asyncio.create_eager_task_factory(asyncio.Task))
+        self.addCleanup(loop.set_task_factory, None)
+
+    async def test_stack_eager_task(self):
+        self.set_eager_task_factory()
+
+        async def child():
+            nonlocal stack
+            stack = capture_test_stack()
+
+        stack = None
+        await asyncio.gather(child())
+
+        self.assertEqual(stack[0][2], [['T<anon>', ['a test_stack_eager_task'], []]])
+
 
 @unittest.skipIf(
     not hasattr(asyncio.futures, "_c_future_add_to_awaited_by"),

Misc/NEWS.d/next/Library/2026-09-11-12-43-04.gh-issue-157299.HQ3I4F.rst

@@ -0,0 +1,4 @@
+Fix eagerly started :class:`asyncio.Task` objects missing their awaited-by
+edge during the first step. During that step, even a plain
+:func:`asyncio.create_task` call reports the creating task as an awaiter.
+Patch by Timofey Ivankov.

Modules/_asynciomodule.c

@@ -3491,17 +3491,36 @@ task_eager_start(_PyThreadStateImpl *ts, asyncio_state *state, TaskObj *task)
 
     int retval = 0;
 
+    // gh-157299: the eager step must record the awaited_by edge
+    int eager_edge = TaskOrFuture_Check(state, prevtask);
     PyObject *stepres;
     Py_BEGIN_CRITICAL_SECTION(task);
-    stepres = task_step_impl(state, task, NULL);
-    Py_END_CRITICAL_SECTION();
+    if (eager_edge &&
+        future_awaited_by_add(state, (FutureObj *)task, prevtask) < 0)
+    {
+        eager_edge = 0;
+        task->task_log_destroy_pending = 0;
+        unregister_task(task);
+        stepres = NULL;
+    }
+    else {
+        stepres = task_step_impl(state, task, NULL);
+    }
+
+    PyObject *exc = stepres == NULL ? PyErr_GetRaisedException() : NULL;
+    if (eager_edge) {
+        int res = future_awaited_by_discard(state, (FutureObj *)task, prevtask);
+        if (res) {
+            retval = -1;
+        }
+    }
     if (stepres == NULL) {
-        PyObject *exc = PyErr_GetRaisedException();
         _PyErr_ChainExceptions1(exc);
         retval = -1;
     } else {
         Py_DECREF(stepres);
     }
+    Py_END_CRITICAL_SECTION();
 
     PyObject *curtask = swap_current_task(ts, task->task_loop, prevtask);
     Py_DECREF(prevtask);