Skip legacy tar file-directory conflicts

Author: apstenku123

scripts/streaming_reindex.py

@@ -491,6 +491,36 @@ def _mark_source_cache_complete(repo_dir: Path, repo: str) -> None:
     tmp.replace(sentinel)
 
 
+def _copy_tar_member_file(src, target: Path, *, repo: str, member_name: str) -> bool:
+    """Copy a tar file member, explicitly skipping legacy file/dir conflicts.
+
+    Some old source archives contain path-type collisions, for example a
+    directory and a regular file at the same path. Blocking the whole corpus on
+    that single path is worse than losing one impossible-to-materialize member,
+    but the skip must be visible in logs.
+    """
+    if target.exists() and target.is_dir():
+        print(
+            f"WARN {repo}: skip tar file {member_name!r}; target path is already "
+            f"a directory: {target}",
+            file=sys.stderr,
+            flush=True,
+        )
+        return False
+    if target.parent.exists() and not target.parent.is_dir():
+        print(
+            f"WARN {repo}: skip tar file {member_name!r}; parent path is not a "
+            f"directory: {target.parent}",
+            file=sys.stderr,
+            flush=True,
+        )
+        return False
+    target.parent.mkdir(parents=True, exist_ok=True)
+    with open(target, "wb") as out:
+        shutil.copyfileobj(src, out)
+    return True
+
+
 def _iter_complete_source_cache(
     source_cache_dir: Path,
     should_process: Callable[[str], bool],
@@ -613,9 +643,7 @@ def stream_repo_subtrees(
             if src is None:
                 continue
             target = cur_dir / within
-            target.parent.mkdir(parents=True, exist_ok=True)
-            with open(target, "wb") as out:
-                shutil.copyfileobj(src, out)
+            _copy_tar_member_file(src, target, repo=repo, member_name=name)
         if cur_repo is not None and active:
             assert cur_dir is not None
             yield_dir = cur_dir

tests/test_streaming_reindex_run_checked.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+import io
 import sys
+import tarfile
+import subprocess
 
 
 def test_run_checked_times_out_fail_loud(tmp_path) -> None:
@@ -156,6 +159,50 @@ def test_stream_repo_dirs_yields_extracted_src_without_tar(tmp_path) -> None:
     assert yielded == [("repo", root / "repo" / "_src")]
 
 
+def test_stream_repo_subtrees_skips_legacy_file_directory_conflict(
+    tmp_path,
+    monkeypatch,
+    capsys,
+) -> None:
+    import streaming_reindex
+
+    tar_path = tmp_path / "source.tar"
+    zst_path = tmp_path / "source.tar.zst"
+    with tarfile.open(tar_path, "w") as tf:
+        child_info = tarfile.TarInfo("cpp_all/repo/conflict/child.txt")
+        child_payload = b"child creates the conflict directory\n"
+        child_info.size = len(child_payload)
+        child_info.mode = 0o644
+        tf.addfile(child_info, io.BytesIO(child_payload))
+
+        file_info = tarfile.TarInfo("cpp_all/repo/conflict")
+        payload = b"this cannot be materialized over a directory\n"
+        file_info.size = len(payload)
+        file_info.mode = 0o644
+        tf.addfile(file_info, io.BytesIO(payload))
+
+        good_info = tarfile.TarInfo("cpp_all/repo/good.cpp")
+        good_payload = b"int good;\n"
+        good_info.size = len(good_payload)
+        good_info.mode = 0o644
+        tf.addfile(good_info, io.BytesIO(good_payload))
+
+    subprocess.run(["zstd", "-q", "-f", str(tar_path), "-o", str(zst_path)], check=True)
+    monkeypatch.setattr(streaming_reindex, "TARBALL", zst_path)
+
+    yielded = list(
+        streaming_reindex.stream_repo_subtrees(
+            tmp_path / "work",
+            lambda repo: repo == "repo",
+        )
+    )
+
+    assert yielded == [("repo", tmp_path / "work" / "repo" / "_src")]
+    assert (tmp_path / "work" / "repo" / "_src" / "conflict").is_dir()
+    assert (tmp_path / "work" / "repo" / "_src" / "good.cpp").read_text() == "int good;\n"
+    assert "skip tar file" in capsys.readouterr().err
+
+
 def test_populate_source_cache_only_materializes_cache(tmp_path, monkeypatch) -> None:
     import streaming_reindex