stage: cache: catch CollectCacheError (#9677)

repro: remove on_unchaged kwarg and _repro_callback

dvc/repo/experiments/executor/base.py

@@ -459,25 +459,16 @@ class BaseExecutor(ABC):
             should force overwrite any existing duplicates.
         """
         from dvc.repo.checkout import checkout as dvc_checkout
-        from dvc.repo.reproduce import reproduce as dvc_reproduce
-        from dvc.stage import PipelineStage
         from dvc.ui import ui
 
         auto_push = env2bool(DVC_EXP_AUTO_PUSH)
         git_remote = os.getenv(DVC_EXP_GIT_REMOTE, None)
 
-        unchanged = []
-
         if queue is not None:
             queue.put((rev, os.getpid()))
         if log_errors and log_level is not None:
             cls._set_log_level(log_level)
 
-        def filter_pipeline(stages):
-            unchanged.extend(
-                [stage for stage in stages if isinstance(stage, PipelineStage)]
-            )
-
         exp_hash: Optional[str] = None
         exp_ref: Optional["ExpRefInfo"] = None
         repro_force: bool = False
@@ -519,12 +510,7 @@ class BaseExecutor(ABC):
                     recursive=kwargs.get("recursive", False),
                 )
 
-            stages = dvc_reproduce(
-                dvc,
-                *args,
-                on_unchanged=filter_pipeline,
-                **kwargs,
-            )
+            stages = dvc.reproduce(*args, **kwargs)
             if paths := cls._get_top_level_paths(dvc):
                 logger.debug("Staging top-level files: %s", paths)
                 dvc.scm_context.add(paths)

dvc/repo/reproduce.py

@@ -122,7 +122,6 @@ def _reproduce_stages(  # noqa: C901
     stages,
     downstream=False,
     single_item=False,
-    on_unchanged=None,
     **kwargs,
 ):
     r"""Derive the evaluation of the given node for the given graph.
@@ -175,24 +174,20 @@ def _reproduce_stages(  # noqa: C901
         try:
             kwargs["upstream"] = result + unchanged
             ret = _reproduce_stage(stage, **kwargs)
-
-            if len(ret) == 0:
+            if not ret:
                 unchanged.extend([stage])
-            elif force_downstream:
+                continue
+
+            result.extend(ret)
+            if force_downstream:
                 # NOTE: we are walking our pipeline from the top to the
                 # bottom. If one stage is changed, it will be reproduced,
                 # which tells us that we should force reproducing all of
                 # the other stages down below, even if their direct
                 # dependencies didn't change.
                 kwargs["force"] = True
-
-            if ret:
-                result.extend(ret)
         except Exception as exc:  # noqa: BLE001
             raise ReproductionError(stage.addressing) from exc
-
-    if on_unchanged is not None:
-        on_unchanged(unchanged)
     return result
 
 
@@ -233,7 +228,3 @@ def _get_steps(graph, stages, downstream, single_item):
             steps.append(stage)
 
     return steps
-
-
-def _repro_callback(experiments_callback, unchanged, stages):
-    experiments_callback(unchanged, stages)

dvc/stage/cache.py

@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, List, Optional, Tuple
 from funcy import first
 
 from dvc import fs
-from dvc.exceptions import DvcException
+from dvc.exceptions import CollectCacheError, DvcException
 from dvc.utils import dict_sha256, relpath
 
 if TYPE_CHECKING:
@@ -179,7 +179,7 @@ class StageCache:
         dump_yaml(tmp, cache)
         self.repo.cache.legacy.move(tmp, path)
 
-    def restore(self, stage, run_cache=True, pull=False, dry=False):
+    def restore(self, stage, run_cache=True, pull=False, dry=False):  # noqa: C901
         from .serialize import to_single_stage_lockfile
 
         if not _can_hash(stage):
@@ -203,8 +203,11 @@ class StageCache:
         cached_stage = self._create_stage(cache, wdir=stage.wdir)
 
         if pull and not dry:
-            for objs in cached_stage.get_used_objs().values():
-                self.repo.cloud.pull(objs)
+            try:
+                for objs in cached_stage.get_used_objs().values():
+                    self.repo.cloud.pull(objs)
+            except CollectCacheError as exc:
+                raise RunCacheNotFoundError(stage) from exc
 
         if not cached_stage.outs_cached():
             raise RunCacheNotFoundError(stage)