fix(git-integration): target commit from an outdated branch (CM-717) (#3480)

Author: mbani01

backend/src/database/migrations/U1759851185__addBranchColumnToRepositories.sql

@@ -0,0 +1,3 @@
+-- Remove branch column from git.repositories table
+ALTER TABLE git.repositories 
+DROP COLUMN IF EXISTS "branch";

backend/src/database/migrations/V1759851185__addBranchColumnToRepositories.sql

@@ -0,0 +1,5 @@
+ALTER TABLE git.repositories 
+ADD COLUMN "branch" VARCHAR(255) DEFAULT NULL;
+
+-- Add comment for documentation
+COMMENT ON COLUMN git.repositories."branch" IS 'The default branch being tracked for this repository (e.g., main, master, develop).';

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

@@ -52,7 +52,7 @@ async def acquire_onboarding_repo() -> Repository | None:
         LIMIT 1
         FOR UPDATE SKIP LOCKED
     )
-    RETURNING id, url, state, priority, "lastProcessedAt", "lastProcessedCommit", "lockedAt", "createdAt", "updatedAt", "segmentId", "integrationId", "maintainerFile", "lastMaintainerRunAt"
+    RETURNING id, url, state, priority, "lastProcessedAt", "lastProcessedCommit", "lockedAt", "createdAt", "updatedAt", "segmentId", "integrationId", "maintainerFile", "lastMaintainerRunAt", "branch"
     """
     return await acquire_repository(
         onboarding_repo_sql_query, (RepositoryState.PROCESSING, RepositoryState.PENDING)
@@ -101,7 +101,7 @@ async def acquire_recurrent_repo() -> Repository | None:
         LIMIT 1
         FOR UPDATE SKIP LOCKED
     )
-    RETURNING id, url, state, priority, "lastProcessedAt", "lastProcessedCommit", "lockedAt", "createdAt", "updatedAt", "segmentId", "integrationId", "maintainerFile", "lastMaintainerRunAt"
+    RETURNING id, url, state, priority, "lastProcessedAt", "lastProcessedCommit", "lockedAt", "createdAt", "updatedAt", "segmentId", "integrationId", "maintainerFile", "lastMaintainerRunAt", "branch"
     """
     states_to_exclude = (RepositoryState.PENDING, RepositoryState.PROCESSING)
     return await acquire_repository(
@@ -134,17 +134,18 @@ async def release_repo(repo_id: str):
     return str(result)
 
 
-async def update_last_processed_commit(repo_id: str, commit_hash: str):
+async def update_last_processed_commit(repo_id: str, commit_hash: str, branch: str | None = None):
     """
-    Release repository lock (lockedAt) after processing
+    Update last processed commit and optionally the branch after processing
     """
     sql_query = """
     UPDATE git.repositories
         SET "lastProcessedCommit" = $1,
+        "branch" = $2,
         "updatedAt" = NOW()
-    WHERE id = $2
+    WHERE id = $3
     """
-    result = await execute(sql_query, (commit_hash, repo_id))
+    result = await execute(sql_query, (commit_hash, branch, repo_id))
     return str(result)
 
 

services/apps/git_integration/src/crowdgit/models/clone_batch.py

@@ -19,6 +19,9 @@ class CloneBatchInfo(BaseModel):
         default=None,
         description="The edge commit from the previous batch, used to track progress during incremental processing.",
     )
+    clone_with_batches: bool = Field(
+        default=True, description="Whether repo is cloned with batches"
+    )
 
     class Config:
         """Pydantic configuration"""

services/apps/git_integration/src/crowdgit/models/repository.py

@@ -28,6 +28,10 @@ class Repository(BaseModel):
         None,
         description="Timestamp of when the repository maintainer processing was last executed",
     )
+    branch: str | None = Field(
+        None,
+        description="The default branch being tracked for this repository (e.g., main, master, develop)",
+    )
     created_at: datetime = Field(..., description="Creation timestamp")
     updated_at: datetime = Field(..., description="Last update timestamp")
 

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

@@ -13,7 +13,12 @@
 from crowdgit.errors import CommandExecutionError, CrowdGitError
 from crowdgit.models import CloneBatchInfo, Repository, ServiceExecution
 from crowdgit.services.base.base_service import BaseService
-from crowdgit.services.utils import get_default_branch, get_repo_name, run_shell_command
+from crowdgit.services.utils import (
+    get_default_branch,
+    get_remote_default_branch,
+    get_repo_name,
+    run_shell_command,
+)
 
 DEFAULT_CLONE_BATCH_DEPTH = 10
 DEFAULT_STORAGE_OPTIMIZATION_THRESHOLD_MB = 2000
@@ -47,9 +52,9 @@ async def _check_if_final_batch(self, path: str, target_commit_hash: str | None)
         except CommandExecutionError:
             return False
 
-    async def _init_minimal_clone(self, path: str, remote: str) -> None:
+    async def _perform_minimal_clone(self, path: str, remote: str) -> None:
         """
-        Inits minimal clone of depth=1
+        Perform minimal clone of depth=1
         """
         # increasing post buffer to avoid RPC failed error
         await run_shell_command(
@@ -140,6 +145,7 @@ async def _update_batch_info(
         For batched clones (clone_with_batches=True): Checks if target commit reached or full history fetched.
         """
         batch_info.repo_path = repo_path
+        batch_info.clone_with_batches = clone_with_batches
 
         if batch_info.is_first_batch:
             # Set latest commit only from first batch
@@ -271,16 +277,86 @@ async def _calculate_batch_depth(self, repo_path: str, remote: str) -> int:
         )
         return calculated_depth
 
-    async def _clone_repo(self, repo_path: str, remote: str):
-        """Perform full repository clone for new repositories that haven't been processed before"""
+    async def _perform_full_clone(self, repo_path: str, remote: str):
+        """Perform full repository clone"""
+        self.logger.info(f"Performing full clone for repo {remote}...")
         await run_shell_command(["git", "clone", remote, repo_path], cwd=repo_path)
         self.logger.info(f"Successfully completed full clone of repository: {remote}")
 
+    async def has_default_branch_changed(self, remote: str, saved_branch: str | None) -> bool:
+        """Check if the default branch has changed compared to the saved branch
+        Args:
+            remote: The remote repository URL
+            saved_branch: The branch currently saved in the database (can be None)
+        Returns:
+            True if default branch has changed and requires re-cloning, False otherwise
+        """
+        try:
+            remote_default_branch = await get_remote_default_branch(remote)
+
+            if remote_default_branch is None:
+                self.logger.warning(f"Could not determine default branch for {remote}")
+                return False
+
+            if saved_branch is None:
+                self.logger.info(f"No saved branch for {remote} assuming it's not changed")
+                return False
+
+            if saved_branch != remote_default_branch:
+                self.logger.info(
+                    f"Branch changed for {remote}: saved='{saved_branch}' vs remote='{remote_default_branch}'"
+                )
+                return True
+
+            self.logger.debug(f"Branch unchanged for {remote}: {saved_branch}")
+            return False
+
+        except Exception as e:
+            self.logger.error(f"Error validating branch for {remote}: {e}")
+            # On error, assume no change to avoid unnecessary re-cloning
+            return False
+
+    async def determine_clone_strategy(
+        self, repo_path: str, remote: str, branch: str | None, last_processed_commit: str | None
+    ) -> bool:
+        """Determine whether to use full clone or minimal clone strategy based on repository state.
+
+        Args:
+            repo_path: Local path where repository will be cloned
+            remote: Remote repository URL (e.g., 'https://github.com/user/repo')
+            branch: Current saved branch name or None for new repositories
+            last_processed_commit: Last processed commit hash or None for new repositories
+
+        Returns: (clone_with_batches)
+            bool: False for full clone (clone_with_batches=False), True for minimal clone (clone_with_batches=True)
+
+        Strategy:
+            - Full clone: New repositories (last_processed_commit=None) or branch changed
+            - Minimal clone: Existing repositories with unchanged branch for incremental processing
+        """
+
+        self.logger.info(
+            f"Starting clone decision for {remote} (branch: {branch}, last_commit: {last_processed_commit})"
+        )
+
+        default_branch_changed = await self.has_default_branch_changed(remote, branch)
+
+        if not last_processed_commit or default_branch_changed:
+            reason = "new repository" if not last_processed_commit else "branch changed"
+            self.logger.info(f"Performing full clone for {remote} - reason: {reason}")
+            await self._perform_full_clone(repo_path, remote)
+            return False
+
+        self.logger.info(
+            f"Performing minimal clone for {remote} - existing repository with unchanged branch"
+        )
+        await self._perform_minimal_clone(repo_path, remote)
+        return True
+
     async def clone_batches_generator(
         self,
         repository: Repository,
         working_dir_cleanup: bool | None = False,
-        clone_with_batches: bool | None = True,
     ) -> AsyncIterator[CloneBatchInfo]:
         """
         Async generator that yields CloneBatchInfo for repository cloning.
@@ -302,22 +378,15 @@ async def clone_batches_generator(
             remote=remote,
             is_final_batch=False,
             is_first_batch=True,
-            total_commits_count=0,
         )
         try:
             temp_repo_path = tempfile.mkdtemp(prefix=f"{get_repo_name(remote)}_")
-            batch_info.repo_path = temp_repo_path
             batch_start_time = time.time()
 
-            if not clone_with_batches:
-                self.logger.info(f"Performing full clone for repo: {remote}")
-                await self._clone_repo(temp_repo_path, remote)
-            else:
-                # Incremental processing: start with minimal clone and fetch in batches
-                self.logger.info(
-                    f"Performing incremental batched clone for existing repository: {remote}"
-                )
-                await self._init_minimal_clone(temp_repo_path, remote)
+            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

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

@@ -115,7 +115,6 @@ async def process_single_batch_commits(
         self,
         repository: Repository,
         batch_info: CloneBatchInfo,
-        clone_with_batches: bool,
     ) -> None:
         """
         Process commits from a cloned batch.
@@ -147,7 +146,7 @@ async def process_single_batch_commits(
             )
             raw_commits = await self._execute_git_log(
                 batch_info.repo_path,
-                clone_with_batches,
+                batch_info.clone_with_batches,
                 batch_info.prev_batch_edge_commit,
                 batch_info.edge_commit,
             )

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

@@ -89,6 +89,42 @@ def get_repo_name(remote: str) -> str:
     return "-".join(parts)
 
 
+async def get_remote_default_branch(remote_url: str) -> str | None:
+    """Get the default branch of a remote repository without cloning
+
+    Args:
+        remote_url: The URL of the remote repository.
+
+    Returns:
+        The default branch name, or None if unable to determine.
+    """
+    try:
+        # Use git ls-remote to get the symbolic reference for HEAD
+        output = await run_shell_command(["git", "ls-remote", "--symref", remote_url, "HEAD"])
+
+        # Parse the output to find the symbolic reference
+        # Output format: "ref: refs/heads/main\tHEAD\n<commit_hash>\tHEAD"
+        lines = output.strip().split("\n")
+        for line in lines:
+            if line.startswith("ref: refs/heads/"):
+                # Extract branch name from "ref: refs/heads/main"
+                return line.split("refs/heads/")[-1].split("\t")[0]
+
+        # 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
+            except CommandExecutionError:
+                continue
+
+        return None
+
+    except CommandExecutionError as e:
+        logger.warning(f"Failed to get remote default branch for {remote_url}: {e}")
+        return None
+
+
 async def get_default_branch(repo_path: str) -> str:
     """Get the default branch of the repository using local repo
 

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

@@ -19,7 +19,7 @@
     QueueService,
     SoftwareValueService,
 )
-from crowdgit.services.utils import get_repo_name
+from crowdgit.services.utils import get_default_branch, get_repo_name
 from crowdgit.settings import WORKER_ERROR_BACKOFF_SEC, WORKER_POLLING_INTERVAL_SEC
 
 
@@ -142,27 +142,24 @@ async def _process_single_repository(self, repository: Repository):
         try:
             repo_name = get_repo_name(repository.url)
             self._bind_repository_context(repository, repo_name)
-            # Use full clone for new repositories (no last_processed_commit), batched clone for existing ones
-            clone_with_batches = True if repository.last_processed_commit else False
-            logger.info(
-                f"Starting repository cloning for {repo_name} with batching={clone_with_batches}"
-            )
             async for batch_info in self.clone_service.clone_batches_generator(
                 repository,
                 working_dir_cleanup=True,
-                clone_with_batches=clone_with_batches,
             ):
                 logger.info(f"Clone batch info: {batch_info}")
                 if batch_info.is_first_batch:
                     await self.software_value_service.run(repository.id, batch_info.repo_path)
                     await self.maintainer_service.process_maintainers(repository, batch_info)
                 await self.commit_service.process_single_batch_commits(
-                    repository, batch_info, clone_with_batches
+                    repository,
+                    batch_info,
                 )
 
                 if batch_info.is_final_batch:
                     await update_last_processed_commit(
-                        repo_id=repository.id, commit_hash=batch_info.latest_commit_in_repo
+                        repo_id=repository.id,
+                        commit_hash=batch_info.latest_commit_in_repo,
+                        branch=await get_default_branch(batch_info.repo_path),
                     )
 
             logger.info("Incremental processing completed successfully")