Bundle Analysis: dedupe caching processing (#1184)

Author: JerrySentry

tasks/bundle_analysis_processor.py

@@ -8,7 +8,7 @@
 
 from app import celery_app
 from database.enums import ReportType
-from database.models import Commit, Upload
+from database.models import Commit, CommitReport, Upload
 from services.bundle_analysis.report import (
     BundleAnalysisReportService,
     ProcessingResult,
@@ -114,7 +114,42 @@ def process_impl_within_lock(
         if upload_id is not None:
             upload = db_session.query(Upload).filter_by(id_=upload_id).first()
         else:
-            commit_report = report_service.initialize_and_save_report(commit)
+            # This processor task handles caching for reports. When the 'upload' parameter is missing,
+            # it indicates this task was triggered by a non-BA upload.
+            #
+            # To prevent redundant caching of the same parent report:
+            # 1. We first check if a BA report already exists for this commit
+            # 2. We then verify there are uploads associated with it that aren't in an error state
+            #
+            # If both conditions are met, we can exit the task early since the caching was likely
+            # already handled. Otherwise, we need to:
+            # 1. Create a new BA report and upload
+            # 2. Proceed with caching data from the parent report
+            commit_report = (
+                db_session.query(CommitReport)
+                .filter_by(
+                    commit_id=commit.id,
+                    report_type=ReportType.BUNDLE_ANALYSIS.value,
+                )
+                .first()
+            )
+            if commit_report:
+                upload_states = [upload.state for upload in commit_report.uploads]
+                if upload_states and any(
+                    upload_state != "error" for upload_state in upload_states
+                ):
+                    log.info(
+                        "Bundle analysis report already exists for commit, skipping carryforward",
+                        extra=dict(
+                            repoid=commit.repoid,
+                            commit=commit.commitid,
+                        ),
+                    )
+                    return processing_results
+            else:
+                # If the commit report does not exist, we will create a new one
+                commit_report = report_service.initialize_and_save_report(commit)
+
             upload = report_service.create_report_upload({"url": ""}, commit_report)
             carriedforward = True
 
@@ -161,6 +196,17 @@ def process_impl_within_lock(
 
             processing_results.append(result.as_dict())
         except (CeleryError, SoftTimeLimitExceeded, SQLAlchemyError):
+            log.exception(
+                "Unable to process bundle analysis upload",
+                extra=dict(
+                    repoid=repoid,
+                    commit=commitid,
+                    commit_yaml=commit_yaml,
+                    params=params,
+                    upload_id=upload.id_,
+                    parent_task=self.request.parent_id,
+                ),
+            )
             raise
         except Exception:
             log.exception(

tasks/tests/unit/test_bundle_analysis_processor_task.py

@@ -1293,3 +1293,123 @@ def test_bundle_analysis_processor_task_no_upload(
     assert commit.state == "complete"
     assert upload.state == "processed"
     assert upload.upload_type == "carriedforward"
+
+
+@pytest.mark.django_db(databases={"default", "timeseries"})
+def test_bundle_analysis_processor_task_carryforward(
+    mocker,
+    dbsession,
+    mock_storage,
+):
+    storage_path = (
+        "v1/repos/testing/ed1bdd67-8fd2-4cdb-ac9e-39b99e4a3892/bundle_report.sqlite"
+    )
+    mock_storage.write_file(get_bucket_name(), storage_path, "test-content")
+
+    mocker.patch.object(
+        BundleAnalysisProcessorTask,
+        "app",
+        tasks={
+            bundle_analysis_save_measurements_task_name: mocker.MagicMock(),
+        },
+    )
+
+    commit = CommitFactory.create(state="pending")
+    dbsession.add(commit)
+    dbsession.flush()
+
+    commit_report = CommitReport(
+        commit_id=commit.id_, report_type=ReportType.BUNDLE_ANALYSIS.value
+    )
+    dbsession.add(commit_report)
+    dbsession.flush()
+
+    upload = UploadFactory.create(
+        storage_path=storage_path, report=commit_report, state="processed"
+    )
+    dbsession.add(upload)
+    dbsession.flush()
+
+    BundleAnalysisProcessorTask().run_impl(
+        dbsession,
+        {"results": [{"previous": "result"}]},
+        repoid=commit.repoid,
+        commitid=commit.commitid,
+        commit_yaml={},
+        params={
+            "upload_id": None,
+            "commit": commit.commitid,
+        },
+    )
+
+    # A new upload wasn't created because the caching was skipped
+    total_uploads = (
+        dbsession.query(Upload).filter_by(report_id=commit_report.id).count()
+    )
+    assert total_uploads == 1
+
+    # A new report wasn't created either
+    total_ba_reports = (
+        dbsession.query(CommitReport).filter_by(commit_id=commit.id).count()
+    )
+    assert total_ba_reports == 1
+
+
+@pytest.mark.django_db(databases={"default", "timeseries"})
+def test_bundle_analysis_processor_task_carryforward_error(
+    mocker,
+    dbsession,
+    mock_storage,
+):
+    storage_path = (
+        "v1/repos/testing/ed1bdd67-8fd2-4cdb-ac9e-39b99e4a3892/bundle_report.sqlite"
+    )
+    mock_storage.write_file(get_bucket_name(), storage_path, "test-content")
+
+    mocker.patch.object(
+        BundleAnalysisProcessorTask,
+        "app",
+        tasks={
+            bundle_analysis_save_measurements_task_name: mocker.MagicMock(),
+        },
+    )
+
+    commit = CommitFactory.create(state="pending")
+    dbsession.add(commit)
+    dbsession.flush()
+
+    commit_report = CommitReport(
+        commit_id=commit.id_, report_type=ReportType.BUNDLE_ANALYSIS.value
+    )
+    dbsession.add(commit_report)
+    dbsession.flush()
+
+    upload = UploadFactory.create(
+        storage_path=storage_path, report=commit_report, state="error"
+    )
+    dbsession.add(upload)
+    dbsession.flush()
+
+    BundleAnalysisProcessorTask().run_impl(
+        dbsession,
+        {"results": [{"previous": "result"}]},
+        repoid=commit.repoid,
+        commitid=commit.commitid,
+        commit_yaml={},
+        params={
+            "upload_id": None,
+            "commit": commit.commitid,
+        },
+    )
+
+    # A new upload was created because all the previous uploads were in error states
+    total_uploads = (
+        dbsession.query(Upload).filter_by(report_id=commit_report.id).count()
+    )
+    assert total_uploads == 2
+
+    # There should still only be 1 BA report
+    total_ba_reports = (
+        dbsession.query(CommitReport).filter_by(commit_id=commit.id).count()
+    )
+    assert total_ba_reports == 1