fix(workflows): Rule deletion shouldn't automatically result in Workflow deletion

src/sentry/deletions/defaults/rule.py

@@ -1,12 +1,8 @@
-import logging
 from collections.abc import Sequence
 
 from sentry.deletions.base import BaseRelation, ModelDeletionTask, ModelRelation
 from sentry.deletions.defaults.alertrule import AlertRuleDetectorDeletionTask
 from sentry.models.rule import Rule
-from sentry.workflow_engine.models import Workflow
-
-logger = logging.getLogger(__name__)
 
 
 class RuleDeletionTask(ModelDeletionTask[Rule]):
@@ -15,42 +11,21 @@ def get_child_relations(self, instance: Rule) -> list[BaseRelation]:
         from sentry.models.rule import RuleActivity
         from sentry.workflow_engine.models import AlertRuleDetector, AlertRuleWorkflow
 
-        model_relations: list[BaseRelation] = [
+        # Workflows are org-scoped and must not be deleted when a project-scoped
+        # Rule is deleted. Workflow cleanup happens via the API (which schedules
+        # Workflow deletion explicitly) or OrganizationDeletionTask. We only
+        # clean up the link rows (AlertRuleWorkflow, AlertRuleDetector) here.
+        return [
             ModelRelation(GroupRuleStatus, {"rule_id": instance.id}),
             ModelRelation(RuleActivity, {"rule_id": instance.id}),
             ModelRelation(
                 AlertRuleDetector,
                 {"rule_id": instance.id},
                 task=AlertRuleDetectorDeletionTask,
             ),
+            ModelRelation(AlertRuleWorkflow, {"rule_id": instance.id}),
         ]
 
-        # AlertRuleWorkflow must be deleted before Workflow so the link rows
-        # are gone by the time WorkflowDeletionTask runs — otherwise it would
-        # cascade back to this Rule and loop infinitely.
-        workflow_ids = list(
-            AlertRuleWorkflow.objects.filter(rule_id=instance.id).values_list(
-                "workflow_id", flat=True
-            )
-        )
-        if workflow_ids:
-            model_relations.append(ModelRelation(AlertRuleWorkflow, {"rule_id": instance.id}))
-            model_relations.append(
-                ModelRelation(
-                    Workflow,
-                    {
-                        "id__in": workflow_ids,
-                        "organization_id": instance.project.organization_id,
-                    },
-                )
-            )
-        else:
-            logger.info(
-                "No AlertRuleWorkflow found for rule, skipping", extra={"rule_id": instance.id}
-            )
-
-        return model_relations
-
     def mark_deletion_in_progress(self, instance_list: Sequence[Rule]) -> None:
         from sentry.constants import ObjectStatus
 

tests/sentry/deletions/test_rule.py

@@ -64,5 +64,8 @@ def test_simple(self) -> None:
         ).exists()
         assert not GroupRuleStatus.objects.filter(id=group_rule_status.id).exists()
         assert not RuleActivity.objects.filter(id=rule_activity.id).exists()
+        # Link rows are cleaned up
         assert not AlertRuleDetector.objects.filter(rule_id=rule.id).exists()
         assert not AlertRuleWorkflow.objects.filter(rule_id=rule.id).exists()
+        # Org-scoped Workflow survives Rule deletion
+        assert Workflow.objects.filter(id=workflow.id).exists()

tests/sentry/workflow_engine/migration_helpers/test_issue_alert_dual_write.py

@@ -4,6 +4,7 @@
 from sentry.constants import ObjectStatus
 from sentry.deletions.models.scheduleddeletion import CellScheduledDeletion
 from sentry.deletions.tasks.scheduled import run_scheduled_deletions
+from sentry.models.rule import Rule
 from sentry.models.rulesnooze import RuleSnooze
 from sentry.rules.age import AgeComparisonType
 from sentry.rules.conditions.event_frequency import (
@@ -354,7 +355,13 @@ def setUp(self) -> None:
         self.when_dcg: DataConditionGroup = when_dcg
         self.if_dcg: DataConditionGroup = if_dcg
 
-    def assert_issue_alert_deleted(
+    def assert_rule_deleted_workflow_survives(self, workflow: Workflow) -> None:
+        """Rule and link rows are deleted, but org-scoped Workflow survives."""
+        assert not Rule.objects.filter(id=self.issue_alert.id).exists()
+        assert not AlertRuleWorkflow.objects.filter(rule_id=self.issue_alert.id).exists()
+        assert Workflow.objects.filter(id=workflow.id).exists()
+
+    def assert_everything_deleted(
         self, workflow: Workflow, when_dcg: DataConditionGroup, if_dcg: DataConditionGroup
     ) -> None:
         assert not AlertRuleWorkflow.objects.filter(rule_id=self.issue_alert.id).exists()
@@ -373,7 +380,7 @@ def test_delete_issue_alert__rule_deletion_task(self) -> None:
         with self.tasks():
             run_scheduled_deletions()
 
-        self.assert_issue_alert_deleted(self.workflow, self.when_dcg, self.if_dcg)
+        self.assert_rule_deleted_workflow_survives(self.workflow)
 
     def test_delete_issue_alert__project_deletion_task(self) -> None:
         self.project.update(status=ObjectStatus.PENDING_DELETION)
@@ -382,7 +389,9 @@ def test_delete_issue_alert__project_deletion_task(self) -> None:
         with self.tasks():
             run_scheduled_deletions()
 
-        self.assert_issue_alert_deleted(self.workflow, self.when_dcg, self.if_dcg)
+        # Workflows are org-scoped, not project-scoped, so they survive
+        # project deletion. Only OrganizationDeletionTask cleans them up.
+        self.assert_rule_deleted_workflow_survives(self.workflow)
 
     def test_delete_issue_alert__org_deletion_task(self) -> None:
         self.organization.update(status=ObjectStatus.PENDING_DELETION)
@@ -391,4 +400,4 @@ def test_delete_issue_alert__org_deletion_task(self) -> None:
         with self.tasks():
             run_scheduled_deletions()
 
-        self.assert_issue_alert_deleted(self.workflow, self.when_dcg, self.if_dcg)
+        self.assert_everything_deleted(self.workflow, self.when_dcg, self.if_dcg)