refactor(worker): remove redis-expiry finisher reconstruction fallback

Drop the started-upload database fallback path that was only needed for Redis-state expiry recovery and simplify reconstruction to rely on merge-ready processing state directly.

Made-with: Cursor

Author: thomasrockhu-codecov

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

@@ -16,7 +16,6 @@
 from helpers.exceptions import RepositoryWithoutValidBotError
 from helpers.log_context import LogContext, set_log_context
 from services.lock_manager import LockRetry
-from services.processing.intermediate import intermediate_report_key
 from services.processing.merging import get_joined_flag, update_uploads
 from services.processing.types import MergeResult, ProcessingResult
 from services.timeseries import MeasurementName
@@ -1259,105 +1258,18 @@ def test_idempotency_check_proceeds_when_uploads_not_finished(
         # Verify that _process_reports_with_lock WAS called
         mock_process.assert_called_once()
 
-    @pytest.mark.django_db
-    def test_reconstruct_processing_results_falls_back_to_database_when_redis_expires(
-        self,
-        dbsession,
-        mocker,
-        mock_storage,
-        mock_repo_provider,
-        mock_redis,
-        mock_self_app,
-    ):
-        """Test that finisher falls back to database when Redis ProcessingState expires.
-
-        This tests the edge case where Redis keys expire after 24h TTL, but uploads
-        were processed and have intermediate reports. The finisher should find them
-        via database query and include them in the final report.
-        """
-        commit = CommitFactory.create()
-        dbsession.add(commit)
-        dbsession.flush()
-
-        report = CommitReport(commit_id=commit.id_)
-        dbsession.add(report)
-        dbsession.flush()
-
-        # Create uploads in "started" state (simulating Redis state expired)
-        upload_1 = UploadFactory.create(
-            report=report, state="started", state_id=UploadState.UPLOADED.db_id
-        )
-        upload_2 = UploadFactory.create(
-            report=report, state="started", state_id=UploadState.UPLOADED.db_id
-        )
-        dbsession.add(upload_1)
-        dbsession.add(upload_2)
-        dbsession.flush()
-
-        # Mock Redis to simulate intermediate reports exist (confirms uploads were processed)
-        mock_redis.exists.side_effect = lambda key: (
-            key == intermediate_report_key(upload_1.id)
-            or key == intermediate_report_key(upload_2.id)
-        )
-
-        # Mock ProcessingState to return empty (simulating Redis expiration)
-        mock_state = mocker.MagicMock()
-        mock_state.get_uploads_for_merging.return_value = set()  # Redis expired
-        mock_state.get_upload_numbers.return_value = mocker.MagicMock(
-            processing=0, processed=0
-        )
-        mocker.patch("tasks.upload_finisher.ProcessingState", return_value=mock_state)
-
-        # Mock the processing methods
-        mocker.patch("tasks.upload_finisher.load_intermediate_reports", return_value=[])
-        mocker.patch("tasks.upload_finisher.update_uploads")
-        mock_process = mocker.patch.object(
-            UploadFinisherTask, "_process_reports_with_lock"
-        )
-
-        # Call run_impl without processing_results to trigger reconstruction
-        task = UploadFinisherTask()
-        task.run_impl(
-            dbsession,
-            processing_results=None,  # Triggers reconstruction
-            repoid=commit.repoid,
-            commitid=commit.commitid,
-            commit_yaml={},
-        )
-
-        # Verify that _find_started_uploads_with_reports was called (via reconstruction)
-        # This is verified by checking that _process_reports_with_lock was called
-        # with processing_results containing our uploads
-        mock_process.assert_called_once()
-        call_args = mock_process.call_args
-        # processing_results is the 4th positional argument (index 0 is args tuple)
-        processing_results = call_args[0][3]
-
-        # Verify both uploads are included in processing_results
-        upload_ids_in_results = {r["upload_id"] for r in processing_results}
-        assert upload_1.id in upload_ids_in_results
-        assert upload_2.id in upload_ids_in_results
-        assert len(processing_results) == 2
-
-        # Verify both are marked as successful (have intermediate reports)
-        assert all(r["successful"] for r in processing_results)
-
     @pytest.mark.django_db
     def test_reconstruct_processing_results_returns_empty_when_no_uploads_found(
         self, dbsession, mocker, mock_redis, mock_self_app
     ):
-        """Test that finisher returns empty list when no uploads found in Redis or DB.
-
-        This tests the edge case where Redis expires AND no uploads exist in database
-        in "started" state with intermediate reports.
-        """
+        """Test that finisher returns empty list when no uploads are merge-ready."""
         commit = CommitFactory.create()
         dbsession.add(commit)
         dbsession.flush()
 
-        # Mock ProcessingState to return empty (simulating Redis expiration)
+        # Mock ProcessingState to return empty merge-ready set.
         mock_state = mocker.MagicMock()
-        mock_state.get_uploads_for_merging.return_value = set()  # Redis expired
+        mock_state.get_uploads_for_merging.return_value = set()
         mocker.patch("tasks.upload_finisher.ProcessingState", return_value=mock_state)
 
         # Call run_impl without processing_results to trigger reconstruction

apps/worker/tasks/upload_finisher.py

@@ -157,93 +157,21 @@ def _count_remaining_coverage_uploads(self, db_session, commit: Commit) -> int:
             .count()
         )
 
-    def _find_started_uploads_with_reports(
-        self, db_session, commit: Commit
-    ) -> set[int]:
-        """Find uploads in "started" state that have intermediate reports in Redis.
-
-        This is the fallback when Redis ProcessingState has expired (TTL: PROCESSING_STATE_TTL).
-        We check the database for uploads that were processed but never finalized,
-        and verify they have intermediate reports before including them.
-        """
-        # Query for uploads in "started" state for this commit
-        started_uploads = (
-            db_session.query(Upload)
-            .join(Upload.report)
-            .filter(
-                Upload.report.has(commit=commit),
-                Upload.state == "started",
-                Upload.state_id == UploadState.UPLOADED.db_id,
-            )
-            .all()
-        )
-
-        if not started_uploads:
-            return set()
-
-        log.info(
-            "Found uploads in started state, checking for intermediate reports",
-            extra={
-                "upload_ids": [u.id_ for u in started_uploads],
-                "count": len(started_uploads),
-            },
-        )
-
-        # Check which uploads have intermediate reports (confirms they were processed)
-        redis_connection = get_redis_connection()
-        upload_ids_with_reports = set()
-
-        for upload in started_uploads:
-            report_key = intermediate_report_key(upload.id_)
-            if redis_connection.exists(report_key):
-                upload_ids_with_reports.add(upload.id_)
-            else:
-                log.warning(
-                    "Upload in started state but no intermediate report found (may have expired)",
-                    extra={"upload_id": upload.id_},
-                )
-
-        return upload_ids_with_reports
-
     def _reconstruct_processing_results(
         self, db_session, state: ProcessingState, commit: Commit
     ) -> list[ProcessingResult]:
         """Reconstruct processing_results from ProcessingState when finisher is triggered
         outside of a chord (e.g., from orphaned upload recovery).
 
-        This ensures ALL uploads that were marked as processed in Redis are included
+        This ensures all uploads marked as processed in state tracking are included
         in the final merged report, even if they completed via retry/recovery.
-
-        If Redis state has expired (TTL: PROCESSING_STATE_TTL), falls back to database
-        to find uploads in "started" state that have intermediate reports, preventing data loss.
         """
 
-        # Get all upload IDs that are ready to be merged (in "processed" set)
+        # Get all upload IDs that are ready to be merged.
         upload_ids = state.get_uploads_for_merging()
 
         if not upload_ids:
-            log.warning(
-                "No uploads found in Redis processed set, checking database for started uploads",
-                extra={"repoid": commit.repoid, "commitid": commit.commitid},
-            )
-            # Fallback: Redis state expired (TTL: PROCESSING_STATE_TTL), check DB for uploads
-            # in "started" state that might have been processed but never finalized
-            upload_ids = self._find_started_uploads_with_reports(db_session, commit)
-
-            if not upload_ids:
-                log.warning(
-                    "No started uploads with intermediate reports found in database",
-                    extra={"repoid": commit.repoid, "commitid": commit.commitid},
-                )
-                return []
-
-            log.info(
-                "Found started uploads with intermediate reports (Redis state expired)",
-                extra={
-                    "upload_ids": list(upload_ids),
-                    "count": len(upload_ids),
-                },
-            )
+            return []
 
         log.info(
             "Reconstructing processing results from ProcessingState",