fix(api): skip scan tasks when provider was deleted (#11185)

api/CHANGELOG.md

@@ -13,6 +13,10 @@ All notable changes to the **Prowler API** are documented in this file.
 - Replace `poetry` with `uv` (`0.11.14`) as the API package manager; migrate `pyproject.toml` to `[dependency-groups]` and regenerate as `uv.lock` [(#10775)](https://github.com/prowler-cloud/prowler/pull/10775)
 - Remove orphaned `gin_resources_search_idx` declaration from `Resource.Meta.indexes` (DB index dropped in `0072_drop_unused_indexes`) [(#11001)](https://github.com/prowler-cloud/prowler/pull/11001)
 
+### 🐞 Fixed
+
+- `perform_scan_task` and `perform_scheduled_scan_task` now short-circuit with a warning and `return None` when the target provider no longer exists, instead of letting `handle_provider_deletion` raise `ProviderDeletedException`. `perform_scheduled_scan_task` also removes any orphan `PeriodicTask` it finds so beat stops re-firing scans for deleted providers. Prevents queued messages for deleted providers from being recorded as `FAILURE` and, in one-shot scan-worker deployments, from burning a fresh container per redelivery [(#11185)](https://github.com/prowler-cloud/prowler/pull/11185)
+
 ---
 
 ## [1.27.2] (Prowler UNRELEASED)

api/src/backend/tasks/tasks.py

@@ -69,7 +69,7 @@ from tasks.utils import (
 
 from api.compliance import get_compliance_frameworks
 from api.db_router import READ_REPLICA_ALIAS
-from api.db_utils import rls_transaction
+from api.db_utils import delete_related_daily_task, rls_transaction
 from api.decorators import handle_provider_deletion, set_tenant
 from api.models import Finding, Integration, Provider, Scan, ScanSummary, StateChoices
 from api.utils import initialize_prowler_provider
@@ -274,6 +274,17 @@ def perform_scan_task(
     Returns:
         dict: The result of the scan execution, typically including the status and results of the performed checks.
     """
+    with rls_transaction(tenant_id):
+        if not Provider.objects.filter(pk=provider_id).exists():
+            logger.warning(
+                "scan-perform skipped: provider %s no longer exists "
+                "(tenant=%s, scan=%s)",
+                provider_id,
+                tenant_id,
+                scan_id,
+            )
+            return None
+
     result = perform_prowler_scan(
         tenant_id=tenant_id,
         scan_id=scan_id,
@@ -310,6 +321,16 @@ def perform_scheduled_scan_task(self, tenant_id: str, provider_id: str):
     task_id = self.request.id
 
     with rls_transaction(tenant_id):
+        if not Provider.objects.filter(pk=provider_id).exists():
+            logger.warning(
+                "scheduled scan-perform skipped: provider %s no longer exists "
+                "(tenant=%s)",
+                provider_id,
+                tenant_id,
+            )
+            delete_related_daily_task(provider_id)
+            return None
+
         periodic_task_instance = PeriodicTask.objects.get(
             name=f"scan-perform-scheduled-{provider_id}"
         )

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

@@ -21,6 +21,7 @@ from tasks.tasks import (
     check_lighthouse_provider_connection_task,
     generate_outputs_task,
     perform_attack_paths_scan_task,
+    perform_scan_task,
     perform_scheduled_scan_task,
     reaggregate_all_finding_group_summaries_task,
     refresh_lighthouse_provider_models_task,
@@ -2454,6 +2455,57 @@ class TestPerformScheduledScanTask:
             == 1
         )
 
+    def test_no_op_when_provider_does_not_exist(self, tenants_fixture):
+        """Return None without raising when the provider was already deleted."""
+        tenant = tenants_fixture[0]
+        missing_provider_id = str(uuid.uuid4())
+        task_id = str(uuid.uuid4())
+        self._create_task_result(tenant.id, task_id)
+        # Orphan PeriodicTask left behind from a previous lifecycle.
+        self._create_periodic_task(missing_provider_id, tenant.id)
+        orphan_name = f"scan-perform-scheduled-{missing_provider_id}"
+        assert PeriodicTask.objects.filter(name=orphan_name).exists()
+
+        with (
+            patch("tasks.tasks.perform_prowler_scan") as mock_scan,
+            patch("tasks.tasks._perform_scan_complete_tasks") as mock_complete_tasks,
+            self._override_task_request(perform_scheduled_scan_task, id=task_id),
+        ):
+            result = perform_scheduled_scan_task.run(
+                tenant_id=str(tenant.id), provider_id=missing_provider_id
+            )
+
+        assert result is None
+        mock_scan.assert_not_called()
+        mock_complete_tasks.assert_not_called()
+        # Orphan PeriodicTask is cleaned up so beat stops re-firing it.
+        assert not PeriodicTask.objects.filter(name=orphan_name).exists()
+
+
+@pytest.mark.django_db
+class TestPerformScanTask:
+    """Unit tests for perform_scan_task."""
+
+    def test_no_op_when_provider_does_not_exist(self, tenants_fixture):
+        """Return None without raising when the provider was already deleted."""
+        tenant = tenants_fixture[0]
+        missing_provider_id = str(uuid.uuid4())
+        scan_id = str(uuid.uuid4())
+
+        with (
+            patch("tasks.tasks.perform_prowler_scan") as mock_scan,
+            patch("tasks.tasks._perform_scan_complete_tasks") as mock_complete_tasks,
+        ):
+            result = perform_scan_task.run(
+                tenant_id=str(tenant.id),
+                scan_id=scan_id,
+                provider_id=missing_provider_id,
+            )
+
+        assert result is None
+        mock_scan.assert_not_called()
+        mock_complete_tasks.assert_not_called()
+
 
 @pytest.mark.django_db
 class TestReaggregateAllFindingGroupSummaries: