Retain the chunk-received trace task in read_nowait

CHANGES/13300.bugfix.rst

@@ -0,0 +1,3 @@
+Retained the chunk-received trace task scheduled by
+:py:meth:`~aiohttp.StreamReader.read_nowait`, so the tracing hook can no
+longer be garbage collected before it fires -- by :user:`noron12234`.

aiohttp/streams.py

@@ -102,6 +102,7 @@ class StreamReader:
         "_eof_callbacks",
         "_eof_counter",
         "_on_chunk_received",
+        "_chunk_received_tasks",
         "total_bytes",
         "total_compressed_bytes",
     )
@@ -138,6 +139,10 @@ def __init__(
         self._on_chunk_received: (
             Callable[[bytes], Coroutine[None, None, None]] | None
         ) = None
+        # Allocated lazily: a StreamReader is built per response, but the
+        # chunk-received hook is only set when tracing is enabled, so an
+        # eager set() would cost every response for a rarely-used feature.
+        self._chunk_received_tasks: set[asyncio.Task[None]] | None = None
         self.total_bytes = 0
         self.total_compressed_bytes: int | None = None
 
@@ -536,9 +541,16 @@ def read_nowait(self, n: int = -1) -> bytes:
         chunk = self._read_nowait(n)
         if chunk and (cb := self._on_chunk_received) is not None:
             # read_nowait is sync but the hook is async; schedule it so the
-            # observability event still fires.
-            # TODO: Save and await this task.
-            asyncio.create_task(cb(chunk))  # type: ignore[unused-awaitable]
+            # observability event still fires. The loop only holds a weak
+            # reference to a task, so keep one here until it completes —
+            # otherwise the trace event can be collected before it fires.
+            # TODO: this still bypasses the `self._timer` bound that
+            # _fire_chunk_received applies; awaiting it needs a sync/async split.
+            task = asyncio.create_task(cb(chunk))
+            if (tasks := self._chunk_received_tasks) is None:
+                tasks = self._chunk_received_tasks = set()
+            tasks.add(task)
+            task.add_done_callback(tasks.discard)
         return chunk
 
     def _read_nowait_chunk(self, n: int) -> bytes:

tests/test_streams.py

@@ -1296,6 +1296,42 @@ async def test_read_nowait_fires_via_task(self) -> None:
         await asyncio.sleep(0)
         assert seen == [b"abc"]
 
+    async def test_read_nowait_holds_the_task_then_releases_it(self) -> None:
+        # The loop keeps only a weak reference to a task, so the scheduled
+        # hook needs a strong referent of its own until it completes. This
+        # asserts the reference exists while pending and is dropped after,
+        # so holding it cannot turn into a leak.
+        stream, seen = self._make_one()
+        stream.feed_data(b"abc")
+
+        assert stream.read_nowait() == b"abc"
+        assert stream._chunk_received_tasks is not None
+        assert len(stream._chunk_received_tasks) == 1
+
+        # A second call reuses the already-allocated set rather than
+        # replacing it.
+        stream.feed_data(b"def")
+        assert stream.read_nowait() == b"def"
+        assert len(stream._chunk_received_tasks) == 2
+
+        await asyncio.sleep(0)
+        assert seen == [b"abc", b"def"]
+
+        # add_done_callback is delivered via call_soon, so the discard lands
+        # on the tick after the task finishes rather than synchronously.
+        await asyncio.sleep(0)
+        assert not stream._chunk_received_tasks
+
+    async def test_no_hook_does_not_allocate_the_task_set(self) -> None:
+        # A StreamReader is built per response but the hook is only set when
+        # tracing is enabled, so the set stays unallocated in the common case.
+        stream, _ = self._make_one()
+        stream._on_chunk_received = None
+        stream.feed_data(b"abc")
+
+        assert stream.read_nowait() == b"abc"
+        assert stream._chunk_received_tasks is None
+
     async def test_hook_exception_propagates(self) -> None:
         async def cb(chunk: bytes) -> None:
             raise RuntimeError("boom")