diff --git a/src/sentry/conf/server.py b/src/sentry/conf/server.py index d449e790f083..6c25f2877aa9 100644 --- a/src/sentry/conf/server.py +++ b/src/sentry/conf/server.py @@ -207,6 +207,7 @@ def env( SENTRY_SCM_REDIS_CLUSTER = "default" # Ephemeral dedup markers for self-chaining tasks (e.g. merge_groups, unmerge). SENTRY_SELFCHAIN_IDEMPOTENCY_REDIS_CLUSTER = "default" +SENTRY_UNMERGE_REDIS_CLUSTER = "default" # Hosts that are allowed to use system token authentication. # http://en.wikipedia.org/wiki/Reserved_IP_addresses diff --git a/src/sentry/features/temporary.py b/src/sentry/features/temporary.py index f15f6d52f725..b13ce5b29fea 100644 --- a/src/sentry/features/temporary.py +++ b/src/sentry/features/temporary.py @@ -448,6 +448,8 @@ def register_temporary_features(manager: FeatureManager) -> None: manager.add("organizations:transaction-name-normalize", OrganizationFeature, FeatureHandlerStrategy.INTERNAL, default=True, api_expose=False) # Enable unified table controls manager.add("organizations:unified-table-controls", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True) + # Enable recoverable unmerge read/handoff checkpoints. + manager.add("organizations:unmerge-recovery", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False) # Enables view hierarchy attachment scrubbing manager.add("organizations:view-hierarchy-scrubbing", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False) # Enables flamegraph attachments in issue details and their symbolication diff --git a/src/sentry/issues/endpoints/group_hashes.py b/src/sentry/issues/endpoints/group_hashes.py index 7e3738a755dd..f756875132b5 100644 --- a/src/sentry/issues/endpoints/group_hashes.py +++ b/src/sentry/issues/endpoints/group_hashes.py @@ -1,3 +1,4 @@ +import uuid from collections.abc import Sequence from functools import partial from typing import Annotated, TypedDict @@ -8,6 +9,7 @@ from rest_framework.request import Request from rest_framework.response import Response +from sentry import features from sentry.api.api_owners import ApiOwner from sentry.api.api_publish_status import ApiPublishStatus from sentry.api.base import DEFAULT_DURATION_BUDGET, cell_silo_endpoint @@ -42,6 +44,10 @@ from sentry.utils.snuba import raw_query +class UnmergeRecoveryArguments(TypedDict, total=False): + operation_id: str + + class GroupHashesResult(TypedDict): id: Annotated[str, ApiDescription("Fingerprint hash")] latestEvent: Annotated[ @@ -173,8 +179,20 @@ def put(self, request: Request, group: Group) -> Response: tags={"platform": group.platform or "unknown", "sdk": group.sdk or "unknown"}, ) + task_kwargs: UnmergeRecoveryArguments = {} + if features.has("organizations:unmerge-recovery", group.project.organization): + # Stable across starter and batch redeliveries; a broker activation + # ID alone does not identify duplicate child deliveries. + task_kwargs = { + "operation_id": uuid.uuid4().hex, + } start_unmerge.delay( - group.project_id, group.id, None, grouphashes, acting_user_id(request.user) + group.project_id, + group.id, + None, + grouphashes, + acting_user_id(request.user), + **task_kwargs, ) return Response(status=202) diff --git a/src/sentry/issues/unmerge_recovery.py b/src/sentry/issues/unmerge_recovery.py new file mode 100644 index 000000000000..556bfd2fe7d0 --- /dev/null +++ b/src/sentry/issues/unmerge_recovery.py @@ -0,0 +1,192 @@ +"""Redis ownership and the current batch/stage of a recoverable unmerge.""" + +from __future__ import annotations + +import logging +import uuid +from collections.abc import Generator, Mapping, Sequence +from contextlib import contextmanager +from typing import Any + +from django.conf import settings +from django.db import router, transaction +from redis.client import StrictRedis +from redis.exceptions import RedisError +from sentry_redis_tools.clients import RedisCluster +from taskbroker_client.retry import RetryTaskError + +from sentry.models.grouphash import GroupHash +from sentry.unmerge import UnmergeArgs +from sentry.utils import json, redis + +# A worker attempt has a 300s deadline. Recovery records outlive its lease. +LEASE_SECONDS = 600 +RETENTION_SECONDS = 7 * 24 * 60 * 60 +logger = logging.getLogger(__name__) +release_owned_key = redis.load_redis_script("utils/compare_and_delete.lua") +save_checkpoint_script = redis.load_redis_script("issues/unmerge/save_checkpoint.lua") + + +class UnmergeBusy(RetryTaskError): + pass + + +class UnmergeNeedsRecovery(Exception): + pass + + +def client() -> RedisCluster[str] | StrictRedis[str]: + return redis.redis_clusters.get(settings.SENTRY_UNMERGE_REDIS_CLUSTER) + + +def owner_key(project_id: int, fingerprint: str) -> str: + return f"unmerge:{{{project_id}}}:owner:{fingerprint}" + + +def claim_hashes( + project_id: int, fingerprints: Sequence[str], owner: str, duration: int = RETENTION_SECONDS +) -> None: + claimed: list[str] = [] + for fingerprint in sorted(set(fingerprints)): + key = owner_key(project_id, fingerprint) + if not (client().set(key, owner, nx=True, ex=duration) or client().get(key) == owner): + release_hashes(project_id, claimed, owner) + raise UnmergeNeedsRecovery("A fingerprint belongs to another unmerge") + claimed.append(fingerprint) + + +def release_hashes(project_id: int, fingerprints: Sequence[str], owner: str) -> None: + for fingerprint in fingerprints: + release_owned_key(keys=[owner_key(project_id, fingerprint)], args=[owner], client=client()) + + +@contextmanager +def reserve_legacy_hashes(project_id: int, fingerprints: Sequence[str]) -> Generator[None]: + owner = uuid.uuid4().hex + claim_hashes(project_id, fingerprints, owner, LEASE_SECONDS) + try: + yield + finally: + try: + release_hashes(project_id, fingerprints, owner) + except RedisError: + # SQL already protects the acquired hashes; reservations expire. + logger.exception( + "unmerge.legacy_reservation.release_failed", extra={"project_id": project_id} + ) + + +class UnmergeCheckpoint: + def __init__(self, args: UnmergeArgs) -> None: + assert args.operation_id is not None + self.args = args + self.owner = args.operation_id + self.key = f"unmerge:{{{args.project_id}}}:operation:{self.owner}" + self.lease_key = f"{self.key}:lease" + self.token = uuid.uuid4().hex + self.state: dict[str, Any] = {} + + @property + def phase(self) -> str: + return str(self.state["phase"]) + + @property + def finished(self) -> bool: + return self.phase == "complete" or self.args.batch_number < self.state["batch_number"] + + def __enter__(self) -> UnmergeCheckpoint: + if not client().set(self.lease_key, self.token, nx=True, ex=LEASE_SECONDS): + raise UnmergeBusy("Another worker is running this unmerge") + try: + raw = client().get(self.key) + if raw is None: + if self.args.batch_number: + raise UnmergeNeedsRecovery("Missing unmerge checkpoint") + self.state = {"phase": "new", "batch_number": 0} + else: + self.state = json.loads(raw) + if self.args.batch_number > self.state["batch_number"]: + if ( + self.phase != "handoff" + or self.args.batch_number != self.state["batch_number"] + 1 + ): + raise UnmergeNeedsRecovery("Unexpected unmerge batch") + self.save("reading", batch_number=self.args.batch_number, continuation=None) + return self + except BaseException: + self.__exit__(None, None, None) + raise + + def __exit__(self, *exc: Any) -> None: + release_owned_key(keys=[self.lease_key], args=[self.token], client=client()) + + def save(self, phase: str, **updates: Any) -> None: + state = {**self.state, **updates, "phase": phase} + if not save_checkpoint_script( + keys=[self.lease_key, self.key], + args=[self.token, json.dumps(state), RETENTION_SECONDS], + client=client(), + ): + raise UnmergeBusy("Unmerge worker lease has expired") + self.state = state + + def lock_hashes(self) -> list[str]: + fingerprints = sorted(set(self.args.replacement.primary_hashes_to_lock)) + hashes = GroupHash.objects.filter( + project_id=self.args.project_id, + group_id=self.args.source_id, + hash__in=fingerprints, + ) + if self.phase == "new": + claim_hashes(self.args.project_id, fingerprints, self.owner) + if hashes.filter(state=GroupHash.State.LOCKED_IN_MIGRATION).exists(): + release_hashes(self.args.project_id, fingerprints, self.owner) + raise UnmergeNeedsRecovery("Fingerprints are locked without an owned checkpoint") + # Remember ownership before SQL changes, so a retry can resume + # whether the transaction committed or rolled back. + self.save("reading") + else: + self.check_ownership(fingerprints) + with transaction.atomic(router.db_for_write(GroupHash)): + rows = list(hashes.order_by("id").select_for_update()) + if len(rows) == len(fingerprints): + hashes.update(state=GroupHash.State.LOCKED_IN_MIGRATION) + if len(rows) != len(fingerprints): + # Redis cleanup must happen outside the SQL transaction. + self.release_locks() + raise UnmergeNeedsRecovery("Requested fingerprints no longer belong to the source") + return fingerprints + + def check_ownership(self, fingerprints: Sequence[str]) -> None: + if fingerprints and client().mget( + [owner_key(self.args.project_id, fingerprint) for fingerprint in fingerprints] + ) != [self.owner] * len(fingerprints): + raise UnmergeNeedsRecovery("Unmerge fingerprint ownership is missing or has changed") + + def release_locks(self) -> None: + fingerprints = sorted(set(self.args.replacement.primary_hashes_to_lock)) + if self.phase == "new": + # No SQL locks were acquired by this operation. + self.save("releasing", continuation=None) + elif self.phase != "releasing": + self.check_ownership(fingerprints) + self.save("unlocking", continuation=None) + GroupHash.objects.filter( + project_id=self.args.project_id, + hash__in=fingerprints, + state=GroupHash.State.LOCKED_IN_MIGRATION, + ).update(state=GroupHash.State.UNLOCKED) + # Once owners start being released, retries must not unlock SQL: + # a different operation may already have acquired a released hash. + self.save("releasing") + release_hashes(self.args.project_id, fingerprints, self.owner) + self.complete() + + def begin_mutations(self) -> None: + self.save("applying") + + def handoff(self, arguments: Mapping[str, Any]) -> None: + self.save("handoff", continuation=arguments) + + def complete(self) -> None: + self.save("complete", continuation=None) diff --git a/src/sentry/scripts/issues/unmerge/save_checkpoint.lua b/src/sentry/scripts/issues/unmerge/save_checkpoint.lua new file mode 100644 index 000000000000..d87d52113530 --- /dev/null +++ b/src/sentry/scripts/issues/unmerge/save_checkpoint.lua @@ -0,0 +1,13 @@ +-- Save progress only while this worker still owns the operation lease. +local lease_key = KEYS[1] +local checkpoint_key = KEYS[2] +local worker_token = ARGV[1] +local checkpoint = ARGV[2] +local ttl = ARGV[3] + +if redis.call("GET", lease_key) ~= worker_token then + return 0 +end + +redis.call("SET", checkpoint_key, checkpoint, "EX", ttl) +return 1 diff --git a/src/sentry/tasks/unmerge.py b/src/sentry/tasks/unmerge.py index 3507071ba424..5e3d5317d6f2 100644 --- a/src/sentry/tasks/unmerge.py +++ b/src/sentry/tasks/unmerge.py @@ -5,10 +5,13 @@ from collections.abc import Callable, Collection, Mapping, Sequence from datetime import datetime from functools import reduce +from multiprocessing.context import TimeoutError from typing import Any from django.db import router, transaction from django.db.models.base import Model +from redis.exceptions import RedisError +from taskbroker_client.retry import LastAction, Retry from taskbroker_client.state import current_task from sentry import features, similarity, tsdb @@ -17,6 +20,12 @@ from sentry.issues.action_log import SYSTEM_ACTOR, ActionSource, action_context_scope from sentry.issues.action_log.backfill import BackfillEntry, backfill_actions from sentry.issues.action_log.types import FirstSeenAction, first_seen_idempotency_key +from sentry.issues.unmerge_recovery import ( + UnmergeBusy, + UnmergeCheckpoint, + UnmergeNeedsRecovery, + reserve_legacy_hashes, +) from sentry.killswitches import killswitch_matches_context from sentry.models.activity import Activity from sentry.models.environment import Environment @@ -38,7 +47,12 @@ from sentry.taskworker.selfchain_idempotency import already_spawned, mark_spawned from sentry.tsdb.base import TSDBModel from sentry.types.activity import ActivityType -from sentry.unmerge import InitialUnmergeArgs, SuccessiveUnmergeArgs, UnmergeArgs, UnmergeArgsBase +from sentry.unmerge import ( + InitialUnmergeArgs, + SuccessiveUnmergeArgs, + UnmergeArgs, + UnmergeArgsBase, +) from sentry.utils import metrics from sentry.utils.eventuser import EventUser from sentry.utils.query import task_run_batch_query @@ -470,7 +484,10 @@ def publish_destination_first_seen(project: Project, destination_ids: Collection def lock_hashes(project_id: int, source_id: int, fingerprints: Sequence[str]) -> list[str]: - with transaction.atomic(router.db_for_write(GroupHash)): + with ( + reserve_legacy_hashes(project_id, fingerprints), + transaction.atomic(router.db_for_write(GroupHash)), + ): eligible_hashes = list( GroupHash.objects.filter( project_id=project_id, group_id=source_id, hash__in=fingerprints @@ -502,6 +519,7 @@ def unlock_hashes(project_id: int, locked_primary_hashes: Sequence[str]) -> None namespace=issues_merge_tasks, processing_deadline_duration=300, silo_mode=SiloMode.CELL, + retry=Retry(times=12, delay=60, ignore=(TimeoutError,), times_exceeded=LastAction.Deadletter), ) def start_unmerge( project_id: int, @@ -509,11 +527,12 @@ def start_unmerge( destination_id: int | None, fingerprints: Sequence[str], actor_id: int | None, - batch_size: int = 500, + batch_size: int = 250, + operation_id: str | None = None, ) -> None: task_state = current_task() activation_id = task_state.id if task_state else None - if activation_id and already_spawned(_START_TASK_KEY, activation_id): + if not operation_id and activation_id and already_spawned(_START_TASK_KEY, activation_id): logger.info( "unmerge.start.duplicate_redelivery.skipped", extra={ @@ -536,6 +555,19 @@ def start_unmerge( ) metrics.incr("unmerge.started") + if operation_id: + _send_unmerge( + { + "project_id": project_id, + "source_id": source_id, + "destination_id": destination_id, + "fingerprints": fingerprints, + "actor_id": actor_id, + "batch_size": batch_size, + "operation_id": operation_id, + } + ) + return unmerge.delay( project_id, source_id, destination_id, fingerprints, actor_id, batch_size=batch_size ) @@ -548,18 +580,56 @@ def start_unmerge( namespace=issues_merge_tasks, processing_deadline_duration=300, silo_mode=SiloMode.CELL, + retry=Retry(times=12, delay=60, ignore=(TimeoutError,), times_exceeded=LastAction.Deadletter), ) def unmerge(*posargs: Any, **kwargs: Any) -> None: + args = UnmergeArgsBase.parse_arguments(*posargs, **kwargs) + if args.operation_id is None: + _unmerge(args) + return + + checkpoint = UnmergeCheckpoint(args) + try: + with checkpoint: + if checkpoint.finished: + return + if checkpoint.phase in ("unlocking", "releasing"): + checkpoint.release_locks() + return + if checkpoint.phase == "applying": + raise UnmergeNeedsRecovery("Unmerge batch may have partially applied mutations") + if checkpoint.phase != "handoff": + _unmerge(args, checkpoint) + continuation = checkpoint.state.get("continuation") + # Release the operation lease before delivery: the next batch may start + # immediately, while this worker is still waiting for confirmation. + if continuation: + _send_unmerge(continuation) + except (RedisError, MemoryError) as error: + raise UnmergeBusy("Unmerge read or checkpoint interrupted") from error + + +def _send_unmerge(arguments: Mapping[str, Any]) -> None: + try: + future = unmerge.apply_async_with_future(kwargs=arguments) + if future is not None: + future.result(timeout=30) + except Exception as error: + raise UnmergeBusy("Unmerge delivery was not confirmed") from error + + +def _unmerge(args: UnmergeArgs, checkpoint: UnmergeCheckpoint | None = None) -> None: with action_context_scope(ActionSource.SYSTEM): - args = UnmergeArgsBase.parse_arguments(*posargs, **kwargs) - extra = {"source_id": args.source_id, "project_id": args.project_id} + extra: dict[str, Any] = {"source_id": args.source_id, "project_id": args.project_id} + if checkpoint: + extra.update(operation_id=args.operation_id, batch_number=args.batch_number) # Self-chain idempotency guard. If this activation already produced its continuation in a # prior delivery, this execution is a broker re-pend: no-op so we don't fork the chain (and # skip the redundant re-processing). Only effective inside a worker (current_task() set). task_state = current_task() activation_id = task_state.id if task_state else None - if activation_id and already_spawned(_TASK_KEY, activation_id): + if checkpoint is None and activation_id and already_spawned(_TASK_KEY, activation_id): logger.info( "unmerge.duplicate_redelivery.skipped", extra={**extra, "activation_id": activation_id}, @@ -573,11 +643,17 @@ def unmerge(*posargs: Any, **kwargs: Any) -> None: "unmerge.killswitch-projects", {"project_id": args.project_id} ): logger.warning("unmerge.halted_by_killswitch", extra=extra) + hashes = list(args.replacement.primary_hashes_to_lock) + if checkpoint and checkpoint.phase != "new": + checkpoint.check_ownership(hashes) if isinstance(args, SuccessiveUnmergeArgs): - unlock_hashes(args.project_id, list(args.locked_primary_hashes)) + if checkpoint is None: + unlock_hashes(args.project_id, list(args.locked_primary_hashes)) for _unmerge_key, (_destination_id, eventstream_state) in args.destinations.items(): if eventstream_state: args.replacement.stop_snuba_replacement(eventstream_state) + if checkpoint: + checkpoint.release_locks() return source = Group.objects.get(project_id=args.project_id, id=args.source_id) @@ -590,15 +666,24 @@ def unmerge(*posargs: Any, **kwargs: Any) -> None: # denormalizations from the source group so that we can have a clean slate # for the new, repaired data. if isinstance(args, InitialUnmergeArgs): - locked_primary_hashes = lock_hashes( - args.project_id, args.source_id, list(args.replacement.primary_hashes_to_lock) + locked_primary_hashes = ( + checkpoint.lock_hashes() + if checkpoint + else lock_hashes( + args.project_id, args.source_id, list(args.replacement.primary_hashes_to_lock) + ) ) - truncate_denormalizations(project, source) last_event = None else: last_event = args.last_event locked_primary_hashes = list(args.locked_primary_hashes) + if not locked_primary_hashes: + if checkpoint: + checkpoint.complete() + return + if checkpoint: + checkpoint.check_ownership(locked_primary_hashes) last_event, raw_events = task_run_batch_query( filter=eventstore.Filter(project_ids=[args.project_id], group_ids=[source.id]), batch_size=args.batch_size, @@ -612,20 +697,32 @@ def unmerge(*posargs: Any, **kwargs: Any) -> None: # Log info related to this unmerge logger.info("unmerge.check", extra={**extra, "num_events": len(events)}) + # Fetching event payloads is read-only and can be repeated after a worker + # death. Do not truncate the source before that recovery boundary. + if isinstance(args, InitialUnmergeArgs): + if checkpoint: + checkpoint.begin_mutations() + truncate_denormalizations(project, source) + # If there are no more events to process, we're done with the migration. if not events: publish_destination_first_seen( project, [destination_id for destination_id, _ in args.destinations.values()] ) - unlock_hashes(args.project_id, locked_primary_hashes) + if checkpoint is None: + unlock_hashes(args.project_id, locked_primary_hashes) for unmerge_key, (_, eventstream_state) in args.destinations.items(): logger.warning( "Unmerge complete (eventstream state: %s)", eventstream_state, extra=extra ) if eventstream_state: args.replacement.stop_snuba_replacement(eventstream_state) + if checkpoint: + checkpoint.release_locks() return + if checkpoint and isinstance(args, SuccessiveUnmergeArgs): + checkpoint.begin_mutations() source_events = [] destination_events: dict[str, list[GroupEvent]] = {} @@ -689,8 +786,13 @@ def unmerge(*posargs: Any, **kwargs: Any) -> None: destinations=destinations, locked_primary_hashes=locked_primary_hashes, source_fields_reset=source_fields_reset, + operation_id=args.operation_id, + batch_number=args.batch_number + 1, ) + if checkpoint: + checkpoint.handoff(new_args.dump_arguments()) + return unmerge.delay(**new_args.dump_arguments()) # Record that this activation has spawned its continuation. A subsequent re-pend of this same # activation will short-circuit at the guard above instead of spawning again. diff --git a/src/sentry/testutils/factories.py b/src/sentry/testutils/factories.py index 9bbadef9a38a..f288516a1cfd 100644 --- a/src/sentry/testutils/factories.py +++ b/src/sentry/testutils/factories.py @@ -122,6 +122,7 @@ from sentry.models.files.control_file import ControlFile from sentry.models.files.file import File from sentry.models.group import Group +from sentry.models.grouphash import GroupHash from sentry.models.grouphistory import GroupHistory from sentry.models.grouplink import GroupLink from sentry.models.groupopenperiod import GroupOpenPeriod @@ -1603,6 +1604,11 @@ def create_group(project: Project, create_open_period: bool = True, **kwargs: An return group + @staticmethod + @assume_test_silo_mode(SiloMode.CELL) + def create_group_hash(group: Group, hash: str, **kwargs: Any) -> GroupHash: + return GroupHash.objects.create(project=group.project, group=group, hash=hash, **kwargs) + @staticmethod @assume_test_silo_mode(SiloMode.CELL) def create_group_activity(group: Group, *args: Any, **kwargs: Any) -> Activity: diff --git a/src/sentry/unmerge.py b/src/sentry/unmerge.py index b16b09c495b1..5b1bf8441255 100644 --- a/src/sentry/unmerge.py +++ b/src/sentry/unmerge.py @@ -30,6 +30,7 @@ class UnmergeReplacement(abc.ABC): def parse_arguments(fingerprints: Any = None, replacement: Any = None) -> "UnmergeReplacement": if replacement is not None: if isinstance(replacement, dict): + replacement = dict(replacement) replacement = _REPLACEMENT_TYPE_LABELS.get_key(replacement.pop("type"))( **replacement ) @@ -153,6 +154,8 @@ class UnmergeArgsBase(abc.ABC): replacement: UnmergeReplacement actor_id: int | None batch_size: int + operation_id: str | None = dataclasses.field(default=None, kw_only=True) + batch_number: int = dataclasses.field(default=0, kw_only=True) @staticmethod def parse_arguments( @@ -162,12 +165,14 @@ def parse_arguments( fingerprints: Sequence[str], actor_id: int | None, last_event: Mapping[str, Any] | None = None, - batch_size: int = 500, + batch_size: int = 250, source_fields_reset: bool = False, eventstream_state: EventstreamState | None = None, replacement: UnmergeReplacement | None = None, locked_primary_hashes: Collection[str] | None = None, destinations: Destinations | None = None, + operation_id: str | None = None, + batch_number: int = 0, ) -> "UnmergeArgs": if destinations is None: if destination_id is not None: @@ -186,6 +191,8 @@ def parse_arguments( actor_id=actor_id, batch_size=batch_size, destinations=destinations, + operation_id=operation_id, + batch_number=batch_number, ) else: assert locked_primary_hashes is not None or fingerprints is not None @@ -199,10 +206,15 @@ def parse_arguments( destinations=destinations, locked_primary_hashes=locked_primary_hashes or fingerprints or [], source_fields_reset=source_fields_reset, + operation_id=operation_id, + batch_number=batch_number, ) def dump_arguments(self) -> Mapping[str, Any]: rv = dataclasses.asdict(self) + if self.operation_id is None: + for key in ("operation_id", "batch_number"): + del rv[key] rv["fingerprints"] = None rv["destination_id"] = None rv["replacement"]["type"] = _REPLACEMENT_TYPE_LABELS[type(self.replacement)] diff --git a/tests/sentry/issues/endpoints/test_group_hashes.py b/tests/sentry/issues/endpoints/test_group_hashes.py index 95925c7ba3e7..dc8089b83623 100644 --- a/tests/sentry/issues/endpoints/test_group_hashes.py +++ b/tests/sentry/issues/endpoints/test_group_hashes.py @@ -9,6 +9,7 @@ from sentry.issues.endpoints.group_hashes import GroupHashesEndpoint from sentry.models.grouphash import GroupHash from sentry.testutils.cases import APITestCase, SnubaTestCase +from sentry.testutils.factories import Factories from sentry.testutils.helpers.datetime import before_now from sentry.testutils.helpers.features import with_feature from sentry.viewer_context import Actor, ActorType, ViewerContext, encode_viewer_context @@ -234,6 +235,20 @@ def test_unmerge(self) -> None: tags={"platform": "javascript", "sdk": "sentry.javascript.nextjs"}, ) + @with_feature("organizations:unmerge-recovery") + def test_unmerge_assigns_operation_before_enqueue(self) -> None: + self.login_as(user=self.user) + group = self.create_group() + fingerprint = "a" * 32 + Factories.create_group_hash(group=group, hash=fingerprint) + with patch("sentry.issues.endpoints.group_hashes.start_unmerge.delay") as enqueue: + response = self.client.put( + f"/api/0/organizations/{self.organization.slug}/issues/{group.id}/hashes/?id={fingerprint}", + format="json", + ) + assert response.status_code == 202 + assert len(enqueue.call_args.kwargs["operation_id"]) == 32 + def test_unmerge_put_member(self) -> None: member_user = self.create_user(is_superuser=False) member = self.create_member(organization=self.organization, user=member_user, role="member") diff --git a/tests/sentry/test_unmerge.py b/tests/sentry/test_unmerge.py index ca4ff9887b94..48780a9046a8 100644 --- a/tests/sentry/test_unmerge.py +++ b/tests/sentry/test_unmerge.py @@ -1,3 +1,25 @@ +import uuid +from dataclasses import replace +from typing import Any +from unittest.mock import Mock, patch + +import pytest +from redis.exceptions import ConnectionError as RedisConnectionError + +from sentry.db.models.manager.base_query_set import BaseQuerySet +from sentry.issues.unmerge_recovery import ( + UnmergeBusy, + UnmergeCheckpoint, + UnmergeNeedsRecovery, + client, + owner_key, + release_hashes, + reserve_legacy_hashes, +) +from sentry.models.grouphash import GroupHash +from sentry.tasks.unmerge import lock_hashes, start_unmerge, unmerge +from sentry.testutils.cases import TestCase +from sentry.testutils.factories import Factories from sentry.unmerge import ( InitialUnmergeArgs, PrimaryHashUnmergeReplacement, @@ -19,13 +41,13 @@ def test_argument_parsing_endpoint() -> None: fingerprints=["a" * 32], ), actor_id=None, - batch_size=500, + batch_size=250, ) dumped = args.dump_arguments() assert dumped == { "actor_id": None, - "batch_size": 500, + "batch_size": 250, "destination_id": None, "destinations": {}, "fingerprints": None, @@ -75,3 +97,423 @@ def test_argument_parsing_page2() -> None: ) assert UnmergeArgsBase.parse_arguments(**args.dump_arguments()) == args + + +class UnmergeRecoveryTest(TestCase): + def setUp(self) -> None: + super().setUp() + self.source = self.create_group(project=self.project) + self.fingerprint = "a" * 32 + self.group_hash = Factories.create_group_hash(group=self.source, hash=self.fingerprint) + self.args = InitialUnmergeArgs( + project_id=self.project.id, + source_id=self.source.id, + replacement=PrimaryHashUnmergeReplacement(fingerprints=[self.fingerprint]), + actor_id=None, + batch_size=500, + destinations={}, + operation_id=uuid.uuid4().hex, + ) + + def test_recovery_arguments_round_trip(self) -> None: + arguments = self.args.dump_arguments() + assert UnmergeArgsBase.parse_arguments(**arguments) == self.args + assert UnmergeArgsBase.parse_arguments(**arguments) == self.args + + def test_lock_intent_survives_transaction_rollback(self) -> None: + queryset_type = type(GroupHash.objects.all()) + original_update = queryset_type.update + + def interrupted_update( + queryset: BaseQuerySet[GroupHash, GroupHash], **kwargs: object + ) -> int: + original_update(queryset, **kwargs) + raise KeyboardInterrupt() + + with UnmergeCheckpoint(self.args) as checkpoint: + with ( + patch.object(queryset_type, "update", interrupted_update), + pytest.raises(KeyboardInterrupt), + ): + checkpoint.lock_hashes() + assert checkpoint.phase == "reading" + + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.UNLOCKED + with UnmergeCheckpoint(self.args) as redelivery: + assert redelivery.lock_hashes() == [self.fingerprint] + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.LOCKED_IN_MIGRATION + + def test_hard_death_after_sql_commit(self) -> None: + # A killed worker cannot release its lease. Simulate its expiry. + killed = UnmergeCheckpoint(self.args).__enter__() + assert killed.lock_hashes() == [self.fingerprint] + with pytest.raises(UnmergeBusy): + UnmergeCheckpoint(self.args).__enter__() + client().delete(killed.lease_key) + with UnmergeCheckpoint(self.args) as redelivery: + assert redelivery.lock_hashes() == [self.fingerprint] + with pytest.raises(UnmergeBusy): + killed.begin_mutations() + killed.__exit__(None, None, None) + assert client().get(redelivery.lease_key) == redelivery.token + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_ownership_conflict_rejects_the_whole_request(self, query: Mock) -> None: + other_hash = "b" * 32 + Factories.create_group_hash(group=self.source, hash=other_hash) + client().set(owner_key(self.project.id, other_hash), "another-operation", ex=600) + arguments = replace( + self.args, + replacement=PrimaryHashUnmergeReplacement(fingerprints=[self.fingerprint, other_hash]), + ).dump_arguments() + + with pytest.raises(UnmergeNeedsRecovery, match="another unmerge"): + unmerge(**arguments) + query.assert_not_called() + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.UNLOCKED + assert client().get(owner_key(self.project.id, self.fingerprint)) is None + assert client().get(owner_key(self.project.id, other_hash)) == "another-operation" + + def test_legacy_and_recoverable_tasks_respect_ownership(self) -> None: + with reserve_legacy_hashes(self.project.id, [self.fingerprint]): + with UnmergeCheckpoint(self.args) as checkpoint: + with pytest.raises(UnmergeNeedsRecovery): + checkpoint.lock_hashes() + with UnmergeCheckpoint(self.args) as checkpoint: + assert checkpoint.lock_hashes() == [self.fingerprint] + with pytest.raises(UnmergeNeedsRecovery): + lock_hashes(self.project.id, self.source.id, [self.fingerprint]) + + @patch("sentry.tasks.unmerge.task_run_batch_query", return_value=(None, [])) + def test_legacy_release_failure_does_not_strand_sql_locks(self, query: Mock) -> None: + with ( + patch( + "sentry.issues.unmerge_recovery.release_hashes", + side_effect=RedisConnectionError("reservation cleanup interrupted"), + ), + patch("sentry.tasks.unmerge.truncate_denormalizations"), + ): + unmerge(self.project.id, self.source.id, None, [self.fingerprint], None) + query.assert_called_once() + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.UNLOCKED + assert client().ttl(owner_key(self.project.id, self.fingerprint)) > 0 + + def test_locked_hash_without_checkpoint_is_not_adopted(self) -> None: + self.group_hash.update(state=GroupHash.State.LOCKED_IN_MIGRATION) + with UnmergeCheckpoint(self.args) as redelivery, pytest.raises(UnmergeNeedsRecovery): + redelivery.lock_hashes() + assert client().get(owner_key(self.project.id, self.fingerprint)) is None + + @patch("sentry.tasks.unmerge.task_run_batch_query") + @patch("sentry.tasks.unmerge.truncate_denormalizations") + def test_legacy_redelivery_with_no_owned_hashes_does_not_reset_source( + self, truncate: Mock, query: Mock + ) -> None: + self.group_hash.update(state=GroupHash.State.LOCKED_IN_MIGRATION) + unmerge(self.project.id, self.source.id, None, [self.fingerprint], None) + truncate.assert_not_called() + query.assert_not_called() + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_killswitch_releases_recovered_initial_locks(self, query: Mock) -> None: + with UnmergeCheckpoint(self.args) as checkpoint: + checkpoint.lock_hashes() + with self.options({"unmerge.killswitch-projects": [{"project_id": self.project.id}]}): + unmerge(**self.args.dump_arguments()) + query.assert_not_called() + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.UNLOCKED + assert client().get(owner_key(self.project.id, self.fingerprint)) is None + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_moved_hash_releases_ownership_and_retires_request(self, query: Mock) -> None: + self.group_hash.update(group=self.create_group(project=self.project)) + with pytest.raises(UnmergeNeedsRecovery, match="no longer belong"): + unmerge(**self.args.dump_arguments()) + self.assert_cleanup_complete() + unmerge(**self.args.dump_arguments()) + query.assert_not_called() + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_moved_hash_releases_previously_acquired_sql_lock(self, query: Mock) -> None: + with UnmergeCheckpoint(self.args) as checkpoint: + checkpoint.lock_hashes() + self.group_hash.update(group=self.create_group(project=self.project)) + with pytest.raises(UnmergeNeedsRecovery, match="no longer belong"): + unmerge(**self.args.dump_arguments()) + self.assert_cleanup_complete() + query.assert_not_called() + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_killswitch_does_not_unlock_an_unowned_sql_lock(self, query: Mock) -> None: + self.group_hash.update(state=GroupHash.State.LOCKED_IN_MIGRATION) + client().set(owner_key(self.project.id, self.fingerprint), "another-operation", ex=600) + with self.options({"unmerge.killswitch-projects": [{"project_id": self.project.id}]}): + unmerge(**self.args.dump_arguments()) + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.LOCKED_IN_MIGRATION + assert client().get(owner_key(self.project.id, self.fingerprint)) == "another-operation" + query.assert_not_called() + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_deleted_hash_releases_previously_acquired_ownership(self, query: Mock) -> None: + with UnmergeCheckpoint(self.args) as checkpoint: + checkpoint.lock_hashes() + self.group_hash.delete() + with pytest.raises(UnmergeNeedsRecovery, match="no longer belong"): + unmerge(**self.args.dump_arguments()) + assert client().get(owner_key(self.project.id, self.fingerprint)) is None + with UnmergeCheckpoint(self.args) as checkpoint: + assert checkpoint.finished + unmerge(**self.args.dump_arguments()) + query.assert_not_called() + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_killswitch_cleanup_resumes_before_sql_unlock(self, query: Mock) -> None: + with UnmergeCheckpoint(self.args) as checkpoint: + checkpoint.lock_hashes() + with ( + self.options({"unmerge.killswitch-projects": [{"project_id": self.project.id}]}), + patch.object(type(GroupHash.objects.all()), "update", side_effect=KeyboardInterrupt), + pytest.raises(KeyboardInterrupt), + ): + unmerge(**self.args.dump_arguments()) + # Cleanup must continue even if the killswitch has since been disabled. + unmerge(**self.args.dump_arguments()) + self.assert_cleanup_complete() + query.assert_not_called() + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_killswitch_cleanup_resumes_after_sql_unlock(self, query: Mock) -> None: + with UnmergeCheckpoint(self.args) as checkpoint: + checkpoint.lock_hashes() + original_save = UnmergeCheckpoint.save + + def interrupted_save(checkpoint: UnmergeCheckpoint, phase: str, **updates: object) -> None: + if phase == "releasing": + raise KeyboardInterrupt() + original_save(checkpoint, phase, **updates) + + with ( + self.options({"unmerge.killswitch-projects": [{"project_id": self.project.id}]}), + patch.object(UnmergeCheckpoint, "save", interrupted_save), + pytest.raises(KeyboardInterrupt), + ): + unmerge(**self.args.dump_arguments()) + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.UNLOCKED + unmerge(**self.args.dump_arguments()) + self.assert_cleanup_complete() + query.assert_not_called() + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_killswitch_cleanup_resumes_partial_release_without_unlocking_new_owner( + self, query: Mock + ) -> None: + other_hash = "b" * 32 + other_group_hash = Factories.create_group_hash(group=self.source, hash=other_hash) + args = replace( + self.args, + replacement=PrimaryHashUnmergeReplacement(fingerprints=[self.fingerprint, other_hash]), + ) + with UnmergeCheckpoint(args) as checkpoint: + checkpoint.lock_hashes() + + def interrupted_release(project_id: int, fingerprints: list[str], owner: str) -> None: + release_hashes(project_id, fingerprints[:1], owner) + raise KeyboardInterrupt() + + with ( + self.options({"unmerge.killswitch-projects": [{"project_id": self.project.id}]}), + patch("sentry.issues.unmerge_recovery.release_hashes", side_effect=interrupted_release), + pytest.raises(KeyboardInterrupt), + ): + unmerge(**args.dump_arguments()) + with UnmergeCheckpoint(replace(self.args, operation_id=uuid.uuid4().hex)) as newer: + newer.lock_hashes() + newer_owner = newer.owner + unmerge(**args.dump_arguments()) + other_group_hash.refresh_from_db() + assert other_group_hash.state == GroupHash.State.UNLOCKED + assert client().get(owner_key(self.project.id, other_hash)) is None + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.LOCKED_IN_MIGRATION + assert client().get(owner_key(self.project.id, self.fingerprint)) == newer_owner + with UnmergeCheckpoint(args) as checkpoint: + assert checkpoint.finished + query.assert_not_called() + + def assert_cleanup_complete(self) -> None: + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.UNLOCKED + assert client().get(owner_key(self.project.id, self.fingerprint)) is None + with UnmergeCheckpoint(self.args) as checkpoint: + assert checkpoint.phase == "complete" + + @patch("sentry.tasks.unmerge.truncate_denormalizations") + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_read_failure_resumes_owned_hashes_before_truncating( + self, query: Mock, truncate: Mock + ) -> None: + query.side_effect = MemoryError("event hydration interrupted") + with pytest.raises(UnmergeBusy): + unmerge(**self.args.dump_arguments()) + truncate.assert_not_called() + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.LOCKED_IN_MIGRATION + + query.side_effect = None + query.return_value = (None, []) + unmerge(**self.args.dump_arguments()) + assert query.call_args.kwargs["batch_size"] == 500 + truncate.assert_called_once() + self.group_hash.refresh_from_db() + assert self.group_hash.state == GroupHash.State.UNLOCKED + assert client().get(owner_key(self.project.id, self.fingerprint)) is None + + query.reset_mock() + unmerge(**self.args.dump_arguments()) + query.assert_not_called() + + @patch("sentry.tasks.unmerge.task_run_batch_query") + def test_later_read_failure_resumes_its_cursor(self, query: Mock) -> None: + arguments = { + **self.args.dump_arguments(), + "batch_number": 1, + "last_event": {"timestamp": "old-cursor", "event_id": "b" * 32}, + "locked_primary_hashes": [self.fingerprint], + } + with UnmergeCheckpoint(self.args) as parent: + parent.lock_hashes() + parent.begin_mutations() + parent.handoff(arguments) + query.side_effect = MemoryError() + with patch("sentry.tasks.unmerge.truncate_denormalizations") as truncate: + with pytest.raises(UnmergeBusy): + unmerge(**arguments) + query.side_effect = None + query.return_value = (None, []) + unmerge(**arguments) + truncate.assert_not_called() + assert query.call_count == 2 + assert query.call_args.kwargs["state"] == arguments["last_event"] + + def prepare_successive_batch(self) -> dict[str, Any]: + arguments = { + **self.args.dump_arguments(), + "batch_number": 1, + "last_event": {"timestamp": "old-cursor", "event_id": "b" * 32}, + "locked_primary_hashes": [self.fingerprint], + } + with UnmergeCheckpoint(self.args) as parent: + parent.lock_hashes() + parent.begin_mutations() + parent.handoff(arguments) + return arguments + + @patch("sentry.tasks.unmerge.truncate_denormalizations") + @patch("sentry.tasks.unmerge.task_run_batch_query", return_value=(None, [])) + def test_empty_final_batch_retries_before_lock_cleanup( + self, query: Mock, truncate: Mock + ) -> None: + arguments = self.prepare_successive_batch() + with ( + patch( + "sentry.tasks.unmerge.publish_destination_first_seen", + side_effect=RedisConnectionError("completion interrupted"), + ), + pytest.raises(UnmergeBusy), + ): + unmerge(**arguments) + unmerge(**arguments) + assert query.call_count == 2 + truncate.assert_not_called() + self.assert_cleanup_complete() + + @patch("sentry.tasks.unmerge.task_run_batch_query", return_value=(None, [])) + def test_empty_final_batch_resumes_sql_unlock(self, query: Mock) -> None: + arguments = self.prepare_successive_batch() + with ( + patch.object(type(GroupHash.objects.all()), "update", side_effect=KeyboardInterrupt), + pytest.raises(KeyboardInterrupt), + ): + unmerge(**arguments) + unmerge(**arguments) + query.assert_called_once() + self.assert_cleanup_complete() + + @patch("sentry.tasks.unmerge.task_run_batch_query", return_value=(None, [])) + def test_empty_final_batch_resumes_redis_release(self, query: Mock) -> None: + arguments = self.prepare_successive_batch() + with ( + patch( + "sentry.issues.unmerge_recovery.release_hashes", + side_effect=RedisConnectionError("owner release interrupted"), + ), + pytest.raises(UnmergeBusy), + ): + unmerge(**arguments) + unmerge(**arguments) + query.assert_called_once() + self.assert_cleanup_complete() + + @patch("sentry.tasks.unmerge.truncate_denormalizations") + @patch("sentry.tasks.unmerge.task_run_batch_query", return_value=(None, [])) + def test_interrupted_mutations_are_not_replayed(self, query: Mock, truncate: Mock) -> None: + truncate.side_effect = RuntimeError("partial aggregate changes") + with pytest.raises(RuntimeError, match="partial aggregate"): + unmerge(**self.args.dump_arguments()) + query.reset_mock() + with pytest.raises(UnmergeNeedsRecovery, match="partially applied"): + unmerge(**self.args.dump_arguments()) + query.assert_not_called() + truncate.assert_called_once() + + @patch.object(unmerge, "apply_async_with_future") + def test_handoff_retries_delivery_without_reapplying_batch(self, send: Mock) -> None: + next_arguments = {**self.args.dump_arguments(), "batch_number": 1} + + def apply_batch(args: object, checkpoint: UnmergeCheckpoint) -> None: + checkpoint.begin_mutations() + checkpoint.handoff(next_arguments) + raise KeyboardInterrupt() + + with patch("sentry.tasks.unmerge._unmerge", side_effect=apply_batch) as apply: + with pytest.raises(KeyboardInterrupt): + unmerge(**self.args.dump_arguments()) + send.assert_not_called() + send.return_value.result.side_effect = TimeoutError("delivery unknown") + with pytest.raises(UnmergeBusy): + unmerge(**self.args.dump_arguments()) + send.return_value.result.side_effect = None + unmerge(**self.args.dump_arguments()) + + with UnmergeCheckpoint(replace(self.args, batch_number=1)) as child: + child.complete() + unmerge(**self.args.dump_arguments()) + unmerge(**next_arguments) + apply.assert_called_once() + assert send.call_count == 2 + send.return_value.result.assert_called_with(timeout=30) + + @patch.object(unmerge, "apply_async_with_future") + def test_starter_resends_same_operation_when_delivery_is_unknown(self, send: Mock) -> None: + send.return_value.result.side_effect = TimeoutError() + with pytest.raises(UnmergeBusy): + start_unmerge( + self.project.id, + self.source.id, + None, + [self.fingerprint], + None, + operation_id=self.args.operation_id, + ) + assert send.call_args.kwargs["kwargs"]["operation_id"] == self.args.operation_id + + def test_missing_checkpoint_cannot_resume_a_later_batch(self) -> None: + with pytest.raises(UnmergeNeedsRecovery): + UnmergeCheckpoint(replace(self.args, batch_number=1)).__enter__() diff --git a/tests/snuba/tasks/test_unmerge.py b/tests/snuba/tasks/test_unmerge.py index 241f188e465b..30836910f80a 100644 --- a/tests/snuba/tasks/test_unmerge.py +++ b/tests/snuba/tasks/test_unmerge.py @@ -49,6 +49,7 @@ from sentry.testutils.outbox import outbox_runner from sentry.tsdb.base import TSDBModel from sentry.utils import redis +from sentry.utils.query import task_run_batch_query # Use the default redis client as a cluster client in the similarity index index = _make_index_backend(redis.redis_clusters.get_binary("default")) @@ -210,6 +211,23 @@ def test_get_group_backfill_attributes(self) -> None: @with_feature("projects:similarity-indexing") def test_unmerge(self) -> None: + self._assert_unmerge({}) + + @with_feature("projects:similarity-indexing") + def test_unmerge_with_recovery_and_small_batches(self) -> None: + with patch( + "sentry.tasks.unmerge.task_run_batch_query", wraps=task_run_batch_query + ) as query: + self._assert_unmerge( + { + "operation_id": uuid.uuid4().hex, + }, + batch_size=2, + ) + assert len(query.call_args_list) > 5 + assert {call.kwargs["batch_size"] for call in query.call_args_list} == {2} + + def _assert_unmerge(self, operation_kwargs: Mapping[str, Any], batch_size: int = 5) -> None: # Replace second=0 to ensure all 17 events (now+0s to now+17s) # stay within the same hour bucket. Without this, if now is within # 17 seconds of an hour boundary, events can cross into the next @@ -330,7 +348,13 @@ def create_message_event( with self.tasks(): unmerge.delay( - project.id, source.id, destination.id, [list(events.keys())[0]], None, batch_size=5 + project.id, + source.id, + destination.id, + [list(events.keys())[0]], + None, + batch_size=batch_size, + **operation_kwargs, ) assert ( @@ -768,10 +792,9 @@ def test_delegates_to_unmerge(self) -> None: destination_id=None, fingerprints=["abc", "def"], actor_id=3, - batch_size=500, ) - mock_delay.assert_called_once_with(1, 2, None, ["abc", "def"], 3, batch_size=500) + mock_delay.assert_called_once_with(1, 2, None, ["abc", "def"], 3, batch_size=250) class PublishDestinationFirstSeenTest(TestCase):