fix: add retries to insertsions & deletions (#3433)

Author: mbani01

services/apps/git_integration/src/crowdgit/services/commit/commit_service.py

@@ -66,6 +66,12 @@ def _get_or_create_pool(self) -> ProcessPoolExecutor:
             self.process_pool = ProcessPoolExecutor(max_workers=max_workers)
         return self.process_pool
 
+    def cleanup_process_pool(self):
+        """Cleanup process pool to prevent resource leaks"""
+        if self.process_pool:
+            self.process_pool.shutdown(wait=False)
+            self.process_pool = None
+
     @property
     def git_log_format(self) -> str:
         """Git log format string with commit splitter"""
@@ -703,34 +709,46 @@ def _validate_commit_data(commit_dict: dict[str, Any]) -> bool:
 
     @staticmethod
     def get_insertions_deletions(commit_hash: str, repo_path: str) -> tuple[int, int]:
+        """Get insertions and deletions for a commit with retry logic and fallback."""
         try:
-            # Use git show which works for all cases: normal commits, root commits, and shallow boundary commits
-            result = subprocess.run(
-                ["git", "-C", repo_path, "show", "--numstat", "--format=", commit_hash],
-                capture_output=True,
-                text=True,
-                check=True,
+            return CommitService._get_insertions_deletions_with_retry(commit_hash, repo_path)
+        except Exception as e:
+            logger.error(
+                f"All retries failed for insertions/deletions for commit {commit_hash}: {e}"
             )
-
-            insertions, deletions = 0, 0
-            # Process the multi-line output directly in Python.
-            for line in result.stdout.splitlines():
-                if line.strip():  # Skip empty lines
-                    parts = line.split("\t")  # --numstat uses tabs
-                    if len(parts) >= 2:
-                        try:
-                            insertions += int(parts[0])
-                            deletions += int(parts[1])
-                        except ValueError:
-                            # Skip lines that don't have numeric values (e.g., binary files)
-                            continue
-
-            return insertions, deletions
-
-        except (subprocess.CalledProcessError, ValueError, IndexError) as e:
-            logger.error(f"Error getting insertions/deletions for commit {commit_hash}: {e}")
             return 0, 0
 
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_fixed(2),
+    )
+    @staticmethod
+    def _get_insertions_deletions_with_retry(commit_hash: str, repo_path: str) -> tuple[int, int]:
+        """Internal method that performs the actual git operation with retries."""
+        # Use git show which works for all cases: normal commits, root commits, and shallow boundary commits
+        result = subprocess.run(
+            ["git", "-C", repo_path, "show", "--numstat", "--format=", commit_hash],
+            capture_output=True,
+            text=True,
+            check=True,
+            timeout=120,
+        )
+
+        insertions, deletions = 0, 0
+        # Process the multi-line output directly in Python.
+        for line in result.stdout.splitlines():
+            if line.strip():  # Skip empty lines
+                parts = line.split("\t")  # --numstat uses tabs
+                if len(parts) >= 2:
+                    try:
+                        insertions += int(parts[0])
+                        deletions += int(parts[1])
+                    except ValueError:
+                        # Skip lines that don't have numeric values (e.g., binary files)
+                        continue
+
+        return insertions, deletions
+
     @staticmethod
     def _is_valid_email(email: str) -> bool:
         """Check if a string is a valid email format using Pydantic validation."""

services/apps/git_integration/src/crowdgit/worker/repository_worker.py

@@ -171,6 +171,7 @@ async def _process_single_repository(self, repository: Repository):
             processing_state = RepositoryState.FAILED
             logger.error(f"Processing failed with error: {repr(e)}")
         finally:
+            self.commit_service.cleanup_process_pool()
             # Reset logger context for all services
             self._reset_all_contexts()