diff --git a/migrations_lockfile.txt b/migrations_lockfile.txt index 026db7990b72..79ba6647f0e6 100644 --- a/migrations_lockfile.txt +++ b/migrations_lockfile.txt @@ -31,7 +31,7 @@ replays: 0001_squashed_0007_organizationmember_replay_access seer: 0027_add_seer_run_coding_agent_handoff -sentry: 1140_organizationcontributors_unique_provider_hostname +sentry: 1141_add_generated_at_to_group_derived_data social_auth: 0001_squashed_0003_social_auth_json_field diff --git a/src/sentry/issues/derived/processing.py b/src/sentry/issues/derived/processing.py index 8c501a3e1f3e..418af645e1c9 100644 --- a/src/sentry/issues/derived/processing.py +++ b/src/sentry/issues/derived/processing.py @@ -2,19 +2,22 @@ import logging import time from datetime import datetime, timedelta +from typing import NamedTuple from django.core.exceptions import ObjectDoesNotExist from django.db import IntegrityError, router, transaction from django.db.models import Q +from django.utils import timezone from sentry.issues.derived.aggregators import AGGREGATORS from sentry.issues.derived.framework import Pipeline from sentry.issues.derived.store import GroupDerivedDataStore -from sentry.issues.derived.tasks import process_group_log_task +from sentry.issues.derived.tasks import generate_group_derived_data, process_group_log_task from sentry.issues.models.groupactionlogentry import GroupActionLogEntry from sentry.issues.models.groupderiveddata import EPOCH, GroupDerivedData from sentry.models.group import Group from sentry.utils import metrics +from sentry.workflow_engine.caches.mapping import CacheMapping logger = logging.getLogger(__name__) @@ -23,6 +26,30 @@ DEFAULT_BATCH_SIZE = 1000 INLINE_BATCH_SIZE = 100 +# Fields that constitute the derived state, written by promote_to_live. +# Derived by excluding identity, control, and auto-managed fields from the +# model — new columns are automatically included unless explicitly excluded. +_EXCLUDED_FIELDS = frozenset({"id", "group_id", "date_added", "date_updated"}) +_STATE_FIELDS = tuple( + f.attname for f in GroupDerivedData._meta.concrete_fields if f.attname not in _EXCLUDED_FIELDS +) + + +class GenerationId(NamedTuple): + """Uniquely identifies a generation attempt for a group.""" + + group_id: int + generated_at: datetime # when this generation started; reflects log state observed + pipeline_hash: str + + +# Cache for in-progress generation state. +_generation_cache = CacheMapping[GenerationId, GroupDerivedData]( + lambda k: f"{k.group_id}:{k.generated_at.isoformat()}:{k.pipeline_hash}", + namespace="gdd-generation", + ttl_seconds=86400, +) + class ProcessingStrategy(enum.Enum): SYNC = "sync" # process all pending actions now @@ -40,6 +67,9 @@ def _ensure_derived(group_id: int, pipeline_hash: str) -> GroupDerivedData: except GroupDerivedData.DoesNotExist: pass + if not Group.objects.filter(id=group_id).exists(): + raise Group.DoesNotExist(f"Group {group_id} does not exist") + try: derived, _created = GroupDerivedData.objects.get_or_create( group_id=group_id, @@ -51,6 +81,7 @@ def _ensure_derived(group_id: int, pipeline_hash: str) -> GroupDerivedData: }, ) except IntegrityError: + # The group was deleted between our exists() check and the INSERT. raise Group.DoesNotExist(f"Group {group_id} does not exist") return derived @@ -68,15 +99,12 @@ def _entries_after_cursor( ) -def _cursor_lte(cursor_date: datetime, cursor_id: int) -> Q: - return Q(cursor_date__lt=cursor_date) | Q(cursor_date=cursor_date, cursor_id__lte=cursor_id) - - def _process_batch( p: Pipeline[GroupActionLogEntry], derived: GroupDerivedData, - group_id: int, batch_size: int, + *, + persist: bool = True, ) -> bool: """ Process up to `batch_size` entries for a group. Updates derived in place. @@ -87,7 +115,7 @@ def _process_batch( 1. The action log is append-only and the pipeline is deterministic, so any caller processing the same entries produces the same result. - 2. The UPDATE uses a cursor guard (_cursor_lte) that only succeeds if no + 2. The UPDATE uses a cursor guard that only succeeds if no other caller has already advanced the cursor past our batch. If it fails (updated == 0), a concurrent caller already wrote a superset of our work, so we refresh and check if more remains. @@ -95,7 +123,13 @@ def _process_batch( This is an optimistic concurrency scheme — no locks are held, and the last-writer-wins semantics are safe because all writers compute the same deterministic result for overlapping entry ranges. + + When *persist* is False, only the in-memory object is updated — the + caller is responsible for persisting the result (e.g. via + ``promote_to_live``). Used for full generations that accumulate + state in memory and write once at the end. """ + group_id = derived.group_id entries = _entries_after_cursor(group_id, derived.cursor_date, derived.cursor_id, batch_size) if not entries: @@ -108,9 +142,15 @@ def _process_batch( last_id = last.id state_update = GroupDerivedDataStore.build_update(p, result) + if not persist: + derived.cursor_date = last_date + derived.cursor_id = last_id + GroupDerivedDataStore.apply_to_instance(derived, state_update) + return len(entries) == batch_size + updated = GroupDerivedData.objects.filter( - Q(group_id=group_id) - & _cursor_lte(last_date, last_id) + Q(id=derived.id, generated_at=derived.generated_at) + & (Q(cursor_date__lt=last_date) | Q(cursor_date=last_date, cursor_id__lte=last_id)) & Q(pipeline_hash=derived.pipeline_hash) ).update(cursor_date=last_date, cursor_id=last_id, **state_update) @@ -153,33 +193,69 @@ def _process_batch( class GroupLogTimeout(Exception): - """Raised when process_group_log cannot finish within its timeout.""" + """Raised when processing cannot finish within its time budget.""" + + def __init__(self, group_id: int, generation_id: GenerationId | None = None) -> None: + self.group_id = group_id + self.generation_id = generation_id + super().__init__(group_id) + + +DEFAULT_TIME_LIMIT = timedelta(seconds=8) + + +def _drain_log( + derived: GroupDerivedData, + batch_size: int = DEFAULT_BATCH_SIZE, + pipeline: Pipeline[GroupActionLogEntry] | None = None, + time_limit: timedelta = DEFAULT_TIME_LIMIT, + *, + persist: bool = True, +) -> bool: + """Process pending log entries into *derived*, batching as needed. + + Returns True if all entries were processed, False if the time limit was + reached and more entries remain. The limit is checked between batches, + so a single slow batch can exceed it. + + When *persist* is False, batches update only the in-memory object. + """ + deadline = time.monotonic() + time_limit.total_seconds() + p = pipeline or PIPELINE + while _process_batch(p, derived, batch_size, persist=persist): + if time.monotonic() >= deadline: + return False + return True + + +# --------------------------------------------------------------------------- +# Incremental processing (on event arrival) +# --------------------------------------------------------------------------- def process_group_log( group_id: int, batch_size: int = DEFAULT_BATCH_SIZE, - target_pipeline: Pipeline[GroupActionLogEntry] | None = None, + pipeline: Pipeline[GroupActionLogEntry] | None = None, timeout: timedelta | None = None, ) -> GroupDerivedData: - """Fully drain all pending entries for a group, processing in batches. + """Fully drain all pending entries for a group's row. Raises Group.DoesNotExist if the group has been deleted. Raises GroupLogTimeout if *timeout* elapses before all entries are processed. """ - p = target_pipeline or PIPELINE - timeout_seconds = timeout.total_seconds() if timeout is not None else None - start = time.monotonic() + p = pipeline or PIPELINE with transaction.atomic(using=router.db_for_write(GroupDerivedData)): derived = _ensure_derived(group_id, p.pipeline_hash) - has_more = _process_batch(p, derived, group_id, batch_size) - while has_more: - if timeout_seconds is not None and time.monotonic() - start >= timeout_seconds: + time_limit = timeout if timeout is not None else DEFAULT_TIME_LIMIT + drained = _drain_log(derived, batch_size, p, time_limit=time_limit) + if not drained: + if timeout is not None: raise GroupLogTimeout(group_id) - has_more = _process_batch(p, derived, group_id, batch_size) + process_group_log_task.delay(group_id) return derived @@ -187,7 +263,7 @@ def process_group_log( def trigger_group_log_processing(group_id: int, *, strategy: ProcessingStrategy) -> None: """Trigger derived data processing for a group. - Silently returns if the group has been deleted. + Silently returns if the group has been deleted or no row exists. Strategy controls how processing is dispatched: SYNC — process all pending actions now @@ -216,44 +292,241 @@ def trigger_group_log_processing(group_id: int, *, strategy: ProcessingStrategy) except ObjectDoesNotExist: return - has_more = _process_batch(pipeline, derived, group_id, INLINE_BATCH_SIZE) + has_more = _process_batch(pipeline, derived, INLINE_BATCH_SIZE) if has_more: - # Derived data will be stale for any code running between now and - # when the task completes. metrics.incr("issues.derived.inline_fallback_to_async") process_group_log_task.delay(group_id) -def invalidate_group_derived_data( +# --------------------------------------------------------------------------- +# Generation lifecycle: build in memory, upsert, cache partial progress +# --------------------------------------------------------------------------- + + +class PromotionResult(enum.Enum): + PROMOTED = "promoted" + SUPERSEDED = "superseded" # a newer generation already promoted + CURSOR_BEHIND = "cursor_behind" # same generation, but cursor is more advanced + + +def promote_to_live(candidate: GroupDerivedData) -> PromotionResult: + """Upsert the candidate's state into the row for its group. + + The UPDATE guard requires that ``candidate.generated_at`` is >= the + row's (newer generation wins) and the cursor is at or ahead. On + success, all state fields (including ``generated_at``) are stamped. + + Returns SUPERSEDED if the row has a newer ``generated_at``. + Returns CURSOR_BEHIND if the cursor guard failed. + + The candidate object itself is not persisted — it may be an unsaved + in-memory instance used only to carry the computed state. + """ + generated_at = candidate.generated_at + values = {f: getattr(candidate, f) for f in _STATE_FIELDS} + + updated = ( + GroupDerivedData.objects.filter( + group_id=candidate.group_id, + ) + .filter( + Q(generated_at__lt=generated_at) + | Q( + generated_at=generated_at, + cursor_date__lt=candidate.cursor_date, + ) + | Q( + generated_at=generated_at, + cursor_date=candidate.cursor_date, + cursor_id__lte=candidate.cursor_id, + ) + ) + .update(**values) + ) + + if updated: + return PromotionResult.PROMOTED + + # Check why we failed: row missing or newer generation? + row = ( + GroupDerivedData.objects.filter(group_id=candidate.group_id) + .values_list("id", "generated_at") + .first() + ) + + if row is None: + # Row doesn't exist — try to create it. + try: + with transaction.atomic(using=router.db_for_write(GroupDerivedData)): + GroupDerivedData.objects.create( + group_id=candidate.group_id, + **values, + ) + except IntegrityError: + return PromotionResult.CURSOR_BEHIND + return PromotionResult.PROMOTED + + _row_id, current_generated_at = row + if current_generated_at > generated_at: + return PromotionResult.SUPERSEDED + return PromotionResult.CURSOR_BEHIND + + +MAX_PROMOTION_ATTEMPTS = 5 + + +def build_and_promote_derived_data( group_id: int, - cursor: tuple[datetime, int] | None = None, + batch_size: int = DEFAULT_BATCH_SIZE, + generation_id: GenerationId | None = None, + time_limit: timedelta = DEFAULT_TIME_LIMIT, ) -> None: - """Delete derived state so it is rebuilt from scratch on the next pass, - then kicks off an async task to regenerate the derived data. + """Build derived data from scratch and upsert into the row. - If *cursor* is ``(date_added, id)`` of the earliest affected entry, the - row is only deleted when its cursor is at or past that point; otherwise - the mutation is still ahead of processing and no invalidation is needed. - Without a cursor the invalidation is unconditional. + Drains the full action log into an in-memory object, then upserts + via ``promote_to_live`` with a CAS on ``generated_at``. The + generation's ``generated_at`` is captured at start — it reflects the + log state observed, not when the generation finished. + + When *generation_id* is provided, previously cached partial progress + is loaded and resumed. + + Raises GroupLogTimeout (with ``generation_id`` set) if the time-limited + drain could not finish, so the caller can re-enqueue. """ - if cursor is None: - GroupDerivedData.objects.filter(group_id=group_id).delete() - process_group_log_task.delay(group_id) - return + pipeline_hash = PIPELINE.pipeline_hash + generated_at: datetime + + # Try to resume from cache. + derived: GroupDerivedData | None = None + if generation_id is not None: + derived = _generation_cache.get(generation_id) + generated_at = generation_id.generated_at + if derived is None: + logger.info( + "issues.derived.build_and_promote.cache_miss", + extra={"group_id": group_id, "generation_id": generation_id}, + ) - # Only invalidate if the row has already processed past the affected point. - cursor_date, cursor_id = cursor - deleted, _ = GroupDerivedData.objects.filter( - Q(group_id=group_id) - & (Q(cursor_date__gt=cursor_date) | Q(cursor_date=cursor_date, cursor_id__gte=cursor_id)), - ).delete() - if deleted: - logger.info( - "issues.derived.invalidated", + if derived is None: + generated_at = timezone.now() + derived = GroupDerivedData( + group_id=group_id, + generated_at=generated_at, + cursor_date=EPOCH, + cursor_id=0, + data={}, + pipeline_hash=pipeline_hash, + ) + + current_gen_id = GenerationId(group_id, generated_at, pipeline_hash) + + result = PromotionResult.CURSOR_BEHIND + attempt = 0 + for attempt in range(MAX_PROMOTION_ATTEMPTS): + drained = _drain_log(derived, batch_size, time_limit=time_limit, persist=False) + if not drained: + _generation_cache.set(current_gen_id, derived) + raise GroupLogTimeout(group_id, generation_id=current_gen_id) + + result = promote_to_live(derived) + if result is PromotionResult.PROMOTED: + logger.info( + "issues.derived.promoted", + extra={ + "group_id": group_id, + "cursor_date": str(derived.cursor_date), + "cursor_id": derived.cursor_id, + "attempts": attempt + 1, + }, + ) + _generation_cache.delete(current_gen_id) + return + + if result is PromotionResult.SUPERSEDED: + # A newer generation already won — retrying is futile. + break + + # CURSOR_BEHIND: the live row's cursor advanced past ours (via + # incremental processing). Load the live row's full state so the + # next _drain_log builds on the correct base, not stale aggregations. + live_row = GroupDerivedData.objects.filter(group_id=group_id).first() + if live_row is None: + break + for field in _STATE_FIELDS: + setattr(derived, field, getattr(live_row, field)) + # Keep our generated_at — the live row's may be older. + derived.generated_at = generated_at + + _generation_cache.delete(current_gen_id) + + if result is PromotionResult.CURSOR_BEHIND: + metrics.incr("issues.derived.promotion_exhausted", sample_rate=1.0) + logger.warning( + "issues.derived.promotion_exhausted", extra={ "group_id": group_id, - "cursor_date": str(cursor_date), - "cursor_id": cursor_id, + "attempts": attempt + 1, }, ) - process_group_log_task.delay(group_id) + + logger.info( + "issues.derived.promotion_failed", + extra={ + "group_id": group_id, + "result": result.value, + }, + ) + + +# --------------------------------------------------------------------------- +# Invalidation +# --------------------------------------------------------------------------- + + +def invalidate_group_derived_data( + group_id: int, + cursor: tuple[datetime, int] | None = None, + *, + hard_delete: bool = False, +) -> None: + """Invalidate derived state so it is regenerated. + + *hard_delete* controls the strategy: + + - ``False`` (default): enqueue a generation task. The existing row + continues serving reads until the generation promotes and stamps + a newer ``generated_at``. + - ``True``: delete the row immediately, then enqueue a generation + task to create a new one. Use when the existing data is known to + be wrong and must not be served. + + If *cursor* is ``(date_added, id)`` of the earliest affected entry, the + invalidation only fires when the row's cursor is at or past that + point; otherwise the mutation is still ahead of processing and no + invalidation is needed. *cursor* is only meaningful with + ``hard_delete=True``. + """ + if hard_delete: + if cursor is None: + GroupDerivedData.objects.filter(group_id=group_id).delete() + else: + cursor_date, cursor_id = cursor + deleted, _ = GroupDerivedData.objects.filter( + Q(group_id=group_id) + & ( + Q(cursor_date__gt=cursor_date) + | Q(cursor_date=cursor_date, cursor_id__gte=cursor_id) + ) + ).delete() + if not deleted: + return + logger.info( + "issues.derived.invalidated", + extra={ + "group_id": group_id, + "cursor_date": str(cursor_date), + "cursor_id": cursor_id, + }, + ) + generate_group_derived_data.delay(group_id) diff --git a/src/sentry/issues/derived/tasks.py b/src/sentry/issues/derived/tasks.py index 96600a29c26a..a7a4a0327913 100644 --- a/src/sentry/issues/derived/tasks.py +++ b/src/sentry/issues/derived/tasks.py @@ -11,7 +11,15 @@ BATCH_PROCESSING_DEADLINE = timedelta(seconds=30) # taskworker hard kill timeout BATCH_RETRIGGER_TIMEOUT = timedelta(seconds=20) # self-reschedule before the hard kill + _BATCH_TASK_KEY = "process_project_derived_data_batch" +_GENERATE_BATCH_TASK_KEY = "generate_project_derived_data_batch" +_GENERATE_GROUP_TASK_KEY = "generate_group_derived_data" + +# Cap self-rescheduling rebuilds to avoid infinite loops on very large groups. +_MAX_GENERATION_RUNS = 20 +# Hard limit on group IDs loaded per project-level task to bound memory. +_MAX_PROJECT_GROUPS = 10_000 @instrumented_task( @@ -30,6 +38,81 @@ def process_group_log_task(group_id: int, **kwargs: object) -> None: logger.info("process_group_log_task.group_not_found", extra={"group_id": group_id}) +@instrumented_task( + name="sentry.issues.derived.tasks.generate_group_derived_data", + namespace=issues_tasks, + silo_mode=SiloMode.CELL, +) +def generate_group_derived_data( + group_id: int, + resume_generated_at: str | None = None, + resume_pipeline_hash: str | None = None, + prior_runs: int = 0, + **kwargs: object, +) -> None: + """Generate derived data for a group by draining its action log.""" + from taskbroker_client.state import current_task + + from sentry.issues.derived.processing import ( + GenerationId, + GroupLogTimeout, + build_and_promote_derived_data, + ) + from sentry.taskworker.selfchain_idempotency import already_spawned, mark_spawned + + task_state = current_task() + activation_id = task_state.id if task_state else None + if activation_id and already_spawned(_GENERATE_GROUP_TASK_KEY, activation_id): + logger.info( + "generate_group_derived_data.duplicate_skipped", + extra={"group_id": group_id, "activation_id": activation_id}, + ) + metrics.incr( + "taskworker.selfchain.duplicate_skipped", + tags={"task": _GENERATE_GROUP_TASK_KEY}, + ) + return + + generation_id: GenerationId | None = None + if resume_generated_at is not None and resume_pipeline_hash is not None: + from datetime import datetime, timezone + + generation_id = GenerationId( + group_id, + datetime.fromisoformat(resume_generated_at).replace(tzinfo=timezone.utc), + resume_pipeline_hash, + ) + + from sentry.models.group import Group + + try: + build_and_promote_derived_data(group_id, generation_id=generation_id) + except Group.DoesNotExist: + logger.info("generate_group_derived_data.group_not_found", extra={"group_id": group_id}) + return + except GroupLogTimeout as e: + if prior_runs + 1 >= _MAX_GENERATION_RUNS: + logger.error( + "generate_group_derived_data.max_runs_exceeded", + extra={ + "group_id": group_id, + "generation_id": e.generation_id, + "prior_runs": prior_runs + 1, + }, + ) + metrics.incr("issues.derived.generate_max_runs_exceeded", sample_rate=1.0) + return + gen_id = e.generation_id + generate_group_derived_data.delay( + group_id, + resume_generated_at=gen_id.generated_at.isoformat() if gen_id else None, + resume_pipeline_hash=gen_id.pipeline_hash if gen_id else None, + prior_runs=prior_runs + 1, + ) + if activation_id: + mark_spawned(_GENERATE_GROUP_TASK_KEY, activation_id) + + @instrumented_task( name="sentry.issues.derived.tasks.process_project_derived_data", namespace=issues_tasks, @@ -50,11 +133,12 @@ def process_project_derived_data(project_id: int, **kwargs: object) -> None: batch_size = options.get("issues.derived.project-batch-size") max_tasks = options.get("issues.derived.project-max-tasks") + # TODO: support very large projects via paginated iteration group_ids = list( Group.objects.filter(project_id=project_id) .exclude(Exists(GroupDerivedData.objects.filter(group_id=OuterRef("id")))) .order_by("id") - .values_list("id", flat=True) + .values_list("id", flat=True)[:_MAX_PROJECT_GROUPS] ) if not group_ids: @@ -64,6 +148,15 @@ def process_project_derived_data(project_id: int, **kwargs: object) -> None: ) return + if len(group_ids) >= _MAX_PROJECT_GROUPS: + logger.error( + "process_project_derived_data.too_many_groups", + extra={ + "project_id": project_id, + "limit": _MAX_PROJECT_GROUPS, + }, + ) + starts = [group_ids[i] for i in range(0, len(group_ids), batch_size)] ends = starts[1:] + [group_ids[-1] + 1] ranges = list(zip(starts, ends)) @@ -145,7 +238,7 @@ def process_project_derived_data_batch( rescheduled = False for group_id in group_ids: - remaining = timedelta(seconds=timeout_seconds - (time.monotonic() - start)) + remaining = timedelta(seconds=max(0, timeout_seconds - (time.monotonic() - start))) try: process_group_log(group_id, timeout=remaining) processed += 1 @@ -203,3 +296,187 @@ def process_project_derived_data_batch( "elapsed": time.monotonic() - start, }, ) + + +# --------------------------------------------------------------------------- +# Project-level generation: build-and-promote for all groups +# --------------------------------------------------------------------------- + + +@instrumented_task( + name="sentry.issues.derived.tasks.generate_project_derived_data", + namespace=issues_tasks, + silo_mode=SiloMode.CELL, +) +def generate_project_derived_data(project_id: int, **kwargs: object) -> None: + """Generate derived data for every group in a project. + + Partitions groups into ID ranges and fans out a batch task for each + range. Each batch calls ``build_and_promote_derived_data`` per group, + which replaces existing rows via CAS while they continue serving reads. + """ + from sentry import options + from sentry.models.group import Group + + batch_size = options.get("issues.derived.project-batch-size") + max_tasks = options.get("issues.derived.project-max-tasks") + + # TODO: support very large projects via paginated iteration + group_ids = list( + Group.objects.filter(project_id=project_id) + .order_by("id") + .values_list("id", flat=True)[:_MAX_PROJECT_GROUPS] + ) + + if not group_ids: + return + + if len(group_ids) >= _MAX_PROJECT_GROUPS: + logger.error( + "generate_project_derived_data.too_many_groups", + extra={ + "project_id": project_id, + "limit": _MAX_PROJECT_GROUPS, + }, + ) + + starts = [group_ids[i] for i in range(0, len(group_ids), batch_size)] + ends = starts[1:] + [group_ids[-1] + 1] + ranges = list(zip(starts, ends)) + + if len(ranges) > max_tasks: + logger.error( + "generate_project_derived_data.too_many_tasks", + extra={ + "project_id": project_id, + "task_count": len(ranges), + "max_tasks": max_tasks, + }, + ) + return + + for start, end in ranges: + generate_project_derived_data_batch.delay( + project_id=project_id, + group_id_start=start, + group_id_end=end, + ) + + logger.info( + "generate_project_derived_data.scheduled", + extra={ + "project_id": project_id, + "group_count": len(group_ids), + "task_count": len(ranges), + }, + ) + + +@instrumented_task( + name="sentry.issues.derived.tasks.generate_project_derived_data_batch", + namespace=issues_tasks, + silo_mode=SiloMode.CELL, + processing_deadline_duration=int(BATCH_PROCESSING_DEADLINE.total_seconds()), +) +def generate_project_derived_data_batch( + project_id: int, + group_id_start: int, + group_id_end: int, + **kwargs: object, +) -> None: + """Generate derived data for groups in [group_id_start, group_id_end). + + Calls build_and_promote_derived_data for each group. On per-group + GroupLogTimeout, re-enqueues the individual group for resumption and + continues the batch. Reschedules the remaining range on batch timeout. + """ + from taskbroker_client.state import current_task + + from sentry.issues.derived.processing import GroupLogTimeout, build_and_promote_derived_data + from sentry.models.group import Group + from sentry.taskworker.selfchain_idempotency import already_spawned, mark_spawned + + task_state = current_task() + activation_id = task_state.id if task_state else None + if activation_id and already_spawned(_GENERATE_BATCH_TASK_KEY, activation_id): + logger.info( + "generate_project_derived_data_batch.duplicate_skipped", + extra={"project_id": project_id, "activation_id": activation_id}, + ) + metrics.incr( + "taskworker.selfchain.duplicate_skipped", + tags={"task": _GENERATE_BATCH_TASK_KEY}, + ) + return + + timeout_seconds = BATCH_RETRIGGER_TIMEOUT.total_seconds() + start = time.monotonic() + + group_ids = list( + Group.objects.filter( + project_id=project_id, + id__gte=group_id_start, + id__lt=group_id_end, + ) + .order_by("id") + .values_list("id", flat=True) + ) + + processed = 0 + rescheduled = False + + for group_id in group_ids: + remaining = timedelta(seconds=max(0, timeout_seconds - (time.monotonic() - start))) + try: + build_and_promote_derived_data(group_id, time_limit=remaining) + processed += 1 + except Group.DoesNotExist: + logger.info( + "generate_project_derived_data_batch.group_not_found", + extra={"group_id": group_id, "project_id": project_id}, + ) + except GroupLogTimeout as e: + # Re-enqueue the single group so the partially-drained + # state is resumed, then continue the batch. + gen_id = e.generation_id + generate_group_derived_data.delay( + group_id, + resume_generated_at=gen_id.generated_at.isoformat() + if gen_id and gen_id.generated_at + else None, + resume_pipeline_hash=gen_id.pipeline_hash if gen_id else None, + ) + + if time.monotonic() - start >= timeout_seconds: + rescheduled = True + metrics.incr( + "issues.derived.generate_batch_rescheduled", + sample_rate=1.0, + tags={"reason": "batch_timeout"}, + ) + generate_project_derived_data_batch.delay( + project_id=project_id, + group_id_start=group_id + 1, + group_id_end=group_id_end, + ) + if activation_id: + mark_spawned(_GENERATE_BATCH_TASK_KEY, activation_id) + break + + metrics.incr( + "issues.derived.generate_project_groups_processed", + amount=processed, + sample_rate=1.0, + ) + logger.info( + "generate_project_derived_data_batch.complete", + extra={ + "project_id": project_id, + "group_id_start": group_id_start, + "group_id_end": group_id_end, + "processed": processed, + "total": len(group_ids), + "rescheduled": rescheduled, + "elapsed": time.monotonic() - start, + }, + ) diff --git a/src/sentry/issues/models/groupderiveddata.py b/src/sentry/issues/models/groupderiveddata.py index 93e4415620f4..1c93873b5c62 100644 --- a/src/sentry/issues/models/groupderiveddata.py +++ b/src/sentry/issues/models/groupderiveddata.py @@ -3,6 +3,7 @@ from datetime import UTC, datetime from django.db import models +from django.utils import timezone from sentry.backup.scopes import RelocationScope from sentry.db.models import ( @@ -23,12 +24,43 @@ class GroupDerivedData(DefaultFieldsModel): """ Materialized state derived from GroupActionLogEntry entries. - One row per group. The cursor tracks the last entry processed. + One row per group (enforced by ``unique=True`` on the FK). + + Update safety + ~~~~~~~~~~~~~ + The pipeline is deterministic: replaying the same log produces the same + state. However, the log is not strictly append-only — historical entries + may be inserted, which is a primary reason generations are triggered. + + * **generated_at** — timestamp of when the generation that produced + this row's current state *started* processing. This is a CAS version: + incremental writes only succeed if ``generated_at`` hasn't changed + since the row was read, and generation promotes only succeed if their + ``generated_at`` is newer than the row's. The start time (rather + than finish time) reflects the log state the generation observed. + + * **cursor guard** — incremental writes only succeed if the writer's + ``(cursor_date, cursor_id)`` is at or ahead of the row's, preventing + cursor regression. + + * **pipeline_hash** — stamped at row creation, incremental writes only + succeed if the pipeline version hasn't changed since the row was + read. A pipeline upgrade invalidates in-flight incremental work. + + See ``processing.py`` for the full lifecycle. """ __relocation_scope__ = RelocationScope.Excluded group = FlexibleForeignKey("sentry.Group", unique=True) + + # Timestamp of when the generation that produced this state *started* + # processing. The start time (not finish time) reflects the log state + # the generation observed. Used as a CAS version — incremental writes + # check equality, generation promotes require their timestamp to be + # newer. Defaults to row creation time. + generated_at = models.DateTimeField(default=timezone.now, db_default=models.functions.Now()) + cursor_date = models.DateTimeField(default=EPOCH) cursor_id = BoundedBigIntegerField(default=0) @@ -39,12 +71,11 @@ class GroupDerivedData(DefaultFieldsModel): # Column-backed features — promoted from JSON for indexing/querying. - # This is here just for demonstration purposes. view_count = BoundedPositiveIntegerField(default=0) # Stores the current Progress value as a string. progress = models.CharField(max_length=32, null=True, default="identified") - # The last time the above column was changed. + # The last time ``progress`` was changed. last_progressed_at = models.DateTimeField(null=True, default=None) # Pipeline hash stamped at row creation. If it doesn't match the current @@ -60,4 +91,4 @@ class Meta: models.Index(fields=["last_progressed_at", "group"]), ] - __repr__ = sane_repr("group_id", "cursor_date", "cursor_id") + __repr__ = sane_repr("group_id", "generated_at", "cursor_date", "cursor_id") diff --git a/src/sentry/migrations/1141_add_generated_at_to_group_derived_data.py b/src/sentry/migrations/1141_add_generated_at_to_group_derived_data.py new file mode 100644 index 000000000000..c0a2e19d49a6 --- /dev/null +++ b/src/sentry/migrations/1141_add_generated_at_to_group_derived_data.py @@ -0,0 +1,26 @@ +# Generated by Django 5.2.14 + +import django.db.models.functions.datetime +import django.utils.timezone +from django.db import migrations, models + +from sentry.new_migrations.migrations import CheckedMigration + + +class Migration(CheckedMigration): + is_post_deployment = False + + dependencies = [ + ("sentry", "1140_organizationcontributors_unique_provider_hostname"), + ] + + operations = [ + migrations.AddField( + model_name="groupderiveddata", + name="generated_at", + field=models.DateTimeField( + default=django.utils.timezone.now, + db_default=django.db.models.functions.datetime.Now(), + ), + ), + ] diff --git a/tests/sentry/issues/derived/test_processing.py b/tests/sentry/issues/derived/test_processing.py index 82dc25f7835e..f79bafe2e735 100644 --- a/tests/sentry/issues/derived/test_processing.py +++ b/tests/sentry/issues/derived/test_processing.py @@ -1,4 +1,5 @@ from datetime import datetime, timedelta, timezone +from unittest.mock import patch import pytest from django.db import router, transaction @@ -44,13 +45,16 @@ from sentry.issues.derived.processing import ( PIPELINE, GroupLogTimeout, + PromotionResult, _entries_after_cursor, + build_and_promote_derived_data, invalidate_group_derived_data, process_group_log, + promote_to_live, ) from sentry.issues.derived.store import GroupDerivedDataStore from sentry.issues.models.groupactionlogentry import GroupActionLogEntry -from sentry.issues.models.groupderiveddata import GroupDerivedData +from sentry.issues.models.groupderiveddata import EPOCH, GroupDerivedData from sentry.issues.progress_state import IssueProgressState from sentry.models.group import Group from sentry.testutils.cases import TestCase @@ -99,6 +103,7 @@ def test_records_and_processes(self) -> None: assert entries[0].actor_id == user.id derived = process_group_log(group.id) + assert derived.cursor_id == entries[-1].id assert isinstance(derived.data, dict) @@ -108,10 +113,12 @@ def test_incremental_processing(self) -> None: _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(user.id)) derived = process_group_log(group.id) + first_cursor = derived.cursor_id _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(user.id)) derived = process_group_log(group.id) + assert derived.cursor_id > first_cursor def test_noop_when_no_new_entries(self) -> None: @@ -120,9 +127,11 @@ def test_noop_when_no_new_entries(self) -> None: _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(user.id)) derived = process_group_log(group.id) + old_updated = derived.date_updated derived = process_group_log(group.id) + assert derived.date_updated == old_updated def test_process_group_log_only_affects_target(self) -> None: @@ -222,6 +231,7 @@ def test_status_starts_open(self) -> None: _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) derived = process_group_log(group.id) + state = GroupDerivedDataStore.load(PIPELINE, derived) assert state[STATUS] == IssueStatus.OPEN @@ -231,6 +241,7 @@ def test_resolve_closes(self) -> None: _publish(group=group, action=ResolveAction(), actor=GroupActionActor.user(user.id)) derived = process_group_log(group.id) + assert derived.data["status"] == "closed" def test_unresolve_reopens(self) -> None: @@ -240,6 +251,7 @@ def test_unresolve_reopens(self) -> None: _publish(group=group, action=ResolveAction(), actor=GroupActionActor.user(user.id)) _publish(group=group, action=UnresolveAction(), actor=GroupActionActor.user(user.id)) derived = process_group_log(group.id) + assert derived.data["status"] == "open" def test_duplicate_resolve_ignored(self) -> None: @@ -249,6 +261,7 @@ def test_duplicate_resolve_ignored(self) -> None: _publish(group=group, action=ResolveAction(), actor=GroupActionActor.user(user.id)) _publish(group=group, action=ResolveAction(), actor=GroupActionActor.user(user.id)) derived = process_group_log(group.id) + assert derived.data["status"] == "closed" def test_duplicate_unresolve_ignored(self) -> None: @@ -257,6 +270,7 @@ def test_duplicate_unresolve_ignored(self) -> None: _publish(group=group, action=UnresolveAction(), actor=GroupActionActor.user(user.id)) derived = process_group_log(group.id) + state = GroupDerivedDataStore.load(PIPELINE, derived) assert state[STATUS] == IssueStatus.OPEN @@ -268,35 +282,37 @@ def test_status_toggle(self) -> None: _publish(group=group, action=UnresolveAction(), actor=GroupActionActor.user(user.id)) _publish(group=group, action=ResolveAction(), actor=GroupActionActor.user(user.id)) derived = process_group_log(group.id) + assert derived.data["status"] == "closed" # --- invalidation --- - def test_invalidate_deletes_row(self) -> None: + def test_hard_invalidate_deletes_row(self) -> None: group = self.create_group() _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) process_group_log(group.id) assert GroupDerivedData.objects.filter(group_id=group.id).exists() - invalidate_group_derived_data(group.id) + invalidate_group_derived_data(group.id, hard_delete=True) assert not GroupDerivedData.objects.filter(group_id=group.id).exists() - def test_invalidate_with_cursor_deletes_if_past(self) -> None: + def test_hard_invalidate_with_cursor_deletes_if_past(self) -> None: group = self.create_group() _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) derived = process_group_log(group.id) - # Cursor at the processed entry — row should be deleted. - invalidate_group_derived_data(group.id, cursor=(derived.cursor_date, derived.cursor_id)) + invalidate_group_derived_data( + group.id, cursor=(derived.cursor_date, derived.cursor_id), hard_delete=True + ) assert not GroupDerivedData.objects.filter(group_id=group.id).exists() def test_invalidate_with_cursor_noop_if_not_reached(self) -> None: group = self.create_group() _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) derived = process_group_log(group.id) + old_cursor = derived.cursor_id - # Cursor beyond what we've processed — row should be untouched. future = derived.cursor_date.replace(year=derived.cursor_date.year + 1) invalidate_group_derived_data(group.id, cursor=(future, old_cursor + 1000)) derived.refresh_from_db() @@ -308,12 +324,29 @@ def test_invalidate_then_reprocess(self) -> None: _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(user.id)) _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(user.id)) derived = process_group_log(group.id) + assert derived.view_count == 2 - invalidate_group_derived_data(group.id) + invalidate_group_derived_data(group.id, hard_delete=True) derived = process_group_log(group.id) + assert derived.view_count == 2 # rebuilt from scratch + def test_invalidate_soft_enqueues_generation(self) -> None: + group = self.create_group() + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + derived = process_group_log(group.id) + + original_id = derived.id + + with patch("sentry.issues.derived.processing.generate_group_derived_data") as mock_task: + invalidate_group_derived_data(group.id) + mock_task.delay.assert_called_once_with(group.id) + + # Row is still in place — soft invalidation doesn't modify it. + derived.refresh_from_db() + assert derived.id == original_id + def test_resolved_in_pull_request_proposes_fix(self) -> None: group = self.create_group() user = self.user @@ -324,6 +357,7 @@ def test_resolved_in_pull_request_proposes_fix(self) -> None: actor=GroupActionActor.user(user.id), ) derived = process_group_log(group.id) + # An open PR referencing the issue proposes a fix; the issue stays open. assert derived.data["status"] == "open" assert derived.progress == IssueProgressState.FIX_PROPOSED.value @@ -339,6 +373,7 @@ def test_pull_request_close_demotes_progress(self) -> None: actor=actor, ) derived = process_group_log(group.id) + assert derived.progress == IssueProgressState.FIX_PROPOSED.value _publish( @@ -347,6 +382,7 @@ def test_pull_request_close_demotes_progress(self) -> None: actor=actor, ) derived = process_group_log(group.id) + assert derived.progress == IssueProgressState.DIAGNOSED.value def test_pull_request_close_with_remaining_keeps_progress(self) -> None: @@ -364,6 +400,7 @@ def test_pull_request_close_with_remaining_keeps_progress(self) -> None: actor=actor, ) derived = process_group_log(group.id) + assert derived.progress == IssueProgressState.FIX_PROPOSED.value def test_pull_request_close_invalidate_and_replay_matches(self) -> None: @@ -382,12 +419,14 @@ def test_pull_request_close_invalidate_and_replay_matches(self) -> None: actor=actor, ) first = process_group_log(group.id) + first_data = first.data.copy() first_progress = first.progress first_last_progressed_at = first.last_progressed_at - invalidate_group_derived_data(group.id) + invalidate_group_derived_data(group.id, hard_delete=True) second = process_group_log(group.id) + assert second is not None assert second.data == first_data assert second.progress == first_progress @@ -398,6 +437,7 @@ def test_pipeline_hash_set_on_create(self) -> None: group = self.create_group() _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) derived = process_group_log(group.id) + assert derived.pipeline_hash == PIPELINE.pipeline_hash def test_pipeline_hash_concurrent_change_skips_cursor_update(self) -> None: @@ -405,6 +445,7 @@ def test_pipeline_hash_concurrent_change_skips_cursor_update(self) -> None: _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) derived = process_group_log(group.id) + first_cursor = derived.cursor_id # Insert a log entry directly to avoid inline processing from _publish @@ -422,23 +463,293 @@ def test_pipeline_hash_concurrent_change_skips_cursor_update(self) -> None: # between our load and the UPDATE in _process_batch. GroupDerivedData.objects.filter(group_id=group.id).update(pipeline_hash="reset") - processing._process_batch(processing.PIPELINE, derived, group.id, 1) + processing._process_batch(processing.PIPELINE, derived, 1) # The UPDATE should not have matched because the DB hash changed derived.refresh_from_db() assert derived.cursor_id == first_cursor assert derived.pipeline_hash == "reset" + def test_generated_at_change_skips_incremental_write(self) -> None: + from django.utils import timezone + + group = self.create_group() + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + + derived = process_group_log(group.id) + + first_cursor = derived.cursor_id + + GroupActionLogEntry.objects.create( + group_id=group.id, + project_id=group.project_id, + type=GroupActionType.VIEW, + actor_type=GroupActorType.SYSTEM, + actor_id=0, + source=SOURCE, + data={}, + ) + + # Simulate a generation promoting between our read and the UPDATE + # in _process_batch — generated_at changed. + GroupDerivedData.objects.filter(id=derived.id).update(generated_at=timezone.now()) + + processing._process_batch(processing.PIPELINE, derived, 1) + + derived.refresh_from_db() + assert derived.cursor_id == first_cursor + def test_invalidate_and_reprocess_restores_pipeline_hash(self) -> None: group = self.create_group() _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) process_group_log(group.id) - invalidate_group_derived_data(group.id) + invalidate_group_derived_data(group.id, hard_delete=True) derived = process_group_log(group.id) + assert derived.pipeline_hash == PIPELINE.pipeline_hash +# --- Generation lifecycle --- + + +@with_feature("projects:issue-action-log-write-to-db") +class PromoteToLiveTest(TestCase): + def test_promote_inserts_when_no_row(self) -> None: + from django.utils import timezone + + group = self.create_group() + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + GroupDerivedData.objects.filter(group_id=group.id).delete() + + gen_time = timezone.now() + candidate = GroupDerivedData( + group_id=group.id, generated_at=gen_time, cursor_date=EPOCH, cursor_id=0, data={} + ) + processing._drain_log(candidate, persist=False) + assert promote_to_live(candidate) is PromotionResult.PROMOTED + + live = GroupDerivedData.objects.get(group_id=group.id) + assert live.view_count == 1 + assert live.generated_at == gen_time + + def test_promote_updates_existing_row(self) -> None: + from django.utils import timezone + + group = self.create_group() + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + + old = process_group_log(group.id) + + old_id = old.id + + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + + gen_time = timezone.now() + candidate = GroupDerivedData( + group_id=group.id, generated_at=gen_time, cursor_date=EPOCH, cursor_id=0, data={} + ) + processing._drain_log(candidate, persist=False) + assert promote_to_live(candidate) is PromotionResult.PROMOTED + + live = GroupDerivedData.objects.get(group_id=group.id) + assert live.id == old_id + assert live.view_count == 2 + assert live.generated_at == gen_time + + def test_promote_rejected_if_cursor_behind(self) -> None: + from django.utils import timezone + + group = self.create_group() + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + + # Stamp a generated_at on the live row so the candidate can't + # win just by having a non-NULL generated_at. + gen_time = timezone.now() + process_group_log(group.id) + live = GroupDerivedData.objects.get(group_id=group.id) + GroupDerivedData.objects.filter(id=live.id).update(generated_at=gen_time) + live.refresh_from_db() + + # Candidate uses the same generated_at but only processes one entry. + candidate = GroupDerivedData( + group_id=group.id, generated_at=gen_time, cursor_date=EPOCH, cursor_id=0, data={} + ) + processing._process_batch(PIPELINE, candidate, batch_size=1, persist=False) + + assert (candidate.cursor_date, candidate.cursor_id) < ( + live.cursor_date, + live.cursor_id, + ) + assert promote_to_live(candidate) is PromotionResult.CURSOR_BEHIND + + def test_promote_superseded_by_newer_generation(self) -> None: + from django.utils import timezone + + group = self.create_group() + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + process_group_log(group.id) + + # Simulate a newer generation having already promoted. + newer_time = timezone.now() + GroupDerivedData.objects.filter(group_id=group.id).update(generated_at=newer_time) + + # An older generation tries to promote. + old_time = newer_time - timedelta(seconds=10) + candidate = GroupDerivedData( + group_id=group.id, generated_at=old_time, cursor_date=EPOCH, cursor_id=0, data={} + ) + processing._drain_log(candidate, persist=False) + assert promote_to_live(candidate) is PromotionResult.SUPERSEDED + + def test_build_and_promote(self) -> None: + group = self.create_group() + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + _publish(group=group, action=ResolveAction(), actor=GroupActionActor.user(self.user.id)) + + build_and_promote_derived_data(group.id) + derived = GroupDerivedData.objects.get(group_id=group.id) + assert derived.view_count == 1 + assert derived.data["status"] == "closed" + assert derived.generated_at is not None + + def test_build_and_promote_updates_existing_row(self) -> None: + group = self.create_group() + user = self.user + + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(user.id)) + old = process_group_log(group.id) + + old_id = old.id + + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(user.id)) + build_and_promote_derived_data(group.id) + + live = GroupDerivedData.objects.get(group_id=group.id) + assert live.id == old_id + assert live.view_count == 2 + assert live.generated_at is not None + + def test_generation_prevents_stale_incremental_write(self) -> None: + """End-to-end ABA test: incremental write computed from pre-generation + state must not overwrite a generation's result.""" + group = self.create_group() + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + + # Incremental processing reads the row. + derived = process_group_log(group.id) + + pre_gen_generated_at = derived.generated_at # None (never generated) + + # A generation runs and promotes (stamps generated_at). + build_and_promote_derived_data(group.id) + derived.refresh_from_db() + assert derived.generated_at is not None + + # A new entry arrives. + GroupActionLogEntry.objects.create( + group_id=group.id, + project_id=group.project_id, + type=GroupActionType.VIEW, + actor_type=GroupActorType.SYSTEM, + actor_id=0, + source=SOURCE, + data={}, + ) + + # Create a stale incremental writer with the pre-generation state. + stale = GroupDerivedData( + group_id=group.id, + generated_at=pre_gen_generated_at, + cursor_date=derived.cursor_date, + cursor_id=derived.cursor_id, + data=derived.data.copy(), + pipeline_hash=derived.pipeline_hash, + ) + # The stale writer processes the new entry. + processing._process_batch(PIPELINE, stale, batch_size=1) + + # The write should have been rejected because generated_at changed. + derived.refresh_from_db() + assert derived.generated_at is not None + # Cursor should NOT have advanced (stale write rejected). + entries = list(GroupActionLogEntry.objects.filter(group_id=group.id).order_by("id")) + assert derived.cursor_id == entries[-2].id # still at the pre-new-entry position + + def test_drain_log_respects_time_limit(self) -> None: + group = self.create_group() + for _ in range(5): + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + + candidate = GroupDerivedData(group_id=group.id, cursor_date=EPOCH, cursor_id=0, data={}) + + drained = processing._drain_log( + candidate, batch_size=2, time_limit=timedelta(0), persist=False + ) + assert not drained + assert candidate.cursor_id > 0 + entries = list(GroupActionLogEntry.objects.filter(group_id=group.id).order_by("id")) + assert candidate.cursor_id < entries[-1].id + + def test_process_group_log_reenqueues_on_partial_drain(self) -> None: + group = self.create_group() + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + + with ( + patch("sentry.issues.derived.processing._drain_log", return_value=False), + patch("sentry.issues.derived.processing.process_group_log_task") as mock_task, + ): + process_group_log(group.id) + + mock_task.delay.assert_called_once_with(group.id) + + def test_build_and_promote_caches_on_timeout_for_resumption(self) -> None: + group = self.create_group() + for _ in range(5): + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + + with patch("sentry.issues.derived.processing._drain_log", return_value=False): + with pytest.raises(GroupLogTimeout) as exc_info: + build_and_promote_derived_data(group.id) + + assert exc_info.value.group_id == group.id + assert exc_info.value.generation_id is not None + + # Resuming completes the promotion. + build_and_promote_derived_data(group.id, generation_id=exc_info.value.generation_id) + promoted = GroupDerivedData.objects.get(group_id=group.id) + assert promoted.view_count == 5 + + def test_resumed_generation_advances_cursor_on_repeat_timeout(self) -> None: + from sentry.issues.derived.processing import _generation_cache + + group = self.create_group() + for _ in range(5): + _publish(group=group, action=ViewAction(), actor=GroupActionActor.user(self.user.id)) + + with pytest.raises(GroupLogTimeout) as exc_info: + build_and_promote_derived_data(group.id, batch_size=2, time_limit=timedelta(0)) + + gen_id = exc_info.value.generation_id + assert gen_id is not None + state = _generation_cache.get(gen_id) + assert state is not None + first_cursor = state.cursor_id + assert first_cursor > 0 + + with pytest.raises(GroupLogTimeout) as exc_info: + build_and_promote_derived_data( + group.id, generation_id=gen_id, batch_size=2, time_limit=timedelta(0) + ) + + gen_id2 = exc_info.value.generation_id + assert gen_id2 is not None + state = _generation_cache.get(gen_id2) + assert state is not None + assert state.cursor_id > first_cursor + + # --- Pure Python tests (no DB) --- @@ -648,8 +959,9 @@ def test_round_trip_preserves_state(self) -> None: first_progress = first.progress first_last_progressed_at = first.last_progressed_at - invalidate_group_derived_data(group.id) + invalidate_group_derived_data(group.id, hard_delete=True) second = process_group_log(group.id) + assert second is not None assert second.data == first_data assert second.view_count == first_view_count @@ -719,8 +1031,6 @@ class _IntentionalRollback(Exception): @with_feature("projects:issue-action-log-write-to-db") class DerivedDataTransactionTest(TestCase): - """Verify derived data processing respects transaction boundaries.""" - def test_rolled_back_action_does_not_produce_derived_data(self) -> None: group = self.create_group() @@ -775,4 +1085,5 @@ def test_completes_with_generous_timeout(self) -> None: GroupDerivedData.objects.filter(group_id=group.id).delete() derived = process_group_log(group.id, timeout=timedelta(minutes=5)) + assert derived.view_count == 3 diff --git a/tests/sentry/issues/derived/test_tasks.py b/tests/sentry/issues/derived/test_tasks.py index a8b7bd6f8f32..77dfb79b0992 100644 --- a/tests/sentry/issues/derived/test_tasks.py +++ b/tests/sentry/issues/derived/test_tasks.py @@ -156,7 +156,7 @@ def test_reschedules_on_group_log_timeout(self) -> None: patch.object( processing, "process_group_log", - side_effect=GroupLogTimeout("timed out"), + side_effect=GroupLogTimeout(0), ), patch.object(process_project_derived_data_batch, "delay") as mock_delay, ): diff --git a/tests/sentry/tasks/test_merge.py b/tests/sentry/tasks/test_merge.py index b03c8348b322..70dc2555136a 100644 --- a/tests/sentry/tasks/test_merge.py +++ b/tests/sentry/tasks/test_merge.py @@ -3,7 +3,7 @@ from unittest.mock import patch from sentry import buffer, eventstream -from sentry.issues.models.groupderiveddata import EPOCH, GroupDerivedData +from sentry.issues.models.groupderiveddata import GroupDerivedData from sentry.models.group import Group from sentry.models.groupenvironment import GroupEnvironment from sentry.models.groupmeta import GroupMeta @@ -286,12 +286,9 @@ def test_merge_invalidates_derived_data(self) -> None: assert Group.objects.filter(id=old_group.id).exists() is False - # The derived data is invalidated: the stale row is deleted and rebuilt - # from scratch by the scheduled processing task. - rebuilt = GroupDerivedData.objects.get(group_id=new_group.id) - assert rebuilt.id != derived.id - assert rebuilt.cursor_date == EPOCH - assert rebuilt.cursor_id == 0 + # The generation task ran and stamped a new generated_at. + derived.refresh_from_db() + assert derived.generated_at is not None @mock_redis_buffer() def test_merge_original_group_id(self) -> None: