Add SourcemapDetectorHandler with atomic state transitions

src/sentry/processing_errors/grouptype.py

@@ -1,12 +1,212 @@
 from __future__ import annotations
 
+import enum
+import logging
+from collections.abc import Mapping
 from dataclasses import dataclass
+from typing import Any, override
 
+from django.db import IntegrityError
+from django.utils import timezone
 from sentry_redis_tools.sliding_windows_rate_limiter import Quota
 
 from sentry.issues.grouptype import GroupCategory, GroupType, NotificationConfig
+from sentry.issues.issue_occurrence import IssueEvidence
+from sentry.models.eventerror import EventErrorType
 from sentry.types.group import PriorityLevel
-from sentry.workflow_engine.types import DetectorSettings
+from sentry.utils import metrics
+from sentry.workflow_engine.handlers.detector.base import (
+    DetectorOccurrence,
+    EventData,
+    GroupedDetectorEvaluationResult,
+)
+from sentry.workflow_engine.handlers.detector.stateful import (
+    DetectorThresholds,
+    StatefulDetectorHandler,
+)
+from sentry.workflow_engine.models import DataPacket, DetectorState
+from sentry.workflow_engine.processors.data_condition_group import ProcessedDataConditionGroup
+from sentry.workflow_engine.types import (
+    DetectorEvaluationResult,
+    DetectorGroupKey,
+    DetectorPriorityLevel,
+    DetectorSettings,
+)
+
+logger = logging.getLogger(__name__)
+
+# Error types from symbolicator that indicate sourcemap configuration problems
+JS_SOURCEMAP_ERROR_TYPES = frozenset(
+    {
+        EventErrorType.JS_MISSING_SOURCE,
+        EventErrorType.JS_INVALID_SOURCEMAP,
+        EventErrorType.JS_MISSING_SOURCES_CONTENT,
+        EventErrorType.JS_SCRAPING_DISABLED,
+        EventErrorType.JS_INVALID_SOURCEMAP_LOCATION,
+    }
+)
+
+
+class SourcemapCheckStatus(enum.IntEnum):
+    """
+    Status values used as the comparison value for detector conditions.
+    These must match the values used in DataCondition.comparison when
+    provisioning the detector.
+    """
+
+    SUCCESS = 0
+    FAILURE = 1
+
+
+@dataclass(frozen=True)
+class SourcemapPacketValue:
+    """
+    The data payload passed into the sourcemap detector via DataPacket.
+    Represents the error event that triggered detection
+    """
+
+    event_id: str
+    event_data: Mapping[str, Any]
+
+
+class SourcemapDetectorHandler(StatefulDetectorHandler[SourcemapPacketValue, SourcemapCheckStatus]):
+    @override
+    @property
+    def thresholds(self) -> DetectorThresholds:
+        return {
+            DetectorPriorityLevel.OK: 1,
+            DetectorPriorityLevel.HIGH: 1,
+        }
+
+    @override
+    def extract_value(self, data_packet: DataPacket[SourcemapPacketValue]) -> SourcemapCheckStatus:
+        errors = data_packet.packet.event_data.get("errors", [])
+        has_js_errors = any(e.get("type") in JS_SOURCEMAP_ERROR_TYPES for e in errors)
+        return SourcemapCheckStatus.FAILURE if has_js_errors else SourcemapCheckStatus.SUCCESS
+
+    @override
+    def extract_dedupe_value(self, data_packet: DataPacket[SourcemapPacketValue]) -> int:
+        # Not used — we override evaluate_impl and skip dedupe logic
+        return 0
+
+    @override
+    def build_issue_fingerprint(self, group_key: DetectorGroupKey = None) -> list[str]:
+        return [f"{self.detector.project_id}:sourcemap"]
+
+    @override
+    def create_occurrence(
+        self,
+        evaluation_result: ProcessedDataConditionGroup,
+        data_packet: DataPacket[SourcemapPacketValue],
+        priority: DetectorPriorityLevel,
+    ) -> tuple[DetectorOccurrence, EventData]:
+        event_data_dict = data_packet.packet.event_data
+        errors = event_data_dict.get("errors", [])
+        js_errors = [e for e in errors if e.get("type") in JS_SOURCEMAP_ERROR_TYPES]
+        error_types = {e.get("type", "unknown") for e in js_errors}
+
+        evidence_data: dict[str, Any] = {
+            "error_types": sorted(error_types),
+            "sample_event_id": data_packet.packet.event_id,
+        }
+
+        evidence_display = [
+            IssueEvidence(
+                name="Error types",
+                value=", ".join(sorted(error_types)),
+                important=True,
+            ),
+        ]
+
+        occurrence = DetectorOccurrence(
+            issue_title="Broken source maps detected",
+            subtitle="Source maps are not configured correctly for this project",
+            evidence_data=evidence_data,
+            evidence_display=evidence_display,
+            type=SourcemapConfigurationType,
+            level="warning",
+            culprit="",
+            priority=priority,
+        )
+
+        event_data: EventData = {
+            "platform": event_data_dict.get("platform", "other"),
+            "sdk": event_data_dict.get("sdk"),
+        }
+
+        return (occurrence, event_data)
+
+    @override
+    def evaluate_impl(
+        self, data_packet: DataPacket[SourcemapPacketValue]
+    ) -> GroupedDetectorEvaluationResult:
+        """
+        Custom evaluation that skips dedupe and threshold counting.
+        Uses atomic DB updates for state transitions instead of the
+        parent's batched state manager approach.
+        """
+        data_value = self.extract_value(data_packet)
+        results: dict[DetectorGroupKey, DetectorEvaluationResult] = {}
+
+        condition_results, evaluated_priority = self._evaluation_detector_conditions(data_value)
+
+        if condition_results is None or condition_results.logic_result.triggered is False:
+            return GroupedDetectorEvaluationResult(result=results, tainted=False)
+
+        # Only handle triggering (FAILURE → HIGH). Resolution is handled
+        # by a separate periodic task, not by the detector handler.
+        if evaluated_priority != DetectorPriorityLevel.HIGH:
+            return GroupedDetectorEvaluationResult(result=results, tainted=False)
+
+        # Atomic state transition: use filter().update() as a lock.
+        # If another process already triggered, rows_updated will be 0.
+        rows_updated = self._try_state_transition(DetectorPriorityLevel.HIGH)
+
+        if not rows_updated:
+            metrics.incr("workflow_engine.sourcemap_detector.state_transition_conflict")
+            return GroupedDetectorEvaluationResult(result=results, tainted=False)
+
+        results[None] = self._build_detector_evaluation_result(
+            None,
+            DetectorPriorityLevel.HIGH,
+            condition_results,
+            data_packet,
+            data_value,
+        )
+
+        return GroupedDetectorEvaluationResult(result=results, tainted=False)
+
+    def _try_state_transition(self, new_priority: DetectorPriorityLevel) -> int:
+        """
+        Attempt an atomic state transition on DetectorState.
+
+        Uses filter().update() so that concurrent processes racing to make
+        the same transition will see rows_updated=0 and bail out.
+        """
+        detector_states = self.state_manager.bulk_get_detector_state([None])
+        detector_state = detector_states.get(None)
+
+        if detector_state is None:
+            try:
+                DetectorState.objects.create(
+                    detector=self.detector,
+                    detector_group_key=None,
+                    is_triggered=True,
+                    state=new_priority,
+                )
+                return 1
+            except IntegrityError:
+                # Another process created the row first, just exit
+                return 0
+
+        return DetectorState.objects.filter(
+            id=detector_state.id,
+            is_triggered=False,
+        ).update(
+            is_triggered=True,
+            state=new_priority,
+            date_updated=timezone.now(),
+        )
 
 
 @dataclass(frozen=True)
@@ -23,7 +223,7 @@ class SourcemapConfigurationType(GroupType):
     creation_quota = Quota(3600, 60, 100)
     notification_config = NotificationConfig(context=[])
     detector_settings = DetectorSettings(
-        handler=None,
+        handler=SourcemapDetectorHandler,
         validator=None,
         config_schema={},
     )

tests/sentry/processing_errors/test_grouptype.py

@@ -0,0 +1,147 @@
+from typing import Any
+
+from sentry.issues.issue_occurrence import IssueOccurrence
+from sentry.processing_errors.grouptype import (
+    SourcemapCheckStatus,
+    SourcemapConfigurationType,
+    SourcemapDetectorHandler,
+    SourcemapPacketValue,
+)
+from sentry.testutils.cases import TestCase
+from sentry.workflow_engine.models.data_condition import Condition
+from sentry.workflow_engine.models.data_source import DataPacket
+from sentry.workflow_engine.models.detector import Detector
+from sentry.workflow_engine.models.detector_state import DetectorState
+from sentry.workflow_engine.types import DetectorEvaluationResult, DetectorPriorityLevel
+
+
+class TestSourcemapDetectorHandler(TestCase):
+    def create_sourcemap_detector(
+        self,
+        detector_state: DetectorPriorityLevel = DetectorPriorityLevel.OK,
+    ) -> Detector:
+        condition_group = self.create_data_condition_group(
+            organization=self.project.organization,
+        )
+        self.create_data_condition(
+            comparison=SourcemapCheckStatus.FAILURE,
+            type=Condition.EQUAL,
+            condition_result=DetectorPriorityLevel.HIGH,
+            condition_group=condition_group,
+        )
+        self.create_data_condition(
+            comparison=SourcemapCheckStatus.SUCCESS,
+            type=Condition.EQUAL,
+            condition_result=DetectorPriorityLevel.OK,
+            condition_group=condition_group,
+        )
+        detector = self.create_detector(
+            type=SourcemapConfigurationType.slug,
+            project=self.project,
+            name="Sourcemap Configuration",
+            config={},
+            workflow_condition_group=condition_group,
+        )
+        self.create_detector_state(
+            detector=detector,
+            state=detector_state,
+            is_triggered=detector_state == DetectorPriorityLevel.HIGH,
+        )
+        return detector
+
+    def make_packet(
+        self,
+        errors: list | None = None,
+        event_id: str = "abc123",
+        platform: str = "javascript",
+    ) -> DataPacket[SourcemapPacketValue]:
+        if errors is None:
+            errors = []
+        event_data: dict[str, Any] = {
+            "errors": errors,
+            "platform": platform,
+            "sdk": {"name": "sentry.javascript.browser", "version": "7.0.0"},
+        }
+        return DataPacket(
+            source_id=str(self.project.id),
+            packet=SourcemapPacketValue(
+                event_id=event_id,
+                event_data=event_data,
+            ),
+        )
+
+    def handle_result(
+        self, detector: Detector, data_packet: DataPacket[SourcemapPacketValue]
+    ) -> DetectorEvaluationResult | None:
+        handler = SourcemapDetectorHandler(detector)
+        evaluation = handler.evaluate(data_packet)
+        if None not in evaluation:
+            return None
+        return evaluation[None]
+
+    def test_failure_creates_occurrence(self) -> None:
+        detector = self.create_sourcemap_detector()
+
+        errors = [
+            {"type": "js_no_source", "url": "https://example.com/app.js"},
+            {"type": "js_invalid_source", "url": "https://example.com/vendor.js"},
+        ]
+
+        result = self.handle_result(
+            detector,
+            self.make_packet(errors=errors, event_id="test-event-123", platform="javascript"),
+        )
+
+        assert result is not None
+        assert result.priority == DetectorPriorityLevel.HIGH
+        assert isinstance(result.result, IssueOccurrence)
+        assert result.result.issue_title == "Broken source maps detected"
+        assert result.result.evidence_data["error_types"] == ["js_invalid_source", "js_no_source"]
+        assert result.result.evidence_data["sample_event_id"] == "test-event-123"
+        assert result.event_data is not None
+        assert result.event_data["platform"] == "javascript"
+
+        state = DetectorState.objects.get(detector=detector)
+        assert state.is_triggered is True
+        assert state.state == str(DetectorPriorityLevel.HIGH)
+
+    def test_duplicate_failure_does_not_trigger(self) -> None:
+        detector = self.create_sourcemap_detector()
+        packet = self.make_packet(
+            errors=[{"type": "js_no_source", "url": "https://example.com/app.js"}],
+        )
+
+        result = self.handle_result(detector, packet)
+        assert result is not None
+        assert isinstance(result.result, IssueOccurrence)
+
+        result = self.handle_result(detector, packet)
+        assert result is None
+
+    def test_no_sourcemap_errors_does_not_trigger(self) -> None:
+        detector = self.create_sourcemap_detector()
+
+        assert self.handle_result(detector, self.make_packet(errors=[])) is None
+        assert (
+            self.handle_result(
+                detector,
+                self.make_packet(errors=[{"type": "native_missing_dsym", "image": "libfoo.so"}]),
+            )
+            is None
+        )
+
+    def test_failure_without_detector_state_creates_it(self) -> None:
+        detector = self.create_sourcemap_detector()
+        DetectorState.objects.filter(detector=detector).delete()
+
+        result = self.handle_result(
+            detector,
+            self.make_packet(
+                errors=[{"type": "js_no_source", "url": "https://example.com/app.js"}],
+            ),
+        )
+
+        assert result is not None
+        assert isinstance(result.result, IssueOccurrence)
+        state = DetectorState.objects.get(detector=detector)
+        assert state.is_triggered is True