From e22b4795fd14799f8ea2e5c018b63ee393bed02c Mon Sep 17 00:00:00 2001 From: stroblme Date: Sun, 23 Aug 2026 17:44:52 +0200 Subject: [PATCH] Let teardown and activate be asked for one flow Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu --- backend/fluksio/flow/controller.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index dead426..980d7e7 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -510,12 +510,14 @@ class FlowController: } ) - async def _teardown(self) -> None: - """Stop everything the previous pipeline started.""" + async def _teardown(self, flow: str | None = None) -> None: + """Stop everything the previous pipeline started, or one flow's share.""" for entry in self.loaded.values(): node = entry.node if node is None: continue + if flow is not None and entry.flow != flow: + continue try: await asyncio.wait_for(node.stop(self.app), NODE_STOP_TIMEOUT) except asyncio.TimeoutError: @@ -529,9 +531,12 @@ class FlowController: except Exception: logger.exception("Error stopping node '%s'", entry.id) # After the nodes, so a loop still winding down is not restarted. - await self.supervisor.cancel_all() + if flow is None: + await self.supervisor.cancel_all() + else: + await self.supervisor.cancel_flow(flow) - async def _activate(self) -> None: + async def _activate(self, flow: str | None = None) -> None: """Start subscriptions, schedules and webhooks of the new pipeline. What that means per node type is the node's own business — a connector @@ -541,6 +546,8 @@ class FlowController: node = entry.node if node is None: continue + if flow is not None and entry.flow != flow: + continue # A stopped flow gets no subscriptions, schedules or webhooks — # that is what stopping it means. Nor does a batch flow, which has # no outside to listen to.