From 068ff45108bfdb40a0556e30c8beb5e61d9baa54 Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 15 Oct 2026 10:54:22 +0200 Subject: [PATCH] Un-key stage-cache rows written before the current key shape CACHE_SHAPE_SINCE marks the last change that moved every cache key. The run service's periodic sweep clears cache_key on rows of runs created before it, which nothing can match any more. Their outputs stay. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx --- backend/fluksio/flow/controller.py | 3 ++- backend/fluksio/flow/pipeline.py | 8 ++++++ backend/fluksio/flow/runs.py | 23 +++++++++++++++++- backend/tests/api/routes/test_runs.py | 35 ++++++++++++++++++++++++++- docs/concepts/runs.md | 4 ++- 5 files changed, 69 insertions(+), 4 deletions(-) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 91d046e..8c8bf1c 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -1216,7 +1216,8 @@ class FlowController: # reaches hash to, which `sync` worked out by following # its imports. The ports are here because declaring a new # one changes what the node answers with, which a hit - # would otherwise restore without. + # would otherwise restore without. Changing what goes in + # here changes every key: move `CACHE_SHAPE_SINCE` with it. node.fingerprint = hashlib.sha256( json.dumps( { diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 76375a6..0060d3c 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -23,6 +23,7 @@ from collections import deque from collections.abc import Callable, Iterator from concurrent.futures import Future, ThreadPoolExecutor, wait from contextlib import contextmanager +from datetime import UTC, datetime from functools import lru_cache from typing import Any, Literal, Protocol @@ -166,6 +167,13 @@ class RunCacheLookup(Protocol): """What an equal execution produced, or None when there is no entry.""" +#: When the current shape of a cache key landed: the last change to what a +#: fingerprint hashes (`FlowController`) or to how the key below reduces its +#: inputs that moved every key. A row written before it matches nothing now, +#: so the run service's sweep un-keys it. Move it forward with such a change. +CACHE_SHAPE_SINCE = datetime(2026, 8, 26, 19, 27, 10, tzinfo=UTC) + + def run_cache_key(fingerprint: str, inputs: dict[str, Any], flow: str = "") -> str: """What this node, with these inputs, is known by. diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 43c0b83..4f5b3de 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -52,7 +52,7 @@ from fluksio.core.db import engine as db_engine from fluksio.flow.artifacts import ArtifactStore, is_reference, valid_digest from fluksio.flow.controller import NODE_TYPES, FlowController, RunContext from fluksio.flow.messages import qualify -from fluksio.flow.pipeline import CacheHit, NodeOutcome, Pipeline +from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, CacheHit, NodeOutcome, Pipeline from fluksio.flow.queue import WorkItem, WorkQueue from fluksio.flow.schemas import FlowDef, NodeDef from fluksio.flow.state import MemoryState, StateBackend @@ -759,6 +759,26 @@ def _cacheable(outcome: NodeOutcome) -> str | None: return text if len(text) <= OUTPUT_CAP else None +def prune_stale_cache() -> int: + """Un-key the stage-cache rows written before the current key shape. + + Such a row can match no key a node computes any more, so it is a cache + entry in name only. The row and its outputs stay: they are the record of + what that run's node returned. Returns how many rows were un-keyed. + """ + with Session(db_engine) as session: + older = select(Run.id).where(col(Run.created_at) < CACHE_SHAPE_SINCE) + result = session.exec( + update(RunNode) + .where(col(RunNode.cache_key) != "", col(RunNode.run_id).in_(older)) + .values(cache_key="") + ) + session.commit() + if result.rowcount: + logger.info("Un-keyed %d stage-cache rows of an older shape", result.rowcount) + return int(result.rowcount) + + class RunCache: """What earlier runs already worked out, looked up by cache key. @@ -1353,6 +1373,7 @@ class RunService: if time.monotonic() - last_sweep >= SWEEP_INTERVAL_S: last_sweep = time.monotonic() self._sweep_abandoned(now) + prune_stale_cache() except Exception: logger.exception("Could not refresh run leases") diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 3f512ff..0f1d23c 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -19,7 +19,7 @@ from fluksio.core.db import engine as db_engine from fluksio.flow.artifacts import ArtifactStore from fluksio.flow.controller import FlowController from fluksio.flow.messages import DType, MessageSpec -from fluksio.flow.pipeline import NodeOutcome +from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, NodeOutcome from fluksio.flow.runs import ( LEASE_STALE_S, OUTPUT_CAP, @@ -28,6 +28,7 @@ from fluksio.flow.runs import ( RunService, _cacheable, new_run_id, + prune_stale_cache, resolve_references, seed_values, ) @@ -106,6 +107,38 @@ def test_what_may_be_stored_as_a_cache_entry(): assert _cacheable(ok.model_copy(update={"output_values": big})) is None +def test_a_row_keyed_before_the_current_shape_is_unkeyed(): + """It matches nothing any more, but it is still the run's record.""" + outputs = json.dumps({"study.loss": 1.0}) + with Session(db_engine) as session: + for run_id, made in ( + ("shape-old", CACHE_SHAPE_SINCE - timedelta(days=1)), + ("shape-new", CACHE_SHAPE_SINCE + timedelta(seconds=1)), + ): + session.add(Run(id=run_id, flow="study", status="ok", created_at=made)) + session.add( + RunNode( + run_id=run_id, + node="study.a", + cache_key=run_id, + outputs=outputs, + ) + ) + session.commit() + try: + assert prune_stale_cache() == 1 + with Session(db_engine) as session: + old = session.get(RunNode, ("shape-old", "study.a")) + new = session.get(RunNode, ("shape-new", "study.a")) + assert old is not None and old.cache_key == "" and old.outputs == outputs + assert new is not None and new.cache_key == "shape-new" + finally: + with Session(db_engine) as session: + for run_id in ("shape-old", "shape-new"): + session.delete(session.get(Run, run_id)) + session.commit() + + # ----------------------------------------------------------------------------- # Naming an artifact from outside the process that made it # diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index ea77265..ecee531 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -139,7 +139,9 @@ Use `fluksio run ... --no-cache`, `fluksio sweep ... --no-cache`, or Only Python nodes are cached. Built-in and connector nodes are not. The cache uses run history; outputs too large to store there are not cacheable, and -missing artifact bytes make an entry a miss. +missing artifact bytes make an entry a miss. An entry written before the cache +key last changed shape can never match again, so the engine drops it from the +cache and keeps the node's recorded outputs. A cache hit does not replay emissions. Its metrics are read from the original run, so comparisons still show the recorded curve. Deleting that original run