remote: make copy links atomic for local/hdfs/ssh

Fixes #2942

dvc/remote/hdfs.py

@@ -100,8 +100,14 @@ class RemoteHDFS(RemoteBASE):
             # NOTE: this is how `hadoop fs -cp` works too: it copies through
             # your local machine.
             with hdfs.open(from_info.path, "rb") as from_fobj:
-                with hdfs.open(to_info.path, "wb") as to_fobj:
-                    to_fobj.upload(from_fobj)
+                tmp_info = to_info.parent / tmp_fname(to_info.name)
+                try:
+                    with hdfs.open(tmp_info.path, "wb") as tmp_fobj:
+                        tmp_fobj.upload(from_fobj)
+                    hdfs.rename(tmp_info.path, to_info.path)
+                except Exception:
+                    self.remove(tmp_info)
+                    raise
 
     def remove(self, path_info):
         if path_info.scheme != "hdfs":

dvc/remote/local.py

@@ -164,9 +164,14 @@ class RemoteLOCAL(RemoteBASE):
 
         move(from_info, to_info, mode=mode)
 
-    @staticmethod
-    def copy(from_info, to_info):
-        System.copy(from_info, to_info)
+    def copy(self, from_info, to_info):
+        tmp_info = to_info.parent / tmp_fname(to_info.name)
+        try:
+            System.copy(from_info, tmp_info)
+            os.rename(fspath_py35(tmp_info), fspath_py35(to_info))
+        except Exception:
+            self.remove(tmp_info)
+            raise
 
     @staticmethod
     def symlink(from_info, to_info):

dvc/remote/ssh/__init__.py

@@ -178,7 +178,7 @@ class RemoteSSH(RemoteBASE):
             raise NotImplementedError
 
         with self.ssh(from_info) as ssh:
-            ssh.copy(from_info.path, to_info.path)
+            ssh.atomic_copy(from_info.path, to_info.path)
 
     def symlink(self, from_info, to_info):
         if not from_info.scheme == to_info.scheme == self.scheme:

dvc/remote/ssh/connection.py

@@ -195,10 +195,10 @@ class SSHConnection:
         try:
             self.sftp.rename(src, dst)
         except OSError:
-            self._atomic_copy(src, dst)
+            self.atomic_copy(src, dst)
             self.remove(src)
 
-    def _atomic_copy(self, src, dst):
+    def atomic_copy(self, src, dst):
         tmp = tmp_fname(dst)
 
         try:

tests/func/test_add.py

@@ -566,12 +566,14 @@ def test_should_not_checkout_when_adding_cached_copy(tmp_dir, dvc, mocker):
 def test_should_relink_on_repeated_add(
     link, new_link, link_test_func, tmp_dir, dvc
 ):
+    from dvc.path_info import PathInfo
+
     dvc.config.set("cache", "type", link)
 
     tmp_dir.dvc_gen({"foo": "foo", "bar": "bar"})
 
     os.remove("foo")
-    getattr(dvc.cache.local, link)("bar", "foo")
+    getattr(dvc.cache.local, link)(PathInfo("bar"), PathInfo("foo"))
 
     dvc.cache.local.cache_types = [new_link]