fix(api): resolve attack paths scan reliability issues (#12019)

api/src/backend/tasks/jobs/attack_paths/aws.py

@@ -8,6 +8,8 @@ import aioboto3
 import boto3
 import botocore
 import neo4j
+import neo4j.exceptions
+from api.attack_paths.database import DATABASE_NOT_FOUND_CODE
 from api.models import (
     AttackPathsScan as ProwlerAPIAttackPathsScan,
 )
@@ -347,6 +349,12 @@ def sync_aws_account(
                 )
 
             except Exception as e:
+                if (
+                    isinstance(e, neo4j.exceptions.Neo4jError)
+                    and e.code == DATABASE_NOT_FOUND_CODE
+                ):
+                    raise
+
                 logger.info(
                     f"Synced function {func_name} for AWS account {prowler_api_provider.uid} in {time.perf_counter() - func_t0:.3f}s (FAILED)"
                 )

api/src/backend/tasks/jobs/attack_paths/config.py

@@ -10,13 +10,10 @@ NormalizedList = _provider_config.NormalizedList
 PROVIDER_CONFIGS = _provider_config.PROVIDER_CONFIGS
 ProviderConfig = _provider_config.ProviderConfig
 
-# Batch size for Neo4j write operations (resource labeling, cleanup)
-BATCH_SIZE = env.int("ATTACK_PATHS_BATCH_SIZE", 1000)
+# Batch size for graph mutation operations (resource labeling and subgraph deletion)
+GRAPH_MUTATION_BATCH_SIZE = env.int("ATTACK_PATHS_GRAPH_MUTATION_BATCH_SIZE", 1000)
 # Batch size for Postgres findings fetch (keyset pagination page size)
 FINDINGS_BATCH_SIZE = env.int("ATTACK_PATHS_FINDINGS_BATCH_SIZE", 1000)
-# Batch size for temp-to-tenant graph sync (nodes and relationships per cursor page)
-SYNC_BATCH_SIZE = env.int("ATTACK_PATHS_SYNC_BATCH_SIZE", 1000)
-
 # Neo4j internal labels (Prowler-specific, not provider-specific)
 # - `Internet`: Singleton node representing external internet access for exposed-resource queries
 # - `ProwlerFinding`: Label for finding nodes created by Prowler and linked to cloud resources

api/src/backend/tasks/jobs/attack_paths/findings.py

@@ -21,8 +21,8 @@ from cartography.config import Config as CartographyConfig
 from celery.utils.log import get_task_logger
 from prowler.config import config as ProwlerConfig
 from tasks.jobs.attack_paths.config import (
-    BATCH_SIZE,
     FINDINGS_BATCH_SIZE,
+    GRAPH_MUTATION_BATCH_SIZE,
     get_node_uid_field,
     get_provider_resource_label,
     get_root_node_label,
@@ -135,7 +135,7 @@ def add_resource_label(
     while labeled_count > 0:
         result = neo4j_session.run(
             query,
-            {"provider_uid": provider_uid, "batch_size": BATCH_SIZE},
+            {"provider_uid": provider_uid, "batch_size": GRAPH_MUTATION_BATCH_SIZE},
         )
         labeled_count = result.single().get("labeled_count", 0)
         total_labeled += labeled_count

api/src/backend/tasks/jobs/attack_paths/scan.py

@@ -372,7 +372,19 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
 
     except Exception as e:
         exception_message = utils.stringify_exception(e, "Attack Paths scan failed")
-        logger.exception(exception_message)
+        temporary_database_missing = (
+            isinstance(e, graph_database.GraphDatabaseQueryException)
+            and e.code == graph_database.DATABASE_NOT_FOUND_CODE
+            and tmp_database_name in str(e)
+        )
+        if temporary_database_missing:
+            logger.warning(exception_message)
+        else:
+            logger.exception(exception_message)
+        cleanup_log_level = (
+            logging.WARNING if temporary_database_missing else logging.ERROR
+        )
+        cleanup_exc_info = not temporary_database_missing
         ingestion_exceptions["global_error"] = exception_message
 
         # Recover `graph_data_ready` based on how far the swap got
@@ -387,19 +399,24 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
                 )
 
         except Exception:
-            logger.error(
-                f"Failed to recover `graph_data_ready` for provider {attack_paths_scan.provider_id}",
-                exc_info=True,
+            logger.log(
+                cleanup_log_level,
+                "Failed to recover `graph_data_ready` for provider "
+                f"{attack_paths_scan.provider_id}",
+                exc_info=cleanup_exc_info,
             )
 
         # Dropping the temporary database if it still exists
         try:
             graph_database.drop_database(tmp_cartography_config.neo4j_database)
 
-        except Exception as e:
-            logger.error(
-                f"Failed to drop temporary Neo4j database `{tmp_cartography_config.neo4j_database}` during cleanup: {e}",
-                exc_info=True,
+        except Exception as cleanup_error:
+            logger.log(
+                cleanup_log_level,
+                "Failed to drop temporary Neo4j database "
+                f"`{tmp_cartography_config.neo4j_database}` during cleanup: "
+                f"{cleanup_error}",
+                exc_info=cleanup_exc_info,
             )
 
         # Set Attack Paths scan state to FAILED
@@ -407,10 +424,12 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
             db_utils.finish_attack_paths_scan(
                 attack_paths_scan, StateChoices.FAILED, ingestion_exceptions
             )
-        except Exception as e:
-            logger.error(
-                f"Could not mark Attack Paths scan {attack_paths_scan.id} as `FAILED` (row may have been deleted): {e}",
-                exc_info=True,
+        except Exception as cleanup_error:
+            logger.log(
+                cleanup_log_level,
+                f"Could not mark Attack Paths scan {attack_paths_scan.id} as `FAILED` "
+                f"(row may have been deleted): {cleanup_error}",
+                exc_info=cleanup_exc_info,
             )
 
         raise

api/src/backend/tasks/jobs/attack_paths/sync.py

@@ -30,7 +30,6 @@ from tasks.jobs.attack_paths.config import (
     PROVIDER_CONFIGS,
     PROVIDER_ISOLATION_PROPERTIES,
     PROVIDER_RESOURCE_LABEL,
-    SYNC_BATCH_SIZE,
     NormalizedList,
     get_provider_label,
     get_tenant_label,
@@ -116,6 +115,7 @@ def sync_nodes(
     Source and target sessions are opened sequentially per batch to avoid
     holding two Bolt connections simultaneously for the entire sync duration.
     """
+    batch_size = sink.sync_batch_size
     t0 = time.perf_counter()
     last_id = -1
     parents_synced = 0
@@ -137,7 +137,7 @@ def sync_nodes(
         with graph_database.get_session(source_database) as source_session:
             result = source_session.run(
                 NODE_FETCH_QUERY,
-                {"last_id": last_id, "batch_size": SYNC_BATCH_SIZE},
+                {"last_id": last_id, "batch_size": batch_size},
             )
             for record in result:
                 batch_count += 1
@@ -156,17 +156,17 @@ def sync_nodes(
 
         for labels, batch in parent_groups.items():
             rendered_labels = _render_labels(labels, extra_labels)
-            for sink_batch in _iter_sink_batches(batch):
+            for sink_batch in _iter_sink_batches(batch, batch_size):
                 sink.write_nodes(target_database, rendered_labels, sink_batch)
 
         for child_label, batch in child_groups.items():
             rendered_labels = _render_labels((child_label,), extra_labels)
-            for sink_batch in _iter_sink_batches(batch):
+            for sink_batch in _iter_sink_batches(batch, batch_size):
                 sink.write_nodes(target_database, rendered_labels, sink_batch)
             children_synced += len(batch)
 
         for rel_type, batch in rel_groups.items():
-            for sink_batch in _iter_sink_batches(batch):
+            for sink_batch in _iter_sink_batches(batch, batch_size):
                 sink.write_relationships(
                     target_database, rel_type, provider_id, sink_batch
                 )
@@ -205,6 +205,7 @@ def sync_relationships(
     Source and target sessions are opened sequentially per batch to avoid
     holding two Bolt connections simultaneously for the entire sync duration.
     """
+    batch_size = sink.sync_batch_size
     t0 = time.perf_counter()
     last_id = -1
     total_synced = 0
@@ -217,7 +218,7 @@ def sync_relationships(
         with graph_database.get_session(source_database) as source_session:
             result = source_session.run(
                 RELATIONSHIPS_FETCH_QUERY,
-                {"last_id": last_id, "batch_size": SYNC_BATCH_SIZE},
+                {"last_id": last_id, "batch_size": batch_size},
             )
             for record in result:
                 batch_count += 1
@@ -229,7 +230,7 @@ def sync_relationships(
             break
 
         for rel_type, batch in grouped.items():
-            for sink_batch in _iter_sink_batches(batch):
+            for sink_batch in _iter_sink_batches(batch, batch_size):
                 sink.write_relationships(
                     target_database, rel_type, provider_id, sink_batch
                 )
@@ -247,10 +248,9 @@ def sync_relationships(
 
 def _iter_sink_batches(
     rows: list[dict[str, Any]],
-    batch_size: int | None = None,
+    batch_size: int,
 ) -> Iterator[list[dict[str, Any]]]:
     """Yield final sink write batches after source rows have been transformed."""
-    batch_size = SYNC_BATCH_SIZE if batch_size is None else batch_size
     if batch_size <= 0:
         raise ValueError("Sink batch size must be greater than zero")
 

api/src/backend/tasks/tasks.py

@@ -11,6 +11,7 @@ from api.compliance import (
 from api.db_router import READ_REPLICA_ALIAS
 from api.db_utils import delete_related_daily_task, rls_transaction
 from api.decorators import handle_provider_deletion, set_tenant
+from api.exceptions import ProviderDeletedException
 from api.models import (
     Finding,
     Integration,
@@ -666,7 +667,13 @@ class AttackPathsScanRLSTask(RLSTask):
         scan_id = kwargs.get("scan_id")
 
         if tenant_id and scan_id:
-            logger.error(f"Attack paths scan task {task_id} failed: {exc}")
+            if isinstance(exc, ProviderDeletedException):
+                logger.warning(
+                    f"Attack paths scan task {task_id} stopped because its provider "
+                    f"or tenant was deleted: {exc}"
+                )
+            else:
+                logger.error(f"Attack paths scan task {task_id} failed: {exc}")
             attack_paths_db_utils.fail_attack_paths_scan(tenant_id, scan_id, str(exc))
 
 

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

@@ -0,0 +1,102 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import neo4j.exceptions
+import pytest
+from tasks.jobs.attack_paths import aws
+
+DATABASE_NOT_FOUND_CODE = "Neo.ClientError.Database.DatabaseNotFound"
+
+
+def _make_neo4j_error(code: str) -> neo4j.exceptions.Neo4jError:
+    return neo4j.exceptions.Neo4jError._hydrate_neo4j(
+        code=code,
+        message="graph query failed",
+    )
+
+
+def _resource_functions(failing_sync, following_sync):
+    return {
+        "failing_sync": failing_sync,
+        "following_sync": following_sync,
+        "permission_relationships": MagicMock(),
+        "resourcegroupstaggingapi": MagicMock(),
+    }
+
+
+def test_sync_aws_account_reraises_database_not_found_immediately():
+    error = _make_neo4j_error(DATABASE_NOT_FOUND_CODE)
+    failing_sync = MagicMock(side_effect=error)
+    following_sync = MagicMock()
+
+    with (
+        patch.object(
+            aws.cartography_aws,
+            "RESOURCE_FUNCTIONS",
+            _resource_functions(failing_sync, following_sync),
+        ),
+        patch.object(aws.db_utils, "update_attack_paths_scan_progress"),
+        patch.object(aws.utils, "stringify_exception") as stringify_exception,
+        patch.object(aws.logger, "warning") as warning,
+        pytest.raises(neo4j.exceptions.Neo4jError) as exc_info,
+    ):
+        aws.sync_aws_account(
+            SimpleNamespace(uid="123456789012"),
+            [
+                "failing_sync",
+                "following_sync",
+                "permission_relationships",
+                "resourcegroupstaggingapi",
+            ],
+            {},
+            MagicMock(),
+        )
+
+    assert exc_info.value is error
+    following_sync.assert_not_called()
+    stringify_exception.assert_not_called()
+    warning.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        _make_neo4j_error("Neo.ClientError.Statement.SyntaxError"),
+        RuntimeError("resource sync failed"),
+    ],
+    ids=["different-neo4j-error", "non-neo4j-error"],
+)
+def test_sync_aws_account_warns_and_continues_for_other_exceptions(error):
+    failing_sync = MagicMock(side_effect=error)
+    following_sync = MagicMock()
+
+    with (
+        patch.object(
+            aws.cartography_aws,
+            "RESOURCE_FUNCTIONS",
+            _resource_functions(failing_sync, following_sync),
+        ),
+        patch.object(aws.db_utils, "update_attack_paths_scan_progress"),
+        patch.object(
+            aws.utils,
+            "stringify_exception",
+            return_value="formatted failure",
+        ),
+        patch.object(aws.logger, "warning") as warning,
+    ):
+        failed_syncs = aws.sync_aws_account(
+            SimpleNamespace(uid="123456789012"),
+            [
+                "failing_sync",
+                "following_sync",
+                "permission_relationships",
+                "resourcegroupstaggingapi",
+            ],
+            {},
+            MagicMock(),
+        )
+
+    assert failed_syncs == {"failing_sync": "formatted failure"}
+    following_sync.assert_called_once_with()
+    warning.assert_called_once()
+    assert "Continuing to the next AWS sync function" in warning.call_args.args[0]

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

@@ -1,3 +1,4 @@
+import logging
 from contextlib import nullcontext
 from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
@@ -5,7 +6,9 @@ from unittest.mock import MagicMock, call, patch
 from uuid import uuid4
 
 import pytest
+from api.attack_paths.database import GraphDatabaseQueryException
 from api.db_utils import rls_transaction
+from api.exceptions import ProviderDeletedException
 from api.models import (
     AttackPathsScan,
     Finding,
@@ -250,6 +253,32 @@ class TestAttackPathsRun:
         mock_starting.assert_not_called()
         mock_create_db.assert_not_called()
 
+    @pytest.mark.parametrize(
+        ("ingestion_error", "temporary_database_missing"),
+        [
+            (RuntimeError("ingestion boom"), False),
+            (
+                GraphDatabaseQueryException(
+                    message="Graph not found: db-scan-id",
+                    code="Neo.ClientError.Database.DatabaseNotFound",
+                ),
+                True,
+            ),
+            (
+                GraphDatabaseQueryException(
+                    message="Graph not found: db-tenant-id",
+                    code="Neo.ClientError.Database.DatabaseNotFound",
+                ),
+                False,
+            ),
+        ],
+        ids=[
+            "regular-error",
+            "temporary-database-missing",
+            "sink-database-missing",
+        ],
+    )
+    @patch("tasks.jobs.attack_paths.scan.logger")
     @patch(
         "tasks.jobs.attack_paths.scan.utils.stringify_exception",
         return_value="Cartography failed: ingestion boom",
@@ -302,6 +331,9 @@ class TestAttackPathsRun:
         mock_drop_db,
         mock_event_loop,
         mock_stringify,
+        mock_logger,
+        ingestion_error,
+        temporary_database_missing,
         tenants_fixture,
         aws_provider,
         scans_fixture,
@@ -321,7 +353,11 @@ class TestAttackPathsRun:
         session_ctx = MagicMock()
         session_ctx.__enter__.return_value = mock_session
         session_ctx.__exit__.return_value = False
-        ingestion_fn = MagicMock(side_effect=RuntimeError("ingestion boom"))
+        ingestion_fn = MagicMock(side_effect=ingestion_error)
+        if temporary_database_missing:
+            mock_finish.side_effect = DatabaseError(
+                "Save with update_fields did not affect any rows"
+            )
 
         with (
             patch(
@@ -337,13 +373,28 @@ class TestAttackPathsRun:
                 return_value=ingestion_fn,
             ),
         ):
-            with pytest.raises(RuntimeError, match="ingestion boom"):
+            with pytest.raises(type(ingestion_error)):
                 attack_paths_run(str(tenant.id), str(scan.id), "task-456")
 
         failure_args = mock_finish.call_args[0]
         assert failure_args[0] is attack_paths_scan
         assert failure_args[1] == StateChoices.FAILED
         assert failure_args[2] == {"global_error": "Cartography failed: ingestion boom"}
+        mock_drop_db.assert_called_once_with("db-scan-id")
+        if temporary_database_missing:
+            mock_logger.warning.assert_any_call("Cartography failed: ingestion boom")
+            mock_logger.exception.assert_not_called()
+            mock_logger.log.assert_called_once_with(
+                logging.WARNING,
+                f"Could not mark Attack Paths scan {attack_paths_scan.id} as `FAILED` "
+                "(row may have been deleted): Save with update_fields did not affect "
+                "any rows",
+                exc_info=False,
+            )
+        else:
+            mock_logger.exception.assert_called_once_with(
+                "Cartography failed: ingestion boom"
+            )
 
     @patch(
         "tasks.jobs.attack_paths.scan.utils.stringify_exception",
@@ -1265,6 +1316,33 @@ class TestAttackPathsScanRLSTaskOnFailure:
 
         mock_fail.assert_called_once_with("t-1", "s-1", "boom")
 
+    def test_on_failure_logs_provider_deletion_as_warning(self):
+        from tasks.tasks import AttackPathsScanRLSTask
+
+        task = AttackPathsScanRLSTask()
+        error = ProviderDeletedException("provider deleted")
+
+        with (
+            patch("tasks.tasks.logger") as mock_logger,
+            patch(
+                "tasks.tasks.attack_paths_db_utils.fail_attack_paths_scan"
+            ) as mock_fail,
+        ):
+            task.on_failure(
+                exc=error,
+                task_id="task-abc",
+                args=(),
+                kwargs={"tenant_id": "t-1", "scan_id": "s-1"},
+                _einfo=None,
+            )
+
+        mock_logger.warning.assert_called_once_with(
+            "Attack paths scan task task-abc stopped because its provider or tenant "
+            "was deleted: provider deleted"
+        )
+        mock_logger.error.assert_not_called()
+        mock_fail.assert_called_once_with("t-1", "s-1", "provider deleted")
+
     def test_on_failure_skips_when_missing_kwargs(self):
         from tasks.tasks import AttackPathsScanRLSTask
 
@@ -1896,7 +1974,7 @@ class TestSyncNodes:
         mock_source_1.run.return_value = [row]
         mock_source_2 = MagicMock()
         mock_source_2.run.return_value = []
-        sink = MagicMock()
+        sink = MagicMock(sync_batch_size=1000)
 
         with patch(
             "tasks.jobs.attack_paths.sync.graph_database.get_session",
@@ -1933,7 +2011,7 @@ class TestSyncNodes:
         src_1.run.return_value = [row]
         src_2 = MagicMock()
         src_2.run.return_value = []
-        sink = MagicMock()
+        sink = MagicMock(sync_batch_size=1000)
         sink.write_nodes.side_effect = lambda *_a, **_kw: call_order.append(
             "sink:write"
         )
@@ -1969,18 +2047,15 @@ class TestSyncNodes:
         src_2.run.return_value = [row_b]
         src_3 = MagicMock()
         src_3.run.return_value = []
-        sink = MagicMock()
+        sink = MagicMock(sync_batch_size=1)
 
-        with (
-            patch(
-                "tasks.jobs.attack_paths.sync.graph_database.get_session",
-                side_effect=[
-                    _make_session_ctx(src_1),
-                    _make_session_ctx(src_2),
-                    _make_session_ctx(src_3),
-                ],
-            ),
-            patch("tasks.jobs.attack_paths.sync.SYNC_BATCH_SIZE", 1),
+        with patch(
+            "tasks.jobs.attack_paths.sync.graph_database.get_session",
+            side_effect=[
+                _make_session_ctx(src_1),
+                _make_session_ctx(src_2),
+                _make_session_ctx(src_3),
+            ],
         ):
             result = sync_module.sync_nodes("src", "tgt", "t-1", "p-1", sink, [])
 
@@ -2009,17 +2084,14 @@ class TestSyncNodes:
         src_1.run.return_value = [row]
         src_2 = MagicMock()
         src_2.run.return_value = []
-        sink = MagicMock()
+        sink = MagicMock(sync_batch_size=2)
 
-        with (
-            patch(
-                "tasks.jobs.attack_paths.sync.graph_database.get_session",
-                side_effect=[
-                    _make_session_ctx(src_1),
-                    _make_session_ctx(src_2),
-                ],
-            ),
-            patch("tasks.jobs.attack_paths.sync.SYNC_BATCH_SIZE", 2),
+        with patch(
+            "tasks.jobs.attack_paths.sync.graph_database.get_session",
+            side_effect=[
+                _make_session_ctx(src_1),
+                _make_session_ctx(src_2),
+            ],
         ):
             result = sync_module.sync_nodes(
                 "src", "tgt", "t-1", "p-1", sink, normalized_lists
@@ -2037,7 +2109,7 @@ class TestSyncNodes:
     def test_sync_nodes_empty_source_returns_zero(self):
         src = MagicMock()
         src.run.return_value = []
-        sink = MagicMock()
+        sink = MagicMock(sync_batch_size=1000)
 
         with patch(
             "tasks.jobs.attack_paths.sync.graph_database.get_session",
@@ -2066,7 +2138,7 @@ class TestSyncRelationships:
         src_1.run.return_value = [row]
         src_2 = MagicMock()
         src_2.run.return_value = []
-        sink = MagicMock()
+        sink = MagicMock(sync_batch_size=1000)
         sink.write_relationships.side_effect = lambda *_a, **_kw: call_order.append(
             "sink:write"
         )
@@ -2104,18 +2176,15 @@ class TestSyncRelationships:
         src_2.run.return_value = [row_b]
         src_3 = MagicMock()
         src_3.run.return_value = []
-        sink = MagicMock()
+        sink = MagicMock(sync_batch_size=1)
 
-        with (
-            patch(
-                "tasks.jobs.attack_paths.sync.graph_database.get_session",
-                side_effect=[
-                    _make_session_ctx(src_1),
-                    _make_session_ctx(src_2),
-                    _make_session_ctx(src_3),
-                ],
-            ),
-            patch("tasks.jobs.attack_paths.sync.SYNC_BATCH_SIZE", 1),
+        with patch(
+            "tasks.jobs.attack_paths.sync.graph_database.get_session",
+            side_effect=[
+                _make_session_ctx(src_1),
+                _make_session_ctx(src_2),
+                _make_session_ctx(src_3),
+            ],
         ):
             total = sync_module.sync_relationships("src", "tgt", "p-1", sink)
 
@@ -2140,17 +2209,14 @@ class TestSyncRelationships:
         src_1.run.return_value = rows
         src_2 = MagicMock()
         src_2.run.return_value = []
-        sink = MagicMock()
+        sink = MagicMock(sync_batch_size=2)
 
-        with (
-            patch(
-                "tasks.jobs.attack_paths.sync.graph_database.get_session",
-                side_effect=[
-                    _make_session_ctx(src_1),
-                    _make_session_ctx(src_2),
-                ],
-            ),
-            patch("tasks.jobs.attack_paths.sync.SYNC_BATCH_SIZE", 2),
+        with patch(
+            "tasks.jobs.attack_paths.sync.graph_database.get_session",
+            side_effect=[
+                _make_session_ctx(src_1),
+                _make_session_ctx(src_2),
+            ],
         ):
             total = sync_module.sync_relationships("src", "tgt", "p-1", sink)
 
@@ -2163,7 +2229,7 @@ class TestSyncRelationships:
     def test_sync_relationships_empty_source_returns_zero(self):
         src = MagicMock()
         src.run.return_value = []
-        sink = MagicMock()
+        sink = MagicMock(sync_batch_size=1000)
 
         with patch(
             "tasks.jobs.attack_paths.sync.graph_database.get_session",
@@ -3056,6 +3122,61 @@ class TestCleanupStaleAttackPathsScans:
         ap_scan.refresh_from_db()
         assert ap_scan.state == StateChoices.FAILED
 
+    @pytest.mark.parametrize(
+        ("age_seconds", "should_clean"),
+        [
+            (960 * 60 - 1, False),
+            (960 * 60, False),
+            (960 * 60 + 1, True),
+        ],
+    )
+    @patch("tasks.jobs.attack_paths.cleanup.recover_graph_data_ready")
+    @patch("tasks.jobs.attack_paths.cleanup.graph_database.drop_database")
+    @patch(
+        "tasks.jobs.attack_paths.cleanup.rls_transaction",
+        new=lambda *args, **kwargs: nullcontext(),
+    )
+    @patch("tasks.jobs.attack_paths.cleanup._revoke_task")
+    @patch("tasks.jobs.attack_paths.cleanup._ping_workers")
+    def test_stale_threshold_boundary_is_strict(
+        self,
+        mock_ping,
+        mock_revoke,
+        mock_drop_db,
+        mock_recover,
+        age_seconds,
+        should_clean,
+        tenants_fixture,
+        aws_provider,
+    ):
+        from tasks.jobs.attack_paths.cleanup import cleanup_stale_attack_paths_scans
+
+        now = datetime.now(tz=UTC)
+        ap_scan, task_result = self._create_executing_scan(
+            tenants_fixture[0],
+            aws_provider,
+            started_at=now - timedelta(seconds=age_seconds),
+            worker="live-worker@host",
+        )
+        mock_ping.return_value = ({"live-worker@host"}, set())
+
+        with patch("tasks.jobs.attack_paths.cleanup.datetime") as mock_datetime:
+            mock_datetime.now.return_value = now
+            result = cleanup_stale_attack_paths_scans()
+
+        assert result["cleaned_up_count"] == int(should_clean)
+        ap_scan.refresh_from_db()
+        expected_state = StateChoices.FAILED if should_clean else StateChoices.EXECUTING
+        assert ap_scan.state == expected_state
+        if should_clean:
+            mock_revoke.assert_called_once_with(task_result, terminate=True)
+            mock_drop_db.assert_called_once()
+            mock_recover.assert_called_once()
+        else:
+            mock_revoke.assert_not_called()
+            mock_drop_db.assert_not_called()
+            mock_recover.assert_not_called()
+
     @patch("tasks.jobs.attack_paths.cleanup.recover_graph_data_ready")
     @patch("tasks.jobs.attack_paths.cleanup.graph_database.drop_database")
     @patch(