Avoid race condition when reporting job statuses after retriggering (#3121)

Avoid race condition when reporting job statuses after retriggering

Retriggering starts a new job run that shares the same status check with the one that is about to be canceled. Cancel request is sent to Copr/Koji/TF, status check is updated by the new job run, but then a build end event or TF webhook arrives and the status check is updated with a wrong status (canceled in this case, but this could happen on e.g. a build failure as well).
Prevent this by checking if there is a newer run of the same job before reporting status in the handlers.

Reviewed-by: gemini-code-assist[bot]
Reviewed-by: Maja Massarini

Author: centosinfra-prod-github-app[bot]

packit_service/models.py

@@ -2374,6 +2374,34 @@ def get_by_id(cls, id_: int) -> Optional["CoprBuildTargetModel"]:
         with sa_session_transaction() as session:
             return session.query(CoprBuildTargetModel).filter_by(id=id_).first()
 
+    @classmethod
+    def has_newer_run(cls, run: "CoprBuildTargetModel") -> bool:
+        """Check if a newer build exists for the same target+identifier
+        on the same project object (e.g. same PR).
+
+        Used to avoid overwriting a newer build's check status when
+        processing results for an older (e.g. canceled) build.
+        """
+        project_event = run.get_project_event_model()
+        if not project_event or not run.submitted_time:
+            return False
+        with sa_session_transaction() as session:
+            return session.query(
+                session.query(CoprBuildTargetModel)
+                .join(CoprBuildGroupModel)
+                .join(PipelineModel)
+                .join(ProjectEventModel)
+                .filter(
+                    ProjectEventModel.type == project_event.type,
+                    ProjectEventModel.event_id == project_event.event_id,
+                    CoprBuildTargetModel.identifier == run.identifier,
+                    CoprBuildTargetModel.target == run.target,
+                    CoprBuildTargetModel.submitted_time > run.submitted_time,
+                    CoprBuildTargetModel.id != run.id,
+                )
+                .exists()
+            ).scalar()
+
     @classmethod
     def get_all(cls) -> Iterable["CoprBuildTargetModel"]:
         with sa_session_transaction() as session:
@@ -3014,6 +3042,33 @@ def get_by_id(cls, id_: int) -> Optional["KojiBuildTargetModel"]:
         with sa_session_transaction() as session:
             return session.query(KojiBuildTargetModel).filter_by(id=id_).first()
 
+    @classmethod
+    def has_newer_run(cls, run: "KojiBuildTargetModel") -> bool:
+        """Check if a newer build exists for the same target
+        on the same project object (e.g. same PR).
+
+        Used to avoid overwriting a newer build's check status when
+        processing results for an older (e.g. canceled) build.
+        """
+        project_event = run.get_project_event_model()
+        if not project_event or not run.submitted_time:
+            return False
+        with sa_session_transaction() as session:
+            return session.query(
+                session.query(KojiBuildTargetModel)
+                .join(KojiBuildGroupModel)
+                .join(PipelineModel)
+                .join(ProjectEventModel)
+                .filter(
+                    ProjectEventModel.type == project_event.type,
+                    ProjectEventModel.event_id == project_event.event_id,
+                    KojiBuildTargetModel.target == run.target,
+                    KojiBuildTargetModel.submitted_time > run.submitted_time,
+                    KojiBuildTargetModel.id != run.id,
+                )
+                .exists()
+            ).scalar()
+
     @classmethod
     def get_all(cls) -> Iterable["KojiBuildTargetModel"]:
         with sa_session_transaction() as session:
@@ -3855,6 +3910,34 @@ def get_by_pipeline_id(cls, pipeline_id: str) -> Optional["TFTTestRunTargetModel
         with sa_session_transaction() as session:
             return session.query(TFTTestRunTargetModel).filter_by(pipeline_id=pipeline_id).first()
 
+    @classmethod
+    def has_newer_run(cls, run: "TFTTestRunTargetModel") -> bool:
+        """Check if a newer test run exists for the same target+identifier
+        on the same project object (e.g. same PR).
+
+        Used to avoid overwriting a newer run's check status when
+        processing results for an older (e.g. canceled) run.
+        """
+        project_event = run.get_project_event_model()
+        if not project_event or not run.submitted_time:
+            return False
+        with sa_session_transaction() as session:
+            return session.query(
+                session.query(TFTTestRunTargetModel)
+                .join(TFTTestRunGroupModel)
+                .join(PipelineModel)
+                .join(ProjectEventModel)
+                .filter(
+                    ProjectEventModel.type == project_event.type,
+                    ProjectEventModel.event_id == project_event.event_id,
+                    TFTTestRunTargetModel.identifier == run.identifier,
+                    TFTTestRunTargetModel.target == run.target,
+                    TFTTestRunTargetModel.submitted_time > run.submitted_time,
+                    TFTTestRunTargetModel.id != run.id,
+                )
+                .exists()
+            ).scalar()
+
     @classmethod
     def get_all_by_status(
         cls,

packit_service/worker/handlers/copr.py

@@ -316,6 +316,7 @@ def _run(self) -> TaskResults:
         if self.build.status in [
             BuildStatus.success,
             BuildStatus.failure,
+            BuildStatus.canceled,
         ]:
             msg = (
                 f"Copr build {self.copr_event.build_id} is already"
@@ -346,18 +347,25 @@ def _run(self) -> TaskResults:
             packit_dashboard_url = get_copr_build_info_url(self.build.id)
             # if SRPM build failed it has been reported already so skip reporting
             if self.build.get_srpm_build().status != BuildStatus.failure:
-                self.copr_build_helper.report_status_to_all_for_chroot(
-                    state=BaseCommitStatus.failure,
-                    description=failed_msg,
-                    url=packit_dashboard_url,
-                    chroot=self.copr_event.chroot,
-                )
-                self.measure_time_after_reporting()
-                self.copr_build_helper.notify_about_failure_if_configured(
-                    packit_dashboard_url=packit_dashboard_url,
-                    external_dashboard_url=self.build.web_url,
-                    logs_url=self.build.build_logs_url,
-                )
+                if CoprBuildTargetModel.has_newer_run(self.build):
+                    logger.info(
+                        f"Skipping status report for Copr build "
+                        f"{self.copr_event.build_id} ({self.build.target}): "
+                        f"a newer build exists for the same target."
+                    )
+                else:
+                    self.copr_build_helper.report_status_to_all_for_chroot(
+                        state=BaseCommitStatus.failure,
+                        description=failed_msg,
+                        url=packit_dashboard_url,
+                        chroot=self.copr_event.chroot,
+                    )
+                    self.measure_time_after_reporting()
+                    self.copr_build_helper.notify_about_failure_if_configured(
+                        packit_dashboard_url=packit_dashboard_url,
+                        external_dashboard_url=self.build.web_url,
+                        logs_url=self.build.build_logs_url,
+                    )
             self.build.set_status(BuildStatus.failure)
             report_long_runtime("Copr build failed end", 120, run_start_time)
             return TaskResults(success=False, details={"msg": failed_msg})

packit_service/worker/handlers/koji.py

@@ -262,26 +262,34 @@ def _run(self) -> TaskResults:
         else:
             self.push_metrics()
             self.build.set_status(new_commit_status.value)
-            self.report(description, new_commit_status, dashboard_url, koji_web_url)
             koji_build_logs = self.koji_task_event.get_koji_build_rpm_tasks_logs_urls(
                 self.service_config.koji_logs_url,
             )
-
             self.build.set_build_logs_urls(koji_build_logs)
             self.build.set_web_url(koji_web_url)
 
+            newer_run_exists = KojiBuildTargetModel.has_newer_run(self.build)
+            if newer_run_exists:
+                logger.info(
+                    f"Skipping status report for Koji build "
+                    f"{self.koji_task_event.task_id} ({self.build.target}): "
+                    f"a newer build exists for the same target."
+                )
+            else:
+                self.report(description, new_commit_status, dashboard_url, koji_web_url)
+
             if self.koji_task_event.state == KojiTaskState.failed:
                 logger.info(
                     f"Failed Koji scratch build (parent taskID = {self.koji_task_event.task_id})"
                 )
                 self.trigger_log_detective_if_configured()
-                # Convert dict of logs URLs to a string representation
-                logs_url_str = ", ".join(koji_build_logs.values()) if koji_build_logs else ""
-                self.notify_about_failure_if_configured(
-                    packit_dashboard_url=dashboard_url,
-                    external_dashboard_url=koji_web_url,
-                    logs_url=logs_url_str,
-                )
+                if not newer_run_exists:
+                    logs_url_str = ", ".join(koji_build_logs.values()) if koji_build_logs else ""
+                    self.notify_about_failure_if_configured(
+                        packit_dashboard_url=dashboard_url,
+                        external_dashboard_url=koji_web_url,
+                        logs_url=logs_url_str,
+                    )
 
         msg = (
             f"Build on {self.build.target} in koji changed state "

packit_service/worker/handlers/testing_farm.py

@@ -712,6 +712,15 @@ def _run(self) -> TaskResults:
             )
             self.pushgateway.test_run_finished_time.observe(test_run_time)
 
+        if TFTTestRunTargetModel.has_newer_run(test_run_model):
+            logger.info(
+                f"Skipping status report for TF run {self.pipeline_id} "
+                f"({test_run_model.target}/{test_run_model.identifier}): "
+                f"a newer run exists for the same target."
+            )
+            test_run_model.set_status(self.result, created=self.created)
+            return TaskResults(success=True, details={})
+
         test_run_model.set_web_url(self.log_url)
         url = get_testing_farm_info_url(test_run_model.id) if test_run_model else None
         self.testing_farm_job_helper.report_status_to_tests_for_test_target(

tests/conftest.py

@@ -239,6 +239,7 @@ def __init__(self, **kwargs):
             },
         ],
         task_accepted_time=datetime.now(),
+        submitted_time=datetime.now(),
         build_start_time=None,
         build_logs_url="https://log-url",
     )
@@ -316,6 +317,7 @@ def koji_build_pr():
         web_url="https://some-url",
         target="some-target",
         status="some-status",
+        submitted_time=datetime.now(),
         runs=runs,
     )
     koji_build_model._srpm_build_for_mocking = srpm_build

tests/integration/test_listen_to_fedmsg.py

@@ -1207,6 +1207,7 @@ def test_copr_build_end_testing_farm_manual_trigger(
         copr_build_pr,
     )
     flexmock(CoprBuildTargetModel).should_receive("get_by_id").and_return(copr_build_pr)
+    flexmock(CoprBuildTargetModel).should_receive("has_newer_run").and_return(False)
     copr_build_pr.should_call("set_status").with_args(build_status).once()
     copr_build_pr.should_receive("set_end_time").once()
     copr_build_pr.should_receive("get_package_name").and_return(None)
@@ -2504,6 +2505,7 @@ def test_koji_build_start(koji_build_scratch_start, pc_koji_build_pr, koji_build
     flexmock(KojiBuildTargetModel).should_receive("get_by_task_id").and_return(
         koji_build_pr,
     )
+    flexmock(KojiBuildTargetModel).should_receive("has_newer_run").and_return(False)
     url = get_koji_build_info_url(1)
     flexmock(requests).should_receive("get").and_return(requests.Response())
     flexmock(requests.Response).should_receive("raise_for_status").and_return(None)
@@ -2566,6 +2568,7 @@ def test_koji_build_end(koji_build_scratch_end, pc_koji_build_pr, koji_build_pr)
     flexmock(KojiBuildTargetModel).should_receive("get_by_task_id").and_return(
         koji_build_pr,
     )
+    flexmock(KojiBuildTargetModel).should_receive("has_newer_run").and_return(False)
     url = get_koji_build_info_url(1)
     flexmock(requests).should_receive("get").and_return(requests.Response())
     flexmock(requests.Response).should_receive("raise_for_status").and_return(None)

tests/unit/test_copr_build.py

@@ -1109,3 +1109,68 @@ def test_get_running_jobs_rebuild_failed_passes_targets(github_pr_event):
         targets={"fedora-rawhide-x86_64", "fedora-40-x86_64"},
     ).and_return(sentinel).once()
     assert list(helper.get_running_jobs()) == sentinel
+
+
+def test_copr_build_end_skips_already_canceled():
+    from packit_service.worker.handlers.copr import CoprBuildEndHandler
+
+    handler = flexmock(
+        build=flexmock(status=BuildStatus.canceled),
+        copr_event=flexmock(
+            build_id="12345",
+            chroot="fedora-rawhide-x86_64",
+            build=flexmock(status=BuildStatus.canceled),
+        ),
+    )
+    result = CoprBuildEndHandler._run(handler)
+    assert result["success"]
+
+
+def test_copr_build_end_skips_reporting_for_superseded_build():
+    from packit_service.worker.handlers.copr import CoprBuildEndHandler
+
+    srpm_build = flexmock(status=BuildStatus.success, url="https://some.url/my.srpm")
+    build = flexmock(
+        id=1,
+        status=BuildStatus.pending,
+        target="fedora-rawhide-x86_64",
+        task_accepted_time=None,
+        web_url="https://copr.url/build/1",
+        build_logs_url="https://copr.url/logs/1",
+        build_id="12345",
+    )
+    build.should_receive("get_srpm_build").and_return(srpm_build)
+    build.should_receive("set_status").with_args(BuildStatus.failure).once()
+
+    copr_event = flexmock(
+        build_id="12345",
+        chroot="fedora-rawhide-x86_64",
+        status=0,
+        timestamp=None,
+    )
+
+    copr_build_helper = flexmock()
+    copr_build_helper.should_receive("report_status_to_all_for_chroot").never()
+    copr_build_helper.should_receive("notify_about_failure_if_configured").never()
+
+    pushgateway = flexmock(
+        copr_builds_finished=flexmock(inc=lambda: None),
+    )
+    pushgateway.should_receive("push").and_return()
+
+    handler = flexmock(
+        build=build,
+        copr_event=copr_event,
+        copr_build_helper=copr_build_helper,
+        pushgateway=pushgateway,
+    )
+    handler.should_receive("set_end_time").once()
+    handler.should_receive("set_srpm_url").once()
+
+    flexmock(CoprBuildTargetModel).should_receive("has_newer_run").with_args(build).and_return(
+        True,
+    ).once()
+
+    result = CoprBuildEndHandler._run(handler)
+    assert not result["success"]
+    assert result["details"]["msg"] == "RPMs failed to be built."