feat(workflow_engine): Split fast and slow condition evaluation

src/sentry/workflow_engine/handlers/detector/stateful.py

@@ -177,8 +177,9 @@ def evaluate_group_key_value(
         # store these in `DetectorStateData.counter_updates`, but we don't have anywhere to set the required
         # thresholds at the moment. Probably should be a field on the Detector? Could also be on the condition
         # level, but usually we want to set this at a higher level.
+        # TODO 2: Validate that we will never have slow conditions here.
         new_status = DetectorPriorityLevel.OK
-        is_group_condition_met, condition_results = evaluate_condition_group(
+        is_group_condition_met, condition_results, _ = evaluate_condition_group(
             self.condition_group, value
         )
 

src/sentry/workflow_engine/models/workflow.py

@@ -67,17 +67,19 @@ class Meta:
             )
         ]
 
-    def evaluate_trigger_conditions(self, job: WorkflowJob) -> bool:
+    def evaluate_trigger_conditions(self, job: WorkflowJob) -> tuple[bool, list[DataCondition]]:
         """
         Evaluate the conditions for the workflow trigger and return if the evaluation was successful.
         If there aren't any workflow trigger conditions, the workflow is considered triggered.
         """
         if self.when_condition_group is None:
-            return True
+            return True, []
 
         job["workflow"] = self
-        evaluation, _ = evaluate_condition_group(self.when_condition_group, job)
-        return evaluation
+        evaluation, _, remaining_conditions = evaluate_condition_group(
+            self.when_condition_group, job
+        )
+        return evaluation, remaining_conditions
 
 
 def get_slow_conditions(workflow: Workflow) -> list[DataCondition]:

src/sentry/workflow_engine/processors/action.py

@@ -7,9 +7,14 @@
 
 from sentry.db.models.manager.base_query_set import BaseQuerySet
 from sentry.models.group import Group
-from sentry.workflow_engine.models import Action, ActionGroupStatus, DataConditionGroup, Workflow
-from sentry.workflow_engine.processors.data_condition_group import evaluate_condition_group
-from sentry.workflow_engine.types import WorkflowJob
+from sentry.workflow_engine.models import (
+    Action,
+    ActionGroupStatus,
+    DataCondition,
+    DataConditionGroup,
+)
+
+EnqueuedAction = tuple[DataConditionGroup, list[DataCondition]]
 
 
 def get_action_last_updated_statuses(now: datetime, actions: BaseQuerySet[Action], group: Group):
@@ -74,29 +79,3 @@ def filter_recently_fired_workflow_actions(
     filtered_actions = actions.filter(id__in=actions_to_include | actions_without_statuses_ids)
 
     return filtered_actions
-
-
-def evaluate_workflow_action_filters(
-    workflows: set[Workflow], job: WorkflowJob
-) -> BaseQuerySet[Action]:
-    filtered_action_groups: set[DataConditionGroup] = set()
-
-    # gets the list of the workflow ids, and then get the workflow_data_condition_groups for those workflows
-    workflow_ids = {workflow.id for workflow in workflows}
-
-    action_conditions = DataConditionGroup.objects.filter(
-        workflowdataconditiongroup__workflow_id__in=workflow_ids
-    ).distinct()
-
-    for action_condition in action_conditions:
-        evaluation, result = evaluate_condition_group(action_condition, job)
-
-        if evaluation:
-            filtered_action_groups.add(action_condition)
-
-    # get the actions for any of the triggered data condition groups
-    actions = Action.objects.filter(
-        dataconditiongroupaction__condition_group__in=filtered_action_groups
-    ).distinct()
-
-    return filter_recently_fired_workflow_actions(actions, job["event"].group)

src/sentry/workflow_engine/processors/data_condition.py

@@ -0,0 +1,16 @@
+from sentry.workflow_engine.models.data_condition import DataCondition, is_slow_condition
+
+
+def split_conditions_by_speed(
+    conditions: list[DataCondition],
+) -> tuple[list[DataCondition], list[DataCondition]]:
+    fast_conditions: list[DataCondition] = []
+    slow_conditions: list[DataCondition] = []
+
+    for condition in conditions:
+        if is_slow_condition(condition):
+            slow_conditions.append(condition)
+        else:
+            fast_conditions.append(condition)
+
+    return fast_conditions, slow_conditions

src/sentry/workflow_engine/processors/data_condition_group.py

@@ -3,12 +3,15 @@
 
 from sentry.utils.function_cache import cache_func_for_models
 from sentry.workflow_engine.models import DataCondition, DataConditionGroup
-from sentry.workflow_engine.types import ProcessedDataConditionResult
+from sentry.workflow_engine.processors.data_condition import split_conditions_by_speed
+from sentry.workflow_engine.types import DataConditionResult, ProcessedDataConditionResult
 
 logger = logging.getLogger(__name__)
 
 T = TypeVar("T")
 
+DataConditionGroupResult = tuple[bool, list[DataConditionResult], list[DataCondition]]
+
 
 @cache_func_for_models(
     [(DataCondition, lambda condition: (condition.condition_group_id,))],
@@ -18,19 +21,54 @@ def get_data_conditions_for_group(data_condition_group_id: int) -> list[DataCond
     return list(DataCondition.objects.filter(condition_group_id=data_condition_group_id))
 
 
+def process_condition_group_results(
+    results: list[tuple[bool, DataConditionResult]],
+    logic_type: str,
+) -> ProcessedDataConditionResult:
+    logic_result = False
+    condition_results = []
+
+    if logic_type == DataConditionGroup.Type.NONE:
+        # if we get to this point, no conditions were met
+        # because we would have short-circuited
+        logic_result = True
+
+    elif logic_type == DataConditionGroup.Type.ANY:
+        logic_result = any([result[0] for result in results])
+
+        if logic_result:
+            condition_results = [result[1] for result in results if result[0]]
+
+    elif logic_type == DataConditionGroup.Type.ALL:
+        conditions_met = [result[0] for result in results]
+        logic_result = all(conditions_met)
+
+        if logic_result:
+            condition_results = [result[1] for result in results if result[0]]
+
+    return logic_result, condition_results
+
+
 def evaluate_condition_group(
     data_condition_group: DataConditionGroup,
     value: T,
-) -> ProcessedDataConditionResult:
+    is_fast: bool = True,
+) -> DataConditionGroupResult:
     """
     Evaluate the conditions for a given group and value.
     """
-    results = []
+    results: list[tuple[bool, DataConditionResult]] = []
     conditions = get_data_conditions_for_group(data_condition_group.id)
 
+    if is_fast:
+        conditions, remaining_conditions = split_conditions_by_speed(conditions)
+    else:
+        _, conditions = split_conditions_by_speed(conditions)
+        remaining_conditions = []
+
     if len(conditions) == 0:
         # if we don't have any conditions, always return True
-        return True, []
+        return True, [], remaining_conditions
 
     for condition in conditions:
         evaluation_result = condition.evaluate_value(value)
@@ -39,46 +77,41 @@ def evaluate_condition_group(
         if is_condition_triggered:
             # Check for short-circuiting evaluations
             if data_condition_group.logic_type == data_condition_group.Type.ANY_SHORT_CIRCUIT:
-                return is_condition_triggered, [evaluation_result]
+                return is_condition_triggered, [evaluation_result], []
 
             if data_condition_group.logic_type == data_condition_group.Type.NONE:
-                return False, []
+                return False, [], []
 
         results.append((is_condition_triggered, evaluation_result))
 
-    if data_condition_group.logic_type == data_condition_group.Type.NONE:
-        # if we get to this point, no conditions were met
-        return True, []
-
-    elif data_condition_group.logic_type == data_condition_group.Type.ANY:
-        is_any_condition_met = any([result[0] for result in results])
+    logic_type = data_condition_group.logic_type
+    logic_result, condition_results = process_condition_group_results(
+        results,
+        logic_type,
+    )
 
-        if is_any_condition_met:
-            condition_results = [result[1] for result in results if result[0]]
-            return is_any_condition_met, condition_results
-
-    elif data_condition_group.logic_type == data_condition_group.Type.ALL:
-        conditions_met = [result[0] for result in results]
-        is_all_conditions_met = all(conditions_met)
+    if (not logic_result and logic_type == DataConditionGroup.Type.ALL) or (
+        logic_result and logic_type == DataConditionGroup.Type.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 completd logic result
+        remaining_conditions = []
 
-        if is_all_conditions_met:
-            condition_results = [result[1] for result in results if result[0]]
-            return is_all_conditions_met, condition_results
-
-    return False, []
+    return logic_result, condition_results, remaining_conditions
 
 
 def process_data_condition_group(
     data_condition_group_id: int,
     value: Any,
-) -> ProcessedDataConditionResult:
+) -> DataConditionGroupResult:
     try:
         group = DataConditionGroup.objects.get_from_cache(id=data_condition_group_id)
     except DataConditionGroup.DoesNotExist:
         logger.exception(
             "DataConditionGroup does not exist",
             extra={"id": data_condition_group_id},
         )
-        return False, []
+        return False, [], []
 
     return evaluate_condition_group(group, value)

src/sentry/workflow_engine/processors/workflow.py

@@ -4,10 +4,17 @@
 import sentry_sdk
 
 from sentry import buffer
+from sentry.db.models.manager.base_query_set import BaseQuerySet
 from sentry.utils import json, metrics
-from sentry.workflow_engine.models import Detector, Workflow, WorkflowDataConditionGroup
-from sentry.workflow_engine.models.workflow import get_slow_conditions
-from sentry.workflow_engine.processors.action import evaluate_workflow_action_filters
+from sentry.workflow_engine.models import (
+    Action,
+    DataCondition,
+    DataConditionGroup,
+    Detector,
+    Workflow,
+    WorkflowDataConditionGroup,
+)
+from sentry.workflow_engine.processors.action import filter_recently_fired_workflow_actions
 from sentry.workflow_engine.processors.data_condition_group import evaluate_condition_group
 from sentry.workflow_engine.processors.detector import get_detector_by_event
 from sentry.workflow_engine.types import WorkflowJob
@@ -17,6 +24,7 @@
 WORKFLOW_ENGINE_BUFFER_LIST_KEY = "workflow_engine_delayed_processing_buffer"
 
 
+# TODO remove this method
 def get_data_condition_groups_to_fire(
     workflows: set[Workflow], job: WorkflowJob
 ) -> dict[int, list[int]]:
@@ -30,7 +38,7 @@ def get_data_condition_groups_to_fire(
 
     for workflow_dcg in workflow_dcgs:
         action_condition = workflow_dcg.condition_group
-        evaluation, result = evaluate_condition_group(action_condition, job)
+        evaluation, result, _ = evaluate_condition_group(action_condition, job)
 
         if evaluation:
             workflow_action_groups[workflow_dcg.workflow_id].append(action_condition.id)
@@ -69,19 +77,54 @@ def evaluate_workflow_triggers(workflows: set[Workflow], job: WorkflowJob) -> se
     workflows_to_enqueue: set[Workflow] = set()
 
     for workflow in workflows:
-        if workflow.evaluate_trigger_conditions(job):
-            triggered_workflows.add(workflow)
+        evaluation, remaining_conditions = workflow.evaluate_trigger_conditions(job)
+        if remaining_conditions:
+            workflows_to_enqueue.add(workflow)
         else:
-            if get_slow_conditions(workflow):
-                # enqueue to be evaluated later
-                workflows_to_enqueue.add(workflow)
+            if evaluation:
+                # Only add workflows that have no remaining conditions to check
+                triggered_workflows.add(workflow)
 
     if workflows_to_enqueue:
         enqueue_workflows(workflows_to_enqueue, job)
 
     return triggered_workflows
 
 
+def evaluate_workflows_action_filters(
+    workflows: set[Workflow],
+    job: WorkflowJob,
+) -> BaseQuerySet[Action]:
+    filtered_action_groups: set[DataConditionGroup] = set()
+    enqueued_conditions: list[DataCondition] = []
+
+    # gets the list of the workflow ids, and then get the workflow_data_condition_groups for those workflows
+    workflow_ids = {workflow.id for workflow in workflows}
+
+    action_conditions = DataConditionGroup.objects.filter(
+        workflowdataconditiongroup__workflow_id__in=workflow_ids
+    ).distinct()
+
+    for action_condition in action_conditions:
+        evaluation, result, remaining_conditions = evaluate_condition_group(action_condition, job)
+
+        if remaining_conditions:
+            # If there are remaining conditions for the action filter to evaluate,
+            # then return the list of conditions to enqueue
+            enqueued_conditions.extend(remaining_conditions)
+        else:
+            # if we don't have any other conditions to evaluate, add the action to the list
+            if evaluation:
+                filtered_action_groups.add(action_condition)
+
+    # get the actions for any of the triggered data condition groups
+    actions = Action.objects.filter(
+        dataconditiongroupaction__condition_group__in=filtered_action_groups
+    ).distinct()
+
+    return filter_recently_fired_workflow_actions(actions, job["event"].group)
+
+
 def process_workflows(job: WorkflowJob) -> set[Workflow]:
     """
     This method will get the detector based on the event, and then gather the associated workflows.
@@ -101,7 +144,7 @@ def process_workflows(job: WorkflowJob) -> set[Workflow]:
     # Get the workflows, evaluate the when_condition_group, finally evaluate the actions for workflows that are triggered
     workflows = set(Workflow.objects.filter(detectorworkflow__detector_id=detector.id).distinct())
     triggered_workflows = evaluate_workflow_triggers(workflows, job)
-    actions = evaluate_workflow_action_filters(triggered_workflows, job)
+    actions = evaluate_workflows_action_filters(triggered_workflows, job)
 
     with sentry_sdk.start_span(op="workflow_engine.process_workflows.trigger_actions"):
         for action in actions:

tests/sentry/workflow_engine/models/test_workflow.py

@@ -1,3 +1,4 @@
+from sentry.workflow_engine.models.data_condition import Condition
 from sentry.workflow_engine.types import WorkflowJob
 from tests.sentry.workflow_engine.test_base import BaseWorkflowTest
 
@@ -12,19 +13,32 @@ def setUp(self):
         self.job = WorkflowJob({"event": self.group_event})
 
     def test_evaluate_trigger_conditions__condition_new_event__True(self):
-        evaluation = self.workflow.evaluate_trigger_conditions(self.job)
+        evaluation, _ = self.workflow.evaluate_trigger_conditions(self.job)
         assert evaluation is True
 
     def test_evaluate_trigger_conditions__condition_new_event__False(self):
         # Update event to have been seen before
         self.group_event.group.times_seen = 5
 
-        evaluation = self.workflow.evaluate_trigger_conditions(self.job)
+        evaluation, _ = self.workflow.evaluate_trigger_conditions(self.job)
         assert evaluation is False
 
     def test_evaluate_trigger_conditions__no_conditions(self):
         self.workflow.when_condition_group = None
         self.workflow.save()
 
-        evaluation = self.workflow.evaluate_trigger_conditions(self.job)
+        evaluation, _ = self.workflow.evaluate_trigger_conditions(self.job)
         assert evaluation is True
+
+    def test_evaluate_trigger_conditions__slow_condition(self):
+        # Update group to _all_, since the fast condition is met
+        self.data_condition_group.update(logic_type="all")
+
+        slow_condition = self.create_data_condition(
+            type=Condition.EVENT_FREQUENCY_COUNT, comparison={"interval": "1d", "value": 7}
+        )
+        self.data_condition_group.conditions.add(slow_condition)
+        evaluation, remaining_conditions = self.workflow.evaluate_trigger_conditions(self.job)
+
+        assert evaluation is True
+        assert remaining_conditions == [slow_condition]