fix(aci): record WorkflowFireHistory before deduping actions (#97746)

Author: cathteng

src/sentry/workflow_engine/processors/action.py

@@ -28,7 +28,9 @@
     Workflow,
     WorkflowActionGroupStatus,
 )
+from sentry.workflow_engine.models.detector import Detector
 from sentry.workflow_engine.registry import action_handler_registry
+from sentry.workflow_engine.tasks.actions import build_trigger_action_task_params, trigger_action
 from sentry.workflow_engine.types import WorkflowEventData
 
 logger = logging.getLogger(__name__)
@@ -137,6 +139,15 @@ def deduplicate_actions(
     return actions_queryset.filter(id__in=dedup_key_to_action_id.values())
 
 
+def fire_actions(
+    actions: BaseQuerySet[Action], detector: Detector, event_data: WorkflowEventData
+) -> None:
+    deduped_actions = deduplicate_actions(actions)
+    for action in deduped_actions:
+        task_params = build_trigger_action_task_params(action, detector, event_data)
+        trigger_action.apply_async(kwargs=task_params, headers={"sentry-propagate-traces": False})
+
+
 def filter_recently_fired_workflow_actions(
     filtered_action_groups: set[DataConditionGroup], event_data: WorkflowEventData
 ) -> BaseQuerySet[Action]:
@@ -182,12 +193,10 @@ def filter_recently_fired_workflow_actions(
         for action_id, workflow_id in action_to_workflow_ids.items()
     ]
 
-    decorated_actions = actions_queryset.annotate(
+    return 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/processors/delayed_workflow.py

@@ -44,15 +44,17 @@
     SLOW_CONDITIONS,
     Condition,
 )
-from sentry.workflow_engine.processors.action import filter_recently_fired_workflow_actions
+from sentry.workflow_engine.processors.action import (
+    filter_recently_fired_workflow_actions,
+    fire_actions,
+)
 from sentry.workflow_engine.processors.data_condition_group import (
     evaluate_data_conditions,
     get_slow_conditions_for_groups,
 )
 from sentry.workflow_engine.processors.detector import get_detectors_by_groupevents_bulk
 from sentry.workflow_engine.processors.log_util import track_batch_performance
 from sentry.workflow_engine.processors.workflow_fire_history import create_workflow_fire_histories
-from sentry.workflow_engine.tasks.actions import build_trigger_action_task_params, trigger_action
 from sentry.workflow_engine.types import WorkflowEventData
 from sentry.workflow_engine.utils import log_context
 
@@ -747,14 +749,7 @@ def fire_actions_for_groups(
                 total_actions += len(filtered_actions)
 
                 if should_trigger_actions(group_event.group.type):
-                    for action in filtered_actions:
-                        # TODO: populate workflow env in WorkflowEventData correctly
-                        task_params = build_trigger_action_task_params(
-                            action, detector, workflow_event_data
-                        )
-                        trigger_action.apply_async(
-                            kwargs=task_params, headers={"sentry-propagate-traces": False}
-                        )
+                    fire_actions(filtered_actions, detector, workflow_event_data)
 
     logger.info(
         "workflow_engine.delayed_workflow.triggered_actions_summary",

src/sentry/workflow_engine/processors/workflow.py

@@ -16,15 +16,17 @@
 from sentry.utils import json
 from sentry.workflow_engine.models import Action, DataConditionGroup, Detector, Workflow
 from sentry.workflow_engine.models.workflow_data_condition_group import WorkflowDataConditionGroup
-from sentry.workflow_engine.processors.action import filter_recently_fired_workflow_actions
+from sentry.workflow_engine.processors.action import (
+    filter_recently_fired_workflow_actions,
+    fire_actions,
+)
 from sentry.workflow_engine.processors.contexts.workflow_event_context import (
     WorkflowEventContext,
     WorkflowEventContextData,
 )
 from sentry.workflow_engine.processors.data_condition_group import process_data_condition_group
 from sentry.workflow_engine.processors.detector import get_detector_by_event
 from sentry.workflow_engine.processors.workflow_fire_history import create_workflow_fire_histories
-from sentry.workflow_engine.tasks.actions import build_trigger_action_task_params, trigger_action
 from sentry.workflow_engine.types import WorkflowEventData
 from sentry.workflow_engine.utils import log_context
 from sentry.workflow_engine.utils.metrics import metrics_incr
@@ -458,13 +460,9 @@ def process_workflows(
         return triggered_workflows
 
     should_trigger_actions = should_fire_workflow_actions(organization, event_data.group.type)
-
     create_workflow_fire_histories(
         detector, actions, event_data, should_trigger_actions, is_delayed=False
     )
-
-    for action in actions:
-        task_params = build_trigger_action_task_params(action, detector, event_data)
-        trigger_action.apply_async(kwargs=task_params, headers={"sentry-propagate-traces": False})
+    fire_actions(actions, detector, event_data)
 
     return triggered_workflows

tests/sentry/workflow_engine/processors/test_workflow.py

@@ -23,6 +23,7 @@
     Workflow,
 )
 from sentry.workflow_engine.models.data_condition import Condition
+from sentry.workflow_engine.models.workflow_fire_history import WorkflowFireHistory
 from sentry.workflow_engine.processors.workflow import (
     DelayedWorkflowItem,
     delete_workflow,
@@ -271,6 +272,41 @@ def test_metrics_triggered_workflows(self, mock_incr: MagicMock) -> None:
             tags={"detector_type": self.error_detector.type},
         )
 
+    @patch("sentry.workflow_engine.processors.action.trigger_action.apply_async")
+    def test_workflow_fire_history_with_action_deduping(
+        self, mock_trigger_action: MagicMock
+    ) -> None:
+        """Fire a single action, but record that it was fired for multiple workflows"""
+
+        self.action_group, self.action = self.create_workflow_action(workflow=self.error_workflow)
+
+        error_workflow_2 = self.create_workflow(
+            name="error_workflow_2",
+            when_condition_group=self.create_data_condition_group(),
+        )
+        self.create_detector_workflow(
+            detector=self.error_detector,
+            workflow=error_workflow_2,
+        )
+        self.action_group_2, self.action_2 = self.create_workflow_action(workflow=error_workflow_2)
+
+        error_workflow_3 = self.create_workflow(
+            name="error_workflow_3",
+            when_condition_group=self.create_data_condition_group(),
+        )
+        self.create_detector_workflow(
+            detector=self.error_detector,
+            workflow=error_workflow_3,
+        )
+        self.action_group_3, self.action_3 = self.create_workflow_action(workflow=error_workflow_3)
+
+        process_workflows(self.event_data)
+
+        assert WorkflowFireHistory.objects.count() == 3
+        assert (
+            mock_trigger_action.call_count == 1
+        )  # TODO: determine if this is ideal, the actions are duped but not in the same workflows
+
 
 @mock_redis_buffer()
 class TestEvaluateWorkflowTriggers(BaseWorkflowTest):

tests/sentry/workflow_engine/test_integration.py

@@ -198,7 +198,7 @@ def test_workflow_engine__workflows__no_flag(self) -> None:
             mock_process_workflow.assert_not_called()
 
 
-@mock.patch("sentry.workflow_engine.processors.workflow.trigger_action.apply_async")
+@mock.patch("sentry.workflow_engine.processors.action.trigger_action.apply_async")
 @mock_redis_buffer()
 class TestWorkflowEngineIntegrationFromErrorPostProcess(BaseWorkflowIntegrationTest):
     def setUp(self) -> None: