feat(worker): schedule one upload_finisher via redis gate

apps/worker/services/processing/processing.py

@@ -5,20 +5,32 @@
 from celery.exceptions import CeleryError
 from sqlalchemy.orm import Session as DbSession
 
+from app import celery_app
 from database.models.core import Commit
 from database.models.reports import Upload
+from helpers.checkpoint_logger import _kwargs_key
+from helpers.checkpoint_logger.flows import UploadFlow
 from helpers.reports import delete_archive_setting
 from services.report import ProcessingError, RawReportInfo, ReportService
 from services.report.parser.types import VersionOneParsedRawReport
 from shared.api_archive.archive import ArchiveService
+from shared.celery_config import upload_finisher_task_name
+from shared.helpers.redis import get_redis_connection
 from shared.yaml import UserYaml
 
 from .intermediate import save_intermediate_report
-from .state import ProcessingState
+from .state import ProcessingState, should_perform_merge
 from .types import ProcessingResult, UploadArguments
 
 log = logging.getLogger(__name__)
 
+FINISHER_GATE_KEY_PREFIX = "upload_merger_lock"
+FINISHER_GATE_TTL_SECONDS = 900
+
+
+def finisher_gate_key(repo_id: int, commit_sha: str) -> str:
+    return f"{FINISHER_GATE_KEY_PREFIX}_{repo_id}_{commit_sha}"
+
 
 @sentry_sdk.trace
 def process_upload(
@@ -69,6 +81,34 @@ def process_upload(
             save_intermediate_report(upload_id, processing_result.report)
         state.mark_upload_as_processed(upload_id)
 
+        upload_numbers = state.get_upload_numbers()
+        if should_perform_merge(upload_numbers):
+            gate_key = finisher_gate_key(repo_id, commit_sha)
+            redis = get_redis_connection()
+            if redis.set(gate_key, "1", nx=True, ex=FINISHER_GATE_TTL_SECONDS):
+                log.info(
+                    "Enqueuing upload finisher via gate",
+                    extra={
+                        "repo_id": repo_id,
+                        "commit_sha": commit_sha,
+                        "upload_id": upload_id,
+                        "gate_key": gate_key,
+                    },
+                )
+                finisher_kwargs = {
+                    "repoid": repo_id,
+                    "commitid": commit_sha,
+                    "commit_yaml": commit_yaml.to_dict(),
+                }
+                checkpoint_kwargs_key = _kwargs_key(UploadFlow)
+                if checkpoint_kwargs_key in arguments:
+                    finisher_kwargs[checkpoint_kwargs_key] = arguments[
+                        checkpoint_kwargs_key
+                    ]
+                celery_app.tasks[upload_finisher_task_name].apply_async(
+                    kwargs=finisher_kwargs
+                )
+
         rewrite_or_delete_upload(archive_service, commit_yaml, report_info)
 
     except CeleryError:

apps/worker/services/tests/test_processing.py

@@ -13,8 +13,10 @@
 
 
 @pytest.mark.django_db(databases={"default"})
-class TestProcessUpload:
-    def test_processor_does_not_enqueue_finisher(self, dbsession, mocker, mock_storage):
+class TestProcessUploadFinisherGate:
+    def test_triggers_finisher_when_gate_is_acquired(
+        self, dbsession, mocker, mock_storage
+    ):
         # Setup
         repository = RepositoryFactory.create()
         commit = CommitFactory.create(repository=repository)
@@ -54,6 +56,92 @@ def test_processor_does_not_enqueue_finisher(self, dbsession, mocker, mock_stora
             return_value=mock_state_instance,
         )
 
+        # Mock celery app to capture finisher task scheduling
+        mock_celery_app = mocker.patch("services.processing.processing.celery_app")
+        mock_finisher_task = MagicMock()
+        mock_celery_app.tasks = {"app.tasks.upload.UploadFinisher": mock_finisher_task}
+        mock_redis = mocker.patch("services.processing.processing.get_redis_connection")
+        mock_redis.return_value.set.return_value = True
+
+        # Mock other dependencies
+        mocker.patch("services.processing.processing.save_intermediate_report")
+        mocker.patch("services.processing.processing.rewrite_or_delete_upload")
+
+        commit_yaml = UserYaml({})
+
+        # Execute
+        result = process_upload(
+            on_processing_error=lambda error: None,
+            db_session=dbsession,
+            repo_id=repository.repoid,
+            commit_sha=commit.commitid,
+            commit_yaml=commit_yaml,
+            arguments=arguments,
+        )
+
+        # Verify
+        assert result["successful"] is True
+        assert result["upload_id"] == upload.id_
+
+        # Verify finisher was triggered
+        mock_finisher_task.apply_async.assert_called_once_with(
+            kwargs={
+                "repoid": repository.repoid,
+                "commitid": commit.commitid,
+                "commit_yaml": commit_yaml.to_dict(),
+            }
+        )
+        mock_redis.return_value.set.assert_called_once()
+
+    def test_does_not_trigger_finisher_when_gate_exists(
+        self, dbsession, mocker, mock_storage
+    ):
+        # Setup
+        repository = RepositoryFactory.create()
+        commit = CommitFactory.create(repository=repository)
+        upload = UploadFactory.create(
+            report__commit=commit,
+            state="started",
+        )
+        dbsession.add_all([repository, commit, upload])
+        dbsession.flush()
+
+        arguments: UploadArguments = {
+            "commit": commit.commitid,
+            "upload_id": upload.id_,
+            "version": "v4",
+            "reportid": str(upload.report.external_id),
+        }
+
+        # Mock dependencies
+        mock_report_service = mocker.patch(
+            "services.processing.processing.ReportService"
+        )
+        mock_processing_result = MagicMock()
+        mock_processing_result.error = None
+        mock_processing_result.report = None
+        mock_report_service.return_value.build_report_from_raw_content.return_value = (
+            mock_processing_result
+        )
+
+        # Mock ProcessingState to indicate other uploads still processing
+        mock_state_instance = MagicMock()
+        mock_state_instance.get_upload_numbers.return_value = MagicMock(
+            processing=2,
+            processed=1,  # Other uploads still in progress
+        )
+        mocker.patch(
+            "services.processing.processing.ProcessingState",
+            return_value=mock_state_instance,
+        )
+
+        # Mock celery app to capture finisher task scheduling
+        mock_celery_app = mocker.patch("services.processing.processing.celery_app")
+        mock_finisher_task = MagicMock()
+        mock_celery_app.tasks = {"app.tasks.upload.UploadFinisher": mock_finisher_task}
+        mock_redis = mocker.patch("services.processing.processing.get_redis_connection")
+        mock_redis.return_value.set.return_value = False
+
         # Mock other dependencies
         mocker.patch("services.processing.processing.save_intermediate_report")
         mocker.patch("services.processing.processing.rewrite_or_delete_upload")
@@ -74,4 +162,6 @@ def test_processor_does_not_enqueue_finisher(self, dbsession, mocker, mock_stora
         assert result["successful"] is True
         assert result["upload_id"] == upload.id_
 
-        # Finisher enqueue is handled elsewhere; processor no longer triggers it directly.
+        # Verify finisher was NOT triggered.
+        mock_finisher_task.apply_async.assert_not_called()
+        mock_redis.return_value.set.assert_not_called()

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

@@ -40,6 +40,7 @@
 from tasks.upload_finisher import (
     FINISHER_BASE_RETRY_COUNTDOWN_SECONDS,
     FINISHER_BLOCKING_TIMEOUT_SECONDS,
+    FINISHER_MAX_SWEEP_ATTEMPTS,
     ReportService,
     ShouldCallNotifyResult,
     UploadFinisherTask,
@@ -991,6 +992,9 @@ def test_soft_time_limit_handling(self, dbsession, mocker, mock_self_app):
         mocker.patch(
             "tasks.upload_finisher.load_commit_diff", side_effect=SoftTimeLimitExceeded
         )
+        mock_delete_gate = mocker.patch.object(
+            UploadFinisherTask, "_delete_finisher_gate"
+        )
 
         commit = CommitFactory.create()
         dbsession.add(commit)
@@ -1020,6 +1024,7 @@ def test_soft_time_limit_handling(self, dbsession, mocker, mock_self_app):
                 "sentry_trace_id": None,
             }
         )
+        mock_delete_gate.assert_called_once_with(commit.repoid, commit.commitid)
 
     @pytest.mark.django_db
     def test_generic_exception_handling(self, dbsession, mocker, mock_self_app):
@@ -1038,6 +1043,9 @@ def test_generic_exception_handling(self, dbsession, mocker, mock_self_app):
 
         previous_results = [{"upload_id": 0, "successful": True, "arguments": {}}]
 
+        mock_delete_gate = mocker.patch.object(
+            UploadFinisherTask, "_delete_finisher_gate"
+        )
         result = UploadFinisherTask().run_impl(
             dbsession,
             previous_results,
@@ -1074,6 +1082,7 @@ def test_generic_exception_handling(self, dbsession, mocker, mock_self_app):
                 "sentry_trace_id": None,
             }
         )
+        mock_delete_gate.assert_called_once_with(commit.repoid, commit.commitid)
 
     @pytest.mark.django_db
     def test_idempotency_check_skips_already_processed_uploads(
@@ -1115,7 +1124,12 @@ def test_idempotency_check_skips_already_processed_uploads(
             {"upload_id": upload_3.id, "successful": False, "arguments": {}},
         ]
 
-        result = UploadFinisherTask().run_impl(
+        task = UploadFinisherTask()
+        mock_count_pending = mocker.patch.object(
+            task, "_count_remaining_coverage_uploads", return_value=0
+        )
+        mock_delete_gate = mocker.patch.object(task, "_delete_finisher_gate")
+        result = task.run_impl(
             dbsession,
             previous_results,
             repoid=commit.repoid,
@@ -1131,6 +1145,78 @@ def test_idempotency_check_skips_already_processed_uploads(
 
         # Verify that _process_reports_with_lock was NOT called
         mock_process.assert_not_called()
+        mock_count_pending.assert_called_once()
+        mock_delete_gate.assert_called_once()
+
+    @pytest.mark.django_db
+    def test_idempotency_schedules_sweep_when_pending_uploads_remain(
+        self, dbsession, mocker, mock_self_app
+    ):
+        commit = CommitFactory.create()
+        dbsession.add(commit)
+        dbsession.flush()
+
+        report = CommitReport(commit_id=commit.id_)
+        dbsession.add(report)
+        dbsession.flush()
+
+        upload_1 = UploadFactory.create(report=report, state="merged")
+        dbsession.add(upload_1)
+        dbsession.flush()
+
+        task = UploadFinisherTask()
+        mock_count_pending = mocker.patch.object(
+            task, "_count_remaining_coverage_uploads", return_value=2
+        )
+        mock_schedule_sweep = mocker.patch.object(task, "_schedule_sweep")
+        mock_delete_gate = mocker.patch.object(task, "_delete_finisher_gate")
+
+        result = task.run_impl(
+            dbsession,
+            [{"upload_id": upload_1.id, "successful": True, "arguments": {}}],
+            repoid=commit.repoid,
+            commitid=commit.commitid,
+            commit_yaml={},
+        )
+
+        assert result == {"sweep_scheduled": True, "remaining_uploads": 2}
+        mock_count_pending.assert_called_once()
+        mock_schedule_sweep.assert_called_once()
+        mock_delete_gate.assert_not_called()
+
+    @pytest.mark.django_db
+    def test_idempotency_does_not_reschedule_sweep_after_max_attempts(
+        self, dbsession, mocker, mock_self_app
+    ):
+        commit = CommitFactory.create()
+        dbsession.add(commit)
+        dbsession.flush()
+
+        report = CommitReport(commit_id=commit.id_)
+        dbsession.add(report)
+        dbsession.flush()
+
+        upload_1 = UploadFactory.create(report=report, state="merged")
+        dbsession.add(upload_1)
+        dbsession.flush()
+
+        task = UploadFinisherTask()
+        mocker.patch.object(task, "_count_remaining_coverage_uploads", return_value=2)
+        mock_schedule_sweep = mocker.patch.object(task, "_schedule_sweep")
+        mock_delete_gate = mocker.patch.object(task, "_delete_finisher_gate")
+
+        result = task.run_impl(
+            dbsession,
+            [{"upload_id": upload_1.id, "successful": True, "arguments": {}}],
+            repoid=commit.repoid,
+            commitid=commit.commitid,
+            commit_yaml={},
+            sweep_attempt=FINISHER_MAX_SWEEP_ATTEMPTS,
+        )
+
+        assert result == {"sweep_scheduled": False, "remaining_uploads": 2}
+        mock_schedule_sweep.assert_not_called()
+        mock_delete_gate.assert_called_once()
 
     @pytest.mark.django_db
     def test_idempotency_check_proceeds_when_uploads_not_finished(
@@ -1448,6 +1534,66 @@ def mock_process_reports(
         # This is the key assertion - notifications should NOT be blocked by test_results uploads
         mock_handle_finisher_lock.assert_called_once()
 
+    @pytest.mark.django_db
+    def test_run_impl_deletes_gate_after_successful_completion(
+        self, dbsession, mocker, mock_self_app
+    ):
+        commit = CommitFactory.create()
+        dbsession.add(commit)
+        dbsession.flush()
+
+        task = UploadFinisherTask()
+        mocker.patch.object(
+            task,
+            "_reconstruct_processing_results",
+            return_value=[{"upload_id": 1, "successful": True, "arguments": {}}],
+        )
+        mocker.patch.object(task, "_process_reports_with_lock")
+        mocker.patch.object(task, "_count_remaining_coverage_uploads", return_value=0)
+        mocker.patch.object(task, "_handle_finisher_lock", return_value={"done": True})
+        mock_delete_gate = mocker.patch.object(task, "_delete_finisher_gate")
+
+        result = task.run_impl(
+            dbsession,
+            processing_results=[],
+            repoid=commit.repoid,
+            commitid=commit.commitid,
+            commit_yaml={},
+        )
+
+        assert result == {"done": True}
+        mock_delete_gate.assert_called_once()
+
+    @pytest.mark.django_db
+    def test_run_impl_deletes_gate_when_finisher_lock_is_exhausted(
+        self, dbsession, mocker, mock_self_app
+    ):
+        commit = CommitFactory.create()
+        dbsession.add(commit)
+        dbsession.flush()
+
+        task = UploadFinisherTask()
+        mocker.patch.object(
+            task,
+            "_reconstruct_processing_results",
+            return_value=[{"upload_id": 1, "successful": True, "arguments": {}}],
+        )
+        mocker.patch.object(task, "_process_reports_with_lock")
+        mocker.patch.object(task, "_count_remaining_coverage_uploads", return_value=0)
+        mocker.patch.object(task, "_handle_finisher_lock", return_value=None)
+        mock_delete_gate = mocker.patch.object(task, "_delete_finisher_gate")
+
+        result = task.run_impl(
+            dbsession,
+            processing_results=[],
+            repoid=commit.repoid,
+            commitid=commit.commitid,
+            commit_yaml={},
+        )
+
+        assert result is None
+        mock_delete_gate.assert_called_once()
+
 
 class TestLockManagerConfiguration:
     """Tests for lock manager configuration: finite blocking_timeout