Repository: iterative/dvc
Pull request: add: warn on cache link errors instead of failing
* [x] ❗ I have followed the [Contributing to DVC](https://dvc.org/doc/user-guide/contributing/core) checklist.

* [x] 📖 If this PR requires [documentation](https://dvc.org/doc) updates, I have created a separate PR (or issue, at least) in [dvc.org](https://github.com/iterative/dvc.org) and linked it here.

Thank you for the contribution - we'll try to review it as soon as possible. 🙏

Will fix #4797.
Docs PR https://github.com/iterative/dvc.org/pull/1892

* `output.commit()` no longer fails if file cannot be linked during `cache.save()` (as long as the file was still moved to cache successfully)
* `dvc add` will succesfully create .dvc file if the link step fails, and will warn user about the link failure and suggest re-running `dvc checkout`

Link failure during add now looks like:

```
$ dvc add foo
100% Add|██████████████████████████████████████████████████████████████████████████████████████████████████████████████|1/1 [00:00,  3.35file/s]
WARNING: Some targets could not be linked from cache to workspace.
See <https://dvc.org/doc/user-guide/troubleshooting#cache-types> for more information.
To re-link these targets, reconfigure cache types and run:

        dvc checkout foo.dvc

To track the changes with git, run:

        git add foo.dvc

$ ls -l
total 4.0K
-rw-r--r-- 1 pmrowla staff 58 Oct 29 12:07 foo.dvc
```

Link failure during `dvc checkout` fails like before, but will include the paths which failed and the troubleshooting link.

```
$ dvc checkout foo.dvc
ERROR: No possible cache types for 'foo'. See <https://dvc.org/doc/user-guide/troubleshooting#cache-types> for more information.
```

Problem statement:
dvc add deletes file when cache type fails on Windows
### Setup
* Windows client, where support for symlinks clearly didn't get applied. This has happened on all our Windows machines. It may have something to do with the way our corporate IT has set up Windows.
* Shared dvc cache on a samba mount. Samba server happens to be 3000 miles away from client.
* cache type symlink

### Situation
* Try to add a file via a line like `dvc add foo.dat`.
* File gets hashed, appears to get uploaded to cache, but then no .dvc file is created, the local copy of `foo.data` is _deleted_, and then dvc raises a "No cache types left to try" error when trying to symlink from the cache back to the local directory. 
* Basically, you're left with no local copy of the file, but also no md5 hash of the file to locate it in the cache!
* Change the local cache.type to `copy` and everything works fine.

I understand the symlink creation failing, but I don't understand why that error should result in the loss of data!

```
DVC version: 1.8.1 (exe)
 Platform: Python 3.7.9 on Windows-10-10.0.17763-SP0
 Supports: All remotes
 Cache types: symlink
 Cache directory: NTFS on Z:\
 Workspace directory: NTFS on D:\
 Repo: dvc, git
```

Hints from issue discussion:
So in this case what's probably happening is that technically the data is not lost since the files are properly moved into the cache (on your SMB share). However, due to our `cache.save` behavior we end up making it look like a loss of data because of the symlink failure (and we don't provide any useful information for recovering the file from cache).

The issue is that for `cache.save` the move from workspace to cache on its own is atomic, but the entire "move + link" step is not if the link fails.

I think what we should probably be doing here (when the cache file is created, but only the link fails) is to consider the `add` successful and create the .dvc file, and then raise a troubleshooting message suggesting something along the lines of "try configuring a different link type and then rerun `dvc checkout ...`".

Alternatively, we would need to adjust `cache.save` to be properly atomic - so if the link step fails, we would need to copy the file from cache back into the original workspace location.

@efiop thoughts?
Technically no loss of data, agreed, but data that you can't access is still data loss from the user's perspective!

Commit message: add test case

Patch to review:
diff --git a/dvc/cache/base.py b/dvc/cache/base.py
index 7d2509bc5d..5dca5a0963 100644
--- a/dvc/cache/base.py
+++ b/dvc/cache/base.py
@@ -8,6 +8,7 @@
 
 import dvc.prompt as prompt
 from dvc.exceptions import (
+    CacheLinkError,
     CheckoutError,
     ConfirmRemoveError,
     DvcException,
@@ -192,7 +193,7 @@ def _try_links(self, from_info, to_info, link_types):
                 )
                 del link_types[0]
 
-        raise DvcException("no possible cache types left to try out.")
+        raise CacheLinkError([to_info])
 
     def _do_link(self, from_info, to_info, link_method):
         if self.tree.exists(to_info):
diff --git a/dvc/exceptions.py b/dvc/exceptions.py
index 783e4eec9c..9d551a77bc 100644
--- a/dvc/exceptions.py
+++ b/dvc/exceptions.py
@@ -338,3 +338,18 @@ def __init__(self, target, file):
 
 class MergeError(DvcException):
     pass
+
+
+class CacheLinkError(DvcException):
+    SUPPORT_LINK = "See {} for more information.".format(
+        format_link(
+            "https://dvc.org/doc/user-guide/" "troubleshooting#cache-types"
+        )
+    )
+
+    def __init__(self, path_infos):
+        msg = "No possible cache link types for '{}'. {}".format(
+            ", ".join([str(path) for path in path_infos]), self.SUPPORT_LINK,
+        )
+        super().__init__(msg)
+        self.path_infos = path_infos
diff --git a/dvc/repo/add.py b/dvc/repo/add.py
index dd1bf9594a..187de18941 100644
--- a/dvc/repo/add.py
+++ b/dvc/repo/add.py
@@ -6,6 +6,7 @@
 from dvc.dvcfile import Dvcfile, is_dvc_file
 
 from ..exceptions import (
+    CacheLinkError,
     OutputDuplicationError,
     OverlappingOutputPathsError,
     RecursiveAddingWhileUsingFilename,
@@ -30,6 +31,7 @@ def add(
     if isinstance(targets, str):
         targets = [targets]
 
+    link_failures = []
     stages_list = []
     num_targets = len(targets)
     with Tqdm(total=num_targets, desc="Add", unit="file", leave=True) as pbar:
@@ -79,33 +81,56 @@ def add(
                     exc.output, set(exc.stages) - set(stages)
                 )
 
-            with Tqdm(
-                total=len(stages),
-                desc="Processing",
-                unit="file",
-                disable=len(stages) == 1,
-            ) as pbar_stages:
-                for stage in stages:
-                    try:
-                        stage.save()
-                    except OutputDoesNotExistError:
-                        pbar.n -= 1
-                        raise
-
-                    if not no_commit:
-                        stage.commit()
-
-                    Dvcfile(repo, stage.path).dump(stage)
-                    pbar_stages.update()
-
+            link_failures.extend(
+                _process_stages(repo, stages, no_commit, pbar)
+            )
             stages_list += stages
 
         if num_targets == 1:  # restore bar format for stats
             pbar.bar_format = pbar.BAR_FMT_DEFAULT
 
+    if link_failures:
+        msg = (
+            "Some targets could not be linked from cache to workspace.\n{}\n"
+            "To re-link these targets, reconfigure cache types and then run:\n"
+            "\n\tdvc checkout {}"
+        ).format(
+            CacheLinkError.SUPPORT_LINK,
+            " ".join([str(stage.relpath) for stage in link_failures]),
+        )
+        logger.warning(msg)
+
     return stages_list
 
 
+def _process_stages(repo, stages, no_commit, pbar):
+    link_failures = []
+
+    with Tqdm(
+        total=len(stages),
+        desc="Processing",
+        unit="file",
+        disable=len(stages) == 1,
+    ) as pbar_stages:
+        for stage in stages:
+            try:
+                stage.save()
+            except OutputDoesNotExistError:
+                pbar.n -= 1
+                raise
+
+            try:
+                if not no_commit:
+                    stage.commit()
+            except CacheLinkError:
+                link_failures.append(stage)
+
+            Dvcfile(repo, stage.path).dump(stage)
+            pbar_stages.update()
+
+    return link_failures
+
+
 def _find_all_targets(repo, target, recursive):
     if os.path.isdir(target) and recursive:
         return [
diff --git a/dvc/stage/__init__.py b/dvc/stage/__init__.py
index 8ef65c1418..4b7b8328e2 100644
--- a/dvc/stage/__init__.py
+++ b/dvc/stage/__init__.py
@@ -7,7 +7,12 @@
 
 import dvc.dependency as dependency
 import dvc.prompt as prompt
-from dvc.exceptions import CheckoutError, DvcException, MergeError
+from dvc.exceptions import (
+    CacheLinkError,
+    CheckoutError,
+    DvcException,
+    MergeError,
+)
 from dvc.output.base import OutputDoesNotExistError
 from dvc.utils import relpath
 
@@ -422,12 +427,17 @@ def changed_entries(self):
 
     @rwlocked(write=["outs"])
     def commit(self, allow_missing=False):
+        link_failures = []
         for out in self.outs:
             try:
                 out.commit()
             except OutputDoesNotExistError:
                 if not allow_missing:
                     raise
+            except CacheLinkError:
+                link_failures.append(out.path_info)
+        if link_failures:
+            raise CacheLinkError(link_failures)
 
     @rwlocked(read=["deps"], write=["outs"])
     def run(
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index ff35e012fa..ed05a9f704 100644
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -820,3 +820,20 @@ def test_add_file_in_symlink_dir(make_tmp_dir, tmp_dir, dvc, external):
 
     with pytest.raises(DvcException):
         dvc.add(os.path.join("dir", "foo"))
+
+
+def test_add_with_cache_link_error(tmp_dir, dvc, mocker, caplog):
+    tmp_dir.gen("foo", "foo")
+
+    mocker.patch.object(
+        dvc.cache.local, "_do_link", side_effect=DvcException("link failed")
+    )
+    with caplog.at_level(logging.WARNING, logger="dvc"):
+        dvc.add("foo")
+        assert "reconfigure cache types" in caplog.text
+
+    assert not (tmp_dir / "foo").exists()
+    assert (tmp_dir / "foo.dvc").exists()
+    assert (tmp_dir / ".dvc" / "cache").read_text() == {
+        "ac": {"bd18db4cc2f85cedef654fccc4a4d8": "foo"}
+    }


Code review comments:
diff_hunk: @@ -338,3 +338,18 @@ def __init__(self, target, file):
 
 class MergeError(DvcException):
     pass
+
+
+class CacheLinkError(DvcException):
+    SUPPORT_LINK = "See {} for more information.".format(
+        format_link(
+            "https://dvc.org/doc/user-guide/" "troubleshooting#cache-types"
path: dvc/exceptions.py
text: @user1:
```suggestion
            "https://dvc.org/doc/user-guide/troubleshooting#cache-types"
```