From 293a94d4e44d45bab0af3959f3f90cdbc65af6bf Mon Sep 17 00:00:00 2001 From: Herman Brunborg Date: Sun, 31 May 2026 12:14:33 -0700 Subject: [PATCH] Recheck dynamic dependencies before scheduling - Add recheck intervals for dependencies with TTL-based caching - Refresh declared DAG edges and prune obsolete orphan nodes - Track declared vs runtime dependencies in the manager --- src/furu/__init__.py | 3 +- src/furu/dag.py | 144 +++++++++++++++++++++++++++++++--- src/furu/dependencies.py | 123 +++++++++++++++++++++-------- src/furu/execution/manager.py | 21 ++++- tests/test_core.py | 98 +++++++++++++++++++---- tests/test_dag.py | 38 ++++++++- tests/test_worker_manager.py | 126 +++++++++++++++++++++++++++++ 7 files changed, 493 insertions(+), 60 deletions(-) diff --git a/src/furu/__init__.py b/src/furu/__init__.py index f72f86a5..cccd0b50 100644 --- a/src/furu/__init__.py +++ b/src/furu/__init__.py @@ -1,7 +1,7 @@ from importlib.metadata import version from furu.core import Furu -from furu.dependencies import dependency +from furu.dependencies import RecheckInterval, dependency from furu.execution import load_or_create from furu.logging import get_logger from furu.migration import Migration @@ -18,6 +18,7 @@ "LazyResult", "Migration", "ResourceRequirements", + "RecheckInterval", "dependency", "ResultCodec", "ResultRegistry", diff --git a/src/furu/dag.py b/src/furu/dag.py index dfc8a253..b493374f 100644 --- a/src/furu/dag.py +++ b/src/furu/dag.py @@ -17,6 +17,43 @@ class DagNode: obj: Furu dependencies: list[DagNode] = field(default_factory=list) dependents: list[DagNode] = field(default_factory=list) + declared_dependency_ids: set[str] = field(default_factory=set) + runtime_dependency_ids: set[str] = field(default_factory=set) + + +def _ensure_edge(node: DagNode, dep_node: DagNode) -> None: + if dep_node not in node.dependencies: + node.dependencies.append(dep_node) + if node not in dep_node.dependents: + dep_node.dependents.append(node) + + +def _remove_edge(node: DagNode, dep_node: DagNode) -> None: + if dep_node in node.dependencies: + node.dependencies.remove(dep_node) + if node in dep_node.dependents: + dep_node.dependents.remove(node) + + +def _is_running(manager: Manager, node: DagNode) -> bool: + return any(running.node is node for running in manager.running.values()) + + +def _move_ready_or_blocked(manager: Manager, node: DagNode) -> None: + object_id = node.obj.object_id + if ( + object_id in manager.completed + or object_id in manager.failed + or _is_running(manager, node) + ): + return + + if node.dependencies: + manager.ready.pop(object_id, None) + manager.blocked[object_id] = node + else: + manager.blocked.pop(object_id, None) + manager.ready[object_id] = node def _add_to_dag(manager: Manager, objs: Sequence[Furu]) -> None: @@ -54,9 +91,13 @@ def _add_to_dag(manager: Manager, objs: Sequence[Furu]) -> None: for obj_id, refs in refs_by_id.items(): node = manager.nodes_by_id[obj_id] for ref in refs: + if ref.object_id in manager.completed: + continue if dep_node := manager.nodes_by_id.get(ref.object_id): - node.dependencies.append(dep_node) - dep_node.dependents.append(node) + if dep_node.obj.status() == "completed": + continue + node.declared_dependency_ids.add(ref.object_id) + _ensure_edge(node, dep_node) for node in newly_added: if node.dependencies: @@ -94,12 +135,95 @@ def _update_dag_blocking_dependencies( for dependency_id in dependency_ids: dep_node = manager.nodes_by_id[dependency_id] - if dep_node not in node.dependencies: - node.dependencies.append(dep_node) - if node not in dep_node.dependents: - dep_node.dependents.append(node) + node.runtime_dependency_ids.add(dependency_id) + _ensure_edge(node, dep_node) - if node.dependencies: - manager.blocked[node.obj.object_id] = node - else: - manager.ready[node.obj.object_id] = node + _move_ready_or_blocked(manager, node) + + +def _refresh_dag_declared_dependencies(manager: Manager) -> None: + nodes = tuple(manager.blocked.values()) + tuple(manager.ready.values()) + for node in nodes: + if manager.nodes_by_id.get(node.obj.object_id) is not node: + continue + refs = collect_declared_refs(node.obj) + _reconcile_declared_dependencies(manager, node, refs) + + +def _reconcile_declared_dependencies( + manager: Manager, + node: DagNode, + refs: Sequence[Furu], +) -> None: + wanted_declared_ids: set[str] = set() + wanted_declared_ids_in_order: list[str] = [] + missing_dependencies: list[Furu] = [] + + for ref in refs: + object_id = ref.object_id + if object_id == node.obj.object_id or object_id in manager.completed: + continue + + dep_node = manager.nodes_by_id.get(object_id) + if dep_node is not None: + if dep_node.obj.status() != "completed": + wanted_declared_ids.add(object_id) + wanted_declared_ids_in_order.append(object_id) + continue + + match ref.status(): + case "completed": + continue + case "running" | "missing" | "failed": + wanted_declared_ids.add(object_id) + wanted_declared_ids_in_order.append(object_id) + missing_dependencies.append(ref) + case x: + assert_never(x) + + _add_to_dag(manager, missing_dependencies) + + for dependency_id in tuple(node.declared_dependency_ids - wanted_declared_ids): + node.declared_dependency_ids.discard(dependency_id) + if dependency_id in node.runtime_dependency_ids: + continue + + dep_node = manager.nodes_by_id.get(dependency_id) + if dep_node is None: + continue + + _remove_edge(node, dep_node) + _prune_orphan_dependency(manager, dep_node) + + for dependency_id in wanted_declared_ids_in_order: + dep_node = manager.nodes_by_id.get(dependency_id) + if dep_node is None: + continue + + node.declared_dependency_ids.add(dependency_id) + _ensure_edge(node, dep_node) + + _move_ready_or_blocked(manager, node) + + +def _prune_orphan_dependency(manager: Manager, node: DagNode) -> None: + object_id = node.obj.object_id + if ( + object_id in manager.root_ids + or node.dependents + or _is_running(manager, node) + or manager.nodes_by_id.get(object_id) is not node + ): + return + + manager.ready.pop(object_id, None) + manager.blocked.pop(object_id, None) + manager.completed.pop(object_id, None) + manager.failed.pop(object_id, None) + manager.nodes_by_id.pop(object_id, None) + + for dep_node in tuple(node.dependencies): + node.declared_dependency_ids.discard(dep_node.obj.object_id) + node.runtime_dependency_ids.discard(dep_node.obj.object_id) + _remove_edge(node, dep_node) + _prune_orphan_dependency(manager, dep_node) diff --git a/src/furu/dependencies.py b/src/furu/dependencies.py index 8c7d3508..f45b9b34 100644 --- a/src/furu/dependencies.py +++ b/src/furu/dependencies.py @@ -4,7 +4,7 @@ from contextlib import contextmanager from contextvars import ContextVar from dataclasses import fields, is_dataclass -from functools import cached_property +import time from typing import TYPE_CHECKING, Any, Callable, Literal, overload from pydantic import BaseModel as PydanticBaseModel @@ -13,53 +13,110 @@ from furu.core import Furu -class _CachedDependency[T](cached_property): - __furu_dependency__ = True +type RecheckInterval = int | Literal["never"] +type _DependencyCacheEntry[T] = tuple[float | None, T] -class _UncachedDependency[T](property): +class _Dependency[T](property): __furu_dependency__ = True - -@overload -def dependency[TFuru: Furu[Any], T]( - func: Callable[[TFuru], T], / -) -> _CachedDependency[T]: ... - - -@overload -def dependency[TFuru: Furu[Any], T]( - *, cached: Literal[True] = True -) -> Callable[[Callable[[TFuru], T]], _CachedDependency[T]]: ... + def __init__( + self, + func: Callable[[Any], T], + *, + recheck_interval: RecheckInterval, + ) -> None: + if isinstance(recheck_interval, bool): + raise TypeError( + "recheck_interval must be a non-negative integer or 'never'" + ) + if isinstance(recheck_interval, int): + if recheck_interval < 0: + raise ValueError("recheck_interval must be non-negative") + elif recheck_interval != "never": + raise TypeError( + "recheck_interval must be a non-negative integer or 'never'" + ) + + super().__init__(func) + self._recheck_interval = recheck_interval + func_name = getattr(func, "__name__", type(func).__name__) + self._cache_key = f"__furu_dependency_cache_{func_name}" + + def __set_name__(self, owner: type[object], name: str) -> None: + self._cache_key = f"__furu_dependency_cache_{name}" + + @overload + def __get__( + self, + obj: None, + objtype: type[object] | None = None, + ) -> _Dependency[T]: ... + + @overload + def __get__( + self, + obj: object, + objtype: type[object] | None = None, + ) -> T: ... + + def __get__( + self, + obj: object | None, + objtype: type[object] | None = None, + ) -> T | _Dependency[T]: + if obj is None: + return self + + cached: _DependencyCacheEntry[T] | None = getattr(obj, self._cache_key, None) + now: float | None = None + if cached is not None: + expires_at, value = cached + if expires_at is None: + return value + + now = time.monotonic() + if now < expires_at: + return value + + if self.fget is None: + raise AttributeError("unreadable dependency") + + recheck_interval = self._recheck_interval + if isinstance(recheck_interval, int) and now is None: + now = time.monotonic() + + value = self.fget(obj) + if recheck_interval == "never": + expires_at = None + else: + assert isinstance(recheck_interval, int) + assert now is not None + expires_at = now + recheck_interval + object.__setattr__(obj, self._cache_key, (expires_at, value)) + return value @overload def dependency[TFuru: Furu[Any], T]( - *, cached: Literal[False] -) -> Callable[[Callable[[TFuru], T]], _UncachedDependency[T]]: ... + func: Callable[[TFuru], T], / +) -> _Dependency[T]: ... @overload def dependency[TFuru: Furu[Any], T]( - *, cached: bool -) -> Callable[ - [Callable[[TFuru], T]], _CachedDependency[T] | _UncachedDependency[T] -]: ... + *, recheck_interval: RecheckInterval = "never" +) -> Callable[[Callable[[TFuru], T]], _Dependency[T]]: ... def dependency[TFuru: Furu[Any], T]( - func: Callable[[TFuru], T] | None = None, /, *, cached: bool = True -) -> ( - _CachedDependency[T] - | _UncachedDependency[T] - | Callable[[Callable[[TFuru], T]], _CachedDependency[T] | _UncachedDependency[T]] -): - def decorate( - func: Callable[[TFuru], T], - ) -> _CachedDependency[T] | _UncachedDependency[T]: - if cached: - return _CachedDependency(func) - return _UncachedDependency(func) + func: Callable[[TFuru], T] | None = None, + /, + *, + recheck_interval: RecheckInterval = "never", +) -> _Dependency[T] | Callable[[Callable[[TFuru], T]], _Dependency[T]]: + def decorate(func: Callable[[TFuru], T]) -> _Dependency[T]: + return _Dependency(func, recheck_interval=recheck_interval) if func is not None: return decorate(func) diff --git a/src/furu/execution/manager.py b/src/furu/execution/manager.py index c3e5be80..9a4b815f 100644 --- a/src/furu/execution/manager.py +++ b/src/furu/execution/manager.py @@ -12,7 +12,12 @@ from furu._storage_layout import manager_log_path_in from furu.config import get_config from furu.core import Furu -from furu.dag import DagNode, _add_to_dag, _update_dag_blocking_dependencies +from furu.dag import ( + DagNode, + _add_to_dag, + _refresh_dag_declared_dependencies, + _update_dag_blocking_dependencies, +) from furu.logging import _scoped_log_files, get_logger from furu.metadata import ArtifactSpec from furu.resources import ResourceRequest, resource_request_satisfies @@ -53,11 +58,13 @@ def __init__(self, objs: Sequence[Furu]) -> None: self.running: dict[str, RunningJob] = {} self.completed: dict[str, DagNode] = {} self.failed: dict[str, FailedJob] = {} + self.root_ids: set[str] = set() self.lock = threading.Lock() self.done = threading.Event() self._finish_error: str | None = None _add_to_dag(self, objs) + self.root_ids = {obj.object_id for obj in objs} digest = hashlib.blake2s(digest_size=16) for obj in objs: @@ -90,6 +97,7 @@ def log_context(self) -> Iterator[None]: def lease_job(self, *, resources: ResourceRequest) -> LeaseJobResponse: with self.log_context(), self.lock: + _refresh_dag_declared_dependencies(self) self._maybe_finish_locked() if self.done.is_set(): return "stop" @@ -133,6 +141,11 @@ def count_satisfiable_jobs( self, *, resources: ResourceRequest, max_workers: int ) -> int: with self.lock: + _refresh_dag_declared_dependencies(self) + self._maybe_finish_locked() + if self.done.is_set(): + return 0 + count = 0 for node in self.ready.values(): if resource_request_satisfies( @@ -149,9 +162,14 @@ def job_result(self, lease_id: str, request: JobResultRequest) -> None: match request: case JobCompletedResult(): self.completed[running_job.node.obj.object_id] = running_job.node + completed_id = running_job.node.obj.object_id for dependent in tuple(running_job.node.dependents): if running_job.node in dependent.dependencies: dependent.dependencies.remove(running_job.node) + if dependent in running_job.node.dependents: + running_job.node.dependents.remove(dependent) + dependent.declared_dependency_ids.discard(completed_id) + dependent.runtime_dependency_ids.discard(completed_id) dependent_id = dependent.obj.object_id if not dependent.dependencies and dependent_id in self.blocked: @@ -196,6 +214,7 @@ def job_result(self, lease_id: str, request: JobResultRequest) -> None: ) case _: assert_never(request) + _refresh_dag_declared_dependencies(self) self._maybe_finish_locked() def raise_for_failure(self) -> None: diff --git a/tests/test_core.py b/tests/test_core.py index d5045c8c..47745e04 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -13,6 +13,7 @@ from pydantic import BaseModel, ConfigDict import furu +import furu.dependencies as dependencies_module import furu.execution as execution_module from furu import Furu, ResourceRequirements, load_or_create, validate from furu.config import get_config @@ -389,20 +390,22 @@ def create(self) -> str: class ComputedDependencyParent(Furu[str]): name: str + calls: ClassVar[int] = 0 @furu.dependency def child(self) -> Node: + type(self).calls += 1 return Node(name=self.name) def create(self) -> str: return self.child.load_or_create() -class ExplicitCachedDependencyParent(Furu[str]): +class ExplicitNeverDependencyParent(Furu[str]): name: str calls: ClassVar[int] = 0 - @furu.dependency(cached=True) + @furu.dependency(recheck_interval="never") def child(self) -> Node: type(self).calls += 1 return Node(name=f"{self.name}-{type(self).calls}") @@ -411,11 +414,11 @@ def create(self) -> str: return self.child.load_or_create() -class UncachedDependencyParent(Furu[str]): +class ZeroIntervalDependencyParent(Furu[str]): name: str calls: ClassVar[int] = 0 - @furu.dependency(cached=False) + @furu.dependency(recheck_interval=0) def child(self) -> Node: type(self).calls += 1 return Node(name=f"{self.name}-{type(self).calls}") @@ -1040,28 +1043,50 @@ def test_field_dependencies_are_eager_but_metadata_stores_only_loaded_objects() assert _dependency_object_ids(parent) == [first.object_id] -def test_computed_dependency_is_cached_property_and_eager_loaded_dependency() -> None: +def test_default_dependency_is_cached_forever_and_eager_loaded_dependency() -> None: + ComputedDependencyParent.calls = 0 parent = ComputedDependencyParent(name="computed") assert parent.child is parent.child assert collect_declared_refs(parent) == (parent.child,) assert parent.load_or_create() == "Node(computed)" + assert ComputedDependencyParent.calls == 1 assert _dependency_object_ids(parent) == [parent.child.object_id] -def test_dependency_accepts_explicit_cached_true() -> None: - ExplicitCachedDependencyParent.calls = 0 - parent = ExplicitCachedDependencyParent(name="explicit-cached") +def test_dependency_call_syntax_defaults_to_cached_forever() -> None: + class CallSyntaxDependencyParent(Furu[str]): + name: str + calls: ClassVar[int] = 0 + + @furu.dependency() + def child(self) -> Node: + type(self).calls += 1 + return Node(name=f"{self.name}-{type(self).calls}") + + def create(self) -> str: + return self.child.load_or_create() + + parent = CallSyntaxDependencyParent(name="call-syntax") + + assert parent.child is parent.child + assert collect_declared_refs(parent) == (parent.child,) + assert CallSyntaxDependencyParent.calls == 1 + + +def test_dependency_explicit_never_interval_is_cached_forever() -> None: + ExplicitNeverDependencyParent.calls = 0 + parent = ExplicitNeverDependencyParent(name="explicit-never") assert parent.child is parent.child assert collect_declared_refs(parent) == (parent.child,) - assert ExplicitCachedDependencyParent.calls == 1 + assert ExplicitNeverDependencyParent.calls == 1 -def test_dependency_can_be_uncached_property() -> None: - UncachedDependencyParent.calls = 0 - parent = UncachedDependencyParent(name="uncached") +def test_dependency_zero_interval_recomputes_every_access() -> None: + ZeroIntervalDependencyParent.calls = 0 + parent = ZeroIntervalDependencyParent(name="zero") first = parent.child second = parent.child @@ -1069,8 +1094,53 @@ def test_dependency_can_be_uncached_property() -> None: assert first.object_id != second.object_id declared_refs = collect_declared_refs(parent) assert len(declared_refs) == 1 - assert declared_refs[0].object_id == Node(name="uncached-3").object_id - assert UncachedDependencyParent.calls == 3 + assert declared_refs[0].object_id == Node(name="zero-3").object_id + assert ZeroIntervalDependencyParent.calls == 3 + + +def test_dependency_positive_interval_uses_monotonic_ttl( + monkeypatch: pytest.MonkeyPatch, +) -> None: + now = 100.0 + + def monotonic() -> float: + return now + + class TtlDependencyParent(Furu[str]): + name: str + calls: ClassVar[int] = 0 + + @furu.dependency(recheck_interval=60) + def child(self) -> Node: + type(self).calls += 1 + return Node(name=f"{self.name}-{type(self).calls}") + + def create(self) -> str: + return self.child.load_or_create() + + monkeypatch.setattr(dependencies_module.time, "monotonic", monotonic) + parent = TtlDependencyParent(name="ttl") + + first = parent.child + now = 159.0 + assert parent.child is first + assert TtlDependencyParent.calls == 1 + + now = 160.0 + second = parent.child + assert second is not first + assert second.object_id == Node(name="ttl-2").object_id + assert TtlDependencyParent.calls == 2 + + +def test_dependency_rejects_negative_recheck_interval() -> None: + with pytest.raises(ValueError, match="non-negative"): + furu.dependency(recheck_interval=-1)(lambda self: None) + + +def test_dependency_rejects_bool_recheck_interval() -> None: + with pytest.raises(TypeError, match="non-negative integer or 'never'"): + furu.dependency(recheck_interval=True)(lambda self: None) def test_load_or_create_inside_create_is_recorded_and_deduped() -> None: diff --git a/tests/test_dag.py b/tests/test_dag.py index 968f61a4..20cb9dc1 100644 --- a/tests/test_dag.py +++ b/tests/test_dag.py @@ -5,8 +5,9 @@ import furu from furu import Furu -from furu.dag import DagNode +from furu.dag import DagNode, _update_dag_blocking_dependencies from furu.execution.manager import Manager +from furu.metadata import ArtifactSpec from furu._storage_layout import ( compute_lock_path_in, run_log_path_in, @@ -77,6 +78,8 @@ def test_add_to_dag_single_object_no_dependencies(): assert root.obj is leaf assert root.dependencies == [] assert root.dependents == [] + assert root.declared_dependency_ids == set() + assert root.runtime_dependency_ids == set() assert manager.nodes_by_id == {leaf.object_id: root} assert manager.blocked == {} @@ -112,12 +115,21 @@ def test_add_to_dag_traverses_declared_refs_recursively(): assert leaf_a_node.dependencies == [] assert leaf_a_node.dependents == [mid_left_node] + assert leaf_a_node.declared_dependency_ids == set() + assert leaf_a_node.runtime_dependency_ids == set() assert mid_left_node.dependencies == [leaf_a_node] assert mid_left_node.dependents == [top_node] + assert mid_left_node.declared_dependency_ids == {leaf_a.object_id} + assert mid_left_node.runtime_dependency_ids == set() assert {dep.obj.object_id for dep in top_node.dependencies} == { mid_left.object_id, mid_right.object_id, } + assert top_node.declared_dependency_ids == { + mid_left.object_id, + mid_right.object_id, + } + assert top_node.runtime_dependency_ids == set() assert top_node.dependents == [] @@ -240,12 +252,36 @@ def test_add_to_dag_walks_computed_dependencies(): (child_root,) = manager.ready.values() assert child_root.obj.object_id == parent.computed_child.object_id assert {n.obj.object_id for n in child_root.dependents} == {parent.object_id} + parent_node = manager.nodes_by_id[parent.object_id] + assert parent_node.declared_dependency_ids == {parent.computed_child.object_id} + assert parent_node.runtime_dependency_ids == set() assert set(manager.nodes_by_id) == { parent.object_id, parent.computed_child.object_id, } +def test_update_dag_blocking_dependencies_tracks_runtime_source_separately(): + declared = Leaf(name="declared") + runtime = Leaf(name="runtime") + parent = Mid(label="m", child=declared) + manager = Manager([parent]) + parent_node = manager.nodes_by_id[parent.object_id] + + _update_dag_blocking_dependencies( + manager, + parent_node, + [ArtifactSpec.from_furu(runtime)], + ) + + assert parent_node.declared_dependency_ids == {declared.object_id} + assert parent_node.runtime_dependency_ids == {runtime.object_id} + assert {node.obj.object_id for node in parent_node.dependencies} == { + declared.object_id, + runtime.object_id, + } + + def test_add_to_dag_rejects_non_furu_values(): with pytest.raises(TypeError, match="expected Furu objects"): Manager([Leaf(name="ok"), "not-a-furu"]) # ty: ignore[invalid-argument-type] diff --git a/tests/test_worker_manager.py b/tests/test_worker_manager.py index 571fe9e4..f0694b50 100644 --- a/tests/test_worker_manager.py +++ b/tests/test_worker_manager.py @@ -8,6 +8,7 @@ from fastapi.testclient import TestClient from pydantic import TypeAdapter, ValidationError +import furu import furu.worker.loop as worker_loop_module from furu import Furu from furu.config import get_config @@ -90,6 +91,18 @@ def create(self) -> int: return ManagerLeaf(value=self.value).load_or_create() + 1 +class DynamicDeclaredParent(Furu[list[int]]): + prefix: str + count: ClassVar[int] = 0 + + @furu.dependency(recheck_interval=0) + def children(self) -> list[ManagerLeaf]: + return [ManagerLeaf(value=i) for i in range(type(self).count)] + + def create(self) -> list[int]: + return [child.load_or_create() for child in self.children] + + def test_manager_init_partitions_ready_and_blocked() -> None: leaf = ManagerLeaf(value=1) parent = ManagerParent(child=leaf) @@ -180,6 +193,119 @@ def test_manager_job_result_blocked_discovers_lazy_dependency_and_reruns_parent( assert manager.blocked == {} +def test_dynamic_declared_refresh_adds_new_dependency_before_parent_runs() -> None: + DynamicDeclaredParent.count = 1 + parent = DynamicDeclaredParent(prefix="adds") + child_0 = ManagerLeaf(value=0) + child_1 = ManagerLeaf(value=1) + manager = Manager([parent]) + + assert set(manager.ready) == {child_0.object_id} + assert set(manager.blocked) == {parent.object_id} + + child_0_job = manager.lease_job(resources=ANY_RESOURCES) + assert isinstance(child_0_job, Job) + assert child_0_job.artifact.object_id == child_0.object_id + manager.job_result(child_0_job.lease_id, JobCompletedResult()) + + DynamicDeclaredParent.count = 2 + child_1_job = manager.lease_job(resources=ANY_RESOURCES) + + assert isinstance(child_1_job, Job) + assert child_1_job.artifact.object_id == child_1.object_id + assert set(manager.blocked) == {parent.object_id} + assert parent.object_id not in manager.ready + + manager.job_result(child_1_job.lease_id, JobCompletedResult()) + parent_job = manager.lease_job(resources=ANY_RESOURCES) + assert isinstance(parent_job, Job) + assert parent_job.artifact.object_id == parent.object_id + + +def test_dynamic_declared_refresh_removes_obsolete_orphan_dependency() -> None: + DynamicDeclaredParent.count = 2 + parent = DynamicDeclaredParent(prefix="removes") + child_0 = ManagerLeaf(value=0) + child_1 = ManagerLeaf(value=1) + manager = Manager([parent]) + + DynamicDeclaredParent.count = 1 + assert manager.count_satisfiable_jobs(resources=ANY_RESOURCES, max_workers=10) == 1 + + parent_node = manager.nodes_by_id[parent.object_id] + assert parent_node.declared_dependency_ids == {child_0.object_id} + assert {node.obj.object_id for node in parent_node.dependencies} == { + child_0.object_id + } + assert child_1.object_id not in manager.nodes_by_id + assert child_1.object_id not in manager.ready + assert child_1.object_id not in manager.blocked + + +def test_dynamic_declared_refresh_preserves_runtime_dependency() -> None: + parent = ManagerLazyParent(value=20) + dependency = ManagerLeaf(value=20) + manager = Manager([parent]) + + parent_job = manager.lease_job(resources=ANY_RESOURCES) + assert isinstance(parent_job, Job) + manager.job_result( + parent_job.lease_id, + JobBlockedResult(dependencies=[ArtifactSpec.from_furu(dependency)]), + ) + + assert manager.count_satisfiable_jobs(resources=ANY_RESOURCES, max_workers=10) == 1 + assert set(manager.blocked) == {parent.object_id} + assert set(manager.ready) == {dependency.object_id} + + dependency_job = manager.lease_job(resources=ANY_RESOURCES) + assert isinstance(dependency_job, Job) + assert dependency_job.artifact.object_id == dependency.object_id + + +def test_dynamic_declared_refresh_does_not_prune_roots() -> None: + DynamicDeclaredParent.count = 2 + parent = DynamicDeclaredParent(prefix="keeps-root") + child_0 = ManagerLeaf(value=0) + child_1 = ManagerLeaf(value=1) + manager = Manager([parent, child_1]) + + DynamicDeclaredParent.count = 1 + manager.count_satisfiable_jobs(resources=ANY_RESOURCES, max_workers=10) + + parent_node = manager.nodes_by_id[parent.object_id] + child_1_node = manager.nodes_by_id[child_1.object_id] + assert child_1.object_id not in parent_node.declared_dependency_ids + assert child_1_node not in parent_node.dependencies + assert child_1.object_id in manager.nodes_by_id + assert child_1.object_id in manager.ready + assert parent_node.declared_dependency_ids == {child_0.object_id} + + +def test_dynamic_declared_refresh_does_not_prune_running_nodes() -> None: + DynamicDeclaredParent.count = 2 + parent = DynamicDeclaredParent(prefix="keeps-running") + manager = Manager([parent]) + + running_child_job = manager.lease_job(resources=ANY_RESOURCES) + assert isinstance(running_child_job, Job) + running_child_id = running_child_job.artifact.object_id + + DynamicDeclaredParent.count = 0 + manager.count_satisfiable_jobs(resources=ANY_RESOURCES, max_workers=10) + + parent_node = manager.nodes_by_id[parent.object_id] + assert parent_node.declared_dependency_ids == set() + assert running_child_id in manager.nodes_by_id + assert running_child_job.lease_id in manager.running + assert all( + node.obj.object_id != running_child_id for node in parent_node.dependencies + ) + + manager.job_result(running_child_job.lease_id, JobCompletedResult()) + assert running_child_job.lease_id not in manager.running + + def test_manager_job_result_blocked_ignores_completed_lazy_dependency() -> None: parent = ManagerLazyParent(value=2) dependency = ManagerLeaf(value=2)