Repository: iterative/dvc
Pull request #9712 (2023-07-07 16:18:04)

repro: introduce --keep-going/--ignore to continue on error
`-k`/`--keep-going` will skip all dependents of a stage if it fails. It'll still try to reproduce the other dependencies of the pending targets, before it exits with a non-zero status.

`--ignore-errors` will treat any error during stage execution as a success, and will continue executing other dependencies as if it succeeded. Exits with 0 even on failure.

Also fixes:
* --force-downstream used to force everything even if one upstream nodes changed. This used to work when we supported a single targets, but there could be independent pipelines, so  we should not force everything if any node changes. Now we'll force reproduce the stage only if one of its dependency changed upstream or it's upstream node was also forced.

* Makes `--downstream` more correct, in relation to `--allow-missing` and `--pull`. Now we use an active graph for downstream/upstream logic instead of a subgraph. Note that this does not affect the downstream logic though, as dvc always goes downstream, so subgraph was no different than active graph. The subgraph only did not have the upstream nodes.


Fixes https://github.com/iterative/dvc/issues/8647 and fixes https://github.com/iterative/dvc/issues/7292.



Patch:
diff --git a/dvc/api/data.py b/dvc/api/data.py
index 51c5437b7..f5f8a6ed0 100644
--- a/dvc/api/data.py
+++ b/dvc/api/data.py
@@ -45,22 +45,13 @@ def get_url(path, repo=None, rev=None, remote=None):
     with Repo.open(
         repo, rev=rev, subrepos=True, uninitialized=True, **repo_kwargs
     ) as _repo:
-        with _wrap_exceptions(_repo, path):
-            fs_path = _repo.dvcfs.from_os_path(path)
-            fs = _repo.dvcfs.fs
-            # pylint: disable-next=protected-access
-            key = fs._get_key_from_relative(fs_path)
-            # pylint: disable-next=protected-access
-            subrepo, _, subkey = fs._get_subrepo_info(key)
-            index = subrepo.index.data["repo"]
-            with reraise(KeyError, OutputNotFoundError(path, repo)):
-                entry = index[subkey]
-            with reraise(
-                (StorageKeyError, ValueError),
-                NoRemoteError(f"no remote specified in {_repo}"),
-            ):
-                remote_fs, remote_path = index.storage_map.get_remote(entry)
-                return remote_fs.unstrip_protocol(remote_path)
+        index, entry = _repo.get_data_index_entry(path)
+        with reraise(
+            (StorageKeyError, ValueError),
+            NoRemoteError(f"no remote specified in {_repo}"),
+        ):
+            remote_fs, remote_path = index.storage_map.get_remote(entry)
+            return remote_fs.unstrip_protocol(remote_path)
 
 
 class _OpenContextManager(GCM):
diff --git a/dvc/commands/repro.py b/dvc/commands/repro.py
index f5fb50fff..6a448142b 100644
--- a/dvc/commands/repro.py
+++ b/dvc/commands/repro.py
@@ -33,6 +33,7 @@ class CmdRepro(CmdBase):
             "force_downstream": self.args.force_downstream,
             "pull": self.args.pull,
             "allow_missing": self.args.allow_missing,
+            "on_error": self.args.on_error,
         }
 
     @property
@@ -137,6 +138,26 @@ and then the stage name name.
             "Only print the commands that would be executed without actually executing."
         ),
     )
+    repro_parser.add_argument(
+        "-k",
+        "--keep-going",
+        action="store_const",
+        default="fail",
+        const="keep-going",
+        dest="on_error",
+        help=(
+            "Continue executing, skipping stages having dependencies "
+            "on the failed stages"
+        ),
+    )
+    repro_parser.add_argument(
+        "--ignore-errors",
+        action="store_const",
+        default="fail",
+        const="ignore",
+        dest="on_error",
+        help="Ignore errors from stages.",
+    )
 
 
 def add_parser(subparsers, parent_parser):
diff --git a/dvc/exceptions.py b/dvc/exceptions.py
index 1ca14ec48..618eeaef4 100644
--- a/dvc/exceptions.py
+++ b/dvc/exceptions.py
@@ -167,9 +167,7 @@ class InitError(DvcException):
 
 
 class ReproductionError(DvcException):
-    def __init__(self, name):
-        self.name = name
-        super().__init__(f"failed to reproduce '{name}'")
+    pass
 
 
 class BadMetricError(DvcException):
diff --git a/dvc/fs/dvc.py b/dvc/fs/dvc.py
index 8117b019f..4ba009928 100644
--- a/dvc/fs/dvc.py
+++ b/dvc/fs/dvc.py
@@ -412,7 +412,7 @@ class DVCFileSystem(FileSystem):
 
     @functools.cached_property
     # pylint: disable-next=invalid-overridden-method
-    def fs(self) -> "DVCFileSystem":
+    def fs(self) -> "_DVCFileSystem":
         return _DVCFileSystem(**self.fs_args)
 
     def isdvc(self, path, **kwargs) -> bool:
diff --git a/dvc/parsing/__init__.py b/dvc/parsing/__init__.py
index 6d3f47ab0..fce30b6af 100644
--- a/dvc/parsing/__init__.py
+++ b/dvc/parsing/__init__.py
@@ -135,6 +135,7 @@ def make_definition(
 class DataResolver:
     def __init__(self, repo: "Repo", wdir: str, d: dict):
         self.fs = fs = repo.fs
+        self.parsing_config = repo.config.get("parsing", {})
 
         if os.path.isabs(wdir):
             wdir = fs.path.relpath(wdir)
@@ -301,7 +302,10 @@ class EntryDefinition:
     ) -> "DictStrAny":
         try:
             return context.resolve(
-                value, skip_interpolation_checks=skip_checks, key=key
+                value,
+                skip_interpolation_checks=skip_checks,
+                key=key,
+                config=self.resolver.parsing_config,
             )
         except (ParseError, KeyNotInContext) as exc:
             format_and_raise(exc, f"'{self.where}.{self.name}.{key}'", self.relpath)
diff --git a/dvc/parsing/context.py b/dvc/parsing/context.py
index 221c5feab..6484f0829 100644
--- a/dvc/parsing/context.py
+++ b/dvc/parsing/context.py
@@ -495,7 +495,12 @@ class Context(CtxDict):
                 self.data.pop(key, None)
 
     def resolve(
-        self, src, unwrap=True, skip_interpolation_checks=False, key=None
+        self,
+        src,
+        unwrap=True,
+        skip_interpolation_checks=False,
+        key=None,
+        config=None,
     ) -> Any:
         """Recursively resolves interpolation and returns resolved data.
 
@@ -511,10 +516,15 @@ class Context(CtxDict):
         {'lst': [1, 2, 3]}
         """
         func = recurse(self.resolve_str)
-        return func(src, unwrap, skip_interpolation_checks, key)
+        return func(src, unwrap, skip_interpolation_checks, key, config)
 
     def resolve_str(
-        self, src: str, unwrap=True, skip_interpolation_checks=False, key=None
+        self,
+        src: str,
+        unwrap=True,
+        skip_interpolation_checks=False,
+        key=None,
+        config=None,
     ) -> str:
         """Resolves interpolated string to it's original value,
         or in case of multiple interpolations, a combined string.
@@ -535,7 +545,12 @@ class Context(CtxDict):
             return value
         # but not "${num} days"
         return str_interpolate(
-            src, matches, self, skip_checks=skip_interpolation_checks, key=key
+            src,
+            matches,
+            self,
+            skip_checks=skip_interpolation_checks,
+            key=key,
+            config=config,
         )
 
 
diff --git a/dvc/parsing/interpolate.py b/dvc/parsing/interpolate.py
index a423c0a1b..4b9b0686c 100644
--- a/dvc/parsing/interpolate.py
+++ b/dvc/parsing/interpolate.py
@@ -82,20 +82,18 @@ def escape_str(value):
 
 
 @singledispatch
-def to_str(obj) -> str:
+def to_str(obj, config=None) -> str:  # noqa: ARG001, pylint: disable=unused-argument
     return str(obj)
 
 
 @to_str.register(bool)
-def _(obj: bool):
+def _(obj: bool, config=None):  # noqa: ARG001, pylint: disable=unused-argument
     return "true" if obj else "false"
 
 
 @to_str.register(dict)
-def _(obj: dict):  # noqa: C901
-    from dvc.config import Config
-
-    config = Config.from_cwd().get("parsing", {})
+def _(obj: dict, config=None):  # noqa: C901
+    config = config or {}
 
     result = ""
     for k, v in flatten(obj).items():
@@ -211,6 +209,7 @@ def str_interpolate(
     context: "Context",
     skip_checks: bool = False,
     key=None,
+    config=None,
 ):
     index, buf = 0, ""
     for match in matches:
@@ -218,7 +217,7 @@ def str_interpolate(
         expr = get_expression(match, skip_checks=skip_checks)
         value = context.select(expr, unwrap=True)
         validate_value(value, key)
-        buf += template[index:start] + to_str(value)
+        buf += template[index:start] + to_str(value, config=config)
         index = end
     buf += template[index:]
     # regex already backtracks and avoids any `${` starting with
diff --git a/dvc/render/match.py b/dvc/render/match.py
index 3f36e9adc..86eba55c3 100644
--- a/dvc/render/match.py
+++ b/dvc/render/match.py
@@ -1,3 +1,4 @@
+import logging
 import os
 from collections import defaultdict
 from typing import TYPE_CHECKING, DefaultDict, Dict, List, NamedTuple, Optional
@@ -17,6 +18,7 @@ if TYPE_CHECKING:
 
 
 dpath.options.ALLOW_EMPTY_STRING_KEYS = True
+logger = logging.getLogger(__name__)
 
 
 def _squash_plots_properties(data: List) -> Dict:
@@ -113,6 +115,7 @@ def match_defs_renderers(  # noqa: C901, PLR0912
             try:
                 dps, rev_props = converter.flat_datapoints(rev)
             except Exception as e:  # noqa: BLE001, pylint: disable=broad-except
+                logger.warning("In %r, %s", rev, str(e).lower())
                 def_errors[rev] = e
                 continue
 
diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py
index 544fd1f69..66de6e52f 100644
--- a/dvc/repo/__init__.py
+++ b/dvc/repo/__init__.py
@@ -29,7 +29,7 @@ if TYPE_CHECKING:
     from dvc.stage import Stage
     from dvc.types import DictStrAny
     from dvc_data.hashfile.state import StateBase
-    from dvc_data.index import DataIndex
+    from dvc_data.index import DataIndex, DataIndexEntry
 
     from .experiments import Experiments
     from .index import Index
@@ -285,7 +285,7 @@ class Repo:
             new.check_graph()
 
     @staticmethod
-    def open(url, *args, **kwargs):  # noqa: A003
+    def open(url: Optional[str], *args, **kwargs) -> "Repo":  # noqa: A003
         from .open_repo import open_repo
 
         return open_repo(url, *args, **kwargs)
@@ -375,6 +375,28 @@ class Repo:
         self.data_index.commit()
         self._reset()
 
+    def get_data_index_entry(
+        self,
+        path: str,
+        workspace: str = "repo",
+    ) -> Tuple["DataIndex", "DataIndexEntry"]:
+        if self.subrepos:
+            fs_path = self.dvcfs.from_os_path(path)
+            fs = self.dvcfs.fs
+            # pylint: disable-next=protected-access
+            key = fs._get_key_from_relative(fs_path)
+            # pylint: disable-next=protected-access
+            subrepo, _, key = fs._get_subrepo_info(key)
+            index = subrepo.index.data[workspace]
+        else:
+            index = self.index.data[workspace]
+            key = self.fs.path.relparts(path, self.root_dir)
+
+        try:
+            return index, index[key]
+        except KeyError as exc:
+            raise OutputNotFoundError(path, self) from exc
+
     def __repr__(self):
         return f"{self.__class__.__name__}: '{self.root_dir}'"
 
diff --git a/dvc/repo/open_repo.py b/dvc/repo/open_repo.py
index 8a0901166..1b5e0bbeb 100644
--- a/dvc/repo/open_repo.py
+++ b/dvc/repo/open_repo.py
@@ -2,8 +2,7 @@ import logging
 import os
 import tempfile
 import threading
-from contextlib import contextmanager
-from typing import TYPE_CHECKING, Dict, Iterator, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, Optional, Tuple
 
 from funcy import retry, wrap_with
 
@@ -18,16 +17,14 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-@contextmanager
 @map_scm_exception()
 def _external_repo(
     url,
     rev: Optional[str] = None,
-    for_write: bool = False,
     **kwargs,
-) -> Iterator["Repo"]:
+) -> "Repo":
     logger.debug("Creating external repo %s@%s", url, rev)
-    path = _cached_clone(url, rev, for_write=for_write)
+    path = _cached_clone(url, rev)
     # Local HEAD points to the tip of whatever branch we first cloned from
     # (which may not be the default branch), use origin/HEAD here to get
     # the tip of the default branch
@@ -38,11 +35,6 @@ def _external_repo(
     config.update(kwargs.pop("config", None) or {})
 
     main_root = "/"
-    if for_write:
-        # we already checked out needed revision
-        rev = None
-        main_root = path
-
     repo_kwargs = dict(
         root_dir=path,
         url=url,
@@ -52,14 +44,7 @@ def _external_repo(
         **kwargs,
     )
 
-    repo = Repo(**repo_kwargs)
-
-    try:
-        yield repo
-    finally:
-        repo.close()
-        if for_write:
-            _remove(path)
+    return Repo(**repo_kwargs)
 
 
 def open_repo(url, *args, **kwargs):
@@ -140,20 +125,19 @@ def _get_remote_config(url):
         repo.close()
 
 
-def _cached_clone(url, rev, for_write=False):
+def _cached_clone(url, rev):
     """Clone an external git repo to a temporary directory.
 
     Returns the path to a local temporary directory with the specified
-    revision checked out. If for_write is set prevents reusing this dir via
-    cache.
+    revision checked out.
     """
     from shutil import copytree
 
     # even if we have already cloned this repo, we may need to
     # fetch/fast-forward to get specified rev
-    clone_path, shallow = _clone_default_branch(url, rev, for_write=for_write)
+    clone_path, shallow = _clone_default_branch(url, rev)
 
-    if not for_write and (url) in CLONES:
+    if url in CLONES:
         return CLONES[url][0]
 
     # Copy to a new dir to keep the clone clean
@@ -161,16 +145,12 @@ def _cached_clone(url, rev, for_write=False):
     logger.debug("erepo: making a copy of %s clone", url)
     copytree(clone_path, repo_path)
 
-    # Check out the specified revision
-    if for_write:
-        _git_checkout(repo_path, rev)
-    else:
-        CLONES[url] = (repo_path, shallow)
+    CLONES[url] = (repo_path, shallow)
     return repo_path
 
 
 @wrap_with(threading.Lock())
-def _clone_default_branch(url, rev, for_write=False):  # noqa: C901, PLR0912
+def _clone_default_branch(url, rev):  # noqa: C901, PLR0912
     """Get or create a clean clone of the url.
 
     The cloned is reactualized with git pull unless rev is a known sha.
@@ -204,7 +184,7 @@ def _clone_default_branch(url, rev, for_write=False):  # noqa: C901, PLR0912
 
             logger.debug("erepo: git clone '%s' to a temporary dir", url)
             clone_path = tempfile.mkdtemp("dvc-clone")
-            if not for_write and rev and not Git.is_sha(rev):
+            if rev and not Git.is_sha(rev):
                 # If rev is a tag or branch name try shallow clone first
 
                 try:
@@ -249,17 +229,6 @@ def _merge_upstream(git: "Git"):
         pass
 
 
-def _git_checkout(repo_path, rev):
-    from dvc.scm import Git
-
-    logger.debug("erepo: git checkout %s@%s", repo_path, rev)
-    git = Git(repo_path)
-    try:
-        git.checkout(rev)
-    finally:
-        git.close()
-
-
 def _remove(path):
     from dvc.utils.fs import remove
 
diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py
index 65ee8c735..666b12e02 100644
--- a/dvc/repo/reproduce.py
+++ b/dvc/repo/reproduce.py
@@ -1,11 +1,25 @@
 import logging
-from typing import TYPE_CHECKING, Iterable, List, Optional, Union, cast
+from typing import (
+    TYPE_CHECKING,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    NoReturn,
+    Optional,
+    Set,
+    Tuple,
+    TypeVar,
+    Union,
+    cast,
+)
 
 from funcy import ldistinct
 
 from dvc.exceptions import ReproductionError
 from dvc.repo.scm_context import scm_context
 from dvc.stage.cache import RunCacheNotSupported
+from dvc.utils import humanize
 from dvc.utils.collections import ensure_list
 
 from . import locked
@@ -18,6 +32,7 @@ if TYPE_CHECKING:
     from . import Repo
 
 logger = logging.getLogger(__name__)
+T = TypeVar("T")
 
 
 def collect_stages(
@@ -50,7 +65,7 @@ def get_subgraph(
     return nx.compose_all(used_pipelines)
 
 
-def _remove_frozen_stages(graph: "DiGraph") -> "DiGraph":
+def get_active_graph(graph: "DiGraph") -> "DiGraph":
     g = cast("DiGraph", graph.copy())
     for stage in graph:
         if stage.frozen:
@@ -59,23 +74,12 @@ def _remove_frozen_stages(graph: "DiGraph") -> "DiGraph":
     return g
 
 
-def get_active_graph(
-    graph: "DiGraph",
-    stages: Optional[List["Stage"]] = None,
-    pipeline: bool = False,
-    downstream: bool = False,
-) -> "DiGraph":
-    """Return the graph to operate."""
-    processed = _remove_frozen_stages(graph)
-    return get_subgraph(processed, stages, pipeline=pipeline, downstream=downstream)
-
-
 def plan_repro(
     graph: "DiGraph",
-    stages: Optional[List["Stage"]] = None,
+    stages: Optional[List["T"]] = None,
     pipeline: bool = False,
     downstream: bool = False,
-) -> List["Stage"]:
+) -> List["T"]:
     r"""Derive the evaluation of the given node for the given graph.
 
     When you _reproduce a stage_, you want to _evaluate the descendants_
@@ -113,16 +117,14 @@ def plan_repro(
     """
     import networkx as nx
 
-    active = get_active_graph(graph, stages, pipeline=pipeline, downstream=downstream)
-    return list(nx.dfs_postorder_nodes(active))
+    sub = get_subgraph(graph, stages, pipeline=pipeline, downstream=downstream)
+    return list(nx.dfs_postorder_nodes(sub))
 
 
 def _reproduce_stage(stage: "Stage", **kwargs) -> Optional["Stage"]:
     if stage.frozen and not stage.is_import:
-        logger.warning(
-            "%s is frozen. Its dependencies are not going to be reproduced.",
-            stage,
-        )
+        msg = "%s is frozen. Its dependencies are not going to be reproduced."
+        logger.warning(msg, stage)
 
     ret = stage.reproduce(**kwargs)
     if ret and not kwargs.get("dry", False):
@@ -130,31 +132,88 @@ def _reproduce_stage(stage: "Stage", **kwargs) -> Optional["Stage"]:
     return ret
 
 
-def _reproduce_stages(
+def _get_upstream_downstream_nodes(
+    graph: Optional["DiGraph"], node: T
+) -> Tuple[List[T], List[T]]:
+    succ = list(graph.successors(node)) if graph else []
+    pre = list(graph.predecessors(node)) if graph else []
+    return succ, pre
+
+
+def _repr(stages: Iterable["Stage"]) -> str:
+    return humanize.join(repr(stage.addressing) for stage in stages)
+
+
+def handle_error(
+    graph: Optional["DiGraph"], on_error: str, exc: Exception, stage: "Stage"
+) -> Set["Stage"]:
+    import networkx as nx
+
+    logger.warning("%s%s", exc, " (ignored)" if on_error == "ignore" else "")
+    if not graph or on_error == "ignore":
+        return set()
+
+    dependents = set(nx.dfs_postorder_nodes(graph.reverse(), stage)) - {stage}
+    if dependents:
+        names = _repr(dependents)
+        msg = "%s %s will be skipped due to this failure"
+        logger.warning(msg, "Stages" if len(dependents) > 1 else "Stage", names)
+    return dependents
+
+
+def _raise_error(exc: Optional[Exception], *stages: "Stage") -> NoReturn:
+    names = _repr(stages)
+    segment = " stages:" if len(stages) > 1 else ""
+    raise ReproductionError(f"failed to reproduce{segment} {names}") from exc
+
+
+def _reproduce(
     stages: List["Stage"],
+    graph: Optional["DiGraph"] = None,
     force_downstream: bool = False,
+    on_error: str = "fail",
+    force: bool = False,
+    repro_fn: Callable = _reproduce_stage,
     **kwargs,
 ) -> List["Stage"]:
+    assert on_error in ("fail", "keep-going", "ignore")
+
     result: List["Stage"] = []
-    for i, stage in enumerate(stages):
-        try:
-            ret = _reproduce_stage(stage, upstream=stages[:i], **kwargs)
-        except Exception as exc:  # noqa: BLE001
-            raise ReproductionError(stage.addressing) from exc
+    failed: List["Stage"] = []
+    to_skip: Dict["Stage", "Stage"] = {}
+    ret: Optional["Stage"] = None
+
+    force_state = {node: force for node in stages}
 
-        if not ret:
+    for stage in stages:
+        if stage in to_skip:
             continue
 
-        result.append(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 i < len(stages) - 1:
+        if ret:
             logger.info("")  # add a newline
+
+        upstream, downstream = _get_upstream_downstream_nodes(graph, stage)
+        force_stage = force_state[stage]
+
+        try:
+            ret = repro_fn(stage, upstream=upstream, force=force_stage, **kwargs)
+        except Exception as exc:  # noqa: BLE001, pylint: disable=broad-exception-caught
+            failed.append(stage)
+            if on_error == "fail":
+                _raise_error(exc, stage)
+
+            dependents = handle_error(graph, on_error, exc, stage)
+            to_skip.update({node: stage for node in dependents})
+            continue
+
+        if force_downstream and (ret or force_stage):
+            force_state.update({node: True for node in downstream})
+
+        if ret:
+            result.append(ret)
+
+    if on_error != "ignore" and failed:
+        _raise_error(None, *failed)
     return result
 
 
@@ -169,6 +228,7 @@ def reproduce(
     downstream: bool = False,
     single_item: bool = False,
     glob: bool = False,
+    on_error: Optional[str] = "fail",
     **kwargs,
 ):
     from dvc.dvcfile import PROJECT_FILE
@@ -192,8 +252,9 @@ def reproduce(
         except RunCacheNotSupported as e:
             logger.warning("Failed to pull run cache: %s", e)
 
+    graph = None
     steps = stages
     if not single_item:
-        graph = self.index.graph
+        graph = get_active_graph(self.index.graph)
         steps = plan_repro(graph, stages, pipeline=pipeline, downstream=downstream)
-    return _reproduce_stages(steps, **kwargs)
+    return _reproduce(steps, graph=graph, on_error=on_error or "fail", **kwargs)
diff --git a/dvc/stage/cache.py b/dvc/stage/cache.py
index 2a66ad997..3495ed169 100644
--- a/dvc/stage/cache.py
+++ b/dvc/stage/cache.py
@@ -123,12 +123,15 @@ class StageCache:
 
     @contextmanager
     def _cache_type_copy(self):
-        cache_types = self.repo.cache.legacy.cache_types
+        cache_types = self.repo.cache.local.cache_types
+        legacy_cache_types = self.repo.cache.legacy.cache_types
+        self.repo.cache.local.cache_types = ["copy"]
         self.repo.cache.legacy.cache_types = ["copy"]
         try:
             yield
         finally:
-            self.repo.cache.legacy.cache_types = cache_types
+            self.repo.cache.local.cache_types = cache_types
+            self.repo.cache.legacy.cache_types = legacy_cache_types
 
     def _uncached_outs(self, stage, cache):
         # NOTE: using temporary stage to avoid accidentally modifying original


Test patch:
diff --git a/tests/func/repro/test_repro.py b/tests/func/repro/test_repro.py
index a2b10cb0d..64047dc78 100644
--- a/tests/func/repro/test_repro.py
+++ b/tests/func/repro/test_repro.py
@@ -943,6 +943,18 @@ def test_repro_force_downstream(tmp_dir, dvc, copy_script):
     assert stages[1].addressing == file3_stage.addressing
 
 
+def test_repro_force_downstream_do_not_force_independent_stages(tmp_dir, dvc, run_copy):
+    tmp_dir.gen({"foo": "foo", "bar": "bar"})
+    foo1 = run_copy("foo", "foo1", name="foo1")
+    foo2 = run_copy("foo1", "foo2", name="foo2")
+    run_copy("bar", "bar1", name="bar1")
+    run_copy("bar1", "bar2", name="bar2")
+    cat = dvc.run(cmd="cat bar2 foo2", deps=["foo2", "bar2"], name="cat")
+
+    tmp_dir.gen("foo", "foobar")
+    assert dvc.reproduce(force_downstream=True) == [foo1, foo2, cat]
+
+
 def test_repro_pipeline(
     tmp_dir,
     dvc,
@@ -1296,3 +1308,53 @@ def test_repro_single_item_with_multiple_targets(tmp_dir, dvc, copy_script):
         stage2,
         stage1,
     ]
+
+
+def test_repro_keep_going(mocker, tmp_dir, dvc, copy_script):
+    from dvc.repo import reproduce
+
+    (bar_stage, foo_stage) = tmp_dir.dvc_gen({"bar": "bar", "foo": "foo"})
+    stage1 = dvc.stage.add(
+        cmd=["python copy.py bar foobar", "exit 1"],
+        deps=["bar"],
+        outs=["foobar"],
+        name="copy-bar-foobar",
+    )
+    dvc.stage.add(cmd="cat foobar foo", deps=["foobar", "foo"], name="cat")
+    spy = mocker.spy(reproduce, "_reproduce_stage")
+
+    with pytest.raises(ReproductionError):
+        dvc.reproduce(on_error="keep-going", repro_fn=spy)
+
+    assert spy.call_args_list == [
+        mocker.call(bar_stage, upstream=[], force=False, interactive=False),
+        mocker.call(stage1, upstream=[bar_stage], force=False, interactive=False),
+        mocker.call(foo_stage, upstream=[], force=False, interactive=False),
+    ]
+
+
+def test_repro_ignore_errors(M, mocker, tmp_dir, dvc, copy_script):
+    from dvc.repo import reproduce
+
+    (bar_stage, foo_stage) = tmp_dir.dvc_gen({"bar": "bar", "foo": "foo"})
+    stage1 = dvc.stage.add(
+        cmd=["python copy.py bar foobar", "exit 1"],
+        deps=["bar"],
+        outs=["foobar"],
+        name="copy-bar-foobar",
+    )
+    stage2 = dvc.stage.add(cmd="cat foobar foo", deps=["foobar", "foo"], name="cat")
+    spy = mocker.spy(reproduce, "_reproduce_stage")
+    dvc.reproduce(on_error="ignore", repro_fn=spy)
+
+    assert spy.call_args_list == [
+        mocker.call(bar_stage, upstream=[], force=False, interactive=False),
+        mocker.call(stage1, upstream=[bar_stage], force=False, interactive=False),
+        mocker.call(foo_stage, upstream=[], force=False, interactive=False),
+        mocker.call(
+            stage2,
+            upstream=[foo_stage, stage1],
+            force=False,
+            interactive=False,
+        ),
+    ]
diff --git a/tests/unit/command/test_repro.py b/tests/unit/command/test_repro.py
index 3474ce0c9..7010ee466 100644
--- a/tests/unit/command/test_repro.py
+++ b/tests/unit/command/test_repro.py
@@ -14,6 +14,7 @@ common_arguments = {
     "pull": False,
     "allow_missing": False,
     "targets": [],
+    "on_error": "fail",
 }
 repro_arguments = {
     "run_cache": True,
diff --git a/tests/unit/repo/test_reproduce.py b/tests/unit/repo/test_reproduce.py
index a360284dc..cbc9580d7 100644
--- a/tests/unit/repo/test_reproduce.py
+++ b/tests/unit/repo/test_reproduce.py
@@ -1,70 +1,71 @@
 from itertools import chain
 
 from networkx import DiGraph
+from networkx.utils import graphs_equal
 
-from dvc.repo.reproduce import plan_repro
+from dvc.repo.reproduce import get_active_graph, plan_repro
 
 
-def test_number_reproduces(tmp_dir, dvc, mocker):
-    reproduce_stage_mock = mocker.patch(
-        "dvc.repo.reproduce._reproduce_stage", returns=[]
-    )
-    tmp_dir.dvc_gen({"pre-foo": "pre-foo"})
-
-    dvc.run(name="echo-foo", outs=["foo"], cmd="echo foo > foo")
-    dvc.run(name="echo-bar", deps=["foo"], outs=["bar"], cmd="echo bar > bar")
-    dvc.run(name="echo-baz", deps=["foo"], outs=["baz"], cmd="echo baz > baz")
-    dvc.run(name="echo-boop", deps=["bar"], outs=["boop"], cmd="echo boop > boop")
+def test_active_graph(mocker):
+    n = mocker.sentinel
+    n1, n2, n3, n4, n5, n6, n7, n8, n9 = (getattr(n, f"n{i}") for i in range(1, 10))
+    edges = {n1: [n2, n3], n2: [n4, n5], n3: [n6, n7], n8: [n9]}
+    for node in chain.from_iterable([n, *v] for n, v in edges.items()):
+        node.frozen = False
 
-    reproduce_stage_mock.reset_mock()
+    g = DiGraph(edges)
 
-    dvc.reproduce(all_pipelines=True)
+    active = get_active_graph(g)
+    assert graphs_equal(g, active)
 
-    assert reproduce_stage_mock.call_count == 5
+    n2.frozen = True
+    active = get_active_graph(g)
+    assert g.edges() - active.edges() == {(n2, n5), (n2, n4)}
+    assert n2 in active
+    assert not active.edges() - g.edges()
+    assert not graphs_equal(g, active)
 
 
-def test_repro_plan(mocker):
+def test_repro_plan(M):
     r"""
-             n1
-           /   \
-          n2    m3    n8
-         / \   / \     |
-        n4 n5 n6 n7    n9
+             1
+           /  \
+          2    3    8
+         / \  / \   |
+        4  5 6  7   9
     """
+    g = DiGraph({1: [2, 3], 2: [4, 5], 3: [6, 7], 8: [9]})
 
-    # note: downstream steps may not be stable, use AnyOf in such cases
-    class AnyOf:
-        def __init__(self, *items):
-            self.items = items
+    assert plan_repro(g) == [4, 5, 2, 6, 7, 3, 1, 9, 8]
+    assert plan_repro(g, [1]) == [4, 5, 2, 6, 7, 3, 1]
+    assert plan_repro(g, [4], downstream=True) == [4, 2, 1]
+    assert plan_repro(g, [8], True) == plan_repro(g, [9], True) == [9, 8]
+    assert plan_repro(g, [2, 8], True) == [4, 5, 2, 6, 7, 3, 1, 9, 8]
+    assert plan_repro(g, [2, 3], downstream=True) == [
+        M.any_of(2, 3),
+        M.any_of(2, 3),
+        1,
+    ]
 
-        def __eq__(self, other: object) -> bool:
-            return any(item == other for item in self.items)
 
-    n = mocker.sentinel
-    n1, n2, n3, n4, n5, n6, n7, n8, n9 = (getattr(n, f"n{i}") for i in range(1, 10))
-    edges = {n1: [n2, n3], n2: [n4, n5], n3: [n6, n7], n8: [n9]}
-    for node in chain.from_iterable([n, *v] for n, v in edges.items()):
-        node.frozen = False
+def test_number_reproduces(tmp_dir, dvc, mocker):
+    mock = mocker.Mock(return_value=None)
+    tmp_dir.dvc_gen({"pre-foo": "pre-foo"})
 
-    g = DiGraph(edges)
-    assert plan_repro(g) == [n4, n5, n2, n6, n7, n3, n1, n9, n8]
-    assert plan_repro(g, [n1]) == [n4, n5, n2, n6, n7, n3, n1]
-    assert plan_repro(g, [n4], downstream=True) == [n4, n2, n1]
-    assert plan_repro(g, [n8], True) == plan_repro(g, [n9], True) == [n9, n8]
-    assert plan_repro(g, [n2, n8], True) == [n4, n5, n2, n6, n7, n3, n1, n9, n8]
-    assert plan_repro(g, [n2, n3], downstream=True) == [
-        AnyOf(n2, n3),
-        AnyOf(n2, n3),
-        n1,
-    ]
+    dvc.stage.add(name="echo-foo", outs=["foo"], cmd="echo foo > foo", verify=False)
+    dvc.stage.add(
+        name="echo-bar", deps=["foo"], outs=["bar"], cmd="echo bar > bar", verify=False
+    )
+    dvc.stage.add(
+        name="echo-baz", deps=["foo"], outs=["baz"], cmd="echo baz > baz", verify=False
+    )
+    dvc.stage.add(
+        name="echo-boop",
+        deps=["bar"],
+        outs=["boop"],
+        cmd="echo boop > boop",
+        verify=False,
+    )
 
-    n2.frozen = True
-    assert plan_repro(g) == [n2, n6, n7, n3, n1, n9, n8, n4, n5]
-    assert plan_repro(g, [n1]) == [n2, n6, n7, n3, n1]
-    assert plan_repro(g, [n4], downstream=True) == [n4]
-    assert plan_repro(g, [n2, n8], pipeline=True) == [n2, n6, n7, n3, n1, n9, n8]
-    assert plan_repro(g, [n2, n3], downstream=True) == [
-        AnyOf(n2, n3),
-        AnyOf(n2, n3),
-        n1,
-    ]
+    dvc.reproduce(all_pipelines=True, repro_fn=mock)
+    assert mock.call_count == 5