fix(derived data): Adjust auto-heal timeouts in response to operational realities

src/sentry/issues/derived/tasks.py

@@ -42,7 +42,7 @@
 _MAX_PROJECT_GROUPS = 10_000
 # Hard cap on distinct stale hashes discovered per scan.
 _MAX_STALE_HASHES = 5
-_STALE_HASH_DISCOVERY_TIMEOUT = timedelta(seconds=5)
+_STALE_HASH_DISCOVERY_TIMEOUT = timedelta(seconds=15)
 
 
 def _stale_pipeline_filter(qs: BaseQuerySet[Group], pipeline_hash: str) -> BaseQuerySet[Group]:
@@ -476,7 +476,7 @@ def _discover_stale_pipeline_hashes(current_hash: str, limit: int) -> list[str]:
     name="sentry.issues.derived.tasks.heal_stale_derived_data",
     namespace=issues_tasks,
     silo_mode=SiloMode.CELL,
-    processing_deadline_duration=60,
+    processing_deadline_duration=120,
 )
 def heal_stale_derived_data(**kwargs: object) -> None:
     """Rebuild a chunk of GroupDerivedData rows whose ``pipeline_hash`` is stale/NULL."""
@@ -583,19 +583,36 @@ def heal_stale_derived_data(**kwargs: object) -> None:
             "heal_stale_derived_data.range_selection_started",
             extra={"hash_kind": hash_kind, "remaining_budget": remaining},
         )
-        range_result = group_id_ranges_for_hash(
-            stale_hash,
-            chunk_size=batch_size,
-            max_chunks=remaining,
-            group_id_lower_bound=lower_bound,
-        )
+        range_selection_started_at = time.monotonic()
+        try:
+            range_result = group_id_ranges_for_hash(
+                stale_hash,
+                chunk_size=batch_size,
+                max_chunks=remaining,
+                group_id_lower_bound=lower_bound,
+            )
+        except OperationalError:
+            logger.exception(
+                "heal_stale_derived_data.range_selection_failed",
+                extra={
+                    "hash_kind": hash_kind,
+                    "elapsed": time.monotonic() - range_selection_started_at,
+                },
+            )
+            metrics.incr(
+                "issues.derived.heal_range_selection_failed",
+                sample_rate=1.0,
+                tags={"hash_kind": hash_kind},
+            )
+            continue
         ranges = range_result.ranges
         logger.info(
             "heal_stale_derived_data.range_selection_complete",
             extra={
                 "hash_kind": hash_kind,
                 "range_count": len(ranges),
                 "remaining_budget": remaining,
+                "elapsed": time.monotonic() - range_selection_started_at,
             },
         )
         if stale_hash is not None and range_result.drained:

src/sentry/issues/derived/tasks_util.py

@@ -3,7 +3,7 @@
 import logging
 import random
 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Protocol
 
 from django.db import connections, router
@@ -13,6 +13,7 @@
 from sentry.issues.models.groupderiveddata import GroupDerivedData
 from sentry.taskworker.selfchain_idempotency import already_spawned, mark_spawned
 from sentry.utils import metrics
+from sentry.utils.db import statement_timeout
 
 logger = logging.getLogger(__name__)
 
@@ -21,6 +22,7 @@
 # Safety valve on the number of group IDs one ``group_id_ranges_for_hash`` call may
 # walk, however large the requested chunking is.
 _MAX_SCANNED_GROUP_IDS = 2_000_000
+_GROUP_ID_RANGE_TIMEOUT = timedelta(seconds=50)
 
 
 @dataclass(frozen=True)
@@ -170,6 +172,7 @@ def group_id_ranges_for_hash(
 
     using = router.db_for_read(GroupDerivedData)
     with (
+        statement_timeout(using, _GROUP_ID_RANGE_TIMEOUT),
         metrics.timer("issues.derived.group_id_range_query"),
         connections[using].cursor() as cursor,
     ):

tests/sentry/issues/derived/test_tasks.py

@@ -461,6 +461,12 @@ def test_logs_progress_through_scheduling_stages(self) -> None:
             "heal_stale_derived_data.checks_scheduled",
             "heal_stale_derived_data.complete",
         ]
+        range_selection_logs = [
+            log_call
+            for log_call in mock_logger.info.call_args_list
+            if log_call.args[0] == "heal_stale_derived_data.range_selection_complete"
+        ]
+        assert all(log_call.kwargs["extra"]["elapsed"] >= 0 for log_call in range_selection_logs)
 
     def test_missing_state_discovers_and_reports_metric(self) -> None:
         with (
@@ -597,6 +603,48 @@ def test_discovery_timeout_still_schedules_null_and_saves_retryable_state(self)
             "heal_stale_derived_data.stale_hash_discovery_failed"
         )
 
+    def test_range_selection_timeout_is_reported_and_other_hashes_continue(self) -> None:
+        stale_hash = self._pick_stale_hash()
+        state = HealSchedulerState(
+            head_hash=PIPELINE.pipeline_hash,
+            stale={stale_hash: 10},
+            discovered_at=datetime.now(timezone.utc),
+        )
+        with (
+            override_options(
+                {
+                    "issues.derived.heal-max-tasks": 1,
+                    "issues.derived.check-task-count": 0,
+                }
+            ),
+            patch("sentry.issues.derived.tasks.load_state", return_value=state),
+            patch(
+                "sentry.issues.derived.tasks_util.group_id_ranges_for_hash",
+                side_effect=[
+                    OperationalError,
+                    GroupIdRangeResult(ranges=[(10, 20)], drained=False),
+                ],
+            ),
+            patch.object(regenerate_stale_derived_data_batch, "delay") as delay,
+            patch("sentry.issues.derived.tasks.metrics.incr") as mock_incr,
+            patch("sentry.issues.derived.tasks.logger") as mock_logger,
+        ):
+            heal_stale_derived_data()
+
+        delay.assert_called_once()
+        failure_log = mock_logger.exception.call_args
+        assert failure_log.args == ("heal_stale_derived_data.range_selection_failed",)
+        assert failure_log.kwargs["extra"]["hash_kind"] == "null"
+        assert failure_log.kwargs["extra"]["elapsed"] >= 0
+        assert (
+            call(
+                "issues.derived.heal_range_selection_failed",
+                sample_rate=1.0,
+                tags={"hash_kind": "null"},
+            )
+            in mock_incr.call_args_list
+        )
+
     def test_discovered_hashes_are_saved_before_range_selection(self) -> None:
         stale_hash = self._pick_stale_hash()
         with (
@@ -1347,6 +1395,12 @@ def test_no_matching_rows(self) -> None:
             ranges=[], drained=True
         )
 
+    def test_query_has_statement_timeout(self) -> None:
+        with patch("sentry.issues.derived.tasks_util.statement_timeout") as timeout:
+            group_id_ranges_for_hash(self.HASH, chunk_size=2, max_chunks=5)
+
+        assert timeout.call_args.args[1] == timedelta(seconds=50)
+
     def test_short_tail_is_one_range(self) -> None:
         null_ids = self._seed(3, None)
         hash_ids = self._seed(3, self.HASH)
@@ -1635,6 +1689,12 @@ def test_returns_empty_when_only_current_hash_present(self) -> None:
     def test_returns_empty_when_table_empty(self) -> None:
         assert _discover_stale_pipeline_hashes(PIPELINE.pipeline_hash, limit=5) == []
 
+    def test_query_has_statement_timeout(self) -> None:
+        with patch("sentry.issues.derived.tasks.statement_timeout") as timeout:
+            _discover_stale_pipeline_hashes(PIPELINE.pipeline_hash, limit=5)
+
+        assert timeout.call_args.args[1] == timedelta(seconds=15)
+
     def test_excludes_null_pipeline_hash(self) -> None:
         current = PIPELINE.pipeline_hash
         self._seed_hashes([None, None])