fix: prevent missing commits during shallow clones (#4127)

Signed-off-by: Mouad BANI <mouad-mb@outlook.com>

Author: mbani01

services/apps/git_integration/src/crowdgit/database/crud.py

@@ -8,6 +8,7 @@
 from crowdgit.models.repository import Repository
 from crowdgit.models.service_execution import ServiceExecution
 from crowdgit.settings import (
+    FAILED_RETRY_INTERVAL_HOURS,
     MAX_CONCURRENT_ONBOARDINGS,
     MAX_INTEGRATION_RESULTS,
     REPOSITORY_UPDATE_INTERVAL_HOURS,
@@ -143,7 +144,9 @@ async def acquire_recurrent_repo() -> Repository | None:
         WHERE NOT (rp.state = ANY($2))
             AND rp."lockedAt" IS NULL
             AND r."deletedAt" IS NULL
-            AND rp."lastProcessedAt" < NOW() - INTERVAL '1 hour' * $3
+            AND rp."lastProcessedAt" < NOW() - INTERVAL '1 hour' * (
+                CASE WHEN rp.state = 'failed' THEN $4::numeric ELSE $3::numeric END
+            )
             AND NOT (
                 r.url LIKE '%gerrit.automotivelinux.org%'
                 AND EXISTS (SELECT 1 FROM automotivelinux_processing)
@@ -170,7 +173,12 @@ async def acquire_recurrent_repo() -> Repository | None:
     )
     return await acquire_repository(
         recurrent_repo_sql_query,
-        (RepositoryState.PROCESSING, states_to_exclude, REPOSITORY_UPDATE_INTERVAL_HOURS),
+        (
+            RepositoryState.PROCESSING,
+            states_to_exclude,
+            REPOSITORY_UPDATE_INTERVAL_HOURS,
+            FAILED_RETRY_INTERVAL_HOURS,
+        ),
     )
 
 

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

@@ -47,7 +47,7 @@ class NetworkError(CrowdGitError):
 
 
 @dataclass
-class PermissionError(CrowdGitError):
+class RepoPermissionError(CrowdGitError):
     error_message: str = "Permission denied"
     error_code: ErrorCode = ErrorCode.PERMISSION_ERROR
 

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

@@ -33,6 +33,9 @@
 )
 
 DEFAULT_STORAGE_OPTIMIZATION_THRESHOLD_MB = 10000
+INITIAL_BATCH_DEEPEN = 50
+MAX_BATCH_DEEPEN = 10000
+BATCH_DEEPEN_MULTIPLIER = 2
 
 RETRYABLE_CLONE_ERRORS = (RateLimitError, NetworkError, RemoteServerError)
 
@@ -57,6 +60,16 @@ def __init__(self):
     def _is_gerrit_remote(remote: str) -> bool:
         return any(pattern in remote for pattern in GERRIT_PATTERNS)
 
+    async def _configure_global_git_client(self, path: str) -> None:
+        # Increase post buffer to reduce RPC failures on large repos
+        await run_shell_command(
+            ["git", "config", "--global", "http.postBuffer", "524288000"], cwd=path
+        )
+        # Disable recursive submodule fetches
+        await run_shell_command(
+            ["git", "config", "--global", "fetch.recurseSubmodules", "false"], cwd=path
+        )
+
     async def _check_if_final_batch(self, path: str, target_commit_hash: str | None) -> bool:
         """
         Final batch is determined if:
@@ -74,20 +87,62 @@ async def _check_if_final_batch(self, path: str, target_commit_hash: str | None)
             await run_shell_command(
                 ["git", "rev-parse", "--verify", f"{target_commit_hash}^{{commit}}"], cwd=path
             )
+            has_boundary_in_required_range = await self._has_boundary_in_required_range(
+                path, target_commit_hash
+            )
+            if has_boundary_in_required_range:
+                self.logger.info(
+                    f"Target commit {target_commit_hash} reached, but shallow boundary still intersects required range. Continuing deepen."
+                )
+                return False
             self.logger.info(f"Target commit {target_commit_hash} reached")
             return True
         except CommandExecutionError:
             return False
 
+    async def _read_shallow_file(self, repo_path: str) -> list[str]:
+        """
+        Read commit hashes from .git/shallow.
+        Returns an empty list when the file does not exist (full clone).
+        """
+        shallow_file = os.path.join(repo_path, ".git", "shallow")
+        try:
+            async with aiofiles.open(shallow_file, "r", encoding="utf-8") as f:
+                return [line.strip() for line in await f.readlines() if line.strip()]
+        except FileNotFoundError:
+            return []
+
+    async def _get_shallow_boundary_commits(self, repo_path: str) -> set[str]:
+        """
+        Return all shallow boundary commits from .git/shallow.
+        """
+        return set(await self._read_shallow_file(repo_path))
+
+    async def _has_boundary_in_required_range(
+        self, repo_path: str, target_commit_hash: str
+    ) -> bool:
+        """
+        Check if any shallow boundary commit is inside the current required range.
+
+        If true, the required range is still truncated and the clone should be deepened.
+        """
+        shallow_boundaries = await self._get_shallow_boundary_commits(repo_path)
+        if not shallow_boundaries:
+            return False
+
+        required_commits_output = await run_shell_command(
+            ["git", "rev-list", f"{target_commit_hash}..HEAD"],
+            cwd=repo_path,
+        )
+        required_commits = {line.strip() for line in required_commits_output.splitlines() if line}
+        problematic_boundaries = (required_commits & shallow_boundaries) - {target_commit_hash}
+        return bool(problematic_boundaries)
+
     @retry_on_clone_error
     async def _perform_minimal_clone(self, path: str, remote: str) -> None:
         """
         Perform minimal clone of depth=1
         """
-        # increasing post buffer to avoid RPC failed error
-        await run_shell_command(
-            ["git", "config", "--global", "http.postBuffer", "524288000"], cwd=path
-        )
         self.logger.info("Initializing minimal clone")
         await run_shell_command(
             ["git", "clone", "--depth=1", "--no-tags", "--single-branch", remote, "."], cwd=path
@@ -196,21 +251,19 @@ async def _update_batch_info(
         batch_info.is_final_batch = await self._check_if_final_batch(repo_path, target_commit_hash)
         batch_info.edge_commit = await self._get_edge_commit(repo_path)
 
-    async def _get_edge_commit(self, repo_path: str):
+    async def _get_edge_commit(self, repo_path: str) -> str | None:
         """
         Returns the edge commit of a shallow clone by reading the .git/shallow file,
         which contains the boundary commit(s) when history is truncated.
 
         If the full history has been cloned, the .git/shallow file does not exist.
         """
-        shallow_file = os.path.join(repo_path, ".git", "shallow")
-        try:
-            async with aiofiles.open(shallow_file, "r", encoding="utf-8") as f:
-                oldest_commit = (await f.readline()).strip()
-            self.logger.info(f"Edge commit: {oldest_commit}")
-            return oldest_commit
-        except FileNotFoundError:
+        boundaries = await self._read_shallow_file(repo_path)
+        if not boundaries:
             return None
+        oldest_commit = boundaries[0]
+        self.logger.info(f"Edge commit: {oldest_commit}")
+        return oldest_commit
 
     async def _cleanup_temp_directory(self, temp_repo_path: str, repo_id: str) -> None:
         """
@@ -288,26 +341,6 @@ async def _cleanup_working_directory(self, repo_path: str) -> None:
 
         self.logger.info("Working directory cleanup completed")
 
-    async def _calculate_batch_depth(self, repo_path: str, remote: str) -> int:
-        calculated_depth = None
-        total_branches_tags = await run_shell_command(
-            ["git", "ls-remote", "--heads", "--tags", remote], cwd=repo_path
-        )
-        total_branches_tags = len(total_branches_tags.splitlines())
-        if total_branches_tags <= 200:
-            # Small repo, get a decent amount of history
-            calculated_depth = 100
-        elif total_branches_tags <= 1000:
-            # Medium repo, get a moderate amount of history
-            calculated_depth = 50
-        else:
-            # Large repo, get less history
-            calculated_depth = 5
-        self.logger.info(
-            f"total_branches_tags={total_branches_tags}, calculated_depth={calculated_depth}"
-        )
-        return calculated_depth
-
     @retry_on_clone_error
     async def _perform_full_clone(self, repo_path: str, remote: str):
         """Perform full repository clone"""
@@ -372,6 +405,7 @@ async def determine_clone_strategy(
         self.logger.info(
             f"Starting clone decision for {remote} (branch: {branch}, last_commit: {last_processed_commit})"
         )
+        await self._configure_global_git_client(repo_path)
 
         default_branch_changed = await self.has_default_branch_changed(remote, branch)
 
@@ -405,6 +439,7 @@ async def clone_batches_generator(
         error_code = None
         error_message = None
         total_execution_time = 0.0
+        batch_depth = INITIAL_BATCH_DEEPEN
         remote = repository.url.removesuffix(".git")
 
         batch_info = CloneBatchInfo(
@@ -420,8 +455,6 @@ async def clone_batches_generator(
             clone_with_batches = await self.determine_clone_strategy(
                 temp_repo_path, remote, repository.branch, repository.last_processed_commit
             )
-            if clone_with_batches:
-                batch_depth = await self._calculate_batch_depth(temp_repo_path, remote)
             await self._update_batch_info(
                 batch_info, temp_repo_path, repository.last_processed_commit, clone_with_batches
             )
@@ -447,6 +480,12 @@ async def clone_batches_generator(
                 total_execution_time += round(batch_end_time - batch_start_time, 2)
 
                 yield batch_info
+                if not batch_info.is_final_batch:
+                    # exponential deepen increment to speed up fetching
+                    batch_depth = min(
+                        batch_depth * BATCH_DEEPEN_MULTIPLIER,
+                        MAX_BATCH_DEEPEN,
+                    )
 
         except Exception as e:
             # Handle both CrowdGitError and generic Exception

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

@@ -9,10 +9,10 @@
     EmptyRepoError,
     ForbiddenError,
     NetworkError,
-    PermissionError,
     RateLimitError,
     RemoteServerError,
     RepoAuthRequiredError,
+    RepoPermissionError,
     ValidationError,
 )
 from crowdgit.logger import logger
@@ -118,8 +118,11 @@ async def get_remote_default_branch(remote_url: str) -> str | None:
         # Fallback: if symbolic ref not available, try common branches
         for branch in ["main", "master"]:
             try:
-                await run_shell_command(["git", "ls-remote", "--heads", remote_url, branch])
-                return branch
+                output = await run_shell_command(
+                    ["git", "ls-remote", "--heads", remote_url, branch]
+                )
+                if output.strip():
+                    return branch
             except CommandExecutionError:
                 continue
 
@@ -171,7 +174,7 @@ async def get_default_branch(repo_path: str) -> str:
     # (stderr_patterns, exception_class)
     ({"No space left on device"}, DiskSpaceError),
     ({"Network is unreachable", "Connection refused", "Connection timed out"}, NetworkError),
-    ({"Permission denied"}, PermissionError),
+    ({"Permission denied"}, RepoPermissionError),
     ({"The requested URL returned error: 403"}, ForbiddenError),
     ({"The requested URL returned error: 429"}, RateLimitError),
     ({"The requested URL returned error: 5"}, RemoteServerError),
@@ -231,7 +234,7 @@ async def run_shell_command(
         CommandTimeoutError: When command times out
         DiskSpaceError: When disk space is insufficient
         NetworkError: When network connectivity issues occur
-        PermissionError: When permission is denied
+        RepoPermissionError: When permission is denied
         CommandExecutionError: For other command failures
     """
     process = None

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

@@ -9,7 +9,7 @@ def load_env_var(key: str, required=True, default=None):
 
 
 CROWD_DB_WRITE_HOST = load_env_var("CROWD_DB_WRITE_HOST")
-CROWD_DB_PORT = load_env_var("CROWD_DB_PORT")
+CROWD_DB_PORT = int(load_env_var("CROWD_DB_PORT"))
 CROWD_DB_USERNAME = load_env_var("CROWD_DB_USERNAME")
 CROWD_DB_PASSWORD = load_env_var("CROWD_DB_PASSWORD")
 CROWD_DB_DATABASE = load_env_var("CROWD_DB_DATABASE")
@@ -18,6 +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)
 )
+FAILED_RETRY_INTERVAL_HOURS = int(load_env_var("FAILED_RETRY_INTERVAL_HOURS", default="6"))
 DEFAULT_TENANT_ID = load_env_var(
     "CROWD_SSO_LF_TENANT_ID", default="875c38bd-2b1b-4e91-ad07-0cfbabb4c49f"
 )

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

@@ -145,7 +145,7 @@ async def _process_repositories(self):
             )
         finally:
             if available_repo_to_process:
-                logger.info("releasing repo: ", available_repo_to_process.url)
+                logger.info(f"releasing repo: {available_repo_to_process.url}")
                 await release_repo(available_repo_to_process.id)
                 logger.info(f"Repo {available_repo_to_process.url} released!")