Retire a shared node in use with ?detach=true
DELETE /flows/library/{name} still answers 409 while nodes use the source.
With detach=true each of them keeps a copy of the code, in the published
flow and the draft, the source is deleted and the running users rebuild.
This commit is contained in:
@@ -336,8 +336,23 @@ def read_library(controller: FlowControllerDep) -> Any:
|
||||
|
||||
|
||||
@router.delete("/library/{lib_name}", response_model=Message)
|
||||
async def delete_shared_node(lib_name: str, controller: FlowControllerDep) -> Any:
|
||||
"""Remove a shared source, as long as no flow still runs it."""
|
||||
async def delete_shared_node(
|
||||
lib_name: str, controller: FlowControllerDep, detach: bool = False
|
||||
) -> Any:
|
||||
"""Remove a shared source, as long as no flow still runs it.
|
||||
|
||||
With ``detach``, every node using it keeps a copy of the code as its own
|
||||
first, so the source can be retired while flows still use it.
|
||||
"""
|
||||
if detach:
|
||||
running = await run_in_threadpool(controller.store.usages, lib_name, False)
|
||||
try:
|
||||
await run_in_threadpool(controller.store.detach_lib, lib_name)
|
||||
except LibNotFound:
|
||||
raise HTTPException(status_code=404, detail=f"No shared node '{lib_name}'")
|
||||
for flow in sorted({usage.split(".", 1)[0] for usage in running}):
|
||||
await controller.reload_flow(flow)
|
||||
return Message(message=f"Deleted shared node '{lib_name}'")
|
||||
used_by = controller.store.usages(lib_name)
|
||||
if used_by:
|
||||
raise HTTPException(
|
||||
|
||||
@@ -400,6 +400,37 @@ class FlowStore:
|
||||
self._set_source_ref(flow, node_id, None)
|
||||
self._commit(f"Unshare node '{flow}.{node_id}'")
|
||||
|
||||
def detach_lib(self, name: str) -> None:
|
||||
"""Give every node running this shared source a copy, then delete it.
|
||||
|
||||
Published flows and drafts alike, so a running flow keeps the same code
|
||||
and nothing waits on a publish. Each document changed goes up a version,
|
||||
so an editor holding the old one gets a conflict on its next save
|
||||
instead of pointing a node back at a source that is gone.
|
||||
"""
|
||||
with self._write_lock:
|
||||
code = self.read_lib_source(name)
|
||||
for flow in self.list_flows():
|
||||
for path, node_file in (
|
||||
(self._flow_file(flow), self._node_file),
|
||||
(self._draft_file(flow), self._draft_node_file),
|
||||
):
|
||||
if not path.exists():
|
||||
continue
|
||||
definition = FlowDef.model_validate_json(path.read_text())
|
||||
users = [n for n in definition.nodes if n.source_ref == name]
|
||||
if not users:
|
||||
continue
|
||||
for node in users:
|
||||
target = node_file(flow, node.id)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_text(code)
|
||||
node.source_ref = None
|
||||
definition.version += 1
|
||||
path.write_text(definition.model_dump_json(indent=2) + "\n")
|
||||
self._lib_file(name).unlink()
|
||||
self._commit(f"Detach and delete shared node '{name}'")
|
||||
|
||||
def usages(self, lib_name: str, draft: bool = True) -> list[str]:
|
||||
"""Every ``flow.node`` pointing at this shared source.
|
||||
|
||||
|
||||
@@ -550,3 +550,34 @@ async def test_the_live_socket_ends_when_its_client_goes(
|
||||
await asyncio.sleep(0.05)
|
||||
parked = [t for t in asyncio.all_tasks() if "Queue.get" in repr(t.get_coro())]
|
||||
assert parked == []
|
||||
|
||||
|
||||
def test_a_shared_node_in_use_is_retired_only_with_detach(
|
||||
client: TestClient, superuser_token_headers: dict[str, str]
|
||||
) -> None:
|
||||
client.put(
|
||||
f"{PREFIX}/shared", headers=superuser_token_headers, json=a_flow("shared")
|
||||
)
|
||||
client.put(
|
||||
f"{PREFIX}/shared/nodes/sensor/source",
|
||||
headers=superuser_token_headers,
|
||||
json={"code": WORKING_NODE},
|
||||
)
|
||||
client.post(
|
||||
f"{PREFIX}/shared/nodes/sensor/share",
|
||||
headers=superuser_token_headers,
|
||||
json={"lib_name": "reading"},
|
||||
)
|
||||
|
||||
url = f"{PREFIX}/library/reading"
|
||||
assert client.delete(url, headers=superuser_token_headers).status_code == 409
|
||||
response = client.delete(f"{url}?detach=true", headers=superuser_token_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
assert client.get(f"{PREFIX}/library", headers=superuser_token_headers).json() == []
|
||||
source = client.get(
|
||||
f"{PREFIX}/shared/nodes/sensor/source", headers=superuser_token_headers
|
||||
)
|
||||
assert source.json()["code"] == WORKING_NODE
|
||||
|
||||
client.delete(f"{PREFIX}/shared", headers=superuser_token_headers)
|
||||
|
||||
@@ -120,3 +120,27 @@ def test_unsharing_takes_a_private_copy_back(store: FlowStore):
|
||||
def test_a_missing_shared_source_is_reported(store: FlowStore):
|
||||
with pytest.raises(LibNotFound):
|
||||
store.read_lib_source("nothing_here")
|
||||
|
||||
|
||||
def test_retiring_a_shared_source_leaves_each_user_a_copy(store: FlowStore):
|
||||
store.write_flow(a_flow())
|
||||
store.write_node_source("heating", "sensor", SOURCE)
|
||||
store.share_node("heating", "sensor", "read_temp")
|
||||
store.publish_flow("heating", store.read_flow("heating", draft=True).version)
|
||||
version = store.read_flow("heating").version
|
||||
# Shared in a draft that was never published.
|
||||
cooling = a_flow("cooling")
|
||||
cooling.nodes[0].source_ref = "read_temp"
|
||||
store.write_draft(cooling, 0)
|
||||
|
||||
store.detach_lib("read_temp")
|
||||
|
||||
assert store.list_lib() == []
|
||||
heating = store.read_flow("heating")
|
||||
assert heating.nodes[0].source_ref is None
|
||||
assert heating.version == version + 1
|
||||
assert store.read_node_source("heating", "sensor") == SOURCE
|
||||
# What runs is the same code, so there is nothing to publish.
|
||||
assert not store.has_draft("heating")
|
||||
assert store.read_flow("cooling", draft=True).nodes[0].source_ref is None
|
||||
assert store.read_node_source("cooling", "sensor", draft=True) == SOURCE
|
||||
|
||||
@@ -83,6 +83,9 @@ Ordinary source edits stay in the draft until publishing. Shared library
|
||||
source changes affect its users immediately.
|
||||
`POST /flows/{name}/nodes/{id}/share` promotes a source to the library;
|
||||
`/unshare` makes it private again. `GET /flows/library` lists shared sources.
|
||||
`DELETE /flows/library/{name}` retires one, and answers 409 while nodes still
|
||||
use it. With `?detach=true` each of those nodes keeps a copy of the code, in
|
||||
the published flow and the draft, and the shared source is deleted.
|
||||
|
||||
### Running and controlling
|
||||
|
||||
|
||||
@@ -819,8 +819,12 @@ export class FlowsService {
|
||||
/**
|
||||
* Delete Shared Node
|
||||
* Remove a shared source, as long as no flow still runs it.
|
||||
*
|
||||
* With ``detach``, every node using it keeps a copy of the code as its own
|
||||
* first, so the source can be retired while flows still use it.
|
||||
* @param data The data for the request.
|
||||
* @param data.libName
|
||||
* @param data.detach
|
||||
* @returns Message Successful Response
|
||||
* @throws ApiError
|
||||
*/
|
||||
@@ -831,6 +835,9 @@ export class FlowsService {
|
||||
path: {
|
||||
lib_name: data.libName
|
||||
},
|
||||
query: {
|
||||
detach: data.detach
|
||||
},
|
||||
errors: {
|
||||
422: 'Validation Error'
|
||||
}
|
||||
|
||||
@@ -1716,6 +1716,7 @@ export type FlowsReadGraphResponse = (BrainGraph);
|
||||
export type FlowsReadLibraryResponse = (Array<LibraryNode>);
|
||||
|
||||
export type FlowsDeleteSharedNodeData = {
|
||||
detach?: boolean;
|
||||
libName: string;
|
||||
};
|
||||
|
||||
|
||||
Reference in New Issue
Block a user