check uploads for notify (#558)

* check uploads for notify

* update tests

* add test

* add changes back

* one more test

* use db for truth

* tests

* cleanup test

Author: maxweng-sentry

apps/worker/tasks/manual_trigger.py

@@ -92,10 +92,12 @@ def process_impl_within_lock(
                 | (CommitReport.report_type == ReportType.COVERAGE.value),
             )
         )
+
         still_processing = 0
         for upload in uploads:
             if not upload.state or upload.state_id == UploadState.UPLOADED.db_id:
                 still_processing += 1
+
         if still_processing == 0:
             self.trigger_notifications(repoid, commitid, commit_yaml)
             if commit.pullid:
@@ -108,18 +110,23 @@ def process_impl_within_lock(
             # reschedule the task
             try:
                 log.info(
-                    "Retrying ManualTriggerTask. Some uploads are still being processed."
+                    "Retrying ManualTriggerTask. Some uploads are still being processed.",
+                    extra={
+                        "repoid": repoid,
+                        "commitid": commitid,
+                        "uploads_still_processing": still_processing,
+                    },
                 )
                 retry_in = 60 * 3**self.request.retries
                 self.retry(max_retries=5, countdown=retry_in)
             except MaxRetriesExceededError:
                 log.warning(
                     "Not attempting to wait for all uploads to get processed since we already retried too many times",
                     extra={
-                        "repoid": commit.repoid,
                         "commit": commit.commitid,
                         "max_retries": 5,
                         "next_countdown_would_be": retry_in,
+                        "repoid": commit.repoid,
                     },
                 )
                 return {

apps/worker/tasks/tests/unit/test_manual_trigger.py

@@ -34,7 +34,12 @@ def test_manual_upload_completion_trigger(
 
         dbsession.add(commit)
 
-        upload = UploadFactory.create(report__commit=commit)
+        # Upload is complete in database (state = PROCESSED)
+        upload = UploadFactory.create(
+            report__commit=commit,
+            state="processed",
+            state_id=UploadState.PROCESSED.db_id,
+        )
         compared_to = CommitFactory.create(repository=commit.repository)
         pull.compared_to = compared_to.commitid
 
@@ -105,3 +110,45 @@ def test_manual_upload_completion_trigger_uploads_still_processing(
                 "notifications_called": False,
                 "message": "Uploads are still in process and the task got retired so many times. Not triggering notifications.",
             } == result
+
+    def test_manual_upload_completion_trigger_with_uploaded_state(
+        self,
+        mocker,
+        mock_configuration,
+        dbsession,
+        mock_storage,
+        mock_redis,
+        celery_app,
+    ):
+        """Test that task retries when DB shows uploads in UPLOADED state"""
+        mocker.patch.object(
+            ManualTriggerTask,
+            "app",
+            celery_app,
+        )
+        commit = CommitFactory.create()
+        # One upload is still in UPLOADED state (being processed)
+        upload1 = UploadFactory.create(
+            report__commit=commit,
+            state="started",
+            state_id=UploadState.UPLOADED.db_id,
+        )
+        # Another upload is complete
+        upload2 = UploadFactory.create(
+            report__commit=commit,
+            state="processed",
+            state_id=UploadState.PROCESSED.db_id,
+        )
+        dbsession.add(commit)
+        dbsession.add(upload1)
+        dbsession.add(upload2)
+        dbsession.flush()
+
+        # Should retry because DB shows upload still in UPLOADED state
+        with pytest.raises(Retry):
+            ManualTriggerTask().run_impl(
+                dbsession,
+                repoid=commit.repoid,
+                commitid=commit.commitid,
+                current_yaml={},
+            )

apps/worker/tasks/tests/unit/test_upload_finisher_task.py

@@ -1,3 +1,4 @@
+from datetime import UTC, datetime
 from pathlib import Path
 from unittest.mock import ANY, call
 
@@ -180,6 +181,7 @@ def test_upload_finisher_task_call(
             branch="thisbranch",
             ci_passed=True,
             repository__branch="thisbranch",
+            repository__updatestamp=None,
             repository__author__unencrypted_oauth_token="testulk3d54rlhxkjyzomq2wh8b7np47xabcrkx8",
             repository__author__username="ThiagoCodecov",
             repository__author__service="github",
@@ -208,6 +210,11 @@ def test_upload_finisher_task_call(
         dbsession.refresh(commit)
         assert commit.message == "dsidsahdsahdsa"
 
+        # Verify repository timestamp is updated
+        dbsession.refresh(commit.repository)
+        assert commit.repository.updatestamp is not None
+        assert (datetime.now(tz=UTC) - commit.repository.updatestamp).seconds < 60
+
         mock_checkpoint_submit.assert_any_call(
             "batch_processing_duration",
             UploadFlow.INITIAL_PROCESSING_COMPLETE,
@@ -394,7 +401,7 @@ def test_upload_finisher_task_call_different_branch(
             ]
         )
 
-    def test_should_call_notifications(self, dbsession):
+    def test_should_call_notifications(self, dbsession, mocker):
         commit_yaml = {"codecov": {"max_report_age": "1y ago"}}
         commit = CommitFactory.create(
             message="dsidsahdsahdsa",
@@ -432,7 +439,7 @@ def test_should_call_notifications_manual_trigger(self, dbsession):
             == ShouldCallNotifyResult.DO_NOT_NOTIFY
         )
 
-    def test_should_call_notifications_manual_trigger_off(self, dbsession):
+    def test_should_call_notifications_manual_trigger_off(self, dbsession, mocker):
         commit_yaml = {
             "codecov": {"max_report_age": "1y ago", "notify": {"manual_trigger": False}}
         }
@@ -463,7 +470,7 @@ def test_should_call_notifications_manual_trigger_off(self, dbsession):
         ],
     )
     def test_should_call_notifications_no_successful_reports(
-        self, dbsession, notify_error, result
+        self, dbsession, mocker, notify_error, result
     ):
         commit_yaml = {
             "codecov": {
@@ -544,6 +551,44 @@ def test_should_call_notifications_more_than_enough_builds(self, dbsession, mock
             == ShouldCallNotifyResult.NOTIFY
         )
 
+    def test_should_call_notifications_with_pending_uploads_in_db(
+        self, dbsession, mocker
+    ):
+        """Test that notifications are not called when DB shows pending uploads"""
+        commit_yaml = {"codecov": {"max_report_age": "1y ago"}}
+        commit = CommitFactory.create(
+            message="dsidsahdsahdsa",
+            commitid="abf6d4df662c47e32460020ab14abf9303581429",
+            repository__author__unencrypted_oauth_token="testulk3d54rlhxkjyzomq2wh8b7np47xabcrkx8",
+            repository__author__username="ThiagoCodecov",
+            repository__yaml=commit_yaml,
+        )
+        # Create uploads in UPLOADED state (still being processed)
+        upload1 = UploadFactory.create(
+            report__commit=commit,
+            state="started",
+            state_id=UploadState.UPLOADED.db_id,
+        )
+        upload2 = UploadFactory.create(
+            report__commit=commit,
+            state="started",
+            state_id=UploadState.UPLOADED.db_id,
+        )
+        dbsession.add(commit)
+        dbsession.add(upload1)
+        dbsession.add(upload2)
+        dbsession.flush()
+
+        assert (
+            UploadFinisherTask().should_call_notifications(
+                commit,
+                commit_yaml,
+                [{"arguments": {"url": "url"}, "successful": True}],
+                db_session=dbsession,
+            )
+            == ShouldCallNotifyResult.DO_NOT_NOTIFY
+        )
+
     def test_finish_reports_processing(self, dbsession, mocker, mock_self_app):
         commit_yaml = {}
         commit = CommitFactory.create(

apps/worker/tasks/upload_finisher.py

@@ -27,7 +27,7 @@
     load_intermediate_reports,
 )
 from services.processing.merging import merge_reports, update_uploads
-from services.processing.state import ProcessingState, should_trigger_postprocessing
+from services.processing.state import ProcessingState
 from services.processing.types import ProcessingResult
 from services.report import ReportService
 from services.repository import get_repo_provider_service
@@ -319,8 +319,24 @@ def run_impl(
                 state,
             )
 
-            if not should_trigger_postprocessing(state.get_upload_numbers()):
-                log.info("run_impl: Postprocessing should not be triggered")
+            # Check if there are still unprocessed uploads in the database
+            # Use DB as source of truth - if any uploads are still in UPLOADED state,
+            # another finisher will process them and we shouldn't send notifications yet
+            remaining_uploads = (
+                db_session.query(Upload)
+                .join(Upload.report)
+                .filter(
+                    Upload.report.has(commit=commit),
+                    Upload.state_id == UploadState.UPLOADED.db_id,
+                )
+                .count()
+            )
+
+            if remaining_uploads > 0:
+                log.info(
+                    "run_impl: Postprocessing should not be triggered - uploads still pending",
+                    extra={"remaining_uploads": remaining_uploads},
+                )
                 UploadFlow.log(UploadFlow.PROCESSING_COMPLETE)
                 UploadFlow.log(UploadFlow.SKIPPING_NOTIFICATION)
                 self._call_upload_breadcrumb_task(
@@ -556,7 +572,7 @@ def finish_reports_processing(
         notifications_called = False
         if not regexp_ci_skip.search(commit.message or ""):
             should_call_notifications = self.should_call_notifications(
-                commit, commit_yaml, processing_results
+                commit, commit_yaml, processing_results, db_session
             )
             log.info(
                 "finish_reports_processing: should_call_notifications",
@@ -667,6 +683,7 @@ def should_call_notifications(
         commit: Commit,
         commit_yaml: UserYaml,
         processing_results: list[ProcessingResult],
+        db_session=None,
     ) -> ShouldCallNotifyResult:
         extra_dict = {
             "repoid": commit.repoid,
@@ -676,6 +693,25 @@ def should_call_notifications(
             "parent_task": self.request.parent_id,
         }
 
+        # Check if there are still pending uploads in the database
+        # Use DB as source of truth for upload completion status
+        if db_session:
+            remaining_uploads = (
+                db_session.query(Upload)
+                .join(Upload.report)
+                .filter(
+                    Upload.report.has(commit=commit),
+                    Upload.state_id == UploadState.UPLOADED.db_id,
+                )
+                .count()
+            )
+            if remaining_uploads > 0:
+                log.info(
+                    "Not scheduling notify because there are still pending uploads",
+                    extra={**extra_dict, "remaining_uploads": remaining_uploads},
+                )
+                return ShouldCallNotifyResult.DO_NOT_NOTIFY
+
         manual_trigger = read_yaml_field(
             commit_yaml, ("codecov", "notify", "manual_trigger")
         )