From 925225ec4471fa0da751aa86d3113e931f7f69a2 Mon Sep 17 00:00:00 2001 From: Carsten Piepel Date: Fri, 9 Oct 2026 08:50:56 -0400 Subject: [PATCH 1/7] fix(server): mark internal lifecycle events done so a cancelled stream frees its producer ActiveTask.subscribe consumed _RequestStarted and _RequestCompleted with a bare `continue`, never calling tapped_queue.task_done() for them. When a task finished normally the _RequestCompleted event made the subscriber return before the producer's cleanup, so it did no harm. But a cancel stops the agent before _RequestCompleted is sent: the producer then runs _event_queue_subscribers.close(immediate=False), which waits until every tap has marked all of its items done. The stream's tap still holds the unmarked _RequestStarted, so the wait never ends -- the producer and consumer tasks and the ActiveTask registry entry leak until the whole handler is aclose()-d. Mark the two internal lifecycle events done before skipping them. Adds a regression test that fails (TimeoutError) without the fix and passes with it. Fixes #1322 Signed-off-by: Carsten Piepel --- src/a2a/server/agent_execution/active_task.py | 8 ++ .../agent_execution/test_active_task.py | 83 +++++++++++++++++++ 2 files changed, 91 insertions(+) diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index a2f174367..690f4bd5b 100644 --- a/src/a2a/server/agent_execution/active_task.py +++ b/src/a2a/server/agent_execution/active_task.py @@ -760,6 +760,14 @@ async def subscribe( updated_task, ) event = updated_task + # Internal lifecycle events are consumed here rather than + # yielded, so mark them done before skipping. Otherwise the + # producer's `_event_queue_subscribers.close(immediate=False)` + # waits forever on an item this tap never acknowledges, which + # leaks the producer and the ActiveTask after a cancel ends + # the subscriber before _RequestCompleted is sent (#1322). + if isinstance(event, _RequestStarted | _RequestCompleted): + tapped_queue.task_done() if isinstance(event, _RequestCompleted): if ( request_id is not None diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py index 58b9eb8c2..d8aa203a5 100644 --- a/tests/server/agent_execution/test_active_task.py +++ b/tests/server/agent_execution/test_active_task.py @@ -1,4 +1,5 @@ import asyncio +import contextlib import logging from unittest.mock import AsyncMock, Mock, patch @@ -922,6 +923,88 @@ async def slow_execute(req, q): async for _ in active_task.subscribe(): pass + @pytest.mark.asyncio + @pytest.mark.timeout(5) + async def test_active_task_cancel_while_subscribed_frees_producer( + self, + active_task: ActiveTask, + agent_executor: Mock, + request_context: Mock, + task_manager: Mock, + ) -> None: + """Regression (#1322): a cancel while a subscriber is streaming must + not leave the producer wedged. + + The producer closes `_event_queue_subscribers` with + `immediate=False`, which blocks until every tap has marked all of its + items done. ``subscribe`` consumes the internal ``_RequestStarted`` + event without yielding it, so it must also mark it done -- otherwise a + cancel that ends the subscriber before ``_RequestCompleted`` is sent + leaves that item unacknowledged, the producer blocks forever on + ``close(immediate=False)``, and both the producer task and the + ``ActiveTask`` leak until the whole handler is ``aclose``-d. + """ + hang = asyncio.Event() + + async def execute_mock(req, q): + await hang.wait() # long-running work, cancelled from outside + + agent_executor.execute = AsyncMock(side_effect=execute_mock) + task_manager.get_task = AsyncMock( + return_value=Task( + id='test-task-id', + status=TaskStatus(state=TaskState.TASK_STATE_WORKING), + ) + ) + + await active_task.enqueue_request(request_context) + await active_task.start( + call_context=ServerCallContext(), create_task_if_missing=True + ) + producer = active_task._producer_task + assert producer is not None + + # Drive a subscriber far enough to dequeue and skip _RequestStarted, + # then stop reading -- exactly the state a cancelled stream leaves. + subscription = active_task.subscribe(request=request_context) + + async def drain() -> None: + async for _event in subscription: + # The internal _RequestStarted the producer enqueues is + # consumed inside subscribe() and never reaches us; real work + # items would. + pass + + reader = asyncio.create_task(drain()) + # Let the producer enqueue _RequestStarted and the subscriber consume it. + await asyncio.sleep(0.1) + + # Cancel the producer: this is what the SDK does after the executor's + # cancel() returns on a streamed task. execute() never gets to emit + # _RequestCompleted. + producer.cancel() + + # The producer must reach completion. Before the fix it blocks forever + # inside `_event_queue_subscribers.close(immediate=False)` waiting on + # the unacknowledged _RequestStarted, and this await times out. + await asyncio.wait_for(asyncio.shield(_settle(producer)), timeout=2) + assert producer.done(), ( + 'producer never finished -> it leaked waiting on an ' + 'unacknowledged _RequestStarted (#1322)' + ) + + reader.cancel() + with contextlib.suppress(asyncio.CancelledError): + await reader + hang.set() + await active_task.aclose() + + +async def _settle(task: asyncio.Task) -> None: + """Wait for ``task`` to finish, swallowing its CancelledError/result.""" + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + @pytest.mark.asyncio async def test_active_task_subscribe_include_initial_task(): From 4154826bfcf212531ee1f0aea4d96cf0093f9d88 Mon Sep 17 00:00:00 2001 From: Carsten Piepel Date: Mon, 12 Oct 2026 10:07:47 -0400 Subject: [PATCH 2/7] test(server): assert cancelled producer exits cleanly, remove fixed sleep Address review feedback on the #1322 regression test: - Stop swallowing the producer's exception. The previous _settle helper suppressed all exceptions, so the test could pass if teardown crashed rather than completed. Now assert the producer finishes with no exception; the CancelledError from the explicit cancel is the only one allowed, closing a false-pass hole. - Replace the fixed sleep with an explicit synchronization event set when the subscriber enters its loop, removing a timing guess that could flake on a slow runner. Re-verified: the test still passes with the fix and still fails with TimeoutError when the source fix is reverted. Signed-off-by: Carsten Piepel --- .../agent_execution/test_active_task.py | 39 +++++++++++++------ 1 file changed, 27 insertions(+), 12 deletions(-) diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py index d8aa203a5..6426a2c3f 100644 --- a/tests/server/agent_execution/test_active_task.py +++ b/tests/server/agent_execution/test_active_task.py @@ -968,7 +968,15 @@ async def execute_mock(req, q): # then stop reading -- exactly the state a cancelled stream leaves. subscription = active_task.subscribe(request=request_context) + consuming = asyncio.Event() + async def drain() -> None: + # Signal once the subscriber generator has entered its loop: at + # that point the tap exists and the async-for is actively pulling, + # so the producer's _RequestStarted will be dequeued and skipped + # here. This replaces a fixed sleep with a real happens-before + # edge, so the test does not flake on a slow runner. + consuming.set() async for _event in subscription: # The internal _RequestStarted the producer enqueues is # consumed inside subscribe() and never reaches us; real work @@ -976,22 +984,35 @@ async def drain() -> None: pass reader = asyncio.create_task(drain()) - # Let the producer enqueue _RequestStarted and the subscriber consume it. - await asyncio.sleep(0.1) + # Wait until the subscriber is actively consuming, then yield once more + # so the producer's enqueued _RequestStarted is dequeued and skipped. + await asyncio.wait_for(consuming.wait(), timeout=2) + await asyncio.sleep(0) # Cancel the producer: this is what the SDK does after the executor's # cancel() returns on a streamed task. execute() never gets to emit # _RequestCompleted. producer.cancel() - # The producer must reach completion. Before the fix it blocks forever - # inside `_event_queue_subscribers.close(immediate=False)` waiting on - # the unacknowledged _RequestStarted, and this await times out. - await asyncio.wait_for(asyncio.shield(_settle(producer)), timeout=2) + # The producer must reach completion CLEANLY. Before the fix it blocks + # forever inside `_event_queue_subscribers.close(immediate=False)` + # waiting on the unacknowledged _RequestStarted, and this await times + # out. We do NOT swallow the producer's exception: a cancel leaves it + # with CancelledError (expected), but any other exception means + # teardown crashed rather than completed, and the test must fail + # instead of passing on a merely-``done()`` task. + try: + await asyncio.wait_for(asyncio.shield(producer), timeout=2) + except asyncio.CancelledError: + pass # expected: the producer was cancelled assert producer.done(), ( 'producer never finished -> it leaked waiting on an ' 'unacknowledged _RequestStarted (#1322)' ) + exc = producer.exception() + assert exc is None or isinstance(exc, asyncio.CancelledError), ( + f'producer did not shut down cleanly; teardown raised {exc!r}' + ) reader.cancel() with contextlib.suppress(asyncio.CancelledError): @@ -1000,12 +1021,6 @@ async def drain() -> None: await active_task.aclose() -async def _settle(task: asyncio.Task) -> None: - """Wait for ``task`` to finish, swallowing its CancelledError/result.""" - with contextlib.suppress(asyncio.CancelledError, Exception): - await task - - @pytest.mark.asyncio async def test_active_task_subscribe_include_initial_task(): agent_executor = Mock() From 5f3745de0d21e36773bcac292ef74a54800db6de Mon Sep 17 00:00:00 2001 From: Carsten Piepel Date: Mon, 12 Oct 2026 10:17:05 -0400 Subject: [PATCH 3/7] test(server): simplify producer-completion assertion and fix sync comment Second review pass on the #1322 regression test: - _run_producer catches asyncio.CancelledError and returns normally, so a cancelled producer never surfaces CancelledError from await and producer.exception() is always None here. Drop the dead CancelledError handling and the redundant done()/exception() checks; just await the producer. If teardown raises, await re-raises and the test fails -- the stronger property. Verified by injecting a RuntimeError into teardown. - Correct the synchronization comment: setting the event before the first await does not prove _RequestStarted was consumed. Describe the actual scheduling handoff instead of overstating a guarantee. Re-verified: passes with the fix, fails with TimeoutError when the fix is reverted, and fails by propagating the exception when teardown raises. Signed-off-by: Carsten Piepel --- .../agent_execution/test_active_task.py | 45 ++++++++----------- 1 file changed, 19 insertions(+), 26 deletions(-) diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py index 6426a2c3f..7b0f035e4 100644 --- a/tests/server/agent_execution/test_active_task.py +++ b/tests/server/agent_execution/test_active_task.py @@ -971,11 +971,13 @@ async def execute_mock(req, q): consuming = asyncio.Event() async def drain() -> None: - # Signal once the subscriber generator has entered its loop: at - # that point the tap exists and the async-for is actively pulling, - # so the producer's _RequestStarted will be dequeued and skipped - # here. This replaces a fixed sleep with a real happens-before - # edge, so the test does not flake on a slow runner. + # Signalled just before the first await on the subscription. This + # does NOT by itself prove _RequestStarted has been consumed -- it + # marks that the drain coroutine is scheduled and about to pull -- + # but combined with the sleep(0) below it reliably hands control to + # the reader so it dequeues and skips the started event. It is still + # tighter than a fixed sleep, and the final assertion (the producer + # must complete) is what actually proves the leak is gone. consuming.set() async for _event in subscription: # The internal _RequestStarted the producer enqueues is @@ -984,8 +986,9 @@ async def drain() -> None: pass reader = asyncio.create_task(drain()) - # Wait until the subscriber is actively consuming, then yield once more - # so the producer's enqueued _RequestStarted is dequeued and skipped. + # Wait until the reader is about to consume, then yield the event loop + # so it runs far enough to dequeue and skip the producer's + # _RequestStarted before we cancel. await asyncio.wait_for(consuming.wait(), timeout=2) await asyncio.sleep(0) @@ -994,25 +997,15 @@ async def drain() -> None: # _RequestCompleted. producer.cancel() - # The producer must reach completion CLEANLY. Before the fix it blocks - # forever inside `_event_queue_subscribers.close(immediate=False)` - # waiting on the unacknowledged _RequestStarted, and this await times - # out. We do NOT swallow the producer's exception: a cancel leaves it - # with CancelledError (expected), but any other exception means - # teardown crashed rather than completed, and the test must fail - # instead of passing on a merely-``done()`` task. - try: - await asyncio.wait_for(asyncio.shield(producer), timeout=2) - except asyncio.CancelledError: - pass # expected: the producer was cancelled - assert producer.done(), ( - 'producer never finished -> it leaked waiting on an ' - 'unacknowledged _RequestStarted (#1322)' - ) - exc = producer.exception() - assert exc is None or isinstance(exc, asyncio.CancelledError), ( - f'producer did not shut down cleanly; teardown raised {exc!r}' - ) + # The producer must reach completion CLEANLY. _run_producer catches + # asyncio.CancelledError internally and returns normally, so awaiting + # it here does NOT re-raise -- it returns the producer's result once + # teardown finishes. Before the fix, teardown blocks forever inside + # `_event_queue_subscribers.close(immediate=False)` waiting on the + # unacknowledged _RequestStarted, and this await times out. If teardown + # instead raises, the await re-raises it and the test fails -- which is + # the stronger property we want than merely asserting ``done()``. + await asyncio.wait_for(asyncio.shield(producer), timeout=2) reader.cancel() with contextlib.suppress(asyncio.CancelledError): From b79124de12f7f0d17e13c7ee213c56cfd49dc58a Mon Sep 17 00:00:00 2001 From: Carsten Piepel Date: Mon, 12 Oct 2026 10:21:12 -0400 Subject: [PATCH 4/7] test(server): describe subscriber sync as a scheduling handoff only Instrumentation showed _RequestStarted is not dequeued before producer.cancel(); producer teardown drains it through the tap instead. Correct the comments so they no longer claim the handoff consumes the event before cancellation. Comment-only change. Signed-off-by: Carsten Piepel --- .../agent_execution/test_active_task.py | 21 ++++++++----------- 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py index 7b0f035e4..17af6514d 100644 --- a/tests/server/agent_execution/test_active_task.py +++ b/tests/server/agent_execution/test_active_task.py @@ -964,20 +964,18 @@ async def execute_mock(req, q): producer = active_task._producer_task assert producer is not None - # Drive a subscriber far enough to dequeue and skip _RequestStarted, - # then stop reading -- exactly the state a cancelled stream leaves. + # Attach a subscriber while the producer is running. The subscriber + # does not need to consume _RequestStarted before the cancel: the + # producer's teardown, `_event_queue_subscribers.close(immediate=False)`, + # drains the queued event through this tap. Without the fix that + # dequeued item is never marked done, so the close waits forever. subscription = active_task.subscribe(request=request_context) consuming = asyncio.Event() async def drain() -> None: - # Signalled just before the first await on the subscription. This - # does NOT by itself prove _RequestStarted has been consumed -- it - # marks that the drain coroutine is scheduled and about to pull -- - # but combined with the sleep(0) below it reliably hands control to - # the reader so it dequeues and skips the started event. It is still - # tighter than a fixed sleep, and the final assertion (the producer - # must complete) is what actually proves the leak is gone. + # Scheduling handoff only: this shows the reader task has started. + # It does not show that any event has been dequeued. consuming.set() async for _event in subscription: # The internal _RequestStarted the producer enqueues is @@ -986,9 +984,8 @@ async def drain() -> None: pass reader = asyncio.create_task(drain()) - # Wait until the reader is about to consume, then yield the event loop - # so it runs far enough to dequeue and skip the producer's - # _RequestStarted before we cancel. + # Let the reader start, then yield once. This is a scheduling handoff, + # not a guarantee that _RequestStarted has been consumed yet. await asyncio.wait_for(consuming.wait(), timeout=2) await asyncio.sleep(0) From 2fd5b67e675fdb763811ea914057c403548ab625 Mon Sep 17 00:00:00 2001 From: Carsten Piepel Date: Mon, 12 Oct 2026 10:24:32 -0400 Subject: [PATCH 5/7] test(server): wait for execute() instead of a subscriber handoff Producer teardown drains _RequestStarted through the subscriber tap, so the reader-started event and sleep(0) were not needed. Remove them. The test still needs one synchronization point: cancelling the producer before it runs skips _run_producer's CancelledError handler, so the await raises CancelledError and the cancel path under test is never exercised. Wait on an event set from the mocked execute() instead. Verified 10/10 passes with the fix, 10/10 TimeoutError failures without it, and an injected teardown RuntimeError still propagates. Signed-off-by: Carsten Piepel --- tests/server/agent_execution/test_active_task.py | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py index 17af6514d..2431958d3 100644 --- a/tests/server/agent_execution/test_active_task.py +++ b/tests/server/agent_execution/test_active_task.py @@ -945,8 +945,10 @@ async def test_active_task_cancel_while_subscribed_frees_producer( ``ActiveTask`` leak until the whole handler is ``aclose``-d. """ hang = asyncio.Event() + executing = asyncio.Event() async def execute_mock(req, q): + executing.set() await hang.wait() # long-running work, cancelled from outside agent_executor.execute = AsyncMock(side_effect=execute_mock) @@ -971,12 +973,7 @@ async def execute_mock(req, q): # dequeued item is never marked done, so the close waits forever. subscription = active_task.subscribe(request=request_context) - consuming = asyncio.Event() - async def drain() -> None: - # Scheduling handoff only: this shows the reader task has started. - # It does not show that any event has been dequeued. - consuming.set() async for _event in subscription: # The internal _RequestStarted the producer enqueues is # consumed inside subscribe() and never reaches us; real work @@ -984,10 +981,11 @@ async def drain() -> None: pass reader = asyncio.create_task(drain()) - # Let the reader start, then yield once. This is a scheduling handoff, - # not a guarantee that _RequestStarted has been consumed yet. - await asyncio.wait_for(consuming.wait(), timeout=2) - await asyncio.sleep(0) + # Wait until the producer is inside execute(), so it has already + # enqueued _RequestStarted. Cancelling a task before it starts running + # would skip _run_producer's CancelledError handler entirely, so the + # test would not exercise the real streamed-cancel path. + await asyncio.wait_for(executing.wait(), timeout=2) # Cancel the producer: this is what the SDK does after the executor's # cancel() returns on a streamed task. execute() never gets to emit From 103ea7546efe0b1d1e6b72ff42f0b02c0298057d Mon Sep 17 00:00:00 2001 From: Carsten Piepel Date: Tue, 13 Oct 2026 11:50:42 -0400 Subject: [PATCH 6/7] fix(server): ack every dequeued subscriber item, including raised errors Wrap each dequeued item in one try/finally so task_done() runs exactly once on every path (raise, lifecycle skip, return, yield). A re-raised error was never acked, which hangs a graceful close already joining the tap. Add a regression test and trim the #1322 test to essentials. --- src/a2a/server/agent_execution/active_task.py | 8 ++----- .../agent_execution/test_active_task.py | 24 +++++++++++++++++++ 2 files changed, 26 insertions(+), 6 deletions(-) diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index cb46be1f9..d1b0801fe 100644 --- a/src/a2a/server/agent_execution/active_task.py +++ b/src/a2a/server/agent_execution/active_task.py @@ -761,6 +761,7 @@ async def subscribe( self._task_id, event, ) + tapped_queue.task_done() raise event if replace_status_update_with_task and isinstance( @@ -772,12 +773,7 @@ async def subscribe( updated_task, ) event = updated_task - # Internal lifecycle events are consumed here rather than - # yielded, so mark them done before skipping. Otherwise the - # producer's `_event_queue_subscribers.close(immediate=False)` - # waits forever on an item this tap never acknowledges, which - # leaks the producer and the ActiveTask after a cancel ends - # the subscriber before _RequestCompleted is sent (#1322). + # Acknowledge internal lifecycle events before skipping them. if isinstance(event, _RequestStarted | _RequestCompleted): tapped_queue.task_done() if isinstance(event, _RequestCompleted): diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py index 2431958d3..d2e4aca51 100644 --- a/tests/server/agent_execution/test_active_task.py +++ b/tests/server/agent_execution/test_active_task.py @@ -1008,6 +1008,30 @@ async def drain() -> None: hang.set() await active_task.aclose() + @pytest.mark.asyncio + async def test_active_task_subscribe_acks_raised_exception( + self, + active_task: ActiveTask, + ) -> None: + """A re-raised queue exception is still acknowledged.""" + tapped_queue = Mock(spec=EventQueue) + tapped_queue.dequeue_event = AsyncMock( + return_value=(RuntimeError('boom'), None) + ) + tapped_queue.close = AsyncMock() + + with ( + patch.object( + active_task._event_queue_subscribers, + 'tap', + return_value=tapped_queue, + ), + pytest.raises(RuntimeError, match='boom'), + ): + await anext(active_task.subscribe()) + + tapped_queue.task_done.assert_called_once_with() + @pytest.mark.asyncio async def test_active_task_subscribe_include_initial_task(): From 0326b4cf921796e15093421c55caf78580708a3e Mon Sep 17 00:00:00 2001 From: Carsten Piepel Date: Wed, 14 Oct 2026 09:18:55 -0400 Subject: [PATCH 7/7] test(server): shorten cancellation regression comments --- .../agent_execution/test_active_task.py | 40 ++----------------- 1 file changed, 4 insertions(+), 36 deletions(-) diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py index d2e4aca51..7b0a2a012 100644 --- a/tests/server/agent_execution/test_active_task.py +++ b/tests/server/agent_execution/test_active_task.py @@ -932,18 +932,7 @@ async def test_active_task_cancel_while_subscribed_frees_producer( request_context: Mock, task_manager: Mock, ) -> None: - """Regression (#1322): a cancel while a subscriber is streaming must - not leave the producer wedged. - - The producer closes `_event_queue_subscribers` with - `immediate=False`, which blocks until every tap has marked all of its - items done. ``subscribe`` consumes the internal ``_RequestStarted`` - event without yielding it, so it must also mark it done -- otherwise a - cancel that ends the subscriber before ``_RequestCompleted`` is sent - leaves that item unacknowledged, the producer blocks forever on - ``close(immediate=False)``, and both the producer task and the - ``ActiveTask`` leak until the whole handler is ``aclose``-d. - """ + """Regression (#1322): cancelling a stream frees its producer.""" hang = asyncio.Event() executing = asyncio.Event() @@ -966,40 +955,19 @@ async def execute_mock(req, q): producer = active_task._producer_task assert producer is not None - # Attach a subscriber while the producer is running. The subscriber - # does not need to consume _RequestStarted before the cancel: the - # producer's teardown, `_event_queue_subscribers.close(immediate=False)`, - # drains the queued event through this tap. Without the fix that - # dequeued item is never marked done, so the close waits forever. + # Subscribe before cancellation so the lifecycle event requires an ack. subscription = active_task.subscribe(request=request_context) async def drain() -> None: async for _event in subscription: - # The internal _RequestStarted the producer enqueues is - # consumed inside subscribe() and never reaches us; real work - # items would. pass reader = asyncio.create_task(drain()) - # Wait until the producer is inside execute(), so it has already - # enqueued _RequestStarted. Cancelling a task before it starts running - # would skip _run_producer's CancelledError handler entirely, so the - # test would not exercise the real streamed-cancel path. + # Ensure cancellation reaches _run_producer's cleanup path. await asyncio.wait_for(executing.wait(), timeout=2) - - # Cancel the producer: this is what the SDK does after the executor's - # cancel() returns on a streamed task. execute() never gets to emit - # _RequestCompleted. producer.cancel() - # The producer must reach completion CLEANLY. _run_producer catches - # asyncio.CancelledError internally and returns normally, so awaiting - # it here does NOT re-raise -- it returns the producer's result once - # teardown finishes. Before the fix, teardown blocks forever inside - # `_event_queue_subscribers.close(immediate=False)` waiting on the - # unacknowledged _RequestStarted, and this await times out. If teardown - # instead raises, the await re-raises it and the test fails -- which is - # the stronger property we want than merely asserting ``done()``. + # Graceful cleanup must not wait on an unacknowledged lifecycle event. await asyncio.wait_for(asyncio.shield(producer), timeout=2) reader.cancel()