fix(tracing): Serialize spans outside the SpanBatcher lock (#8023)

sentry_sdk/_span_batcher.py

@@ -127,26 +127,30 @@ def add(self, span: "SpanJSON") -> None:
             with self._lock:
                 size = len(self._span_buffer[span["trace_id"]])
                 if size >= self.MAX_BEFORE_DROP:
-                    self._record_lost_func(
-                        reason="queue_overflow",
-                        data_category="span",
-                        quantity=1,
-                    )
-                    return None
-
-                self._span_buffer[span["trace_id"]].append(span)
-                self._running_size[span["trace_id"]] += self._estimate_size(span)
-
-                if (
-                    span["is_segment"] is True
-                    or size + 1 >= self.MAX_BEFORE_FLUSH
-                    or self._running_size[span["trace_id"]]
-                    >= self.MAX_BYTES_BEFORE_FLUSH
-                ):
-                    self._pending_flush.add(span["trace_id"])
-                    notify = True
+                    dropped = True
                 else:
-                    notify = False
+                    dropped = False
+                    self._span_buffer[span["trace_id"]].append(span)
+                    self._running_size[span["trace_id"]] += self._estimate_size(span)
+
+                    if (
+                        span["is_segment"] is True
+                        or size + 1 >= self.MAX_BEFORE_FLUSH
+                        or self._running_size[span["trace_id"]]
+                        >= self.MAX_BYTES_BEFORE_FLUSH
+                    ):
+                        self._pending_flush.add(span["trace_id"])
+                        notify = True
+                    else:
+                        notify = False
+
+            if dropped:
+                self._record_lost_func(
+                    reason="queue_overflow",
+                    data_category="span",
+                    quantity=1,
+                )
+                return None
 
             if notify:
                 self._flush_event.set()
@@ -195,50 +199,55 @@ def _flush(self, only_pending: bool = False) -> None:
             if not buckets:
                 return
 
-            envelopes = []
-
+            flushed_buckets: list[tuple[str, list["SpanJSON"]]] = []
             for bucket_id in buckets:
-                spans = self._span_buffer.get(bucket_id)
-                if not spans:
-                    continue
-
-                dsc = spans[0]["_segment_span"]._dynamic_sampling_context()
-
-                # Max per envelope is 1000, so if we happen to have more than
-                # 1000 spans in one bucket, we'll need to separate them.
-                for start in range(0, len(spans), self.MAX_ENVELOPE_SIZE):
-                    end = min(start + self.MAX_ENVELOPE_SIZE, len(spans))
-
-                    envelope = Envelope(
+                spans = self._span_buffer.pop(bucket_id, None)
+                if spans:
+                    flushed_buckets.append((bucket_id, spans))
+                    self._running_size.pop(bucket_id, None)
+
+        if not flushed_buckets:
+            return
+
+        envelopes = []
+
+        # Serialize outside the lock to prevent blocking concurrent additions
+        # and avoid lock-order inversion during serialization or GC finalizers.
+        for _, spans in flushed_buckets:
+            dsc = spans[0]["_segment_span"]._dynamic_sampling_context()
+
+            # Max per envelope is 1000, so if we happen to have more than
+            # 1000 spans in one bucket, we'll need to separate them.
+            for start in range(0, len(spans), self.MAX_ENVELOPE_SIZE):
+                end = min(start + self.MAX_ENVELOPE_SIZE, len(spans))
+
+                envelope = Envelope(
+                    headers={
+                        "sent_at": format_timestamp(datetime.now(timezone.utc)),
+                        "trace": dsc,
+                    }
+                )
+
+                envelope.add_item(
+                    Item(
+                        type=self.TYPE,
+                        content_type=self.CONTENT_TYPE,
                         headers={
-                            "sent_at": format_timestamp(datetime.now(timezone.utc)),
-                            "trace": dsc,
-                        }
+                            "item_count": end - start,
+                        },
+                        payload=PayloadRef(
+                            json={
+                                "version": 2,
+                                "items": [
+                                    self._to_transport_format(spans[j])
+                                    for j in range(start, end)
+                                ],
+                            }
+                        ),
                     )
+                )
 
-                    envelope.add_item(
-                        Item(
-                            type=self.TYPE,
-                            content_type=self.CONTENT_TYPE,
-                            headers={
-                                "item_count": end - start,
-                            },
-                            payload=PayloadRef(
-                                json={
-                                    "version": 2,
-                                    "items": [
-                                        self._to_transport_format(spans[j])
-                                        for j in range(start, end)
-                                    ],
-                                }
-                            ),
-                        )
-                    )
-
-                    envelopes.append(envelope)
-
-                del self._span_buffer[bucket_id]
-                del self._running_size[bucket_id]
+                envelopes.append(envelope)
 
         for envelope in envelopes:
             self._capture_func(envelope)

tests/tracing/test_span_batcher.py

@@ -572,3 +572,112 @@ def _flush(self, only_pending=False):
     batcher._flush_loop()
 
     assert calls == [1]
+
+
+def test_flush_serializes_outside_the_lock(sentry_init, capture_envelopes):
+    """Regression test for https://github.com/getsentry/sentry-python/issues/8023."""
+    sentry_init(
+        traces_sample_rate=1.0,
+        trace_lifecycle="stream",
+    )
+    capture_envelopes()
+
+    client = sentry_sdk.get_client()
+    batcher = client.span_batcher
+
+    lock_held_while_serializing = []
+    original_to_transport_format = batcher._to_transport_format
+
+    def record_lock_state(item):
+        lock_held_while_serializing.append(batcher._lock.locked())
+        return original_to_transport_format(item)
+
+    batcher._to_transport_format = record_lock_state
+
+    with sentry_sdk.traces.start_span(name="test span"):
+        pass
+
+    client.flush()
+
+    assert lock_held_while_serializing
+    assert not any(lock_held_while_serializing)
+
+
+def test_concurrent_add_not_blocked_during_serialization(
+    sentry_init, capture_envelopes
+):
+    """Verify concurrent additions are not blocked while spans are serialized."""
+    import threading
+
+    sentry_init(
+        traces_sample_rate=1.0,
+        trace_lifecycle="stream",
+    )
+    capture_envelopes()
+
+    client = sentry_sdk.get_client()
+    batcher = client.span_batcher
+
+    serializing_event = threading.Event()
+    continue_event = threading.Event()
+    original_to_transport_format = batcher._to_transport_format
+
+    def pause_serializing(item):
+        serializing_event.set()
+        continue_event.wait(timeout=5)
+        return original_to_transport_format(item)
+
+    batcher._to_transport_format = pause_serializing
+
+    with sentry_sdk.traces.start_span(name="span 1"):
+        pass
+
+    flush_thread = threading.Thread(target=client.flush)
+    flush_thread.start()
+
+    assert serializing_event.wait(timeout=5)
+
+    # The lock must NOT be held while serialization is paused
+    lock_acquired = batcher._lock.acquire(blocking=False)
+    if lock_acquired:
+        batcher._lock.release()
+
+    continue_event.set()
+    flush_thread.join()
+
+    assert lock_acquired is True
+
+
+def test_record_lost_called_outside_lock(
+    sentry_init, capture_record_lost_event_calls, monkeypatch
+):
+    """Verify record_lost is called outside the batcher lock when buffer overflows."""
+    monkeypatch.setattr(SpanBatcher, "MAX_BEFORE_DROP", 1)
+    monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000)
+
+    sentry_init(
+        traces_sample_rate=1.0,
+        trace_lifecycle="stream",
+    )
+    record_lost_event_calls = capture_record_lost_event_calls()
+    client = sentry_sdk.get_client()
+    batcher = client.span_batcher
+
+    lock_held_on_lost = []
+    original_record_lost = batcher._record_lost_func
+
+    def record_lost(*args, **kwargs):
+        lock_held_on_lost.append(batcher._lock.locked())
+        return original_record_lost(*args, **kwargs)
+
+    batcher._record_lost_func = record_lost
+
+    with sentry_sdk.traces.start_span(name="parent"):
+        with sentry_sdk.traces.start_span(name="span 1"):
+            pass
+        with sentry_sdk.traces.start_span(name="span 2"):
+            pass
+
+    assert all(not held for held in lock_held_on_lost)
+    assert len(lock_held_on_lost) >= 1
+    assert ("queue_overflow", "span", None, 1) in record_lost_event_calls