Artifact sweep keeps digests named in node settings
A Service's weights or a wren node's model held only in a setting were swept once the run that produced them was deleted.
This commit is contained in:
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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)
|
||||
|
||||
+4
-3
@@ -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/<digest>`. 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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user