diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 3e8cdfc..ee434aa 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -837,7 +837,8 @@ class FlowController: async def reload_lib_users(self, ref: str) -> list[str]: """Rebuild the flows running a shared node whose source has changed.""" - flows = sorted({usage.split(".", 1)[0] for usage in self.store.usages(ref)}) + usages = self.store.usages(ref, draft=False) + flows = sorted({usage.split(".", 1)[0] for usage in usages}) for flow in flows: await self.reload_flow(flow) return flows diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index d414c63..e100ede 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -355,12 +355,16 @@ class FlowStore: self._set_source_ref(flow, node_id, None) self._commit(f"Unshare node '{flow}.{node_id}'") - def usages(self, lib_name: str) -> list[str]: - """Every ``flow.node`` pointing at this shared source.""" + def usages(self, lib_name: str, draft: bool = True) -> list[str]: + """Every ``flow.node`` pointing at this shared source. + + Read from the drafts the editor shows, or with ``draft`` off from the + published flows the engine runs. + """ used = [] for name in self.list_flows(): try: - definition = self.read_flow(name, draft=True) + definition = self.read_flow(name, draft=draft) except FlowNotFound: continue used += [ diff --git a/backend/tests/flow/test_library.py b/backend/tests/flow/test_library.py index 03eb208..6a0619d 100644 --- a/backend/tests/flow/test_library.py +++ b/backend/tests/flow/test_library.py @@ -1,9 +1,11 @@ """Nodes shared across flows: one source, many instances.""" +import asyncio from pathlib import Path import pytest +from fluksio.flow.controller import FlowController from fluksio.flow.messages import MessageSpec from fluksio.flow.schemas import FlowDef, NodeDef from fluksio.flow.store import FlowStore, LibExists, LibNotFound @@ -67,6 +69,30 @@ def test_a_second_flow_can_use_the_same_source(store: FlowStore): assert store.usages("read_temp") == ["cooling.sensor", "heating.sensor"] +def test_a_shared_source_change_rebuilds_only_what_is_published(store: FlowStore): + store.write_flow(a_flow()) + store.write_node_source("heating", "sensor", SOURCE) + store.share_node("heating", "sensor", "read_temp") + # Published with the shared node, which the draft has since dropped. + store.publish_flow("heating", store.read_flow("heating", draft=True).version) + store.unshare_node("heating", "sensor") + # Shared in a draft that was never published. + cooling = a_flow("cooling") + cooling.nodes[0].source_ref = "read_temp" + store.write_draft(cooling, 0) + + controller = FlowController(store) + rebuilt: list[str] = [] + + async def reload_flow(name: str) -> None: + rebuilt.append(name) + + controller.reload_flow = reload_flow # type: ignore[method-assign] + asyncio.run(controller.reload_lib_users("read_temp")) + + assert rebuilt == ["heating"] + + def test_a_shared_name_is_not_taken_twice(store: FlowStore): store.write_flow(a_flow()) store.write_node_source("heating", "sensor", SOURCE)