feat: support edge case of forked repo having frequent force pushes (CM-745) (#3663)

Author: mbani01

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

@@ -32,7 +32,7 @@ async def insert_repository(url: str, priority: int = 0) -> str:
 async def get_repository_by_url(url: str) -> dict[str, Any] | None:
     """Get repository by URL"""
     sql_query = """
-    SELECT id, url, state, priority, "lastProcessedAt", "lockedAt", "createdAt", "updatedAt", "maintainerFile", "forkedFrom"
+    SELECT id, url, state, priority, "lastProcessedAt", "lockedAt", "createdAt", "updatedAt", "maintainerFile", "forkedFrom", "stuckRequiresReOnboard"
     FROM git.repositories
     WHERE url = $1 AND "deletedAt" IS NULL
     """
@@ -49,7 +49,7 @@ async def get_recently_processed_repository_by_url(url: str) -> Repository | Non
     Used to check if a repository needs reprocessing based on the update interval.
     """
     sql_query = """
-    SELECT id, url, state, priority, "lastProcessedAt", "lockedAt", "createdAt", "updatedAt", "maintainerFile", "forkedFrom", "segmentId"
+    SELECT id, url, state, priority, "lastProcessedAt", "lockedAt", "createdAt", "updatedAt", "maintainerFile", "forkedFrom", "segmentId", "stuckRequiresReOnboard"
     FROM git.repositories
     WHERE url = $1
         AND "deletedAt" IS NULL
@@ -88,7 +88,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", "branch", "forkedFrom"
+    RETURNING id, url, state, priority, "lastProcessedAt", "lastProcessedCommit", "lockedAt", "createdAt", "updatedAt", "segmentId", "integrationId", "maintainerFile", "lastMaintainerRunAt", "branch", "forkedFrom", "stuckRequiresReOnboard"
     """
     return await acquire_repository(
         onboarding_repo_sql_query,
@@ -138,7 +138,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", "branch", "forkedFrom"
+    RETURNING id, url, state, priority, "lastProcessedAt", "lastProcessedCommit", "lockedAt", "createdAt", "updatedAt", "segmentId", "integrationId", "maintainerFile", "lastMaintainerRunAt", "branch", "forkedFrom", "stuckRequiresReOnboard"
     """
     states_to_exclude = (
         RepositoryState.PENDING,

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

@@ -41,6 +41,10 @@ class Repository(BaseModel):
     parent_repo: Repository | None = Field(
         None, description="The parent repository (in case of fork) object from our database"
     )
+    stuck_requires_re_onboard: bool = Field(
+        default=False,
+        description="Indicates if the stuck repository is resolved by a re-onboarding",
+    )
     created_at: datetime = Field(..., description="Creation timestamp")
     updated_at: datetime = Field(..., description="Last update timestamp")
 
@@ -67,6 +71,7 @@ def from_db(cls, db_data: dict[str, Any]) -> Repository:
             "maintainerFile": "maintainer_file",
             "lastMaintainerRunAt": "last_maintainer_run_at",
             "forkedFrom": "forked_from",
+            "stuckRequiresReOnboard": "stuck_requires_re_onboard",
         }
         for db_field, model_field in field_mapping.items():
             if db_field in repo_data:

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

@@ -137,15 +137,7 @@ async def process_single_batch_commits(
                 repository.last_processed_commit,
             )
 
-            await self._process_activities_from_commits(
-                raw_commits,
-                batch_info.repo_path,
-                batch_info.edge_commit,
-                batch_info.remote,
-                repository.segment_id,
-                repository.integration_id,
-                repository.parent_repo,
-            )
+            await self._process_activities_from_commits(raw_commits, batch_info, repository)
 
             batch_end_time = time.time()
             batch_time = round(batch_end_time - batch_start_time, 2)
@@ -616,14 +608,14 @@ def create_activities_from_commit(
 
         return activities_db, activities_queue
 
-    async def _filter_parent_repo_activities(
+    async def _filter_existing_activities(
         self,
         activities_db: list[tuple],
         activities_queue: list[dict],
-        parent_repo: Repository,
+        source_repo: Repository,
     ) -> tuple[list[tuple], list[dict], int]:
         """
-        Filter out activities that exist in parent repo for forked repositories.
+        Filter out activities that exist in specific repo, used for both forked and frequently reonboarded repos.
         Done in post-processing phase using batch lookup to avoid N+1 queries.
 
         Returns: (filtered_activities_db, filtered_activities_queue, skipped_activities_count)
@@ -639,8 +631,8 @@ async def _filter_parent_repo_activities(
         # Batch check which activities exist in parent repo
         parent_source_ids = await batch_check_parent_activities(
             activity_keys,
-            parent_repo.url,
-            parent_repo.segment_id,
+            source_repo.url,
+            source_repo.segment_id,
         )
 
         if not parent_source_ids:
@@ -664,20 +656,16 @@ async def _filter_parent_repo_activities(
 
         if skipped_activities_count > 0:
             self.logger.info(
-                f"Filtered out {skipped_activities_count} activities from parent repo {parent_repo.url}"
+                f"Filtered out {skipped_activities_count} existing activity from {source_repo.url}"
             )
 
         return filtered_activities_db, filtered_activities_queue, skipped_activities_count
 
     async def process_commits_chunk(
         self,
         commit_texts_chunk: list[str | None],
-        repo_path: str,
-        edge_commit_hash: str | None,
-        remote: str,
-        segment_id: str,
-        integration_id: str,
-        parent_repo: Repository | None,
+        batch_info: CloneBatchInfo,
+        repository: Repository,
     ) -> None:
         """
         Process a chunk of raw commit texts into activities and write them to DB and Kafka.
@@ -697,15 +685,15 @@ async def process_commits_chunk(
         commit = None
 
         for full_commit_text in commit_texts_chunk:
-            if self.should_skip_commit(full_commit_text, edge_commit_hash):
+            if self.should_skip_commit(full_commit_text, batch_info.edge_commit):
                 continue
             commit_text, numstats_text = full_commit_text.split(self.NUMSTAT_SPLITTER)
             commit_lines = commit_text.strip().splitlines()
             del full_commit_text
             del commit_text
             if not self._validate_commit_structure(commit_lines):
                 self.logger.warning(
-                    f"Invalid commit structure in {repo_path}: {len(commit_lines)} fields"
+                    f"Invalid commit structure in {batch_info.repo_path}: {len(commit_lines)} fields"
                 )
                 bad_commits += 1
                 del commit_lines
@@ -716,7 +704,7 @@ async def process_commits_chunk(
                 commit = self._construct_commit_dict(commit_lines, numstats_text)
                 if self._validate_commit_data(commit):
                     activity_db_records, activity_kafka = self.create_activities_from_commit(
-                        remote, commit, segment_id, integration_id
+                        batch_info.remote, commit, repository.segment_id, repository.integration_id
                     )
                     activities_db.extend(activity_db_records)
                     activities_queue.extend(activity_kafka)
@@ -727,7 +715,7 @@ async def process_commits_chunk(
                     bad_commits += 1
 
             except Exception as e:
-                self.logger.warning(f"Failed to parse commit in {repo_path}: {e}")
+                self.logger.warning(f"Failed to parse commit in {batch_info.repo_path}: {e}")
                 bad_commits += 1
                 continue
             finally:
@@ -737,17 +725,26 @@ async def process_commits_chunk(
 
         # Filter out activities from parent repo (for forks)
         skipped_activities = 0
-        if parent_repo:
+        if repository.parent_repo:
             (
                 activities_db,
                 activities_queue,
                 skipped_activities,
-            ) = await self._filter_parent_repo_activities(
-                activities_db, activities_queue, parent_repo
+            ) = await self._filter_existing_activities(
+                activities_db, activities_queue, repository.parent_repo
+            )
+        if repository.stuck_requires_re_onboard:
+            self.logger.info(
+                f"Frequent re-onboardings detected! excluding existing activities from repo: {repository.url}"
             )
+            (
+                activities_db,
+                activities_queue,
+                skipped_activities,
+            ) = await self._filter_existing_activities(activities_db, activities_queue, repository)
 
         self.logger.info(
-            f"Processed {processed_commits} commits, skipped {bad_commits} invalid commits, filtered {skipped_activities} activities from parent repo in {repo_path}"
+            f"Processed {processed_commits} commits, skipped {bad_commits} invalid commits, filtered {skipped_activities} activities from parent repo in {batch_info.repo_path}"
         )
         # Update metrics context
         if self._metrics_context:
@@ -766,14 +763,7 @@ async def process_commits_chunk(
         del activities_db, activities_queue
 
     async def _process_activities_from_commits(
-        self,
-        raw_commits: str,
-        repo_path: str,
-        edge_commit_hash: str | None,
-        remote: str,
-        segment_id: str,
-        integration_id: str,
-        parent_repo: Repository | None = None,
+        self, raw_commits: str, batch_info: CloneBatchInfo, repository: Repository
     ):
         """
         Parse raw git log output, process commits into activities, and save to database.
@@ -815,12 +805,8 @@ async def process_single_chunk(chunk_start_idx: int, chunk_end_idx: int):
                     # Process chunk and write to DB/Kafka
                     await self.process_commits_chunk(
                         chunk,
-                        repo_path,
-                        edge_commit_hash,
-                        remote,
-                        segment_id,
-                        integration_id,
-                        parent_repo,
+                        batch_info,
+                        repository,
                     )
                     completed_chunks += 1
                     self.logger.info(f"Progress: {completed_chunks}/{total_chunks} chunks")

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

@@ -115,7 +115,7 @@ async def _ensure_repo_not_stuck(self, repository: Repository):
             logger.warning(
                 f"Repo {repository.url} is stuck for {processing_duration_hours} hours!"
             )
-            if repository.forked_from == repository.url:
+            if repository.stuck_requires_re_onboard:
                 logger.warning(
                     f"Repo {repository.url} is stuck due to force-push or dangling commit. Will be re-onboarded"
                 )
@@ -197,10 +197,6 @@ 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..."
         )