fix(dedupe): stop imports failing when findings are deleted under them

docs/content/triage_findings/finding_deduplication/about_deduplication.md

@@ -170,6 +170,21 @@ For example, let’s say that you had your Maximum Duplicates field set to ‘1
 2. **Later, you import Test 2 contains the same vulnerability. This will be recorded as Finding B, and Finding B will be marked as a duplicate of Finding A.**
 3. Later still, you import **Test 3** which also contains that vulnerability. This will be recorded as Finding C, which will be marked as a duplicate of Finding A. At this time, Finding B will be deleted from DefectDojo as the threshold for maximum duplicates has been crossed.
 
+### When duplicates are deleted
+
+Excess duplicates are deleted by a background task that runs every minute. Two kinds of duplicate wait for a later run:
+
+- Duplicates in a Test that an import or reimport is working on right now. The import holds a database lock on its Test until it finishes, and the task skips that Test's duplicates while the lock is held. The next run after the import ends deletes them, so a Test that is reimported every few minutes is still cleaned up between reimports.
+- A Finding saved in the last minute, so a request that is still writing it (for example a Finding created through the API) can finish first.
+
+Every other excess duplicate is deleted on the next run, as before.
+
+Self-hosted instances can tune this with three environment variables:
+
+- `DD_DUPE_DELETE_GRACE_MINUTES` (default `1`) sets how long after its last save a duplicate waits. `0` turns the wait off.
+- `DD_DUPE_DELETE_SKIP_TESTS_UNDER_IMPORT` (default `True`) turns the import lock on. Set it to `False` only if DefectDojo reaches PostgreSQL through a connection pooler in transaction mode (such as PgBouncer with `pool_mode = transaction`), where the lock cannot be held across an import. Without the lock, an import can fail if the task deletes Findings it is still writing.
+- `DD_DUPE_DELETE_ABANDONED_IMPORT_LOCK_MINUTES` (default `30`) is a safety valve. If every connection holding a Test's import lock has been idle for this long, no import is running, so the task deletes that Test's duplicates anyway and logs a warning naming the database connections involved. `0` turns the valve off.
+
 ### Applying this setting
 
 Applying **Delete Deduplicate Findings** will begin a deletion process immediately. This setting can be applied on the **System Settings** page. See Enabling Deduplication for more information.

dojo/finding/helper.py

@@ -30,6 +30,7 @@
     do_false_positive_history_batch,
     get_finding_models_for_deduplication,
 )
+from dojo.importers.import_lock import lock_tests_of_findings_under_import, unlock_tests_under_import
 from dojo.jira import services as jira_services
 from dojo.location.feature import locations_enabled
 from dojo.location.models import Location
@@ -549,6 +550,33 @@ def post_process_findings_batch(
     if not finding_ids:
         return
 
+    # Deduplication may mark these findings duplicate, and everything below goes on writing them,
+    # so hold the import lock on their tests until it is done (see dojo.importers.import_lock).
+    locked_tests = lock_tests_of_findings_under_import(finding_ids)
+    try:
+        _post_process_findings_batch(
+            finding_ids,
+            dedupe_option=dedupe_option,
+            product_grading_option=product_grading_option,
+            issue_updater_option=issue_updater_option,
+            push_to_jira=push_to_jira,
+            jira_instance_id=jira_instance_id,
+            force_sync=force_sync,
+        )
+    finally:
+        unlock_tests_under_import(locked_tests)
+
+
+def _post_process_findings_batch(
+    finding_ids,
+    *,
+    dedupe_option,
+    product_grading_option,
+    issue_updater_option,
+    push_to_jira,
+    jira_instance_id,
+    force_sync,
+):
     system_settings = System_Settings.objects.get()
 
     # The status-changing dedup / false-positive-history writes below update dojo_finding

dojo/importers/base_importer.py

@@ -1,7 +1,10 @@
 import base64
 import binascii
 import logging
+import re
 import time
+from contextlib import contextmanager
+from functools import wraps
 from itertools import batched
 from pathlib import Path
 
@@ -27,8 +30,9 @@
 import dojo.finding.helper as finding_helper
 import dojo.risk_acceptance.helper as ra_helper
 from dojo.celery_dispatch import dojo_dispatch_task
-from dojo.db_utils import is_transient_db_conflict
+from dojo.db_utils import is_foreign_key_conflict, is_transient_db_conflict
 from dojo.finding.cwe import finding_cwe_labels
+from dojo.importers.import_lock import lock_tests_under_import, unlock_tests_under_import
 from dojo.importers.options import ImporterOptions
 from dojo.jira import services as jira_services
 from dojo.jira.services import is_keep_in_sync
@@ -68,6 +72,79 @@
 
 logger = logging.getLogger(__name__)
 
+
+def deleted_target_message(instance) -> str:
+    return (
+        f"The {instance._meta.verbose_name} this scan was being imported into "
+        f"(id {instance.pk}) was deleted while the scan was being processed, so "
+        f"the import could not be completed. Nothing was imported. Re-run the "
+        f"import against a {instance._meta.verbose_name} that still exists."
+    )
+
+
+# PostgreSQL's DETAIL for a foreign-key violation on insert/update names the missing row.
+_MISSING_REFERENCED_ROW = re.compile(
+    r'Key \((?P<column>[^)]+)\)=\((?P<value>[^)]*)\) is not present in table "(?P<table>[^"]+)"',
+)
+
+
+def _missing_referenced_row(exception):
+    """Return (table, key) of the row a foreign-key violation says is missing, or None."""
+    detail = getattr(getattr(exception.__cause__, "diag", None), "message_detail", None) or str(exception)
+    match = _MISSING_REFERENCED_ROW.search(detail)
+    return (match["table"], match["value"]) if match else None
+
+
+def findings_deleted_message(test) -> str:
+    return (
+        f"Findings this scan was writing were deleted while the scan was being processed (for "
+        f"example because test {test.pk}, or its engagement or product, is being deleted), so the "
+        f"import could not be completed. Re-run the import once the delete has finished."
+    )
+
+
+def reports_a_deleted_target(process_scan):
+    """
+    Turn the foreign-key failure an import hits when its test is deleted mid-run into a clear error.
+
+    An import writes a batch's child rows (endpoint or location statuses, tags, vulnerability ids,
+    finding groups) well after it created or matched the findings they belong to. A delete in
+    between -- of the test by a user, cascading from its engagement or product, or the API's
+    asynchronous delete, which removes a test's findings in chunks and the test row last -- removes
+    those findings, and the next of those writes fails on its foreign key with a PostgreSQL
+    constraint name that says nothing about why. The import cannot complete: report that, the way
+    save_without_resurrecting does for the test itself.
+
+    Only a violation naming a finding or test that no longer exists is translated; one naming a row
+    that is still there is a genuine bug and is re-raised unchanged. Only the failure path pays for
+    the existence check.
+    """
+
+    @wraps(process_scan)
+    def wrapper(self, *args, **kwargs):
+        try:
+            with self.import_lock_scope():
+                return process_scan(self, *args, **kwargs)
+        except IntegrityError as exception:
+            test = getattr(self, "test", None)
+            missing = _missing_referenced_row(exception) if is_foreign_key_conflict(exception) else None
+            models_by_table = {Finding._meta.db_table: Finding, Test._meta.db_table: Test}
+            if test is None or test.pk is None or missing is None or missing[0] not in models_by_table:
+                raise
+            try:
+                referenced_row_exists = models_by_table[missing[0]].objects.filter(pk=missing[1]).exists()
+                test_exists = Test.objects.filter(pk=test.pk).exists()
+            except DatabaseError:
+                # An aborted enclosing transaction cannot answer; the original error is the report.
+                raise exception from None
+            if referenced_row_exists:
+                raise
+            message = findings_deleted_message(test) if test_exists else deleted_target_message(test)
+            raise ValidationError(message) from exception
+
+    return wrapper
+
+
 # Number of times update_test_tags() re-runs Test.tags.set() when it loses a race against
 # tagulous' cleanup of unused tag rows (see set_test_tags_safe for the full explanation).
 TEST_TAG_SET_MAX_ATTEMPTS = 3
@@ -811,6 +888,72 @@ def _is_vanished_row_error(exception: DatabaseError) -> bool:
         """
         return type(exception) is DatabaseError and "did not affect any rows" in str(exception)
 
+    @contextmanager
+    def import_lock_scope(self):
+        """
+        Hold the import lock taken by lock_test_under_import until the outermost scope exits.
+
+        process_scan opens one; a caller that goes on writing this import's findings after
+        process_scan returns opens its own around the call, and the lock is held until that one
+        exits. See dojo.importers.import_lock.
+        """
+        depth = getattr(self, "_import_lock_depth", 0)
+        self._import_lock_depth = depth + 1
+        try:
+            yield
+        finally:
+            self._import_lock_depth = depth
+            if depth == 0:
+                unlock_tests_under_import(getattr(self, "_import_locked_tests", []))
+                self._import_locked_tests = []
+
+    def lock_test_under_import(self, test: Test) -> None:
+        """
+        Tell the excess-duplicate delete this test is being imported, until the import ends.
+
+        Call before the first finding is created or matched. Waits if a delete is removing this
+        test's duplicates right now, so the import starts from what that delete left.
+        """
+        if getattr(self, "_import_lock_depth", 0) == 0:
+            msg = "lock_test_under_import must run inside import_lock_scope"
+            raise RuntimeError(msg)
+        locked = getattr(self, "_import_locked_tests", [])
+        if test is None or test.pk is None or test.pk in locked:
+            return
+        self._import_locked_tests = locked + lock_tests_under_import([test.pk])
+
+    def save_matched_finding(self, finding: Finding, *, no_options: bool = False) -> bool:
+        """
+        Save a finding this import matched earlier in the run, unless it has been deleted since.
+
+        A reimport matches report rows against findings it loaded at the start of a batch, and a
+        report can carry several rows for one finding (one per CVE for the same package, say), each
+        of which saves it again. If the finding is deleted in between -- by a user or client
+        deleting it through the API, a test or engagement delete, or the excess-duplicate delete --
+        a plain save() finds no row to UPDATE and falls back to an INSERT that re-creates it from the
+        stale in-memory copy, and a companion saved first (or a deferred field read on the way)
+        fails instead. force_update=True turns that into an error the vanished-row check
+        recognises, from the same statement, so nothing is re-created.
+
+        Returns False when the finding is gone: the caller skips the rest of its handling for that
+        finding and the import carries on with the others.
+        """
+        save = finding.save_no_options if no_options else finding.save
+        try:
+            save(force_update=True)
+        except type(finding).DoesNotExist:
+            pass  # a deferred field read found the row gone before the UPDATE ran
+        except DatabaseError as exception:
+            if not self._is_vanished_row_error(exception):
+                raise
+        else:
+            return True
+        using = finding._state.db or DEFAULT_DB_ALIAS
+        if connections[using].in_atomic_block:
+            transaction.set_rollback(False, using=using)
+        logger.info("Finding %s was deleted while this import was processing it; leaving it deleted", finding.pk)
+        return False
+
     def save_without_resurrecting(self, instance: Engagement | Test) -> None:
         """
         Persist an import target, refusing to re-create it if it was deleted mid-import.
@@ -859,13 +1002,7 @@ def save_without_resurrecting(self, instance: Engagement | Test) -> None:
             using = instance._state.db or DEFAULT_DB_ALIAS
             if connections[using].in_atomic_block:
                 transaction.set_rollback(False, using=using)
-            msg = (
-                f"The {instance._meta.verbose_name} this scan was being imported into "
-                f"(id {instance.pk}) was deleted while the scan was being processed, so "
-                f"the import could not be completed. Nothing was imported. Re-run the "
-                f"import against a {instance._meta.verbose_name} that still exists."
-            )
-            raise ValidationError(msg) from exception
+            raise ValidationError(deleted_target_message(instance)) from exception
 
     def update_test_progress(
         self,

dojo/importers/default_importer.py

@@ -7,7 +7,7 @@
 
 from dojo.celery_dispatch import dojo_dispatch_task
 from dojo.finding import helper as finding_helper
-from dojo.importers.base_importer import BaseImporter, Parser
+from dojo.importers.base_importer import BaseImporter, Parser, reports_a_deleted_target
 from dojo.importers.base_location_manager import LocationHandler
 from dojo.importers.options import ImporterOptions
 from dojo.jira import services as jira_services
@@ -90,6 +90,7 @@ def create_test(
         )
         return self.test
 
+    @reports_a_deleted_target
     def process_scan(
         self,
         scan: TemporaryUploadedFile,
@@ -117,6 +118,8 @@ def process_scan(
         # Note: for fresh imports, parse_findings() calls create_test() internally,
         # so self.test is guaranteed to be set after this call.
         parsed_findings = self.parse_findings(scan, parser) or []
+        # The excess-duplicate delete leaves the duplicates of a test being imported alone.
+        self.lock_test_under_import(self.test)
         new_findings = self.process_findings(parsed_findings, **kwargs)
         # Close any old findings in the processed list if the the user specified for that
         # to occur in the form that is then passed to the kwargs

dojo/importers/default_reimporter.py

@@ -16,7 +16,7 @@
     find_candidates_for_deduplication_unique_id,
     find_candidates_for_reimport_legacy,
 )
-from dojo.importers.base_importer import BaseImporter, Parser
+from dojo.importers.base_importer import BaseImporter, Parser, reports_a_deleted_target
 from dojo.importers.base_location_manager import LocationHandler
 from dojo.importers.options import ImporterOptions
 from dojo.jira import services as jira_services
@@ -91,6 +91,7 @@ def __init__(self, *args, **kwargs):
         )
         self.location_handler = LocationHandler(self.test.engagement.product)
 
+    @reports_a_deleted_target
     def process_scan(
         self,
         scan: TemporaryUploadedFile,
@@ -116,6 +117,9 @@ def process_scan(
         # Get the findings from the parser based on what methods the parser supplies
         # This could either mean traditional file parsing, or API pull parsing
         parsed_findings = self.parse_findings(scan, parser) or []
+        # The excess-duplicate delete leaves the duplicates of a test being imported alone; this
+        # one is about to have its findings matched and written, so say so before the first match.
+        self.lock_test_under_import(self.test)
         (
             new_findings,
             reactivated_findings,
@@ -1176,7 +1180,8 @@ def process_matched_mitigated_finding(
             # restart the sla start date to the current date, finding.save() will set new sla_expiration_date
             existing_finding.sla_start_date = self.now
         # don't dedupe before endpoints/locations are added, postprocessing will be done on next save (in calling method)
-        existing_finding.save_no_options()
+        if not self.save_matched_finding(existing_finding, no_options=True):
+            return existing_finding, True
 
         if existing_finding.duplicate:
             note_entry = f"Un-mitigated by {self.scan_type} re-upload but kept inactive because the finding is a duplicate."
@@ -1228,7 +1233,8 @@ def process_matched_active_finding(
                 if self.verified is not None:
                     existing_finding.verified = self.verified
                 existing_finding = self.process_cve(existing_finding)
-                existing_finding.save_no_options()
+                if not self.save_matched_finding(existing_finding, no_options=True):
+                    return existing_finding, True
                 existing_finding.notes.create(
                     author=self.user,
                     entry=f"Mitigated by {self.test.test_type} re-upload.",
@@ -1248,7 +1254,8 @@ def process_matched_active_finding(
                 if self.verified is not None:
                     existing_finding.verified = self.verified
                 existing_finding = self.process_cve(existing_finding)
-                existing_finding.save_no_options()
+                if not self.save_matched_finding(existing_finding, no_options=True):
+                    return existing_finding, True
                 existing_finding.notes.create(
                     author=self.user,
                     entry=f"Mitigated by {self.test.test_type} re-upload.",
@@ -1266,7 +1273,8 @@ def process_matched_active_finding(
         ):
             existing_finding.component_name = existing_finding.component_name or component_name
             existing_finding.component_version = existing_finding.component_version or component_version
-            existing_finding.save_no_options()
+            if not self.save_matched_finding(existing_finding, no_options=True):
+                return existing_finding, True
         # Return False here to make sure further processing happens
         return existing_finding, False
 
@@ -1450,7 +1458,7 @@ def finding_post_processing(
         # update, and saving here would insert it early -- defeating the buffering and splitting
         # one batched INSERT into per-finding ones. The value rides along when the buffer flushes.
         if finding.cve != old_cve and finding.pk:
-            finding.save()
+            self.save_matched_finding(finding)
         return finding
 
     def process_groups_for_all_findings(