feat(issues): Wire EAP double-read into issue feed search

src/sentry/search/snuba/executors.py

@@ -37,6 +37,8 @@
 from sentry.models.group import Group
 from sentry.models.organization import Organization
 from sentry.models.project import Project
+from sentry.search.eap.occurrences.rollout_utils import EAPOccurrencesComparator
+from sentry.search.eap.occurrences.search_executor import EAP_SORT_STRATEGIES, run_eap_group_search
 from sentry.search.events.filter import convert_search_filter_to_snuba_query, format_search_filter
 from sentry.snuba.dataset import Dataset
 from sentry.users.models.user import User
@@ -45,6 +47,8 @@
 from sentry.utils.cursors import Cursor, CursorResult
 from sentry.utils.snuba import SnubaQueryParams, aliased_query_params, bulk_raw_query
 
+logger = logging.getLogger(__name__)
+
 FIRST_RELEASE_FILTERS = ["first_release", "firstRelease"]
 
 
@@ -90,6 +94,19 @@ class Clauses(Enum):
 ENTITY_SEARCH_ISSUES = "search_issues"
 
 
+def _reasonable_search_result_match(
+    control: tuple[list[tuple[int, Any]], int],
+    experimental: tuple[list[tuple[int, Any]], int],
+) -> bool:
+    control_group_ids = {gid for gid, _ in control[0]}
+    experimental_group_ids = {gid for gid, _ in experimental[0]}
+
+    if not experimental_group_ids:
+        return True
+
+    return experimental_group_ids.issubset(control_group_ids)
+
+
 @dataclass
 class TrendsParams:
     # (event or issue age_hours) / (event or issue halflife hours)
@@ -510,7 +527,50 @@ def snuba_search(
         if get_sample:
             sort_field = "sample"
 
-        return [(row["group_id"], row[sort_field]) for row in rows], total  # type: ignore[literal-required]
+        snuba_result = [(row["group_id"], row[sort_field]) for row in rows], total  # type: ignore[literal-required]
+        result = snuba_result
+
+        # Double-read from EAP for supported sort strategies
+        callsite = "PostgresSnubaQueryExecutor.snuba_search"
+        if (
+            not get_sample
+            and sort_field in EAP_SORT_STRATEGIES
+            and EAPOccurrencesComparator.should_check_experiment(callsite)
+        ):
+            try:
+                eap_result = run_eap_group_search(
+                    start=start,
+                    end=end,
+                    project_ids=project_ids,
+                    environment_ids=environment_ids,
+                    sort_field=sort_field,
+                    organization=organization,
+                    group_ids=group_ids,
+                    limit=limit,
+                    offset=offset,
+                    search_filters=snuba_search_filters,
+                    referrer=referrer,
+                )
+                result = EAPOccurrencesComparator.check_and_choose(
+                    snuba_result,
+                    eap_result,
+                    callsite,
+                    is_experimental_data_a_null_result=len(eap_result[0]) == 0,
+                    reasonable_match_comparator=_reasonable_search_result_match,
+                    debug_context={
+                        "sort_field": sort_field,
+                        "organization_id": organization.id,
+                        "num_group_ids": len(group_ids) if group_ids else 0,
+                        "num_filters": len(snuba_search_filters),
+                    },
+                )
+            except Exception:
+                logger.exception(
+                    "eap.double_read.snuba_search_failed",
+                    extra={"callsite": callsite, "sort_field": sort_field},
+                )
+
+        return result
 
     def has_sort_strategy(self, sort_by: str) -> bool:
         return sort_by in self.sort_strategies.keys()