fix(api): duplicate finding rows in outputs on tasks re-run (#12097)

api/changelog.d/output-generation-duplicate-rows.fixed.md

@@ -0,0 +1 @@
+Output generation now removes the scan's temporary output directory before writing, so a re-run of the task for the same scan (e.g. broker redelivery after a worker is killed mid-run) no longer appends to the previous run's files and duplicates finding rows in the exported CSV and other outputs

api/src/backend/tasks/tasks.py

@@ -797,12 +797,34 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
         if name not in frameworks_bulk and universal_bulk[name].outputs
     }
     frameworks_avail = get_compliance_frameworks(provider_type)
+    # Idempotency: a previous run of this task for the same scan may have left
+    # output files behind (e.g. broker redelivery after a worker was killed
+    # mid-run with task_acks_late, or a successful run on a deployment without
+    # S3 where the tmp dir is not removed). Output writers open files in append
+    # mode with a deterministic path (derived from scan.started_at), so reusing
+    # them would append every finding row again and duplicate the CSV/output
+    # rows. Start from a clean slate before (re)generating.
+    scan_tmp_dir = _scan_tmp_output_directory(tenant_id, scan_id)
+    if os.path.exists(scan_tmp_dir):
+        rmtree(scan_tmp_dir, ignore_errors=True)
+        # The writers below open output files in append mode with deterministic
+        # paths (derived from scan.started_at). Any stale file that survives the
+        # cleanup would get every finding row appended again, which is the exact
+        # duplication this guards against. Continuing is therefore unsafe: abort
+        # so `ScanReportRLSTask.on_failure` removes the tmp dir and the retry
+        # starts from a clean slate instead of publishing duplicated rows.
+        if os.path.exists(scan_tmp_dir):
+            raise RuntimeError(
+                "Could not remove stale output directory for scan "
+                f"{scan_id} before generating outputs; aborting to avoid "
+                "duplicated rows in appended outputs."
+            )
+
     out_dir, comp_dir = _generate_output_directory(
         DJANGO_TMP_OUTPUT_DIRECTORY, provider_uid, tenant_id, scan_id
     )
     # Removed on success here and on failure by ScanReportRLSTask.on_failure,
     # so partial artifacts do not accumulate and fill the disk (ENOSPC).
-    scan_tmp_dir = _scan_tmp_output_directory(tenant_id, scan_id)
 
     def get_writer(writer_map, name, factory, is_last):
         """

api/src/backend/tasks/tests/test_tasks.py

@@ -420,6 +420,124 @@ class TestGenerateOutputs:
             assert result == {"upload": False}
             mock_scan_update.return_value.update.assert_called_once()
 
+    def test_generate_outputs_removes_previous_run_artifacts(self):
+        """Regression for PROWLER-2266.
+
+        Output writers open files in append mode with a deterministic path
+        (derived from scan.started_at). If this task runs again for the same
+        scan (e.g. broker redelivery after a worker is killed mid-run with
+        task_acks_late), reusing the leftover files appends every finding row
+        again, duplicating rows in the CSV/output while the API console keeps
+        showing a single finding. The task must start from a clean slate by
+        removing the scan's tmp output directory before (re)generating.
+        """
+        import tempfile
+        from pathlib import Path
+
+        with tempfile.TemporaryDirectory() as tmp_root:
+            # Simulate artifacts left behind by a previous run of the same scan.
+            scan_tmp_dir = Path(tmp_root) / self.tenant_id / self.scan_id
+            scan_tmp_dir.mkdir(parents=True)
+            stale_artifact = scan_tmp_dir / "prowler-output-aws-20260723120000.csv"
+            stale_artifact.write_text("HEADER\nold-finding-row\n")
+
+            with (
+                patch("tasks.tasks.DJANGO_TMP_OUTPUT_DIRECTORY", tmp_root),
+                patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter,
+                patch("tasks.tasks.Provider.objects.get"),
+                patch("tasks.tasks.initialize_prowler_provider"),
+                patch("tasks.tasks.Compliance.get_bulk"),
+                patch("tasks.tasks.get_compliance_frameworks"),
+                patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
+                patch("tasks.tasks.Finding.all_objects.filter") as mock_findings,
+                patch(
+                    "tasks.tasks._generate_output_directory",
+                    return_value=("/tmp/test/out", "/tmp/test/comp"),
+                ),
+                patch("tasks.tasks.FindingOutput._transform_findings_stats"),
+                patch("tasks.tasks.FindingOutput.transform_api_finding"),
+                patch(
+                    "tasks.tasks.OUTPUT_FORMATS_MAPPING",
+                    {
+                        "json": {
+                            "class": MagicMock(name="Writer"),
+                            "suffix": ".json",
+                            "kwargs": {},
+                        }
+                    },
+                ),
+                patch("tasks.tasks.COMPLIANCE_CLASS_MAP", {"aws": []}),
+                patch(
+                    "tasks.tasks._compress_output_files", return_value="/tmp/compressed"
+                ),
+                patch("tasks.tasks._upload_to_s3", return_value=None),
+                patch("tasks.tasks.Scan.all_objects.filter"),
+            ):
+                mock_filter.return_value.exists.return_value = True
+                mock_findings.return_value.order_by.return_value.iterator.return_value = [
+                    [MagicMock()],
+                    True,
+                ]
+
+                generate_outputs_task(
+                    scan_id=self.scan_id,
+                    provider_id=self.provider_id,
+                    tenant_id=self.tenant_id,
+                )
+
+            # The stale artifacts from the previous run must be gone, so the
+            # append-mode writers cannot duplicate rows onto them.
+            assert not stale_artifact.exists()
+            assert not scan_tmp_dir.exists()
+
+    def test_generate_outputs_aborts_when_stale_cleanup_fails(self):
+        """Regression for PROWLER-2266.
+
+        If the stale output directory cannot be removed (e.g. permission error),
+        the leftover files would be reopened in append mode and every finding
+        row would be duplicated. The task must abort instead of continuing and
+        publishing duplicated rows, so the retry can start from a clean slate.
+        """
+        import tempfile
+        from pathlib import Path
+
+        with tempfile.TemporaryDirectory() as tmp_root:
+            scan_tmp_dir = Path(tmp_root) / self.tenant_id / self.scan_id
+            scan_tmp_dir.mkdir(parents=True)
+            stale_artifact = scan_tmp_dir / "prowler-output-aws-20260723120000.csv"
+            stale_artifact.write_text("HEADER\nold-finding-row\n")
+
+            with (
+                patch("tasks.tasks.DJANGO_TMP_OUTPUT_DIRECTORY", tmp_root),
+                patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter,
+                patch("tasks.tasks.Provider.objects.get"),
+                patch("tasks.tasks.initialize_prowler_provider"),
+                patch("tasks.tasks.Compliance.get_bulk"),
+                patch("tasks.tasks.get_compliance_frameworks"),
+                patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
+                # `rmtree(ignore_errors=True)` swallows the failure and leaves the
+                # directory behind; simulate that with a no-op so the guard fires.
+                patch("tasks.tasks.rmtree"),
+                patch("tasks.tasks._generate_output_directory") as mock_gen_dir,
+                patch("tasks.tasks._compress_output_files") as mock_compress,
+                patch("tasks.tasks._upload_to_s3") as mock_upload,
+                patch("tasks.tasks.Scan.all_objects.filter") as mock_scan_update,
+            ):
+                mock_filter.return_value.exists.return_value = True
+
+                with pytest.raises(RuntimeError, match="stale output directory"):
+                    generate_outputs_task(
+                        scan_id=self.scan_id,
+                        provider_id=self.provider_id,
+                        tenant_id=self.tenant_id,
+                    )
+
+            # The task must abort before generating/publishing any output.
+            mock_gen_dir.assert_not_called()
+            mock_compress.assert_not_called()
+            mock_upload.assert_not_called()
+            mock_scan_update.assert_not_called()
+
     def test_generate_outputs_triggers_html_extra_update(self):
         mock_finding_output = MagicMock()
         mock_finding_output.compliance = {"cis": ["requirement-1", "requirement-2"]}