diff --git a/migrations_lockfile.txt b/migrations_lockfile.txt index eab14bdc1d72..eef89c59d978 100644 --- a/migrations_lockfile.txt +++ b/migrations_lockfile.txt @@ -31,7 +31,7 @@ releases: 0004_cleanup_failed_safe_deletes replays: 0001_squashed_0007_organizationmember_replay_access -seer: 0029_add_agent_write_grant +seer: 0030_add_night_shift_run_schedule_id sentry: 1146_add_gcp_service_account diff --git a/src/sentry/seer/migrations/0030_add_night_shift_run_schedule_id.py b/src/sentry/seer/migrations/0030_add_night_shift_run_schedule_id.py new file mode 100644 index 000000000000..ca290ac17b49 --- /dev/null +++ b/src/sentry/seer/migrations/0030_add_night_shift_run_schedule_id.py @@ -0,0 +1,47 @@ +# Generated by Django 5.2.14 on 2026-07-24 22:38 + +from django.db import migrations, models + +from sentry.new_migrations.migrations import CheckedMigration + + +class Migration(CheckedMigration): + # This flag is used to mark that a migration shouldn't be automatically run in production. + # This should only be used for operations where it's safe to run the migration after your + # code has deployed. So this should not be used for most operations that alter the schema + # of a table. + # Here are some things that make sense to mark as post deployment: + # - Large data migrations. Typically we want these to be run manually so that they can be + # monitored and not block the deploy for a long period of time while they run. + # - Adding indexes to large tables. Since this can take a long time, we'd generally prefer to + # run this outside deployments so that we don't block them. Note that while adding an index + # is a schema change, it's completely safe to run the operation after the code has deployed. + # Once deployed, run these manually via: https://develop.sentry.dev/database-migrations/#migration-deployment + + is_post_deployment = False + + dependencies = [ + ("seer", "0029_add_agent_write_grant"), + ("sentry", "1146_add_gcp_service_account"), + ] + + operations = [ + migrations.AddField( + model_name="seernightshiftrun", + name="schedule_id", + field=models.CharField(max_length=256, null=True), + ), + migrations.AddField( + model_name="seernightshiftrun", + name="date_completed", + field=models.DateTimeField(null=True), + ), + migrations.AddConstraint( + model_name="seernightshiftrun", + constraint=models.UniqueConstraint( + condition=models.Q(("schedule_id__isnull", False)), + fields=("organization", "workflow_config", "schedule_id"), + name="seer_nightshiftrun_unique_org_config_schedule", + ), + ), + ] diff --git a/src/sentry/seer/models/night_shift.py b/src/sentry/seer/models/night_shift.py index 129cebbe0554..30077685c3d5 100644 --- a/src/sentry/seer/models/night_shift.py +++ b/src/sentry/seer/models/night_shift.py @@ -10,9 +10,10 @@ @cell_silo_model class SeerNightShiftRun(DefaultFieldsModel): - """ - Records each night shift invocation for an organization. - One row is created per org each time run_night_shift_for_org executes. + """Records each night shift invocation for an organization. + + Cron invocations create one row per organization, workflow config, and + schedule window. Manual invocations create one row per execution. """ __relocation_scope__ = RelocationScope.Excluded @@ -21,6 +22,10 @@ class SeerNightShiftRun(DefaultFieldsModel): workflow_config = FlexibleForeignKey( "seer.SeerWorkflowConfig", on_delete=models.SET_NULL, null=True ) + # Cron-derived schedule window (currently YYYY-MM-DDTHH:MM); nullable for + # manual and historical runs, with standard capacity for future formats. + schedule_id = models.CharField(max_length=256, null=True) + date_completed = models.DateTimeField(null=True) extras = models.JSONField(db_default={}, default=dict) class Meta: @@ -31,6 +36,13 @@ class Meta: models.Index(fields=["date_added"]), models.Index(fields=["workflow_config", "date_added"]), ] + constraints = [ + models.UniqueConstraint( + fields=["organization", "workflow_config", "schedule_id"], + condition=models.Q(schedule_id__isnull=False), + name="seer_nightshiftrun_unique_org_config_schedule", + ) + ] __repr__ = sane_repr("organization_id", "workflow_config_id", "date_added") diff --git a/src/sentry/tasks/seer/night_shift/cron.py b/src/sentry/tasks/seer/night_shift/cron.py index 8f332e1617d9..afeaf4e65370 100644 --- a/src/sentry/tasks/seer/night_shift/cron.py +++ b/src/sentry/tasks/seer/night_shift/cron.py @@ -4,11 +4,16 @@ import logging import time from collections.abc import Mapping, Sequence -from datetime import timedelta +from datetime import UTC, datetime, timedelta from typing import Any, Literal, TypedDict import sentry_sdk +from cronsim import CronSim +from django.conf import settings +from django.db import router, transaction +from django.utils import timezone from django.utils.translation import ngettext +from taskbroker_client.scheduler.config import crontab from sentry import features, options, quotas from sentry.constants import ( @@ -105,6 +110,21 @@ class SeerNightShiftRunOptionsPartial(TypedDict, total=False): extra_triage_instructions: str +def _night_shift_cron_expr() -> str: + schedule = settings.TASKWORKER_SCHEDULES["seer-night-shift"]["schedule"] + if not isinstance(schedule, crontab): + raise TypeError( + "The seer-night-shift schedule must use taskbroker_client.scheduler.config.crontab" + ) + return str(schedule) + + +def _current_schedule_id(now: datetime, cron_expr: str) -> str: + """Return the most recent scheduled fire time at or before ``now``.""" + base = now.replace(second=0, microsecond=0) + timedelta(minutes=1) + return next(CronSim(cron_expr, base, reverse=True)).strftime("%Y-%m-%dT%H:%M") + + @instrumented_task( name="sentry.tasks.seer.night_shift.schedule_night_shift", namespace=seer_tasks, @@ -128,7 +148,11 @@ def schedule_night_shift( if not options.get("seer.night_shift.enable"): return - logger.info("night_shift.schedule_start") + schedule_id: str | None = None + if run_options is None: + schedule_id = _current_schedule_id(datetime.now(tz=UTC), _night_shift_cron_expr()) + + logger.info("night_shift.schedule_start", extra={"schedule_id": schedule_id}) start_time = time.monotonic() seer_org_ids: set[int] = set() @@ -150,7 +174,11 @@ def schedule_night_shift( spread_seconds = int(NIGHT_SHIFT_SPREAD_DURATION.total_seconds()) batch_index = 0 - task_kwargs: dict[str, Any] = {"options": dict(run_options)} if run_options else {} + task_kwargs: dict[str, Any] = {} + if run_options is not None: + task_kwargs["options"] = dict(run_options) + if schedule_id is not None: + task_kwargs["schedule_id"] = schedule_id for chunk_index, org_id_chunk in enumerate(chunked(seer_org_ids, 100)): org_batch = list( @@ -187,6 +215,7 @@ def schedule_night_shift( extra={ "orgs_dispatched": batch_index, "elapsed_seconds": time.monotonic() - start_time, + "schedule_id": schedule_id, }, ) @@ -203,6 +232,7 @@ def run_night_shift_for_org( project_ids: list[int] | None = None, triggering_user_id: int | None = None, execute_in_task: bool = False, + schedule_id: str | None = None, **kwargs: Any, ) -> int | None: """Run night shift for one organization. `options` is a partial dict — @@ -244,11 +274,43 @@ def run_night_shift_for_org( if triggering_user_id is not None: extras["triggering_user_id"] = triggering_user_id - run = SeerNightShiftRun.objects.create( - organization=organization, - workflow_config=workflow_config, - extras=extras, - ) + created = schedule_id is None + if schedule_id is None: + run = SeerNightShiftRun.objects.create( + organization=organization, + workflow_config=workflow_config, + extras=extras, + ) + else: + run, created = SeerNightShiftRun.objects.get_or_create( + organization=organization, + workflow_config=workflow_config, + schedule_id=schedule_id, + defaults={"extras": extras}, + ) + + if not created and run.date_completed is not None: + logger.info( + "night_shift.duplicate_run_skipped", + extra={ + "organization_id": organization.id, + "schedule_id": schedule_id, + "night_shift_run_id": run.id, + }, + ) + sentry_sdk.metrics.count("night_shift.duplicate_run_skipped", 1) + return run.id + + if not created: + logger.info( + "night_shift.incomplete_run_resumed", + extra={ + "organization_id": organization.id, + "schedule_id": schedule_id, + "night_shift_run_id": run.id, + }, + ) + sentry_sdk.metrics.count("night_shift.incomplete_run_resumed", 1) task_kwargs: dict[str, Any] = {"options": dict(resolved_options)} if project_ids is not None: @@ -295,9 +357,18 @@ def run_night_shift_execution( {"organization_id": organization.id, "organization_slug": organization.slug} ) + if run.date_completed is not None: + logger.info("night_shift.execute_already_complete", extra=log_extra) + return None + start_time = time.monotonic() logger.info("night_shift.execute.start", extra=log_extra) + if run.shards.exists(): + if _dispatch_pending_shards(run, organization, log_extra, start_time): + _complete_run(run) + return None + if not quotas.backend.check_seer_quota( org_id=organization.id, data_category=DataCategory.SEER_AUTOFIX, @@ -325,9 +396,13 @@ def run_night_shift_execution( if not eligible: logger.info("night_shift.no_eligible_projects", extra=log_extra) + _complete_run(run) return None - _dispatch_to_seer_feature(run, organization, eligible, resolved_options, log_extra, start_time) + if _plan_and_dispatch_shards( + run, organization, eligible, resolved_options, log_extra, start_time + ): + _complete_run(run) def _run_option_defaults(data: Mapping[str, Any]) -> SeerNightShiftRunOptions: @@ -437,8 +512,25 @@ def _get_eligible_orgs_from_batch( return eligible +def _complete_run(run: SeerNightShiftRun) -> None: + using = router.db_for_write(SeerNightShiftRun) + with transaction.atomic(using=using): + locked_run = SeerNightShiftRun.objects.select_for_update().get(id=run.id) + if locked_run.date_completed is not None: + return + + extras = dict(locked_run.extras or {}) + extras.pop("error_message", None) + locked_run.update(extras=extras, date_completed=timezone.now()) + + def _record_run_error(run: SeerNightShiftRun, message: str) -> None: - run.update(extras={**(run.extras or {}), "error_message": message}) + using = router.db_for_write(SeerNightShiftRun) + with transaction.atomic(using=using): + locked_run = SeerNightShiftRun.objects.select_for_update().get(id=run.id) + if locked_run.date_completed is not None: + return + locked_run.update(extras={**(locked_run.extras or {}), "error_message": message}) def _fail_run( @@ -579,18 +671,19 @@ def _build_triage_payload( ) -def _dispatch_to_seer_feature( +def _plan_and_dispatch_shards( run: SeerNightShiftRun, organization: Organization, eligible: Sequence[EligibleProject], resolved_options: SeerNightShiftRunOptions, log_extra: dict[str, object], start_time: float, -) -> None: - """Shard the scored candidates into chunks of seer.night_shift.shard_size and - dispatch each chunk as its own Seer feature run, recorded as a - SeerNightShiftRunShard. Seer pushes verdicts back per shard via - deliver_feature_result.""" +) -> bool: + """Persist the full shard plan before creating any Seer runs. + + A redelivered task can then dispatch only plan rows without a SeerRun, + avoiding duplicate feature runs after an interrupted execution. + """ eligible_projects = [ep.project for ep in eligible] repos_by_project = {ep.project.id: ep.connected_repos for ep in eligible} tuning_by_project = { @@ -606,22 +699,19 @@ def _dispatch_to_seer_feature( run.update(extras={**(run.extras or {}), "num_candidates": len(scored)}) if not scored: logger.info("night_shift.no_candidates", extra=log_extra) - return + return True try: - client = SeerAgentClient(organization) + SeerAgentClient(organization) except SeerPermissionError: logger.info("night_shift.no_seer_access", extra=log_extra) _record_run_error(run, "Organization does not have Seer access") - return - - def _link_shard(created: SeerRun) -> None: - SeerNightShiftRunShard.objects.create(run=run, seer_run=created) + return False shard_size = max(1, options.get("seer.night_shift.shard_size")) - shards = list(chunked(scored, shard_size)) - dispatched = 0 - for shard_index, chunk in enumerate(shards): + chunks = list(chunked(scored, shard_size)) + shard_plan: list[dict[str, object]] = [] + for shard_index, chunk in enumerate(chunks): payload = _build_triage_payload( chunk, resolved_options, repos_by_project, tuning_by_project ) @@ -631,47 +721,111 @@ def _link_shard(created: SeerRun) -> None: "Agentic triage (%(count)d candidates)", num_candidates, ) % {"count": num_candidates} - if len(shards) > 1: - title += f" — part {shard_index + 1} of {len(shards)}" - try: - client.start_feature_run( - feature_id="night_shift", - payload=payload.dict(), - title=title, - flush=False, - on_run_created=_link_shard, - ) - except Exception: - logger.exception( - "night_shift.shard_dispatch_failed", - extra={**log_extra, "shard_index": shard_index, "num_shards": len(shards)}, - ) - continue - dispatched += 1 + if len(chunks) > 1: + title += f" — part {shard_index + 1} of {len(chunks)}" + shard_plan.append({"payload": payload.dict(), "title": title}) + + _get_or_create_shard_plan(run, shard_plan) + return _dispatch_pending_shards(run, organization, log_extra, start_time) + + +def _get_or_create_shard_plan( + run: SeerNightShiftRun, shard_plan: Sequence[dict[str, object]] +) -> list[SeerNightShiftRunShard]: + using = router.db_for_write(SeerNightShiftRunShard) + with transaction.atomic(using=using): + locked_run = SeerNightShiftRun.objects.select_for_update().get(id=run.id) + planned_shards = list(locked_run.shards.order_by("id")) + if planned_shards: + return planned_shards + + SeerNightShiftRunShard.objects.bulk_create( + [SeerNightShiftRunShard(run=locked_run, extras=plan) for plan in shard_plan] + ) + return list(locked_run.shards.order_by("id")) - if dispatched == 0: - sentry_sdk.metrics.count("night_shift.run_error", 1) - _record_run_error(run, "Night shift dispatch failed") - logger.error("night_shift.dispatch_failed", extra={**log_extra, "num_shards": len(shards)}) - return - failed_shards = len(shards) - dispatched - if failed_shards: +def _dispatch_pending_shards( + run: SeerNightShiftRun, + organization: Organization, + log_extra: dict[str, object], + start_time: float, +) -> bool: + try: + client = SeerAgentClient(organization) + except SeerPermissionError: + logger.info("night_shift.no_seer_access", extra=log_extra) + _record_run_error(run, "Organization does not have Seer access") + return False + + using = router.db_for_write(SeerNightShiftRunShard) + planned_shards = list(run.shards.order_by("id")) + dispatched = 0 + for shard_index, planned_shard in enumerate(planned_shards): + with transaction.atomic(using=using): + shard = SeerNightShiftRunShard.objects.select_for_update().get(id=planned_shard.id) + if shard.seer_run_id is not None: + dispatched += 1 + continue + + payload = shard.extras.get("payload") + title = shard.extras.get("title") + if not isinstance(payload, dict) or not isinstance(title, str): + logger.error( + "night_shift.invalid_shard_plan", + extra={**log_extra, "shard_index": shard_index}, + ) + return False + + def _link_shard(created: SeerRun) -> None: + shard.seer_run = created + shard.save(update_fields=["seer_run"]) + + try: + client.start_feature_run( + feature_id="night_shift", + payload=payload, + title=title, + flush=False, + on_run_created=_link_shard, + ) + except Exception: + logger.exception( + "night_shift.shard_dispatch_failed", + extra={ + **log_extra, + "shard_index": shard_index, + "num_shards": len(planned_shards), + }, + ) + continue + dispatched += 1 + + if dispatched != len(planned_shards): + failed_shards = len(planned_shards) - dispatched sentry_sdk.metrics.count("night_shift.shard_dispatch_failure", failed_shards) - _record_run_error(run, f"Failed to dispatch {failed_shards} of {len(shards)} triage shards") + _record_run_error( + run, f"Failed to dispatch {failed_shards} of {len(planned_shards)} triage shards" + ) logger.warning( "night_shift.partial_dispatch_failure", - extra={**log_extra, "num_shards": len(shards), "num_shards_dispatched": dispatched}, + extra={ + **log_extra, + "num_shards": len(planned_shards), + "num_shards_dispatched": dispatched, + }, ) + return False sentry_sdk.metrics.distribution("night_shift.org_run_duration", time.monotonic() - start_time) logger.info( "night_shift.feature_dispatched", extra={ **log_extra, - "num_eligible_projects": len(eligible_projects), - "num_candidates": len(scored), - "num_shards": len(shards), + "num_eligible_projects": run.extras.get("num_eligible_projects"), + "num_candidates": run.extras.get("num_candidates"), + "num_shards": len(planned_shards), "num_shards_dispatched": dispatched, }, ) + return True diff --git a/tests/sentry/tasks/seer/test_night_shift.py b/tests/sentry/tasks/seer/test_night_shift.py index 5fe6434fa7f6..4b7d0ff35881 100644 --- a/tests/sentry/tasks/seer/test_night_shift.py +++ b/tests/sentry/tasks/seer/test_night_shift.py @@ -1,5 +1,11 @@ +from datetime import UTC, datetime from unittest.mock import Mock, patch +import pytest +from django.conf import settings +from django.db import IntegrityError +from taskbroker_client.scheduler.config import crontab + from sentry.hybridcloud.models.outbox import CellOutbox from sentry.hybridcloud.outbox.category import OutboxCategory from sentry.issues.search import group_types_from @@ -7,6 +13,7 @@ from sentry.models.organization import OrganizationStatus from sentry.models.project import Project from sentry.processing_errors.grouptype import LowValueSpanConfigurationType +from sentry.seer.agent.client import SeerAgentClient from sentry.seer.autofix.constants import AutofixAutomationTuningSettings from sentry.seer.autofix.utils import AutofixStoppingPoint, bulk_read_preferences_from_sentry_db from sentry.seer.models.night_shift import ( @@ -17,7 +24,9 @@ from sentry.seer.models.run import SeerRun, SeerRunMirrorStatus, SeerRunType from sentry.seer.models.workflow import SeerWorkflowStrategy from sentry.tasks.seer.night_shift.cron import ( + _current_schedule_id, _get_eligible_projects, + _night_shift_cron_expr, build_run_options, run_night_shift_for_org, schedule_night_shift, @@ -34,7 +43,7 @@ from sentry.tasks.seer.night_shift.skip_cache import mark_skipped from sentry.testutils.cases import SnubaTestCase, TestCase from sentry.testutils.fixtures import Fixtures -from sentry.testutils.helpers.datetime import before_now +from sentry.testutils.helpers.datetime import before_now, freeze_time from sentry.testutils.outbox import outbox_runner from sentry.testutils.pytest.fixtures import django_db_all from sentry.utils.cursors import Cursor @@ -146,6 +155,61 @@ def test_manual_overrides_win(self) -> None: assert resolved["max_candidates"] == 3 +class TestCurrentScheduleId(TestCase): + def test_exact_fire_minute(self) -> None: + cron_expr = "0 10,22 * * *" + assert _current_schedule_id(datetime(2024, 7, 22, 22, 0, tzinfo=UTC), cron_expr) == ( + "2024-07-22T22:00" + ) + assert _current_schedule_id(datetime(2024, 7, 22, 22, 0, 59, tzinfo=UTC), cron_expr) == ( + "2024-07-22T22:00" + ) + + def test_redelivery_offsets(self) -> None: + cron_expr = "0 10,22 * * *" + assert _current_schedule_id(datetime(2024, 7, 22, 22, 30, tzinfo=UTC), cron_expr) == ( + "2024-07-22T22:00" + ) + assert _current_schedule_id(datetime(2024, 7, 22, 23, 30, tzinfo=UTC), cron_expr) == ( + "2024-07-22T22:00" + ) + + def test_day_rollover(self) -> None: + assert ( + _current_schedule_id(datetime(2024, 7, 23, 1, 30, tzinfo=UTC), "0 10,22 * * *") + == "2024-07-22T22:00" + ) + + def test_window_edges(self) -> None: + cron_expr = "0 10,22 * * *" + assert _current_schedule_id(datetime(2024, 7, 22, 9, 59, tzinfo=UTC), cron_expr) == ( + "2024-07-21T22:00" + ) + assert _current_schedule_id(datetime(2024, 7, 22, 10, 0, 5, tzinfo=UTC), cron_expr) == ( + "2024-07-22T10:00" + ) + + def test_flexible_cadence(self) -> None: + cron_expr = "0 2,6,10,14,18,22 * * *" + assert _current_schedule_id(datetime(2024, 7, 22, 14, 0, 30, tzinfo=UTC), cron_expr) == ( + "2024-07-22T14:00" + ) + assert _current_schedule_id(datetime(2024, 7, 22, 1, 59, tzinfo=UTC), cron_expr) == ( + "2024-07-21T22:00" + ) + + def test_night_shift_beat_schedule(self) -> None: + schedule_entry = settings.TASKWORKER_SCHEDULES["seer-night-shift"] + assert schedule_entry["task"] == "seer:sentry.tasks.seer.night_shift.schedule_night_shift" + assert isinstance(schedule_entry["schedule"], crontab) + assert ( + _current_schedule_id( + datetime(2024, 7, 22, 22, 30, tzinfo=UTC), _night_shift_cron_expr() + ) + == "2024-07-22T22:00" + ) + + @django_db_all class TestScheduleNightShift(TestCase): def create_org_with_seer(self): @@ -168,6 +232,7 @@ def test_dispatches_eligible_orgs(self) -> None: org = self.create_org_with_seer() with ( + freeze_time("2024-07-22 22:30:00Z"), self.options({"seer.night_shift.enable": True}), self.feature( { @@ -177,14 +242,22 @@ def test_dispatches_eligible_orgs(self) -> None: } ), patch("sentry.tasks.seer.night_shift.cron.run_night_shift_for_org") as mock_worker, + patch("sentry.tasks.seer.night_shift.cron.logger") as mock_logger, ): schedule_night_shift() mock_worker.apply_async.assert_called_once() assert mock_worker.apply_async.call_args.kwargs["args"] == [org.id] - assert mock_worker.apply_async.call_args.kwargs["kwargs"] == {} + assert mock_worker.apply_async.call_args.kwargs["kwargs"] == { + "schedule_id": "2024-07-22T22:00" + } assert mock_worker.apply_async.call_args.kwargs["headers"] == { "sentry-propagate-traces": False } + log_extras = { + call.args[0]: call.kwargs["extra"] for call in mock_logger.info.call_args_list + } + assert log_extras["night_shift.schedule_start"]["schedule_id"] == "2024-07-22T22:00" + assert log_extras["night_shift.schedule_complete"]["schedule_id"] == "2024-07-22T22:00" def test_dispatches_with_run_options(self) -> None: org = self.create_org_with_seer() @@ -209,6 +282,29 @@ def test_dispatches_with_run_options(self) -> None: "options": {"source": "manual", "dry_run": True, "max_candidates": 3}, } + def test_redelivery_dispatches_same_schedule_id(self) -> None: + org = self.create_org_with_seer() + + with ( + freeze_time("2024-07-22 22:30:00Z"), + self.options({"seer.night_shift.enable": True}), + self.feature( + { + "organizations:seer-night-shift": [org.slug], + "organizations:gen-ai-features": [org.slug], + "organizations:seat-based-seer-enabled": [org.slug], + } + ), + patch("sentry.tasks.seer.night_shift.cron.run_night_shift_for_org") as mock_worker, + ): + schedule_night_shift() + schedule_night_shift() + + assert [call.kwargs["kwargs"] for call in mock_worker.apply_async.call_args_list] == [ + {"schedule_id": "2024-07-22T22:00"}, + {"schedule_id": "2024-07-22T22:00"}, + ] + def test_skips_orgs_without_seat_based_seer(self) -> None: org = self.create_org_with_seer() @@ -519,6 +615,68 @@ def test_nonexistent_org(self) -> None: run_night_shift_for_org(999999999) mock_logger.info.assert_not_called() + def test_incomplete_schedule_id_resumes_execution(self) -> None: + org = self.create_organization() + + with ( + patch("sentry.tasks.seer.night_shift.cron.run_night_shift_execution") as mock_execute, + patch("sentry.tasks.seer.night_shift.cron.logger") as mock_logger, + patch("sentry.tasks.seer.night_shift.cron.sentry_sdk.metrics.count") as mock_count, + ): + first_run_id = run_night_shift_for_org(org.id, schedule_id="2024-07-22T22:00") + second_run_id = run_night_shift_for_org(org.id, schedule_id="2024-07-22T22:00") + + assert first_run_id == second_run_id + assert SeerNightShiftRun.objects.filter(organization=org).count() == 1 + assert mock_execute.call_count == 2 + mock_logger.info.assert_called_once_with( + "night_shift.incomplete_run_resumed", + extra={ + "organization_id": org.id, + "schedule_id": "2024-07-22T22:00", + "night_shift_run_id": first_run_id, + }, + ) + mock_count.assert_called_once_with("night_shift.incomplete_run_resumed", 1) + + def test_different_schedule_ids_create_separate_runs(self) -> None: + org = self.create_organization() + + with patch("sentry.tasks.seer.night_shift.cron.run_night_shift_execution"): + run_night_shift_for_org(org.id, schedule_id="2024-07-22T22:00") + run_night_shift_for_org(org.id, schedule_id="2024-07-23T10:00") + + assert SeerNightShiftRun.objects.filter(organization=org).count() == 2 + + def test_null_schedule_id_preserves_manual_semantics(self) -> None: + org = self.create_organization() + + with patch("sentry.tasks.seer.night_shift.cron.run_night_shift_execution"): + run_night_shift_for_org(org.id) + run_night_shift_for_org(org.id) + + assert ( + SeerNightShiftRun.objects.filter(organization=org, schedule_id__isnull=True).count() + == 2 + ) + + def test_unique_schedule_constraint(self) -> None: + org = self.create_organization() + + with patch("sentry.tasks.seer.night_shift.cron.run_night_shift_execution"): + run_id = run_night_shift_for_org(org.id, schedule_id="2024-07-22T22:00") + assert run_id is not None + + run = SeerNightShiftRun.objects.get(id=run_id) + pytest.raises( + IntegrityError, + SeerNightShiftRun.objects.create, + organization=org, + workflow_config=run.workflow_config, + schedule_id="2024-07-22T22:00", + extras={}, + ) + def test_no_eligible_projects(self) -> None: org = self.create_organization() self.create_project(organization=org) @@ -534,21 +692,27 @@ def test_no_eligible_projects(self) -> None: assert run.extras.get("error_message") is None assert not SeerNightShiftRunResult.objects.filter(run=run).exists() - def test_eligible_projects_error_records_error_message(self) -> None: + def test_eligible_projects_error_resumes_same_schedule_run(self) -> None: org = self.create_organization() self.create_project(organization=org) + schedule_id = "2024-07-22T22:00" - with ( - patch( - "sentry.tasks.seer.night_shift.cron._get_eligible_projects", - side_effect=RuntimeError("boom"), - ), + with patch( + "sentry.tasks.seer.night_shift.cron._get_eligible_projects", + side_effect=RuntimeError("boom"), ): - run_night_shift_for_org(org.id) + run_night_shift_for_org(org.id, schedule_id=schedule_id) run = SeerNightShiftRun.objects.get(organization=org) assert run.extras["error_message"] == "Failed to get eligible projects" - assert not SeerNightShiftRunResult.objects.filter(run=run).exists() + assert run.date_completed is None + + run_night_shift_for_org(org.id, schedule_id=schedule_id) + + resumed_run = SeerNightShiftRun.objects.get(id=run.id) + assert resumed_run.date_completed is not None + assert resumed_run.extras.get("error_message") is None + assert not SeerNightShiftRunResult.objects.filter(run=resumed_run).exists() def test_filters_recently_skipped_groups(self) -> None: org = self.create_organization() @@ -573,25 +737,33 @@ def test_filters_recently_skipped_groups(self) -> None: candidate_ids = [c["group_id"] for c in body["payload"]["candidates"]] assert candidate_ids == [other_group.id] - def test_skips_dispatch_when_no_seer_quota(self) -> None: + def test_no_seer_quota_resumes_same_schedule_run(self) -> None: org = self.create_organization() - project = self.create_project(organization=org) - self._make_eligible(project) - - self._store_event_and_update_group( - project, "fixable", seer_fixability_score=0.9, times_seen=5 - ) + schedule_id = "2024-07-22T22:00" with patch( "sentry.tasks.seer.night_shift.cron.quotas.backend.check_seer_quota", return_value=False, ): - run_night_shift_for_org(org.id) + run_night_shift_for_org(org.id, schedule_id=schedule_id) run = SeerNightShiftRun.objects.get(organization=org) assert run.extras["error_message"] == "No Seer quota available" + assert run.date_completed is None assert not SeerRun.objects.filter(organization=org).exists() - assert not SeerNightShiftRunResult.objects.filter(run=run).exists() + + with ( + patch( + "sentry.tasks.seer.night_shift.cron.quotas.backend.check_seer_quota", + return_value=True, + ), + patch("sentry.tasks.seer.night_shift.cron._get_eligible_projects", return_value=[]), + ): + run_night_shift_for_org(org.id, schedule_id=schedule_id) + + resumed_run = SeerNightShiftRun.objects.get(id=run.id) + assert resumed_run.date_completed is not None + assert resumed_run.extras.get("error_message") is None def test_max_candidates_defaults_to_global_option(self) -> None: org = self.create_organization() @@ -932,36 +1104,58 @@ def test_shards_candidates_across_feature_runs(self) -> None: assert sorted(dispatched_group_ids) == sorted(g.id for g in groups) assert run.extras.get("error_message") is None - def test_partial_shard_failure_still_dispatches(self) -> None: + def test_partial_shard_failure_resumes_without_duplicate_runs(self) -> None: org = self.create_organization() project = self.create_project(organization=org) self._make_eligible(project) - for i in range(2): - self._store_event_and_update_group( - project, f"fixable-{i}", seer_fixability_score=0.9, times_seen=5 + i - ) - - real_create = SeerNightShiftRunShard.objects.create - calls: list[int] = [] - - def flaky_create(*args, **kwargs): - calls.append(1) - if len(calls) == 2: + groups = [self.create_group(project=project) for _ in range(2)] + scored = [ScoredCandidate(group=group, fixability=0.9) for group in groups] + real_start_feature_run = SeerAgentClient.start_feature_run + calls = 0 + + def fail_second_dispatch(client, *args, **kwargs): + nonlocal calls + calls += 1 + if calls == 2: raise RuntimeError("boom") - return real_create(*args, **kwargs) + return real_start_feature_run(client, *args, **kwargs) with ( self.options({"seer.night_shift.shard_size": 1}), self.feature("organizations:gen-ai-features"), - patch.object(SeerNightShiftRunShard.objects, "create", side_effect=flaky_create), + patch( + "sentry.tasks.seer.night_shift.cron.fixability_score_strategy", + return_value=scored, + ) as mock_score, + patch.object( + SeerAgentClient, + "start_feature_run", + autospec=True, + side_effect=fail_second_dispatch, + ), ): - run_night_shift_for_org(org.id) + run_night_shift_for_org(org.id, schedule_id="2024-07-22T22:00") - run = SeerNightShiftRun.objects.get(organization=org) - # One shard dispatched; the failed one is recorded so it isn't invisible. - assert SeerNightShiftRunShard.objects.filter(run=run).count() == 1 - assert SeerRun.objects.filter(organization=org, type=SeerRunType.FEATURE_RUN).count() == 1 - assert run.extras["error_message"] == "Failed to dispatch 1 of 2 triage shards" + run = SeerNightShiftRun.objects.get(organization=org) + assert run.date_completed is None + assert run.shards.filter(seer_run__isnull=True).count() == 1 + assert ( + SeerRun.objects.filter(organization=org, type=SeerRunType.FEATURE_RUN).count() == 1 + ) + + run_night_shift_for_org(org.id, schedule_id="2024-07-22T22:00") + resumed_run = SeerNightShiftRun.objects.get(id=run.id) + assert resumed_run.date_completed is not None + assert resumed_run.extras.get("error_message") is None + assert resumed_run.shards.filter(seer_run__isnull=True).count() == 0 + assert ( + SeerRun.objects.filter(organization=org, type=SeerRunType.FEATURE_RUN).count() == 2 + ) + + run_night_shift_for_org(org.id, schedule_id="2024-07-22T22:00") + + mock_score.assert_called_once() + assert SeerRun.objects.filter(organization=org, type=SeerRunType.FEATURE_RUN).count() == 2 def test_no_candidates_skips_dispatch(self) -> None: org = self.create_organization() @@ -1009,8 +1203,9 @@ def test_dispatch_failure_records_error(self) -> None: run_night_shift_for_org(org.id) run = SeerNightShiftRun.objects.get(organization=org) - assert not run.shards.exists() - assert run.extras["error_message"] == "Night shift dispatch failed" + assert run.shards.filter(seer_run__isnull=True).count() == 1 + assert run.date_completed is None + assert run.extras["error_message"] == "Failed to dispatch 1 of 1 triage shards" def test_outbox_drain_mirrors_run_against_seer(self) -> None: org = self.create_organization()