chore: fix ruff

api/CHANGELOG.md

@@ -2,6 +2,14 @@
 
 All notable changes to the **Prowler API** are documented in this file.
 
+## [1.34.1] (Prowler UNRELEASED)
+
+### 🐞 Fixed
+
+- Partitioned `compliance_requirements_overviews` and hardened its COPY ingest (batched, single reused connection, first-run DELETE skipped) to stop the database writer from running out of memory during compliance overview ingestion [(#11870)](https://github.com/prowler-cloud/prowler/pull/11870)
+
+---
+
 ## [1.34.0] (Prowler v5.33.0)
 
 ### 🚀 Added

api/src/backend/api/migrations/0098_partition_compliance_requirement_overviews.py

@@ -0,0 +1,209 @@
+"""Convert ``compliance_requirements_overviews`` into a RANGE-partitioned table.
+
+The compliance-overview ingest rewrites every scan's rows (delete-then-reinsert),
+which — together with old scans ageing out — generates heavy autovacuum churn on a
+plain table. Partitioning RANGE by the UUIDv7 ``id`` (same strategy as ``findings``)
+lets old data be reclaimed with ``DROP PARTITION`` instead of ``DELETE`` + CASCADE.
+
+Existing data is preserved WITHOUT a copy: the current table is renamed and
+re-attached as the ``default`` partition of a new partitioned parent (ATTACH is a
+metadata operation, so there is no row rewrite). Existing rows keep their uuid4
+ids and live in ``default``; new rows carry uuid7 ids and route to the monthly
+partitions created by ``pgpartition``.
+
+NOTE for production: rename + ATTACH take a brief ACCESS EXCLUSIVE lock and adding
+the parent FKs validates the default partition. Run during a low-traffic window (or
+``--fake`` + apply manually) on large tables, and rehearse against a prod snapshot —
+this is delicate DDL with no in-repo precedent. See skills/django-migration-psql.
+"""
+
+import api.rls
+import psqlextra.manager.manager
+import uuid6
+from api.rls import RowLevelSecurityConstraint
+from django.db import migrations, models
+
+TABLE = "compliance_requirements_overviews"
+DEFAULT = f"{TABLE}_default"
+INDEX = "cro_scan_comp_reg_idx"
+UNIQUE = "unique_tenant_compliance_requirement_overview"
+PARENT_RLS = "rls_on_compliancerequirementoverview"
+DEFAULT_RLS = "rls_on_compliancerequirementoverview_default"
+
+
+def _partition_table(apps, schema_editor):
+    model = apps.get_model("api", "ComplianceRequirementOverview")
+    cursor = schema_editor.connection.cursor
+
+    # 1. Drop the old RLS (created in 0027 with SELECT/INSERT/UPDATE/DELETE) via the
+    #    constraint object so the DB_USER-scoped policy/grant names resolve correctly.
+    schema_editor.remove_constraint(
+        model,
+        RowLevelSecurityConstraint(
+            "tenant_id",
+            name=PARENT_RLS,
+            statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
+        ),
+    )
+
+    with cursor() as cur:
+        # 2. Drop the old business unique (5 cols) — after partitioning, dedup is
+        #    handled by the delete-then-reinsert ingest (the model keeps no unique).
+        cur.execute(f"ALTER TABLE {TABLE} DROP CONSTRAINT {UNIQUE};")
+
+        # 3. Rename the existing table and every object whose name the new parent
+        #    will reuse, moving them under the ``_default`` name. The default keeps
+        #    its own FKs, so the parent FKs added below skip re-validation on ATTACH.
+        cur.execute(f"ALTER TABLE {TABLE} RENAME TO {DEFAULT};")
+        cur.execute(
+            f"ALTER TABLE {DEFAULT} RENAME CONSTRAINT {TABLE}_pkey TO {DEFAULT}_pkey;"
+        )
+        cur.execute(f"ALTER INDEX {INDEX} RENAME TO {INDEX}_default;")
+
+        # 4. Create the partitioned parent and re-establish its structure.
+        cur.execute(
+            f"CREATE TABLE {TABLE} (LIKE {DEFAULT} INCLUDING DEFAULTS) "
+            "PARTITION BY RANGE (id);"
+        )
+        cur.execute(
+            f"ALTER TABLE {TABLE} ADD CONSTRAINT {TABLE}_pkey PRIMARY KEY (id);"
+        )
+        # Parent index ON ONLY (no data), then attach the pre-built default index so
+        # the existing rows are NOT re-indexed.
+        cur.execute(
+            f"CREATE INDEX {INDEX} ON ONLY {TABLE} "
+            "(tenant_id, scan_id, compliance_id, region);"
+        )
+
+        # 5. Add FKs on the STILL-EMPTY parent (instant, no scan) so new partitions
+        #    inherit referential integrity. Postgres cannot add NOT VALID FKs to a
+        #    partitioned table, but because the default partition already carries an
+        #    equivalent valid FK, the ATTACH below skips FK re-validation. Cascade on
+        #    scan/tenant deletion is still handled at the ORM level.
+        cur.execute(
+            f"ALTER TABLE {TABLE} ADD CONSTRAINT {TABLE}_scan_id_fk "
+            "FOREIGN KEY (scan_id) REFERENCES scans (id) DEFERRABLE INITIALLY DEFERRED;"
+        )
+        cur.execute(
+            f"ALTER TABLE {TABLE} ADD CONSTRAINT {TABLE}_tenant_id_fk "
+            "FOREIGN KEY (tenant_id) REFERENCES tenants (id) DEFERRABLE INITIALLY DEFERRED;"
+        )
+
+        # 6. Attach the existing table (with all its rows) as the DEFAULT partition
+        #    and attach its pre-built index to the parent's partitioned index.
+        cur.execute(f"ALTER TABLE {TABLE} ATTACH PARTITION {DEFAULT} DEFAULT;")
+        cur.execute(f"ALTER INDEX {INDEX} ATTACH PARTITION {INDEX}_default;")
+
+    # 7. Re-establish RLS on the parent and the default partition (DB_USER-aware).
+    schema_editor.add_constraint(
+        model,
+        RowLevelSecurityConstraint(
+            "tenant_id", name=PARENT_RLS, statements=["SELECT", "INSERT", "DELETE"]
+        ),
+    )
+    schema_editor.add_constraint(
+        model,
+        RowLevelSecurityConstraint(
+            "tenant_id",
+            name=DEFAULT_RLS,
+            partition_name="default",
+            statements=["SELECT", "INSERT", "DELETE"],
+        ),
+    )
+
+
+def _unpartition_table(apps, schema_editor):
+    """Best-effort reverse: detach the default partition and restore a plain table."""
+    model = apps.get_model("api", "ComplianceRequirementOverview")
+    cursor = schema_editor.connection.cursor
+
+    schema_editor.remove_constraint(
+        model,
+        RowLevelSecurityConstraint(
+            "tenant_id",
+            name=DEFAULT_RLS,
+            partition_name="default",
+            statements=["SELECT", "INSERT", "DELETE"],
+        ),
+    )
+    schema_editor.remove_constraint(
+        model,
+        RowLevelSecurityConstraint(
+            "tenant_id", name=PARENT_RLS, statements=["SELECT", "INSERT", "DELETE"]
+        ),
+    )
+
+    with cursor() as cur:
+        # Detach the default partition (keeps its own pkey/index/FKs) and drop the
+        # partitioned parent, then rename the default back to the original table.
+        cur.execute(f"ALTER INDEX {INDEX} DETACH PARTITION {INDEX}_default;")
+        cur.execute(f"ALTER TABLE {TABLE} DETACH PARTITION {DEFAULT};")
+        cur.execute(f"DROP TABLE {TABLE};")
+        cur.execute(f"ALTER TABLE {DEFAULT} RENAME TO {TABLE};")
+        cur.execute(
+            f"ALTER TABLE {TABLE} RENAME CONSTRAINT {DEFAULT}_pkey TO {TABLE}_pkey;"
+        )
+        cur.execute(f"ALTER INDEX {INDEX}_default RENAME TO {INDEX};")
+        # Restore the original 5-column business unique (the FKs survived on the
+        # detached table, so they are not re-created here).
+        cur.execute(
+            f"ALTER TABLE {TABLE} ADD CONSTRAINT {UNIQUE} "
+            "UNIQUE (tenant_id, scan_id, compliance_id, requirement_id, region);"
+        )
+
+    schema_editor.add_constraint(
+        model,
+        RowLevelSecurityConstraint(
+            "tenant_id",
+            name=PARENT_RLS,
+            statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
+        ),
+    )
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0097_attack_paths_scan_db_defaults"),
+    ]
+
+    operations = [
+        migrations.SeparateDatabaseAndState(
+            # State: mirror what pgmakemigrations detected (minus unrelated stowaways)
+            # so the ORM state matches the partitioned model definition.
+            state_operations=[
+                migrations.AlterModelOptions(
+                    name="compliancerequirementoverview",
+                    options={"base_manager_name": "objects"},
+                ),
+                migrations.AlterModelManagers(
+                    name="compliancerequirementoverview",
+                    managers=[
+                        ("objects", psqlextra.manager.manager.PostgresManager()),
+                    ],
+                ),
+                migrations.RemoveConstraint(
+                    model_name="compliancerequirementoverview",
+                    name=UNIQUE,
+                ),
+                migrations.AlterField(
+                    model_name="compliancerequirementoverview",
+                    name="id",
+                    field=models.UUIDField(
+                        default=uuid6.uuid7,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                migrations.AddConstraint(
+                    model_name="compliancerequirementoverview",
+                    constraint=api.rls.RowLevelSecurityConstraint(
+                        "tenant_id", name=DEFAULT_RLS
+                    ),
+                ),
+            ],
+            database_operations=[
+                migrations.RunPython(_partition_table, _unpartition_table),
+            ],
+        ),
+    ]

api/src/backend/api/models.py

@@ -1587,8 +1587,25 @@ class ComplianceOverview(RowLevelSecurityProtectedModel):
         resource_name = "compliance-overviews"
 
 
-class ComplianceRequirementOverview(RowLevelSecurityProtectedModel):
-    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
+class ComplianceRequirementOverview(
+    PostgresPartitionedModel, RowLevelSecurityProtectedModel
+):
+    """
+    Per-requirement compliance rows, materialized once per scan via COPY.
+
+    Partitioned RANGE by the UUIDv7 ``id`` (same strategy as ``Finding``) so that
+    ageing scans' compliance rows can be reclaimed with ``DROP PARTITION`` instead
+    of ``DELETE`` + CASCADE, removing the autovacuum churn that a per-scan delete
+    otherwise generates.
+
+    Note when creating migrations you must use ``python manage.py pgmakemigrations``.
+    """
+
+    class PartitioningMeta:
+        method = PostgresPartitioningMethod.RANGE
+        key = ["id"]
+
+    id = models.UUIDField(primary_key=True, default=uuid7, editable=False)
     inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
     compliance_id = models.TextField(blank=False)
     framework = models.TextField(blank=False)
@@ -1613,23 +1630,25 @@ class ComplianceRequirementOverview(RowLevelSecurityProtectedModel):
 
     class Meta(RowLevelSecurityProtectedModel.Meta):
         db_table = "compliance_requirements_overviews"
+        base_manager_name = "objects"
 
+        # No UniqueConstraint on (tenant, scan, compliance, requirement, region):
+        # Postgres would force the partition key ``id`` into it, making it trivial
+        # (``id`` is the PK). Business-level dedup is guaranteed by the
+        # delete-then-reinsert ingest in ``create_compliance_requirements`` instead
+        # (same approach as the partitioned ``Finding`` model, which has none).
         constraints = [
-            models.UniqueConstraint(
-                fields=(
-                    "tenant_id",
-                    "scan_id",
-                    "compliance_id",
-                    "requirement_id",
-                    "region",
-                ),
-                name="unique_tenant_compliance_requirement_overview",
-            ),
             RowLevelSecurityConstraint(
                 field="tenant_id",
                 name="rls_on_%(class)s",
                 statements=["SELECT", "INSERT", "DELETE"],
             ),
+            RowLevelSecurityConstraint(
+                field="tenant_id",
+                name="rls_on_%(class)s_default",
+                partition_name="default",
+                statements=["SELECT", "INSERT", "DELETE"],
+            ),
         ]
         indexes = [
             models.Index(

api/src/backend/api/partitions.py

@@ -1,7 +1,7 @@
 from collections.abc import Generator
 from datetime import UTC, datetime
 
-from api.models import Finding, ResourceFindingMapping
+from api.models import ComplianceRequirementOverview, Finding, ResourceFindingMapping
 from api.rls import RowLevelSecurityConstraint
 from api.uuid_utils import datetime_to_uuid7
 from dateutil.relativedelta import relativedelta
@@ -196,5 +196,22 @@ manager = PostgresPartitioningManager(
                 rls_statements=["SELECT"],
             ),
         ),
+        # ComplianceRequirementOverview: partition by its own UUIDv7 id so old
+        # scans' compliance rows age out via DROP PARTITION instead of DELETE.
+        PostgresPartitioningConfig(
+            model=ComplianceRequirementOverview,
+            strategy=PostgresUUIDv7PartitioningStrategy(
+                start_date=datetime.now(UTC),
+                size=PostgresTimePartitionSize(
+                    months=settings.COMPLIANCE_REQ_OVERVIEW_PARTITION_MONTHS
+                ),
+                count=settings.COMPLIANCE_REQ_OVERVIEW_PARTITION_COUNT,
+                max_age=relative_days_or_none(
+                    settings.COMPLIANCE_REQ_OVERVIEW_PARTITION_MAX_AGE_MONTHS
+                ),
+                name_format="%Y_%b",
+                rls_statements=["SELECT", "INSERT", "DELETE"],
+            ),
+        ),
     ]
 )

api/src/backend/config/settings/partitions.py

@@ -14,3 +14,17 @@ FINDINGS_TABLE_PARTITION_COUNT = env.int("FINDINGS_TABLE_PARTITION_COUNT", 7)
 FINDINGS_TABLE_PARTITION_MAX_AGE_MONTHS = env.int(
     "FINDINGS_TABLE_PARTITION_MAX_AGE_MONTHS", None
 )
+
+# Compliance requirement overviews partitioning (RANGE by UUIDv7 id).
+# Mirrors the findings settings; kept separate so the ingest-heavy compliance
+# table can be tuned (and aged out) independently from findings.
+COMPLIANCE_REQ_OVERVIEW_PARTITION_MONTHS = env.int(
+    "COMPLIANCE_REQ_OVERVIEW_PARTITION_MONTHS", 1
+)
+COMPLIANCE_REQ_OVERVIEW_PARTITION_COUNT = env.int(
+    "COMPLIANCE_REQ_OVERVIEW_PARTITION_COUNT", 7
+)
+# Setting this to None keeps partitions indefinitely (matches findings default).
+COMPLIANCE_REQ_OVERVIEW_PARTITION_MAX_AGE_MONTHS = env.int(
+    "COMPLIANCE_REQ_OVERVIEW_PARTITION_MAX_AGE_MONTHS", None
+)

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

@@ -4,9 +4,9 @@ import json
 import random
 import re
 import time
-import uuid
 from collections import defaultdict
 from collections.abc import Iterable
+from contextlib import ExitStack
 from datetime import UTC, datetime
 from typing import Any
 
@@ -71,6 +71,7 @@ from tasks.jobs.queries import (
     COMPLIANCE_UPSERT_TENANT_SUMMARY_SQL,
 )
 from tasks.utils import CustomEncoder, batched
+from uuid6 import uuid7
 
 logger = get_task_logger(__name__)
 
@@ -99,6 +100,17 @@ COMPLIANCE_REQUIREMENT_COPY_COLUMNS = (
 FINDINGS_MICRO_BATCH_SIZE = env.int("DJANGO_FINDINGS_MICRO_BATCH_SIZE", default=3000)
 # Controls how many rows each ORM bulk_create/bulk_update call sends to Postgres.
 SCAN_DB_BATCH_SIZE = env.int("DJANGO_SCAN_DB_BATCH_SIZE", default=1000)
+# Controls how many rows each compliance-overview COPY batch streams to Postgres.
+# Smaller batches keep each COPY (and its commit/fsync) short so the writer can
+# checkpoint and autovacuum between flushes, avoiding the FreeableMemory cliff
+# observed when a single client streams back-to-back multi-second COPY statements.
+COMPLIANCE_COPY_BATCH_SIZE = env.int("DJANGO_COMPLIANCE_COPY_BATCH_SIZE", default=5000)
+# Optional pause (seconds) inserted between compliance-overview COPY batches so the
+# writer gets breathing room for checkpoints/autovacuum under sustained ingest.
+# Default 0 preserves current behavior; ops can raise it to spread the write load.
+COMPLIANCE_COPY_THROTTLE_SECONDS = env.float(
+    "DJANGO_COMPLIANCE_COPY_THROTTLE_SECONDS", default=0.0
+)
 # Throttle scan progress persistence: minimum progress delta (fraction 0-1)
 # between two persisted progress updates.
 PROGRESS_THROTTLE_DELTA = env.float("DJANGO_SCAN_PROGRESS_THROTTLE_DELTA", default=0.01)
@@ -357,15 +369,21 @@ def _bulk_update_resource_failed_findings_counts(
 
 
 def _copy_compliance_requirement_rows(
-    tenant_id: str, rows: list[dict[str, Any]]
+    connection, tenant_id: str, rows: list[dict[str, Any]]
 ) -> None:
     """Stream compliance requirement rows into Postgres using COPY.
 
-    We leverage the admin connection (when available) to bypass the COPY + RLS
-    restriction, writing only the fields required by
-    ``ComplianceRequirementOverview``.
+    Runs on a caller-supplied admin connection (which bypasses the COPY + RLS
+    restriction), writing only the fields required by
+    ``ComplianceRequirementOverview``. Each batch is committed on its own so the
+    writer releases locks and can checkpoint between flushes; the connection
+    itself is reused across batches to avoid per-batch reconnection churn. The
+    tenant GUC is set inside every batch's transaction because
+    ``SET_CONFIG_QUERY`` uses ``is_local=TRUE`` (transaction-scoped), so it is
+    cleared on each commit.
 
     Args:
+        connection: Open admin psycopg connection (``autocommit=False``).
         tenant_id: Target tenant UUID.
         rows: List of row dictionaries prepared by
             :func:`create_compliance_requirements`.
@@ -405,34 +423,38 @@ def _copy_compliance_requirement_rows(
     )
 
     try:
-        with psycopg_connection(MainRouter.admin_db) as connection:
-            connection.autocommit = False
-            try:
-                with connection.cursor() as cursor:
-                    cursor.execute(SET_CONFIG_QUERY, [POSTGRES_TENANT_VAR, tenant_id])
-                    cursor.copy_expert(copy_sql, csv_buffer)
-                connection.commit()
-            except Exception:
-                connection.rollback()
-                raise
+        with connection.cursor() as cursor:
+            # is_local=TRUE ⇒ must be re-applied inside each batch's transaction.
+            cursor.execute(SET_CONFIG_QUERY, [POSTGRES_TENANT_VAR, tenant_id])
+            cursor.copy_expert(copy_sql, csv_buffer)
+        connection.commit()
+    except Exception:
+        connection.rollback()
+        raise
     finally:
         csv_buffer.close()
 
 
 def _persist_compliance_requirement_rows(
-    tenant_id: str, rows: Iterable[dict[str, Any]], batch_size: int = 10000
+    tenant_id: str,
+    rows: Iterable[dict[str, Any]],
+    batch_size: int = COMPLIANCE_COPY_BATCH_SIZE,
+    throttle_seconds: float = COMPLIANCE_COPY_THROTTLE_SECONDS,
 ) -> int:
     """Persist compliance requirement rows using batched COPY with ORM fallback.
 
     ``rows`` is consumed lazily in batches, so peak memory stays at ~``batch_size``
-    rows instead of the full set. A batch that fails COPY falls back to an ORM
-    ``bulk_create`` of just that batch.
+    rows instead of the full set. A single admin connection is reused across all
+    batches (each batch committed separately); an optional pause between batches
+    lets the writer checkpoint/autovacuum under sustained ingest. A batch that
+    fails COPY falls back to an ORM ``bulk_create`` of just that batch.
 
     Args:
         tenant_id: Target tenant UUID.
         rows: Iterable of row dictionaries reflecting the compliance overview
             state for a scan.
-        batch_size: Number of rows per COPY batch (default: 10000).
+        batch_size: Number of rows per COPY batch.
+        throttle_seconds: Optional pause between batches (0 disables throttling).
 
     Returns:
         int: total number of rows persisted.
@@ -440,49 +462,63 @@ def _persist_compliance_requirement_rows(
     total_rows = 0
     batch_num = 0
 
-    for batch, _is_last in batched(rows, batch_size):
-        if not batch:
-            continue
-        batch_num += 1
-        try:
-            _copy_compliance_requirement_rows(tenant_id, batch)
-        except Exception as error:
-            logger.exception(
-                f"COPY bulk insert for compliance requirements batch {batch_num} "
-                "failed; falling back to ORM bulk_create for this batch",
-                exc_info=error,
-            )
-            fallback_objects = [
-                ComplianceRequirementOverview(
-                    id=row["id"],
-                    tenant_id=row["tenant_id"],
-                    inserted_at=row["inserted_at"],
-                    compliance_id=row["compliance_id"],
-                    framework=row["framework"],
-                    version=row["version"],
-                    description=row["description"],
-                    region=row["region"],
-                    requirement_id=row["requirement_id"],
-                    requirement_status=row["requirement_status"],
-                    passed_checks=row["passed_checks"],
-                    failed_checks=row["failed_checks"],
-                    total_checks=row["total_checks"],
-                    passed_findings=row.get("passed_findings", 0),
-                    total_findings=row.get("total_findings", 0),
-                    scan_id=row["scan_id"],
-                )
-                for row in batch
-            ]
-            with rls_transaction(tenant_id):
-                ComplianceRequirementOverview.objects.bulk_create(
-                    fallback_objects, batch_size=500
+    # A single admin connection is opened lazily (only when the first non-empty
+    # batch arrives) and reused for every batch; ``ExitStack`` guarantees it is
+    # closed on exit. A COPY failure rolls back inside
+    # ``_copy_compliance_requirement_rows``, leaving the connection reusable, so
+    # the same connection is kept for the ORM-fallback path and later batches.
+    with ExitStack() as stack:
+        copy_connection = None
+        for batch, _is_last in batched(rows, batch_size):
+            if not batch:
+                continue
+            batch_num += 1
+            if throttle_seconds > 0 and batch_num > 1:
+                time.sleep(throttle_seconds)
+            try:
+                if copy_connection is None:
+                    copy_connection = stack.enter_context(
+                        psycopg_connection(MainRouter.admin_db)
+                    )
+                    copy_connection.autocommit = False
+                _copy_compliance_requirement_rows(copy_connection, tenant_id, batch)
+            except Exception as error:
+                logger.exception(
+                    f"COPY bulk insert for compliance requirements batch {batch_num} "
+                    "failed; falling back to ORM bulk_create for this batch",
+                    exc_info=error,
                 )
+                fallback_objects = [
+                    ComplianceRequirementOverview(
+                        id=row["id"],
+                        tenant_id=row["tenant_id"],
+                        inserted_at=row["inserted_at"],
+                        compliance_id=row["compliance_id"],
+                        framework=row["framework"],
+                        version=row["version"],
+                        description=row["description"],
+                        region=row["region"],
+                        requirement_id=row["requirement_id"],
+                        requirement_status=row["requirement_status"],
+                        passed_checks=row["passed_checks"],
+                        failed_checks=row["failed_checks"],
+                        total_checks=row["total_checks"],
+                        passed_findings=row.get("passed_findings", 0),
+                        total_findings=row.get("total_findings", 0),
+                        scan_id=row["scan_id"],
+                    )
+                    for row in batch
+                ]
+                with rls_transaction(tenant_id):
+                    ComplianceRequirementOverview.objects.bulk_create(
+                        fallback_objects, batch_size=500
+                    )
 
-        total_rows += len(batch)
-        logger.info(
-            f"Compliance COPY batch {batch_num}: inserted {len(batch)} rows "
-            f"({total_rows} total)"
-        )
+            total_rows += len(batch)
+            logger.info(
+                f"Compliance COPY batch {batch_num}: inserted {len(batch)} rows "
+                f"({total_rows} total)"
+            )
 
     return total_rows
 
@@ -885,15 +921,19 @@ def _process_finding_micro_batch(
                         # Denormalized resource arrays populated directly on insert
                         # (was previously a separate bulk_update; saves a CASE WHEN
                         # over thousands of rows per micro-batch).
-                        resource_regions=[resource_instance.region]
-                        if resource_instance.region
-                        else [],
-                        resource_services=[resource_instance.service]
-                        if resource_instance.service
-                        else [],
-                        resource_types=[resource_instance.type]
-                        if resource_instance.type
-                        else [],
+                        resource_regions=(
+                            [resource_instance.region]
+                            if resource_instance.region
+                            else []
+                        ),
+                        resource_services=(
+                            [resource_instance.service]
+                            if resource_instance.service
+                            else []
+                        ),
+                        resource_types=(
+                            [resource_instance.type] if resource_instance.type else []
+                        ),
                     )
                     findings_to_create.append(finding_instance)
                     resource_denormalized_data.append(
@@ -1746,7 +1786,10 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
                                 requirement_statuses[key]["pass_count"] += 1
 
                             yield {
-                                "id": uuid.uuid4(),
+                                # UUIDv7 so the COPY-inserted rows route to the
+                                # dated partitions of compliance_requirements_overviews
+                                # (uuid4 would always land in the default partition).
+                                "id": uuid7(),
                                 "tenant_id": tenant_id_str,
                                 "inserted_at": utc_datetime_now,
                                 "compliance_id": compliance_id,
@@ -1765,8 +1808,15 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
                             }
 
             # Idempotent re-run: clear this scan's rows before re-inserting.
+            # First-run scans have no prior rows, so skip the DELETE entirely to
+            # avoid an empty write transaction and needless dead-tuple churn; the
+            # ``.exists()`` probe is an index-only lookup on cro_scan_comp_reg_idx.
             with rls_transaction(tenant_id):
-                ComplianceRequirementOverview.objects.filter(scan_id=scan_id).delete()
+                existing_rows = ComplianceRequirementOverview.objects.filter(
+                    scan_id=scan_id
+                )
+                if existing_rows.exists():
+                    existing_rows.delete()
 
             requirements_created = _persist_compliance_requirement_rows(
                 tenant_id, _iter_compliance_requirement_rows()

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

@@ -2409,25 +2409,13 @@ class TestCreateComplianceRequirements:
 
 
 class TestComplianceRequirementCopy:
-    @patch("tasks.jobs.scan.psycopg_connection")
-    def test_copy_compliance_requirement_rows_streams_csv(
-        self, mock_psycopg_connection, settings
-    ):
-        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
-
+    def test_copy_compliance_requirement_rows_streams_csv(self):
         connection = MagicMock()
         cursor = MagicMock()
         cursor_context = MagicMock()
         cursor_context.__enter__.return_value = cursor
         cursor_context.__exit__.return_value = False
         connection.cursor.return_value = cursor_context
-        connection.__enter__.return_value = connection
-        connection.__exit__.return_value = False
-
-        context_manager = MagicMock()
-        context_manager.__enter__.return_value = connection
-        context_manager.__exit__.return_value = False
-        mock_psycopg_connection.return_value = context_manager
 
         captured = {}
 
@@ -2454,9 +2442,8 @@ class TestComplianceRequirementCopy:
         }
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(connection, str(row["tenant_id"]), [row])
 
-        mock_psycopg_connection.assert_called_once_with("admin")
         connection.cursor.assert_called_once()
         cursor.execute.assert_called_once()
         cursor.copy_expert.assert_called_once()
@@ -2466,6 +2453,7 @@ class TestComplianceRequirementCopy:
         assert csv_rows[0][5] == ""
         assert csv_rows[0][-1] == str(row["scan_id"])
 
+    @patch("tasks.jobs.scan.psycopg_connection")
     @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.bulk_create")
     @patch("tasks.jobs.scan.rls_transaction")
     @patch(
@@ -2473,7 +2461,7 @@ class TestComplianceRequirementCopy:
         side_effect=Exception("copy failed"),
     )
     def test_persist_compliance_requirement_rows_fallback(
-        self, mock_copy, mock_rls_transaction, mock_bulk_create
+        self, mock_copy, mock_rls_transaction, mock_bulk_create, mock_psycopg_connection
     ):
         inserted_at = datetime.now(UTC)
         row = {
@@ -2495,6 +2483,10 @@ class TestComplianceRequirementCopy:
 
         tenant_id = row["tenant_id"]
 
+        conn = MagicMock()
+        mock_psycopg_connection.return_value.__enter__.return_value = conn
+        mock_psycopg_connection.return_value.__exit__.return_value = False
+
         ctx = MagicMock()
         ctx.__enter__.return_value = None
         ctx.__exit__.return_value = False
@@ -2502,7 +2494,7 @@ class TestComplianceRequirementCopy:
 
         _persist_compliance_requirement_rows(tenant_id, [row])
 
-        mock_copy.assert_called_once_with(tenant_id, [row])
+        mock_copy.assert_called_once_with(conn, tenant_id, [row])
         mock_rls_transaction.assert_called_once_with(tenant_id)
         mock_bulk_create.assert_called_once()
 
@@ -2525,26 +2517,14 @@ class TestComplianceRequirementCopy:
         mock_rls_transaction.assert_not_called()
         mock_bulk_create.assert_not_called()
 
-    @patch("tasks.jobs.scan.psycopg_connection")
-    def test_copy_compliance_requirement_rows_multiple_rows(
-        self, mock_psycopg_connection, settings
-    ):
+    def test_copy_compliance_requirement_rows_multiple_rows(self):
         """Test COPY with multiple rows to ensure batch processing works correctly."""
-        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
-
         connection = MagicMock()
         cursor = MagicMock()
         cursor_context = MagicMock()
         cursor_context.__enter__.return_value = cursor
         cursor_context.__exit__.return_value = False
         connection.cursor.return_value = cursor_context
-        connection.__enter__.return_value = connection
-        connection.__exit__.return_value = False
-
-        context_manager = MagicMock()
-        context_manager.__enter__.return_value = connection
-        context_manager.__exit__.return_value = False
-        mock_psycopg_connection.return_value = context_manager
 
         captured = {}
 
@@ -2610,9 +2590,8 @@ class TestComplianceRequirementCopy:
         ]
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(tenant_id, rows)
+            _copy_compliance_requirement_rows(connection, tenant_id, rows)
 
-        mock_psycopg_connection.assert_called_once_with("admin")
         connection.cursor.assert_called_once()
         cursor.execute.assert_called_once()
         cursor.copy_expert.assert_called_once()
@@ -2644,26 +2623,14 @@ class TestComplianceRequirementCopy:
         assert csv_rows[2][5] == "2.0"
         assert csv_rows[2][9] == "MANUAL"
 
-    @patch("tasks.jobs.scan.psycopg_connection")
-    def test_copy_compliance_requirement_rows_null_values(
-        self, mock_psycopg_connection, settings
-    ):
+    def test_copy_compliance_requirement_rows_null_values(self):
         """Test COPY handles NULL/None values correctly in nullable fields."""
-        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
-
         connection = MagicMock()
         cursor = MagicMock()
         cursor_context = MagicMock()
         cursor_context.__enter__.return_value = cursor
         cursor_context.__exit__.return_value = False
         connection.cursor.return_value = cursor_context
-        connection.__enter__.return_value = connection
-        connection.__exit__.return_value = False
-
-        context_manager = MagicMock()
-        context_manager.__enter__.return_value = connection
-        context_manager.__exit__.return_value = False
-        mock_psycopg_connection.return_value = context_manager
 
         captured = {}
 
@@ -2691,7 +2658,7 @@ class TestComplianceRequirementCopy:
         }
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(connection, str(row["tenant_id"]), [row])
 
         csv_rows = list(csv.reader(StringIO(captured["data"])))
         assert len(csv_rows) == 1
@@ -2700,26 +2667,14 @@ class TestComplianceRequirementCopy:
         assert csv_rows[0][5] == ""  # version
         assert csv_rows[0][6] == ""  # description
 
-    @patch("tasks.jobs.scan.psycopg_connection")
-    def test_copy_compliance_requirement_rows_special_characters(
-        self, mock_psycopg_connection, settings
-    ):
+    def test_copy_compliance_requirement_rows_special_characters(self):
         """Test COPY correctly escapes special characters in CSV."""
-        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
-
         connection = MagicMock()
         cursor = MagicMock()
         cursor_context = MagicMock()
         cursor_context.__enter__.return_value = cursor
         cursor_context.__exit__.return_value = False
         connection.cursor.return_value = cursor_context
-        connection.__enter__.return_value = connection
-        connection.__exit__.return_value = False
-
-        context_manager = MagicMock()
-        context_manager.__enter__.return_value = connection
-        context_manager.__exit__.return_value = False
-        mock_psycopg_connection.return_value = context_manager
 
         captured = {}
 
@@ -2747,7 +2702,7 @@ class TestComplianceRequirementCopy:
         }
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(connection, str(row["tenant_id"]), [row])
 
         # Verify CSV was generated (csv module handles escaping automatically)
         csv_rows = list(csv.reader(StringIO(captured["data"])))
@@ -2759,26 +2714,14 @@ class TestComplianceRequirementCopy:
         assert "quotes" in csv_rows[0][6]
         assert "commas" in csv_rows[0][6]
 
-    @patch("tasks.jobs.scan.psycopg_connection")
-    def test_copy_compliance_requirement_rows_missing_inserted_at(
-        self, mock_psycopg_connection, settings
-    ):
+    def test_copy_compliance_requirement_rows_missing_inserted_at(self):
         """Test COPY uses current datetime when inserted_at is missing."""
-        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
-
         connection = MagicMock()
         cursor = MagicMock()
         cursor_context = MagicMock()
         cursor_context.__enter__.return_value = cursor
         cursor_context.__exit__.return_value = False
         connection.cursor.return_value = cursor_context
-        connection.__enter__.return_value = connection
-        connection.__exit__.return_value = False
-
-        context_manager = MagicMock()
-        context_manager.__enter__.return_value = connection
-        context_manager.__exit__.return_value = False
-        mock_psycopg_connection.return_value = context_manager
 
         captured = {}
 
@@ -2808,7 +2751,7 @@ class TestComplianceRequirementCopy:
 
         before_call = datetime.now(UTC)
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(connection, str(row["tenant_id"]), [row])
         after_call = datetime.now(UTC)
 
         csv_rows = list(csv.reader(StringIO(captured["data"])))
@@ -2819,26 +2762,14 @@ class TestComplianceRequirementCopy:
         inserted_at = datetime.fromisoformat(inserted_at_str)
         assert before_call <= inserted_at <= after_call
 
-    @patch("tasks.jobs.scan.psycopg_connection")
-    def test_copy_compliance_requirement_rows_transaction_rollback_on_copy_error(
-        self, mock_psycopg_connection, settings
-    ):
+    def test_copy_compliance_requirement_rows_transaction_rollback_on_copy_error(self):
         """Test transaction is rolled back when copy_expert fails."""
-        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
-
         connection = MagicMock()
         cursor = MagicMock()
         cursor_context = MagicMock()
         cursor_context.__enter__.return_value = cursor
         cursor_context.__exit__.return_value = False
         connection.cursor.return_value = cursor_context
-        connection.__enter__.return_value = connection
-        connection.__exit__.return_value = False
-
-        context_manager = MagicMock()
-        context_manager.__enter__.return_value = connection
-        context_manager.__exit__.return_value = False
-        mock_psycopg_connection.return_value = context_manager
 
         # Simulate copy_expert failure
         cursor.copy_expert.side_effect = Exception("COPY command failed")
@@ -2861,32 +2792,24 @@ class TestComplianceRequirementCopy:
 
         with patch.object(MainRouter, "admin_db", "admin"):
             with pytest.raises(Exception, match="COPY command failed"):
-                _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+                _copy_compliance_requirement_rows(
+                    connection, str(row["tenant_id"]), [row]
+                )
 
         # Verify rollback was called
         connection.rollback.assert_called_once()
         connection.commit.assert_not_called()
 
-    @patch("tasks.jobs.scan.psycopg_connection")
     def test_copy_compliance_requirement_rows_transaction_rollback_on_set_config_error(
-        self, mock_psycopg_connection, settings
+        self,
     ):
         """Test transaction is rolled back when SET_CONFIG fails."""
-        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
-
         connection = MagicMock()
         cursor = MagicMock()
         cursor_context = MagicMock()
         cursor_context.__enter__.return_value = cursor
         cursor_context.__exit__.return_value = False
         connection.cursor.return_value = cursor_context
-        connection.__enter__.return_value = connection
-        connection.__exit__.return_value = False
-
-        context_manager = MagicMock()
-        context_manager.__enter__.return_value = connection
-        context_manager.__exit__.return_value = False
-        mock_psycopg_connection.return_value = context_manager
 
         # Simulate cursor.execute failure
         cursor.execute.side_effect = Exception("SET prowler.tenant_id failed")
@@ -2909,32 +2832,22 @@ class TestComplianceRequirementCopy:
 
         with patch.object(MainRouter, "admin_db", "admin"):
             with pytest.raises(Exception, match="SET prowler.tenant_id failed"):
-                _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+                _copy_compliance_requirement_rows(
+                    connection, str(row["tenant_id"]), [row]
+                )
 
         # Verify rollback was called
         connection.rollback.assert_called_once()
         connection.commit.assert_not_called()
 
-    @patch("tasks.jobs.scan.psycopg_connection")
-    def test_copy_compliance_requirement_rows_commit_on_success(
-        self, mock_psycopg_connection, settings
-    ):
+    def test_copy_compliance_requirement_rows_commit_on_success(self):
         """Test transaction is committed on successful COPY."""
-        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
-
         connection = MagicMock()
         cursor = MagicMock()
         cursor_context = MagicMock()
         cursor_context.__enter__.return_value = cursor
         cursor_context.__exit__.return_value = False
         connection.cursor.return_value = cursor_context
-        connection.__enter__.return_value = connection
-        connection.__exit__.return_value = False
-
-        context_manager = MagicMock()
-        context_manager.__enter__.return_value = connection
-        context_manager.__exit__.return_value = False
-        mock_psycopg_connection.return_value = context_manager
 
         cursor.copy_expert.return_value = None  # Success
 
@@ -2955,19 +2868,24 @@ class TestComplianceRequirementCopy:
         }
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(connection, str(row["tenant_id"]), [row])
 
         # Verify commit was called and rollback was not
         connection.commit.assert_called_once()
         connection.rollback.assert_not_called()
-        # Verify autocommit was disabled
-        assert connection.autocommit is False
 
+    @patch("tasks.jobs.scan.psycopg_connection")
     @patch("tasks.jobs.scan._copy_compliance_requirement_rows")
-    def test_persist_compliance_requirement_rows_success(self, mock_copy):
+    def test_persist_compliance_requirement_rows_success(
+        self, mock_copy, mock_psycopg_connection
+    ):
         """Test successful COPY path without fallback to ORM."""
         mock_copy.return_value = None  # Success, no exception
 
+        conn = MagicMock()
+        mock_psycopg_connection.return_value.__enter__.return_value = conn
+        mock_psycopg_connection.return_value.__exit__.return_value = False
+
         tenant_id = str(uuid.uuid4())
         rows = [
             {
@@ -2991,8 +2909,9 @@ class TestComplianceRequirementCopy:
         _persist_compliance_requirement_rows(tenant_id, rows)
 
         # Verify COPY was called
-        mock_copy.assert_called_once_with(tenant_id, rows)
+        mock_copy.assert_called_once_with(conn, tenant_id, rows)
 
+    @patch("tasks.jobs.scan.psycopg_connection")
     @patch("tasks.jobs.scan.logger")
     @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.bulk_create")
     @patch("tasks.jobs.scan.rls_transaction")
@@ -3001,7 +2920,12 @@ class TestComplianceRequirementCopy:
         side_effect=Exception("COPY failed"),
     )
     def test_persist_compliance_requirement_rows_fallback_logging(
-        self, mock_copy, mock_rls_transaction, mock_bulk_create, mock_logger
+        self,
+        mock_copy,
+        mock_rls_transaction,
+        mock_bulk_create,
+        mock_logger,
+        mock_psycopg_connection,
     ):
         """Test logger.exception is called when COPY fails and fallback occurs."""
         tenant_id = str(uuid.uuid4())
@@ -3036,6 +2960,7 @@ class TestComplianceRequirementCopy:
         assert "falling back to ORM" in args[0]
         assert kwargs.get("exc_info") is not None
 
+    @patch("tasks.jobs.scan.psycopg_connection")
     @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.bulk_create")
     @patch("tasks.jobs.scan.rls_transaction")
     @patch(
@@ -3043,7 +2968,7 @@ class TestComplianceRequirementCopy:
         side_effect=Exception("copy failed"),
     )
     def test_persist_compliance_requirement_rows_fallback_multiple_rows(
-        self, mock_copy, mock_rls_transaction, mock_bulk_create
+        self, mock_copy, mock_rls_transaction, mock_bulk_create, mock_psycopg_connection
     ):
         """Test ORM fallback with multiple rows."""
         tenant_id = str(uuid.uuid4())
@@ -3085,6 +3010,10 @@ class TestComplianceRequirementCopy:
             },
         ]
 
+        conn = MagicMock()
+        mock_psycopg_connection.return_value.__enter__.return_value = conn
+        mock_psycopg_connection.return_value.__exit__.return_value = False
+
         ctx = MagicMock()
         ctx.__enter__.return_value = None
         ctx.__exit__.return_value = False
@@ -3092,7 +3021,7 @@ class TestComplianceRequirementCopy:
 
         _persist_compliance_requirement_rows(tenant_id, rows)
 
-        mock_copy.assert_called_once_with(tenant_id, rows)
+        mock_copy.assert_called_once_with(conn, tenant_id, rows)
         mock_rls_transaction.assert_called_once_with(tenant_id)
         mock_bulk_create.assert_called_once()
 
@@ -3117,6 +3046,7 @@ class TestComplianceRequirementCopy:
         assert objects[1].passed_checks == 2
         assert objects[1].failed_checks == 3
 
+    @patch("tasks.jobs.scan.psycopg_connection")
     @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.bulk_create")
     @patch("tasks.jobs.scan.rls_transaction")
     @patch(
@@ -3124,7 +3054,7 @@ class TestComplianceRequirementCopy:
         side_effect=Exception("copy failed"),
     )
     def test_persist_compliance_requirement_rows_fallback_all_fields(
-        self, mock_copy, mock_rls_transaction, mock_bulk_create
+        self, mock_copy, mock_rls_transaction, mock_bulk_create, mock_psycopg_connection
     ):
         """Test ORM fallback correctly maps all fields from row dict to model."""
         tenant_id = str(uuid.uuid4())