diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py
index 0301fce3c..06df65a4a 100644
--- a/dvc/remote/hdfs.py
+++ b/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":
diff --git a/dvc/remote/local.py b/dvc/remote/local.py
index d8a269458..9c68e3a80 100644
--- a/dvc/remote/local.py
+++ b/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):
diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py
index 7509d7504..78da815e1 100644
--- a/dvc/remote/ssh/__init__.py
+++ b/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:
diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py
index dafde3e15..abc131173 100644
--- a/dvc/remote/ssh/connection.py
+++ b/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:
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index 300cccc45..8a76263da 100644
--- a/tests/func/test_add.py
+++ b/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]