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.