feat(worker): make upload_finisher source-of-truth for commit uploads

apps/worker/services/processing/merging.py

@@ -129,8 +129,8 @@ def update_uploads(
 
         if result["successful"]:
             update = {
-                "state_id": UploadState.PROCESSED.db_id,
-                "state": "processed",
+                "state_id": UploadState.MERGED.db_id,
+                "state": "merged",
             }
             report = reports.get(upload_id)
             if report is not None:

apps/worker/services/processing/processing.py

@@ -5,18 +5,16 @@
 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.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.yaml import UserYaml
 
 from .intermediate import save_intermediate_report
-from .state import ProcessingState, should_trigger_postprocessing
+from .state import ProcessingState
 from .types import ProcessingResult, UploadArguments
 
 log = logging.getLogger(__name__)
@@ -71,27 +69,6 @@ def process_upload(
             save_intermediate_report(upload_id, processing_result.report)
         state.mark_upload_as_processed(upload_id)
 
-        # Check if all uploads are now processed and trigger finisher if needed
-        # This handles the case where a processor task retries outside of a chord
-        # (e.g., from visibility timeout or task_reject_on_worker_lost)
-        upload_numbers = state.get_upload_numbers()
-        if should_trigger_postprocessing(upload_numbers):
-            log.info(
-                "All uploads processed, triggering finisher",
-                extra={
-                    "repo_id": repo_id,
-                    "commit_sha": commit_sha,
-                    "upload_id": upload_id,
-                },
-            )
-            celery_app.tasks[upload_finisher_task_name].apply_async(
-                kwargs={
-                    "repoid": repo_id,
-                    "commitid": commit_sha,
-                    "commit_yaml": commit_yaml.to_dict(),
-                }
-            )
-
         rewrite_or_delete_upload(archive_service, commit_yaml, report_info)
 
     except CeleryError:

apps/worker/services/tests/test_processing.py

@@ -13,27 +13,8 @@
 
 
 @pytest.mark.django_db(databases={"default"})
-class TestProcessUploadOrphanedTaskRecovery:
-    """
-    Tests for the orphaned upload recovery mechanism.
-
-    When a processor task retries outside of a chord (e.g., from visibility timeout
-    or task_reject_on_worker_lost), it should trigger the finisher if all uploads
-    are now processed.
-    """
-
-    def test_triggers_finisher_when_last_upload_completes(
-        self, dbsession, mocker, mock_storage
-    ):
-        """
-        Test that finisher is triggered when the last upload completes processing.
-
-        This simulates the scenario where:
-        1. A processor task died before ACK, missing the chord callback
-        2. Task retries standalone (visibility timeout or worker rejection)
-        3. Completes successfully and notices all uploads are done
-        4. Triggers finisher to complete the upload flow
-        """
+class TestProcessUpload:
+    def test_processor_does_not_enqueue_finisher(self, dbsession, mocker, mock_storage):
         # Setup
         repository = RepositoryFactory.create()
         commit = CommitFactory.create(repository=repository)
@@ -73,105 +54,6 @@ def test_triggers_finisher_when_last_upload_completes(
             return_value=mock_state_instance,
         )
 
-        # Mock should_trigger_postprocessing to return True
-        mocker.patch(
-            "services.processing.processing.should_trigger_postprocessing",
-            return_value=True,
-        )
-
-        # 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 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(),
-            }
-        )
-
-    def test_does_not_trigger_finisher_when_uploads_still_processing(
-        self, dbsession, mocker, mock_storage
-    ):
-        """
-        Test that finisher is NOT triggered when other uploads are still processing.
-
-        This verifies we don't prematurely trigger the finisher when only some
-        uploads have completed.
-        """
-        # 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 should_trigger_postprocessing to return False
-        mocker.patch(
-            "services.processing.processing.should_trigger_postprocessing",
-            return_value=False,
-        )
-
-        # 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 other dependencies
         mocker.patch("services.processing.processing.save_intermediate_report")
         mocker.patch("services.processing.processing.rewrite_or_delete_upload")
@@ -192,5 +74,4 @@ def test_does_not_trigger_finisher_when_uploads_still_processing(
         assert result["successful"] is True
         assert result["upload_id"] == upload.id_
 
-        # Verify finisher was NOT triggered
-        mock_finisher_task.apply_async.assert_not_called()
+        # Finisher enqueue is handled elsewhere; processor no longer triggers it directly.