fix(dedupe): only count import locks held in the current database

dojo/importers/import_lock.py

@@ -164,6 +164,10 @@ def _import_lock_holders(test_ids=None):
 
     The state and its age are NULL for a backend this database user may not see, such as one of
     another role.
+
+    pg_locks lists the advisory locks of every database on the server, but an advisory lock only
+    conflicts within its own database, so only this database's holders count. Another DefectDojo
+    database on the same server (or another parallel test worker's) has its own test with this id.
     """
     with connection.cursor() as cursor:
         cursor.execute(
@@ -173,6 +177,7 @@ def _import_lock_holders(test_ids=None):
             FROM pg_locks AS locks
             LEFT JOIN pg_stat_activity AS activity ON activity.pid = locks.pid
             WHERE locks.locktype = 'advisory' AND locks.granted AND locks.objsubid = 2
+              AND locks.database = (SELECT oid FROM pg_database WHERE datname = current_database())
               AND locks.classid::bigint = %s AND (%s::bigint[] IS NULL OR locks.objid::bigint = ANY(%s::bigint[]))
             """,
             [TEST_IMPORT_LOCK_CLASS, test_ids, test_ids],

unittests/test_dedupe_delete_commit_race.py

@@ -33,6 +33,7 @@
 from types import SimpleNamespace
 from unittest.mock import patch
 
+import psycopg
 from crum import impersonate
 from django.core.exceptions import ValidationError
 from django.core.files.uploadedfile import SimpleUploadedFile
@@ -206,6 +207,20 @@ def hold():
         holder.join(WAIT_TIMEOUT_SECONDS)
 
 
+@contextmanager
+def _import_lock_held_in_another_database(test_id):
+    """
+    Hold the import lock for this test id in a different database on the same server.
+
+    Advisory locks are scoped to a database, so this is another DefectDojo instance (or another
+    parallel test worker) whose own test happens to have the same id: it locks nothing here.
+    """
+    params = {**connection.get_connection_params(), "dbname": "postgres"}
+    with psycopg.connect(**params, autocommit=True) as other:
+        other.execute("SELECT pg_advisory_lock_shared(%s, %s)", [TEST_IMPORT_LOCK_CLASS, test_id])
+        yield
+
+
 def _import_lock_is_held(test_id):
     """Ask from another connection, as the delete would, whether an import holds this test's lock."""
     def in_a_transaction():
@@ -809,6 +824,24 @@ def test_matched_duplicate_survives_while_its_test_is_being_imported(self):
         persisted = Finding.objects.filter(id=matched.id).first()
         self.assertIsNotNone(persisted, msg="a matched duplicate in a test under import was deleted mid-import")
 
+    def test_import_lock_in_another_database_does_not_hold_back_the_delete(self):
+        """
+        Regression: the selection read every import lock on the server from pg_locks, including those
+        of other databases, so a test was skipped whenever another database's test with the same id
+        was being imported (flaky under --parallel, where each worker has its own database).
+        """
+        original, duplicate = self._create_findings("Dupe delete original", "Dupe delete other database")
+        self._commit_as_duplicate(duplicate, original)
+        self._backdate(finding_ids=[original.id, duplicate.id])
+
+        with _import_lock_held_in_another_database(self.test.id):
+            self._run_dupe_delete()
+
+        self.assertFalse(
+            Finding.objects.filter(id=duplicate.id).exists(),
+            msg="an import lock held in another database kept this database's excess duplicate",
+        )
+
     def test_duplicate_of_a_test_just_reimported_is_deleted_once_the_import_has_ended(self):
         """
         Control, and the reason for the lock: a test reimported every few minutes still has its