fix: resolve mkdir race condition in concurrent operations (#551)

Author: CyMule

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.0.52
+
+* **Fix mkdir race condition in concurrent operations**
+
 ## 1.0.51
 
 * **Fix SharePoint connector UnboundLocalError when site not found**

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "1.0.51"  # pragma: no cover
+__version__ = "1.0.52"  # pragma: no cover

unstructured_ingest/processes/connectors/fsspec/fsspec.py

@@ -30,6 +30,7 @@
 )
 from unstructured_ingest.logger import logger
 from unstructured_ingest.processes.connectors.fsspec.utils import sterilize_dict
+from unstructured_ingest.utils.filesystem import mkdir_concurrent_safe
 
 if TYPE_CHECKING:
     from fsspec import AbstractFileSystem
@@ -270,7 +271,7 @@ def wrap_error(self, e: Exception) -> Exception:
 
     def run(self, file_data: FileData, **kwargs: Any) -> DownloadResponse:
         download_path = self.get_download_path(file_data=file_data)
-        download_path.parent.mkdir(parents=True, exist_ok=True)
+        mkdir_concurrent_safe(download_path.parent)
         try:
             rpath = file_data.additional_metadata["original_file_path"]
             with self.connection_config.get_client(protocol=self.protocol) as client:
@@ -282,7 +283,7 @@ def run(self, file_data: FileData, **kwargs: Any) -> DownloadResponse:
 
     async def async_run(self, file_data: FileData, **kwargs: Any) -> DownloadResponse:
         download_path = self.get_download_path(file_data=file_data)
-        download_path.parent.mkdir(parents=True, exist_ok=True)
+        mkdir_concurrent_safe(download_path.parent)
         try:
             rpath = file_data.additional_metadata["original_file_path"]
             with self.connection_config.get_client(protocol=self.protocol) as client:

unstructured_ingest/utils/__init__.py

@@ -0,0 +1,5 @@
+"""Utility functions for unstructured-ingest."""
+
+from unstructured_ingest.utils.filesystem import mkdir_concurrent_safe
+
+__all__ = ["mkdir_concurrent_safe"]

unstructured_ingest/utils/compression.py

@@ -6,6 +6,7 @@
 from typing import Optional
 
 from unstructured_ingest.logger import logger
+from unstructured_ingest.utils.filesystem import mkdir_concurrent_safe
 
 ZIP_FILE_EXT = [".zip"]
 TAR_FILE_EXT = [".tar", ".tar.gz", ".tgz"]
@@ -17,7 +18,7 @@ def uncompress_file(filename: str, path: Optional[str] = None) -> str:
     """
     # Create path if it doesn't already exist
     if path:
-        Path(path).mkdir(parents=True, exist_ok=True)
+        mkdir_concurrent_safe(Path(path))
 
     if any(filename.endswith(ext) for ext in ZIP_FILE_EXT):
         return uncompress_zip_file(zip_filename=filename, path=path)

unstructured_ingest/utils/filesystem.py

@@ -0,0 +1,27 @@
+"""
+Filesystem utilities for concurrent operations.
+
+This module provides race-condition-safe filesystem operations that are needed
+when multiple processes operate on the same directory structures simultaneously.
+"""
+
+from pathlib import Path
+
+
+def mkdir_concurrent_safe(path: Path) -> None:
+    """
+    Create directory safely in concurrent environments, handling race conditions.
+    
+    This addresses the issue where Path.mkdir(parents=True, exist_ok=True) can still
+    raise FileExistsError when multiple processes attempt to create overlapping 
+    directory structures simultaneously. In this codebase, this occurs when multiple
+    files are being downloaded in parallel and archive extraction is happening in parallel.
+    
+    Related: https://github.com/python/cpython/pull/112966/files
+    Python core team used the same approach to fix zipfile race conditions.
+    """
+    try:
+        path.mkdir(parents=True, exist_ok=True)
+    except FileExistsError:
+        if not (path.exists() and path.is_dir()):
+            raise