diff --git a/dvc/stage.py b/dvc/stage.py
index a569b245af..62ea6278f5 100644
--- a/dvc/stage.py
+++ b/dvc/stage.py
@@ -394,7 +394,21 @@ def is_cached(self):
             out.pop(RemoteLOCAL.PARAM_CHECKSUM, None)
             out.pop(RemoteS3.PARAM_CHECKSUM, None)
 
-        return old_d == new_d
+        if old_d != new_d:
+            return False
+
+        # NOTE: committing to prevent potential data duplication. For example
+        #
+        #    $ dvc config cache.type hardlink
+        #    $ echo foo > foo
+        #    $ dvc add foo
+        #    $ rm -f foo
+        #    $ echo foo > foo
+        #    $ dvc add foo # should replace foo with a link to cache
+        #
+        old.commit()
+
+        return True
 
     @staticmethod
     def create(
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index de4108b447..4e7e66ad5e 100644
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -475,3 +475,24 @@ def test(self):
         created_stages_filenames = stage_creator_spy.mock.call_args[0][0]
         for fname in created_stages_filenames:
             self.assertNotIn(".git", fname)
+
+
+class TestAddUnprotected(TestDvc):
+    def test(self):
+        ret = main(["config", "cache.type", "hardlink"])
+        self.assertEqual(ret, 0)
+
+        ret = main(["config", "cache.protected", "true"])
+        self.assertEqual(ret, 0)
+
+        ret = main(["add", self.FOO])
+        self.assertEqual(ret, 0)
+
+        ret = main(["unprotect", self.FOO])
+        self.assertEqual(ret, 0)
+
+        ret = main(["add", self.FOO])
+        self.assertEqual(ret, 0)
+
+        self.assertFalse(os.access(self.FOO, os.W_OK))
+        self.assertTrue(System.is_hardlink(self.FOO))