perf(attack-paths): cleanup task prioritization, restore default batch sizes to 1000, upgrade Cartography to 0.135.0 (#10729)

api/src/backend/api/migrations/0090_attack_paths_cleanup_priority.py

@@ -0,0 +1,23 @@
+from django.db import migrations
+
+TASK_NAME = "attack-paths-cleanup-stale-scans"
+
+
+def set_cleanup_priority(apps, schema_editor):
+    PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+    PeriodicTask.objects.filter(name=TASK_NAME).update(priority=0)
+
+
+def unset_cleanup_priority(apps, schema_editor):
+    PeriodicTask = apps.get_model("django_celery_beat", "PeriodicTask")
+    PeriodicTask.objects.filter(name=TASK_NAME).update(priority=None)
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0089_backfill_finding_group_status_muted"),
+    ]
+
+    operations = [
+        migrations.RunPython(set_cleanup_priority, unset_cleanup_priority),
+    ]

api/src/backend/config/celery.py

@@ -17,8 +17,10 @@ celery_app.config_from_object("django.conf:settings", namespace="CELERY")
 celery_app.conf.update(result_extended=True, result_expires=None)
 
 celery_app.conf.broker_transport_options = {
-    "visibility_timeout": BROKER_VISIBILITY_TIMEOUT
+    "visibility_timeout": BROKER_VISIBILITY_TIMEOUT,
+    "queue_order_strategy": "priority",
 }
+celery_app.conf.task_default_priority = 6
 celery_app.conf.result_backend_transport_options = {
     "visibility_timeout": BROKER_VISIBILITY_TIMEOUT
 }

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

@@ -1,6 +1,8 @@
 # Portions of this file are based on code from the Cartography project
 # (https://github.com/cartography-cncf/cartography), which is licensed under the Apache 2.0 License.
 
+import time
+
 from typing import Any
 
 import aioboto3
@@ -33,7 +35,7 @@ def start_aws_ingestion(
 
     For the scan progress updates:
         - The caller of this function (`tasks.jobs.attack_paths.scan.run`) has set it to 2.
-        - When the control returns to the caller, it will be set to 95.
+        - When the control returns to the caller, it will be set to 93.
     """
 
     # Initialize variables common to all jobs
@@ -89,34 +91,50 @@ def start_aws_ingestion(
         logger.info(
             f"Syncing function permission_relationships for AWS account {prowler_api_provider.uid}"
         )
+        t0 = time.perf_counter()
         cartography_aws.RESOURCE_FUNCTIONS["permission_relationships"](**sync_args)
+        logger.info(
+            f"Synced function permission_relationships for AWS account {prowler_api_provider.uid} in {time.perf_counter() - t0:.3f}s"
+        )
     db_utils.update_attack_paths_scan_progress(attack_paths_scan, 88)
 
     if "resourcegroupstaggingapi" in requested_syncs:
         logger.info(
             f"Syncing function resourcegroupstaggingapi for AWS account {prowler_api_provider.uid}"
         )
+        t0 = time.perf_counter()
         cartography_aws.RESOURCE_FUNCTIONS["resourcegroupstaggingapi"](**sync_args)
+        logger.info(
+            f"Synced function resourcegroupstaggingapi for AWS account {prowler_api_provider.uid} in {time.perf_counter() - t0:.3f}s"
+        )
     db_utils.update_attack_paths_scan_progress(attack_paths_scan, 89)
 
     logger.info(
         f"Syncing ec2_iaminstanceprofile scoped analysis for AWS account {prowler_api_provider.uid}"
     )
+    t0 = time.perf_counter()
     cartography_aws.run_scoped_analysis_job(
         "aws_ec2_iaminstanceprofile.json",
         neo4j_session,
         common_job_parameters,
     )
+    logger.info(
+        f"Synced ec2_iaminstanceprofile scoped analysis for AWS account {prowler_api_provider.uid} in {time.perf_counter() - t0:.3f}s"
+    )
     db_utils.update_attack_paths_scan_progress(attack_paths_scan, 90)
 
     logger.info(
         f"Syncing lambda_ecr analysis for AWS account {prowler_api_provider.uid}"
     )
+    t0 = time.perf_counter()
     cartography_aws.run_analysis_job(
         "aws_lambda_ecr.json",
         neo4j_session,
         common_job_parameters,
     )
+    logger.info(
+        f"Synced lambda_ecr analysis for AWS account {prowler_api_provider.uid} in {time.perf_counter() - t0:.3f}s"
+    )
 
     if all(
         s in requested_syncs
@@ -125,25 +143,34 @@ def start_aws_ingestion(
         logger.info(
             f"Syncing lb_container_exposure scoped analysis for AWS account {prowler_api_provider.uid}"
         )
+        t0 = time.perf_counter()
         cartography_aws.run_scoped_analysis_job(
             "aws_lb_container_exposure.json",
             neo4j_session,
             common_job_parameters,
         )
+        logger.info(
+            f"Synced lb_container_exposure scoped analysis for AWS account {prowler_api_provider.uid} in {time.perf_counter() - t0:.3f}s"
+        )
 
     if all(s in requested_syncs for s in ["ec2:network_acls", "ec2:load_balancer_v2"]):
         logger.info(
             f"Syncing lb_nacl_direct scoped analysis for AWS account {prowler_api_provider.uid}"
         )
+        t0 = time.perf_counter()
         cartography_aws.run_scoped_analysis_job(
             "aws_lb_nacl_direct.json",
             neo4j_session,
             common_job_parameters,
         )
+        logger.info(
+            f"Synced lb_nacl_direct scoped analysis for AWS account {prowler_api_provider.uid} in {time.perf_counter() - t0:.3f}s"
+        )
 
     db_utils.update_attack_paths_scan_progress(attack_paths_scan, 91)
 
     logger.info(f"Syncing metadata for AWS account {prowler_api_provider.uid}")
+    t0 = time.perf_counter()
     cartography_aws.merge_module_sync_metadata(
         neo4j_session,
         group_type="AWSAccount",
@@ -152,24 +179,23 @@ def start_aws_ingestion(
         update_tag=cartography_config.update_tag,
         stat_handler=cartography_aws.stat_handler,
     )
+    logger.info(
+        f"Synced metadata for AWS account {prowler_api_provider.uid} in {time.perf_counter() - t0:.3f}s"
+    )
     db_utils.update_attack_paths_scan_progress(attack_paths_scan, 92)
 
     # Removing the added extra field
     del common_job_parameters["AWS_ID"]
 
-    logger.info(f"Syncing cleanup_job for AWS account {prowler_api_provider.uid}")
-    cartography_aws.run_cleanup_job(
-        "aws_post_ingestion_principals_cleanup.json",
-        neo4j_session,
-        common_job_parameters,
-    )
-    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 93)
-
     logger.info(f"Syncing analysis for AWS account {prowler_api_provider.uid}")
+    t0 = time.perf_counter()
     cartography_aws._perform_aws_analysis(
         requested_syncs, neo4j_session, common_job_parameters
     )
-    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 94)
+    logger.info(
+        f"Synced analysis for AWS account {prowler_api_provider.uid} in {time.perf_counter() - t0:.3f}s"
+    )
+    db_utils.update_attack_paths_scan_progress(attack_paths_scan, 93)
 
     return failed_syncs
 
@@ -234,6 +260,8 @@ def sync_aws_account(
             )
 
             try:
+                func_t0 = time.perf_counter()
+
                 # `ecr:image_layers` uses `aioboto3_session` instead of `boto3_session`
                 if func_name == "ecr:image_layers":
                     cartography_aws.RESOURCE_FUNCTIONS[func_name](
@@ -257,7 +285,15 @@ def sync_aws_account(
                 else:
                     cartography_aws.RESOURCE_FUNCTIONS[func_name](**sync_args)
 
+                logger.info(
+                    f"Synced function {func_name} for AWS account {prowler_api_provider.uid} in {time.perf_counter() - func_t0:.3f}s"
+                )
+
             except Exception as e:
+                logger.info(
+                    f"Synced function {func_name} for AWS account {prowler_api_provider.uid} in {time.perf_counter() - func_t0:.3f}s (FAILED)"
+                )
+
                 exception_message = utils.stringify_exception(
                     e, f"Exception for AWS sync function: {func_name}"
                 )

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

@@ -8,9 +8,9 @@ from tasks.jobs.attack_paths import aws
 # Batch size for Neo4j write operations (resource labeling, cleanup)
 BATCH_SIZE = env.int("ATTACK_PATHS_BATCH_SIZE", 1000)
 # Batch size for Postgres findings fetch (keyset pagination page size)
-FINDINGS_BATCH_SIZE = env.int("ATTACK_PATHS_FINDINGS_BATCH_SIZE", 500)
+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", 250)
+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

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

@@ -12,6 +12,7 @@ from typing import Any, Generator
 from uuid import UUID
 
 import neo4j
+
 from cartography.config import Config as CartographyConfig
 from celery.utils.log import get_task_logger
 from tasks.jobs.attack_paths.config import (
@@ -86,17 +87,21 @@ def analysis(
     prowler_api_provider: Provider,
     scan_id: str,
     config: CartographyConfig,
-) -> None:
+) -> tuple[int, int]:
     """
     Main entry point for Prowler findings analysis.
 
     Adds resource labels and loads findings.
+    Returns (labeled_nodes, findings_loaded).
     """
-    add_resource_label(
+    total_labeled = add_resource_label(
         neo4j_session, prowler_api_provider.provider, str(prowler_api_provider.uid)
     )
     findings_data = stream_findings_with_resources(prowler_api_provider, scan_id)
-    load_findings(neo4j_session, findings_data, prowler_api_provider, config)
+    total_loaded = load_findings(
+        neo4j_session, findings_data, prowler_api_provider, config
+    )
+    return total_labeled, total_loaded
 
 
 def add_resource_label(
@@ -146,12 +151,11 @@ def load_findings(
     findings_batches: Generator[list[dict[str, Any]], None, None],
     prowler_api_provider: Provider,
     config: CartographyConfig,
-) -> None:
+) -> int:
     """Load Prowler findings into the graph, linking them to resources."""
     query = render_cypher_template(
         INSERT_FINDING_TEMPLATE,
         {
-            "__ROOT_NODE_LABEL__": get_root_node_label(prowler_api_provider.provider),
             "__NODE_UID_FIELD__": get_node_uid_field(prowler_api_provider.provider),
             "__RESOURCE_LABEL__": get_provider_resource_label(
                 prowler_api_provider.provider
@@ -160,7 +164,6 @@ def load_findings(
     )
 
     parameters = {
-        "provider_uid": str(prowler_api_provider.uid),
         "last_updated": config.update_tag,
         "prowler_version": ProwlerConfig.prowler_version,
     }
@@ -178,6 +181,7 @@ def load_findings(
         neo4j_session.run(query, parameters)
 
     logger.info(f"Finished loading {total_records} records in {batch_num} batches")
+    return total_records
 
 
 # Findings Streaming (Generator-based)

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

@@ -32,17 +32,14 @@ ADD_RESOURCE_LABEL_TEMPLATE = """
 """
 
 INSERT_FINDING_TEMPLATE = f"""
-    MATCH (account:__ROOT_NODE_LABEL__ {{id: $provider_uid}})
     UNWIND $findings_data AS finding_data
 
-    OPTIONAL MATCH (account)-->(resource_by_uid:__RESOURCE_LABEL__)
-        WHERE resource_by_uid.__NODE_UID_FIELD__ = finding_data.resource_uid
-    WITH account, finding_data, resource_by_uid
+    OPTIONAL MATCH (resource_by_uid:__RESOURCE_LABEL__ {{__NODE_UID_FIELD__: finding_data.resource_uid}})
+    WITH finding_data, resource_by_uid
 
-    OPTIONAL MATCH (account)-->(resource_by_id:__RESOURCE_LABEL__)
+    OPTIONAL MATCH (resource_by_id:__RESOURCE_LABEL__ {{id: finding_data.resource_uid}})
         WHERE resource_by_uid IS NULL
-            AND resource_by_id.id = finding_data.resource_uid
-    WITH account, finding_data, COALESCE(resource_by_uid, resource_by_id) AS resource
+    WITH finding_data, COALESCE(resource_by_uid, resource_by_id) AS resource
         WHERE resource IS NOT NULL
 
     MERGE (finding:{PROWLER_FINDING_LABEL} {{id: finding_data.id}})

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

@@ -55,6 +55,7 @@ exception propagates to Celery.
 
 import logging
 import time
+
 from typing import Any
 
 from cartography.config import Config as CartographyConfig
@@ -144,6 +145,12 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
         attack_paths_scan, task_id, tenant_cartography_config
     )
 
+    scan_t0 = time.perf_counter()
+    logger.info(
+        f"Starting Attack Paths scan ({attack_paths_scan.id}) for "
+        f"{prowler_api_provider.provider.upper()} provider {prowler_api_provider.id}"
+    )
+
     subgraph_dropped = False
     sync_completed = False
     provider_gated = False
@@ -169,6 +176,7 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
             db_utils.update_attack_paths_scan_progress(attack_paths_scan, 2)
 
             # The real scan, where iterates over cloud services
+            t0 = time.perf_counter()
             ingestion_exceptions = utils.call_within_event_loop(
                 cartography_ingestion_function,
                 tmp_neo4j_session,
@@ -177,19 +185,23 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
                 prowler_sdk_provider,
                 attack_paths_scan,
             )
+            logger.info(
+                f"Cartography ingestion completed in {time.perf_counter() - t0:.3f}s "
+                f"(failed_syncs={len(ingestion_exceptions)})"
+            )
 
             # Post-processing: Just keeping it to be more Cartography compliant
             logger.info(
                 f"Syncing Cartography ontology for AWS account {prowler_api_provider.uid}"
             )
             cartography_ontology.run(tmp_neo4j_session, tmp_cartography_config)
-            db_utils.update_attack_paths_scan_progress(attack_paths_scan, 95)
+            db_utils.update_attack_paths_scan_progress(attack_paths_scan, 94)
 
             logger.info(
                 f"Syncing Cartography analysis for AWS account {prowler_api_provider.uid}"
             )
             cartography_analysis.run(tmp_neo4j_session, tmp_cartography_config)
-            db_utils.update_attack_paths_scan_progress(attack_paths_scan, 96)
+            db_utils.update_attack_paths_scan_progress(attack_paths_scan, 95)
 
             # Creating Internet node and CAN_ACCESS relationships
             logger.info(
@@ -198,14 +210,20 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
             internet.analysis(
                 tmp_neo4j_session, prowler_api_provider, tmp_cartography_config
             )
+            db_utils.update_attack_paths_scan_progress(attack_paths_scan, 96)
 
             # Adding Prowler Finding nodes and relationships
             logger.info(
                 f"Syncing Prowler analysis for AWS account {prowler_api_provider.uid}"
             )
-            findings.analysis(
+            t0 = time.perf_counter()
+            labeled_nodes, findings_loaded = findings.analysis(
                 tmp_neo4j_session, prowler_api_provider, scan_id, tmp_cartography_config
             )
+            logger.info(
+                f"Prowler analysis completed in {time.perf_counter() - t0:.3f}s "
+                f"(findings={findings_loaded}, labeled_nodes={labeled_nodes})"
+            )
             db_utils.update_attack_paths_scan_progress(attack_paths_scan, 97)
 
         logger.info(
@@ -227,22 +245,33 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
         logger.info(f"Deleting existing provider graph in {tenant_database_name}")
         db_utils.set_provider_graph_data_ready(attack_paths_scan, False)
         provider_gated = True
-        graph_database.drop_subgraph(
+
+        t0 = time.perf_counter()
+        deleted_nodes = graph_database.drop_subgraph(
             database=tenant_database_name,
             provider_id=str(prowler_api_provider.id),
         )
+        logger.info(
+            f"Deleted existing provider graph in {time.perf_counter() - t0:.3f}s "
+            f"(deleted_nodes={deleted_nodes})"
+        )
         subgraph_dropped = True
         db_utils.update_attack_paths_scan_progress(attack_paths_scan, 98)
 
         logger.info(
             f"Syncing graph from {tmp_database_name} into {tenant_database_name}"
         )
-        sync.sync_graph(
+        t0 = time.perf_counter()
+        sync_result = sync.sync_graph(
             source_database=tmp_database_name,
             target_database=tenant_database_name,
             tenant_id=str(prowler_api_provider.tenant_id),
             provider_id=str(prowler_api_provider.id),
         )
+        logger.info(
+            f"Synced graph in {time.perf_counter() - t0:.3f}s "
+            f"(nodes={sync_result['nodes']}, relationships={sync_result['relationships']})"
+        )
         sync_completed = True
         db_utils.set_graph_data_ready(attack_paths_scan, True)
         db_utils.update_attack_paths_scan_progress(attack_paths_scan, 99)
@@ -250,17 +279,16 @@ def run(tenant_id: str, scan_id: str, task_id: str) -> dict[str, Any]:
         logger.info(f"Clearing Neo4j cache for database {tenant_database_name}")
         graph_database.clear_cache(tenant_database_name)
 
-        logger.info(
-            f"Completed Cartography ({attack_paths_scan.id}) for "
-            f"{prowler_api_provider.provider.upper()} provider {prowler_api_provider.id}"
-        )
-
         logger.info(f"Dropping temporary Neo4j database {tmp_database_name}")
         graph_database.drop_database(tmp_database_name)
 
         db_utils.finish_attack_paths_scan(
             attack_paths_scan, StateChoices.COMPLETED, ingestion_exceptions
         )
+        logger.info(
+            f"Attack Paths scan completed in {time.perf_counter() - scan_t0:.3f}s "
+            f"(state=completed, failed_syncs={len(ingestion_exceptions)})"
+        )
         return ingestion_exceptions
 
     except Exception as e:

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

@@ -5,6 +5,8 @@ This module handles syncing graph data from temporary scan databases
 to the tenant database, adding provider isolation labels and properties.
 """
 
+import time
+
 from collections import defaultdict
 from typing import Any
 
@@ -81,6 +83,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.
     """
+    t0 = time.perf_counter()
     last_id = -1
     total_synced = 0
 
@@ -117,7 +120,7 @@ def sync_nodes(
 
         total_synced += batch_count
         logger.info(
-            f"Synced {total_synced} nodes from {source_database} to {target_database}"
+            f"Synced {total_synced} nodes from {source_database} to {target_database} in {time.perf_counter() - t0:.3f}s"
         )
 
     return total_synced
@@ -136,6 +139,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.
     """
+    t0 = time.perf_counter()
     last_id = -1
     total_synced = 0
 
@@ -166,7 +170,7 @@ def sync_relationships(
 
         total_synced += batch_count
         logger.info(
-            f"Synced {total_synced} relationships from {source_database} to {target_database}"
+            f"Synced {total_synced} relationships from {source_database} to {target_database} in {time.perf_counter() - t0:.3f}s"
         )
 
     return total_synced

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

@@ -38,11 +38,14 @@ class TestAttackPathsRun:
     @patch("tasks.jobs.attack_paths.scan.db_utils.finish_attack_paths_scan")
     @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
     @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
-    @patch("tasks.jobs.attack_paths.scan.sync.sync_graph")
-    @patch("tasks.jobs.attack_paths.scan.graph_database.drop_subgraph")
+    @patch(
+        "tasks.jobs.attack_paths.scan.sync.sync_graph",
+        return_value={"nodes": 0, "relationships": 0},
+    )
+    @patch("tasks.jobs.attack_paths.scan.graph_database.drop_subgraph", return_value=0)
     @patch("tasks.jobs.attack_paths.scan.indexes.create_sync_indexes")
     @patch("tasks.jobs.attack_paths.scan.internet.analysis")
-    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis", return_value=(0, 0))
     @patch("tasks.jobs.attack_paths.scan.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
@@ -188,7 +191,7 @@ class TestAttackPathsRun:
     @patch("tasks.jobs.attack_paths.scan.db_utils.set_provider_graph_data_ready")
     @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
     @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
-    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis", return_value=(0, 0))
     @patch("tasks.jobs.attack_paths.scan.internet.analysis")
     @patch("tasks.jobs.attack_paths.scan.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
@@ -287,7 +290,7 @@ class TestAttackPathsRun:
     @patch("tasks.jobs.attack_paths.scan.db_utils.set_provider_graph_data_ready")
     @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
     @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
-    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis", return_value=(0, 0))
     @patch("tasks.jobs.attack_paths.scan.internet.analysis")
     @patch("tasks.jobs.attack_paths.scan.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
@@ -390,7 +393,7 @@ class TestAttackPathsRun:
     @patch("tasks.jobs.attack_paths.scan.db_utils.set_provider_graph_data_ready")
     @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
     @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
-    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis", return_value=(0, 0))
     @patch("tasks.jobs.attack_paths.scan.internet.analysis")
     @patch("tasks.jobs.attack_paths.scan.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
@@ -489,14 +492,17 @@ class TestAttackPathsRun:
     @patch("tasks.jobs.attack_paths.scan.db_utils.set_provider_graph_data_ready")
     @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
     @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
-    @patch("tasks.jobs.attack_paths.scan.sync.sync_graph")
+    @patch(
+        "tasks.jobs.attack_paths.scan.sync.sync_graph",
+        return_value={"nodes": 0, "relationships": 0},
+    )
     @patch(
         "tasks.jobs.attack_paths.scan.graph_database.drop_subgraph",
         side_effect=RuntimeError("drop failed"),
     )
     @patch("tasks.jobs.attack_paths.scan.indexes.create_sync_indexes")
     @patch("tasks.jobs.attack_paths.scan.internet.analysis")
-    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis", return_value=(0, 0))
     @patch("tasks.jobs.attack_paths.scan.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
@@ -609,7 +615,7 @@ class TestAttackPathsRun:
     @patch("tasks.jobs.attack_paths.scan.graph_database.drop_subgraph")
     @patch("tasks.jobs.attack_paths.scan.indexes.create_sync_indexes")
     @patch("tasks.jobs.attack_paths.scan.internet.analysis")
-    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis", return_value=(0, 0))
     @patch("tasks.jobs.attack_paths.scan.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
@@ -718,11 +724,14 @@ class TestAttackPathsRun:
     @patch("tasks.jobs.attack_paths.scan.db_utils.set_provider_graph_data_ready")
     @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
     @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
-    @patch("tasks.jobs.attack_paths.scan.sync.sync_graph")
+    @patch(
+        "tasks.jobs.attack_paths.scan.sync.sync_graph",
+        return_value={"nodes": 0, "relationships": 0},
+    )
     @patch("tasks.jobs.attack_paths.scan.graph_database.drop_subgraph")
     @patch("tasks.jobs.attack_paths.scan.indexes.create_sync_indexes")
     @patch("tasks.jobs.attack_paths.scan.internet.analysis")
-    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis", return_value=(0, 0))
     @patch("tasks.jobs.attack_paths.scan.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
@@ -833,14 +842,17 @@ class TestAttackPathsRun:
     @patch("tasks.jobs.attack_paths.scan.db_utils.set_provider_graph_data_ready")
     @patch("tasks.jobs.attack_paths.scan.db_utils.update_attack_paths_scan_progress")
     @patch("tasks.jobs.attack_paths.scan.db_utils.starting_attack_paths_scan")
-    @patch("tasks.jobs.attack_paths.scan.sync.sync_graph")
+    @patch(
+        "tasks.jobs.attack_paths.scan.sync.sync_graph",
+        return_value={"nodes": 0, "relationships": 0},
+    )
     @patch(
         "tasks.jobs.attack_paths.scan.graph_database.drop_subgraph",
         side_effect=RuntimeError("drop failed"),
     )
     @patch("tasks.jobs.attack_paths.scan.indexes.create_sync_indexes")
     @patch("tasks.jobs.attack_paths.scan.internet.analysis")
-    @patch("tasks.jobs.attack_paths.scan.findings.analysis")
+    @patch("tasks.jobs.attack_paths.scan.findings.analysis", return_value=(0, 0))
     @patch("tasks.jobs.attack_paths.scan.indexes.create_findings_indexes")
     @patch("tasks.jobs.attack_paths.scan.cartography_ontology.run")
     @patch("tasks.jobs.attack_paths.scan.cartography_analysis.run")
@@ -1274,10 +1286,6 @@ class TestAttackPathsFindingsHelpers:
         mock_session = MagicMock()
 
         with (
-            patch(
-                "tasks.jobs.attack_paths.findings.get_root_node_label",
-                return_value="AWSAccount",
-            ),
             patch(
                 "tasks.jobs.attack_paths.findings.get_node_uid_field",
                 return_value="arn",
@@ -1294,7 +1302,6 @@ class TestAttackPathsFindingsHelpers:
         assert mock_session.run.call_count == 2
         for call_args in mock_session.run.call_args_list:
             params = call_args.args[1]
-            assert params["provider_uid"] == str(provider.uid)
             assert params["last_updated"] == config.update_tag
             assert "findings_data" in params
 
@@ -1673,10 +1680,6 @@ class TestAttackPathsFindingsHelpers:
             yield  # Make it a generator
 
         with (
-            patch(
-                "tasks.jobs.attack_paths.findings.get_root_node_label",
-                return_value="AWSAccount",
-            ),
             patch(
                 "tasks.jobs.attack_paths.findings.get_node_uid_field",
                 return_value="arn",