From 423968d9a0d31cb677af3a052b781460d03881cc Mon Sep 17 00:00:00 2001 From: stroblme Date: Sun, 23 Aug 2026 16:36:48 +0200 Subject: [PATCH] Bound the pipeline teardown so a stuck node cannot wedge the controller MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu --- NOTEPAD.md | 29 ++++++----- backend/fluksio/flow/controller.py | 57 ++++++++++++++++++++-- backend/fluksio/flow/supervision.py | 32 ++++++++++-- backend/fluksio/main.py | 14 +++++- backend/tests/flow/test_runtime_control.py | 33 +++++++++++++ 5 files changed, 142 insertions(+), 23 deletions(-) diff --git a/NOTEPAD.md b/NOTEPAD.md index a4f6312..b57dc2c 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -15,7 +15,8 @@ Deferring because out of scope is fine, but don't mention deferring than. - FEAT/UI: we promise testing, but currently don't provide an UI for testing e.g. mock values or probing edge cases of a flow. This should be resolved (in a dedicated session); I'm thinking of a "Labs" page, which allows simulating an installation with all the flows (using their draft states) and which allows injecting values or mocking values based on events in the past - FEAT/UI: check if PWA (https://whatpwacando.today/) notifications could be used to have a panel sending notifications to the device event bus (or generally using PWA to retrieve e.g. location etc). We could introduce a general concept of having a panel (a device, like a wall panel or a phone where the pwa (dashboard) runs) being effectively a node with various outputs. Then various inputs could trigger actions like authentification (i.e. you get home and get a notification which allows you to authenticate the door unlock), get notified on alarms (native alarm connector) or to query geolocation (check where the user is before turning of all lights) etc - BUG/UI sync the theme state between panels and installations - +- BUG/UI some dashboard widgets (like the color picker) are scrollable; we should make sure that no widgets (except for text widgets or list-related widgets) are scrollable +- INFRA document `make update` in the docs; this command is intended to run as a fire-and-forget command when updating a local installation - CHORE/UI: the house panels are laid out for 1280x800 — twelve columns, twelve rows. A chart's fixed chrome is now its title line and legend: the range picker moved up onto the title and gave back its row, so the budget is nearer forty @@ -34,16 +35,22 @@ Deferring because out of scope is fine, but don't mention deferring than. `Segment` now takes an optional `label`; the house dashboard sets one (`Solar`), and it shows on the next frontend build. -- BUG/FLOW: **a cancelled request can leave `FlowController._lock` held forever.** - Seeding nineteen flows over a client that timed out mid-request left the next - `POST /flows/{name}/start` waiting on the lock indefinitely — ten minutes, until - the container was restarted. A py-spy dump showed *no* thread in the reload path, - so the coroutine that holds it is suspended at an `await` inside `reload()`, most - likely in `_teardown()` awaiting a supervised task's cancellation. Everything else - kept working — health, reads, MQTT — so the engine looked fine and only anything - needing a rebuild hung. Two things worth doing: release the lock on cancellation - (`asyncio.timeout` around the teardown, or a `finally` that cannot be skipped), and - fail a `start` that waits more than a few seconds for the lock rather than hanging. +- BUG/NODE: **an MQTT client can hang forever on the way out.** `MqttNode` builds + `aiomqtt.Client` without a `timeout`, so `Client.__aexit__` waits for the broker's + disconnect acknowledgement with no deadline — `_wait_for(..., timeout=None)` falls + through to `self.timeout`, which is `None` too. A subscriber cancelled while its + socket is dead never finishes unwinding. This was what wedged the rebuild lock; + the teardown now abandons such a task after five seconds rather than waiting on + it, so what is left is the task itself, which is stopped only by the second + cancellation it is sent on the way out. Passing a `timeout` to the client fixes it + at the source, but the same number also bounds `subscribe` and `publish`, so it + wants choosing deliberately. +- BUG/FLOW: `MqttNode.stop_publisher`, `MqttNode.stop_subscription` and + `DelayNode.stop_cron` catch `CancelledError` around the task they have just + cancelled, which swallows a cancellation meant for the caller — the trap + `Supervisor.cancel_all` was just fixed for. Latent rather than live: a supervised + node leaves those handles `None`, so only a node built on its own (a test, a + preview) awaits there. - BUG/INFRA: **475 zombie `git` processes** in the API container after a seeding session. `FlowStore._git` uses `subprocess.run`, which reaps its own child — these are the `git gc --auto` daemons `git commit` spawns, reparented to PID 1 when their diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 1eddce6..29192c8 100644 --- a/backend/fluksio/flow/controller.py +++ b/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. diff --git a/backend/fluksio/flow/supervision.py b/backend/fluksio/flow/supervision.py index 6d7f92b..5a1404f 100644 --- a/backend/fluksio/flow/supervision.py +++ b/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: diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 9afa9fa..f257102 100644 --- a/backend/fluksio/main.py +++ b/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( diff --git a/backend/tests/flow/test_runtime_control.py b/backend/tests/flow/test_runtime_control.py index 73c4bce..2ce3738 100644 --- a/backend/tests/flow/test_runtime_control.py +++ b/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))