Use atomic update when adding a testing farm test (#2952)

Use atomic update when adding a testing farm test

To prevent race conditions.
Both threads see that the pipeline model has not yet been linked to the test. So both link it to the test. The latest one overwrites the previous value.
The with_for_update clause isn't enough.

SELECT ... FOR UPDATE + Python check: Two separate operations (read, then check in Python, then write) → race condition possible
Atomic UPDATE with WHERE clause: One operation that checks and updates atomically → no race condition

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

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

packit_service/models.py

@@ -3447,32 +3447,40 @@ def create(cls, run_model: "PipelineModel", ranch: str) -> "TFTTestRunGroupModel
             # Flush to get test_run_group.id before using it
             session.flush()
 
-            # lock the run_model in the DB by using SELECT ... FOR UPDATE
-            # all other workers accessing this run_model will block here
-            # until the context is exited
-            locked_run_model = (
-                session.query(PipelineModel).filter_by(id=run_model.id).with_for_update().first()
+            # Attempt to update the existing run_model if test_run_group_id is NULL
+            # This is an atomic operation at the DB level, preventing race conditions
+            # Only one thread will successfully update (updated_count=1)
+            # The other thread will get updated_count=0 and proceed to clone
+            updated_count = (
+                session.query(PipelineModel)
+                .filter_by(id=run_model.id, test_run_group_id=None)
+                .update({"test_run_group_id": test_run_group.id})
             )
 
-            if not locked_run_model:
-                logger.warning(f"PipelineModel with id={run_model.id} not found in DB.")
-                return test_run_group
+            if updated_count == 0:
+                # If 0 rows were updated, it means another thread already set test_run_group_id
+                # So, clone the run model
+                locked_run_model = (
+                    session.query(PipelineModel)
+                    .filter_by(id=run_model.id)
+                    .with_for_update()
+                    .first()
+                )
+                if not locked_run_model:
+                    logger.warning(f"PipelineModel with id={run_model.id} not found in DB.")
+                    return test_run_group
 
-            # Check test_run_group_id directly to avoid stale relationship cache.
-            # The relationship might not be loaded or might be stale from a different session.
-            if locked_run_model.test_run_group_id:
-                # Clone run model
                 new_run_model = PipelineModel()
+                # Add to session first so relationships can be set properly
+                session.add(new_run_model)
                 new_run_model.project_event = locked_run_model.project_event
                 new_run_model.package_name = locked_run_model.package_name
                 new_run_model.srpm_build = locked_run_model.srpm_build
                 new_run_model.copr_build_group = locked_run_model.copr_build_group
                 new_run_model.koji_build_group = locked_run_model.koji_build_group
                 new_run_model.test_run_group_id = test_run_group.id
-                session.add(new_run_model)
-            else:
-                locked_run_model.test_run_group_id = test_run_group.id
-                session.add(locked_run_model)
+            # else: (updated_count == 1) - the current thread successfully set test_run_group_id
+            # No need to do anything further for the existing run_model as it's already updated.
 
             return test_run_group
 

tests_openshift/conftest.py

@@ -58,6 +58,7 @@
     TFTTestRunTargetModel,
     sa_session_transaction,
     sync_release_pr_association_table,
+    tf_copr_association_table,
 )
 
 
@@ -210,6 +211,7 @@ def clean_db():
         session.query(PipelineModel).delete()
         session.query(ProjectEventModel).delete()
 
+        session.query(tf_copr_association_table).delete()
         session.query(TFTTestRunTargetModel).delete()
         session.query(TFTTestRunGroupModel).delete()
         session.query(CoprBuildTargetModel).delete()

tests_openshift/database/test_tasks.py

@@ -1,6 +1,8 @@
 # Copyright Contributors to the Packit project.
 # SPDX-License-Identifier: MIT
 
+import threading
+
 import pytest
 from celery.canvas import Signature
 from copr.v3 import BuildChrootProxy, BuildProxy, Client
@@ -21,9 +23,16 @@
     BuildStatus,
     CoprBuildGroupModel,
     CoprBuildTargetModel,
+    PipelineModel,
     ProjectEventModel,
+    ProjectEventModelType,
+    PullRequestModel,
     SRPMBuildModel,
+    TFTTestRunGroupModel,
+    TFTTestRunTargetModel,
+    sa_session_transaction,
 )
+from packit_service.worker.handlers import TestingFarmHandler
 from packit_service.worker.helpers.build import babysit
 from packit_service.worker.helpers.build.babysit import check_copr_build
 from packit_service.worker.monitoring import Pushgateway
@@ -260,3 +269,213 @@ def celery_run_async_stub(signatures, handlers) -> None:
     build = CoprBuildTargetModel.get_by_id(packit_build_752.id)
     assert build is not None
     assert build.status == BuildStatus.success
+
+
+def test_testing_farm_race_condition_concurrent_identifiers(
+    clean_before_and_after,
+):
+    """
+    Test that two concurrent testing_farm tasks with different identifiers
+    (test_c and test_d) for the same copr build create separate pipelines
+    without race conditions.
+
+    This test simulates the race condition where two handlers both try to
+    create a test_run_group for the same pipeline at the same time.
+    """
+    # Setup: Create a PR and project event
+    pr = PullRequestModel.get_or_create(
+        pr_id=1,
+        namespace="test-namespace",
+        repo_name="test-repo",
+        project_url="https://github.com/test-namespace/test-repo",
+    )
+    project_event = ProjectEventModel.get_or_create(
+        type=ProjectEventModelType.pull_request,
+        event_id=pr.id,
+        commit_sha="4f4403b44107aae0b820f2a940623d3fa54dfcb6",
+    )
+    # Store the ID to avoid thread-safety issues with SQLAlchemy sessions
+    project_event_id = project_event.id
+
+    # Create SRPM build and pipeline
+    _, run_model = SRPMBuildModel.create_with_new_run(
+        project_event_model=project_event,
+    )
+    copr_build_group = CoprBuildGroupModel.create(run_model)
+
+    # Create copr build with specific build_id
+    # The commit_sha is already set in the project_event we created above
+    _ = CoprBuildTargetModel.create(
+        build_id="10034382",
+        project_name="test-project",
+        owner="test-owner",
+        web_url="https://copr.fedorainfracloud.org/coprs/build/10034382/",
+        target="fedora-rawhide-x86_64",
+        status=BuildStatus.success,
+        copr_build_group=copr_build_group,
+    )
+
+    # Results storage for threads
+    results = {}
+    errors = {}
+
+    # Use a barrier to synchronize both threads to start at exactly the same time
+    # This ensures they execute concurrently and increases the chance of a race condition
+    barrier = threading.Barrier(2, timeout=30)
+
+    def create_test_run_group(identifier: str):
+        """Helper function to create test run group in a thread"""
+        try:
+            # Query the copr build fresh in this thread
+            fresh_copr_build = CoprBuildTargetModel.get_by_build_id(
+                build_id="10034382",
+                target="fedora-rawhide-x86_64",
+            )
+            assert fresh_copr_build is not None, "Copr build should exist"
+
+            handler_instance = TestingFarmHandler(
+                package_config=flexmock(),
+                job_config=JobConfig(
+                    type=JobType.tests,
+                    trigger=JobConfigTriggerType.pull_request,
+                    packages={
+                        "package": CommonPackageConfig(identifier=identifier),
+                    },
+                ),
+                event={},
+                celery_task=flexmock(request=flexmock(retries=0)),
+            )
+            # Query project_event fresh in this thread to avoid thread-safety issues
+            with sa_session_transaction() as session:
+                fresh_project_event = (
+                    session.query(ProjectEventModel).filter_by(id=project_event_id).first()
+                )
+                assert fresh_project_event is not None, "Project event should exist"
+
+            # Set up the handler's required attributes
+            handler_instance._db_project_event = fresh_project_event
+            handler_instance.data = flexmock(
+                db_project_event=fresh_project_event,
+                commit_sha="4f4403b44107aae0b820f2a940623d3fa54dfcb6",
+            )
+            handler_instance._project = flexmock(
+                get_web_url=lambda: "https://github.com/test-namespace/test-repo"
+            )
+
+            # Mock the testing_farm_job_helper
+            handler_instance._testing_farm_job_helper = flexmock(
+                skip_build=False,
+                tft_client=flexmock(default_ranch="public"),
+                tests_targets={"fedora-rawhide-x86_64"},
+                test_target2build_target=lambda target: target,
+                get_latest_copr_build=lambda target, commit_sha: fresh_copr_build,
+                run_testing_farm=lambda test_run, build: {"success": True, "details": {}},
+                build_required=lambda: False,  # Build not required, proceed with tests
+                job_owner="test-owner",
+                job_project="test-project",
+            )
+
+            # Wait for both threads to be ready before calling run()
+            barrier.wait()
+
+            # Call run() which will exercise the full logic including _get_or_create_group
+            result = handler_instance.run()
+            assert result["success"], (
+                f"Handler run() failed for {identifier}: {result.get('details', {})}"
+            )
+
+            # Query the database to get the created group
+            with sa_session_transaction() as session:
+                test_run = (
+                    session.query(TFTTestRunTargetModel)
+                    .join(TFTTestRunGroupModel)
+                    .join(PipelineModel)
+                    .filter(PipelineModel.project_event_id == project_event_id)
+                    .filter(TFTTestRunTargetModel.identifier == identifier)
+                    .order_by(TFTTestRunTargetModel.id.desc())
+                    .first()
+                )
+                assert test_run is not None, f"Test run should have been created for {identifier}"
+                group_id = test_run.group_of_targets.id
+                results[identifier] = (group_id, [test_run])
+        except Exception as e:
+            errors[identifier] = e
+            raise
+
+    # Create two threads that will execute concurrently
+    thread_c = threading.Thread(target=create_test_run_group, args=("test_c",))
+    thread_d = threading.Thread(target=create_test_run_group, args=("test_d",))
+
+    # Start both threads at the same time
+    thread_c.start()
+    thread_d.start()
+
+    # Wait for both threads to complete
+    thread_c.join()
+    thread_d.join()
+
+    # Verify no errors occurred
+    assert not errors, f"Errors occurred: {errors}"
+
+    # Verify both test_run_groups were created
+    assert "test_c" in results
+    assert "test_d" in results
+
+    group_c_id, _ = results["test_c"]
+    group_d_id, _ = results["test_d"]
+
+    # Verify both groups have different IDs
+    assert group_c_id != group_d_id, "Each identifier should have its own group"
+
+    # Verify both groups and test run targets exist
+    with sa_session_transaction() as session:
+        # Verify that both test_run_groups were created in the same second
+        # This ensures the race condition test is actually testing concurrent creation
+        group_c = session.query(TFTTestRunGroupModel).filter_by(id=group_c_id).first()
+        group_d = session.query(TFTTestRunGroupModel).filter_by(id=group_d_id).first()
+
+        assert group_c is not None, "test_c group should exist"
+        assert group_d is not None, "test_d group should exist"
+        assert group_c.submitted_time is not None, "group_c should have a submitted_time"
+        assert group_d.submitted_time is not None, "group_d should have a submitted_time"
+
+        # Verify that there exists a tft_test_run_targets row for every identifier
+        test_run_target_c = (
+            session.query(TFTTestRunTargetModel)
+            .filter_by(identifier="test_c")
+            .filter_by(tft_test_run_group_id=group_c_id)
+            .first()
+        )
+        test_run_target_d = (
+            session.query(TFTTestRunTargetModel)
+            .filter_by(identifier="test_d")
+            .filter_by(tft_test_run_group_id=group_d_id)
+            .first()
+        )
+
+        assert test_run_target_c is not None, "test_c should have a test_run_target row"
+        assert test_run_target_d is not None, "test_d should have a test_run_target row"
+
+        # Verify that both test_run_targets are linked to their respective groups
+        assert test_run_target_c.tft_test_run_group_id == group_c_id
+        assert test_run_target_d.tft_test_run_group_id == group_d_id
+
+        # Verify both groups have pipelines linked to them (check last)
+        pipeline_c = session.query(PipelineModel).filter_by(test_run_group_id=group_c_id).first()
+        pipeline_d = session.query(PipelineModel).filter_by(test_run_group_id=group_d_id).first()
+
+        assert pipeline_c is not None, "test_c group should have a pipeline"
+        assert pipeline_d is not None, "test_d group should have a pipeline"
+        assert pipeline_c.id != pipeline_d.id, "Each group should have its own pipeline"
+
+        # Verify the pipelines are linked to the same project_event
+        assert pipeline_c.project_event_id == project_event_id
+        assert pipeline_d.project_event_id == project_event_id
+
+        # Truncate to seconds and compare
+        group_c_second = group_c.submitted_time.replace(microsecond=0)
+        group_d_second = group_d.submitted_time.replace(microsecond=0)
+        assert group_c_second == group_d_second, (
+            f"Test run groups should be created in the same second. "
+            f"group_c: {group_c.submitted_time}, group_d: {group_d.submitted_time}"
+        )