feat(eap): Sets up double deletion of occurrences with EAP

src/sentry/deletions/tasks/nodestore.py

@@ -6,8 +6,9 @@
 import sentry_sdk
 from snuba_sdk import DeleteQuery, Request
 
-from sentry import eventstream, nodestore
+from sentry import eventstream, nodestore, options
 from sentry.deletions.tasks.scheduled import MAX_RETRIES, logger
+from sentry.eventstream.eap import delete_groups_from_eap_rpc
 from sentry.exceptions import DeleteAborted
 from sentry.models.eventattachment import EventAttachment
 from sentry.models.userreport import UserReport
@@ -205,6 +206,51 @@ def delete_events_from_eventstore(
         eventstream_state = eventstream.backend.start_delete_groups(project_id, group_ids)
         eventstream.backend.end_delete_groups(eventstream_state)
 
+    delete_events_from_eap(organization_id, project_id, group_ids, dataset)
+
+
+def delete_events_from_eap(
+    organization_id: int,
+    project_id: int,
+    group_ids: Sequence[int],
+    dataset: Dataset,
+) -> None:
+    eap_deletion_allowlist = options.get("eventstream.eap.deletion_enabled.organization_allowlist")
+    if organization_id not in eap_deletion_allowlist:
+        return
+
+    try:
+        response = delete_groups_from_eap_rpc(
+            organization_id=organization_id,
+            project_id=project_id,
+            group_ids=group_ids,
+            referrer="deletions.group.eap",
+        )
+        logger.info(
+            "eap.delete_groups.completed",
+            extra={
+                "organization_id": organization_id,
+                "project_id": project_id,
+                "group_count": len(group_ids),
+                "matching_items_count": response.matching_items_count,
+            },
+        )
+    except Exception as e:
+        logger.exception(
+            "eap.delete_groups.failed",
+            extra={
+                "organization_id": organization_id,
+                "project_id": project_id,
+                "group_ids": group_ids[:10],
+                "error": str(e),
+            },
+        )
+        metrics.incr(
+            "deletions.eap.failed",
+            tags={"dataset": dataset.value},
+            sample_rate=1.0,
+        )
+
 
 def delete_events_from_eventstore_issue_platform(
     organization_id: int, project_id: int, group_ids: Sequence[int], times_seen_list: Sequence[int]

src/sentry/eventstream/eap.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+
+from sentry_protos.snuba.v1.endpoint_delete_trace_items_pb2 import (
+    DeleteTraceItemsRequest,
+    DeleteTraceItemsResponse,
+)
+from sentry_protos.snuba.v1.request_common_pb2 import (
+    TRACE_ITEM_TYPE_OCCURRENCE,
+    RequestMeta,
+    TraceItemFilterWithType,
+)
+from sentry_protos.snuba.v1.trace_item_attribute_pb2 import AttributeKey, AttributeValue, IntArray
+from sentry_protos.snuba.v1.trace_item_filter_pb2 import (
+    AndFilter,
+    ComparisonFilter,
+    TraceItemFilter,
+)
+
+from sentry.utils import snuba_rpc
+
+logger = logging.getLogger(__name__)
+
+
+def delete_groups_from_eap_rpc(
+    organization_id: int,
+    project_id: int,
+    group_ids: Sequence[int],
+    referrer: str = "deletions.group",
+) -> DeleteTraceItemsResponse:
+    """
+    Delete occurrences from EAP for the given group IDs.
+    """
+
+    if not group_ids:
+        raise ValueError("group_ids must not be empty")
+
+    project_filter = _create_project_filter(project_id)
+    group_id_filter = _create_group_id_filter(list(group_ids))
+    combined_filter = TraceItemFilter(
+        and_filter=AndFilter(filters=[project_filter, group_id_filter])
+    )
+    filter_with_type = TraceItemFilterWithType(
+        item_type=TRACE_ITEM_TYPE_OCCURRENCE,
+        filter=combined_filter,
+    )
+
+    request = DeleteTraceItemsRequest(
+        meta=RequestMeta(
+            organization_id=organization_id,
+            project_ids=[project_id],
+            referrer=referrer,
+            cogs_category="deletions",
+        ),
+        filters=[filter_with_type],
+    )
+    response = snuba_rpc.rpc(request, DeleteTraceItemsResponse)
+
+    return response
+
+
+def _create_project_filter(project_id: int) -> TraceItemFilter:
+    return TraceItemFilter(
+        comparison_filter=ComparisonFilter(
+            key=AttributeKey(
+                type=AttributeKey.TYPE_INT,
+                name="sentry.project_id",
+            ),
+            op=ComparisonFilter.OP_EQUALS,
+            value=AttributeValue(val_int=project_id),
+        )
+    )
+
+
+def _create_group_id_filter(group_ids: list[int]) -> TraceItemFilter:
+    return TraceItemFilter(
+        comparison_filter=ComparisonFilter(
+            key=AttributeKey(
+                type=AttributeKey.TYPE_INT,
+                name="sentry.group_id",
+            ),
+            op=ComparisonFilter.OP_IN,
+            value=AttributeValue(val_int_array=IntArray(values=group_ids)),
+        )
+    )

src/sentry/options/defaults.py

@@ -3578,9 +3578,9 @@
     flags=FLAG_MODIFIABLE_BOOL | FLAG_AUTOMATOR_MODIFIABLE,
 )
 
-# Allow list for projects with LLM issue detection enabled
+# The allowlist of organization IDs for which deletion from EAP is enabled.
 register(
-    "issue-detection.llm-detection.projects-allowlist",
+    "eventstream.eap.deletion_enabled.organization_allowlist",
     type=Sequence,
     default=[],
     flags=FLAG_ALLOW_EMPTY | FLAG_AUTOMATOR_MODIFIABLE,

tests/sentry/eventstream/test_eap.py

@@ -0,0 +1,88 @@
+from unittest.mock import patch
+
+import pytest
+from sentry_protos.snuba.v1.endpoint_delete_trace_items_pb2 import DeleteTraceItemsResponse
+from sentry_protos.snuba.v1.request_common_pb2 import TRACE_ITEM_TYPE_OCCURRENCE, ResponseMeta
+
+from sentry.deletions.tasks.nodestore import delete_events_from_eap
+from sentry.eventstream.eap import delete_groups_from_eap_rpc
+from sentry.snuba.dataset import Dataset
+from sentry.testutils.cases import TestCase
+
+
+class TestEAPDeletion(TestCase):
+    def setUp(self):
+        self.organization_id = 1
+        self.project_id = 123
+        self.group_ids = [1, 2, 3]
+
+    @patch("sentry.eventstream.eap.snuba_rpc.rpc")
+    def test_deletion_with_error_dataset(self, mock_rpc):
+        mock_rpc.return_value = DeleteTraceItemsResponse(
+            meta=ResponseMeta(),
+            matching_items_count=150,
+        )
+
+        with self.options(
+            {"eventstream.eap.deletion_enabled.organization_allowlist": [self.organization_id]}
+        ):
+            delete_events_from_eap(
+                self.organization_id, self.project_id, self.group_ids, Dataset.Events
+            )
+
+        assert mock_rpc.call_count == 1
+
+        request = mock_rpc.call_args[0][0]
+        assert request.meta.organization_id == self.organization_id
+        assert request.meta.project_ids == [self.project_id]
+        assert request.meta.referrer == "deletions.group.eap"
+        assert request.meta.cogs_category == "deletions"
+
+        assert len(request.filters) == 1
+        assert request.filters[0].item_type == TRACE_ITEM_TYPE_OCCURRENCE
+
+    @patch("sentry.eventstream.eap.snuba_rpc.rpc")
+    def test_multiple_group_ids(self, mock_rpc):
+        mock_rpc.return_value = DeleteTraceItemsResponse(
+            meta=ResponseMeta(),
+            matching_items_count=500,
+        )
+
+        many_group_ids = [10, 20, 30, 40, 50]
+
+        with self.options(
+            {"eventstream.eap.deletion_enabled.organization_allowlist": [self.organization_id]}
+        ):
+            delete_events_from_eap(
+                self.organization_id, self.project_id, many_group_ids, Dataset.Events
+            )
+
+        request = mock_rpc.call_args[0][0]
+        group_filter = request.filters[0].filter.and_filter.filters[1]
+        assert list(group_filter.comparison_filter.value.val_int_array.values) == many_group_ids
+
+    @patch("sentry.eventstream.eap.snuba_rpc.rpc")
+    def test_organization_not_in_allowlist_skips_deletion(self, mock_rpc):
+        with self.options({"eventstream.eap.deletion_enabled.organization_allowlist": [456, 789]}):
+            delete_events_from_eap(
+                self.organization_id, self.project_id, self.group_ids, Dataset.Events
+            )
+
+        mock_rpc.assert_not_called()
+
+    @patch("sentry.eventstream.eap.snuba_rpc.rpc")
+    def test_empty_allowlist_skips_deletion(self, mock_rpc):
+        with self.options({"eventstream.eap.deletion_enabled.organization_allowlist": []}):
+            delete_events_from_eap(
+                self.organization_id, self.project_id, self.group_ids, Dataset.Events
+            )
+
+        mock_rpc.assert_not_called()
+
+    def test_empty_group_ids_raises_error(self):
+        with pytest.raises(ValueError, match="group_ids must not be empty"):
+            delete_groups_from_eap_rpc(
+                organization_id=self.organization_id,
+                project_id=self.project_id,
+                group_ids=[],
+            )