diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 4cfb331..01ba8a4 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -151,16 +151,24 @@ class RunCacheLookup(Protocol): """What an equal execution produced, or None when there is no entry.""" -def run_cache_key(fingerprint: str, inputs: dict[str, Any]) -> str: +def run_cache_key(fingerprint: str, inputs: dict[str, Any], flow: str = "") -> str: """What this node, with these inputs, is known by. An artifact input counts as its digest: the reference carries a name and a size beside it, and the same bytes under another name are the same input. A value JSON cannot carry cannot be part of a key, and a node reading one is simply not cacheable. + + Inputs are keyed by the *node's* name for them, not the flow's: the same + node reading the same values through ``study`` and through ``quick`` did + the same work, and the fingerprint beside it already says what the node is. + A name belonging to some other flow keeps its prefix — reading + ``other.metric`` is part of what makes this execution what it is. """ reduced = { - name: value["digest"] if is_reference(value) else value + (name.removeprefix(f"{flow}.") if flow else name): ( + value["digest"] if is_reference(value) else value + ) for name, value in inputs.items() } try: @@ -932,7 +940,7 @@ class Pipeline: key = "" if self.run_cache is not None and node.fingerprint: - key = run_cache_key(node.fingerprint, inputs) + key = run_cache_key(node.fingerprint, inputs, node.flow) if key: hit, restored = self._from_cache(node, key, state, entry_id) if hit: diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py index b53bd36..9b06221 100644 --- a/backend/tests/flow/test_runs.py +++ b/backend/tests/flow/test_runs.py @@ -406,7 +406,7 @@ def test_a_cache_hit_restores_the_outputs_without_running_the_node(): node, calls = counting_node() state = MemoryState() seen: list[NodeOutcome] = [] - key = run_cache_key("fp-train", {"study.lr": 0.5}) + key = run_cache_key("fp-train", {"study.lr": 0.5}, "study") cache = FakeCache({key: {"study.loss": 99.0}}) Pipeline(nodes=[node], state=state, observer=seen.append, run_cache=cache).run( @@ -428,7 +428,7 @@ def test_a_hit_from_another_flow_restores_under_this_flow_s_names(): node, calls = counting_node() state = MemoryState() seen: list[NodeOutcome] = [] - key = run_cache_key("fp-train", {"study.lr": 0.5}) + key = run_cache_key("fp-train", {"study.lr": 0.5}, "study") # Recorded by a run of "other", which is what a node shared between two # flows gets a hit from — the values are the same, the namespace is not. cache = FakeCache({key: {"other.loss": 99.0}}, flow="other") @@ -455,7 +455,7 @@ def test_a_miss_runs_the_node_and_carries_what_would_be_stored(): ).run(seed_values(flow, {"lr": 0.5})) assert calls == [1] - assert cache.asked == [run_cache_key("fp-train", {"study.lr": 0.5})] + assert cache.asked == [run_cache_key("fp-train", {"study.lr": 0.5}, "study")] assert not seen[0].cached assert seen[0].cache_key and seen[0].output_values == {"study.loss": 1.0} @@ -467,6 +467,19 @@ def test_the_key_follows_the_inputs(): assert first == run_cache_key("fp", {"lr": 0.5}) +def test_the_key_does_not_follow_the_flow_an_input_hangs_in(): + # The same node reading the same value through two flows did the same work, + # so a run of one is a hit for the other. + assert run_cache_key("fp", {"study.lr": 0.5}, "study") == run_cache_key( + "fp", {"quick.lr": 0.5}, "quick" + ) + # A name belonging to neither keeps its prefix: reading another flow's + # message is part of what this execution is. + assert run_cache_key("fp", {"other.lr": 0.5}, "study") != run_cache_key( + "fp", {"other.lr": 0.5}, "other" + ) + + def test_an_artifact_input_counts_as_its_digest(): digest = "sha256:" + "0" * 64 # The same bytes under another name, of a size recorded differently, are