feat: allow import/reimport to wait for deduplication to complete

docs/content/en/open_source/upgrading/2.60.md

@@ -2,6 +2,45 @@
 title: 'Upgrading to DefectDojo Version 2.60.x'
 toc_hide: true
 weight: -20260601
-description: No special instructions.
+description: New deduplication execution mode for import/reimport.
 ---
-There are no special instructions for upgrading to 2.60.x. Check the [Release Notes](https://github.com/DefectDojo/django-DefectDojo/releases/tag/2.60.0) for the contents of the release.
+
+## Deduplication execution mode for import/reimport
+
+This release adds a new `deduplication_execution_mode` setting that controls how
+import/reimport deduplication post-processing is dispatched and whether the API
+response waits for it. It can be set per user (profile) and overridden per request
+on the import and reimport endpoints.
+
+Modes:
+
+- `async` (default): deduplication and the rest of post-processing are dispatched
+  to the background and the response returns immediately. This is the historical
+  behavior; nothing changes for existing users.
+- `async_wait`: post-processing is still dispatched to the background, but the
+  request waits for deduplication to finish before responding. As a result the
+  `scan_added` notification and the statistics in the import/reimport response
+  reflect the deduplicated state (findings that turned out to be duplicates are
+  no longer counted/listed as new). JIRA push, product grading and other
+  non-deduplication tasks remain asynchronous and are not awaited.
+- `sync`: import deduplication runs inline in the web request.
+
+The wait in `async_wait` is bounded by the new `DD_DEDUPLICATION_ASYNC_WAIT_TIMEOUT`
+environment variable (default `60` seconds). If no worker picks up the work within
+the timeout, the request responds anyway (degrading to the `async` outcome) rather
+than hanging.
+
+The import/reimport response now also includes a `deduplication_complete` boolean
+indicating whether deduplication had finished by the time the response was produced.
+
+### Relationship to `block_execution`
+
+The existing `block_execution` profile flag is unchanged. It remains the global
+switch that forces **all** of a user's asynchronous tasks (notifications, JIRA
+push, product grading, deduplication, ...) to run in the foreground.
+`deduplication_execution_mode` is independent and narrower — it only affects
+import/reimport deduplication post-processing. A user who has `block_execution`
+enabled continues to get fully synchronous imports; the upgrade migration seeds
+their `deduplication_execution_mode` to `sync` so behavior is unchanged.
+
+No action is required to upgrade. Check the [Release Notes](https://github.com/DefectDojo/django-DefectDojo/releases/tag/2.60.0) for the contents of the release.

dojo/api_v2/serializers.py

@@ -41,6 +41,7 @@
 from dojo.jira import services as jira_services
 from dojo.location.models import Location, LocationFindingReference
 from dojo.models import (
+    DEDUPLICATION_EXECUTION_MODE_CHOICES,
     IMPORT_ACTIONS,
     SEVERITIES,
     SEVERITY_CHOICES,
@@ -1868,6 +1869,16 @@ class CommonImportScanSerializer(serializers.Serializer):
         allow_null=True, default=None, queryset=User.objects.all(),
     )
     push_to_jira = serializers.BooleanField(default=False)
+    deduplication_execution_mode = serializers.ChoiceField(
+        required=False,
+        allow_null=True,
+        choices=DEDUPLICATION_EXECUTION_MODE_CHOICES,
+        help_text="Override how import post-processing (deduplication, jira push, grading, ...) is executed for "
+        "this request. 'async' dispatches post-processing to the background and responds immediately (default). "
+        "'async_wait' dispatches to the background but waits for deduplication to finish before responding, so "
+        "notifications and the returned statistics reflect the deduplicated state. 'sync' runs everything inline. "
+        "If omitted, falls back to the user's profile setting (deduplication_execution_mode).",
+    )
     environment = serializers.CharField(required=False)
     build_id = serializers.CharField(
         required=False, help_text="ID of the build that was scanned.",
@@ -1913,6 +1924,14 @@ class CommonImportScanSerializer(serializers.Serializer):
         help_text=_("Also referred to as 'Organization' ID."),
     )
     statistics = ImportStatisticsSerializer(read_only=True, required=False)
+    deduplication_complete = serializers.BooleanField(
+        read_only=True,
+        required=False,
+        help_text="Whether deduplication had finished by the time this response was produced. "
+        "True for 'sync' and for 'async_wait' when deduplication completed within the timeout; "
+        "False for 'async' (deduplication is still running in the background) or when an "
+        "'async_wait' import timed out waiting for it.",
+    )
     pro = serializers.ListField(read_only=True, required=False)
     apply_tags_to_findings = serializers.BooleanField(
         help_text="If set to True, the tags will be applied to the findings",
@@ -1971,6 +1990,7 @@ def process_scan(
                 data["product_id"] = test.engagement.product.id
                 data["product_type_id"] = test.engagement.product.prod_type.id
                 data["statistics"] = {"after": test.statistics}
+                data["deduplication_complete"] = importer.deduplication_complete
             duration = time.perf_counter() - start_time
             LargeScanSizeProductAnnouncement(response_data=data, duration=duration)
             ScanTypeProductAnnouncement(response_data=data, scan_type=context.get("scan_type"))
@@ -2069,6 +2089,14 @@ def setup_common_context(self, data: dict) -> dict:
         if eng_end_date:
             context["target_end"] = context.get("engagement_end_date")
 
+        # Resolve the effective import execution mode: request override (if any)
+        # takes precedence over the user's profile setting, otherwise default async.
+        request = self.context.get("request")
+        user = getattr(request, "user", None)
+        context["deduplication_execution_mode"] = Dojo_User.resolve_deduplication_execution_mode(
+            user, data.get("deduplication_execution_mode"),
+        )
+
         return context
 
 
@@ -2242,11 +2270,11 @@ def process_scan(
         try:
             logger.debug(f"process_scan called with context: {context}")
             start_time = time.perf_counter()
+            processor = None
             if test := context.get("test"):
                 statistics_before = test.statistics
-                context["test"], _, _, _, _, _, test_import = self.get_reimporter(
-                    **context,
-                ).process_scan(
+                processor = self.get_reimporter(**context)
+                context["test"], _, _, _, _, _, test_import = processor.process_scan(
                     context.pop("scan", None),
                 )
                 if test_import:
@@ -2258,9 +2286,10 @@ def process_scan(
                 # Do not close old findings when creating a brand new test: there are no
                 # existing findings to compare against, and close_old_findings would
                 # incorrectly close findings from other tests in the same scope.
-                context["test"], _, _, _, _, _, _ = self.get_importer(
+                processor = self.get_importer(
                     **{**context, "close_old_findings": False},
-                ).process_scan(
+                )
+                context["test"], _, _, _, _, _, _ = processor.process_scan(
                     context.pop("scan", None),
                 )
             else:
@@ -2279,6 +2308,8 @@ def process_scan(
                 if statistics_delta:
                     data["statistics"]["delta"] = statistics_delta
                 data["statistics"]["after"] = test.statistics
+                if processor is not None:
+                    data["deduplication_complete"] = processor.deduplication_complete
             duration = time.perf_counter() - start_time
             LargeScanSizeProductAnnouncement(response_data=data, duration=duration)
             ScanTypeProductAnnouncement(response_data=data, scan_type=context.get("scan_type"))

dojo/celery_dispatch.py

@@ -61,10 +61,11 @@ def dojo_dispatch_task(task_or_sig: _SupportsSi | _SupportsApplyAsync | Signatur
 
     - Inject `async_user_id` if missing.
     - Capture and inject pghistory context if available.
-    - Respect `force_sync=True` (foreground execution) and user `block_execution`.
+    - Respect `force_sync=True` (foreground execution) and the user's
+      block_execution flag.
     - Respect `force_async=True` (background execution even when the caller
-      would otherwise run synchronously, e.g. user has `block_execution`).
-      `force_async` wins over `force_sync` and `block_execution`.
+      would otherwise run synchronously, e.g. user has block_execution).
+      `force_async` wins over `force_sync` and block_execution.
     - Support `countdown=<seconds>` for async dispatch.
 
     Returns:

dojo/db_migrations/0269_usercontactinfo_deduplication_execution_mode.py

@@ -0,0 +1,16 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dojo', '0268_release_authorization_to_pro'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='usercontactinfo',
+            name='deduplication_execution_mode',
+            field=models.CharField(blank=True, choices=[('async', 'Async (do not wait)'), ('async_wait', 'Async, wait for deduplication'), ('sync', 'Synchronous (block)')], help_text="Controls how import/reimport deduplication post-processing is executed. 'Async' dispatches it to the background and returns immediately (default). 'Async, wait for deduplication' dispatches to the background but waits for deduplication to finish before responding, so notifications and statistics reflect the deduplicated state. 'Synchronous' runs the import deduplication inline. Can be overridden per request. Independent of block_execution, which forces all async tasks (notifications, jira, ...) to the foreground.", max_length=20, null=True),
+        ),
+    ]

dojo/db_migrations/0270_seed_deduplication_execution_mode.py

@@ -0,0 +1,30 @@
+from django.db import migrations
+
+
+def seed_deduplication_execution_mode(apps, schema_editor):
+    """
+    Seed the new import deduplication execution mode from the legacy block_execution flag.
+
+    block_execution remains the global "run all async tasks in the foreground" switch;
+    users who had it enabled get the synchronous deduplication mode so import behavior is
+    unchanged for them.
+    """
+    UserContactInfo = apps.get_model("dojo", "UserContactInfo")
+    UserContactInfo.objects.filter(block_execution=True).update(deduplication_execution_mode="sync")
+
+
+def unseed_deduplication_execution_mode(apps, schema_editor):
+    """Reverse: clear the seeded synchronous mode."""
+    UserContactInfo = apps.get_model("dojo", "UserContactInfo")
+    UserContactInfo.objects.filter(deduplication_execution_mode="sync").update(deduplication_execution_mode=None)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dojo', '0269_usercontactinfo_deduplication_execution_mode'),
+    ]
+
+    operations = [
+        migrations.RunPython(seed_deduplication_execution_mode, unseed_deduplication_execution_mode),
+    ]

dojo/engagement/views.py

@@ -954,6 +954,10 @@ def process_form(
             "create_finding_groups_for_all_findings": form.cleaned_data.get("create_finding_groups_for_all_findings", None),
             "environment": self.get_development_environment(environment_name=form.cleaned_data.get("environment")),
         })
+        # Honor the user's profile deduplication_execution_mode for UI imports. The API resolves
+        # this in the serializer; the UI has no per-import selector, so fall back to the profile
+        # (or block_execution) instead of silently defaulting to async.
+        context["deduplication_execution_mode"] = Dojo_User.resolve_deduplication_execution_mode(request.user)
         # Create the engagement if necessary
         self.create_engagement(context)
         # close_old_findings_product_scope is a modifier of close_old_findings.

dojo/finding/helper.py

@@ -456,7 +456,13 @@ def post_process_finding_save_internal(finding, dedupe_option=True, rules_option
             jira_services.push(finding.finding_group)
 
 
-@app.task
+# ignore_result=False so the 'async_wait' import execution mode can join on the
+# dispatched batch via AsyncResult.get() even when CELERY_TASK_IGNORE_RESULT=True.
+# NOTE: this override may not be strictly necessary — in the past .get()/await
+# appears to have worked with the global CELERY_TASK_IGNORE_RESULT=True and a
+# Redis broker. Needs verification against a real broker/worker setup; if join
+# works without it, this override can be removed to avoid storing extra results.
+@app.task(ignore_result=False)
 def post_process_findings_batch(
     finding_ids,
     *args,

dojo/forms.py

@@ -2393,7 +2393,7 @@ class Meta:
         # Swap order: password_last_reset before token_last_reset
         field_order = [
             "title", "phone_number", "cell_number", "twitter_username", "github_username",
-            "slack_username", "ui_use_tailwind", "block_execution", "force_password_reset", "reset_api_token",
+            "slack_username", "ui_use_tailwind", "block_execution", "deduplication_execution_mode", "force_password_reset", "reset_api_token",
             "password_last_reset", "token_last_reset",
         ]