push: test versioning and ignore missing

src/dvc_data/index/checkout.py

@@ -7,6 +7,7 @@
     Callable,
     Collection,
     Dict,
+    Iterable,
     Iterator,
     List,
     Optional,
@@ -38,22 +39,21 @@ class VersioningNotSupported(Exception):
     pass
 
 
-def test_versioning(
-    src_fs: "FileSystem",
-    src_path: "AnyFSPath",
-    dest_fs: "FileSystem",
-    dest_path: "AnyFSPath",
-    callback: "Callback" = DEFAULT_CALLBACK,
-) -> Meta:
-    transfer(src_fs, src_path, dest_fs, dest_path, callback=callback)
-    info = dest_fs.info(dest_path)
-    meta = Meta.from_info(info, dest_fs.protocol)
-    if meta.version_id in (None, "null"):
-        raise VersioningNotSupported(
-            f"while uploading {dest_path!r}, "
-            "support for versioning could not be detected"
-        )
-    return meta
+def _check_versioning(paths: Iterable["AnyFSPath"], fs: "FileSystem"):
+    if not fs.version_aware:
+        return
+
+    for path in paths:
+        try:
+            info = fs.info(path)
+        except FileNotFoundError:
+            continue
+        meta = Meta.from_info(info, fs.protocol)
+        if meta.version_id in (None, "null"):
+            raise VersioningNotSupported(
+                f"while uploading {path!r}, "
+                "support for versioning could not be detected"
+            )
 
 
 def _delete_files(
@@ -103,16 +103,6 @@ def _create_files(  # noqa: C901
 
         by_storage[storage_obj].append((entry, src_path, dest_path))
 
-    if fs.version_aware and by_storage:
-        storage_obj, items = next(iter(by_storage.items()))
-        src_fs = storage_obj.fs
-        if items:
-            entry, src_path, dest_path = items.pop()
-            entry.meta = test_versioning(
-                src_fs, src_path, fs, dest_path, callback=callback
-            )
-            index.add(entry)
-
     for storage_obj, args in by_storage.items():
         if not args:
             continue
@@ -134,6 +124,8 @@ def _create_files(  # noqa: C901
             on_error=onerror,
         )
 
+        _check_versioning(dest_paths, fs)
+
         if state:
             for entry, _, dest_path in args:
                 if not entry.hash_info:

src/dvc_data/index/push.py

@@ -31,6 +31,15 @@ def _meta_checksum(fs: "FileSystem", meta: "Meta") -> Any:
     return getattr(meta, fs.PARAM_CHECKSUM)
 
 
+def _onerror(src_path, dest_path, _exc):
+    logger.debug(
+        "failed to create '%s' from '%s'",
+        src_path,
+        dest_path,
+        exc_info=True,
+    )
+
+
 def push(
     idxs,
     callback: "Callback" = DEFAULT_CALLBACK,
@@ -87,6 +96,7 @@ def push(
                     jobs=jobs,
                     callback=cb,
                     links=["reflink", "copy"],
+                    onerror=_onerror,
                 )
                 fetched += len(diff.changes.get("added", []))