fix(api): correct writer transaction docs and scope regression test

Docstring corrected to writer-only scope: the findings loop and rendering
still run inside one long read-replica transaction by design, not zero
transactions as previously stated.

Add a task-level regression test asserting no writer transaction is held
during provider authentication, compression, or upload, and that the final
scan output_location update is scoped to its own rls_transaction.

Materialize the scan summary queryset with the scan relation joined,
avoiding redundant queries.

Amend the API skill doc to document the long-running-task exception to the
@set_tenant decorator rule.

api/changelog.d/scan-report-writer-transaction.fixed.md

@@ -0,0 +1 @@
+Scan report generation no longer holds a database transaction open while it renders and uploads files, which could block schema changes for the length of the report

api/src/backend/tasks/tasks.py

@@ -738,7 +738,6 @@ class ScanReportRLSTask(RLSTask):
     name="scan-report",
     queue="scan-reports",
 )
-@set_tenant(keep_tenant=True)
 @handle_provider_deletion
 def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
     """
@@ -751,6 +750,15 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
     batch is being processed. Finally, the output files are compressed and
     uploaded to S3.
 
+    Rendering and uploading take tens of minutes, so every writer access is scoped
+    to its own short `rls_transaction` and the task holds no writer transaction
+    across that work. A task-wide writer transaction would hold ACCESS SHARE on the
+    tables it read for the whole run, and any DDL waiting on one of those tables
+    would queue every later reader behind itself. The findings loop and the report
+    rendering still run inside one long read-replica transaction, which is
+    deliberate: it keeps a consistent snapshot across batches, and locks taken there
+    do not block writers on the primary.
+
     Args:
         tenant_id (str): The tenant identifier.
         scan_id (str): The scan identifier.
@@ -769,15 +777,21 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
             error,
         )
 
-    # Check if the scan has findings
-    if not ScanSummary.objects.filter(scan_id=scan_id).exists():
-        logger.info(f"No findings found for scan {scan_id}")
-        return {"upload": False}
+    with rls_transaction(tenant_id):
+        # Check if the scan has findings
+        if not ScanSummary.objects.filter(scan_id=scan_id).exists():
+            logger.info(f"No findings found for scan {scan_id}")
+            return {"upload": False}
 
-    provider_obj = Provider.objects.get(id=provider_id)
+        # `secret` is selected eagerly because initialize_prowler_provider reads it
+        # but runs outside this transaction, where RLS would resolve it to nothing.
+        provider_obj = Provider.objects.select_related("secret").get(id=provider_id)
+        provider_uid = provider_obj.uid
+        provider_type = provider_obj.provider
+
+    # Kept outside the transaction: provider initialization authenticates against
+    # the cloud provider, and that network call must not hold the providers lock.
     prowler_provider = initialize_prowler_provider(provider_obj)
-    provider_uid = provider_obj.uid
-    provider_type = provider_obj.provider
 
     # Per-framework exporters in `COMPLIANCE_CLASS_MAP` consume the legacy bulk.
     frameworks_bulk = Compliance.get_bulk(provider_type)
@@ -818,19 +832,24 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
     universal_base_dir = os.path.dirname(out_dir)
     universal_output_filename = os.path.basename(out_dir)
 
-    scan_summary = FindingOutput._transform_findings_stats(
-        ScanSummary.objects.filter(scan_id=scan_id)
-    )
-
-    # Check if we need to generate ASFF output for AWS providers with SecurityHub integration
-    generate_asff = False
-    if provider_type == "aws":
-        security_hub_integrations = Integration.objects.filter(
-            integrationproviderrelationship__provider_id=provider_id,
-            integration_type=Integration.IntegrationChoices.AWS_SECURITY_HUB,
-            enabled=True,
+    with rls_transaction(tenant_id):
+        # `_transform_findings_stats` indexes the rows and then iterates them, which
+        # on a lazy queryset runs the query twice, and it reads `scan` off the first
+        # row, which costs a third. Materializing once with the relation joined in
+        # answers all three from a single query.
+        scan_summary = FindingOutput._transform_findings_stats(
+            list(ScanSummary.objects.select_related("scan").filter(scan_id=scan_id))
         )
-        generate_asff = security_hub_integrations.exists()
+
+        # Check if we need to generate ASFF output for AWS providers with SecurityHub integration
+        generate_asff = False
+        if provider_type == "aws":
+            security_hub_integrations = Integration.objects.filter(
+                integrationproviderrelationship__provider_id=provider_id,
+                integration_type=Integration.IntegrationChoices.AWS_SECURITY_HUB,
+                enabled=True,
+            )
+            generate_asff = security_hub_integrations.exists()
 
     qs = (
         Finding.all_objects.filter(tenant_id=tenant_id, scan_id=scan_id)
@@ -943,10 +962,15 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
 
     # S3 integrations (need output_directory)
     with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
-        s3_integrations = Integration.objects.filter(
-            integrationproviderrelationship__provider_id=provider_id,
-            integration_type=Integration.IntegrationChoices.AMAZON_S3,
-            enabled=True,
+        # Materialized inside the transaction: a lazy queryset would run its query
+        # on the next access, once the tenant context is gone and RLS filters
+        # everything out.
+        s3_integrations = list(
+            Integration.objects.filter(
+                integrationproviderrelationship__provider_id=provider_id,
+                integration_type=Integration.IntegrationChoices.AMAZON_S3,
+                enabled=True,
+            )
         )
 
     if s3_integrations:
@@ -977,7 +1001,8 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
     else:
         final_location, did_upload = compressed, False
 
-    Scan.all_objects.filter(id=scan_id).update(output_location=final_location)
+    with rls_transaction(tenant_id):
+        Scan.all_objects.filter(id=scan_id).update(output_location=final_location)
     logger.info(f"Scan outputs at {final_location}")
 
     return {

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

@@ -0,0 +1,257 @@
+import uuid
+from unittest.mock import MagicMock, patch
+
+import pytest
+from api.db_utils import rls_transaction
+from api.models import (
+    Integration,
+    IntegrationProviderRelationship,
+    Provider,
+    ProviderSecret,
+)
+from django.db import DataError, connections
+from tasks.tasks import generate_outputs_task
+
+
+@pytest.fixture
+def enforce_rls():
+    """
+    Run the assertions under the unprivileged `test` role, on a fresh session.
+
+    The suite normally connects as `prowler_admin`, a superuser, and superusers
+    bypass row-level security even on FORCE ROW LEVEL SECURITY tables. Under that
+    role a query with no tenant context still returns rows, so nothing here would
+    fail. `test` is the role the RLS policies are actually written for.
+
+    The reconnect matters: a session that has already run an `rls_transaction`
+    reports the tenant variable as "" rather than NULL, which sends the policy down
+    a different branch. Starting clean keeps each test independent of suite order.
+    """
+    connections["default"].close()
+    with connections["default"].cursor() as cursor:
+        cursor.execute("SET ROLE test;")
+    yield
+    with connections["default"].cursor() as cursor:
+        cursor.execute("RESET ROLE;")
+    connections["default"].close()
+
+
+@pytest.fixture
+def provider_with_secret(tenants_fixture):
+    tenant = tenants_fixture[0]
+    provider = Provider.objects.create(
+        tenant_id=tenant.id,
+        provider=Provider.ProviderChoices.AWS.value,
+        uid="123456789012",
+        alias="rls-scoping",
+    )
+    ProviderSecret.objects.create(
+        tenant_id=tenant.id,
+        provider=provider,
+        secret_type=ProviderSecret.TypeChoices.STATIC,
+        secret={"key": "value"},
+        name=provider.alias,
+    )
+    integration = Integration.objects.create(
+        tenant_id=tenant.id,
+        enabled=True,
+        connected=True,
+        integration_type=Integration.IntegrationChoices.AMAZON_S3,
+        configuration={"key": "value"},
+        credentials={"psswd": "1234"},
+    )
+    IntegrationProviderRelationship.objects.create(
+        tenant_id=tenant.id,
+        integration=integration,
+        provider=provider,
+    )
+    return str(tenant.id), str(provider.id)
+
+
+@pytest.mark.django_db(transaction=True)
+class TestGenerateOutputsTenantScoping:
+    """
+    Pins the assumptions that let `generate_outputs_task` drop `@set_tenant`: every
+    query it runs must sit inside an `rls_transaction`, and anything it reads after
+    one closes must already be in memory.
+
+    `transaction=True` is required, not incidental. Plain `django_db` wraps the test
+    in its own atomic block, which demotes each `rls_transaction` to a savepoint and
+    leaves the transaction-scoped tenant variable set for the rest of the test. Every
+    assertion about a query running *outside* tenant context would then pass for the
+    wrong reason.
+    """
+
+    def test_rls_denies_reads_without_tenant_context(
+        self, provider_with_secret, enforce_rls
+    ):
+        # enforce_rls reconnects as the unprivileged `test` role for the whole
+        # test; pytest injects it by parameter name, so it is referenced
+        # explicitly to keep static analysers from flagging it as unused.
+        del enforce_rls
+        # Guard for the tests below: proves the `test` role really is subject to
+        # RLS, otherwise passing assertions here would mean nothing.
+        _, provider_id = provider_with_secret
+
+        assert not Provider.objects.filter(id=provider_id).exists()
+
+    def test_rls_allows_reads_inside_an_rls_transaction(
+        self, provider_with_secret, enforce_rls
+    ):
+        del enforce_rls
+        tenant_id, provider_id = provider_with_secret
+
+        with rls_transaction(tenant_id):
+            assert Provider.objects.filter(id=provider_id).exists()
+
+    def test_select_related_secret_survives_the_transaction(
+        self, provider_with_secret, enforce_rls
+    ):
+        del enforce_rls
+        tenant_id, provider_id = provider_with_secret
+
+        with rls_transaction(tenant_id):
+            provider = Provider.objects.select_related("secret").get(id=provider_id)
+
+        # initialize_prowler_provider reads `secret` out here, with no tenant
+        # context. It only works because select_related already cached it.
+        assert provider.secret.secret == {"key": "value"}
+
+    def test_secret_without_select_related_is_lost_after_the_transaction(
+        self, provider_with_secret, enforce_rls
+    ):
+        del enforce_rls
+        tenant_id, provider_id = provider_with_secret
+
+        with rls_transaction(tenant_id):
+            provider = Provider.objects.get(id=provider_id)
+
+        # The regression select_related guards against: a lazy relation resolved
+        # after the transaction never reaches the row.
+        with pytest.raises(DataError):
+            provider.secret
+
+    def test_lazy_queryset_does_not_reach_the_rows_after_the_transaction(
+        self, provider_with_secret, enforce_rls
+    ):
+        del enforce_rls
+        tenant_id, provider_id = provider_with_secret
+
+        with rls_transaction(tenant_id):
+            lazy = Integration.objects.filter(
+                integrationproviderrelationship__provider_id=provider_id,
+                integration_type=Integration.IntegrationChoices.AMAZON_S3,
+                enabled=True,
+            )
+            materialized = list(
+                Integration.objects.filter(
+                    integrationproviderrelationship__provider_id=provider_id,
+                    integration_type=Integration.IntegrationChoices.AMAZON_S3,
+                    enabled=True,
+                )
+            )
+
+        # The lazy queryset only runs its query here, and by then the tenant
+        # variable is gone. Once a transaction has set and discarded it,
+        # current_setting returns "" rather than NULL, so the policy's
+        # ""::uuid cast errors instead of taking the NULL branch that yields no
+        # rows (see test_rls_denies_reads_without_tenant_context). Wrong either
+        # way, which is why the S3 lookup is materialized with list().
+        assert len(materialized) == 1
+        with pytest.raises(DataError):
+            list(lazy)
+
+
+@pytest.mark.django_db(transaction=True)
+class TestGenerateOutputsTransactionScope:
+    """
+    Runs the real task and checks where its transactions begin and end.
+
+    `set_tenant` wraps the function it decorates in `transaction.atomic`, so
+    re-adding it to `generate_outputs_task` puts provider authentication, rendering,
+    compression and upload — tens of minutes of work — inside one writer
+    transaction. That is the stall this task was changed to avoid, and it is what
+    the `in_atomic_block` probes below detect.
+
+    `transaction=True` is required: plain `django_db` runs the test inside its own
+    atomic block, which would make every probe report True and the test pass or fail
+    for reasons unrelated to the task.
+    """
+
+    def test_slow_phases_hold_no_transaction_and_writes_stay_scoped(self):
+        scan_id = str(uuid.uuid4())
+        provider_id = str(uuid.uuid4())
+        tenant_id = str(uuid.uuid4())
+
+        # READ_REPLICA_ALIAS is None under the test settings, so the task's replica
+        # transactions also open on `default`; one connection sees every phase.
+        connection = connections["default"]
+        probes = {}
+
+        def probe(name, result=None):
+            def record(*_args, **_kwargs):
+                probes[name] = connection.in_atomic_block
+                return result
+
+            return record
+
+        provider_obj = MagicMock(uid="provider-uid", provider="aws")
+
+        integrations = MagicMock()
+        integrations.exists.return_value = False
+        integrations.__iter__.return_value = iter([])
+
+        scan_update = MagicMock()
+        scan_update.return_value.update.side_effect = probe("scan_output_location")
+
+        with (
+            patch("tasks.tasks._cleanup_stale_tmp_output_directories"),
+            patch("tasks.tasks.ScanSummary.objects.filter") as scan_summary_filter,
+            patch("tasks.tasks.ScanSummary.objects.select_related"),
+            patch("tasks.tasks.Provider.objects.select_related") as provider_select,
+            patch("tasks.tasks.Integration.objects.filter", return_value=integrations),
+            patch("tasks.tasks.Finding.all_objects.filter") as finding_filter,
+            patch(
+                "tasks.tasks.initialize_prowler_provider",
+                side_effect=probe("provider_authentication", MagicMock()),
+            ),
+            patch("tasks.tasks.Compliance.get_bulk", return_value={}),
+            patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
+            patch("tasks.tasks.get_compliance_frameworks", return_value=[]),
+            patch("tasks.tasks.FindingOutput._transform_findings_stats"),
+            patch("tasks.tasks.OUTPUT_FORMATS_MAPPING", {}),
+            patch(
+                "tasks.tasks._generate_output_directory",
+                return_value=("/tmp/test/out-dir", "/tmp/test/comp-dir"),
+            ),
+            patch(
+                "tasks.tasks._compress_output_files",
+                side_effect=probe("compression", "/tmp/zipped.zip"),
+            ),
+            patch(
+                "tasks.tasks._upload_to_s3",
+                side_effect=probe("upload", "s3://bucket/zipped.zip"),
+            ),
+            patch("tasks.tasks.Scan.all_objects.filter", scan_update),
+            patch("tasks.tasks.rmtree"),
+        ):
+            scan_summary_filter.return_value.exists.return_value = True
+            provider_select.return_value.get.return_value = provider_obj
+            finding_filter.return_value.order_by.return_value.iterator.return_value = []
+
+            result = generate_outputs_task(
+                scan_id=scan_id,
+                provider_id=provider_id,
+                tenant_id=tenant_id,
+            )
+
+        assert result == {"upload": True}
+
+        # The phases that made the original transaction long-lived.
+        assert probes["provider_authentication"] is False
+        assert probes["compression"] is False
+        assert probes["upload"] is False
+
+        # The counterpart: dropping the task-wide transaction must not leave the
+        # writer accesses bare, or they would run with no tenant context at all.
+        assert probes["scan_output_location"] is True

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

@@ -290,14 +290,14 @@ class TestGenerateOutputs:
     @patch("tasks.tasks.get_compliance_frameworks")
     @patch("tasks.tasks.Compliance.get_bulk")
     @patch("tasks.tasks.initialize_prowler_provider")
-    @patch("tasks.tasks.Provider.objects.get")
+    @patch("tasks.tasks.Provider.objects.select_related")
     @patch("tasks.tasks.ScanSummary.objects.filter")
     @patch("tasks.tasks.Finding.all_objects.filter")
     def test_generate_outputs_happy_path(
         self,
         mock_finding_filter,
         mock_scan_summary_filter,
-        mock_provider_get,
+        mock_provider_select_related,
         mock_initialize_provider,
         mock_compliance_get_bulk,
         mock_get_available_frameworks,
@@ -309,7 +309,7 @@ class TestGenerateOutputs:
         mock_provider = MagicMock()
         mock_provider.uid = "provider-uid"
         mock_provider.provider = "aws"
-        mock_provider_get.return_value = mock_provider
+        mock_provider_select_related.return_value.get.return_value = mock_provider
 
         prowler_provider = MagicMock()
         mock_initialize_provider.return_value = prowler_provider
@@ -375,7 +375,8 @@ class TestGenerateOutputs:
     def test_generate_outputs_fails_upload(self):
         with (
             patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter,
-            patch("tasks.tasks.Provider.objects.get"),
+            patch("tasks.tasks.ScanSummary.objects.select_related"),
+            patch("tasks.tasks.Provider.objects.select_related"),
             patch("tasks.tasks.initialize_prowler_provider"),
             patch("tasks.tasks.Compliance.get_bulk"),
             patch("tasks.tasks.get_compliance_frameworks"),
@@ -446,7 +447,10 @@ class TestGenerateOutputs:
         with (
             patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
             patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter,
-            patch("tasks.tasks.Provider.objects.get", return_value=mock_provider),
+            patch(
+                "tasks.tasks.Provider.objects.select_related",
+                **{"return_value.get.return_value": mock_provider},
+            ),
             patch("tasks.tasks.initialize_prowler_provider"),
             patch("tasks.tasks.Compliance.get_bulk", return_value={"cis": MagicMock()}),
             patch("tasks.tasks.get_compliance_frameworks", return_value=["cis"]),
@@ -523,7 +527,7 @@ class TestGenerateOutputs:
 
         with (
             patch("tasks.tasks.ScanSummary.objects.filter") as mock_summary,
-            patch("tasks.tasks.Provider.objects.get"),
+            patch("tasks.tasks.Provider.objects.select_related"),
             patch("tasks.tasks.initialize_prowler_provider"),
             patch("tasks.tasks.Compliance.get_bulk"),
             patch("tasks.tasks.get_compliance_frameworks", return_value=[]),
@@ -603,8 +607,12 @@ class TestGenerateOutputs:
             patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
             patch("tasks.tasks.ScanSummary.objects.filter") as mock_summary,
             patch(
-                "tasks.tasks.Provider.objects.get",
-                return_value=MagicMock(uid="UID", provider="aws"),
+                "tasks.tasks.Provider.objects.select_related",
+                **{
+                    "return_value.get.return_value": MagicMock(
+                        uid="UID", provider="aws"
+                    )
+                },
             ),
             patch("tasks.tasks.initialize_prowler_provider"),
             patch(
@@ -677,7 +685,10 @@ class TestGenerateOutputs:
         with (
             patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
             patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter,
-            patch("tasks.tasks.Provider.objects.get", return_value=mock_provider),
+            patch(
+                "tasks.tasks.Provider.objects.select_related",
+                **{"return_value.get.return_value": mock_provider},
+            ),
             patch("tasks.tasks.initialize_prowler_provider"),
             patch("tasks.tasks.Compliance.get_bulk", return_value={"cis": MagicMock()}),
             patch("tasks.tasks.get_compliance_frameworks", return_value=["cis"]),
@@ -735,7 +746,7 @@ class TestGenerateOutputs:
         """Test that generate_outputs_task only processes enabled S3 integrations."""
         with (
             patch("tasks.tasks.ScanSummary.objects.filter") as mock_summary,
-            patch("tasks.tasks.Provider.objects.get"),
+            patch("tasks.tasks.Provider.objects.select_related"),
             patch("tasks.tasks.initialize_prowler_provider"),
             patch("tasks.tasks.Compliance.get_bulk"),
             patch("tasks.tasks.get_compliance_frameworks", return_value=[]),
@@ -1123,7 +1134,7 @@ class TestCheckIntegrationsTask:
     @patch("tasks.tasks.s3_integration_task")
     @patch("tasks.tasks.Integration.objects.filter")
     @patch("tasks.tasks.ScanSummary.objects.filter")
-    @patch("tasks.tasks.Provider.objects.get")
+    @patch("tasks.tasks.Provider.objects.select_related")
     @patch("tasks.tasks.initialize_prowler_provider")
     @patch("tasks.tasks.Compliance.get_bulk")
     @patch("tasks.tasks.get_compliance_frameworks")
@@ -1148,7 +1159,7 @@ class TestCheckIntegrationsTask:
         mock_get_frameworks,
         mock_compliance_bulk,
         mock_initialize_provider,
-        mock_provider_get,
+        mock_provider_select_related,
         mock_scan_summary,
         mock_integration_filter,
         mock_s3_task,
@@ -1164,7 +1175,7 @@ class TestCheckIntegrationsTask:
         mock_provider = MagicMock()
         mock_provider.uid = "aws-account-123"
         mock_provider.provider = "aws"
-        mock_provider_get.return_value = mock_provider
+        mock_provider_select_related.return_value.get.return_value = mock_provider
 
         # Mock SecurityHub integration exists
         mock_security_hub_integrations = MagicMock()
@@ -1253,7 +1264,7 @@ class TestCheckIntegrationsTask:
     @patch("tasks.tasks.s3_integration_task")
     @patch("tasks.tasks.Integration.objects.filter")
     @patch("tasks.tasks.ScanSummary.objects.filter")
-    @patch("tasks.tasks.Provider.objects.get")
+    @patch("tasks.tasks.Provider.objects.select_related")
     @patch("tasks.tasks.initialize_prowler_provider")
     @patch("tasks.tasks.Compliance.get_bulk")
     @patch("tasks.tasks.get_compliance_frameworks")
@@ -1278,7 +1289,7 @@ class TestCheckIntegrationsTask:
         mock_get_frameworks,
         mock_compliance_bulk,
         mock_initialize_provider,
-        mock_provider_get,
+        mock_provider_select_related,
         mock_scan_summary,
         mock_integration_filter,
         mock_s3_task,
@@ -1294,7 +1305,7 @@ class TestCheckIntegrationsTask:
         mock_provider = MagicMock()
         mock_provider.uid = "aws-account-123"
         mock_provider.provider = "aws"
-        mock_provider_get.return_value = mock_provider
+        mock_provider_select_related.return_value.get.return_value = mock_provider
 
         # Mock NO SecurityHub integration
         mock_security_hub_integrations = MagicMock()
@@ -1378,7 +1389,7 @@ class TestCheckIntegrationsTask:
 
     @patch("tasks.tasks.get_prowler_provider_compliance", return_value={})
     @patch("tasks.tasks.ScanSummary.objects.filter")
-    @patch("tasks.tasks.Provider.objects.get")
+    @patch("tasks.tasks.Provider.objects.select_related")
     @patch("tasks.tasks.initialize_prowler_provider")
     @patch("tasks.tasks.Compliance.get_bulk")
     @patch("tasks.tasks.get_compliance_frameworks")
@@ -1403,7 +1414,7 @@ class TestCheckIntegrationsTask:
         mock_get_frameworks,
         mock_compliance_bulk,
         mock_initialize_provider,
-        mock_provider_get,
+        mock_provider_select_related,
         mock_scan_summary,
         mock_get_prowler_compliance,
     ):
@@ -1417,7 +1428,7 @@ class TestCheckIntegrationsTask:
         mock_provider = MagicMock()
         mock_provider.uid = "azure-subscription-123"
         mock_provider.provider = "azure"  # Non-AWS provider
-        mock_provider_get.return_value = mock_provider
+        mock_provider_select_related.return_value.get.return_value = mock_provider
 
         # Mock other necessary components
         mock_initialize_provider.return_value = MagicMock()

skills/prowler-api/SKILL.md

@@ -29,7 +29,7 @@ For **generic DRF patterns** (ViewSets, Serializers, Filters, JSON:API), use `dj
 
 - ALWAYS use `rls_transaction(tenant_id)` when querying outside ViewSet context
 - ALWAYS use `get_role()` before checking permissions (returns FIRST role only)
-- ALWAYS use `@set_tenant` then `@handle_provider_deletion` decorator order
+- ALWAYS use `@set_tenant` then `@handle_provider_deletion` decorator order, EXCEPT in long-running tasks (see note below)
 - ALWAYS use explicit through models for M2M relationships (required for RLS)
 - NEVER access `Provider.objects` without RLS context in Celery tasks
 - NEVER bypass RLS by using raw SQL or `connection.cursor()`
@@ -37,6 +37,8 @@ For **generic DRF patterns** (ViewSets, Serializers, Filters, JSON:API), use `dj
 
 > **Note**: `rls_transaction()` accepts both UUID objects and strings - it converts internally via `str(value)`.
 
+> **Note**: `@set_tenant` is the default for Celery tasks, but it wraps the whole task in `transaction.atomic`. A task that keeps running after it is done with the writer - rendering, compression, uploads - must NOT use it, or it holds ACCESS SHARE on every table it read until the task ends, and any DDL waiting on one of those tables queues every later reader behind itself. Such tasks scope each writer access to its own short `rls_transaction(tenant_id)` instead, which is the first rule above applied strictly. See `generate_outputs_task` in `api/src/backend/tasks/tasks.py`. Anything read inside one of those transactions and used after it closes must be materialized first (`list()`, `select_related()`), since a lazy queryset or relation resolves with no tenant context.
+
 ---
 
 ## Architecture Overview