diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 1082af5..a465891 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -56,7 +56,7 @@ from fluksio.flow.pipeline import CacheHit, NodeOutcome, Pipeline from fluksio.flow.queue import WorkItem, WorkQueue from fluksio.flow.schemas import FlowDef, NodeDef from fluksio.flow.state import MemoryState, StateBackend -from fluksio.flow.store import FlowStore +from fluksio.flow.store import FlowNotFound, FlowStore from fluksio.models import Run, RunArtifact, RunMetric, RunNode from fluksio.sdk.imports import digest_of as files_digest @@ -375,16 +375,20 @@ def _references_in(value: Any, found: set[str], depth: int = 0) -> None: def sweep_artifacts( - store: ArtifactStore, state: StateBackend, grace_s: float = 0.0 + store: ArtifactStore, + state: StateBackend, + grace_s: float = 0.0, + flows: FlowStore | None = None, ) -> int: """Remove artifact bytes nothing refers to any more. Returns how many went. - Two things refer to an artifact: a run that recorded it, and a message - currently holding it. The second is what makes a stream of media - affordable — a camera publishing a frame a second replaces the reference - each time, so yesterday's frames are unreferenced by definition and the - store does not grow without bound. What a node *returns* is recorded - against its run and kept; what it emits along the way is not. + Three things refer to an artifact: 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. The second is what + makes a stream of media affordable — a camera publishing a frame a second + replaces the reference each time, so yesterday's frames are unreferenced by + definition and the store does not grow without bound. What a node *returns* + is recorded against its run and kept, what it emits along the way is not. Skipped entirely while anything is running: a node that stores a checkpoint an hour before it returns has neither a row nor a message naming it yet, @@ -399,6 +403,15 @@ def sweep_artifacts( return 0 keep = set(session.exec(select(col(RunArtifact.digest)).distinct()).all()) + for name in flows.list_flows() if flows else (): + for draft in (False, True): + try: + flow = flows.read_flow(name, draft=draft) + except FlowNotFound: + continue + for declared in flow.inputs: + _references_in(declared.initial, keep) + keys = state.keys() for start in range(0, len(keys), _SWEEP_BATCH): for value in state.get_multi(keys[start : start + _SWEEP_BATCH]).values(): diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 638f1d9..dd534d5 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -145,6 +145,7 @@ async def _sweep_artifacts(store: ArtifactStore, controller: FlowController) -> store, controller.state, settings.ARTIFACT_GC_GRACE_S, + controller.store, ) except Exception: logger.exception("Artifact sweep failed") diff --git a/backend/tests/test_artifacts.py b/backend/tests/test_artifacts.py index ecaa3c4..cbe145c 100644 --- a/backend/tests/test_artifacts.py +++ b/backend/tests/test_artifacts.py @@ -9,8 +9,11 @@ from sqlmodel import Session, col, select from fluksio.core.config import settings 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.state import MemoryState +from fluksio.flow.store import FlowStore from fluksio.models import Run, RunArtifact @@ -112,6 +115,27 @@ def test_a_streamed_chunk_falls_out_once_the_message_moves_on(store): assert store.path(second["digest"]) is not None +def test_a_sweep_keeps_what_a_flow_input_starts_with(store, tmp_path): + """A flow that has never run holds its `initial` in no message yet.""" + published = store.put([b"published initial"]) + drafted = store.put([b"draft initial"]) + orphan = store.put([b"referred to by nothing"]) + + flows = FlowStore(tmp_path / "flows") + clip = MessageSpec(name="clip", dtype="artifact") + flows.write_flow( + FlowDef(name="f", inputs=[FlowInput(spec=clip, initial=published)]) + ) + flows.write_draft( + FlowDef(name="g", inputs=[FlowInput(spec=clip, initial=drafted)]), 0 + ) + + assert sweep_artifacts(store, MemoryState(), grace_s=0, flows=flows) == 1 + assert store.path(orphan["digest"]) is None + for kept in (published, drafted): + assert store.path(kept["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)