fix: run advisory_lock import/sync project in the right place AAP-47111 (#1333)

Fix a bug introduced in #1296
where we are running under the advisory lock not the real import/sync
task but the proxy that schedules the job for rq and dispatcherd.

This fix run the task under the lock in the right place. Also allows to
merge both test_job_uniqueness and test_project_job_uniqueness tests

Jira: AAP-47111

Signed-off-by: Alex <[email protected]>

Author: Alex-Izquierdo

src/aap_eda/tasks/project.py

@@ -28,27 +28,18 @@
 logger = logging.getLogger(__name__)
 PROJECT_TASKS_QUEUE = "default"
 
-# Wrap the django_rq job decorator so its processing is within our retry
-# code.
-
 
 def import_project(project_id: int) -> tp.Optional[str]:
     """Import project async task.
 
     Proxy for import_project_dispatcherd and import_project_rq.
     """
-    with advisory_lock(f"import_project_{project_id}", wait=False) as acquired:
-        if not acquired:
-            logger.debug(
-                f"Another task already importing project {project_id}, exiting"
-            )
-            return None
-        if features.DISPATCHERD:
-            return import_project_dispatcherd(project_id)
+    if features.DISPATCHERD:
+        return import_project_dispatcherd(project_id)
 
-        # rq
-        job_data = import_project_rq(project_id)
-        return job_data
+    # rq
+    job_data = import_project_rq(project_id)
+    return job_data
 
 
 def import_project_dispatcherd(project_id: int) -> str:
@@ -68,6 +59,17 @@ def import_project_rq(project_id: int) -> str:
 
 
 def _import_project(project_id: int):
+    """Wrap _import_project_no_lock with advisory lock."""
+    with advisory_lock(f"import_project_{project_id}", wait=False) as acquired:
+        if not acquired:
+            logger.debug(
+                f"Another task already importing project {project_id}, exiting"
+            )
+            return
+        _import_project_no_lock(project_id)
+
+
+def _import_project_no_lock(project_id: int):
     """Import project without lock.
 
     This function is intended to be run by the tasking system inside the lock.
@@ -88,17 +90,11 @@ def sync_project(project_id: int) -> tp.Optional[str]:
 
     Proxy for sync_project_dispatcherd and sync_project_rq.
     """
-    with advisory_lock(f"sync_project_{project_id}", wait=False) as acquired:
-        if not acquired:
-            logger.debug(
-                f"Another task already syncing project {project_id}, exiting"
-            )
-            return None
-        if features.DISPATCHERD:
-            return sync_project_dispatcherd(project_id)
-        # rq
-        job_data = sync_project_rq(project_id)
-        return job_data
+    if features.DISPATCHERD:
+        return sync_project_dispatcherd(project_id)
+    # rq
+    job_data = sync_project_rq(project_id)
+    return job_data
 
 
 def sync_project_dispatcherd(project_id: int) -> str:
@@ -118,6 +114,21 @@ def sync_project_rq(project_id: int) -> str:
 
 
 def _sync_project(project_id: int):
+    """Wrap _sync_project_no_lock with advisory lock."""
+    with advisory_lock(f"sync_project_{project_id}", wait=False) as acquired:
+        if not acquired:
+            logger.debug(
+                f"Another task already syncing project {project_id}, exiting"
+            )
+            return
+        _sync_project_no_lock(project_id)
+
+
+def _sync_project_no_lock(project_id: int):
+    """Sync project without lock.
+
+    This function is intended to be run by the tasking system inside the lock.
+    """
     logger.info(f"Task started: Sync project ( {project_id=} )")
 
     project = models.Project.objects.get(pk=project_id)

tests/integration/test_advisory_lock.py

@@ -29,104 +29,43 @@
             "fn_call": "monitor_project_tasks",
             "fn_args": [],
         },
-    ],
-    ids=[
-        "gather_analytics",
-        "monitor_rulebook_processes",
-        "monitor_project_tasks",  # only scheduled in RQ
-    ],
-)
-@pytest.mark.django_db
-def test_job_uniqueness(module_data):
-    call_log = []
-    lock = Lock()
-
-    def _wrapper_call(shared_list):
-        """Patch fn mock in a thread-safe way inside each thread."""
-        import importlib
-
-        module = importlib.import_module(module_data["module_path"])
-
-        importlib.reload(module)
-
-        with mock.patch(
-            f"{module_data['module_path']}.{module_data['fn_mock']}",
-        ) as fn_mock:
-
-            def record_call(void=None):
-                time.sleep(1)
-                shared_list.append("called")
-
-            fn_mock.side_effect = record_call
-            try:
-                getattr(module, module_data["fn_call"])(
-                    *module_data["fn_args"],
-                )
-            except Exception as e:
-                pytest.fail(
-                    f"Exception raised in {module_data['fn_call']}: {e}"
-                )
-
-    def thread_safe_call():
-        _wrapper_call(ThreadSafeList(call_log, lock))
-
-    with ThreadPoolExecutor(max_workers=2) as executor:
-        futures = [
-            executor.submit(thread_safe_call),
-            executor.submit(thread_safe_call),
-        ]
-        wait(futures, timeout=3)
-
-    assert len(call_log) == 1
-
-
-@pytest.mark.parametrize(
-    "module_data",
-    [
         {
             "module_path": "aap_eda.tasks.project",
-            "fn_mock_dispatcherd": "import_project_dispatcherd",
-            "fn_mock_rq": "import_project_rq",
-            "fn_call": "import_project",
+            "fn_mock": "_import_project_no_lock",
+            "fn_call": "_import_project",
             "fn_args": [1],
         },
         {
             "module_path": "aap_eda.tasks.project",
-            "fn_mock_dispatcherd": "sync_project_dispatcherd",
-            "fn_mock_rq": "sync_project_rq",
-            "fn_call": "sync_project",
+            "fn_mock": "_sync_project_no_lock",
+            "fn_call": "_sync_project",
             "fn_args": [1],
         },
     ],
     ids=[
-        "import_project",
-        "sync_project",
+        "gather_analytics",
+        "monitor_rulebook_processes",
+        "monitor_project_tasks",  # only scheduled in RQ
+        "_import_project",
+        "_sync_project",
     ],
 )
 @pytest.mark.django_db
-def test_project_job_uniqueness(module_data):
+def test_job_uniqueness(module_data):
     call_log = []
     lock = Lock()
 
     def _wrapper_call(shared_list):
-        """Patch fn_mock in a thread-safe way inside each thread."""
+        """Patch fn mock in a thread-safe way inside each thread."""
         import importlib
 
-        from aap_eda.settings import features
-
         module = importlib.import_module(module_data["module_path"])
 
         importlib.reload(module)
 
-        mock_fn_target = (
-            f"{module_data['fn_mock_dispatcherd']}"
-            if features.DISPATCHERD
-            else f"{module_data['fn_mock_rq']}"
-        )
-
-        mock_path = f"{module_data['module_path']}.{mock_fn_target}"
-
-        with mock.patch(mock_path) as fn_mock:
+        with mock.patch(
+            f"{module_data['module_path']}.{module_data['fn_mock']}",
+        ) as fn_mock:
 
             def record_call(void=None):
                 time.sleep(1)