fix(native): Isolate and bound flamegraph attachment processing

src/sentry/lang/native/flamegraphs.py

@@ -2,6 +2,7 @@
 from typing import Any
 
 import rapidjson
+import sentry_sdk
 from rest_framework import serializers
 
 from sentry import features
@@ -13,11 +14,18 @@
 )
 from sentry.ingest.consumer import CACHE_TIMEOUT
 from sentry.models.project import Project
+from sentry.utils import metrics
 
 # See: https://develop.sentry.dev/sdk/telemetry/attachments/#flame-graph-validation-and-limits
 MAX_FLAMEGRAPH_SIZE = 10 * 1024 * 1024
 MAX_SYMBOLICATION_FRAMES = 2500
 MAX_SAMPLE_COUNT = 9007199254740991
+# Bound node allocations independently of the serialized attachment size.
+MAX_EXPANDED_NODES = 100_000
+
+
+class FlamegraphExpansionLimitExceeded(Exception):
+    pass
 
 
 class _StringField(serializers.Field):
@@ -113,11 +121,15 @@ def _is_valid_flamegraph(payload: Any) -> bool:
 def expand_inline_frames(
     nodes: list[dict[str, Any]], chains: dict[int, list[int]]
 ) -> list[dict[str, Any]]:
+    if len(nodes) > MAX_EXPANDED_NODES:
+        raise FlamegraphExpansionLimitExceeded
     # Keep captured indices stable so roots and shared child references remain valid.
     output = [dict(node) for node in nodes]
     for index, original in enumerate(nodes):
         current = output[index]
         for frame_id in chains.get(original["frame_id"], [])[1:]:
+            if len(output) >= MAX_EXPANDED_NODES:
+                raise FlamegraphExpansionLimitExceeded
             child = {"frame_id": frame_id, "sample_count": original["sample_count"]}
             current["children"] = [len(output)]
             output.append(child)
@@ -141,27 +153,49 @@ def get_flamegraphs(self) -> Generator[tuple[int, dict[str, Any]]]:
         for index, attachment in enumerate(self._attachments):
             if attachment.type != "event.flamegraph":
                 continue
+            reason = "load_error"
             try:
-                raw = attachment.load_data(self._project)
-            except (MissingAttachmentChunks, FileNotFoundError):
-                continue
-            if len(raw) > MAX_FLAMEGRAPH_SIZE:
-                continue
-            try:
+                try:
+                    raw = attachment.load_data(self._project)
+                except (MissingAttachmentChunks, FileNotFoundError):
+                    metrics.incr("process.native.flamegraph.skipped", tags={"reason": "load_error"})
+                    continue
+                if len(raw) > MAX_FLAMEGRAPH_SIZE:
+                    metrics.incr("process.native.flamegraph.skipped", tags={"reason": "raw_size"})
+                    continue
+                reason = "invalid_json"
                 payload = rapidjson.loads(raw, allow_nan=False)
-            except (rapidjson.JSONDecodeError, UnicodeDecodeError):
-                continue
-            if not isinstance(payload, dict):
-                continue
-            state = payload.get("_symbolication")
-            if isinstance(state, dict) and state.get("status") == "completed":
+                if not isinstance(payload, dict):
+                    metrics.incr(
+                        "process.native.flamegraph.skipped", tags={"reason": "invalid_schema"}
+                    )
+                    continue
+                state = payload.get("_symbolication")
+                if isinstance(state, dict) and state.get("status") == "completed":
+                    metrics.incr(
+                        "process.native.flamegraph.skipped", tags={"reason": "already_completed"}
+                    )
+                    continue
+                nodes = payload.get("nodes")
+                if isinstance(nodes, list) and len(nodes) > MAX_EXPANDED_NODES:
+                    metrics.incr("process.native.flamegraph.skipped", tags={"reason": "node_limit"})
+                    continue
+                reason = "invalid_schema"
+                if not _is_valid_flamegraph(payload):
+                    metrics.incr(
+                        "process.native.flamegraph.skipped", tags={"reason": "invalid_schema"}
+                    )
+                    continue
+            except Exception as error:
+                sentry_sdk.capture_exception(error)
+                metrics.incr("process.native.flamegraph.skipped", tags={"reason": reason})
                 continue
-            if _is_valid_flamegraph(payload):
-                yield index, payload
+            yield index, payload
 
     def save(self, attachment_index: int, payload: dict[str, Any]) -> bool:
         serialized = rapidjson.dumps(payload, ensure_ascii=False, allow_nan=False).encode()
         if len(serialized) > MAX_FLAMEGRAPH_SIZE:
+            metrics.incr("process.native.flamegraph.skipped", tags={"reason": "rewrite_size"})
             return False
         attachment = self._attachments[attachment_index]
         metadata = attachment.meta()

src/sentry/lang/native/processing.py

@@ -6,14 +6,14 @@
 from typing import Any
 
 import sentry_sdk
-from requests.exceptions import RequestException
 from symbolic.debuginfo import normalize_debug_id
 from symbolic.exceptions import ParseDebugIdError
 
-from sentry import options
+from sentry import features, options
 from sentry.lang.native.error import SymbolicationFailed, write_error
 from sentry.lang.native.flamegraphs import (
     MAX_SYMBOLICATION_FRAMES,
+    FlamegraphExpansionLimitExceeded,
     Flamegraphs,
     expand_inline_frames,
 )
@@ -32,6 +32,7 @@
     signal_from_data,
 )
 from sentry.models.eventerror import EventErrorType
+from sentry.models.project import Project
 from sentry.options.rollout import in_random_rollout
 from sentry.stacktraces.functions import trim_function_name
 from sentry.stacktraces.processing import StacktraceInfo, find_stacktraces_in_data
@@ -526,30 +527,69 @@ def _merge_flamegraph_frames(
 
 
 def process_native_stacktraces(symbolicator: Symbolicator, data: Any) -> Any:
-    flamegraphs = Flamegraphs(symbolicator.project, data)
+    event_changed = _process_native_event_stacktraces(symbolicator, data)
+    attachments_changed = _process_flamegraphs(symbolicator, data)
+    return data if event_changed or attachments_changed else None
+
+
+def _process_flamegraphs(symbolicator: Symbolicator, data: Any) -> bool:
+    # Imported at invocation time to avoid the symbolication task's processing import cycle.
+    from sentry.tasks.symbolication import SymbolicationTimeout
+
     attachments_changed = False
-    for attachment_index, payload in flamegraphs.get_flamegraphs():
-        modules, frame_ids, stacktraces = _get_flamegraph_stacktraces(payload)
-        if not stacktraces:
-            continue
-        try:
-            response = symbolicator.process_payload(
-                platform=payload["platform"],
-                stacktraces=stacktraces,
-                modules=modules,
-                frame_order=FrameOrder.caller_first,
-                apply_source_context=False,
-            )
-        except (OSError, RequestException) as error:
-            sentry_sdk.capture_exception(error)
-            continue
-        if not response or response.get("status") != "completed":
-            continue
-        resolved_stacktraces = response.get("stacktraces", [])
-        if len(resolved_stacktraces) != len(frame_ids):
-            continue
-        output = _merge_flamegraph_frames(payload, frame_ids, resolved_stacktraces)
-        attachments_changed |= flamegraphs.save(attachment_index, output)
+    try:
+        flamegraphs = Flamegraphs(symbolicator.project, data)
+        for attachment_index, payload in flamegraphs.get_flamegraphs():
+            reason = "processing_error"
+            try:
+                modules, frame_ids, stacktraces = _get_flamegraph_stacktraces(payload)
+                if not stacktraces:
+                    metrics.incr(
+                        "process.native.flamegraph.skipped", tags={"reason": "no_stacktraces"}
+                    )
+                    continue
+                reason = "request_error"
+                response = symbolicator.process_payload(
+                    platform=payload["platform"],
+                    stacktraces=stacktraces,
+                    modules=modules,
+                    frame_order=FrameOrder.caller_first,
+                    apply_source_context=False,
+                )
+                if not response or response.get("status") != "completed":
+                    metrics.incr(
+                        "process.native.flamegraph.skipped", tags={"reason": "request_error"}
+                    )
+                    continue
+                resolved_stacktraces = response.get("stacktraces", [])
+                if len(resolved_stacktraces) != len(frame_ids):
+                    metrics.incr(
+                        "process.native.flamegraph.skipped", tags={"reason": "invalid_response"}
+                    )
+                    continue
+                reason = "merge_error"
+                output = _merge_flamegraph_frames(payload, frame_ids, resolved_stacktraces)
+                reason = "save_error"
+                attachments_changed |= flamegraphs.save(attachment_index, output)
+            except SymbolicationTimeout as error:
+                sentry_sdk.capture_exception(error)
+                metrics.incr("process.native.flamegraph.skipped", tags={"reason": "timeout"})
+                break
+            except FlamegraphExpansionLimitExceeded as error:
+                sentry_sdk.capture_exception(error)
+                metrics.incr(
+                    "process.native.flamegraph.skipped", tags={"reason": "expansion_limit"}
+                )
+            except Exception as error:
+                sentry_sdk.capture_exception(error)
+                metrics.incr("process.native.flamegraph.skipped", tags={"reason": reason})
+    except Exception as error:
+        sentry_sdk.capture_exception(error)
+        metrics.incr("process.native.flamegraph.skipped", tags={"reason": "processing_error"})
+    return attachments_changed
+
+
+def _process_native_event_stacktraces(symbolicator: Symbolicator, data: Any) -> Any:
     stacktrace_infos = [
         stacktrace
         for stacktrace in find_stacktraces_in_data(data)
@@ -572,7 +612,7 @@ def process_native_stacktraces(symbolicator: Symbolicator, data: Any) -> Any:
     ]
 
     if not any(stacktrace["frames"] for stacktrace in stacktraces):
-        return data if attachments_changed else None
+        return None
 
     signal = signal_from_data(data)
 
@@ -726,7 +766,7 @@ def emit_apple_symbol_stats(apple_symbol_stats, data):
 
 
 def get_native_symbolication_functions(
-    data: Mapping[str, Any], stacktraces: list[StacktraceInfo]
+    data: Mapping[str, Any], stacktraces: list[StacktraceInfo], project: Project
 ) -> list[SymbolicatorFunction]:
     """
     Yields the symbolication functions that will process the event,
@@ -741,8 +781,12 @@ def get_native_symbolication_functions(
         functions.append(SymbolicatorFunction.native)
         return functions  # early return to prevent duplicates
 
-    if has_native_stacktraces(stacktraces) or any(
-        attachment.get("type") == "event.flamegraph" for attachment in data.get("_attachments", [])
+    if has_native_stacktraces(stacktraces) or (
+        any(
+            attachment.get("type") == "event.flamegraph"
+            for attachment in data.get("_attachments", [])
+        )
+        and features.has("organizations:flamegraph-attachments", project.organization)
     ):
         # Run native symbolication first, to prevent sending minidump / ACR events to
         # symbolicator twice (after symbolication a minidump event will have native stacktraces).

src/sentry/tasks/store.py

@@ -223,7 +223,7 @@ def _do_preprocess_event(
     # The event will be submitted to Symbolicator for all returned platforms,
     # one after the other, so we handle mixed stacktraces.
     stacktraces = find_stacktraces_in_data(data)
-    symbolicate_functions = get_symbolication_functions(data, stacktraces)
+    symbolicate_functions = get_symbolication_functions(data, stacktraces, project)
     metrics.incr(
         "events.to-symbolicate",
         tags={f.value: True for f in symbolicate_functions},

src/sentry/tasks/symbolication.py

@@ -41,7 +41,7 @@
 
 
 def get_symbolication_functions(
-    data: Mapping[str, Any], stacktraces: list[StacktraceInfo]
+    data: Mapping[str, Any], stacktraces: list[StacktraceInfo], project: Project
 ) -> list[SymbolicatorFunction]:
     """Returns a list of Symbolicator platforms
     that apply to this event."""
@@ -56,7 +56,7 @@ def get_symbolication_functions(
     if is_js_event(data, stacktraces):
         functions.append(SymbolicatorFunction.js)
 
-    functions.extend(get_native_symbolication_functions(data, stacktraces))
+    functions.extend(get_native_symbolication_functions(data, stacktraces, project))
 
     return functions