diff --git a/dvc/cache/local.py b/dvc/cache/local.py
index 1db76fb23..ec4d1d267 100644
--- a/dvc/cache/local.py
+++ b/dvc/cache/local.py
@@ -255,6 +255,7 @@ class LocalCache(CloudCache):
         path_infos = []
         names = []
         hashes = []
+        missing = []
         for md5, info in Tqdm(
             status_info.items(), desc="Analysing status", unit="file"
         ):
@@ -263,6 +264,8 @@ class LocalCache(CloudCache):
                 path_infos.append(remote.tree.hash_to_path_info(md5))
                 names.append(info["name"])
                 hashes.append(md5)
+            elif info["status"] == STATUS_MISSING:
+                missing.append(md5)
 
         if download:
             to_infos = cache
@@ -271,7 +274,7 @@ class LocalCache(CloudCache):
             to_infos = path_infos
             from_infos = cache
 
-        return from_infos, to_infos, names, hashes
+        return (from_infos, to_infos, names, hashes), missing
 
     def _process(
         self,
@@ -312,8 +315,10 @@ class LocalCache(CloudCache):
             download=download,
         )
 
-        dir_plans = self._get_plans(download, remote, dir_status, status)
-        file_plans = self._get_plans(download, remote, file_status, status)
+        dir_plans, _ = self._get_plans(download, remote, dir_status, status)
+        file_plans, missing_files = self._get_plans(
+            download, remote, file_status, status
+        )
 
         total = len(dir_plans[0]) + len(file_plans[0])
         if total == 0:
@@ -339,19 +344,32 @@ class LocalCache(CloudCache):
                         )
                     dir_futures = {}
                     for from_info, to_info, name, dir_hash in zip(*dir_plans):
-                        wait_futures = {
-                            future
-                            for file_hash, future in file_futures.items()
-                            if file_hash in dir_contents[dir_hash]
-                        }
-                        dir_futures[dir_hash] = executor.submit(
-                            self._dir_upload,
-                            func,
-                            wait_futures,
-                            from_info,
-                            to_info,
-                            name,
-                        )
+                        # if for some reason a file contained in this dir is
+                        # missing both locally and in the remote, we want to
+                        # push whatever file content we have, but should not
+                        # push .dir file
+                        for file_hash in missing_files:
+                            if file_hash in dir_contents[dir_hash]:
+                                logger.debug(
+                                    "directory '%s' contains missing files,"
+                                    "skipping .dir file upload",
+                                    name,
+                                )
+                                break
+                        else:
+                            wait_futures = {
+                                future
+                                for file_hash, future in file_futures.items()
+                                if file_hash in dir_contents[dir_hash]
+                            }
+                            dir_futures[dir_hash] = executor.submit(
+                                self._dir_upload,
+                                func,
+                                wait_futures,
+                                from_info,
+                                to_info,
+                                name,
+                            )
                     fails = sum(
                         future.result()
                         for future in concat(
diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py
index 38196ffbd..008b80421 100644
--- a/tests/func/test_remote.py
+++ b/tests/func/test_remote.py
@@ -412,3 +412,23 @@ def test_protect_local_remote(tmp_dir, dvc, local_remote):
 
     assert os.path.exists(remote_cache_file)
     assert stat.S_IMODE(os.stat(remote_cache_file).st_mode) == 0o444
+
+
+def test_push_incomplete_dir(tmp_dir, dvc, mocker, local_remote):
+    (stage,) = tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
+    remote = dvc.cloud.get_remote("upstream")
+
+    cache = dvc.cache.local
+    dir_hash = stage.outs[0].checksum
+    used = stage.get_used_cache(remote=remote)
+
+    # remove one of the cache files for directory
+    file_hashes = list(used.child_keys(cache.tree.scheme, dir_hash))
+    remove(cache.tree.hash_to_path_info(file_hashes[0]))
+
+    dvc.push()
+    assert not remote.tree.exists(remote.tree.hash_to_path_info(dir_hash))
+    assert not remote.tree.exists(
+        remote.tree.hash_to_path_info(file_hashes[0])
+    )
+    assert remote.tree.exists(remote.tree.hash_to_path_info(file_hashes[1]))