diff --git a/backend/fluksio/api/routes/artifacts.py b/backend/fluksio/api/routes/artifacts.py index 2667259..cea16df 100644 --- a/backend/fluksio/api/routes/artifacts.py +++ b/backend/fluksio/api/routes/artifacts.py @@ -217,8 +217,8 @@ def delete_artifact( raise HTTPException( status_code=409, detail=( - "The artifact is still referenced by a run, a current message or " - "a flow input's initial value" + "The artifact is still referenced by a run, a current message, " + "a flow input's initial value or a node setting" ), ) if not store.delete(digest): diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 698080e..50fb47f 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -357,11 +357,20 @@ _REF_DEPTH = 8 _SWEEP_BATCH = 500 -def _references_in(value: Any, found: set[str], depth: int = 0) -> None: - """Every artifact digest inside one stored value.""" +def _references_in( + value: Any, found: set[str], depth: int = 0, bare: bool = False +) -> None: + """Every artifact digest inside one stored value. + + With ``bare``, a plain ``sha256:...`` string counts as well, which is how a + node setting such as a Service's ``weights`` names its artifact. + """ if is_reference(value): found.add(str(value["digest"])) return + if bare and isinstance(value, str) and valid_digest(value): + found.add(value) + return if depth >= _REF_DEPTH: return if isinstance(value, dict): @@ -371,7 +380,7 @@ def _references_in(value: Any, found: set[str], depth: int = 0) -> None: else: return for item in items: - _references_in(item, found, depth + 1) + _references_in(item, found, depth + 1, bare) def sweep_artifacts( @@ -412,9 +421,12 @@ def referenced_artifacts( ) -> set[str]: """Every artifact digest something still refers to. - Three things do: a run that recorded it, a message currently holding it, - and a flow input's ``initial``, published or draft, which a flow that has - never started holds nowhere else. + Four things do: a run that recorded it, a message currently holding it, + a flow input's ``initial``, which a flow that has never started holds + nowhere else, and any node setting naming it, such as a Service's + ``weights``. Both the published flow and its draft count, since the draft + is what gets published next. Older versions live only in git history and + do not count. """ with Session(db_engine) as session: keep = set(session.exec(select(col(RunArtifact.digest)).distinct()).all()) @@ -428,6 +440,8 @@ def referenced_artifacts( continue for declared in flow.inputs: _references_in(declared.initial, keep) + for node in flow.nodes: + _references_in(node.params, keep, bare=True) keys = state.keys() for start in range(0, len(keys), _SWEEP_BATCH): diff --git a/backend/tests/test_artifacts.py b/backend/tests/test_artifacts.py index 649a2f3..2d8fd5e 100644 --- a/backend/tests/test_artifacts.py +++ b/backend/tests/test_artifacts.py @@ -11,7 +11,7 @@ from fluksio.core.db import engine as db_engine from fluksio.flow.artifacts import ArtifactStore, VolatileStore from fluksio.flow.messages import MessageSpec from fluksio.flow.runs import new_run_id, sweep_artifacts -from fluksio.flow.schemas import FlowDef, FlowInput +from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef from fluksio.flow.state import MemoryState from fluksio.flow.store import FlowStore from fluksio.models import Run, RunArtifact @@ -136,6 +136,20 @@ def test_a_sweep_keeps_what_a_flow_input_starts_with(store, tmp_path): assert store.path(kept["digest"]) is not None +def test_a_sweep_keeps_what_a_node_setting_names(store, tmp_path): + """A Service's weights outlive the run that produced them.""" + weights = store.put([b"weights"]) + orphan = store.put([b"referred to by nothing"]) + + flows = FlowStore(tmp_path / "flows") + params = {"nested": [{"weights": weights["digest"]}]} + flows.write_flow(FlowDef(name="f", nodes=[NodeDef(id="n", params=params)])) + + assert sweep_artifacts(store, MemoryState(), grace_s=0, flows=flows) == 1 + assert store.path(orphan["digest"]) is None + assert store.path(weights["digest"]) is not None + + def test_a_ring_drops_the_oldest_once_it_is_full(tmp_path): """What makes a camera affordable: the room is fixed, not the history.""" ring = VolatileStore(tmp_path / "ring", limit_bytes=3000) diff --git a/docs/code/api.md b/docs/code/api.md index 04ccedd..99c5479 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -180,9 +180,10 @@ superusers may see answers 403 to anyone else for the frames its messages hold, the last 8 included (see [Who may watch a camera](../interface/dashboards.md#who-may-watch-a-camera)). A superuser can remove stored bytes with `DELETE /artifacts/`. While -a run, a current message or a flow input's initial value still references the -digest, the call returns `409`. The hourly sweep removes unreferenced bytes -on its own, so this is for reclaiming space sooner. +a run, a current message, a flow input's initial value or a node setting in a +published or draft flow still references the digest, the call returns `409`. +The hourly sweep removes unreferenced bytes on its own, so this is for +reclaiming space sooner. ## Dashboards and devices diff --git a/docs/code/workers.md b/docs/code/workers.md index 0aec879..9b90629 100644 --- a/docs/code/workers.md +++ b/docs/code/workers.md @@ -118,6 +118,8 @@ Set `--service-host` on the worker where that address is wrong. A service's `weights` setting holds an artifact digest. The worker fetches it as node code fetches an artifact and passes the local path as `{weights}`. +The hourly artifact sweep keeps any digest a node setting names in a published +or draft flow, so the weights stay after the run that produced them is deleted. ## Resources and artifacts