Let a teardown's cancellation through, and reap the workers it leaves

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

@@ -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:

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()
+            await self._cancel_task(self._cron_task)
-            try:
-                await self._cron_task
-            except asyncio.CancelledError:
-                pass
 
         self._cron_task = None
         self._stop_cron = None

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()
+            await self._cancel_task(self._publisher_task)
-            try:
-                await self._publisher_task
-            except (asyncio.CancelledError, Exception):  # noqa: B014 - shutting down
-                pass
         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()
+            await self._cancel_task(self._subscription_task)
-            try:
-                await self._subscription_task
-            except asyncio.CancelledError:
-                pass
 
         self._subscription_task = None
         self._stop_event = None

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())

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: