fix(commits): prevent broken pool process error [CM-702]

Author: mbani01

services/apps/git_integration/src/crowdgit/services/clone/clone_service.py

@@ -256,16 +256,16 @@ async def _calculate_batch_depth(self, repo_path: str, remote: str) -> int:
         total_branches_tags = len(total_branches_tags.splitlines())
         if total_branches_tags <= 200:
             # Small repo, get a decent amount of history
-            calculated_depth = 500
+            calculated_depth = 200
         elif total_branches_tags <= 1000:
             # Medium repo, get a moderate amount of history
-            calculated_depth = 300
+            calculated_depth = 100
         elif total_branches_tags <= 5000:
             # Large repo, get less history
-            calculated_depth = 20
+            calculated_depth = 10
         else:
             # Very large repo, get a minimal history
-            calculated_depth = 10
+            calculated_depth = 5
         self.logger.info(
             f"total_branches_tags={total_branches_tags}, calculated_depth={calculated_depth}"
         )

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

@@ -7,6 +7,7 @@
 import time
 import uuid
 from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures.process import BrokenProcessPool
 from decimal import Decimal
 from typing import Any
 
@@ -69,6 +70,7 @@ def _get_or_create_pool(self) -> ProcessPoolExecutor:
     def cleanup_process_pool(self):
         """Cleanup process pool to prevent resource leaks"""
         if self.process_pool:
+            self.logger.info("Cleaning up process pool")
             self.process_pool.shutdown(wait=False)
             self.process_pool = None
 
@@ -659,26 +661,44 @@ async def _process_activities_from_commits(
 
         executor = self._get_or_create_pool()
 
-        futures = [
-            loop.run_in_executor(
-                executor,
-                CommitService.process_commits_chunk,
-                chunk,
-                repo_path,
-                edge_commit_hash,
-                remote,
-                segment_id,
-                integration_id,
-            )
-            for chunk in chunks
-        ]
+        try:
+            futures = [
+                loop.run_in_executor(
+                    executor,
+                    CommitService.process_commits_chunk,
+                    chunk,
+                    repo_path,
+                    edge_commit_hash,
+                    remote,
+                    segment_id,
+                    integration_id,
+                )
+                for chunk in chunks
+            ]
+            self.logger.info(f"Submitted {len(futures)} tasks to process pool")
+        except Exception as e:
+            if isinstance(e, BrokenProcessPool):
+                self.logger.warning("BrokenProcessPool during task submission, cleaning up")
+                self.cleanup_process_pool()
+            raise
 
         # Save each chunk's activities as they complete
+        completed_chunks = 0
         for future in asyncio.as_completed(futures):
-            chunk_activities_db, chunk_activities_queue = await future
-            if chunk_activities_db and chunk_activities_queue:
-                await batch_insert_activities(chunk_activities_db)
-                await self.queue_service.send_batch_activities(chunk_activities_queue)
+            try:
+                chunk_activities_db, chunk_activities_queue = await future
+                completed_chunks += 1
+                self.logger.info(f"Chunk {completed_chunks}/{len(futures)} completed")
+                if chunk_activities_db and chunk_activities_queue:
+                    await batch_insert_activities(chunk_activities_db)
+                    await self.queue_service.send_batch_activities(chunk_activities_queue)
+            except Exception as e:
+                if isinstance(e, BrokenProcessPool):
+                    self.logger.warning(
+                        f"BrokenProcessPool after {completed_chunks}/{len(futures)} chunks, cleaning up"
+                    )
+                    self.cleanup_process_pool()
+                raise
 
     @staticmethod
     def _validate_commit_structure(commit_lines: list[str]) -> bool:

services/apps/git_integration/src/crowdgit/settings.py

@@ -18,7 +18,7 @@ def load_env_var(key: str, required=True, default=None):
 REPOSITORY_UPDATE_INTERVAL_HOURS = int(
     load_env_var("REPOSITORY_UPDATE_INTERVAL_HOURS", default=24)
 )
-MAX_WORKER_PROCESSES = int(load_env_var("MAX_WORKER_PROCESSES", default=min(4, os.cpu_count())))
+MAX_WORKER_PROCESSES = int(load_env_var("MAX_WORKER_PROCESSES", default=min(3, os.cpu_count())))
 DEFAULT_TENANT_ID = load_env_var(
     "CROWD_SSO_LF_TENANT_ID", default="875c38bd-2b1b-4e91-ad07-0cfbabb4c49f"
 )