fix(api): make every endpoint resolve the same latest scan for a provider (#12858)

api/changelog.d/ephemeral-resource-count-reset.fixed.md

@@ -0,0 +1 @@
+Resources no longer keep a stale failed findings count forever when a scoped or imported scan for the same provider completes after a full scan, which used to make the full scan skip its own cleanup

api/changelog.d/latest-scan-null-completed-at.fixed.md

@@ -0,0 +1 @@
+Providers whose most recent completed scan has no `completed_at` timestamp are no longer missing from every endpoint that reports a provider's latest scan, which now falls back to scan creation order instead of skipping the provider

api/changelog.d/latest-scan-selector.changed.md

@@ -0,0 +1 @@
+Unify how every endpoint resolves a provider latest completed scan, so overlapping scans no longer make findings, compliance and mute rules read from different scans

api/src/backend/api/db_utils.py

@@ -409,7 +409,7 @@ def batch_delete(tenant_id, queryset, batch_size=settings.DJANGO_DELETION_BATCH_
 
     Args:
         tenant_id (str): Tenant ID the queryset belongs to.
-        queryset (QuerySet): The queryset of objects to delete.
+        queryset: The queryset of objects to delete.
         batch_size (int): The number of objects to delete in each batch.
 
     Returns:

api/src/backend/api/models.py

@@ -617,9 +617,66 @@ class Task(RowLevelSecurityProtectedModel):
         resource_name = "tasks"
 
 
+class ScanQuerySet(models.QuerySet):
+    """Shared selectors for "the latest scan of a provider".
+
+    The queryset must already be scoped by the caller: manager, tenant, RBAC,
+    providers and database alias.
+    """
+
+    # How "which completed scan is the provider's current one" is ordered.
+    LATEST_ORDER_BY = (
+        models.F("completed_at").desc(nulls_last=True),
+        models.F("inserted_at").desc(),
+        models.F("id").desc(),
+    )
+
+    def _eligible_for_latest(self) -> "ScanQuerySet":
+        """Restrict to the scans that may be a provider's latest.
+
+        Returns:
+            ScanQuerySet: The completed scans.
+        """
+        return self.filter(state=StateChoices.COMPLETED)
+
+    def latest_per_provider(self) -> "ScanQuerySet":
+        """Pick each provider's latest scan with `DISTINCT ON (provider_id)`.
+
+        Returns:
+            ScanQuerySet: One scan per provider, the latest one.
+        """
+        return (
+            self._eligible_for_latest()
+            .order_by("provider_id", *self.LATEST_ORDER_BY)
+            .distinct("provider_id")
+        )
+
+    def latest_ids_per_provider(self) -> list[UUID]:
+        """Evaluate `latest_per_provider` and return the scan ids.
+
+        The ids are materialised so callers can pass them as a literal `IN`
+        list; as a subquery Postgres misestimates the row count and picks a
+        slow nested loop.
+
+        Returns:
+            list[UUID]: The id of each provider's latest scan.
+        """
+        return list(self.latest_per_provider().values_list("id", flat=True))
+
+    def latest_first(self) -> "ScanQuerySet":
+        """Order eligible scans newest first, without deduplicating per provider.
+
+        Expects the queryset to be already filtered to a single provider.
+
+        Returns:
+            ScanQuerySet: The eligible scans, latest first.
+        """
+        return self._eligible_for_latest().order_by(*self.LATEST_ORDER_BY)
+
+
 class Scan(RowLevelSecurityProtectedModel):
-    objects = ActiveProviderManager()
-    all_objects = models.Manager()
+    objects = ActiveProviderManager.from_queryset(ScanQuerySet)()
+    all_objects = ScanQuerySet.as_manager()
 
     _SCOPING_SCANNER_ARG_KEYS_CACHE: tuple[str, ...] | None = None
 
@@ -726,6 +783,12 @@ class Scan(RowLevelSecurityProtectedModel):
                 name="scans_prov_state_ins_desc_idx",
             ),
             # TODO This might replace `scans_prov_state_ins_desc_idx` completely. Review usage
+            # Since `ScanQuerySet`, no code path reads a provider's
+            # completed scans by `-inserted_at`. The only query left that
+            # matches this index (and `scans_prov_state_ins_desc_idx` above)
+            # is `GET /scans?filter[provider]=…&filter[state]=completed` with
+            # the default sort. Both are candidates to drop in a follow-up
+            # once production `pg_stat_user_indexes.idx_scan` confirms it.
             models.Index(
                 fields=["tenant_id", "provider_id", "-inserted_at"],
                 condition=Q(state=StateChoices.COMPLETED),

api/src/backend/api/tests/test_models.py

@@ -1,14 +1,16 @@
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 
 import pytest
 from allauth.socialaccount.models import SocialApp
 from api.db_router import MainRouter
 from api.models import (
+    Provider,
     ProviderComplianceScore,
     Resource,
     ResourceTag,
     SAMLConfiguration,
     SAMLDomainIndex,
+    Scan,
     StateChoices,
     StatusChoices,
     TenantComplianceSummary,
@@ -524,3 +526,226 @@ class TestTenantComplianceSummaryModel:
 
         assert summary1.id != summary2.id
         assert summary1.requirements_passed != summary2.requirements_passed
+
+
+def _latest_scan_fixture(tenant, provider, *, completed_at, inserted_at=None, **kwargs):
+    scan = Scan.objects.create(
+        tenant_id=tenant.id,
+        provider=provider,
+        trigger=kwargs.pop("trigger", Scan.TriggerChoices.MANUAL),
+        state=kwargs.pop("state", StateChoices.COMPLETED),
+        completed_at=completed_at,
+        **kwargs,
+    )
+    if inserted_at is not None:
+        # `inserted_at` is auto_now_add, so it has to be forced after the fact.
+        Scan.all_objects.filter(pk=scan.pk).update(inserted_at=inserted_at)
+        scan.refresh_from_db()
+    return scan
+
+
+@pytest.mark.django_db
+class TestScanQuerySetOrdering:
+    def test_completed_later_wins_over_inserted_later(
+        self, tenants_fixture, aws_provider
+    ):
+        """The scan that FINISHED last is current, not the one that started last."""
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        finished_last = _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            inserted_at=now - timedelta(hours=3),
+            completed_at=now,
+        )
+        _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            inserted_at=now - timedelta(hours=1),
+            completed_at=now - timedelta(hours=1),
+        )
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id
+        ).latest_ids_per_provider() == [finished_last.id]
+
+    def test_null_completed_at_provider_is_still_returned(
+        self, tenants_fixture, aws_provider
+    ):
+        """NULLS LAST, not `completed_at__isnull=False`.
+
+        Excluding NULL `completed_at` would drop the provider from every
+        "latest" endpoint instead of falling back to `inserted_at`.
+        """
+        tenant, *_ = tenants_fixture
+        only_scan = _latest_scan_fixture(tenant, aws_provider, completed_at=None)
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id
+        ).latest_ids_per_provider() == [only_scan.id]
+
+    def test_null_completed_at_never_outranks_a_finished_scan(
+        self, tenants_fixture, aws_provider
+    ):
+        """Postgres sorts NULLs first under DESC; NULLS LAST is what fixes it."""
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        finished = _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            inserted_at=now - timedelta(hours=2),
+            completed_at=now - timedelta(hours=2),
+        )
+        _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            inserted_at=now,
+            completed_at=None,
+        )
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id
+        ).latest_ids_per_provider() == [finished.id]
+
+    def test_id_breaks_an_exact_timestamp_tie_deterministically(
+        self, tenants_fixture, aws_provider
+    ):
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        scans = [
+            _latest_scan_fixture(
+                tenant, aws_provider, inserted_at=now, completed_at=now
+            )
+            for _ in range(3)
+        ]
+        expected = max(scan.id for scan in scans)
+
+        picks = {
+            Scan.all_objects.filter(tenant_id=tenant.id).latest_ids_per_provider()[0]
+            for _ in range(5)
+        }
+        assert picks == {expected}
+
+
+@pytest.mark.django_db
+class TestScanQuerySetEligibility:
+    def test_unfinished_scans_are_excluded(self, tenants_fixture, aws_provider):
+        tenant, *_ = tenants_fixture
+        _latest_scan_fixture(
+            tenant,
+            aws_provider,
+            completed_at=None,
+            state=StateChoices.EXECUTING,
+        )
+        assert (
+            Scan.all_objects.filter(tenant_id=tenant.id).latest_ids_per_provider() == []
+        )
+
+
+@pytest.mark.django_db
+class TestScanQuerySetManagerChoice:
+    def test_active_manager_hides_soft_deleted_providers(
+        self, tenants_fixture, aws_provider
+    ):
+        """`Scan.objects` drops soft-deleted providers, `all_objects` keeps them.
+
+        The queryset must not decide this for the caller.
+        """
+        tenant, *_ = tenants_fixture
+        scan = _latest_scan_fixture(
+            tenant, aws_provider, completed_at=datetime.now(UTC)
+        )
+
+        Provider.all_objects.filter(pk=aws_provider.pk).update(is_deleted=True)
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id
+        ).latest_ids_per_provider() == [scan.id]
+        assert Scan.objects.filter(tenant_id=tenant.id).latest_ids_per_provider() == []
+
+
+@pytest.mark.django_db
+class TestScanQuerySetPerProviderScoping:
+    def test_one_scan_per_provider(self, tenants_fixture, aws_provider_pair):
+        tenant, *_ = tenants_fixture
+        provider_one, provider_two = aws_provider_pair
+        now = datetime.now(UTC)
+
+        newest_one = _latest_scan_fixture(tenant, provider_one, completed_at=now)
+        _latest_scan_fixture(tenant, provider_one, completed_at=now - timedelta(days=1))
+        newest_two = _latest_scan_fixture(tenant, provider_two, completed_at=now)
+
+        assert set(
+            Scan.all_objects.filter(tenant_id=tenant.id).latest_ids_per_provider()
+        ) == {newest_one.id, newest_two.id}
+
+    def test_caller_filters_are_preserved(self, tenants_fixture, aws_provider_pair):
+        tenant, *_ = tenants_fixture
+        provider_one, provider_two = aws_provider_pair
+        now = datetime.now(UTC)
+
+        scan_one = _latest_scan_fixture(tenant, provider_one, completed_at=now)
+        _latest_scan_fixture(tenant, provider_two, completed_at=now)
+
+        assert Scan.all_objects.filter(
+            tenant_id=tenant.id, provider__in=[provider_one]
+        ).latest_ids_per_provider() == [scan_one.id]
+
+    def test_latest_first_is_ordered_not_deduplicated(
+        self, tenants_fixture, aws_provider
+    ):
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        newest = _latest_scan_fixture(tenant, aws_provider, completed_at=now)
+        older = _latest_scan_fixture(
+            tenant, aws_provider, completed_at=now - timedelta(days=1)
+        )
+
+        ordered = list(
+            Scan.all_objects.filter(
+                tenant_id=tenant.id, provider_id=aws_provider.id
+            ).latest_first()
+        )
+        assert [scan.id for scan in ordered] == [newest.id, older.id]
+
+    def test_tenant_isolation(self, tenants_fixture, aws_provider):
+        tenant, other_tenant, *_ = tenants_fixture
+        _latest_scan_fixture(tenant, aws_provider, completed_at=datetime.now(UTC))
+
+        assert (
+            Scan.all_objects.filter(tenant_id=other_tenant.id).latest_ids_per_provider()
+            == []
+        )
+
+    def test_empty_queryset_returns_empty_list(self, tenants_fixture):
+        tenant, *_ = tenants_fixture
+        assert (
+            Scan.all_objects.filter(tenant_id=tenant.id).latest_ids_per_provider() == []
+        )
+
+
+@pytest.mark.django_db
+class TestScanQuerySetPerProviderQuerysetShape:
+    def test_returns_a_queryset_not_a_list(self, tenants_fixture, aws_provider):
+        tenant, *_ = tenants_fixture
+        _latest_scan_fixture(tenant, aws_provider, completed_at=datetime.now(UTC))
+
+        qs = Scan.all_objects.filter(tenant_id=tenant.id).latest_per_provider()
+        # Callers chain .values(...) / .values_list(...) onto this.
+        assert qs.values_list("provider_id", flat=True).count() == 1
+
+
+@pytest.mark.django_db
+class TestScanQuerySetRelatedManager:
+    def test_reverse_relation_exposes_the_methods(self, tenants_fixture, aws_provider):
+        tenant, *_ = tenants_fixture
+        now = datetime.now(UTC)
+
+        newest = _latest_scan_fixture(tenant, aws_provider, completed_at=now)
+        _latest_scan_fixture(tenant, aws_provider, completed_at=now - timedelta(days=1))
+
+        assert aws_provider.scans.latest_first().first().id == newest.id

api/src/backend/api/v1/views.py

@@ -3478,12 +3478,8 @@ class ResourceViewSet(PaginateByPkMixin, BaseRLSViewSet):
         filtered_queryset = self.filter_queryset(self.get_queryset())
 
         latest_scans = (
-            Scan.all_objects.filter(
-                tenant_id=tenant_id,
-                state=StateChoices.COMPLETED,
-            )
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
+            Scan.all_objects.filter(tenant_id=tenant_id)
+            .latest_per_provider()
             .values("provider_id")
         )
 
@@ -3615,11 +3611,9 @@ class ResourceViewSet(PaginateByPkMixin, BaseRLSViewSet):
         tenant_id = request.tenant_id
         query_params = request.query_params
 
-        latest_scans_queryset = (
-            Scan.all_objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-        )
+        latest_scans_queryset = Scan.all_objects.filter(
+            tenant_id=tenant_id
+        ).latest_per_provider()
 
         queryset = ResourceScanSummary.objects.filter(
             tenant_id=tenant_id,
@@ -4206,12 +4200,9 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
         tenant_id = request.tenant_id
         filtered_queryset = self.filter_queryset(self.get_queryset())
 
-        latest_scan_ids = list(
-            Scan.all_objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        latest_scan_ids = Scan.all_objects.filter(
+            tenant_id=tenant_id
+        ).latest_ids_per_provider()
         filtered_queryset = filtered_queryset.filter(
             tenant_id=tenant_id, scan_id__in=latest_scan_ids
         )
@@ -4234,11 +4225,9 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
         tenant_id = request.tenant_id
         query_params = request.query_params
 
-        latest_scans_queryset = (
-            Scan.all_objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-        )
+        latest_scans_queryset = Scan.all_objects.filter(
+            tenant_id=tenant_id
+        ).latest_per_provider()
         raw_latest_scans_ids = list(
             latest_scans_queryset.values_list("id", "unique_resource_count")
         )
@@ -4979,11 +4968,7 @@ class ComplianceOverviewViewSet(
         if provider_filters:
             scans = scans.filter(**provider_filters)
 
-        return list(
-            scans.order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        return scans.latest_ids_per_provider()
 
     def _filtered_queryset_for_latest_provider_scans(self, latest_scan_ids=None):
         if latest_scan_ids is None:
@@ -5725,14 +5710,9 @@ class OverviewViewSet(ProviderFilterParamsMixin, BaseRLSViewSet):
             else {}
         )
 
-        latest_scan_ids = (
-            Scan.all_objects.filter(
-                tenant_id=tenant_id, state=StateChoices.COMPLETED, **provider_filter
-            )
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        latest_scan_ids = Scan.all_objects.filter(
+            tenant_id=tenant_id, **provider_filter
+        ).latest_ids_per_provider()
 
         return filtered_queryset.filter(
             tenant_id=tenant_id, scan_id__in=latest_scan_ids
@@ -5759,16 +5739,10 @@ class OverviewViewSet(ProviderFilterParamsMixin, BaseRLSViewSet):
 
     def _latest_scan_ids_for_allowed_providers(self, tenant_id, provider_filters=None):
         provider_filter = self._get_provider_filter()
-        queryset = Scan.all_objects.filter(
-            tenant_id=tenant_id, state=StateChoices.COMPLETED, **provider_filter
-        )
+        queryset = Scan.all_objects.filter(tenant_id=tenant_id, **provider_filter)
         if provider_filters:
             queryset = queryset.filter(**provider_filters)
-        return (
-            queryset.order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        return queryset.latest_ids_per_provider()
 
     @action(detail=False, methods=["get"], url_name="providers")
     def providers(self, request):
@@ -5780,14 +5754,9 @@ class OverviewViewSet(ProviderFilterParamsMixin, BaseRLSViewSet):
             else {}
         )
 
-        latest_scan_ids = (
-            Scan.all_objects.filter(
-                tenant_id=tenant_id, state=StateChoices.COMPLETED, **provider_filter
-            )
-            .order_by("provider_id", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        latest_scan_ids = Scan.all_objects.filter(
+            tenant_id=tenant_id, **provider_filter
+        ).latest_ids_per_provider()
 
         findings_aggregated = (
             queryset.filter(scan_id__in=latest_scan_ids)
@@ -7934,18 +7903,13 @@ class FindingGroupViewSet(JsonApiFilterMixin, BaseRLSViewSet):
 
     def _get_latest_findings_per_provider(self, filtered_queryset):
         """Keep only findings from each provider's most recent completed scan."""
-        # Materialize to a literal IN list. Left as a subquery, Postgres can't
-        # estimate the match count and picks a serial nested loop on
-        # resource_finding_mappings when one scan dominates findings
-        latest_scan_ids = list(
-            Scan.objects.filter(
-                tenant_id=self.request.tenant_id,
-                state=StateChoices.COMPLETED,
-            )
-            .order_by("provider_id", "-completed_at", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        # `latest_ids_per_provider` materializes to a literal IN list.
+        # Left as a subquery, Postgres can't estimate the match count and picks
+        # a serial nested loop on resource_finding_mappings when one scan
+        # dominates findings
+        latest_scan_ids = Scan.objects.filter(
+            tenant_id=self.request.tenant_id
+        ).latest_ids_per_provider()
         return filtered_queryset.filter(scan_id__in=latest_scan_ids)
 
     def _post_process_aggregation(self, aggregated_data):
@@ -8891,16 +8855,14 @@ class FindingGroupViewSet(JsonApiFilterMixin, BaseRLSViewSet):
         tenant_id = request.tenant_id
         queryset = self._get_finding_queryset()
 
-        # Order by -completed_at (matching the /latest summary path and the
-        # daily summary upsert keyed on midnight(completed_at)) so that
-        # overlapping scans do not make /resources and /latest read from
-        # different scans and report diverging counts.
-        latest_scan_ids = (
-            Scan.objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
-            .order_by("provider_id", "-completed_at", "-inserted_at")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        # The shared selector orders by -completed_at (matching the /latest
+        # summary path and the daily summary upsert keyed on
+        # midnight(completed_at)) so that overlapping scans do not make
+        # /resources and /latest read from different scans and report
+        # diverging counts.
+        latest_scan_ids = Scan.objects.filter(
+            tenant_id=tenant_id
+        ).latest_ids_per_provider()
 
         normalized_params = self._normalize_jsonapi_params(request.query_params)
         # Remove date filters since we're using latest

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

@@ -499,10 +499,13 @@ def backfill_provider_compliance_scores(tenant_id: str) -> dict:
                 provider_id__in=existing_providers
             )
 
+        # `completed_scans` keeps its own `completed_at__isnull=False`: this
+        # task writes a *dated* ProviderComplianceScore row, so unlike the read
+        # paths it genuinely cannot use a scan without a `completed_at`.
         scan_info = list(
-            completed_scans.order_by("provider_id", "-completed_at")
-            .distinct("provider_id")
-            .values("id", "provider_id", "completed_at")
+            completed_scans.latest_per_provider().values(
+                "id", "provider_id", "completed_at"
+            )
         )
 
         if not scan_info:

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

@@ -1,7 +1,7 @@
 from collections.abc import Iterable
 
 from api.db_utils import rls_transaction
-from api.models import Finding, MuteRule, Scan, StateChoices
+from api.models import Finding, MuteRule, Scan
 from celery.utils.log import get_task_logger
 
 logger = get_task_logger(__name__)
@@ -39,17 +39,9 @@ def mute_findings_in_latest_scans(
 
     with rls_transaction(tenant_id):
         mute_rule = MuteRule.objects.get(id=mute_rule_id, tenant_id=tenant_id)
-        latest_scans = list(
-            Scan.objects.filter(
-                tenant_id=tenant_id,
-                provider_id__in=provider_ids,
-                state=StateChoices.COMPLETED,
-                completed_at__isnull=False,
-            )
-            .order_by("provider_id", "-completed_at", "-inserted_at", "-id")
-            .distinct("provider_id")
-            .values_list("id", flat=True)
-        )
+        latest_scans = Scan.objects.filter(
+            tenant_id=tenant_id, provider_id__in=provider_ids
+        ).latest_ids_per_provider()
 
         changed_scan_ids = []
         findings_muted = 0

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

@@ -2700,20 +2700,37 @@ def reset_ephemeral_resource_findings_count(tenant_id: str, scan_id: str) -> dic
     # refreshed). Wiping based on the older scan would zero counts the newer
     # scan just set. Skip and let the newer scan's reset task do the work; if
     # this task was delayed in the queue, that's the correct outcome.
-    # `completed_at__isnull=False` is required: Postgres orders NULL first in
-    # DESC, so a sibling COMPLETED scan with a missing completed_at would sort
-    # as "newest" and incorrectly cause us to skip.
+    #
+    # The comparison must be against the newest *full-scope* scan, which is
+    # what this variable has always been named after but did not use to be:
+    # the query filtered nothing about scope, so any newer scan that is not
+    # full-scope (an imported one, for instance) made the full-scope scan
+    # skip its own cleanup and leave ephemeral resources with a stale
+    # failed_findings_count permanently.
+    #
+    # `is_full_scope()` reads `trigger` plus the scoping keys inside
+    # `scanner_args`, which is not expressible as a WHERE clause, so the
+    # candidates are walked newest-first in Python until the first full-scope
+    # one. The walk needs no cap: `scan` is itself a full-scope candidate, so
+    # it stops at `scan` at the latest, after reading only the scans newer than
+    # it. A fixed window would return None once more newer scoped scans had
+    # landed than it inspected, and skip the cleanup exactly like the bug above.
+    #
+    # NULL `completed_at` no longer needs an explicit filter here: the shared
+    # ordering in `ScanQuerySet.LATEST_ORDER_BY` sorts NULLs last
+    # rather than excluding them, which also fixes the case where a provider
+    # whose completed scans all have a NULL `completed_at` resolved to None and
+    # therefore never ran the reset at all.
     with rls_transaction(tenant_id):
-        latest_full_scope_scan_id = (
-            Scan.objects.filter(
-                tenant_id=tenant_id,
-                provider_id=scan.provider_id,
-                state=StateChoices.COMPLETED,
-                completed_at__isnull=False,
-            )
-            .order_by("-completed_at", "-inserted_at")
-            .values_list("id", flat=True)
-            .first()
+        candidates = (
+            Scan.objects.filter(tenant_id=tenant_id, provider_id=scan.provider_id)
+            .latest_first()
+            .only("id", "trigger", "scanner_args")
+            .iterator(chunk_size=100)
+        )
+        latest_full_scope_scan_id = next(
+            (candidate.id for candidate in candidates if candidate.is_full_scope()),
+            None,
         )
     if latest_full_scope_scan_id != scan.id:
         logger.info(

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

@@ -5967,6 +5967,112 @@ class TestResetEphemeralResourceFindingsCount:
         resource2.refresh_from_db()
         assert resource2.failed_findings_count == 5
 
+    def test_runs_when_newer_scan_is_not_full_scope(
+        self, tenants_fixture, scans_fixture, aws_provider, resources_fixture
+    ):
+        """A newer scoped scan must not block the full-scope scan's cleanup.
+
+        The race guard used to pick the newest COMPLETED scan of any scope
+        despite being named after full-scope ones, so a single scoped scan
+        landing after a complete one made the complete scan skip its cleanup
+        and leave ephemeral resources with a stale count permanently.
+        """
+        from datetime import timedelta
+
+        tenant, *_ = tenants_fixture
+        scan1, *_ = scans_fixture
+        resource1, resource2, _ = resources_fixture
+
+        Resource.objects.filter(id=resource2.id).update(failed_findings_count=5)
+        self._make_scan_summary(tenant.id, scan1.id, resource1)
+
+        newer_completed_at = scan1.completed_at + timedelta(minutes=5)
+        Scan.objects.create(
+            name="Newer scoped scan",
+            provider=aws_provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant_id=tenant.id,
+            started_at=newer_completed_at,
+            completed_at=newer_completed_at,
+            scanner_args={"checks": ["check1"]},
+        )
+
+        result = reset_ephemeral_resource_findings_count(
+            tenant_id=str(tenant.id), scan_id=str(scan1.id)
+        )
+
+        assert result["status"] == "completed"
+
+        resource2.refresh_from_db()
+        assert resource2.failed_findings_count == 0
+
+    def test_runs_when_many_newer_scans_are_not_full_scope(
+        self, tenants_fixture, scans_fixture, aws_provider, resources_fixture
+    ):
+        """The walk must not give up before it reaches the full-scope scan.
+
+        A fixed look-back window returned None once more newer scoped scans
+        had landed than it inspected, and skipped the cleanup exactly like the
+        original bug did with one.
+        """
+        from datetime import timedelta
+
+        tenant, *_ = tenants_fixture
+        scan1, *_ = scans_fixture
+        resource1, resource2, _ = resources_fixture
+
+        Resource.objects.filter(id=resource2.id).update(failed_findings_count=5)
+        self._make_scan_summary(tenant.id, scan1.id, resource1)
+
+        for minutes in range(1, 41):
+            newer_completed_at = scan1.completed_at + timedelta(minutes=minutes)
+            Scan.objects.create(
+                name=f"Newer scoped scan {minutes}",
+                provider=aws_provider,
+                trigger=Scan.TriggerChoices.MANUAL,
+                state=StateChoices.COMPLETED,
+                tenant_id=tenant.id,
+                started_at=newer_completed_at,
+                completed_at=newer_completed_at,
+                scanner_args={"checks": ["check1"]},
+            )
+
+        result = reset_ephemeral_resource_findings_count(
+            tenant_id=str(tenant.id), scan_id=str(scan1.id)
+        )
+
+        assert result["status"] == "completed"
+
+        resource2.refresh_from_db()
+        assert resource2.failed_findings_count == 0
+
+    def test_runs_when_completed_at_is_null(
+        self, tenants_fixture, scans_fixture, aws_provider, resources_fixture
+    ):
+        """NULL `completed_at` used to make the reset never run at all.
+
+        The old guard filtered `completed_at__isnull=False`, so a provider
+        whose completed scans all had a NULL `completed_at` resolved the
+        "latest" scan to None, which never equals `scan.id`.
+        """
+        tenant, *_ = tenants_fixture
+        scan1, *_ = scans_fixture
+        resource1, resource2, _ = resources_fixture
+
+        Scan.all_objects.filter(id=scan1.id).update(completed_at=None)
+        Resource.objects.filter(id=resource2.id).update(failed_findings_count=5)
+        self._make_scan_summary(tenant.id, scan1.id, resource1)
+
+        result = reset_ephemeral_resource_findings_count(
+            tenant_id=str(tenant.id), scan_id=str(scan1.id)
+        )
+
+        assert result["status"] == "completed"
+
+        resource2.refresh_from_db()
+        assert resource2.failed_findings_count == 0
+
     def test_does_not_touch_other_providers_resources(
         self, tenants_fixture, scans_fixture, aws_provider, resources_fixture
     ):