Let a connector's teardown cancellation through too

ConnectorNode.stop cancelled its poll task and then caught CancelledError
around the await — the fourth site of the trap f002e21 closed elsewhere,
swallowing a cancellation aimed at whoever asked for the teardown. It now
calls the shared Node._cancel_task, which keeps retrieving whatever the
loop raised on its way out, as the old `except (CancelledError, Exception)`
did.
This commit is contained in:
2026-08-28 11:56:19 +02:00
parent f4fd26e44d
commit 7d585bbb38
2 changed files with 27 additions and 12 deletions
+1 -5
View File
@@ -161,11 +161,7 @@ class ConnectorNode(Node):
return
self._stop_event.set()
if self._poll_task is not None:
self._poll_task.cancel()
try:
await self._poll_task
except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down
pass
await self._cancel_task(self._poll_task)
self._poll_task = None
self._stop_event = None
self._last_published = {}
+26 -7
View File
@@ -4,17 +4,19 @@ import asyncio
import pytest
from fluksio.flow.connector import ConnectorNode
from fluksio.flow.nodes import DelayNode
def test_stop_cron_lets_the_callers_cancellation_through():
async def stubborn() -> None:
"""A loop whose shutdown does not answer the first cancellation."""
try:
await asyncio.sleep(3600)
except asyncio.CancelledError:
await asyncio.sleep(3600)
async def stubborn() -> None:
"""A loop whose shutdown does not answer the first cancellation."""
try:
await asyncio.sleep(3600)
except asyncio.CancelledError:
await asyncio.sleep(3600)
def test_stop_cron_lets_the_callers_cancellation_through():
async def scenario() -> None:
node = DelayNode(params={"cron": "* * * * *"})
node._stop_cron = asyncio.Event()
@@ -29,3 +31,20 @@ def test_stop_cron_lets_the_callers_cancellation_through():
node._cron_task.cancel()
asyncio.run(scenario())
def test_connector_stop_lets_the_callers_cancellation_through():
async def scenario() -> None:
node = ConnectorNode()
node._stop_event = asyncio.Event()
node._poll_task = asyncio.create_task(stubborn())
stopping = asyncio.create_task(node.stop())
await asyncio.sleep(0.05) # let it reach the await on the poll task
stopping.cancel()
with pytest.raises(asyncio.CancelledError):
await stopping
node._poll_task.cancel()
asyncio.run(scenario())