🔧 fix(aci): deduplicate actions before triggering (#97444)

Author: iamrajjoshi

pyproject.toml

@@ -642,7 +642,6 @@ module = [
     "sentry.web.frontend.cli",
     "sentry.web.frontend.csv",
     "sentry.web.frontend.mixins.*",
-    "sentry.workflow_engine.handlers.action.*",
     "sentry.workflow_engine.handlers.condition.*",
     "sentry.workflow_engine.migrations.*",
     "sentry.workflow_engine.processors.*",

src/sentry/workflow_engine/handlers/action/__init__.py

@@ -110,6 +110,26 @@ def trigger(self, event_data: WorkflowEventData, detector: Detector) -> None:
             },
         )
 
+    def get_dedup_key(self) -> str:
+        key_parts = [self.type]
+
+        if self.integration_id:
+            key_parts.append(str(self.integration_id))
+
+        if self.config:
+            config = self.config.copy()
+            config.pop("target_display", None)
+            key_parts.append(str(config))
+
+        if self.data:
+            data = self.data.copy()
+            if "dynamic_form_fields" in data:
+                data = data["dynamic_form_fields"]
+
+            key_parts.append(str(data))
+
+        return ":".join(key_parts)
+
 
 @receiver(pre_save, sender=Action)
 def enforce_config_schema(sender, instance: Action, **kwargs):

src/sentry/workflow_engine/handlers/action/notification/plugin_handler.py

@@ -121,6 +121,22 @@ def update_workflow_action_group_statuses(
     )
 
 
+def deduplicate_actions(
+    actions_queryset: BaseQuerySet[Action],  # decorated with the workflow_ids
+) -> BaseQuerySet[Action]:
+    """
+    Deduplicates actions based on their handler's dedup_key method.
+    Returns a de-duplicated queryset of actions.
+    """
+    dedup_key_to_action_id: dict[str, int] = {}
+
+    for action in actions_queryset:
+        dedup_key = action.get_dedup_key()
+        dedup_key_to_action_id[dedup_key] = action.id
+
+    return actions_queryset.filter(id__in=dedup_key_to_action_id.values())
+
+
 def filter_recently_fired_workflow_actions(
     filtered_action_groups: set[DataConditionGroup], event_data: WorkflowEventData
 ) -> BaseQuerySet[Action]:
@@ -158,16 +174,20 @@ def filter_recently_fired_workflow_actions(
     )
     update_workflow_action_group_statuses(now, statuses_to_update, missing_statuses)
 
+    actions_queryset = Action.objects.filter(id__in=list(action_to_workflow_ids.keys()))
+
     # annotate actions with workflow_id they are firing for (deduped)
     workflow_id_cases = [
         When(id=action_id, then=Value(workflow_id))
         for action_id, workflow_id in action_to_workflow_ids.items()
     ]
 
-    return Action.objects.filter(id__in=list(action_to_workflow_ids.keys())).annotate(
+    decorated_actions = actions_queryset.annotate(
         workflow_id=Case(*workflow_id_cases, output_field=models.IntegerField()),
     )
 
+    return deduplicate_actions(decorated_actions)
+
 
 def get_available_action_integrations_for_org(organization: Organization) -> list[RpcIntegration]:
     providers = [

src/sentry/workflow_engine/handlers/action/notification/webhook_handler.py

@@ -8,6 +8,7 @@
 from sentry.issues.status_change_consumer import update_status
 from sentry.issues.status_change_message import StatusChangeMessage
 from sentry.models.group import Group
+from sentry.notifications.models.notificationaction import ActionTarget
 from sentry.tasks.post_process import post_process_group
 from sentry.testutils.helpers import with_feature
 from sentry.testutils.helpers.features import Feature
@@ -57,7 +58,16 @@ def create_metric_issue_workflow(self, detector: Detector):
             type=Condition.ISSUE_PRIORITY_DEESCALATING,
             condition_group=critical_dcg,
         )
-        critical_action = self.create_action()
+
+        critical_action = self.create_action(
+            integration_id=self.integration.id,
+            config={
+                "target_type": ActionTarget.SPECIFIC,
+                "target_identifier": "channel-123",
+                "target_display": "Test Channel",
+            },
+        )
+
         self.create_data_condition_group_action(critical_action, critical_dcg)
 
         warning_dcg = self.create_data_condition_group(organization=self.organization)
@@ -75,8 +85,15 @@ def create_metric_issue_workflow(self, detector: Detector):
             type=Condition.ISSUE_PRIORITY_DEESCALATING,
             condition_group=warning_dcg,
         )
+        warning_action = self.create_action(
+            integration_id=self.integration.id,
+            config={
+                "target_type": ActionTarget.SPECIFIC,
+                "target_identifier": "channel-456",
+                "target_display": "Test Channel",
+            },
+        )
 
-        warning_action = self.create_action()
         self.create_data_condition_group_action(warning_action, warning_dcg)
 
         return (
@@ -141,6 +158,30 @@ def test_escalation(self, mock_trigger: MagicMock) -> None:
         self.call_post_process_group(occurrence)
         assert mock_trigger.call_count == 2  # warning + critical actions
 
+    def test_escalation_with_deduped_actions(self, mock_trigger: MagicMock) -> None:
+
+        # make the warning action same as the critical action
+        self.warning_action.config = self.critical_action.config
+        self.warning_action.save()
+
+        value = self.warning_detector_trigger.comparison + 1
+        data_packet = self.create_subscription_packet(value)
+        occurrence = self.process_packet_and_return_result(data_packet)
+        assert isinstance(occurrence, IssueOccurrence)
+        occurrence.save()
+        self.call_post_process_group(occurrence)
+        assert mock_trigger.call_count == 1  # just warning action
+
+        mock_trigger.reset_mock()
+
+        value = self.critical_detector_trigger.comparison + 1
+        data_packet = self.create_subscription_packet(value, 1000)
+        occurrence = self.process_packet_and_return_result(data_packet)
+        assert isinstance(occurrence, IssueOccurrence)
+        occurrence.save()
+        self.call_post_process_group(occurrence)
+        assert mock_trigger.call_count == 1  # just warning action (because we deduped the actions)
+
     def test_deescalation(self, mock_trigger: MagicMock) -> None:
         value = self.critical_detector_trigger.comparison + 1
         data_packet = self.create_subscription_packet(value)

src/sentry/workflow_engine/models/action.py

@@ -75,7 +75,9 @@ def test_multiple_workflows(self) -> None:
             workflow=workflow, action=action_2, group=self.group
         )
 
-        _, action_3 = self.create_workflow_action(workflow=workflow)
+        action_3 = self.create_action(type=Action.Type.PLUGIN)
+        self.create_workflow_action(workflow=workflow, action=action_3)
+
         status_3 = WorkflowActionGroupStatus.objects.create(
             workflow=workflow, action=action_3, group=self.group
         )