stage: cache: catch CollectCacheError (#9677)

repro: remove on_unchaged kwarg and _repro_callback

.dvc/.gitignore

@@ -0,0 +1,3 @@
+/config.local
+/tmp
+/cache

dvc/output.py

@@ -20,6 +20,7 @@ from dvc.exceptions import (
     DvcException,
     MergeError,
 )
+from dvc.utils import format_link
 from dvc.utils.objects import cached_property
 from dvc_data.hashfile import check as ocheck
 from dvc_data.hashfile import load as oload
@@ -678,6 +679,14 @@ class Output:
     # pylint: enable=no-member
 
     def save(self) -> None:
+        if self.use_cache and not self.is_in_repo:
+            raise DvcException(
+                f"Saving cached external output {str(self)} is not supported "
+                "since DVC 3.0. See "
+                f"{format_link('https://dvc.org/doc/user-guide/upgrade')} "
+                "for more info."
+            )
+
         if not self.exists:
             raise self.DoesNotExistError(self)
 
@@ -1270,7 +1279,7 @@ class Output:
             raise FileNotFoundError(  # noqa: B904
                 errno.ENOENT,
                 os.strerror(errno.ENOENT),
-                self.fs.path.relpath(path),
+                path,
             )
 
         new = tree.from_trie(trie)
@@ -1393,8 +1402,7 @@ class Output:
         otransfer(staging, self.cache, {obj.hash_info}, hardlink=relink, shallow=False)
 
         if relink:
-            rel = self.fs.path.relpath(path)
-            with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as callback:
+            with CheckoutCallback(desc="Checking out {path}", unit="files") as callback:
                 self._checkout(
                     path,
                     self.fs,

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)