fix(ACI): Don't send None values to workflow engine processing (#91777)

If an alert rule (that got migrated to become a detector, data
conditions, etc) has a comparison delta and ends up getting a `None`
aggregation value we were sending that value along to the data packet
which causes an error downstream. We can check that it's not None before
sending.

Fixes https://sentry.sentry.io/issues/6606822297/

Author: ceorourke

src/sentry/incidents/subscription_processor.py

@@ -312,6 +312,14 @@ def get_comparison_aggregation_value(
 
         if not comparison_aggregate:
             metrics.incr("incidents.alert_rules.skipping_update_comparison_value_invalid")
+            logger.info(
+                "No comparison aggregate",
+                extra={
+                    "alert_rule_id": self.alert_rule.id,
+                    "subscription_id": subscription_update.get("subscription_id"),
+                    "organization_id": self.alert_rule.organization_id,
+                },
+            )
             return None
 
         return (aggregation_value / comparison_aggregate) * 100
@@ -408,33 +416,34 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
                 },
             )
 
-        if features.has(
-            "organizations:workflow-engine-metric-alert-processing",
-            self.subscription.project.organization,
-        ):
-            packet = MetricDetectorUpdate(
-                entity=subscription_update.get("entity", ""),
-                subscription_id=subscription_update["subscription_id"],
-                values={"value": aggregation_value},
-                timestamp=self.last_update,
-            )
-            data_packet = DataPacket[MetricDetectorUpdate](
-                source_id=str(self.subscription.id), packet=packet
-            )
-            results = process_data_packets([data_packet], DATA_SOURCE_SNUBA_QUERY_SUBSCRIPTION)
+        if aggregation_value is not None:
             if features.has(
-                "organizations:workflow-engine-metric-alert-dual-processing-logs",
-                self.alert_rule.organization,
+                "organizations:workflow-engine-metric-alert-processing",
+                self.subscription.project.organization,
             ):
-                logger.info(
-                    "dual processing results for alert rule",
-                    extra={
-                        "results": results,
-                        "num_results": len(results),
-                        "value": aggregation_value,
-                        "rule_id": self.alert_rule.id,
-                    },
+                packet = MetricDetectorUpdate(
+                    entity=subscription_update.get("entity", ""),
+                    subscription_id=subscription_update["subscription_id"],
+                    values={"value": aggregation_value},
+                    timestamp=self.last_update,
+                )
+                data_packet = DataPacket[MetricDetectorUpdate](
+                    source_id=str(self.subscription.id), packet=packet
                 )
+                results = process_data_packets([data_packet], DATA_SOURCE_SNUBA_QUERY_SUBSCRIPTION)
+                if features.has(
+                    "organizations:workflow-engine-metric-alert-dual-processing-logs",
+                    self.alert_rule.organization,
+                ):
+                    logger.info(
+                        "dual processing results for alert rule",
+                        extra={
+                            "results": results,
+                            "num_results": len(results),
+                            "value": aggregation_value,
+                            "rule_id": self.alert_rule.id,
+                        },
+                    )
 
         has_anomaly_detection = features.has(
             "organizations:anomaly-detection-alerts", self.subscription.project.organization

tests/sentry/incidents/test_subscription_processor.py

@@ -3855,6 +3855,27 @@ def test_process_data_packets_called(self, mock_process_data_packets):
         assert data_packet_list[0].source_id == str(self.sub.id)
         assert data_packet_list[0].packet["values"] == {"value": 10}
 
+    @with_feature("organizations:workflow-engine-metric-alert-processing")
+    @mock.patch("sentry.incidents.subscription_processor.process_data_packets")
+    @mock.patch(
+        "sentry.incidents.subscription_processor.SubscriptionProcessor.get_comparison_aggregation_value"
+    )
+    def test_process_data_packets_not_called(
+        self, mock_get_comparison_aggregation_value, mock_process_data_packets
+    ):
+        rule = self.comparison_rule_above
+        trigger = self.trigger
+
+        detector = self.create_detector(name="hojicha", type=MetricIssue.slug)
+        data_source = self.create_data_source(source_id=str(self.sub.id))
+        data_source.detectors.set([detector])
+
+        mock_get_comparison_aggregation_value.return_value = None
+        self.send_update(
+            rule, trigger.alert_threshold + 1, timedelta(minutes=-10), subscription=self.sub
+        )
+        assert mock_process_data_packets.call_count == 0
+
 
 class MetricsCrashRateAlertProcessUpdateTest(ProcessUpdateBaseClass, BaseMetricsTestCase):
     @pytest.fixture(autouse=True)