feat(ingest-metrics): Ensure billing metrics consumer avoids outcome double-billing

src/sentry/features/temporary.py

@@ -465,4 +465,5 @@ def register_temporary_features(manager: FeatureManager) -> None:
 
     manager.add("projects:workflow-engine-performance-detectors", ProjectFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
 
+    manager.add("organizations:relay-generate-billing-outcome", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # fmt: on

src/sentry/ingest/billing_metrics_consumer.py

@@ -42,6 +42,7 @@ class BillingTxCountMetricConsumerStrategy(ProcessingStrategy[KafkaPayload]):
 
     span_metric_id = SPAN_METRICS_NAMES["c:spans/usage@none"]
     span_is_segment_tag = str(SHARED_TAG_STRINGS["is_segment"])
+    billing_outcome_accepted_tag = str(SHARED_TAG_STRINGS["billing_outcome_emitted"])
 
     def __init__(self, next_step: ProcessingStrategy[Any]) -> None:
         self.__next_step = next_step
@@ -79,6 +80,10 @@ def _count_processed_items(self, generic_metric: GenericMetric) -> Mapping[DataC
         if generic_metric["metric_id"] != self.span_metric_id:
             return {}
 
+        # If relay already produced an outcome for this item, ignore it.
+        if generic_metric["tags"].get(self.billing_outcome_accepted_tag) == "true":
+            return {}
+
         value = generic_metric["value"]
         try:
             quantity = max(int(value), 0)  # type: ignore[arg-type]

src/sentry/sentry_metrics/indexer/strings.py

@@ -202,6 +202,7 @@
     "has_transaction": PREFIX + 281,
     "was_transaction": PREFIX + 282,
     "is_segment": PREFIX + 283,
+    "billing_outcome_emitted": PREFIX + 284,
     # GENERAL/MISC (don't have a category)
     "": PREFIX + 1000,
 }

tests/sentry/ingest/billing_metrics_consumer/test_billing_metrics_consumer_kafka.py

@@ -141,3 +141,102 @@ def generate_kafka_message(generic_metric: GenericMetric) -> Message[KafkaPayloa
 
     strategy.join()
     assert next_step.join.call_count == 1
+
+
+@mock.patch("sentry.ingest.billing_metrics_consumer.track_outcome")
+def test_no_double_billing_outcomes(track_outcome) -> None:
+    topic = Topic("snuba-generic-metrics")
+
+    organization = 123
+    project_1 = 56789
+
+    span_usage_mri = "c:spans/usage@none"
+    span_usage_id = SPAN_METRICS_NAMES[span_usage_mri]
+
+    generic_metrics: list[GenericMetric] = [
+        {
+            "mapping_meta": {"c": {str(span_usage_id): span_usage_mri}},
+            "metric_id": span_usage_id,
+            "type": "d",
+            "org_id": organization,
+            "project_id": project_1,
+            "timestamp": 123456,
+            "value": 65.0,
+            "tags": {str(SHARED_TAG_STRINGS["billing_outcome_emitted"]): "true"},
+            "use_case_id": "spans",
+            "retention_days": 90,
+        },
+        {
+            "mapping_meta": {"c": {str(span_usage_id): span_usage_mri}},
+            "metric_id": span_usage_id,
+            "type": "d",
+            "org_id": organization,
+            "project_id": project_1,
+            "timestamp": 123456,
+            "value": 12.0,
+            "tags": {str(SHARED_TAG_STRINGS["is_segment"]): "true"},
+            "use_case_id": "spans",
+            "retention_days": 90,
+        },
+    ]
+
+    next_step = mock.MagicMock()
+
+    strategy = BillingTxCountMetricConsumerStrategy(
+        next_step=next_step,
+    )
+
+    generate_kafka_message_counter = 0
+
+    def generate_kafka_message(generic_metric: GenericMetric) -> Message[KafkaPayload]:
+        nonlocal generate_kafka_message_counter
+
+        encoded = orjson.dumps(generic_metric)
+        payload = KafkaPayload(key=None, value=encoded, headers=[])
+        message = Message(
+            BrokerValue(
+                payload,
+                Partition(topic, index=0),
+                generate_kafka_message_counter,
+                datetime.now(timezone.utc),
+            )
+        )
+        generate_kafka_message_counter += 1
+        return message
+
+    # Mimick the behavior of StreamProcessor._run_once: Call poll repeatedly,
+    # then call submit when there is a message.
+    strategy.poll()
+    strategy.poll()
+    assert track_outcome.call_count == 0
+    for generic_metric in generic_metrics:
+        strategy.poll()
+        strategy.submit(generate_kafka_message(generic_metric))
+
+    assert track_outcome.mock_calls == [
+        mock.call(
+            org_id=organization,
+            project_id=project_1,
+            key_id=None,
+            outcome=Outcome.ACCEPTED,
+            reason=None,
+            timestamp=mock.ANY,
+            event_id=None,
+            category=DataCategory.SPAN,
+            quantity=12,
+        ),
+        mock.call(
+            org_id=organization,
+            project_id=project_1,
+            key_id=None,
+            outcome=Outcome.ACCEPTED,
+            reason=None,
+            timestamp=mock.ANY,
+            event_id=None,
+            category=DataCategory.TRANSACTION,
+            quantity=12,
+        ),
+    ]
+
+    strategy.join()
+    assert next_step.join.call_count == 1

tests/sentry/sentry_metrics/test_prevent_modifications_to_indexer_strings.py

@@ -1,7 +1,7 @@
 from hashlib import sha256
 
 LOCKED_FILE = "src/sentry/sentry_metrics/indexer/strings.py"
-LOCKED_DIGEST = "f984bf497f587f7d52665714cb9a049cf9f515f0bbb288f8782cc018484e65ea"
+LOCKED_DIGEST = "9a2c7aeb2abf0c8ce704504201073df1d4776ff2d74d7a7af349bfc0198dedb7"
 MESSAGE = f"""{LOCKED_FILE} is locked.
 
 * We have detected you made changes to this file.