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.