From 56396732a4e20d742c77717d2a743c6b0f1fd441 Mon Sep 17 00:00:00 2001 From: stroblme Date: Sun, 23 Aug 2026 18:35:28 +0200 Subject: [PATCH] Stop and start a flow without rebuilding anything MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stopped flow's nodes are built like any other flow's — being stopped means having no subscriptions, schedules or webhooks, not being absent — so a toggle only ever needed the lifecycle call and the gate that goes with it. It was doing a whole-pipeline rebuild instead, which on a populated installation is every node in every flow reconnecting. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu --- NOTEPAD.md | 5 ++- backend/fluksio/flow/controller.py | 43 +++++++++++++++++++++- backend/tests/flow/test_runtime_control.py | 35 ++++++++++++++++++ 3 files changed, 80 insertions(+), 3 deletions(-) diff --git a/NOTEPAD.md b/NOTEPAD.md index 0013195..5d6df71 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -13,7 +13,10 @@ Deferring because out of scope is fine, but don't mention deferring than. ### Dashboard UI rework (dedicated session) Rework the dashboard UI to make it more flexible in terms of the look and feel. -I consider following design options a) overall appearance: material look or liquid glass (einUI https://github.com/einui/einui) b) color palette +I consider following design options a) overall appearance: material look or liquid glass (einUI https://github.com/einui/einui) b) color palette (https://plotly.com/python/discrete-color/). +Based on a) and b) it should be possible to derive almost endles combinations of looks. +The color palette should affect charts/gauges and other widgets + ### To be sorted diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 29cc1f1..ab3ba68 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -426,9 +426,48 @@ class FlowController: entry.last_error_ts = None async def set_enabled(self, flow: str, enabled: bool) -> None: - """Stop or start one flow. Rebuilding is what applies it.""" + """Stop or start one flow, without building anything. + + Its nodes are built whether or not it runs — being stopped means + having no subscriptions, schedules or webhooks, not being absent — so + all this has to do is the lifecycle call and the gate that goes with + it. Nothing is compiled and no other flow is touched. + """ await run_in_threadpool(self.store.write_enabled, flow, enabled) - await self.reload() + if self.pipeline is None: + # Nothing built yet, so there are no nodes to start or stop. + await self.reload() + return + + await self._acquire(FLOW_REBUILD_WAIT) + try: + if enabled: + # The gate first: a subscription that fires the instant it + # starts must not be turned away by a flag on its way out. + self.disabled = self.disabled - {flow} + self.pipeline.set_disabled(self.disabled) + # Lifts any quarantine, which is what a rebuild used to do by + # throwing the whole supervisor away. + await self.supervisor.cancel_flow(flow) + await self._activate(flow) + else: + self.disabled = self.disabled | {flow} + self.pipeline.set_disabled(self.disabled) + await self._teardown(flow) + finally: + self._lock.release() + + self._publish( + { + "type": "pipeline_rebuilt", + "issues": [issue.model_dump() for issue in self.issues], + "nodes": [status.model_dump() for status in self.node_statuses()], + # Every flow's, not this one's: the canvas replaces its paused + # set from this, so a narrowed list would clear the markers of + # flows this never touched. + "paused": self.paused_flows(), + } + ) async def _acquire(self, wait: float | None) -> None: """Take the rebuild lock, or say the engine is busy. diff --git a/backend/tests/flow/test_runtime_control.py b/backend/tests/flow/test_runtime_control.py index 19177b1..680dde4 100644 --- a/backend/tests/flow/test_runtime_control.py +++ b/backend/tests/flow/test_runtime_control.py @@ -297,3 +297,38 @@ def test_a_deleted_flow_leaves_its_consumers_reporting_a_missing_input( assert engine.pipeline.get_node_by_id("a.meter") is None asyncio.run(asyncio.wait_for(scenario(), timeout=10)) + + +def test_stopping_one_flow_does_not_touch_the_others( + tmp_path: Path, lifecycle: list[str] +): + """A toggle is a lifecycle call, not a rebuild. + + A stopped flow's nodes are still built — being stopped means having no + subscriptions, not being absent — so nothing is compiled and no other + flow's node is asked to reconnect. That is where the seconds went. + """ + store = FlowStore(tmp_path / "flows") + for name in ("a", "b"): + store.write_flow(FlowDef(name=name, nodes=[NodeDef(id="io", type="lifecycle")])) + engine = FlowController(store) + + async def scenario() -> None: + await engine.reload() + before = {name: engine.loaded[name].node for name in ("a.io", "b.io")} + lifecycle.clear() + + await engine.set_enabled("a", False) + + assert lifecycle == ["stop a.io"] + # Still built, still the same objects: only the lifecycle changed. + assert {name: engine.loaded[name].node for name in ("a.io", "b.io")} == before + assert engine.disabled == {"a"} + + lifecycle.clear() + await engine.set_enabled("a", True) + + assert lifecycle == ["start a.io"] + assert {name: engine.loaded[name].node for name in ("a.io", "b.io")} == before + + asyncio.run(asyncio.wait_for(scenario(), timeout=10))