From 93e45274d03598573e45915c63b0afb5ca8c7c1b Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 28 Aug 2026 11:27:01 +0200 Subject: [PATCH] Let a teardown's cancellation through, and reap the workers it leaves MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Node stop paths cancelled their background task and then caught CancelledError around the await, which swallows a cancellation aimed at the caller — the trap Supervisor._cancel already documents. One shared Node._cancel_task now waits the way the supervisor does; mqtt's publisher and subscription and delay's cron call it. The api container also collected zombie python workers: orphaned when --reload replaces the process holding their handle, they reparent onto a PID 1 that reaps nothing but its own. `init: true` on the backend service. --- backend/fluksio/flow/nodes/base.py | 16 ++++++++++++ backend/fluksio/flow/nodes/delay.py | 6 +---- backend/fluksio/flow/nodes/mqtt.py | 12 ++------- backend/tests/flow/test_node_teardown.py | 31 ++++++++++++++++++++++++ docker/compose.yml | 7 ++++++ 5 files changed, 57 insertions(+), 15 deletions(-) create mode 100644 backend/tests/flow/test_node_teardown.py diff --git a/backend/fluksio/flow/nodes/base.py b/backend/fluksio/flow/nodes/base.py index e959070..b46a57d 100644 --- a/backend/fluksio/flow/nodes/base.py +++ b/backend/fluksio/flow/nodes/base.py @@ -212,6 +212,22 @@ class Node: return None return asyncio.create_task(factory()) + @staticmethod + async def _cancel_task(task: asyncio.Task[None]) -> None: + """Stop an unsupervised loop and wait for it to be gone. + + `wait` keeps whatever the task raises on its way out to itself, and + lets a cancellation aimed at *this* coroutine through — the + `except CancelledError` around `await task` it replaces swallowed that, + which left whoever asked for the teardown unkillable. The same trap + `Supervisor._cancel` documents. + """ + task.cancel() + await asyncio.wait([task]) + if not task.cancelled(): + # Retrieved so a crash on the way out is not reported at exit. + task.exception() + def report_health(self, status: str, detail: str | None = None) -> None: """Say how this node's connection is doing: ok, degraded or down.""" if self._on_health is not None: diff --git a/backend/fluksio/flow/nodes/delay.py b/backend/fluksio/flow/nodes/delay.py index ac3698a..0cc1cf3 100644 --- a/backend/fluksio/flow/nodes/delay.py +++ b/backend/fluksio/flow/nodes/delay.py @@ -211,11 +211,7 @@ class DelayNode(Node): self._stop_cron.set() if self._cron_task is not None: - self._cron_task.cancel() - try: - await self._cron_task - except asyncio.CancelledError: - pass + await self._cancel_task(self._cron_task) self._cron_task = None self._stop_cron = None diff --git a/backend/fluksio/flow/nodes/mqtt.py b/backend/fluksio/flow/nodes/mqtt.py index f7d9ef2..f936ca8 100644 --- a/backend/fluksio/flow/nodes/mqtt.py +++ b/backend/fluksio/flow/nodes/mqtt.py @@ -461,11 +461,7 @@ class MqttNode(Node): if self._publish_queue is None: return if self._publisher_task is not None: - self._publisher_task.cancel() - try: - await self._publisher_task - except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down - pass + await self._cancel_task(self._publisher_task) self._publisher_task = None self._publish_queue = None self._loop = None @@ -515,11 +511,7 @@ class MqttNode(Node): self._stop_event.set() if self._subscription_task is not None: - self._subscription_task.cancel() - try: - await self._subscription_task - except asyncio.CancelledError: - pass + await self._cancel_task(self._subscription_task) self._subscription_task = None self._stop_event = None diff --git a/backend/tests/flow/test_node_teardown.py b/backend/tests/flow/test_node_teardown.py new file mode 100644 index 0000000..440ce9d --- /dev/null +++ b/backend/tests/flow/test_node_teardown.py @@ -0,0 +1,31 @@ +"""Tearing a node down must not swallow a cancellation meant for the caller.""" + +import asyncio + +import pytest + +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 scenario() -> None: + node = DelayNode(params={"cron": "* * * * *"}) + node._stop_cron = asyncio.Event() + node._cron_task = asyncio.create_task(stubborn()) + + stopping = asyncio.create_task(node.stop_cron()) + await asyncio.sleep(0.05) # let it reach the await on the cron task + stopping.cancel() + with pytest.raises(asyncio.CancelledError): + await stopping + + node._cron_task.cancel() + + asyncio.run(scenario()) diff --git a/docker/compose.yml b/docker/compose.yml index e554fbb..75f5e03 100644 --- a/docker/compose.yml +++ b/docker/compose.yml @@ -75,6 +75,13 @@ services: image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}' container_name: fluksio-api restart: always + # PID 1 that reaps orphans. A python node's worker processes outlive the + # process holding their handle whenever that one goes without stopping the + # pool -- which `--reload` does on every source edit -- and reparent onto + # PID 1, which is the app itself and waits for nobody else's children. The + # container filled up with zombie `python`. Declared here rather than in + # compose.dev.yml because an init closes the whole class, not just reload. + init: true security_opt: - no-new-privileges:true networks: