feat(git-integration): auto detect stuck commits and handle repos with force-pushes (CM-745) (#3658)

Author: mbani01

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

@@ -140,7 +140,11 @@ async def acquire_recurrent_repo() -> Repository | None:
     )
     RETURNING id, url, state, priority, "lastProcessedAt", "lastProcessedCommit", "lockedAt", "createdAt", "updatedAt", "segmentId", "integrationId", "maintainerFile", "lastMaintainerRunAt", "branch", "forkedFrom"
     """
-    states_to_exclude = (RepositoryState.PENDING, RepositoryState.PROCESSING)
+    states_to_exclude = (
+        RepositoryState.PENDING,
+        RepositoryState.PROCESSING,
+        RepositoryState.STUCK,
+    )
     return await acquire_repository(
         recurrent_repo_sql_query,
         (RepositoryState.PROCESSING, states_to_exclude, REPOSITORY_UPDATE_INTERVAL_HOURS),

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

@@ -20,6 +20,8 @@ class ErrorCode(str, Enum):
     MAINTAINER_INTERVAL_NOT_ELAPSED = "maintainer-interval-not-elapsed"
     CLEANUP_FAILED = "cleanup-failed"
     PARENT_REPO_INVALID = "parent-repo-invalid"
+    REONBOARDING_REQUIRED = "reonboarding-required"
+    STUCK_REPO = "stuck-repo"
 
 
 class RepositoryState(str, Enum):
@@ -30,6 +32,7 @@ class RepositoryState(str, Enum):
     COMPLETED = "completed"
     FAILED = "failed"
     REQUIRES_PARENT = "requires_parent"  # fork repo without valid parent repo in out system
+    STUCK = "stuck"  # requires manual resolution
 
 
 class RepositoryPriority(int):

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

@@ -107,3 +107,15 @@ class MaintainerIntervalNotElapsedError(CrowdGitError):
 class ParentRepoInvalidError(CrowdGitError):
     error_message: str = "Parent repository is not valid or not found"
     error_code: ErrorCode = ErrorCode.PARENT_REPO_INVALID
+
+
+@dataclass
+class ReOnboardingRequiredError(CrowdGitError):
+    error_message = "Repository cannot be processed and requires re-onboarding"
+    error_code: ErrorCode = ErrorCode.REONBOARDING_REQUIRED
+
+
+@dataclass
+class StuckRepoError(CrowdGitError):
+    error_message = "Repos stuck in processing state for a long time"
+    error_code: ErrorCode = ErrorCode.STUCK_REPO

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

@@ -38,3 +38,9 @@ def load_env_var(key: str, required=True, default=None):
 WORKER_SHUTDOWN_TIMEOUT_SEC = int(load_env_var("WORKER_SHUTDOWN_TIMEOUT_SEC", default="3600"))
 MAX_CONCURRENT_ONBOARDINGS = int(load_env_var("MAX_CONCURRENT_ONBOARDINGS", default="3"))
 MAX_INTEGRATION_RESULTS = int(load_env_var("MAX_INTEGRATION_RESULTS", default="5000000"))
+STUCK_ONBOARDING_REPO_TIMEOUT_HOURS = int(
+    load_env_var("STUCK_ONBOARDING_REPO_TIMEOUT_HOURS", default="12")
+)
+STUCK_RECURRENT_REPO_TIMEOUT_HOURS = int(
+    load_env_var("STUCK_RECURRENT_REPO_TIMEOUT_HOURS", default="4")
+)

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

@@ -1,4 +1,5 @@
 import asyncio
+from datetime import datetime, timezone
 
 from crowdgit.database.crud import (
     acquire_repo_for_processing,
@@ -8,11 +9,16 @@
     update_last_processed_commit,
 )
 from crowdgit.enums import RepositoryState
-from crowdgit.errors import InternalError, ParentRepoInvalidError
+from crowdgit.errors import (
+    InternalError,
+    ParentRepoInvalidError,
+    ReOnboardingRequiredError,
+    StuckRepoError,
+)
 
 # Import configured loguru logger from crowdgit.logger
 from crowdgit.logger import logger
-from crowdgit.models.repository import Repository
+from crowdgit.models import Repository
 from crowdgit.services import (
     CloneService,
     CommitService,
@@ -21,7 +27,12 @@
     SoftwareValueService,
 )
 from crowdgit.services.utils import get_default_branch, get_repo_name
-from crowdgit.settings import WORKER_ERROR_BACKOFF_SEC, WORKER_POLLING_INTERVAL_SEC
+from crowdgit.settings import (
+    STUCK_ONBOARDING_REPO_TIMEOUT_HOURS,
+    STUCK_RECURRENT_REPO_TIMEOUT_HOURS,
+    WORKER_ERROR_BACKOFF_SEC,
+    WORKER_POLLING_INTERVAL_SEC,
+)
 
 
 class RepositoryWorker:
@@ -78,6 +89,36 @@ async def shutdown(self):
 
         logger.info("Worker services shutdown triggered")
 
+    async def _ensure_repo_not_stuck(self, repository: Repository):
+        """
+        Check if repo is stuck and raise the appropriate exception if so.
+        Repos can get stuck in processing state for different reasons:
+        - Worker crash or restart (e.g. pod eviction due OOM, deployment after timeout, ...)
+        - `last_processed_commit` is no loger valid due to force-push, dangling-commit, or so...
+        - Race condition: remote is going under breaking changes at the same time we're processing it
+        - Network issues breaking the clone/pull operation
+        """
+        # detection
+        processing_duration_hours = (
+            datetime.now(timezone.utc) - repository.locked_at.astimezone(timezone.utc)
+        ).total_seconds() / 3600
+        repo_stuck: bool = (
+            repository.last_processed_commit
+            and processing_duration_hours >= STUCK_RECURRENT_REPO_TIMEOUT_HOURS
+        ) or (
+            repository.last_processed_commit is None  # onboarding
+            and processing_duration_hours >= STUCK_ONBOARDING_REPO_TIMEOUT_HOURS
+        )
+
+        # handling
+        if repo_stuck and repository.forked_from == repository.url:
+            logger.warning(
+                f"Repo {repository.url} is stuck due to force-push or dangling commit. Will be re-onboarded"
+            )
+            raise ReOnboardingRequiredError()
+
+        raise StuckRepoError()
+
     async def _process_repositories(self):
         """
         Process repositories by priority - check acquire_repo_for_processing()
@@ -153,6 +194,10 @@ async def _validate_and_get_parent_repo(self, repository: Repository) -> Reposit
         if not repository.forked_from:
             return None
 
+        if repository.forked_from == repository.url:
+            # EDGE CASE: not a fork but repo get reonboarded a lot and we treat it as a "fork" to avoid producing tons of duplicate activities
+            return repository.forked_from
+
         logger.info(
             f"Repository {repository.url} is forked from {repository.forked_from}, validating parent repo..."
         )
@@ -200,9 +245,24 @@ async def _process_single_repository(self, repository: Repository):
                         commit_hash=batch_info.latest_commit_in_repo,
                         branch=await get_default_branch(batch_info.repo_path),
                     )
+                else:
+                    await self._ensure_repo_not_stuck(repository)
 
             logger.info("Incremental processing completed successfully")
             processing_state = RepositoryState.COMPLETED
+        except StuckRepoError:
+            logger.error(
+                f"Repo {repository.url} is stuck for unkown reason, marking it as stuck until manually resolved!"
+            )
+            processing_state = RepositoryState.STUCK
+        except ReOnboardingRequiredError:
+            logger.info(f"Resetting and queueing {repository.url} for re-onboarding")
+            await update_last_processed_commit(
+                repo_id=repository.id,
+                commit_hash=None,
+                branch=None,
+            )
+            processing_state = RepositoryState.PENDING
         except ParentRepoInvalidError as e:
             logger.error(f"Parent repo validation failed: {repr(e)}")
             processing_state = RepositoryState.REQUIRES_PARENT