fix(aci): ANY_SHORT_CIRCUIT early exit (#87114)

Author: cathteng

src/sentry/workflow_engine/processors/data_condition_group.py

@@ -126,9 +126,13 @@ def process_data_condition_group(
     conditions_to_evaluate = [(condition, value) for condition in conditions]
     logic_result, condition_results = evaluate_data_conditions(conditions_to_evaluate, logic_type)
 
-    if (not logic_result and logic_type == DataConditionGroup.Type.ALL) or (
-        logic_result and logic_type == DataConditionGroup.Type.ANY
-    ):
+    is_short_circuit_all = not logic_result and logic_type == DataConditionGroup.Type.ALL
+    is_short_circuit_any = logic_result and logic_type in (
+        DataConditionGroup.Type.ANY,
+        DataConditionGroup.Type.ANY_SHORT_CIRCUIT,
+    )
+
+    if is_short_circuit_all or is_short_circuit_any:
         # if we have a logic type of all and a False result,
         # or if we have a logic type of any and a True result, then
         #  we can short-circuit any remaining conditions since we have a completed logic result

tests/sentry/workflow_engine/processors/test_workflow.py

@@ -343,6 +343,58 @@ def test_enqueues_workflow_any_logic_type(self):
         )
         assert project_ids[0][0] == self.project.id
 
+    def test_skips_enqueuing_any(self):
+        # skips slow conditions if the condition group evaluates to True without evaluating them
+        assert self.workflow.when_condition_group
+        self.workflow.when_condition_group.update(
+            logic_type=DataConditionGroup.Type.ANY_SHORT_CIRCUIT
+        )
+
+        self.create_data_condition(
+            condition_group=self.workflow.when_condition_group,
+            type=Condition.EVENT_FREQUENCY_COUNT,
+            comparison={
+                "interval": "1h",
+                "value": 100,
+            },
+            condition_result=True,
+        )
+
+        triggered_workflows = evaluate_workflow_triggers({self.workflow}, self.job)
+        assert triggered_workflows == {self.workflow}
+        project_ids = buffer.backend.get_sorted_set(
+            WORKFLOW_ENGINE_BUFFER_LIST_KEY, 0, self.buffer_timestamp
+        )
+        assert len(project_ids) == 0
+
+    def test_skips_enqueuing_all(self):
+        assert self.workflow.when_condition_group
+        self.workflow.when_condition_group.conditions.all().delete()
+        self.workflow.when_condition_group.update(logic_type=DataConditionGroup.Type.ALL)
+
+        self.create_data_condition(
+            condition_group=self.workflow.when_condition_group,
+            type=Condition.EVENT_FREQUENCY_COUNT,
+            comparison={
+                "interval": "1h",
+                "value": 100,
+            },
+            condition_result=True,
+        )
+        self.create_data_condition(
+            condition_group=self.workflow.when_condition_group,
+            type=Condition.REGRESSION_EVENT,  # fast condition, does not pass
+            comparison=True,
+            condition_result=True,
+        )
+
+        triggered_workflows = evaluate_workflow_triggers({self.workflow}, self.job)
+        assert not triggered_workflows
+        project_ids = buffer.backend.get_sorted_set(
+            WORKFLOW_ENGINE_BUFFER_LIST_KEY, 0, self.buffer_timestamp
+        )
+        assert len(project_ids) == 0
+
 
 class TestEvaluateWorkflowActionFilters(BaseWorkflowTest):
     def setUp(self):