Try to minimize concurrency problems (#2617)

Try to minimize concurrency problems

Related to #2604
We have concurrency issues.
This change does not solve them but could minimize them.
Here an example log of the concurrency issue before my change:
worker1
11/5/24 11:47:31.688 AM	2024-11-05T11:47:31.688406049+00:00 stderr F [2024-11-05 11:47:31,688: DEBUG/MainProcess] task.run_copr_build_end_handler[8d4413e0-ba73-4f78-a14f-487b824a1b0f] Searching for base build for 7c08e96170470eb6e3cb99fbf8b705862a1e0a58 commit in @teemtee/tmt Copr project in our DB.
[...]
11/5/24 11:47:32.592 AM	2024-11-05T11:47:32.592982751+00:00 stderr F [2024-11-05 11:47:32,592: DEBUG/MainProcess] task.run_copr_build_end_handler[8d4413e0-ba73-4f78-a14f-487b824a1b0f] Command: osh-cli version-diff-build --srpm=/tmp/tmp27a0u3z_/tmt-1.39.dev888-1.20241105114529529022.pr3342.18.g3fc4607c.src.rpm --base-srpm=/tmp/tmp27a0u3z_/tmt-1.39.dev17+g7c08e961-main.src.rpm --config=fedora-rawhide-x86_64 --nowait --json --comment=Submitted via Packit Service for https://dashboard.packit.dev/results/copr-builds/1994869
11/5/24 11:47:39.070 AM	2024-11-05T11:47:39.070934307+00:00 stderr F [2024-11-05 11:47:39,070: INFO/MainProcess] task.run_copr_build_end_handler[8d4413e0-ba73-4f78-a14f-487b824a1b0f] Scan submitted successfully.

worker2
11/5/24 11:47:30.159 AM	2024-11-05T11:47:30.159087194+00:00 stderr F [2024-11-05 11:47:30,158: DEBUG/ForkPoolWorker-1] task.babysit_copr_build[fd022210-82ed-43ff-9494-18705b393847] Searching for base build for 7c08e96170470eb6e3cb99fbf8b705862a1e0a58 commit in @teemtee/tmt Copr project in our DB.
[...]
11/5/24 11:47:30.795 AM	2024-11-05T11:47:30.795838802+00:00 stderr F [2024-11-05 11:47:30,795: DEBUG/ForkPoolWorker-1] task.babysit_copr_build[fd022210-82ed-43ff-9494-18705b393847] Command: osh-cli version-diff-build --srpm=/tmp/tmpg70py27t/tmt-1.39.dev888-1.20241105114529529022.pr3342.18.g3fc4607c.src.rpm --base-srpm=/tmp/tmpg70py27t/tmt-1.39.dev17+g7c08e961-main.src.rpm --config=fedora-rawhide-x86_64 --nowait --json --comment=Submitted via Packit Service for https://dashboard.packit.dev/results/copr-builds/1994869
11/5/24 11:47:32.774 AM	2024-11-05T11:47:32.774600167+00:00 stderr F [2024-11-05 11:47:32,774: INFO/ForkPoolWorker-1] task.babysit_copr_build[fd022210-82ed-43ff-9494-18705b393847] Scan submitted successfully.

Reviewed-by: Matej Focko
Reviewed-by: Nikola Forró
Reviewed-by: Maja Massarini

packit_service/models.py

@@ -2329,6 +2329,46 @@ def add_scan(self, task_id: int) -> "OSHScanModel":
             session.add(scan)
             return scan
 
+    @contextmanager
+    def add_scan_transaction(self) -> Generator["OSHScanModel"]:
+        """
+        Context manager that creates a ScanModel upon entering the context,
+        provides a corresponding instance of `ScanModel` to be updated within the context
+        and commits the changes upon exiting the context, all within a single transaction.
+
+        This locking mechanism is working on the assumption that just a single scan model
+        for build can exist.
+
+        raise: IntegrityError if the scan model already exists
+        """
+        session = singleton_session or Session()
+        session.begin()
+
+        try:
+            scan = OSHScanModel()
+            scan.copr_build_target = self
+            session.add(scan)
+            session.commit()
+        except Exception as ex:
+            logger.warning(f"Exception while working with database: {ex!r}")
+            session.rollback()
+            raise
+
+        try:
+            yield scan
+        except Exception as ex:
+            logger.warning(f"{ex!r}")
+            session.rollback()
+            raise
+
+        try:
+            session.add(scan)
+            session.commit()
+        except Exception as ex:
+            logger.warning(f"Exception while working with database: {ex!r}")
+            session.rollback()
+            raise
+
 
 class KojiBuildGroupModel(ProjectAndEventsConnector, GroupModel, Base):
     __tablename__ = "koji_build_groups"

packit_service/worker/events/open_scan_hub.py

@@ -38,13 +38,20 @@ def __init__(
                 " It should have been created when receiving the CoprBuildEndEvent"
                 " and should have been associated with the copr build.",
             )
-        else:
-            self.build = self.scan.copr_build_target
-            project_event = self.build.get_project_event_model()
-            # commit_sha is needed by the StatusReporter
-            # and have to be serialized to be later found in the
-            # event metadata
-            self.commit_sha = project_event.commit_sha if not self.commit_sha else self.commit_sha
+            return
+        self.build = self.scan.copr_build_target
+        if not self.build:
+            logger.warning(
+                f"Scan with id {task_id} not associated with a build."
+                " It should have been associated when receiving the CoprBuildEndEvent."
+            )
+            return
+
+        project_event = self.build.get_project_event_model()
+        # commit_sha is needed by the StatusReporter
+        # and have to be serialized to be later found in the
+        # event metadata
+        self.commit_sha = project_event.commit_sha if not self.commit_sha else self.commit_sha
 
     def get_db_project_object(self) -> Optional[AbstractProjectObjectDbType]:
         return self.build.get_project_event_object()

packit_service/worker/helpers/open_scan_hub.py

@@ -15,13 +15,16 @@
     JobConfigTriggerType,
     JobType,
 )
+from packit.exceptions import PackitException
+from sqlalchemy.exc import IntegrityError
 
 from packit_service.constants import (
     OPEN_SCAN_HUB_FEATURE_DESCRIPTION,
 )
 from packit_service.models import (
     BuildStatus,
     CoprBuildTargetModel,
+    OSHScanStatus,
     SRPMBuildModel,
 )
 from packit_service.service.urls import get_copr_build_info_url
@@ -34,6 +37,10 @@
 logger = logging.getLogger(__name__)
 
 
+class OSHNoFeedback(PackitException):
+    pass
+
+
 class OpenScanHubHelper:
     def __init__(
         self,
@@ -65,14 +72,6 @@ def handle_scan(self):
             logger.debug("No base build job needed for diff scan found in the config.")
             return
 
-        if self.build.scan:
-            # see comment https://github.com/packit/packit-service/issues/2604#issuecomment-2444321483
-            logger.debug(
-                f"Scan for build {self.build.id} already submitted, "
-                "scan task_id {self.build.scan.task_id}."
-            )
-            return
-
         if not (base_srpm_model := self.get_base_srpm_model(base_build_job)):
             logger.debug("Successful base SRPM build has not been found.")
             return
@@ -94,53 +93,55 @@ def handle_scan(self):
 
             build_dashboard_url = get_copr_build_info_url(self.build.id)
 
-            output = self.copr_build_helper.api.run_osh_build(
-                srpm_path=paths[1],
-                base_srpm=paths[0],
-                comment=f"Submitted via Packit Service for {build_dashboard_url}",
-            )
+            try:
+                err_msg = "Scan in OpenScanHub was not submitted successfully."
+                with self.build.add_scan_transaction() as scan:
+                    output = self.copr_build_helper.api.run_osh_build(
+                        srpm_path=paths[1],
+                        base_srpm=paths[0],
+                        comment=f"Submitted via Packit Service for {build_dashboard_url}",
+                    )
 
-            if not output:
-                self.report(
-                    state=BaseCommitStatus.neutral,
-                    description="Scan in OpenScanHub was not submitted successfully.",
-                    url=None,
-                )
-                return
+                    if not output:
+                        raise OSHNoFeedback("Something went wrong, skipping the reporting.")
 
-            logger.info("Scan submitted successfully.")
+                    logger.info("Scan submitted successfully.")
 
-            response_dict = self.parse_dict_from_output(output)
+                    response_dict = self.parse_dict_from_output(output)
 
-            logger.debug(f"Parsed dict from output: {response_dict} ")
+                    logger.debug(f"Parsed dict from output: {response_dict} ")
 
-            scan = None
-            if id := response_dict.get("id"):
-                scan = self.build.add_scan(task_id=id)
-            else:
-                logger.debug(
-                    "It was not possible to get the Open Scan Hub task_id from the response.",
-                )
+                    if id := response_dict.get("id"):
+                        scan.task_id = id
+                        scan.status = OSHScanStatus.pending
+                    else:
+                        raise OSHNoFeedback(
+                            "It was not possible to get the Open Scan Hub task_id "
+                            "from the response.",
+                        )
 
-            if not (url := response_dict.get("url")):
-                msg = "It was not possible to get the task URL from the OSH response."
-                logger.debug(msg)
+                    if not (url := response_dict.get("url")):
+                        err_msg = "It was not possible to get the task URL from the OSH response."
+                        raise OSHNoFeedback(err_msg)
+                    scan.url = url
+
+                    self.report(
+                        state=BaseCommitStatus.running,
+                        description=(
+                            "Scan in OpenScanHub submitted successfully. "
+                            "Check the URL for more details."
+                        ),
+                        url=url,
+                    )
+            except IntegrityError as ex:
+                logger.info(f"OpenScanHub already submitted: {ex}")
+            except OSHNoFeedback as ex:
+                logger.info(f"OpenScanHub feedback missing: {ex}")
                 self.report(
                     state=BaseCommitStatus.neutral,
-                    description=msg,
-                    url=None,
+                    description=err_msg,
+                    url=build_dashboard_url,
                 )
-                return
-            if url and scan:
-                scan.set_url(url)
-
-            self.report(
-                state=BaseCommitStatus.running,
-                description=(
-                    "Scan in OpenScanHub submitted successfully. Check the URL for more details."
-                ),
-                url=url,
-            )
 
     def report(
         self,

tests/unit/test_open_scan_hub.py

@@ -158,7 +158,11 @@ def test_handle_scan(build_models):
                 type=ProjectEventModelType.pull_request,
                 get_project_event_object=lambda: flexmock(),
             ),
-        ),
+        )
+        .should_receive("add_scan_transaction")
+        .once()
+        .and_return(flexmock())
+        .mock(),
         copr_build_helper=CoprBuildJobHelper(
             service_config=flexmock(),
             package_config=package_config,