diff --git a/src/sentry/issues/derived/tasks.py b/src/sentry/issues/derived/tasks.py index df06bf33d3d6..5e4a4fbacebe 100644 --- a/src/sentry/issues/derived/tasks.py +++ b/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,12 +583,28 @@ 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", @@ -596,6 +612,7 @@ def heal_stale_derived_data(**kwargs: object) -> None: "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: diff --git a/src/sentry/issues/derived/tasks_util.py b/src/sentry/issues/derived/tasks_util.py index cea61af19728..5da712e5c530 100644 --- a/src/sentry/issues/derived/tasks_util.py +++ b/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, ): diff --git a/tests/sentry/issues/derived/test_tasks.py b/tests/sentry/issues/derived/test_tasks.py index 3ba6bef832e7..5769ce118f75 100644 --- a/tests/sentry/issues/derived/test_tasks.py +++ b/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])