Shared node changes rebuild the flows whose published copy uses them, not their drafts

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

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 += [

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)