Bound the pipeline teardown so a stuck node cannot wedge the controller

A node's stop() and a supervised task's cancellation are both waited on
inside the rebuild lock, and neither had a deadline: an MQTT client whose
broker never acknowledges the disconnect leaves aiomqtt's __aexit__
waiting forever, so reload() never returned and every start, stop or
publish behind it hung until the container was restarted.

Each node now gets five seconds to close and is abandoned after that, and
cancel_all reports what is still running rather than waiting on it — it
also no longer swallows a cancellation aimed at the caller, which used to
make the lock holder unkillable. A rebuild asked for by a request gives up
on the lock after fifteen seconds with RebuildBusy, answered as a 503.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu

backend/fluksio/flow/controller.py

@@ -71,6 +71,27 @@ HOOK_PREFIX = "/hooks"
 # neither the brain graph nor the health summary treats them as a fault.
 ADVISORY_ISSUES = frozenset({"unauthenticated_hook"})
 
+# How long a node gets to close what it opened before the rebuild moves on.
+# A node's `stop` talks to whatever it connected to, and a broker that has gone
+# away can leave it waiting for an acknowledgement that never arrives — which
+# used to hold the rebuild, and everything queued behind it, forever.
+NODE_STOP_TIMEOUT = 5.0
+
+# How long a rebuild asked for by a request waits for one already running.
+# Generous on purpose: a rebuild of a populated installation reconnects every
+# node and takes the better part of ten seconds, and a caller queued behind a
+# healthy one of those should not be turned away. Past that the controller is
+# wedged rather than busy, and an error the caller can act on beats a request
+# that never ends.
+REBUILD_WAIT = 15.0
+
+
+class RebuildBusy(RuntimeError):
+    """A rebuild could not start because the one before it has not finished.
+
+    Answered as a 503: nothing is wrong with the request, the engine is busy.
+    """
+
 
 class NodeStatus(str, Enum):
     ACTIVE = "active"
@@ -343,7 +364,9 @@ class FlowController:
         # Build first: the consumer must have a pipeline to execute against
         # before it claims anything, or work waiting from the last run would be
         # taken and dropped — which is the very case the queue exists for.
-        await self.reload()
+        # Nothing is running to queue behind here, and a lifespan has nobody to
+        # report a timeout to, so this one build waits however long it needs.
+        await self.reload(wait=None)
         if self.execution is not None:
             self.execution.start()
 
@@ -397,9 +420,23 @@ class FlowController:
         await run_in_threadpool(self.store.write_enabled, flow, enabled)
         await self.reload()
 
-    async def reload(self) -> None:
-        """Rebuild the whole pipeline from what is currently stored."""
-        async with self._lock:
+    async def reload(self, wait: float | None = REBUILD_WAIT) -> None:
+        """Rebuild the whole pipeline from what is currently stored.
+
+        Only one rebuild runs at a time. A caller waits *wait* seconds for the
+        one in front of it and then gives up with ``RebuildBusy`` — hanging on
+        a rebuild that is stuck is worse than saying so. ``None`` waits.
+        """
+        if wait is None:
+            await self._lock.acquire()
+        else:
+            try:
+                await asyncio.wait_for(self._lock.acquire(), wait)
+            except asyncio.TimeoutError:
+                raise RebuildBusy(
+                    f"A pipeline rebuild is still running after {wait:.0f}s"
+                ) from None
+        try:
             # Work already claimed belongs to the pipeline it was claimed
             # against; let it finish there before swapping the graph out.
             if self.execution is not None:
@@ -460,6 +497,8 @@ class FlowController:
                 # what the old pipeline parked. Release it here or it is lost.
                 for flow in published:
                     self._release_parked(flow.name)
+        finally:
+            self._lock.release()
 
         self._publish(
             {
@@ -478,7 +517,15 @@ class FlowController:
             if node is None:
                 continue
             try:
-                await node.stop(self.app)
+                await asyncio.wait_for(node.stop(self.app), NODE_STOP_TIMEOUT)
+            except asyncio.TimeoutError:
+                # Abandoned rather than waited on: the next node still gets to
+                # close, and the rebuild still happens.
+                logger.warning(
+                    "Node '%s' did not stop within %.0fs — carrying on without it",
+                    entry.id,
+                    NODE_STOP_TIMEOUT,
+                )
             except Exception:
                 logger.exception("Error stopping node '%s'", entry.id)
         # After the nodes, so a loop still winding down is not restarted.

backend/fluksio/flow/supervision.py

@@ -29,6 +29,11 @@ BACKOFF = (1.0, 5.0, 30.0, 60.0)
 # recover by being restarted again.
 FAILURE_BUDGET = 5
 FAILURE_WINDOW = 300.0
+# How long a cancelled task gets to notice. A loop that is still waiting after
+# this is not going to stop on its own — a client closing a socket the broker
+# no longer answers on is the case seen in the wild — and the rebuild asking
+# for it must not wait on that forever.
+CANCEL_GRACE = 5.0
 
 TaskFactory = Callable[[], Coroutine[Any, Any, None]]
 
@@ -113,11 +118,28 @@ class Supervisor:
         self._tasks.clear()
         for task in tasks:
             task.cancel()
-        for task in tasks:
-            try:
-                await task
-            except (asyncio.CancelledError, Exception):  # noqa: B014 - shutting down
-                pass
+        if not tasks:
+            return
+        # `wait` hands back what is still going instead of waiting on it, and
+        # lets a cancellation aimed at *this* coroutine through — the
+        # `except CancelledError` it replaces swallowed that, which left
+        # whoever asked for the teardown holding their lock and unkillable.
+        done, pending = await asyncio.wait(tasks, timeout=CANCEL_GRACE)
+        for task in done:
+            if not task.cancelled():
+                # Retrieved so a crash on the way out is not reported at exit;
+                # the supervisor has already said what it was.
+                task.exception()
+        for task in pending:
+            # Cancelled once and still running means its shutdown is waiting on
+            # something that is not answering. A second cancellation interrupts
+            # that wait; whether it takes is no longer the rebuild's problem.
+            task.cancel()
+            logger.warning(
+                "Supervised task '%s' did not stop within %.0fs — abandoned",
+                task.get_name(),
+                CANCEL_GRACE,
+            )
 
     def _publish(self, event: dict[str, Any]) -> None:
         if self._events is not None:

backend/fluksio/main.py

@@ -4,7 +4,7 @@ from collections.abc import AsyncIterator
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 
 import sentry_sdk
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import JSONResponse
 from fastapi.routing import APIRoute
@@ -21,7 +21,7 @@ from fluksio.core.db import prepare
 from fluksio.flow import logs, modules
 from fluksio.flow.alerts import AlertManager
 from fluksio.flow.artifacts import ArtifactStore
-from fluksio.flow.controller import FlowController
+from fluksio.flow.controller import FlowController, RebuildBusy
 from fluksio.flow.dashboards import DashboardStore
 from fluksio.flow.events import event_bus
 from fluksio.flow.executor import ExecutionService
@@ -234,6 +234,16 @@ if settings.all_cors_origins:
 app.include_router(api_router, prefix=settings.API_V1_STR)
 
 
+@app.exception_handler(RebuildBusy)
+async def rebuild_busy(request: Request, exc: RebuildBusy) -> JSONResponse:  # noqa: ARG001
+    """Every route that deploys something answers a wedged rebuild the same way.
+
+    The request was fine and retrying it may well work, so this is the engine
+    saying it is busy rather than the request having gone wrong.
+    """
+    return JSONResponse(status_code=503, content={"detail": str(exc)})
+
+
 # Tagged because the operation-id builder reads the first tag; the route
 # itself stays out of the schema.
 @app.get(

backend/tests/flow/test_runtime_control.py

@@ -1,13 +1,16 @@
 """Stopping a flow takes it off the engine; pausing holds its nodes."""
 
+import asyncio
 from pathlib import Path
 from typing import Any
 
+import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
 from fluksio.api.deps import get_current_user
 from fluksio.api.routes.flows import router
+from fluksio.flow.controller import FlowController, LoadedNode
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.nodes import Node
 from fluksio.flow.pipeline import Pipeline
@@ -165,3 +168,33 @@ def test_stepping_an_unknown_flow_is_a_404(tmp_path: Path):
 
     assert response.status_code == 404
     assert controller.calls == []
+
+
+def test_a_node_that_will_not_stop_does_not_wedge_the_rebuild(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    """A node closing a connection nobody answers used to hold the lock forever.
+
+    Everything that deploys rebuilds, so the next publish, start or stop then
+    waited on a lock that was never given back.
+    """
+    monkeypatch.setattr("fluksio.flow.controller.NODE_STOP_TIMEOUT", 0.05)
+
+    class NeverStops(Node):
+        async def stop(self, app: FastAPI | None = None) -> None:
+            await asyncio.Event().wait()
+
+    node = NeverStops(f=lambda params: None, name="stuck")
+    node.assign_flow("heating", "stuck")
+    engine = FlowController(FlowStore(tmp_path / "flows"))
+    engine.loaded = {
+        "heating.stuck": LoadedNode(id="heating.stuck", flow="heating", node=node)
+    }
+
+    async def scenario() -> None:
+        # The first rebuild gives up on the node it cannot stop, and the second
+        # is not left queueing behind a lock the first never released.
+        await engine.reload()
+        await engine.reload()
+
+    asyncio.run(asyncio.wait_for(scenario(), timeout=5))