From 940d5daf091c11cdc4ddbc0d00e090b8e9fcaf55 Mon Sep 17 00:00:00 2001 From: Trevor Elkins Date: Wed, 28 Oct 2026 16:53:31 -0400 Subject: [PATCH 1/5] ref(seer): Use lifecycle locking for workflow configuration Store workflow identity as IDs and load fresh configuration for reads and locked mutations. Update callers and preserve batched config serialization. --- .../organization_seer_workflow_configs.py | 8 +- src/sentry/seer/workflows/action.py | 16 +- src/sentry/seer/workflows/breached_metrics.py | 36 ++--- src/sentry/seer/workflows/workflow.py | 120 ++++++++------ tests/sentry/backup/test_exports.py | 4 +- tests/sentry/backup/test_imports.py | 6 +- .../seer/workflows/test_breached_metrics.py | 44 +++-- tests/sentry/seer/workflows/test_runs.py | 32 ++-- tests/sentry/seer/workflows/test_workflow.py | 152 ++++++++++++++---- 9 files changed, 271 insertions(+), 147 deletions(-) diff --git a/src/sentry/seer/endpoints/organization_seer_workflow_configs.py b/src/sentry/seer/endpoints/organization_seer_workflow_configs.py index 0ba2c7f62b40..a593a92b2e35 100644 --- a/src/sentry/seer/endpoints/organization_seer_workflow_configs.py +++ b/src/sentry/seer/endpoints/organization_seer_workflow_configs.py @@ -26,10 +26,11 @@ from sentry.seer.models.workflow import ( SEER_CONFIGURABLE_WORKFLOW_STRATEGIES, SeerWorkflowConfig, + SeerWorkflowConfigExtras, SeerWorkflowConfigProject, + SeerWorkflowFilters, SeerWorkflowRun, ) -from sentry.seer.workflows.workflow import SeerWorkflow class SeerWorkflowConfigResponse(TypedDict): @@ -198,8 +199,7 @@ def get_config_last_runs( def serialize_workflow_config( config: SeerWorkflowConfig, *, selected_project_ids: list[int], last_run: datetime | None ) -> SeerWorkflowConfigResponse: - workflow = SeerWorkflow(config) - filters = workflow.filters + filters = SeerWorkflowFilters.parse_obj(config.filters) result: SeerWorkflowConfigResponse = { "id": str(config.id), "strategy": config.strategy, @@ -217,7 +217,7 @@ def serialize_workflow_config( snake_to_camel_case, ), "outputPolicy": config.output_policy, - "seededDefault": workflow.extras.seeded_default, + "seededDefault": SeerWorkflowConfigExtras.parse_obj(config.extras).seeded_default, "dateAdded": config.date_added, "dateUpdated": config.date_updated, } diff --git a/src/sentry/seer/workflows/action.py b/src/sentry/seer/workflows/action.py index f592ed97ce2f..82398f974ec4 100644 --- a/src/sentry/seer/workflows/action.py +++ b/src/sentry/seer/workflows/action.py @@ -44,20 +44,18 @@ def execute(invocation: ActionInvocation) -> None: "workflow_id": invocation.workflow_id, "action_id": invocation.action.id, } - binding = ( - SeerWorkflowTriggerBinding.objects.filter( - workflow_config__organization_id=group.project.organization_id, - trigger_action=invocation.action, - ) - .select_related("workflow_config") - .first() - ) + binding = SeerWorkflowTriggerBinding.objects.filter( + workflow_config__organization_id=group.project.organization_id, + trigger_action=invocation.action, + ).first() if binding is None: _record_trigger_outcome("ignored", extra, reason="missing_binding") return extra.update(workflow_config_id=binding.workflow_config_id, binding_id=binding.id) - admission = BreachedMetricWorkflow(binding.workflow_config).accept( + admission = BreachedMetricWorkflow( + config_id=binding.workflow_config_id, organization_id=group.project.organization_id + ).accept( group, event, binding=binding, diff --git a/src/sentry/seer/workflows/breached_metrics.py b/src/sentry/seer/workflows/breached_metrics.py index 05ad72f01898..5dc216dcb484 100644 --- a/src/sentry/seer/workflows/breached_metrics.py +++ b/src/sentry/seer/workflows/breached_metrics.py @@ -5,7 +5,6 @@ from datetime import datetime from typing import Annotated, Any, Literal -from django.db import router, transaction from django.utils import timezone from pydantic import BaseModel, Field, PositiveInt, ValidationError @@ -115,7 +114,7 @@ def create( enabled=enabled, filters=SeerWorkflowFilters(all_projects=True, environments=list(environments)).dict(), ) - return cls(config) + return cls(config_id=config.id, organization_id=organization_id) def accept( self, @@ -128,8 +127,10 @@ def accept( """Check eligibility, then record one admission decision per issue and config.""" if not features.has("organizations:seer-automations", group.organization): return IgnoredMetricBreach("feature_disabled") - if binding.workflow_config_id != self.config.id: + if binding.workflow_config_id != self.config_id: return IgnoredMetricBreach("workflow_unavailable") + if group.organization.id != self.organization_id: + return IgnoredMetricBreach("configuration_unavailable") if group.type != MetricIssue.type_id or not isinstance(event, GroupEvent): return IgnoredMetricBreach("unsupported_trigger") occurrence = event.occurrence @@ -159,23 +160,18 @@ def accept( except ValidationError: return IgnoredMetricBreach("invalid_source") - with transaction.atomic(using=router.db_for_write(SeerWorkflowRun)): - config = ( - SeerWorkflowConfig.objects.select_for_update() - .filter( - id=self.config.id, - organization_id=group.organization.id, - strategy=SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION, - enabled=True, - ) - .first() - ) - if config is None: - return IgnoredMetricBreach("configuration_unavailable") - self.config = config - if not self.matches_scope(group.project_id, trigger.environment): - return IgnoredMetricBreach("out_of_scope") - return self._record_run(group, trigger) + try: + with self._lock_config() as config: + if ( + config.strategy != SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION + or not config.enabled + ): + return IgnoredMetricBreach("configuration_unavailable") + if not self._matches_scope(config, group.project_id, trigger.environment): + return IgnoredMetricBreach("out_of_scope") + return self._record_run(group, trigger) + except SeerWorkflowConfig.DoesNotExist: + return IgnoredMetricBreach("configuration_unavailable") def _record_run( self, diff --git a/src/sentry/seer/workflows/workflow.py b/src/sentry/seer/workflows/workflow.py index 0e54be2ae263..036e9ae4a340 100644 --- a/src/sentry/seer/workflows/workflow.py +++ b/src/sentry/seer/workflows/workflow.py @@ -1,6 +1,7 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Iterator, Mapping +from contextlib import contextmanager from dataclasses import dataclass from typing import Any, assert_never @@ -42,16 +43,22 @@ class SelectedProjects: type ProjectSelection = AllProjects | SelectedProjects -@dataclass +@dataclass(frozen=True, kw_only=True) class SeerWorkflow: - """A Seer automation backed by a saved workflow configuration.""" + """Apply each change to fresh state under the workflow config's row lock. - config: SeerWorkflowConfig + Returned models do not carry a lock or refresh automatically. Configuration + changes, including project selection, should go through these methods. + """ + + config_id: int + organization_id: int @classmethod def ensure_defaults(cls, organization_id: int) -> None: """Seed defaults for strategies with no config; disabled configs count as existing.""" - configs = SeerWorkflowConfig.objects.filter( + database = router.db_for_write(SeerWorkflowConfig) + configs = SeerWorkflowConfig.objects.using(database).filter( organization_id=organization_id, strategy__in=[s.value for s in SEER_DEFAULT_WORKFLOW_STRATEGIES], ) @@ -59,11 +66,13 @@ def ensure_defaults(cls, organization_id: int) -> None: if all(s.value in existing for s in SEER_DEFAULT_WORKFLOW_STRATEGIES): return - with transaction.atomic(router.db_for_write(SeerWorkflowConfig)): + with transaction.atomic(database): # Prevent duplicate seeds without a uniqueness constraint on configurable strategies. - list(Organization.objects.select_for_update().filter(id=organization_id)) + list( + Organization.objects.using(database).select_for_update().filter(id=organization_id) + ) existing = set(configs.values_list("strategy", flat=True)) - SeerWorkflowConfig.objects.bulk_create( + SeerWorkflowConfig.objects.using(database).bulk_create( [ SeerWorkflowConfig( organization_id=organization_id, @@ -78,13 +87,17 @@ def ensure_defaults(cls, organization_id: int) -> None: ] ) + def get_config(self) -> SeerWorkflowConfig: + """Fetch the config without acquiring a lock.""" + return self._config_queryset().get() + @property def extras(self) -> SeerWorkflowConfigExtras: - return SeerWorkflowConfigExtras.parse_obj(self.config.extras) + return SeerWorkflowConfigExtras.parse_obj(self.get_config().extras) @property def filters(self) -> SeerWorkflowFilters: - return SeerWorkflowFilters.parse_obj(self.config.filters) + return SeerWorkflowFilters.parse_obj(self.get_config().filters) def get_or_create_run( self, @@ -97,10 +110,7 @@ def get_or_create_run( if not trigger_key or len(trigger_key) > 256: raise ValueError("Trigger keys must contain between 1 and 256 characters") - with transaction.atomic(router.db_for_write(SeerWorkflowRun)): - self.config = SeerWorkflowConfig.objects.select_for_update().get( - id=self.config.id, organization_id=self.config.organization_id - ) + with self._lock_config() as config: existing = self.get_runs().filter(trigger_key=trigger_key).first() if existing is not None: return WorkflowRunCreationResult( @@ -111,17 +121,17 @@ def get_or_create_run( ) snapshot = WorkflowRunSnapshot( - config_version=self.config.version, - strategy=SeerWorkflowStrategy(self.config.strategy), - filters=self.config.filters, - project_ids=tuple(self.get_project_ids()), - output_policy=self.config.output_policy, + config_version=config.version, + strategy=SeerWorkflowStrategy(config.strategy), + filters=config.filters, + project_ids=tuple(self._get_project_ids(config)), + output_policy=config.output_policy, inputs=json.loads(inputs.json()), ) run_extras = SeerWorkflowRunExtras.parse_obj({**(extras or {}), "snapshot": snapshot}) run = SeerWorkflowRun.objects.create( - organization_id=self.config.organization_id, - workflow_config=self.config, + organization_id=config.organization_id, + workflow_config=config, trigger_key=trigger_key, status=SeerWorkflowRunStatus.QUEUED, extras=json.loads(run_extras.json()), @@ -134,17 +144,15 @@ def get_or_create_run( ) def get_runs(self) -> QuerySet[SeerWorkflowRun]: - return SeerWorkflowRun.objects.filter( - organization_id=self.config.organization_id, workflow_config_id=self.config.id - ).order_by("-date_added", "-id") + return ( + SeerWorkflowRun.objects.using(router.db_for_write(SeerWorkflowRun)) + .filter(organization_id=self.organization_id, workflow_config_id=self.config_id) + .order_by("-date_added", "-id") + ) def matches_scope(self, project_id: int, environment: str | None) -> bool: """Check the automation's scope before recording an admission decision.""" - return ( - self.config.enabled - and (not self.filters.environments or environment in self.filters.environments) - and self._get_projects().filter(id=project_id).exists() - ) + return self._matches_scope(self.get_config(), project_id, environment) def has_quota_available(self) -> bool: """Allow runs until automation quota enforcement is implemented.""" @@ -153,14 +161,11 @@ def has_quota_available(self) -> bool: def get_project_ids(self) -> list[int]: """Return active project IDs matching this workflow's filters, in ascending order.""" - return list(self._get_projects().order_by("id").values_list("id", flat=True)) + return self._get_project_ids(self.get_config()) def set_service_account(self, service_account_id: int | None) -> None: """Store the control account ID after validating its current cell replica.""" - with transaction.atomic(router.db_for_write(SeerWorkflowConfig)): - config = SeerWorkflowConfig.objects.select_for_update().get( - id=self.config.id, organization_id=self.config.organization_id - ) + with self._lock_config() as config: if service_account_id is not None: try: replica = ServiceAccountReplica.objects.get( @@ -179,16 +184,15 @@ def set_service_account(self, service_account_id: int | None) -> None: config.version += 1 config.save(update_fields=["service_account_id", "version", "date_updated"]) - self.config = config - def resolve_service_account(self) -> ServiceAccountReplica: """Resolve authoritative identity against current cell state on every call.""" - if self.config.service_account_id is None: + config = self.get_config() + if config.service_account_id is None: raise ValueError("Workflow has no service account") try: replica = ServiceAccountReplica.objects.get( - service_account_id=self.config.service_account_id, - organization_id=self.config.organization_id, + service_account_id=config.service_account_id, + organization_id=config.organization_id, ) except ServiceAccountReplica.DoesNotExist as exc: raise ValueError( @@ -209,10 +213,7 @@ def set_project_selection(self, selection: ProjectSelection) -> None: case _: assert_never(selection) - with transaction.atomic(router.db_for_write(SeerWorkflowConfig)): - config = SeerWorkflowConfig.objects.select_for_update().get( - id=self.config.id, organization_id=self.config.organization_id - ) + with self._lock_config() as config: if ( config.strategy == SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION and not all_projects @@ -239,17 +240,36 @@ def set_project_selection(self, selection: ProjectSelection) -> None: config.version += 1 config.save(update_fields=["filters", "version", "date_updated"]) - self.config = config + @contextmanager + def _lock_config(self) -> Iterator[SeerWorkflowConfig]: + with transaction.atomic(router.db_for_write(SeerWorkflowConfig)): + yield self._config_queryset().select_for_update().get() + + def _config_queryset(self) -> QuerySet[SeerWorkflowConfig]: + return SeerWorkflowConfig.objects.using(router.db_for_write(SeerWorkflowConfig)).filter( + id=self.config_id, organization_id=self.organization_id + ) - def _get_projects(self) -> QuerySet[Project]: + def _matches_scope( + self, config: SeerWorkflowConfig, project_id: int, environment: str | None + ) -> bool: + filters = SeerWorkflowFilters.parse_obj(config.filters) + return ( + config.enabled + and (not filters.environments or environment in filters.environments) + and self._get_projects(config).filter(id=project_id).exists() + ) + + def _get_project_ids(self, config: SeerWorkflowConfig) -> list[int]: + return list(self._get_projects(config).order_by("id").values_list("id", flat=True)) + + def _get_projects(self, config: SeerWorkflowConfig) -> QuerySet[Project]: projects = Project.objects.filter( - organization_id=self.config.organization_id, status=ObjectStatus.ACTIVE + organization_id=config.organization_id, status=ObjectStatus.ACTIVE ) - if not self.filters.all_projects: + if not SeerWorkflowFilters.parse_obj(config.filters).all_projects: projects = projects.filter( - id__in=SeerWorkflowConfigProject.objects.filter(config=self.config).values( - "project_id" - ) + id__in=SeerWorkflowConfigProject.objects.filter(config=config).values("project_id") ) return projects diff --git a/tests/sentry/backup/test_exports.py b/tests/sentry/backup/test_exports.py index d9a301d6b3b9..0c9203ea3a38 100644 --- a/tests/sentry/backup/test_exports.py +++ b/tests/sentry/backup/test_exports.py @@ -787,7 +787,9 @@ def test_export_workflow_with_service_account(self) -> None: org, strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2 ) Factories.create_seer_workflow_config_project(config, project) - SeerWorkflow(config).set_service_account(account.id) + SeerWorkflow( + config_id=config.id, organization_id=config.organization_id + ).set_service_account(account.id) for scope in (ExportScope.Organization, ExportScope.Global): with TemporaryDirectory() as tmp_dir: diff --git a/tests/sentry/backup/test_imports.py b/tests/sentry/backup/test_imports.py index 39ff43a14bb1..a65040ee1881 100644 --- a/tests/sentry/backup/test_imports.py +++ b/tests/sentry/backup/test_imports.py @@ -2575,7 +2575,7 @@ def test_account_identity_remapped_before_replication(self) -> None: assert config.version == original.version assert SeerWorkflowConfig.objects.filter(service_account_id=None).count() == 1 inactive_config = SeerWorkflowConfig.objects.get(service_account_id=inactive.id) - workflow = SeerWorkflow(config) + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) assert workflow.get_project_ids() == [project.id] # Shared databases may replicate during import; explicitly simulate delayed delivery. ServiceAccountReplica.objects.filter( @@ -2600,7 +2600,9 @@ def test_account_identity_remapped_before_replication(self) -> None: ) ) == [project.id] with pytest.raises(ValueError, match="inactive"): - SeerWorkflow(inactive_config).resolve_service_account() + SeerWorkflow( + config_id=inactive_config.id, organization_id=inactive_config.organization_id + ).resolve_service_account() config.refresh_from_db() assert config.version == original.version diff --git a/tests/sentry/seer/workflows/test_breached_metrics.py b/tests/sentry/seer/workflows/test_breached_metrics.py index b4cd3bf170e3..21f5151bd17a 100644 --- a/tests/sentry/seer/workflows/test_breached_metrics.py +++ b/tests/sentry/seer/workflows/test_breached_metrics.py @@ -58,7 +58,7 @@ def setUp(self) -> None: create_open_period=False, ) self.action = self.create_action(type=Action.Type.SEER_WORKFLOW) - self.config = BreachedMetricWorkflow.create(self.organization.id, enabled=True).config + self.config = BreachedMetricWorkflow.create(self.organization.id, enabled=True).get_config() Factories.create_seer_workflow_trigger_binding(self.config, self.action) def breach_event(self, *, detector_id: int, value: int = 180) -> GroupEvent: @@ -93,23 +93,25 @@ class BreachedMetricWorkflowAdmissionTest(BreachedMetricWorkflowTestCase): def test_creates_all_projects_config_preserving_environment_filter(self) -> None: workflow = BreachedMetricWorkflow.create(self.organization.id, environments=["production"]) - workflow.config.refresh_from_db() - assert workflow.config.strategy == SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION - assert workflow.config.enabled is False - assert workflow.config.filters == {"all_projects": True, "environments": ["production"]} + config = workflow.get_config() + assert config.strategy == SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION + assert config.enabled is False + assert config.filters == {"all_projects": True, "environments": ["production"]} def test_project_selection_is_managed_by_connected_monitors(self) -> None: with pytest.raises(ValueError, match="selected by connecting monitors"): - SeerWorkflow(self.config).set_project_selection( - SelectedProjects(project_ids=(self.project.id,)) - ) + SeerWorkflow( + config_id=self.config.id, organization_id=self.config.organization_id + ).set_project_selection(SelectedProjects(project_ids=(self.project.id,))) self.config.refresh_from_db() assert self.config.filters["all_projects"] is True def test_replaying_old_metadata_logs_validation_error_without_replacing_run(self) -> None: event = self.breach_event(detector_id=1) - workflow = BreachedMetricWorkflow(self.config) + workflow = BreachedMetricWorkflow( + config_id=self.config.id, organization_id=self.config.organization_id + ) result = workflow.accept( self.group, event, binding=self.action.seer_workflow_binding, workflow_id=1 ) @@ -141,7 +143,9 @@ def test_replaying_old_metadata_logs_validation_error_without_replacing_run(self def test_failed_execution_creation_rolls_back_admission(self) -> None: event = self.breach_event(detector_id=1) - workflow = BreachedMetricWorkflow(self.config) + workflow = BreachedMetricWorkflow( + config_id=self.config.id, organization_id=self.config.organization_id + ) with ( mock.patch( "sentry.seer.workflows.runs.SeerWorkflowRunExecution.objects.bulk_create", @@ -156,7 +160,9 @@ def test_failed_execution_creation_rolls_back_admission(self) -> None: assert not SeerWorkflowRunExecution.objects.exists() def test_accept_uses_latest_scope_from_the_locked_config(self) -> None: - workflow = BreachedMetricWorkflow(self.config) + workflow = BreachedMetricWorkflow( + config_id=self.config.id, organization_id=self.config.organization_id + ) SeerWorkflowConfig.objects.filter(id=self.config.id).update( filters={"all_projects": True, "environments": ["staging"]} ) @@ -174,7 +180,9 @@ def test_accept_uses_latest_scope_from_the_locked_config(self) -> None: def test_ignored_configuration_reports_reason_without_starting_work(self) -> None: self.config.update(enabled=False) - rejected = BreachedMetricWorkflow(self.config).accept( + rejected = BreachedMetricWorkflow( + config_id=self.config.id, organization_id=self.config.organization_id + ).accept( self.group, self.breach_event(detector_id=1), binding=self.action.seer_workflow_binding, @@ -190,7 +198,9 @@ def test_ignored_malformed_source_reports_reason(self) -> None: assert event.occurrence is not None event.occurrence = replace(event.occurrence, evidence_data={}) - rejected = BreachedMetricWorkflow(self.config).accept( + rejected = BreachedMetricWorkflow( + config_id=self.config.id, organization_id=self.config.organization_id + ).accept( self.group, event, binding=self.action.seer_workflow_binding, @@ -210,7 +220,9 @@ def test_other_organization_configuration_does_not_accept_the_breach(self) -> No trigger_actions=[other_action], ) - result = BreachedMetricWorkflow(other).accept( + result = BreachedMetricWorkflow( + config_id=other.id, organization_id=other.organization_id + ).accept( self.group, self.breach_event(detector_id=1), binding=other_action.seer_workflow_binding, @@ -345,7 +357,9 @@ def test_distinct_actions_dispatch_but_shared_config_creates_one_run(self) -> No shared_action = self.create_action(type=Action.Type.SEER_WORKFLOW) Factories.create_seer_workflow_trigger_binding(self.config, shared_action) other_action = self.create_action(type=Action.Type.SEER_WORKFLOW) - other_config = BreachedMetricWorkflow.create(self.organization.id, enabled=True).config + other_config = BreachedMetricWorkflow.create( + self.organization.id, enabled=True + ).get_config() Factories.create_seer_workflow_trigger_binding(other_config, other_action) self.create_data_condition_group_action( action=shared_action, condition_group=self.action_conditions diff --git a/tests/sentry/seer/workflows/test_runs.py b/tests/sentry/seer/workflows/test_runs.py index d3ab200e3106..51f19b15d157 100644 --- a/tests/sentry/seer/workflows/test_runs.py +++ b/tests/sentry/seer/workflows/test_runs.py @@ -1,4 +1,3 @@ -from copy import copy from functools import partial from unittest.mock import Mock, patch @@ -191,13 +190,12 @@ def create_run(self) -> SeerRun: class SeerWorkflowRunLifecycleTest(TestCase): def setUp(self) -> None: super().setUp() - self.workflow = SeerWorkflow( - Factories.create_seer_workflow_config( - self.organization, - strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2, - filters={"all_projects": True}, - ) + config = Factories.create_seer_workflow_config( + self.organization, + strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2, + filters={"all_projects": True}, ) + self.workflow = SeerWorkflow(config_id=config.id, organization_id=self.organization.id) def test_config_changes_only_apply_to_new_runs(self) -> None: original_project = self.create_project() @@ -214,7 +212,10 @@ def test_config_changes_only_apply_to_new_runs(self) -> None: assert snapshot.inputs == {"issue_limit": 10} new_project = self.create_project() - self.workflow.set_project_selection(SelectedProjects(project_ids=(new_project.id,))) + other = SeerWorkflow( + config_id=self.workflow.config_id, organization_id=self.organization.id + ) + other.set_project_selection(SelectedProjects(project_ids=(new_project.id,))) retry_result = self.workflow.get_or_create_run( trigger_key="manual:first", inputs=RunInputs(issue_limit=20) ) @@ -234,10 +235,11 @@ def test_config_changes_only_apply_to_new_runs(self) -> None: assert new_run.extras.snapshot.inputs == {"issue_limit": 20} def test_request_identity_and_history_are_scoped_to_config_and_organization(self) -> None: + other_config = Factories.create_seer_workflow_config( + self.organization, strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2 + ) other_workflow = SeerWorkflow( - Factories.create_seer_workflow_config( - self.organization, strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2 - ) + config_id=other_config.id, organization_id=self.organization.id ) first_result = self.workflow.get_or_create_run( trigger_key="event:occurrence", inputs=RunInputs(issue_limit=10) @@ -250,15 +252,15 @@ def test_request_identity_and_history_are_scoped_to_config_and_organization(self assert second_result.created is True assert first.run_id != second.run_id legacy = Factories.create_seer_workflow_run( - self.organization, workflow_config=self.workflow.config + self.organization, workflow_config=self.workflow.get_config() ) assert list(self.workflow.get_runs()) == [legacy, first.get_run()] assert list(other_workflow.get_runs()) == [second.get_run()] other_org = self.create_organization() - wrong_org_config = copy(self.workflow.config) - wrong_org_config.organization_id = other_org.id - wrong_org_workflow = SeerWorkflow(wrong_org_config) + wrong_org_workflow = SeerWorkflow( + config_id=self.workflow.config_id, organization_id=other_org.id + ) assert not wrong_org_workflow.get_runs().exists() with pytest.raises(SeerWorkflowConfig.DoesNotExist): wrong_org_workflow.get_or_create_run( diff --git a/tests/sentry/seer/workflows/test_workflow.py b/tests/sentry/seer/workflows/test_workflow.py index 810a9fab9d70..3f98d71a43d1 100644 --- a/tests/sentry/seer/workflows/test_workflow.py +++ b/tests/sentry/seer/workflows/test_workflow.py @@ -1,6 +1,7 @@ from unittest.mock import patch import pytest +from django.db import router, transaction from sentry.constants import ObjectStatus from sentry.deletions.tasks.hybrid_cloud import schedule_hybrid_cloud_foreign_key_jobs @@ -35,7 +36,7 @@ def test_seeds_defaults_once_per_organization(self) -> None: later = self.create_project() SeerWorkflow.ensure_defaults(self.organization.id) - workflow = SeerWorkflow(config) + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) assert self.v2_configs() == [config] assert self.v2_configs(other) == [] assert workflow.filters.all_projects is True @@ -61,8 +62,9 @@ def test_existing_disabled_config_is_not_reseeded_or_changed(self) -> None: assert config.enabled is False assert config.filters == {"environments": ["production"]} assert config.version == 3 - assert SeerWorkflow(config).extras.seeded_default is None - assert SeerWorkflow(config).get_project_ids() == [selected.id] + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) + assert workflow.extras.seeded_default is None + assert workflow.get_project_ids() == [selected.id] def v2_configs(self, organization: Organization | None = None) -> list[SeerWorkflowConfig]: return list( @@ -83,12 +85,14 @@ def test_scope_requires_active_project_in_organization(self) -> None: ) selected = self.create_project(organization=self.organization) Factories.create_seer_workflow_config_project(config, selected) - workflow = SeerWorkflow(config) + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) assert workflow.matches_scope(selected.id, None) is True later = self.create_project(organization=self.organization) assert workflow.matches_scope(later.id, None) is False - workflow.set_project_selection(AllProjects()) + other = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) + other.set_project_selection(AllProjects()) + assert workflow.filters.all_projects is True assert workflow.matches_scope(later.id, None) is True foreign = self.create_project(organization=self.create_organization()) assert workflow.matches_scope(foreign.id, None) is False @@ -104,21 +108,59 @@ def test_scope_requires_enabled_config_and_matching_environment(self) -> None: filters={"all_projects": True, "environments": ["production"]}, ) - workflow = SeerWorkflow(config) + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) assert workflow.matches_scope(self.project.id, "production") is True assert workflow.matches_scope(self.project.id, "staging") is False assert workflow.matches_scope(self.project.id, None) is False - config.update(enabled=False) + SeerWorkflowConfig.objects.filter(id=config.id).update(enabled=False) assert workflow.matches_scope(self.project.id, "production") is False + def test_reads_fresh_config_without_changing_previous_results(self) -> None: + config = Factories.create_seer_workflow_config( + self.organization, strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2 + ) + workflow = SeerWorkflow(config_id=config.id, organization_id=self.organization.id) + original = workflow.get_config() + + SeerWorkflowConfig.objects.filter(id=config.id).update( + filters={"all_projects": True, "environments": ["production"]}, + extras={"seeded_default": SeerWorkflowStrategy.AGENTIC_TRIAGE_V2}, + version=2, + ) + + assert original.version == 1 + assert workflow.get_config().version == 2 + assert workflow.filters.all_projects is True + assert workflow.filters.environments == ["production"] + assert workflow.extras.seeded_default == SeerWorkflowStrategy.AGENTIC_TRIAGE_V2 + + def test_reads_and_changes_require_matching_organization(self) -> None: + config = Factories.create_seer_workflow_config( + self.organization, strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2 + ) + workflow = SeerWorkflow(config_id=config.id, organization_id=self.create_organization().id) + + with pytest.raises(SeerWorkflowConfig.DoesNotExist): + workflow.get_config() + with pytest.raises(SeerWorkflowConfig.DoesNotExist): + workflow.get_project_ids() + with pytest.raises(SeerWorkflowConfig.DoesNotExist): + workflow.set_project_selection(AllProjects()) + with pytest.raises(SeerWorkflowConfig.DoesNotExist): + workflow.set_service_account(None) + + config.refresh_from_db() + assert config.version == 1 + class SeerWorkflowProjectSelectionTest(TestCase): def test_selected_projects_exclude_inactive_and_deleted_projects(self) -> None: first, second = self.create_project(), self.create_project() self.create_project() - workflow = SeerWorkflow(self.create_config(first, second)) + config = self.create_config(first, second) + workflow = SeerWorkflow(config_id=config.id, organization_id=self.organization.id) assert workflow.get_project_ids() == [first.id, second.id] first.update(status=ObjectStatus.PENDING_DELETION) @@ -132,7 +174,7 @@ def test_replaces_selection_preserving_latest_filters_and_version(self) -> None: previous, selected = self.create_project(), self.create_project() config = self.create_config(previous) other_config = self.create_config(previous) - workflow = SeerWorkflow(config) + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) SeerWorkflowConfig.objects.filter(id=config.id, organization=self.organization).update( filters=SeerWorkflowFilters(environments=["production"]).dict() | {"min_event_count": 10}, @@ -147,16 +189,18 @@ def test_replaces_selection_preserving_latest_filters_and_version(self) -> None: "environments": ["production"], "min_event_count": 10, } - assert config.version == workflow.config.version == 4 + assert config.version == workflow.get_config().version == 4 assert workflow.get_project_ids() == [selected.id] - assert SeerWorkflow(other_config).get_project_ids() == [previous.id] + assert SeerWorkflow( + config_id=other_config.id, organization_id=other_config.organization_id + ).get_project_ids() == [previous.id] def test_switching_to_all_projects_clears_links_and_scopes_active_projects(self) -> None: selected = self.create_project() self.create_project(organization=self.create_organization()) self.create_project(status=ObjectStatus.PENDING_DELETION) config = self.create_config(selected) - workflow = SeerWorkflow(config) + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) initial_filters = workflow.filters assert initial_filters.all_projects is False @@ -164,7 +208,12 @@ def test_switching_to_all_projects_clears_links_and_scopes_active_projects(self) later = self.create_project() config.refresh_from_db() - assert SeerWorkflow(config).filters.all_projects is True + assert ( + SeerWorkflow( + config_id=config.id, organization_id=config.organization_id + ).filters.all_projects + is True + ) assert workflow.filters.all_projects is True assert workflow.get_project_ids() == [selected.id, later.id] assert not SeerWorkflowConfigProject.objects.filter(config=config).exists() @@ -173,12 +222,17 @@ def test_switching_to_all_projects_clears_links_and_scopes_active_projects(self) def test_empty_selection_switches_from_all_to_no_projects(self) -> None: self.create_project() config = self.create_config(all_projects=True) - workflow = SeerWorkflow(config) + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) workflow.set_project_selection(SelectedProjects(project_ids=())) config.refresh_from_db() - assert SeerWorkflow(config).filters.all_projects is False + assert ( + SeerWorkflow( + config_id=config.id, organization_id=config.organization_id + ).filters.all_projects + is False + ) assert workflow.get_project_ids() == [] assert not SeerWorkflowConfigProject.objects.filter(config=config).exists() assert config.version == 2 @@ -187,7 +241,7 @@ def test_rejects_cross_organization_projects_without_changing_selection(self) -> previous, valid = self.create_project(), self.create_project() foreign = self.create_project(organization=self.create_organization()) config = self.create_config(previous) - workflow = SeerWorkflow(config) + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) with pytest.raises(ValueError, match="Selected projects must exist"): workflow.set_project_selection(SelectedProjects(project_ids=(valid.id, foreign.id))) @@ -200,7 +254,7 @@ def test_rejects_cross_organization_projects_without_changing_selection(self) -> def test_failed_insert_restores_previous_selection(self) -> None: previous, selected = self.create_project(), self.create_project() config = self.create_config(previous) - workflow = SeerWorkflow(config) + workflow = SeerWorkflow(config_id=config.id, organization_id=config.organization_id) with ( patch.object( @@ -215,6 +269,23 @@ def test_failed_insert_restores_previous_selection(self) -> None: assert config.filters == {"all_projects": False} assert config.version == 1 + def test_outer_rollback_does_not_leave_stale_config(self) -> None: + previous = self.create_project() + config = self.create_config(previous) + workflow = SeerWorkflow(config_id=config.id, organization_id=self.organization.id) + + with ( + pytest.raises(RuntimeError, match="failed"), + transaction.atomic(router.db_for_write(SeerWorkflowConfig)), + ): + workflow.set_project_selection(AllProjects()) + assert workflow.filters.all_projects is True + raise RuntimeError("failed") + + assert workflow.filters.all_projects is False + assert workflow.get_config().version == 1 + assert workflow.get_project_ids() == [previous.id] + def create_config(self, *selected: Project, all_projects: bool = False) -> SeerWorkflowConfig: config = Factories.create_seer_workflow_config( self.organization, @@ -236,7 +307,9 @@ def setUp(self) -> None: filters={"environments": ["production"]}, output_policy={"notify": True}, ) - self.workflow = SeerWorkflow(self.config) + self.workflow = SeerWorkflow( + config_id=self.config.id, organization_id=self.config.organization_id + ) with outbox_runner(): self.account = Factories.create_service_account(self.organization, id=910001) @@ -254,7 +327,7 @@ def test_assigns_authoritative_identity_and_preserves_latest_settings(self) -> N self.config.refresh_from_db() assert self.config.service_account_id == self.account.id - assert self.config.version == self.workflow.config.version == 5 + assert self.config.version == self.workflow.get_config().version == 5 assert self.config.enabled is True assert self.config.filters == {"environments": ["production"]} assert self.config.output_policy == {"notify": True} @@ -262,19 +335,33 @@ def test_assigns_authoritative_identity_and_preserves_latest_settings(self) -> N def test_assignment_versions_only_changes(self) -> None: self.workflow.set_service_account(None) - assert self.workflow.config.version == 1 + assert self.workflow.get_config().version == 1 self.workflow.set_service_account(self.account.id) self.workflow.set_service_account(self.account.id) - assert self.workflow.config.version == 2 + assert self.workflow.get_config().version == 2 with outbox_runner(): replacement = Factories.create_service_account(self.organization, name="replacement") self.workflow.set_service_account(replacement.id) - assert self.workflow.config.version == 3 + assert self.workflow.get_config().version == 3 self.workflow.set_service_account(None) - assert self.workflow.config.version == 4 + assert self.workflow.get_config().version == 4 self.workflow.set_service_account(None) - assert self.workflow.config.version == 4 + assert self.workflow.get_config().version == 4 + + def test_resolution_reads_assignment_changes_from_another_instance(self) -> None: + other = SeerWorkflow(config_id=self.config.id, organization_id=self.organization.id) + other.set_service_account(self.account.id) + assert self.workflow.resolve_service_account().service_account_id == self.account.id + + with outbox_runner(): + replacement = Factories.create_service_account(self.organization, name="replacement") + other.set_service_account(replacement.id) + assert self.workflow.resolve_service_account().service_account_id == replacement.id + + other.set_service_account(None) + with pytest.raises(ValueError, match="no service account"): + self.workflow.resolve_service_account() def test_rejects_unavailable_and_inactive_accounts_without_mutation(self) -> None: self.workflow.set_service_account(self.account.id) @@ -340,11 +427,11 @@ def test_resolution_rejects_unset_and_differently_owned_associations(self) -> No self.workflow.resolve_service_account() with outbox_runner(): foreign = Factories.create_service_account(self.create_organization()) - self.workflow.config.service_account_id = foreign.id + self.config.update(service_account_id=foreign.id) with pytest.raises(ValueError, match="unavailable"): self.workflow.resolve_service_account() - def test_control_deletion_clears_associations_and_rejects_stale_wrapper(self) -> None: + def test_control_deletion_clears_associations_and_resolution_reads_current_config(self) -> None: selected = self.create_project() Factories.create_seer_workflow_config_project(self.config, selected) self.workflow.set_service_account(self.account.id) @@ -352,13 +439,16 @@ def test_control_deletion_clears_associations_and_rejects_stale_wrapper(self) -> self.organization, strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2, ) - SeerWorkflow(other).set_service_account(self.account.id) + SeerWorkflow(config_id=other.id, organization_id=other.organization_id).set_service_account( + self.account.id + ) with outbox_runner(): unrelated = Factories.create_service_account(self.organization, name="unrelated") + unrelated_config = Factories.create_seer_workflow_config( + self.organization, strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2 + ) unrelated_workflow = SeerWorkflow( - Factories.create_seer_workflow_config( - self.organization, strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE_V2 - ) + config_id=unrelated_config.id, organization_id=self.organization.id ) unrelated_workflow.set_service_account(unrelated.id) with assume_test_silo_mode(SiloMode.CONTROL), outbox_runner(): @@ -375,7 +465,7 @@ def test_control_deletion_clears_associations_and_rejects_stale_wrapper(self) -> assert self.config.version == 2 assert self.workflow.get_project_ids() == [selected.id] assert unrelated_workflow.resolve_service_account().service_account_id == unrelated.id - with pytest.raises(ValueError, match="unavailable"): + with pytest.raises(ValueError, match="no service account"): self.workflow.resolve_service_account() def test_assignment_during_delayed_account_deletion_is_eventually_cleared(self) -> None: From acfc9c94b1ac7662c98810958804c5ca86598d1c Mon Sep 17 00:00:00 2001 From: Trevor Elkins Date: Wed, 28 Oct 2026 17:13:46 -0400 Subject: [PATCH 2/5] perf(seer): Reuse locked config during breach admission --- src/sentry/seer/workflows/breached_metrics.py | 7 +- src/sentry/seer/workflows/workflow.py | 71 +++++++++++-------- .../seer/workflows/test_breached_metrics.py | 29 ++++++++ 3 files changed, 74 insertions(+), 33 deletions(-) diff --git a/src/sentry/seer/workflows/breached_metrics.py b/src/sentry/seer/workflows/breached_metrics.py index 5dc216dcb484..677fb5d4e619 100644 --- a/src/sentry/seer/workflows/breached_metrics.py +++ b/src/sentry/seer/workflows/breached_metrics.py @@ -169,12 +169,13 @@ def accept( return IgnoredMetricBreach("configuration_unavailable") if not self._matches_scope(config, group.project_id, trigger.environment): return IgnoredMetricBreach("out_of_scope") - return self._record_run(group, trigger) + return self._record_run(config, group, trigger) except SeerWorkflowConfig.DoesNotExist: return IgnoredMetricBreach("configuration_unavailable") def _record_run( self, + config: SeerWorkflowConfig, group: Group, trigger: MetricBreachTrigger, ) -> AcceptedMetricBreach | RejectedMetricBreach: @@ -195,8 +196,8 @@ def _record_run( if self.has_quota_available() else RejectedMetricBreachAdmission(reason="quota_exceeded", decided_at=decided_at) ) - result = self.get_or_create_run( - trigger_key=trigger_key, inputs=trigger, extras={"admission": admission.dict()} + result = self._get_or_create_run( + config, trigger_key=trigger_key, inputs=trigger, extras={"admission": admission.dict()} ) workflow_run = result.lifecycle if admission.status == "rejected": diff --git a/src/sentry/seer/workflows/workflow.py b/src/sentry/seer/workflows/workflow.py index 036e9ae4a340..db4eb7314c62 100644 --- a/src/sentry/seer/workflows/workflow.py +++ b/src/sentry/seer/workflows/workflow.py @@ -111,36 +111,8 @@ def get_or_create_run( raise ValueError("Trigger keys must contain between 1 and 256 characters") with self._lock_config() as config: - existing = self.get_runs().filter(trigger_key=trigger_key).first() - if existing is not None: - return WorkflowRunCreationResult( - lifecycle=SeerWorkflowRunLifecycle( - run_id=existing.id, organization_id=existing.organization_id - ), - created=False, - ) - - snapshot = WorkflowRunSnapshot( - config_version=config.version, - strategy=SeerWorkflowStrategy(config.strategy), - filters=config.filters, - project_ids=tuple(self._get_project_ids(config)), - output_policy=config.output_policy, - inputs=json.loads(inputs.json()), - ) - run_extras = SeerWorkflowRunExtras.parse_obj({**(extras or {}), "snapshot": snapshot}) - run = SeerWorkflowRun.objects.create( - organization_id=config.organization_id, - workflow_config=config, - trigger_key=trigger_key, - status=SeerWorkflowRunStatus.QUEUED, - extras=json.loads(run_extras.json()), - ) - return WorkflowRunCreationResult( - lifecycle=SeerWorkflowRunLifecycle( - run_id=run.id, organization_id=run.organization_id - ), - created=True, + return self._get_or_create_run( + config, trigger_key=trigger_key, inputs=inputs, extras=extras ) def get_runs(self) -> QuerySet[SeerWorkflowRun]: @@ -240,6 +212,45 @@ def set_project_selection(self, selection: ProjectSelection) -> None: config.version += 1 config.save(update_fields=["filters", "version", "date_updated"]) + def _get_or_create_run( + self, + config: SeerWorkflowConfig, + *, + trigger_key: str, + inputs: BaseModel, + extras: Mapping[str, Any] | None = None, + ) -> WorkflowRunCreationResult: + """Reuse the config while the caller holds its row lock.""" + existing = self.get_runs().filter(trigger_key=trigger_key).first() + if existing is not None: + return WorkflowRunCreationResult( + lifecycle=SeerWorkflowRunLifecycle( + run_id=existing.id, organization_id=existing.organization_id + ), + created=False, + ) + + snapshot = WorkflowRunSnapshot( + config_version=config.version, + strategy=SeerWorkflowStrategy(config.strategy), + filters=config.filters, + project_ids=tuple(self._get_project_ids(config)), + output_policy=config.output_policy, + inputs=json.loads(inputs.json()), + ) + run_extras = SeerWorkflowRunExtras.parse_obj({**(extras or {}), "snapshot": snapshot}) + run = SeerWorkflowRun.objects.create( + organization_id=config.organization_id, + workflow_config=config, + trigger_key=trigger_key, + status=SeerWorkflowRunStatus.QUEUED, + extras=json.loads(run_extras.json()), + ) + return WorkflowRunCreationResult( + lifecycle=SeerWorkflowRunLifecycle(run_id=run.id, organization_id=run.organization_id), + created=True, + ) + @contextmanager def _lock_config(self) -> Iterator[SeerWorkflowConfig]: with transaction.atomic(router.db_for_write(SeerWorkflowConfig)): diff --git a/tests/sentry/seer/workflows/test_breached_metrics.py b/tests/sentry/seer/workflows/test_breached_metrics.py index 21f5151bd17a..22d0f77f369c 100644 --- a/tests/sentry/seer/workflows/test_breached_metrics.py +++ b/tests/sentry/seer/workflows/test_breached_metrics.py @@ -7,6 +7,8 @@ from uuid import uuid4 import pytest +from django.db import connections, router +from django.test.utils import CaptureQueriesContext from pydantic import ValidationError from sentry.eventstream.base import GroupState @@ -177,6 +179,33 @@ def test_accept_uses_latest_scope_from_the_locked_config(self) -> None: assert result == IgnoredMetricBreach("out_of_scope") assert not SeerWorkflowRun.objects.exists() + def test_accept_reuses_locked_config_for_run_snapshot(self) -> None: + workflow = BreachedMetricWorkflow( + config_id=self.config.id, organization_id=self.organization.id + ) + event = self.breach_event(detector_id=1) + binding = self.action.seer_workflow_binding + SeerWorkflowConfig.objects.filter(id=self.config.id).update( + version=3, + filters={"all_projects": True, "environments": ["production"]}, + output_policy={"review_required": True}, + ) + + with CaptureQueriesContext(connections[router.db_for_write(SeerWorkflowConfig)]) as queries: + result = workflow.accept(self.group, event, binding=binding, workflow_id=1) + + assert isinstance(result, AcceptedMetricBreach) + assert result.created is True + snapshot = MetricBreachRunExtras.parse_obj(result.run.extras).snapshot + assert snapshot.config_version == 3 + assert snapshot.filters == {"all_projects": True, "environments": ["production"]} + assert snapshot.output_policy == {"review_required": True} + config_reads = [ + query["sql"] for query in queries if 'FROM "seer_workflowconfig"' in query["sql"] + ] + assert len(config_reads) == 1 + assert "FOR UPDATE" in config_reads[0] + def test_ignored_configuration_reports_reason_without_starting_work(self) -> None: self.config.update(enabled=False) From bae9d1c88f0553c782f772341fdafd70255e39ba Mon Sep 17 00:00:00 2001 From: Trevor Elkins Date: Wed, 28 Oct 2026 17:17:53 -0400 Subject: [PATCH 3/5] ref(seer): Make run creation lock requirement explicit --- src/sentry/seer/workflows/breached_metrics.py | 2 +- src/sentry/seer/workflows/workflow.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/sentry/seer/workflows/breached_metrics.py b/src/sentry/seer/workflows/breached_metrics.py index 677fb5d4e619..c97f69803c73 100644 --- a/src/sentry/seer/workflows/breached_metrics.py +++ b/src/sentry/seer/workflows/breached_metrics.py @@ -196,7 +196,7 @@ def _record_run( if self.has_quota_available() else RejectedMetricBreachAdmission(reason="quota_exceeded", decided_at=decided_at) ) - result = self._get_or_create_run( + result = self._get_or_create_run_with_locked_config( config, trigger_key=trigger_key, inputs=trigger, extras={"admission": admission.dict()} ) workflow_run = result.lifecycle diff --git a/src/sentry/seer/workflows/workflow.py b/src/sentry/seer/workflows/workflow.py index db4eb7314c62..6b9731ff2cc3 100644 --- a/src/sentry/seer/workflows/workflow.py +++ b/src/sentry/seer/workflows/workflow.py @@ -111,7 +111,7 @@ def get_or_create_run( raise ValueError("Trigger keys must contain between 1 and 256 characters") with self._lock_config() as config: - return self._get_or_create_run( + return self._get_or_create_run_with_locked_config( config, trigger_key=trigger_key, inputs=inputs, extras=extras ) @@ -212,7 +212,7 @@ def set_project_selection(self, selection: ProjectSelection) -> None: config.version += 1 config.save(update_fields=["filters", "version", "date_updated"]) - def _get_or_create_run( + def _get_or_create_run_with_locked_config( self, config: SeerWorkflowConfig, *, From 15b6ca64d29c6b039257d79b8f7b11a844d228f1 Mon Sep 17 00:00:00 2001 From: Trevor Elkins Date: Wed, 28 Oct 2026 17:28:07 -0400 Subject: [PATCH 4/5] ref(seer): Expose scoped workflow transactions --- src/sentry/seer/workflows/breached_metrics.py | 17 ++- src/sentry/seer/workflows/workflow.py | 144 +++++++++++------- tests/sentry/seer/workflows/test_runs.py | 47 +++++- 3 files changed, 146 insertions(+), 62 deletions(-) diff --git a/src/sentry/seer/workflows/breached_metrics.py b/src/sentry/seer/workflows/breached_metrics.py index c97f69803c73..9a89ac70e3b8 100644 --- a/src/sentry/seer/workflows/breached_metrics.py +++ b/src/sentry/seer/workflows/breached_metrics.py @@ -22,7 +22,7 @@ WorkflowRunError, ) from sentry.seer.workflows.runs import SeerWorkflowRunLifecycle -from sentry.seer.workflows.workflow import SeerWorkflow +from sentry.seer.workflows.workflow import SeerWorkflow, SeerWorkflowTransaction from sentry.services.eventstore.models import GroupEvent @@ -161,26 +161,27 @@ def accept( return IgnoredMetricBreach("invalid_source") try: - with self._lock_config() as config: + with self.transaction() as transaction: + config = transaction.config if ( config.strategy != SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION or not config.enabled ): return IgnoredMetricBreach("configuration_unavailable") - if not self._matches_scope(config, group.project_id, trigger.environment): + if not transaction.matches_scope(group.project_id, trigger.environment): return IgnoredMetricBreach("out_of_scope") - return self._record_run(config, group, trigger) + return self._record_run(transaction, group, trigger) except SeerWorkflowConfig.DoesNotExist: return IgnoredMetricBreach("configuration_unavailable") def _record_run( self, - config: SeerWorkflowConfig, + transaction: SeerWorkflowTransaction, group: Group, trigger: MetricBreachTrigger, ) -> AcceptedMetricBreach | RejectedMetricBreach: trigger_key = f"issue_created:{group.id}" - existing = self.get_runs().filter(trigger_key=trigger_key).first() + existing = transaction.get_runs().filter(trigger_key=trigger_key).first() if existing is not None: lifecycle = SeerWorkflowRunLifecycle( run_id=existing.id, organization_id=existing.organization_id @@ -196,8 +197,8 @@ def _record_run( if self.has_quota_available() else RejectedMetricBreachAdmission(reason="quota_exceeded", decided_at=decided_at) ) - result = self._get_or_create_run_with_locked_config( - config, trigger_key=trigger_key, inputs=trigger, extras={"admission": admission.dict()} + result = transaction.get_or_create_run( + trigger_key=trigger_key, inputs=trigger, extras={"admission": admission.dict()} ) workflow_run = result.lifecycle if admission.status == "rejected": diff --git a/src/sentry/seer/workflows/workflow.py b/src/sentry/seer/workflows/workflow.py index 6b9731ff2cc3..86a6fa1a3a51 100644 --- a/src/sentry/seer/workflows/workflow.py +++ b/src/sentry/seer/workflows/workflow.py @@ -1,8 +1,8 @@ from __future__ import annotations from collections.abc import Iterator, Mapping -from contextlib import contextmanager -from dataclasses import dataclass +from contextlib import AbstractContextManager, contextmanager +from dataclasses import dataclass, field from typing import Any, assert_never from django.db import router, transaction @@ -89,7 +89,7 @@ def ensure_defaults(cls, organization_id: int) -> None: def get_config(self) -> SeerWorkflowConfig: """Fetch the config without acquiring a lock.""" - return self._config_queryset().get() + return _config_queryset(self.config_id, self.organization_id).get() @property def extras(self) -> SeerWorkflowConfigExtras: @@ -107,24 +107,15 @@ def get_or_create_run( extras: Mapping[str, Any] | None = None, ) -> WorkflowRunCreationResult: """Retries preserve the original configuration and inputs.""" - if not trigger_key or len(trigger_key) > 256: - raise ValueError("Trigger keys must contain between 1 and 256 characters") - - with self._lock_config() as config: - return self._get_or_create_run_with_locked_config( - config, trigger_key=trigger_key, inputs=inputs, extras=extras - ) + with self.transaction() as locked: + return locked.get_or_create_run(trigger_key=trigger_key, inputs=inputs, extras=extras) def get_runs(self) -> QuerySet[SeerWorkflowRun]: - return ( - SeerWorkflowRun.objects.using(router.db_for_write(SeerWorkflowRun)) - .filter(organization_id=self.organization_id, workflow_config_id=self.config_id) - .order_by("-date_added", "-id") - ) + return _run_queryset(self.config_id, self.organization_id) def matches_scope(self, project_id: int, environment: str | None) -> bool: """Check the automation's scope before recording an admission decision.""" - return self._matches_scope(self.get_config(), project_id, environment) + return _matches_scope(self.get_config(), project_id, environment) def has_quota_available(self) -> bool: """Allow runs until automation quota enforcement is implemented.""" @@ -133,11 +124,12 @@ def has_quota_available(self) -> bool: def get_project_ids(self) -> list[int]: """Return active project IDs matching this workflow's filters, in ascending order.""" - return self._get_project_ids(self.get_config()) + return _get_project_ids(self.get_config()) def set_service_account(self, service_account_id: int | None) -> None: """Store the control account ID after validating its current cell replica.""" - with self._lock_config() as config: + with self.transaction() as locked: + config = locked.config if service_account_id is not None: try: replica = ServiceAccountReplica.objects.get( @@ -185,7 +177,8 @@ def set_project_selection(self, selection: ProjectSelection) -> None: case _: assert_never(selection) - with self._lock_config() as config: + with self.transaction() as locked: + config = locked.config if ( config.strategy == SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION and not all_projects @@ -212,15 +205,42 @@ def set_project_selection(self, selection: ProjectSelection) -> None: config.version += 1 config.save(update_fields=["filters", "version", "date_updated"]) - def _get_or_create_run_with_locked_config( + def transaction(self) -> AbstractContextManager[SeerWorkflowTransaction]: + """Share one locked config across an atomic workflow operation.""" + return SeerWorkflowTransaction.open( + config_id=self.config_id, organization_id=self.organization_id + ) + + +@dataclass +class SeerWorkflowTransaction: + """Workflow operations available only while the config's row lock is held.""" + + _config: SeerWorkflowConfig | None = field(default=None, init=False, repr=False) + + @classmethod + @contextmanager + def open(cls, *, config_id: int, organization_id: int) -> Iterator[SeerWorkflowTransaction]: + with transaction.atomic(router.db_for_write(SeerWorkflowConfig)): + locked = cls() + locked._config = _config_queryset(config_id, organization_id).select_for_update().get() + try: + yield locked + finally: + locked._config = None + + def get_or_create_run( self, - config: SeerWorkflowConfig, *, trigger_key: str, inputs: BaseModel, extras: Mapping[str, Any] | None = None, ) -> WorkflowRunCreationResult: - """Reuse the config while the caller holds its row lock.""" + """Retries preserve the original configuration and inputs.""" + config = self.config + if not trigger_key or len(trigger_key) > 256: + raise ValueError("Trigger keys must contain between 1 and 256 characters") + existing = self.get_runs().filter(trigger_key=trigger_key).first() if existing is not None: return WorkflowRunCreationResult( @@ -234,7 +254,7 @@ def _get_or_create_run_with_locked_config( config_version=config.version, strategy=SeerWorkflowStrategy(config.strategy), filters=config.filters, - project_ids=tuple(self._get_project_ids(config)), + project_ids=tuple(_get_project_ids(config)), output_policy=config.output_policy, inputs=json.loads(inputs.json()), ) @@ -251,41 +271,59 @@ def _get_or_create_run_with_locked_config( created=True, ) - @contextmanager - def _lock_config(self) -> Iterator[SeerWorkflowConfig]: - with transaction.atomic(router.db_for_write(SeerWorkflowConfig)): - yield self._config_queryset().select_for_update().get() - - def _config_queryset(self) -> QuerySet[SeerWorkflowConfig]: - return SeerWorkflowConfig.objects.using(router.db_for_write(SeerWorkflowConfig)).filter( - id=self.config_id, organization_id=self.organization_id - ) - - def _matches_scope( - self, config: SeerWorkflowConfig, project_id: int, environment: str | None - ) -> bool: - filters = SeerWorkflowFilters.parse_obj(config.filters) - return ( - config.enabled - and (not filters.environments or environment in filters.environments) - and self._get_projects(config).filter(id=project_id).exists() - ) + def get_runs(self) -> QuerySet[SeerWorkflowRun]: + config = self.config + return _run_queryset(config.id, config.organization_id) - def _get_project_ids(self, config: SeerWorkflowConfig) -> list[int]: - return list(self._get_projects(config).order_by("id").values_list("id", flat=True)) + def matches_scope(self, project_id: int, environment: str | None) -> bool: + return _matches_scope(self.config, project_id, environment) - def _get_projects(self, config: SeerWorkflowConfig) -> QuerySet[Project]: - projects = Project.objects.filter( - organization_id=config.organization_id, status=ObjectStatus.ACTIVE - ) - if not SeerWorkflowFilters.parse_obj(config.filters).all_projects: - projects = projects.filter( - id__in=SeerWorkflowConfigProject.objects.filter(config=config).values("project_id") - ) - return projects + @property + def config(self) -> SeerWorkflowConfig: + if self._config is None: + raise RuntimeError("Workflow transaction is not active") + return self._config @dataclass(frozen=True) class WorkflowRunCreationResult: lifecycle: SeerWorkflowRunLifecycle created: bool + + +def _config_queryset(config_id: int, organization_id: int) -> QuerySet[SeerWorkflowConfig]: + return SeerWorkflowConfig.objects.using(router.db_for_write(SeerWorkflowConfig)).filter( + id=config_id, organization_id=organization_id + ) + + +def _run_queryset(config_id: int, organization_id: int) -> QuerySet[SeerWorkflowRun]: + return ( + SeerWorkflowRun.objects.using(router.db_for_write(SeerWorkflowRun)) + .filter(organization_id=organization_id, workflow_config_id=config_id) + .order_by("-date_added", "-id") + ) + + +def _matches_scope(config: SeerWorkflowConfig, project_id: int, environment: str | None) -> bool: + filters = SeerWorkflowFilters.parse_obj(config.filters) + return ( + config.enabled + and (not filters.environments or environment in filters.environments) + and _get_projects(config).filter(id=project_id).exists() + ) + + +def _get_project_ids(config: SeerWorkflowConfig) -> list[int]: + return list(_get_projects(config).order_by("id").values_list("id", flat=True)) + + +def _get_projects(config: SeerWorkflowConfig) -> QuerySet[Project]: + projects = Project.objects.filter( + organization_id=config.organization_id, status=ObjectStatus.ACTIVE + ) + if not SeerWorkflowFilters.parse_obj(config.filters).all_projects: + projects = projects.filter( + id__in=SeerWorkflowConfigProject.objects.filter(config=config).values("project_id") + ) + return projects diff --git a/tests/sentry/seer/workflows/test_runs.py b/tests/sentry/seer/workflows/test_runs.py index 51f19b15d157..aa56d14c518c 100644 --- a/tests/sentry/seer/workflows/test_runs.py +++ b/tests/sentry/seer/workflows/test_runs.py @@ -28,7 +28,11 @@ get_workflow_run_status, ) from sentry.seer.workflows.schemas import WorkflowResult -from sentry.seer.workflows.workflow import SeerWorkflow, SelectedProjects +from sentry.seer.workflows.workflow import ( + SeerWorkflow, + SeerWorkflowTransaction, + SelectedProjects, +) from sentry.testutils.cases import TestCase from sentry.testutils.factories import Factories @@ -274,6 +278,47 @@ def test_request_identity_and_history_are_scoped_to_config_and_organization(self assert first.get_run().status == SeerWorkflowRunStatus.QUEUED assert first.get_run().date_completed is None + def test_workflow_transaction_requires_an_active_context(self) -> None: + transaction = SeerWorkflowTransaction() + + with pytest.raises(RuntimeError, match="Workflow transaction is not active"): + transaction.get_or_create_run( + trigger_key="manual:first", inputs=RunInputs(issue_limit=10) + ) + + assert not self.workflow.get_runs().exists() + + def test_workflow_transaction_cannot_be_used_after_exit(self) -> None: + with self.workflow.transaction() as transaction: + result = transaction.get_or_create_run( + trigger_key="manual:first", inputs=RunInputs(issue_limit=10) + ) + + with pytest.raises(RuntimeError, match="Workflow transaction is not active"): + transaction.get_or_create_run( + trigger_key="manual:later", inputs=RunInputs(issue_limit=10) + ) + with pytest.raises(RuntimeError, match="Workflow transaction is not active"): + transaction.get_runs() + with pytest.raises(RuntimeError, match="Workflow transaction is not active"): + transaction.matches_scope(self.project.id, None) + + assert list(self.workflow.get_runs()) == [result.lifecycle.get_run()] + + def test_failed_workflow_transaction_rolls_back_and_cannot_be_reused(self) -> None: + with pytest.raises(ValueError, match="failed"): + with self.workflow.transaction() as transaction: + transaction.get_or_create_run( + trigger_key="manual:first", inputs=RunInputs(issue_limit=10) + ) + raise ValueError("failed") + + assert not self.workflow.get_runs().exists() + with pytest.raises(RuntimeError, match="Workflow transaction is not active"): + transaction.get_or_create_run( + trigger_key="manual:retry", inputs=RunInputs(issue_limit=10) + ) + def test_lifecycle_reads_updates_from_another_instance(self) -> None: run = self.create_run() other = SeerWorkflowRunLifecycle(organization_id=self.organization.id, run_id=run.run_id) From cb9f9a79f0a42199d51df277cd1f4f793fa29f48 Mon Sep 17 00:00:00 2001 From: Trevor Elkins Date: Wed, 28 Oct 2026 18:45:49 -0400 Subject: [PATCH 5/5] ref(seer): Keep workflow locking local to each operation --- src/sentry/seer/workflows/breached_metrics.py | 106 ++++++------ src/sentry/seer/workflows/workflow.py | 155 +++++++----------- .../seer/workflows/test_breached_metrics.py | 6 +- tests/sentry/seer/workflows/test_runs.py | 47 +----- 4 files changed, 118 insertions(+), 196 deletions(-) diff --git a/src/sentry/seer/workflows/breached_metrics.py b/src/sentry/seer/workflows/breached_metrics.py index 9a89ac70e3b8..5ae6bdc69a15 100644 --- a/src/sentry/seer/workflows/breached_metrics.py +++ b/src/sentry/seer/workflows/breached_metrics.py @@ -5,6 +5,7 @@ from datetime import datetime from typing import Annotated, Any, Literal +from django.db import router, transaction from django.utils import timezone from pydantic import BaseModel, Field, PositiveInt, ValidationError @@ -22,7 +23,7 @@ WorkflowRunError, ) from sentry.seer.workflows.runs import SeerWorkflowRunLifecycle -from sentry.seer.workflows.workflow import SeerWorkflow, SeerWorkflowTransaction +from sentry.seer.workflows.workflow import SeerWorkflow, matches_workflow_scope from sentry.services.eventstore.models import GroupEvent @@ -160,56 +161,67 @@ def accept( except ValidationError: return IgnoredMetricBreach("invalid_source") - try: - with self.transaction() as transaction: - config = transaction.config - if ( - config.strategy != SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION - or not config.enabled - ): - return IgnoredMetricBreach("configuration_unavailable") - if not transaction.matches_scope(group.project_id, trigger.environment): - return IgnoredMetricBreach("out_of_scope") - return self._record_run(transaction, group, trigger) - except SeerWorkflowConfig.DoesNotExist: - return IgnoredMetricBreach("configuration_unavailable") + return self._record_run(group, trigger) def _record_run( self, - transaction: SeerWorkflowTransaction, group: Group, trigger: MetricBreachTrigger, - ) -> AcceptedMetricBreach | RejectedMetricBreach: - trigger_key = f"issue_created:{group.id}" - existing = transaction.get_runs().filter(trigger_key=trigger_key).first() - if existing is not None: - lifecycle = SeerWorkflowRunLifecycle( - run_id=existing.id, organization_id=existing.organization_id + ) -> AcceptedMetricBreach | RejectedMetricBreach | IgnoredMetricBreach: + database = router.db_for_write(SeerWorkflowConfig) + with transaction.atomic(database): + config = ( + SeerWorkflowConfig.objects.using(database) + .select_for_update() + .filter( + id=self.config_id, + organization_id=self.organization_id, + strategy=SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION, + enabled=True, + ) + .first() ) - admission = lifecycle.parse_extras(MetricBreachRunExtras).admission - if admission.status == "rejected": - return RejectedMetricBreach(run=existing, reason=admission.reason, created=False) - return AcceptedMetricBreach(run=existing, created=False) - - decided_at = timezone.now() - admission = ( - AcceptedMetricBreachAdmission(decided_at=decided_at) - if self.has_quota_available() - else RejectedMetricBreachAdmission(reason="quota_exceeded", decided_at=decided_at) - ) - result = transaction.get_or_create_run( - trigger_key=trigger_key, inputs=trigger, extras={"admission": admission.dict()} - ) - workflow_run = result.lifecycle - if admission.status == "rejected": - run = workflow_run.reject( - error=WorkflowRunError(code=admission.reason, message="Automation quota exceeded.") + if config is None: + return IgnoredMetricBreach("configuration_unavailable") + if not matches_workflow_scope(config, group.project_id, trigger.environment): + return IgnoredMetricBreach("out_of_scope") + + trigger_key = f"issue_created:{group.id}" + existing = self.get_runs().filter(trigger_key=trigger_key).first() + if existing is not None: + lifecycle = SeerWorkflowRunLifecycle( + run_id=existing.id, organization_id=existing.organization_id + ) + admission = lifecycle.parse_extras(MetricBreachRunExtras).admission + if admission.status == "rejected": + return RejectedMetricBreach( + run=existing, reason=admission.reason, created=False + ) + return AcceptedMetricBreach(run=existing, created=False) + + decided_at = timezone.now() + admission = ( + AcceptedMetricBreachAdmission(decided_at=decided_at) + if self.has_quota_available() + else RejectedMetricBreachAdmission(reason="quota_exceeded", decided_at=decided_at) ) - return RejectedMetricBreach(run=run, reason=admission.reason, created=result.created) - - workflow_run.start() - # TODO(telkins): Replace this no-op execution with investigation dispatch. - (execution,) = workflow_run.create_executions([MetricBreachExecutionExtras()]) - workflow_run.start_execution(execution.id) - workflow_run.complete_execution(execution.id) - return AcceptedMetricBreach(run=workflow_run.get_run(), created=result.created) + result = self.get_or_create_run( + trigger_key=trigger_key, inputs=trigger, extras={"admission": admission.dict()} + ) + workflow_run = result.lifecycle + if admission.status == "rejected": + run = workflow_run.reject( + error=WorkflowRunError( + code=admission.reason, message="Automation quota exceeded." + ) + ) + return RejectedMetricBreach( + run=run, reason=admission.reason, created=result.created + ) + + workflow_run.start() + # TODO(telkins): Replace this no-op execution with investigation dispatch. + (execution,) = workflow_run.create_executions([MetricBreachExecutionExtras()]) + workflow_run.start_execution(execution.id) + workflow_run.complete_execution(execution.id) + return AcceptedMetricBreach(run=workflow_run.get_run(), created=result.created) diff --git a/src/sentry/seer/workflows/workflow.py b/src/sentry/seer/workflows/workflow.py index 86a6fa1a3a51..f51d0540fb94 100644 --- a/src/sentry/seer/workflows/workflow.py +++ b/src/sentry/seer/workflows/workflow.py @@ -1,8 +1,8 @@ from __future__ import annotations from collections.abc import Iterator, Mapping -from contextlib import AbstractContextManager, contextmanager -from dataclasses import dataclass, field +from contextlib import contextmanager +from dataclasses import dataclass from typing import Any, assert_never from django.db import router, transaction @@ -89,7 +89,7 @@ def ensure_defaults(cls, organization_id: int) -> None: def get_config(self) -> SeerWorkflowConfig: """Fetch the config without acquiring a lock.""" - return _config_queryset(self.config_id, self.organization_id).get() + return self._config_queryset().get() @property def extras(self) -> SeerWorkflowConfigExtras: @@ -107,15 +107,52 @@ def get_or_create_run( extras: Mapping[str, Any] | None = None, ) -> WorkflowRunCreationResult: """Retries preserve the original configuration and inputs.""" - with self.transaction() as locked: - return locked.get_or_create_run(trigger_key=trigger_key, inputs=inputs, extras=extras) + if not trigger_key or len(trigger_key) > 256: + raise ValueError("Trigger keys must contain between 1 and 256 characters") + + with self._lock_config() as config: + existing = self.get_runs().filter(trigger_key=trigger_key).first() + if existing is not None: + return WorkflowRunCreationResult( + lifecycle=SeerWorkflowRunLifecycle( + run_id=existing.id, organization_id=existing.organization_id + ), + created=False, + ) + + snapshot = WorkflowRunSnapshot( + config_version=config.version, + strategy=SeerWorkflowStrategy(config.strategy), + filters=config.filters, + project_ids=tuple(_get_project_ids(config)), + output_policy=config.output_policy, + inputs=json.loads(inputs.json()), + ) + run_extras = SeerWorkflowRunExtras.parse_obj({**(extras or {}), "snapshot": snapshot}) + run = SeerWorkflowRun.objects.create( + organization_id=config.organization_id, + workflow_config=config, + trigger_key=trigger_key, + status=SeerWorkflowRunStatus.QUEUED, + extras=json.loads(run_extras.json()), + ) + return WorkflowRunCreationResult( + lifecycle=SeerWorkflowRunLifecycle( + run_id=run.id, organization_id=run.organization_id + ), + created=True, + ) def get_runs(self) -> QuerySet[SeerWorkflowRun]: - return _run_queryset(self.config_id, self.organization_id) + return ( + SeerWorkflowRun.objects.using(router.db_for_write(SeerWorkflowRun)) + .filter(organization_id=self.organization_id, workflow_config_id=self.config_id) + .order_by("-date_added", "-id") + ) def matches_scope(self, project_id: int, environment: str | None) -> bool: """Check the automation's scope before recording an admission decision.""" - return _matches_scope(self.get_config(), project_id, environment) + return matches_workflow_scope(self.get_config(), project_id, environment) def has_quota_available(self) -> bool: """Allow runs until automation quota enforcement is implemented.""" @@ -128,8 +165,7 @@ def get_project_ids(self) -> list[int]: def set_service_account(self, service_account_id: int | None) -> None: """Store the control account ID after validating its current cell replica.""" - with self.transaction() as locked: - config = locked.config + with self._lock_config() as config: if service_account_id is not None: try: replica = ServiceAccountReplica.objects.get( @@ -177,8 +213,7 @@ def set_project_selection(self, selection: ProjectSelection) -> None: case _: assert_never(selection) - with self.transaction() as locked: - config = locked.config + with self._lock_config() as config: if ( config.strategy == SeerWorkflowStrategy.BREACHED_METRIC_INVESTIGATION and not all_projects @@ -205,84 +240,15 @@ def set_project_selection(self, selection: ProjectSelection) -> None: config.version += 1 config.save(update_fields=["filters", "version", "date_updated"]) - def transaction(self) -> AbstractContextManager[SeerWorkflowTransaction]: - """Share one locked config across an atomic workflow operation.""" - return SeerWorkflowTransaction.open( - config_id=self.config_id, organization_id=self.organization_id - ) - - -@dataclass -class SeerWorkflowTransaction: - """Workflow operations available only while the config's row lock is held.""" - - _config: SeerWorkflowConfig | None = field(default=None, init=False, repr=False) - - @classmethod @contextmanager - def open(cls, *, config_id: int, organization_id: int) -> Iterator[SeerWorkflowTransaction]: + def _lock_config(self) -> Iterator[SeerWorkflowConfig]: with transaction.atomic(router.db_for_write(SeerWorkflowConfig)): - locked = cls() - locked._config = _config_queryset(config_id, organization_id).select_for_update().get() - try: - yield locked - finally: - locked._config = None - - def get_or_create_run( - self, - *, - trigger_key: str, - inputs: BaseModel, - extras: Mapping[str, Any] | None = None, - ) -> WorkflowRunCreationResult: - """Retries preserve the original configuration and inputs.""" - config = self.config - if not trigger_key or len(trigger_key) > 256: - raise ValueError("Trigger keys must contain between 1 and 256 characters") - - existing = self.get_runs().filter(trigger_key=trigger_key).first() - if existing is not None: - return WorkflowRunCreationResult( - lifecycle=SeerWorkflowRunLifecycle( - run_id=existing.id, organization_id=existing.organization_id - ), - created=False, - ) + yield self._config_queryset().select_for_update().get() - snapshot = WorkflowRunSnapshot( - config_version=config.version, - strategy=SeerWorkflowStrategy(config.strategy), - filters=config.filters, - project_ids=tuple(_get_project_ids(config)), - output_policy=config.output_policy, - inputs=json.loads(inputs.json()), + def _config_queryset(self) -> QuerySet[SeerWorkflowConfig]: + return SeerWorkflowConfig.objects.using(router.db_for_write(SeerWorkflowConfig)).filter( + id=self.config_id, organization_id=self.organization_id ) - run_extras = SeerWorkflowRunExtras.parse_obj({**(extras or {}), "snapshot": snapshot}) - run = SeerWorkflowRun.objects.create( - organization_id=config.organization_id, - workflow_config=config, - trigger_key=trigger_key, - status=SeerWorkflowRunStatus.QUEUED, - extras=json.loads(run_extras.json()), - ) - return WorkflowRunCreationResult( - lifecycle=SeerWorkflowRunLifecycle(run_id=run.id, organization_id=run.organization_id), - created=True, - ) - - def get_runs(self) -> QuerySet[SeerWorkflowRun]: - config = self.config - return _run_queryset(config.id, config.organization_id) - - def matches_scope(self, project_id: int, environment: str | None) -> bool: - return _matches_scope(self.config, project_id, environment) - - @property - def config(self) -> SeerWorkflowConfig: - if self._config is None: - raise RuntimeError("Workflow transaction is not active") - return self._config @dataclass(frozen=True) @@ -291,21 +257,10 @@ class WorkflowRunCreationResult: created: bool -def _config_queryset(config_id: int, organization_id: int) -> QuerySet[SeerWorkflowConfig]: - return SeerWorkflowConfig.objects.using(router.db_for_write(SeerWorkflowConfig)).filter( - id=config_id, organization_id=organization_id - ) - - -def _run_queryset(config_id: int, organization_id: int) -> QuerySet[SeerWorkflowRun]: - return ( - SeerWorkflowRun.objects.using(router.db_for_write(SeerWorkflowRun)) - .filter(organization_id=organization_id, workflow_config_id=config_id) - .order_by("-date_added", "-id") - ) - - -def _matches_scope(config: SeerWorkflowConfig, project_id: int, environment: str | None) -> bool: +def matches_workflow_scope( + config: SeerWorkflowConfig, project_id: int, environment: str | None +) -> bool: + """Check scope using a config snapshot without modifying it or acquiring a lock.""" filters = SeerWorkflowFilters.parse_obj(config.filters) return ( config.enabled diff --git a/tests/sentry/seer/workflows/test_breached_metrics.py b/tests/sentry/seer/workflows/test_breached_metrics.py index 22d0f77f369c..8ab6b90a5dff 100644 --- a/tests/sentry/seer/workflows/test_breached_metrics.py +++ b/tests/sentry/seer/workflows/test_breached_metrics.py @@ -179,7 +179,7 @@ def test_accept_uses_latest_scope_from_the_locked_config(self) -> None: assert result == IgnoredMetricBreach("out_of_scope") assert not SeerWorkflowRun.objects.exists() - def test_accept_reuses_locked_config_for_run_snapshot(self) -> None: + def test_admission_and_run_creation_lock_current_config(self) -> None: workflow = BreachedMetricWorkflow( config_id=self.config.id, organization_id=self.organization.id ) @@ -203,8 +203,8 @@ def test_accept_reuses_locked_config_for_run_snapshot(self) -> None: config_reads = [ query["sql"] for query in queries if 'FROM "seer_workflowconfig"' in query["sql"] ] - assert len(config_reads) == 1 - assert "FOR UPDATE" in config_reads[0] + assert len(config_reads) == 2 + assert all("FOR UPDATE" in query for query in config_reads) def test_ignored_configuration_reports_reason_without_starting_work(self) -> None: self.config.update(enabled=False) diff --git a/tests/sentry/seer/workflows/test_runs.py b/tests/sentry/seer/workflows/test_runs.py index aa56d14c518c..51f19b15d157 100644 --- a/tests/sentry/seer/workflows/test_runs.py +++ b/tests/sentry/seer/workflows/test_runs.py @@ -28,11 +28,7 @@ get_workflow_run_status, ) from sentry.seer.workflows.schemas import WorkflowResult -from sentry.seer.workflows.workflow import ( - SeerWorkflow, - SeerWorkflowTransaction, - SelectedProjects, -) +from sentry.seer.workflows.workflow import SeerWorkflow, SelectedProjects from sentry.testutils.cases import TestCase from sentry.testutils.factories import Factories @@ -278,47 +274,6 @@ def test_request_identity_and_history_are_scoped_to_config_and_organization(self assert first.get_run().status == SeerWorkflowRunStatus.QUEUED assert first.get_run().date_completed is None - def test_workflow_transaction_requires_an_active_context(self) -> None: - transaction = SeerWorkflowTransaction() - - with pytest.raises(RuntimeError, match="Workflow transaction is not active"): - transaction.get_or_create_run( - trigger_key="manual:first", inputs=RunInputs(issue_limit=10) - ) - - assert not self.workflow.get_runs().exists() - - def test_workflow_transaction_cannot_be_used_after_exit(self) -> None: - with self.workflow.transaction() as transaction: - result = transaction.get_or_create_run( - trigger_key="manual:first", inputs=RunInputs(issue_limit=10) - ) - - with pytest.raises(RuntimeError, match="Workflow transaction is not active"): - transaction.get_or_create_run( - trigger_key="manual:later", inputs=RunInputs(issue_limit=10) - ) - with pytest.raises(RuntimeError, match="Workflow transaction is not active"): - transaction.get_runs() - with pytest.raises(RuntimeError, match="Workflow transaction is not active"): - transaction.matches_scope(self.project.id, None) - - assert list(self.workflow.get_runs()) == [result.lifecycle.get_run()] - - def test_failed_workflow_transaction_rolls_back_and_cannot_be_reused(self) -> None: - with pytest.raises(ValueError, match="failed"): - with self.workflow.transaction() as transaction: - transaction.get_or_create_run( - trigger_key="manual:first", inputs=RunInputs(issue_limit=10) - ) - raise ValueError("failed") - - assert not self.workflow.get_runs().exists() - with pytest.raises(RuntimeError, match="Workflow transaction is not active"): - transaction.get_or_create_run( - trigger_key="manual:retry", inputs=RunInputs(issue_limit=10) - ) - def test_lifecycle_reads_updates_from_another_instance(self) -> None: run = self.create_run() other = SeerWorkflowRunLifecycle(organization_id=self.organization.id, run_id=run.run_id)