diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 06850a0..d883416 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -162,10 +162,15 @@ class RunContext: """ run_id: str - #: What the code-defined flow's repository hashed to when this run started. - #: Part of every node's fingerprint, so editing a function the node calls - #: into invalidates the cache the way editing the node itself does. + #: What this run records as the code it ran: its nodes' digests, hashed. code_digest: str = "" + #: Per node, what the modules that node's function reaches hashed to when + #: the run was claimed. This is what a fingerprint is keyed on, so editing + #: a function the node calls into invalidates the cache the way editing + #: the node itself does — and editing something it does not reach leaves + #: the hit standing. Empty for a node synced before this was recorded, + #: which falls back to `code_digest`. + node_digests: dict[str, str] = field(default_factory=dict) @dataclass(frozen=True, slots=True) @@ -1132,8 +1137,9 @@ class FlowController: # flow is a shim that imports the real function — it does # not move when that function does, and it says nothing # about what the function calls into. `code` is what - # covers both: the digest of the repository the shim - # imports from. The ports are here because declaring a new + # covers both: what the modules this node's function + # 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. node.fingerprint = hashlib.sha256( @@ -1149,7 +1155,8 @@ class FlowController: spec.model_dump(mode="json") for spec in node_def.provides ], - "code": run.code_digest, + "code": run.node_digests.get(node_def.id) + or run.code_digest, }, sort_keys=True, separators=(",", ":"), diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 72bf8e6..39f4260 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -56,6 +56,7 @@ from fluksio.flow.schemas import FlowDef, NodeDef from fluksio.flow.state import MemoryState, StateBackend from fluksio.flow.store import FlowStore from fluksio.models import Run, RunArtifact, RunMetric, RunNode +from fluksio.sdk.imports import digest_of as files_digest logger = logging.getLogger(__name__) @@ -615,12 +616,71 @@ def repo_digest(repo: str) -> str: return answer +#: What a node's own modules hashed to, and the stat state behind that answer. +#: Keyed by the node's file list, the same trade `_digests` makes. +_node_digests: dict[tuple[str, ...], tuple[tuple[tuple[int, int], ...], str]] = {} + + +def node_digest(node: NodeDef, flow: FlowDef) -> str: + """What the code this node reaches hashes to now. + + `sync` recorded which project modules the node's function imports its way + to, so this is the narrow answer the repository-wide digest could not + give: it moves when the node's own code moves and stays put when a + notebook two directories away changes. + + Read live, because the shim imports from disk whenever the worker starts + and a sweep may sit queued for hours. An engine that cannot see the files + keeps what sync recorded instead of nothing — a remote worker's runs used + to share one empty digest, and one key. + """ + files = node.code_files + if not files: + # Drawn on the canvas, or synced before this was recorded. The + # repository is the older, blunter answer and still the only one. + return code_digest(flow) + paths = tuple(sorted(files.values())) + state: list[tuple[int, int]] = [] + for path in paths: + try: + info = os.stat(path) + except OSError: + return node.code_digest + state.append((info.st_mtime_ns, info.st_size)) + stamp = tuple(state) + cached = _node_digests.get(paths) + if cached is not None and cached[0] == stamp: + return cached[1] + answer = files_digest(files) or node.code_digest + _node_digests[paths] = (stamp, answer) + return answer + + +def node_digests(flow: FlowDef) -> dict[str, str]: + """Every node's digest, which is what its fingerprint is keyed on.""" + return {node.id: node_digest(node, flow) for node in flow.nodes} + + def code_digest(flow: FlowDef) -> str: """The digest of the repository a code-defined flow was declared in.""" origin = flow.origin return repo_digest(origin.repo) if origin is not None else "" +def run_digest(flow: FlowDef) -> str: + """What the run records as the code it ran: its nodes' digests, hashed. + + One column over a flow's several nodes, so it changes when any node's own + code does and not when anything else in the repository does — which is + what makes it worth joining an exported table on. + """ + parts = node_digests(flow) + if not any(parts.values()): + return "" + canonical = json.dumps(parts, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(canonical.encode()).hexdigest() + + def _cacheable(outcome: NodeOutcome) -> str | None: """A node's outputs as stored, or None when it may not be reused.""" if not outcome.ok or not outcome.cache_key: @@ -794,7 +854,7 @@ class RunService: flow_version=flow.version, commit=self.controller.store.head(), origin_commit=origin_commit(flow), - code_digest=code_digest(flow), + code_digest=run_digest(flow), params=params, params_digest=digest_of(params, seed), seed=seed, @@ -1070,14 +1130,19 @@ class RunService: # run at once, and the code on disk is free to move in the hours # before the last of them starts. What the record must name is the # state that ran, not the state that was submitted. - run.code_digest = self._restamp(run, code_digest(flow)) + digests = node_digests(flow) + run.code_digest = self._restamp(run, run_digest(flow)) state = self._state_factory(f"{RUN_NAMESPACE}:{run_id}") pipeline = self.controller.build_run_pipeline( flow, state=state, observer=observe, emission_observer=sink.handle, - run=RunContext(run_id=run_id, code_digest=run.code_digest), + run=RunContext( + run_id=run_id, + code_digest=run.code_digest, + node_digests=digests, + ), run_cache=None if run.no_cache else self._cache, ) with self._lock: diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index 8a15757..71acdcb 100644 --- a/backend/fluksio/flow/schemas.py +++ b/backend/fluksio/flow/schemas.py @@ -206,6 +206,23 @@ class NodeDef(BaseModel): "workers, which is right for everything that is not compute-heavy." ), ) + code_files: dict[str, str] = Field( + default_factory=dict, + description=( + "The project modules this node's function reaches, dotted name to " + "the file on the machine that synced it. Written by `fluksio " + "sync`, which is the only side that can import the code and see " + "what it imports. Empty for a node drawn on the canvas." + ), + ) + code_digest: str = Field( + default="", + description=( + "What those modules hashed to at sync. The engine hashes them " + "again when a run is claimed, and falls back to this when it " + "cannot see the files — a worker on another machine." + ), + ) @field_validator("id") @classmethod diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index facf75c..e74e5f6 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -23,6 +23,8 @@ from collections.abc import Callable, Sequence from pathlib import Path from typing import Any, TypeVar +from fluksio.sdk.imports import digest_of, reached + __all__ = [ "FLOWS", "Flow", @@ -549,6 +551,18 @@ def _check(spec: NodeSpec, mode: str) -> dict[str, Any]: } +def _code_of(spec: NodeSpec) -> dict[str, Any]: + """What this node's function calls into, which its shim does not say. + + Read here rather than when the flow is declared, so a second `sync()` in + one process — a notebook — sees the edit that happened between them. This + side is the only one that can work it out at all: it has imported the + code, and the engine never does. + """ + files = reached(spec.fn) + return {"code_files": files, "code_digest": digest_of(files)} + + def _serialisable(value: Any) -> bool: try: json.dumps(value) @@ -676,7 +690,10 @@ class Flow: "name": self.name, "title": self.title, "mode": self.mode, - "nodes": self._nodes, + "nodes": [ + node | _code_of(spec) + for spec, node in zip(self.nodes, self._nodes, strict=True) + ], "inputs": [ {"spec": port.spec(), "initial": port.initial} for port in self.inputs ], diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index e920fc3..a93ca60 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -14,9 +14,10 @@ import importlib import itertools import json import pkgutil +import shutil import sys import time -from collections.abc import Callable, Iterator +from collections.abc import Callable, Iterable, Iterator from contextlib import contextmanager, nullcontext from pathlib import Path from typing import Any @@ -744,11 +745,36 @@ def cmd_status(args: argparse.Namespace) -> int: return _unreachable(exc) -#: How much of a run's inputs the list shows. A flow taking a few kB of JSON -#: would otherwise make the table unreadable; `client.runs()` is where the -#: whole value is read. +#: How much of a run's inputs the list shows when nothing says how wide the +#: terminal is. `client.runs()` and `fluksio export runs` are where the whole +#: value is read. PARAMS_WIDTH = 80 +#: What the columns before the inputs take: the id, status, flow, duration and +#: stamp, with their spacing. +LISTING_WIDTH = 84 + + +def _declared(client: Client, flows: Iterable[str]) -> dict[str, dict[str, Any]]: + """What each flow declares its inputs to be, by flow. + + So the listing can leave out an input that is simply the declared value: + what a reader is looking for across a page of runs is where they differ, + and a flow taking a few kB of JSON crowds that off the line. + """ + known: dict[str, dict[str, Any]] = {} + for name in flows: + try: + stored = client.get_flow(name) or {} + except (SyncError, ApiError, httpx.HTTPError): + # A flow deleted since its runs were recorded still lists them. + stored = {} + known[name] = { + str((entry.get("spec") or {}).get("name", "")): entry.get("initial") + for entry in (stored.get("definition") or {}).get("inputs") or [] + } + return known + def _stamp(row: dict[str, Any]) -> str: """What code a run ran: the commit, whether it was dirty, and the digest. @@ -767,14 +793,29 @@ def cmd_runs(args: argparse.Namespace) -> int: try: with _client_for(args, retries=0) as client: rows = client.runs(flow=args.flow, limit=args.limit) + declared = _declared(client, {str(row["flow"]) for row in rows}) except (SyncError, ApiError) as exc: return _fail(str(exc)) except httpx.HTTPError as exc: return _unreachable(exc) + # Whatever is left of the line after the fixed columns; the fallback is + # what a pipe gets, since there is no width to ask for there. + room = max( + 20, + shutil.get_terminal_size((LISTING_WIDTH + PARAMS_WIDTH, 24)).columns + - LISTING_WIDTH, + ) for row in rows: - params = json.dumps(row["params"]) - if len(params) > PARAMS_WIDTH: - params = params[: PARAMS_WIDTH - 3] + "..." + given = row["params"] or {} + defaults = declared.get(str(row["flow"])) or {} + shown = { + name: value + for name, value in given.items() + if name not in defaults or defaults[name] != value + } + params = json.dumps(shown) + if len(params) > room: + params = params[: room - 3] + "..." _say( f"{row['id']} {_status(row['status'], 9)} {row['flow']:<16} " f"{row['duration_ms'] / 1000:7.1f}s {_stamp(row):<22} {params}" diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 335212b..dad2645 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -726,13 +726,19 @@ def _sync_one( if not force: known = _refuse_on_drift(client, target, definition) - saved = client.put_flow(target.document(origin) | {"version": version}) + document = target.document(origin) + moved = _moved_code(stored, document) + saved = client.put_flow(document | {"version": version}) stored_version = int((saved.get("definition") or {}).get("version") or version) if report.created or stored_version != version: # The store bumps the version only when the content actually changed, # so this is its own no-op detection rather than a second guess at it. report.changed.append("flow") version = stored_version + # Named before the shims are compared, because this is the case a shim + # comparison cannot see: the function is unchanged and something it calls + # into is not. + report.changed.extend(moved) for node_id, code in target.shims().items(): if not report.created: @@ -742,7 +748,8 @@ def _sync_one( if stored_code == code: continue client.put_source(target.name, node_id, code) - report.changed.append(node_id) + if node_id not in report.changed: + report.changed.append(node_id) if publish and (client.get_flow(target.name) or {}).get("has_draft"): client.publish(target.name, version) @@ -750,6 +757,28 @@ def _sync_one( return report +def _moved_code(stored: dict[str, Any] | None, document: dict[str, Any]) -> list[str]: + """The nodes whose reachable code changed since the last sync. + + A node is otherwise named only when its generated shim text changes, and + the shim reflects the declared ports rather than what the function calls + into — so editing a helper printed as nothing at all, which reads as + "there was nothing to do". + """ + if stored is None: + return [] + before = { + node.get("id"): node.get("code_digest") or "" + for node in (stored.get("definition") or {}).get("nodes") or [] + } + return [ + node_id + for node in document.get("nodes") or [] + if (node_id := node.get("id")) in before + and before[node_id] != (node.get("code_digest") or "") + ] + + def _refuse_on_drift( client: Client, target: Flow, definition: dict[str, Any] ) -> dict[str, str]: diff --git a/backend/fluksio/sdk/imports.py b/backend/fluksio/sdk/imports.py new file mode 100644 index 0000000..52acdb7 --- /dev/null +++ b/backend/fluksio/sdk/imports.py @@ -0,0 +1,155 @@ +"""What code a node actually reaches, so a cache key can name it. + +A node's stored body is a shim that imports the real function, so it says +nothing about what that function calls into. Hashing the whole repository +instead says too much: it moves when a notebook two directories away does, +which makes the digest neither necessary nor sufficient for "this run's +numbers came from that code". + +This walks the imports out from the module a node's function lives in and +stops at anything installed or standard, so what is left is the project's own +code — including a package installed editable from its own checkout, whose +files are its source rather than a copy under `site-packages`. + +The walk is static: it reads the imports out of each file rather than running +it. A module imported by a name the code computes is therefore not followed, +which is the same limit the yield check has, and for the same reason. +""" + +from __future__ import annotations + +import ast +import hashlib +import importlib.util +import sys +import sysconfig +from collections.abc import Iterator +from pathlib import Path +from typing import Any + +__all__ = ["digest_of", "reached"] + + +#: Never the project's own code, however it happens to be installed. Every +#: node's module imports the authoring API, and from a checkout or an editable +#: install that is a source tree like any other — so without this, a node's +#: digest is mostly this tool, and upgrading it invalidates every cache entry +#: anybody has. +_SELF = frozenset({"fluksio", "fluksio_worker"}) + + +def _foreign() -> tuple[str, ...]: + """Where code that is not the project's lives. + + The standard library and whatever `pip install` copied. An editable + install is deliberately not here: its files are the checkout, and a + checkout is somebody's code. + """ + roots = { + path + for key in ("stdlib", "platstdlib", "purelib", "platlib") + if (path := sysconfig.get_paths().get(key)) + } + roots.update( + entry + for entry in sys.path + if entry.endswith(("site-packages", "dist-packages")) + ) + return tuple(sorted(roots)) + + +def _project_file(name: str, foreign: tuple[str, ...]) -> Path | None: + """The file a dotted module is, if it is the project's own python.""" + module = sys.modules.get(name) + origin = getattr(module, "__file__", None) if module is not None else None + if origin is None: + try: + spec = importlib.util.find_spec(name) + except (ImportError, ValueError, AttributeError): + return None + origin = spec.origin if spec is not None else None + if not origin or not origin.endswith(".py"): + return None + return None if origin.startswith(foreign) else Path(origin) + + +def _package_of(name: str, path: Path) -> str: + """The package a relative import inside this module resolves against.""" + return name if path.name == "__init__.py" else name.rpartition(".")[0] + + +def _imported(tree: ast.AST, package: str) -> Iterator[str]: + """Every module name a file imports, relative ones resolved. + + `from pkg import thing` yields both `pkg` and `pkg.thing`, since only + trying to resolve them says which of the two `thing` is. + """ + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + yield alias.name + elif isinstance(node, ast.ImportFrom): + base = node.module or "" + if node.level: + parts = package.split(".") if package else [] + parts = parts[: len(parts) - (node.level - 1)] + base = ".".join([*parts, base]) if base else ".".join(parts) + if not base: + continue + yield base + for alias in node.names: + yield f"{base}.{alias.name}" + + +def reached(fn: Any) -> dict[str, str]: + """The project modules ``fn``'s own module reaches, itself included. + + Dotted name to the file it is on this machine — the name is what the + digest is keyed on, so an engine hashing the same code arrives at the same + answer; the path is how it finds the file. + """ + start = getattr(fn, "__module__", "") + if not start: + return {} + foreign = _foreign() + found: dict[str, str] = {} + queue = [start] + while queue: + name = queue.pop() + if name in found or name.partition(".")[0] in _SELF: + continue + path = _project_file(name, foreign) + if path is None: + continue + try: + tree = ast.parse(path.read_bytes()) + except (OSError, SyntaxError, ValueError): + continue + found[name] = str(path) + queue.extend(_imported(tree, _package_of(name, path))) + # The packages above it, whose `__init__` runs on the way to importing + # this module and is as much a part of what it does as an import is. + parts = name.split(".")[:-1] + queue.extend(".".join(parts[: index + 1]) for index in range(len(parts))) + return found + + +def digest_of(files: dict[str, str]) -> str: + """What those modules hash to now, or empty if one is not readable here. + + All or nothing: a digest over the half of the code an engine happens to + have would be a confident answer to a question it cannot see. + """ + if not files: + return "" + digest = hashlib.sha256() + for name, path in sorted(files.items()): + try: + body = Path(path).read_bytes() + except OSError: + return "" + digest.update(name.encode()) + digest.update(b"\0") + digest.update(body) + digest.update(b"\0") + return digest.hexdigest() diff --git a/backend/tests/api/routes/test_sync.py b/backend/tests/api/routes/test_sync.py index 8431561..161bb4a 100644 --- a/backend/tests/api/routes/test_sync.py +++ b/backend/tests/api/routes/test_sync.py @@ -65,6 +65,45 @@ def test_sync_stores_a_runnable_flow_with_its_origin(api, flows): assert "from myresearch.train import fit" in source +def test_a_node_carries_the_modules_its_code_reaches(api, flows): + """The shim names one function; the cache has to key on what it calls.""" + sync([flows["train"]], api, origin=ORIGIN) + + nodes = {n["id"]: n for n in api.get_flow("train")["definition"]["nodes"]} + files = nodes["fit"]["code_files"] + + # The module the function is in, and the package whose `__init__` runs on + # the way to it. Not the engine's own code, and not the standard library. + assert "myresearch.train" in files + assert all(path.endswith(".py") for path in files.values()) + assert not any(name.startswith("fluksio") for name in files) + assert len(nodes["fit"]["code_digest"]) == 64 + + +def test_a_node_whose_helper_moved_is_named_by_sync(): + """Editing a helper changes no shim, so nothing used to say it happened.""" + from fluksio.sdk.client import _moved_code + + stored = { + "definition": { + "nodes": [ + {"id": "fit", "code_digest": "aaa"}, + {"id": "prepare", "code_digest": "bbb"}, + ] + } + } + document = { + "nodes": [ + {"id": "fit", "code_digest": "ccc"}, + {"id": "prepare", "code_digest": "bbb"}, + ] + } + + assert _moved_code(stored, document) == ["fit"] + # Nothing to compare against is not a change. + assert _moved_code(None, document) == [] + + def test_a_second_sync_changes_nothing_and_commits_nothing(api, flows): sync([flows["train"]], api, origin=ORIGIN) before = commits() diff --git a/backend/tests/sdk/test_imports.py b/backend/tests/sdk/test_imports.py new file mode 100644 index 0000000..a21e282 --- /dev/null +++ b/backend/tests/sdk/test_imports.py @@ -0,0 +1,101 @@ +"""What a node's code reaches — the narrow answer a cache key is keyed on.""" + +import sys +import textwrap + +from fluksio.sdk.imports import digest_of, reached + + +def _write(root, name, body): + path = root / f"{name}.py" + path.write_text(textwrap.dedent(body).lstrip()) + return path + + +def _study(tmp_path): + """A package whose node function delegates to a helper beside it.""" + root = tmp_path / "study" + root.mkdir() + (root / "__init__.py").write_text("VERSION = 1\n") + _write( + root, + "train", + """ + import json + + from study.helpers import curve + + + def fit(lr): + return json.dumps(curve(lr)) + """, + ) + _write(root, "helpers", "def curve(lr):\n return [lr]\n") + _write(root, "unused", "def nothing():\n return 0\n") + return root + + +def test_the_walk_follows_a_helper_and_stops_at_installed_code(tmp_path, monkeypatch): + """The shim says nothing about this, and the repository says too much.""" + _study(tmp_path) + monkeypatch.syspath_prepend(str(tmp_path)) + for name in [n for n in sys.modules if n == "study" or n.startswith("study.")]: + del sys.modules[name] + + from study.train import fit + + files = reached(fit) + + # The module, the helper it calls into, and the package whose `__init__` + # runs on the way in. Not the sibling nobody imports, and not `json`. + assert set(files) == {"study", "study.train", "study.helpers"} + assert all(path.endswith(".py") for path in files.values()) + + +def test_the_digest_moves_with_the_helper_and_not_with_the_neighbour( + tmp_path, monkeypatch +): + """The whole point: it is neither necessary nor sufficient otherwise.""" + root = _study(tmp_path) + monkeypatch.syspath_prepend(str(tmp_path)) + for name in [n for n in sys.modules if n == "study" or n.startswith("study.")]: + del sys.modules[name] + + from study.train import fit + + files = reached(fit) + before = digest_of(files) + + _write(root, "unused", "def nothing():\n return 1\n") + assert digest_of(files) == before + + _write(root, "helpers", "def curve(lr):\n return [lr, lr]\n") + assert digest_of(files) != before + + # A file this engine cannot see is not half an answer. + assert digest_of({**files, "study.gone": str(root / "gone.py")}) == "" + + +def test_the_tool_itself_is_not_the_projects_code(tmp_path, monkeypatch): + """Every node imports the authoring API, and from a checkout that is source.""" + root = tmp_path / "arm" + root.mkdir() + (root / "__init__.py").write_text("") + _write( + root, + "work", + """ + from fluksio import node + + + def fit(lr): + return lr + """, + ) + monkeypatch.syspath_prepend(str(tmp_path)) + for name in [n for n in sys.modules if n == "arm" or n.startswith("arm.")]: + del sys.modules[name] + + from arm.work import fit + + assert set(reached(fit)) == {"arm", "arm.work"} diff --git a/docs/code/cli.md b/docs/code/cli.md index c954665..46080ea 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -160,7 +160,14 @@ file path, because the shim has to import the same way. Every sync retires the engine's workers, including one that had nothing to upload — a worker holds your package in memory, so an edit to it is invisible -until the process goes. See +until the process goes. + +It also records, per node, which of your modules that node's function imports +its way to, and what they hash to. That is what the +[stage cache](../concepts/runs.md#stage-caching) keys on, so editing a +helper a node calls into is reported as that node changing — `train: updated +(flow, fit)` — and re-runs it, while editing something the node never reaches +is left alone. See [Getting started: data science](../getting-started/data-science.md). ### `fluksio run` @@ -275,12 +282,17 @@ fluksio runs [--flow train] [--limit 20] ``` The runs an engine has recorded, newest first: id, status, flow, duration, the -commit of the repository it came from, and its parameters. Statuses are -coloured when a terminal is reading the output — `ok` green, `error` red, -`cached` cyan. Parameters are clamped to 80 characters so a flow taking a few -kB of JSON still lists as a table; `Client.runs()` is where the whole value is -read. `--local` reads the same history from an in-process engine, without one -having to be served. +commit of the repository it came from, and the inputs it was given. Statuses +are coloured when a terminal is reading the output — `ok` green, `error` red, +`cached` cyan. + +Only the inputs that *differ from what the flow declares* are shown, and they +are clamped to what is left of the terminal's width — a run that took the +defaults lists none at all, and a flow taking a few kB of JSON does not push +everything else off the line. `Client.runs()` and +[`fluksio export runs`](#fluksio-export) are where the whole value is read. +`--local` reads the same history from an in-process engine, without one having +to be served. ### `fluksio flavors` diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index 0cfa433..bbafd5f 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -191,13 +191,20 @@ on a run carries the `cache_key` it was looked up by. For a [code-defined flow](../getting-started/data-science.md), "its source" is the generated shim, which imports the real function and does not change when -that function does. So the key carries one thing more: a digest of every -`.py` file under the repository the flow was declared in, read when the run -starts. Editing a helper three calls down from the node invalidates it, which -is the point — the alternative is a re-run answering with the previous code's -numbers. It is deliberately blunt: any edit anywhere in the repository re-runs -every node of its flows. An engine that cannot see the repository — a worker on -another machine — records no digest and keys as it did before. +that function does. So the key carries one thing more: a digest of the project +modules that node's function reaches, worked out by `fluksio sync` — which is +the only side that imports your code and can see what it imports — and read +again from those files when the run starts. + +Editing a helper three calls down from the node invalidates it, which is the +point: the alternative is a re-run answering with the previous code's numbers. +Editing something the node does not reach leaves the hit standing, which is +the other half — a notebook two directories away is not a reason to retrain. +The walk follows imports statically and stops at the standard library, at +anything installed, and at Fluksio itself; a module imported under a name the +code computes is not followed. An engine that cannot see the files keeps what +sync recorded instead of nothing, so a worker on another machine no longer +keys every run the same. The run history *is* the cache; there is no second store. A node's returned outputs are kept on its run record as canonical JSON, up to 256000 characters — diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md index 4bf66e8..d156971 100644 --- a/docs/getting-started/data-science.md +++ b/docs/getting-started/data-science.md @@ -542,9 +542,13 @@ Beside your commit is `code_digest`, and `fluksio runs` prints the pair as `a1b2c3d-dirty+9f0e1a2`. The commit alone cannot identify what ran: your node bodies are imports, so the engine executes whatever is on disk when the worker starts, and an uncommitted tree stamps `-dirty` for every run it ever produces. -The digest is read at the moment the run starts — so in a sweep whose runs -queue for hours, each one records the code that actually executed it, not the -code that was there when you submitted. +The digest is over the modules this flow's nodes actually reach — not the +whole repository, so it moves when the code behind a number moves and stays +put when a notebook beside it changes — and it is read at the moment the run +starts, so in a sweep whose runs queue for hours each one records the code +that actually executed it rather than the code that was there when you +submitted. It is what makes the column worth joining an +[exported table](../concepts/runs.md#taking-it-into-a-dataframe) on. ### When the engine is busy