From 0414de55921f2c481338136aeb89091879419ae3 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 13:15:56 +0200 Subject: [PATCH 01/21] Revert "fix(batcher): Add global flush trigger based on the span number (#7150)" This reverts commit c7e9e62cd03785b4c311d66293257f56097a08dc. --- sentry_sdk/_span_batcher.py | 18 +++------ tests/tracing/test_span_batcher.py | 63 +----------------------------- 2 files changed, 7 insertions(+), 74 deletions(-) diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index deadf1b40e..2033cf7845 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -21,10 +21,10 @@ class SpanBatcher(Batcher["SpanJSON"]): # MAX_BEFORE_FLUSH should be lower than MAX_BEFORE_DROP, so that there is # a bit of a buffer for spans that appear between the trigger to flush # and actually flushing the buffer. + # + # The max limits are all per trace (per bucket). MAX_ENVELOPE_SIZE = 1000 # spans - MAX_BEFORE_FLUSH = 1000 - GLOBAL_MAX_BEFORE_FLUSH = 5_000 MAX_BEFORE_DROP = 2000 GLOBAL_MAX_BEFORE_DROP = 10_000 @@ -105,13 +105,9 @@ def _flush_loop(self) -> None: self._flush(only_pending=True) - if ( - self._span_number >= self.GLOBAL_MAX_BEFORE_FLUSH - or self._total_running_size >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH - or ( - time.monotonic() - self._last_full_flush - >= self.FLUSH_WAIT_TIME + jitter - ) + if self._total_running_size >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH or ( + time.monotonic() - self._last_full_flush + >= self.FLUSH_WAIT_TIME + jitter ): self._flush() self._last_full_flush = time.monotonic() @@ -159,9 +155,7 @@ def add(self, span: "SpanJSON") -> None: notify = True else: notify = ( - self._span_number >= self.GLOBAL_MAX_BEFORE_FLUSH - or self._total_running_size - >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH + self._total_running_size >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH ) if notify: diff --git a/tests/tracing/test_span_batcher.py b/tests/tracing/test_span_batcher.py index 172117e742..fa34ab26e1 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -339,67 +339,6 @@ def test_weight_based_flushing_by_attribute_size( def test_global_length_based_flushing(sentry_init, capture_items, monkeypatch): - """A flush event is triggered when the batcher contains GLOBAL_MAX_BEFORE_FLUSH spans.""" - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BEFORE_FLUSH", 2) - # set the time-based flush limit to something huge so that we're not hitting - # it since we want to test GLOBAL_MAX_BEFORE_FLUSH instead - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="span"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span 2"): - pass - - time.sleep(0.1) - - assert len(items) == 2 - assert items[0].payload["name"] == "span" - - -def test_span_number_reset_after_length_based_flushing( - sentry_init, capture_items, monkeypatch -): - """Span is not flushed after a flush reduces the number of spans in the batcher below the global limit.""" - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BEFORE_FLUSH", 2) - # set the time-based flush limit to something huge so that we're not hitting - # it since we want to test GLOBAL_MAX_BYTES_BEFORE_FLUSH instead - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="span"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - assert len(items) == 2 - assert items[0].payload["name"] == "span" - - -def test_global_weight_based_flushing(sentry_init, capture_items, monkeypatch): """When the batcher reaches GLOBAL_MAX_BYTES_BEFORE_FLUSH, all buckets will be flushed.""" # Limit of 2_000 is just above the size of a bare span. monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BYTES_BEFORE_FLUSH", 2_000) @@ -427,7 +366,7 @@ def test_global_weight_based_flushing(sentry_init, capture_items, monkeypatch): assert items[0].payload["name"] == "span" -def test_total_size_reset_after_weight_based_flushing( +def test_total_size_reset_after_length_based_flushing( sentry_init, capture_items, monkeypatch ): """Span is not flushed after a flush reduces the combined span size in bytes below the global limit.""" From dd57e21e763889cff0367efaa1b9a8fdab23a6e2 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 13:16:07 +0200 Subject: [PATCH 02/21] Revert "feat(batcher): Add global flush trigger based on summed size estimates (#7144)" This reverts commit 2d86bfc3081f77c2a87ff74d6de5a82af9a0028b. --- sentry_sdk/_span_batcher.py | 16 ++----- tests/tracing/test_span_batcher.py | 67 ------------------------------ 2 files changed, 3 insertions(+), 80 deletions(-) diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index 2033cf7845..c1c40044ee 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -30,7 +30,6 @@ class SpanBatcher(Batcher["SpanJSON"]): GLOBAL_MAX_BEFORE_DROP = 10_000 MAX_BYTES_BEFORE_FLUSH = 5 * 1024 * 1024 # 5 MB - GLOBAL_MAX_BYTES_BEFORE_FLUSH = 25 * 1024 * 1024 # 25 MB FLUSH_WAIT_TIME = 5.0 @@ -51,8 +50,6 @@ def __init__( self._span_number: int = 0 self._running_size: dict[str, int] = defaultdict(lambda: 0) - self._total_running_size: int = 0 - self._capture_func = capture_func self._record_lost_func = record_lost_func self._running = True @@ -82,8 +79,6 @@ def _reset_thread_state(self) -> None: self._span_number = 0 self._running_size = defaultdict(lambda: 0) - self._total_running_size = 0 - self._running = True self._lock = threading.Lock() @@ -105,7 +100,7 @@ def _flush_loop(self) -> None: self._flush(only_pending=True) - if self._total_running_size >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH or ( + if ( time.monotonic() - self._last_full_flush >= self.FLUSH_WAIT_TIME + jitter ): @@ -142,9 +137,7 @@ def add(self, span: "SpanJSON") -> None: self._span_buffer[span["trace_id"]].append(span) self._span_number += 1 - estimated_size = self._estimate_size(span) - self._running_size[span["trace_id"]] += estimated_size - self._total_running_size += estimated_size + self._running_size[span["trace_id"]] += self._estimate_size(span) if ( len(self._span_buffer[span["trace_id"]]) >= self.MAX_BEFORE_FLUSH @@ -154,9 +147,7 @@ def add(self, span: "SpanJSON") -> None: self._pending_flush.add(span["trace_id"]) notify = True else: - notify = ( - self._total_running_size >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH - ) + notify = False if notify: self._flush_event.set() @@ -250,7 +241,6 @@ def _flush(self, only_pending: bool = False) -> None: self._span_number -= len(self._span_buffer[bucket_id]) del self._span_buffer[bucket_id] - self._total_running_size -= self._running_size[bucket_id] del self._running_size[bucket_id] for envelope in envelopes: diff --git a/tests/tracing/test_span_batcher.py b/tests/tracing/test_span_batcher.py index fa34ab26e1..507fef184b 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -338,69 +338,6 @@ def test_weight_based_flushing_by_attribute_size( assert envelopes[0].items[0].payload.json["items"][1]["name"] == "big span" -def test_global_length_based_flushing(sentry_init, capture_items, monkeypatch): - """When the batcher reaches GLOBAL_MAX_BYTES_BEFORE_FLUSH, all buckets will be flushed.""" - # Limit of 2_000 is just above the size of a bare span. - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BYTES_BEFORE_FLUSH", 2_000) - # set the time-based flush limit to something huge so that it doesn't - # interfere - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="span"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - assert len(items) == 2 - assert items[0].payload["name"] == "span" - - -def test_total_size_reset_after_length_based_flushing( - sentry_init, capture_items, monkeypatch -): - """Span is not flushed after a flush reduces the combined span size in bytes below the global limit.""" - # Limit of 2_000 is just above the size of a bare span. - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BYTES_BEFORE_FLUSH", 2_000) - # set the time-based flush limit to something huge so that it doesn't - # interfere - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="span"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - assert len(items) == 2 - assert items[0].payload["name"] == "span" - - def test_bucket_recreated_after_flush(sentry_init, capture_envelopes, monkeypatch): """Spans for a trace that arrive after that trace's bucket was flushed land in a fresh bucket.""" monkeypatch.setattr(SpanBatcher, "MAX_BEFORE_FLUSH", 2) @@ -608,8 +545,6 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): batcher._span_number = 1 batcher._running_size["test-trace-id"] = 42 - batcher._total_running_size = 42 - batcher._active.flag = True batcher._flush_event.set() batcher._running = False @@ -624,7 +559,6 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): span_number_reset = batcher._span_number == 0 running_size_reset = len(batcher._running_size) == 0 - total_running_size_reset = batcher._total_running_size == 0 active_reset = not getattr(batcher._active, "flag", False) event_reset = not batcher._flush_event.is_set() @@ -638,7 +572,6 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): and span_buffer_reset and span_number_reset and running_size_reset - and total_running_size_reset and active_reset and event_reset and running_reset From 8669777b18ce37bb1b52530fafb4953175f8dfa9 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 13:16:14 +0200 Subject: [PATCH 03/21] Revert "feat(batcher): Add hard span limit (#7143)" This reverts commit 58cba751e5b804fdc0105bdff1ab1dc4b4d819bc. --- sentry_sdk/_span_batcher.py | 19 ++----- tests/tracing/test_span_batcher.py | 87 ------------------------------ 2 files changed, 3 insertions(+), 103 deletions(-) diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index c1c40044ee..79285c3386 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -25,10 +25,7 @@ class SpanBatcher(Batcher["SpanJSON"]): # The max limits are all per trace (per bucket). MAX_ENVELOPE_SIZE = 1000 # spans MAX_BEFORE_FLUSH = 1000 - MAX_BEFORE_DROP = 2000 - GLOBAL_MAX_BEFORE_DROP = 10_000 - MAX_BYTES_BEFORE_FLUSH = 5 * 1024 * 1024 # 5 MB FLUSH_WAIT_TIME = 5.0 @@ -47,8 +44,6 @@ def __init__( # envelope. # trace_id -> span buffer self._span_buffer: dict[str, list["SpanJSON"]] = defaultdict(list) - self._span_number: int = 0 - self._running_size: dict[str, int] = defaultdict(lambda: 0) self._capture_func = capture_func self._record_lost_func = record_lost_func @@ -76,8 +71,6 @@ def _reset_in_child() -> None: def _reset_thread_state(self) -> None: self._span_buffer = defaultdict(list) - self._span_number = 0 - self._running_size = defaultdict(lambda: 0) self._running = True @@ -123,10 +116,8 @@ def add(self, span: "SpanJSON") -> None: return None with self._lock: - if ( - self._span_number >= self.GLOBAL_MAX_BEFORE_DROP - or len(self._span_buffer[span["trace_id"]]) >= self.MAX_BEFORE_DROP - ): + size = len(self._span_buffer[span["trace_id"]]) + if size >= self.MAX_BEFORE_DROP: self._record_lost_func( reason="queue_overflow", data_category="span", @@ -135,12 +126,10 @@ def add(self, span: "SpanJSON") -> None: return None self._span_buffer[span["trace_id"]].append(span) - self._span_number += 1 - self._running_size[span["trace_id"]] += self._estimate_size(span) if ( - len(self._span_buffer[span["trace_id"]]) >= self.MAX_BEFORE_FLUSH + size + 1 >= self.MAX_BEFORE_FLUSH or self._running_size[span["trace_id"]] >= self.MAX_BYTES_BEFORE_FLUSH ): @@ -238,9 +227,7 @@ def _flush(self, only_pending: bool = False) -> None: envelopes.append(envelope) - self._span_number -= len(self._span_buffer[bucket_id]) del self._span_buffer[bucket_id] - del self._running_size[bucket_id] for envelope in envelopes: diff --git a/tests/tracing/test_span_batcher.py b/tests/tracing/test_span_batcher.py index 507fef184b..c21d6cdea4 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -171,88 +171,6 @@ def test_drop_isolated_per_bucket( assert record_lost_event_calls.count(("queue_overflow", "span", None, 1)) == 1 -def test_drop_after_global_max_reached( - sentry_init, capture_envelopes, capture_record_lost_event_calls, monkeypatch -): - """New spans are dropped if the buffer reaches GLOBAL_MAX_BEFORE_DROP spans.""" - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BEFORE_DROP", 2) - # set the time-based flush limit to something huge so that we're not flushing - # prematurely - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - envelopes = capture_envelopes() - record_lost_event_calls = capture_record_lost_event_calls() - - with sentry_sdk.traces.start_span(name="span 1"): - pass - with sentry_sdk.traces.start_span(name="span 2"): - pass - with sentry_sdk.traces.start_span(name="span 3"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span 4"): - pass - - sentry_sdk.flush() - - assert len(envelopes) == 1 - - assert len(envelopes[0].items[0].payload.json["items"]) == 2 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" - assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" - - assert record_lost_event_calls.count(("queue_overflow", "span", None, 1)) == 2 - - -def test_capture_after_flush_with_global_limit( - sentry_init, capture_envelopes, monkeypatch -): - """New spans are captured again after a flush reduces the span number below the global limit.""" - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BEFORE_DROP", 2) - # set the time-based flush limit to something huge so that we're not flushing - # prematurely - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - envelopes = capture_envelopes() - - with sentry_sdk.traces.start_span(name="span 1"): - pass - with sentry_sdk.traces.start_span(name="span 2"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span 3"): - pass - - sentry_sdk.flush() - - # The span is captured even though a span was dropped in the same trace. - with sentry_sdk.traces.start_span(name="span 4"): - pass - - sentry_sdk.flush() - - assert len(envelopes) == 2 - - assert len(envelopes[0].items[0].payload.json["items"]) == 2 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" - assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" - - assert len(envelopes[1].items[0].payload.json["items"]) == 1 - assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 4" - - def test_length_based_flushing(sentry_init, capture_items, monkeypatch): """A flush event is triggered when a bucket contains MAX_BEFORE_FLUSH spans.""" monkeypatch.setattr(SpanBatcher, "MAX_BEFORE_FLUSH", 1) @@ -542,8 +460,6 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): original_lock.acquire() batcher._span_buffer["test-trace-id"].append(object()) - batcher._span_number = 1 - batcher._running_size["test-trace-id"] = 42 batcher._active.flag = True batcher._flush_event.set() @@ -556,8 +472,6 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): flusher_reset = batcher._flusher is None and batcher._flusher_pid is None span_buffer_reset = len(batcher._span_buffer) == 0 - span_number_reset = batcher._span_number == 0 - running_size_reset = len(batcher._running_size) == 0 active_reset = not getattr(batcher._active, "flag", False) @@ -570,7 +484,6 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): and unheld and flusher_reset and span_buffer_reset - and span_number_reset and running_size_reset and active_reset and event_reset From 46cf5ce0214dc09433a94d57f8e9779ebde84d25 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 13:43:57 +0200 Subject: [PATCH 04/21] ref: Flush trace bucket when segment span finishes --- sentry_sdk/_span_batcher.py | 5 +++-- sentry_sdk/client.py | 27 ++++++++++++--------------- tests/tracing/test_span_batcher.py | 19 +++++++++++++++++++ 3 files changed, 34 insertions(+), 17 deletions(-) diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index 79285c3386..1a852530cb 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -100,7 +100,7 @@ def _flush_loop(self) -> None: self._flush() self._last_full_flush = time.monotonic() - def add(self, span: "SpanJSON") -> None: + def add(self, span: "SpanJSON", flush_trace_bucket: "bool" = False) -> None: # Bail out if the current thread is already executing batcher code. # This prevents deadlocks when code running inside the batcher (e.g. # _add_to_envelope during flush, or _flush_event.wait/set) triggers @@ -129,7 +129,8 @@ def add(self, span: "SpanJSON") -> None: self._running_size[span["trace_id"]] += self._estimate_size(span) if ( - size + 1 >= self.MAX_BEFORE_FLUSH + flush_trace_bucket + or size + 1 >= self.MAX_BEFORE_FLUSH or self._running_size[span["trace_id"]] >= self.MAX_BYTES_BEFORE_FLUSH ): diff --git a/sentry_sdk/client.py b/sentry_sdk/client.py index e8414d3f91..b852418a1e 100644 --- a/sentry_sdk/client.py +++ b/sentry_sdk/client.py @@ -1270,6 +1270,12 @@ def _capture_telemetry( if serialized is None: return + if ty == "log": + self.log_batcher.add(serialized) # type: ignore + + elif ty == "metric": + self.metrics_batcher.add(serialized) # type: ignore + elif ty == "span" and isinstance(telemetry, StreamedSpan): # Reset the span to its original value before we attempted # to call the `before_send_span` callback @@ -1292,21 +1298,12 @@ def _capture_telemetry( serialized = telemetry._to_json() - batcher = None - if ty == "log": - batcher = self.log_batcher - - elif ty == "metric": - batcher = self.metrics_batcher - - elif ty == "span": - # We need a reference to the segment span in the batcher to populate - # the dynamic sampling context (DSC) - serialized["_segment_span"] = telemetry._segment # type: ignore - batcher = self.span_batcher - - if batcher is not None: - batcher.add(serialized) # type: ignore + # We need a reference to the segment span in the batcher to populate + # the dynamic sampling context (DSC) + serialized["_segment_span"] = telemetry._segment # type: ignore + self.span_batcher.add( + serialized, flush_trace_bucket=telemetry._is_segment() + ) # type: ignore def _capture_log(self, log: "Optional[Log]", scope: "Scope") -> None: self._capture_telemetry(log, "log", scope) diff --git a/tests/tracing/test_span_batcher.py b/tests/tracing/test_span_batcher.py index c21d6cdea4..9a3674fd36 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -432,6 +432,25 @@ def test_transport_format(sentry_init, capture_envelopes): assert value["type"] in ("string", "boolean", "integer", "double", "array") +def test_trace_bucket_flushes_when_segment_ends( + sentry_init, capture_items, monkeypatch +): + """All currently completed spans in a trace are flushed when the segment is finished.""" + monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) + + sentry_init(traces_sample_rate=1.0, trace_lifecycle="stream") + items = capture_items("span") + + with sentry_sdk.traces.start_span(name="segment span"): + with sentry_sdk.traces.start_span(name="child"): + pass + + time.sleep(0.1) + + assert len(items) == 3 + assert items[0].payload["name"] == "span" + + @pytest.mark.skipif( sys.platform == "win32" or not hasattr(os, "fork") From 51a768e97985e15159e4f04a1cf74c58eeded96b Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 13:59:19 +0200 Subject: [PATCH 05/21] make mypy happy --- sentry_sdk/client.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/sentry_sdk/client.py b/sentry_sdk/client.py index b852418a1e..dd7d3ba3be 100644 --- a/sentry_sdk/client.py +++ b/sentry_sdk/client.py @@ -1270,10 +1270,10 @@ def _capture_telemetry( if serialized is None: return - if ty == "log": + if ty == "log" and self.log_batcher is not None: self.log_batcher.add(serialized) # type: ignore - elif ty == "metric": + elif ty == "metric" and self.metrics_batcher is not None: self.metrics_batcher.add(serialized) # type: ignore elif ty == "span" and isinstance(telemetry, StreamedSpan): @@ -1300,10 +1300,14 @@ def _capture_telemetry( # We need a reference to the segment span in the batcher to populate # the dynamic sampling context (DSC) - serialized["_segment_span"] = telemetry._segment # type: ignore + serialized["_segment_span"] = telemetry._segment + + if self.span_batcher is None: + return + self.span_batcher.add( serialized, flush_trace_bucket=telemetry._is_segment() - ) # type: ignore + ) def _capture_log(self, log: "Optional[Log]", scope: "Scope") -> None: self._capture_telemetry(log, "log", scope) From 379b8c4effc32bb4b751f63d3096c1d9ed074952 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 15:20:08 +0200 Subject: [PATCH 06/21] simplify and fix tests --- sentry_sdk/_span_batcher.py | 4 +- sentry_sdk/client.py | 27 ++- tests/tracing/test_span_batcher.py | 272 ++++++++++++++++------------- 3 files changed, 165 insertions(+), 138 deletions(-) diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index 1a852530cb..c343b533eb 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -100,7 +100,7 @@ def _flush_loop(self) -> None: self._flush() self._last_full_flush = time.monotonic() - def add(self, span: "SpanJSON", flush_trace_bucket: "bool" = False) -> None: + def add(self, span: "SpanJSON") -> None: # Bail out if the current thread is already executing batcher code. # This prevents deadlocks when code running inside the batcher (e.g. # _add_to_envelope during flush, or _flush_event.wait/set) triggers @@ -129,7 +129,7 @@ def add(self, span: "SpanJSON", flush_trace_bucket: "bool" = False) -> None: self._running_size[span["trace_id"]] += self._estimate_size(span) if ( - flush_trace_bucket + span["is_segment"] is True or size + 1 >= self.MAX_BEFORE_FLUSH or self._running_size[span["trace_id"]] >= self.MAX_BYTES_BEFORE_FLUSH diff --git a/sentry_sdk/client.py b/sentry_sdk/client.py index dd7d3ba3be..e8414d3f91 100644 --- a/sentry_sdk/client.py +++ b/sentry_sdk/client.py @@ -1270,12 +1270,6 @@ def _capture_telemetry( if serialized is None: return - if ty == "log" and self.log_batcher is not None: - self.log_batcher.add(serialized) # type: ignore - - elif ty == "metric" and self.metrics_batcher is not None: - self.metrics_batcher.add(serialized) # type: ignore - elif ty == "span" and isinstance(telemetry, StreamedSpan): # Reset the span to its original value before we attempted # to call the `before_send_span` callback @@ -1298,16 +1292,21 @@ def _capture_telemetry( serialized = telemetry._to_json() - # We need a reference to the segment span in the batcher to populate - # the dynamic sampling context (DSC) - serialized["_segment_span"] = telemetry._segment + batcher = None + if ty == "log": + batcher = self.log_batcher - if self.span_batcher is None: - return + elif ty == "metric": + batcher = self.metrics_batcher - self.span_batcher.add( - serialized, flush_trace_bucket=telemetry._is_segment() - ) + elif ty == "span": + # We need a reference to the segment span in the batcher to populate + # the dynamic sampling context (DSC) + serialized["_segment_span"] = telemetry._segment # type: ignore + batcher = self.span_batcher + + if batcher is not None: + batcher.add(serialized) # type: ignore def _capture_log(self, log: "Optional[Log]", scope: "Scope") -> None: self._capture_telemetry(log, "log", scope) diff --git a/tests/tracing/test_span_batcher.py b/tests/tracing/test_span_batcher.py index 9a3674fd36..679a2be4a7 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -18,19 +18,27 @@ def test_envelope_by_trace_id(sentry_init, capture_envelopes, monkeypatch): envelopes = capture_envelopes() - sentry_sdk.traces.new_trace() - - with sentry_sdk.traces.start_span(name="span 1a") as span1: - trace_id1 = span1.trace_id - with sentry_sdk.traces.start_span(name="span 1b"): - pass - - sentry_sdk.traces.new_trace() + with sentry_sdk.new_scope(): + sentry_sdk.traces.new_trace() + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_a = sentry_sdk.traces.start_span(name="parent a") + with sentry_sdk.traces.start_span( + name="span 1a", parent_span=parent_a + ) as span1: + trace_id1 = span1.trace_id + with sentry_sdk.traces.start_span(name="span 1b", parent_span=parent_a): + pass - with sentry_sdk.traces.start_span(name="span 2a") as span2: - trace_id2 = span2.trace_id - with sentry_sdk.traces.start_span(name="span 2b"): - pass + with sentry_sdk.new_scope(): + sentry_sdk.traces.new_trace() + parent_b = sentry_sdk.traces.start_span(name="parent b") + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + with sentry_sdk.traces.start_span( + name="span 2a", parent_span=parent_b + ) as span2: + trace_id2 = span2.trace_id + with sentry_sdk.traces.start_span(name="span 2b", parent_span=parent_b): + pass sentry_sdk.flush() @@ -65,29 +73,30 @@ def test_max_envelope_size(sentry_init, capture_envelopes, monkeypatch): envelopes = capture_envelopes() - with sentry_sdk.traces.start_span(name="span 1"): - pass - with sentry_sdk.traces.start_span(name="span 2"): - pass - with sentry_sdk.traces.start_span(name="span 3"): - pass - with sentry_sdk.traces.start_span(name="span 4"): - pass - with sentry_sdk.traces.start_span(name="span 5"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span 1"): + pass + with sentry_sdk.traces.start_span(name="span 2"): + pass + with sentry_sdk.traces.start_span(name="span 3"): + pass + with sentry_sdk.traces.start_span(name="span 4"): + pass + with sentry_sdk.traces.start_span(name="span 5"): + pass - sentry_sdk.flush() + sentry_sdk.flush() - assert len(envelopes) == 3 + assert len(envelopes) == 3 - assert len(envelopes[0].items[0].payload.json["items"]) == 2 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" - assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" - assert len(envelopes[1].items[0].payload.json["items"]) == 2 - assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 3" - assert envelopes[1].items[0].payload.json["items"][1]["name"] == "span 4" - assert len(envelopes[2].items[0].payload.json["items"]) == 1 - assert envelopes[2].items[0].payload.json["items"][0]["name"] == "span 5" + assert len(envelopes[0].items[0].payload.json["items"]) == 2 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" + assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" + assert len(envelopes[1].items[0].payload.json["items"]) == 2 + assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 3" + assert envelopes[1].items[0].payload.json["items"][1]["name"] == "span 4" + assert len(envelopes[2].items[0].payload.json["items"]) == 1 + assert envelopes[2].items[0].payload.json["items"][0]["name"] == "span 5" def test_drop_after_max_reached( @@ -107,22 +116,23 @@ def test_drop_after_max_reached( envelopes = capture_envelopes() record_lost_event_calls = capture_record_lost_event_calls() - with sentry_sdk.traces.start_span(name="span 1"): - pass - with sentry_sdk.traces.start_span(name="span 2"): - pass - with sentry_sdk.traces.start_span(name="span 3"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span 1"): + pass + with sentry_sdk.traces.start_span(name="span 2"): + pass + with sentry_sdk.traces.start_span(name="span 3"): + pass - sentry_sdk.flush() + sentry_sdk.flush() - assert len(envelopes) == 1 + assert len(envelopes) == 1 - assert len(envelopes[0].items[0].payload.json["items"]) == 2 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" - assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" + assert len(envelopes[0].items[0].payload.json["items"]) == 2 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" + assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" - assert ("queue_overflow", "span", None, 1) in record_lost_event_calls + assert ("queue_overflow", "span", None, 1) in record_lost_event_calls def test_drop_isolated_per_bucket( @@ -140,19 +150,24 @@ def test_drop_isolated_per_bucket( envelopes = capture_envelopes() record_lost_event_calls = capture_record_lost_event_calls() - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="a1") as span_a: - trace_id_a = span_a.trace_id - with sentry_sdk.traces.start_span(name="a2"): - pass - with sentry_sdk.traces.start_span(name="a3"): - pass + with sentry_sdk.new_scope(): + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_a = sentry_sdk.traces.start_span(name="parent a") + with sentry_sdk.traces.start_span(name="a1", parent_span=parent_a) as span_a: + trace_id_a = span_a.trace_id + with sentry_sdk.traces.start_span(name="a2", parent_span=parent_a): + pass + with sentry_sdk.traces.start_span(name="a3"): + pass - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="b1") as span_b: - trace_id_b = span_b.trace_id - with sentry_sdk.traces.start_span(name="b2"): - pass + with sentry_sdk.new_scope(): + sentry_sdk.traces.new_trace() + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_b = sentry_sdk.traces.start_span(name="parent b") + with sentry_sdk.traces.start_span(name="b1", parent_span=parent_b) as span_b: + trace_id_b = span_b.trace_id + with sentry_sdk.traces.start_span(name="b2", parent_span=parent_b): + pass sentry_sdk.flush() @@ -185,13 +200,14 @@ def test_length_based_flushing(sentry_init, capture_items, monkeypatch): items = capture_items("span") - with sentry_sdk.traces.start_span(name="span"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span"): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(items) == 1 - assert items[0].payload["name"] == "span" + assert len(items) == 1 + assert items[0].payload["name"] == "span" def test_weight_based_flushing(sentry_init, capture_envelopes, monkeypatch): @@ -208,15 +224,16 @@ def test_weight_based_flushing(sentry_init, capture_envelopes, monkeypatch): envelopes = capture_envelopes() - with sentry_sdk.traces.start_span(name="span"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span"): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(envelopes) == 1 + assert len(envelopes) == 1 - assert len(envelopes[0].items[0].payload.json["items"]) == 1 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span" + assert len(envelopes[0].items[0].payload.json["items"]) == 1 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span" def test_weight_based_flushing_by_attribute_size( @@ -233,27 +250,30 @@ def test_weight_based_flushing_by_attribute_size( envelopes = capture_envelopes() - with sentry_sdk.traces.start_span(name="small span") as bare_span: - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="small span") as bare_span: + pass - bare_span_size = SpanBatcher._estimate_size(bare_span._to_json()) - big_attr = "x" * bare_span_size + bare_span_size = SpanBatcher._estimate_size(bare_span._to_json()) + big_attr = "x" * bare_span_size - monkeypatch.setattr(SpanBatcher, "MAX_BYTES_BEFORE_FLUSH", bare_span_size * 3) + monkeypatch.setattr(SpanBatcher, "MAX_BYTES_BEFORE_FLUSH", bare_span_size * 3) - time.sleep(0.1) + time.sleep(0.1) - # The first span alone is well under the byte limit, so no flush yet. - assert len(envelopes) == 0 + # The first span alone is well under the byte limit, so no flush yet. + assert len(envelopes) == 0 - with sentry_sdk.traces.start_span(name="big span", attributes={"big": big_attr}): - pass + with sentry_sdk.traces.start_span( + name="big span", attributes={"big": big_attr} + ): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(envelopes) == 1 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "small span" - assert envelopes[0].items[0].payload.json["items"][1]["name"] == "big span" + assert len(envelopes) == 1 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "small span" + assert envelopes[0].items[0].payload.json["items"][1]["name"] == "big span" def test_bucket_recreated_after_flush(sentry_init, capture_envelopes, monkeypatch): @@ -270,37 +290,38 @@ def test_bucket_recreated_after_flush(sentry_init, capture_envelopes, monkeypatc sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span 1") as span1: - trace_id = span1.trace_id - with sentry_sdk.traces.start_span(name="span 2"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span 1") as span1: + trace_id = span1.trace_id + with sentry_sdk.traces.start_span(name="span 2"): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(envelopes) == 1 + assert len(envelopes) == 1 - with sentry_sdk.traces.start_span(name="span 3"): - pass - with sentry_sdk.traces.start_span(name="span 4"): - pass + with sentry_sdk.traces.start_span(name="span 3"): + pass + with sentry_sdk.traces.start_span(name="span 4"): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(envelopes) == 2 + assert len(envelopes) == 2 - assert envelopes[0].headers["trace"]["trace_id"] == trace_id - assert len(envelopes[0].items[0].payload.json["items"]) == 2 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" - assert envelopes[0].items[0].payload.json["items"][0]["trace_id"] == trace_id - assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" - assert envelopes[0].items[0].payload.json["items"][1]["trace_id"] == trace_id + assert envelopes[0].headers["trace"]["trace_id"] == trace_id + assert len(envelopes[0].items[0].payload.json["items"]) == 2 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" + assert envelopes[0].items[0].payload.json["items"][0]["trace_id"] == trace_id + assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" + assert envelopes[0].items[0].payload.json["items"][1]["trace_id"] == trace_id - assert envelopes[1].headers["trace"]["trace_id"] == trace_id - assert len(envelopes[1].items[0].payload.json["items"]) == 2 - assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 3" - assert envelopes[1].items[0].payload.json["items"][0]["trace_id"] == trace_id - assert envelopes[1].items[0].payload.json["items"][1]["name"] == "span 4" - assert envelopes[1].items[0].payload.json["items"][1]["trace_id"] == trace_id + assert envelopes[1].headers["trace"]["trace_id"] == trace_id + assert len(envelopes[1].items[0].payload.json["items"]) == 2 + assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 3" + assert envelopes[1].items[0].payload.json["items"][0]["trace_id"] == trace_id + assert envelopes[1].items[0].payload.json["items"][1]["name"] == "span 4" + assert envelopes[1].items[0].payload.json["items"][1]["trace_id"] == trace_id def test_quiet_buckets_flush_eventually(sentry_init, capture_envelopes, monkeypatch): @@ -317,15 +338,16 @@ def test_quiet_buckets_flush_eventually(sentry_init, capture_envelopes, monkeypa envelopes = capture_envelopes() - with sentry_sdk.traces.start_span(name="span 1"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span 1"): + pass - time.sleep(0.3) + time.sleep(0.3) - assert len(envelopes) == 1 + assert len(envelopes) == 1 - assert len(envelopes[0].items[0].payload.json["items"]) == 1 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" + assert len(envelopes[0].items[0].payload.json["items"]) == 1 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" def test_quiet_buckets_flushed_with_busy_neighbors( @@ -344,17 +366,22 @@ def test_quiet_buckets_flushed_with_busy_neighbors( sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span 1") as span1: - trace_id1 = span1.trace_id + with sentry_sdk.new_scope(): + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_a = sentry_sdk.traces.start_span(name="parent a") + with sentry_sdk.traces.start_span(name="span 1", parent_span=parent_a) as span1: + trace_id1 = span1.trace_id - sentry_sdk.traces.new_trace() - - with sentry_sdk.traces.start_span(name="span 2") as span2: - trace_id2 = span2.trace_id + with sentry_sdk.new_scope(): + sentry_sdk.traces.new_trace() + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_b = sentry_sdk.traces.start_span(name="parent b") + with sentry_sdk.traces.start_span(name="span 2", parent_span=parent_b) as span2: + trace_id2 = span2.trace_id - for i in range(3, 10): - with sentry_sdk.traces.start_span(name=f"span {i}"): - pass + for i in range(3, 10): + with sentry_sdk.traces.start_span(name=f"span {i}", parent_span=parent_b): + pass time.sleep(0.3) @@ -447,8 +474,9 @@ def test_trace_bucket_flushes_when_segment_ends( time.sleep(0.1) - assert len(items) == 3 - assert items[0].payload["name"] == "span" + assert len(items) == 2 + assert items[0].payload["name"] == "child" + assert items[1].payload["name"] == "segment span" @pytest.mark.skipif( From 9241cee48aa069eb691ab8d00098e1d2752e0078 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 16:18:42 +0200 Subject: [PATCH 07/21] . --- sentry_sdk/_span_batcher.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index c343b533eb..148a3806e3 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -129,8 +129,7 @@ def add(self, span: "SpanJSON") -> None: self._running_size[span["trace_id"]] += self._estimate_size(span) if ( - span["is_segment"] is True - or size + 1 >= self.MAX_BEFORE_FLUSH + size + 1 >= self.MAX_BEFORE_FLUSH or self._running_size[span["trace_id"]] >= self.MAX_BYTES_BEFORE_FLUSH ): @@ -139,7 +138,9 @@ def add(self, span: "SpanJSON") -> None: else: notify = False - if notify: + if span["is_segment"] is True: + self._flush(only_pending=True) + elif notify: self._flush_event.set() finally: self._active.flag = False From b3da36f149264fa35ae33a4177a86068ebf768a6 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 16:39:51 +0200 Subject: [PATCH 08/21] add span to pending bucket --- sentry_sdk/_span_batcher.py | 1 + 1 file changed, 1 insertion(+) diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index 148a3806e3..4200d73c48 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -139,6 +139,7 @@ def add(self, span: "SpanJSON") -> None: notify = False if span["is_segment"] is True: + self._pending_flush.add(span["trace_id"]) self._flush(only_pending=True) elif notify: self._flush_event.set() From 76525790a18035dc1c231a30a535feee02c5b2ee Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 16:54:54 +0200 Subject: [PATCH 09/21] modify pending flush inside lock --- sentry_sdk/_span_batcher.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index 4200d73c48..3c2b54b2e2 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -129,7 +129,8 @@ def add(self, span: "SpanJSON") -> None: self._running_size[span["trace_id"]] += self._estimate_size(span) if ( - size + 1 >= self.MAX_BEFORE_FLUSH + span["is_segment"] is True + or size + 1 >= self.MAX_BEFORE_FLUSH or self._running_size[span["trace_id"]] >= self.MAX_BYTES_BEFORE_FLUSH ): @@ -139,7 +140,6 @@ def add(self, span: "SpanJSON") -> None: notify = False if span["is_segment"] is True: - self._pending_flush.add(span["trace_id"]) self._flush(only_pending=True) elif notify: self._flush_event.set() From 78b2893fba8b3cba1a6c0bc6d446181c5da7bd21 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 11 Aug 2026 17:34:00 +0200 Subject: [PATCH 10/21] flush async --- sentry_sdk/_span_batcher.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index 3c2b54b2e2..c343b533eb 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -139,9 +139,7 @@ def add(self, span: "SpanJSON") -> None: else: notify = False - if span["is_segment"] is True: - self._flush(only_pending=True) - elif notify: + if notify: self._flush_event.set() finally: self._active.flag = False From 0478e7188ea3350f6038a18067d7b90b274f1b60 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 08:39:31 +0200 Subject: [PATCH 11/21] fix race on 3.14t --- tests/conftest.py | 37 +++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index 599c075224..1acf8af1b2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,6 +4,7 @@ import json import os import socket +import threading import warnings from collections import namedtuple from contextlib import contextmanager @@ -293,11 +294,47 @@ def inner(identifier): return inner +def _install_flush_completion_handshake(client: "sentry_sdk.Client") -> None: + """Make batcher.flush() wait for the flusher thread to drain. + + Otherwise, test assertions can be run before envelopes are captured + despite `sentry_sdk.flush()`. The span batcher flushes pending items + asynchronously with the main thread. + """ + batcher = client.span_batcher + if batcher is None: + return + + orig_flush_raw = batcher._flush + orig_flush = batcher.flush + done = threading.Event() + + def _flush(*args: "Any", **kwargs: "Any") -> "Any": + try: + return orig_flush_raw(*args, **kwargs) + finally: + done.set() + + def flush() -> None: + # If a segment or threshold already woke the background flusher and + # it drained, `done` is set and we can skip waiting. Otherwise, + # poke the flusher so a drain is guaranteed to happen (otherwise + # done.wait() could hang for tests that rely on flush() itself). + if not done.is_set(): + batcher._flush_event.set() + done.wait() + orig_flush() + + object.__setattr__(batcher, "_flush", _flush) + object.__setattr__(batcher, "flush", flush) + + @pytest.fixture def sentry_init(request): def inner(*a, **kw): kw.setdefault("transport", TestTransport()) client = sentry_sdk.Client(*a, **kw) + _install_flush_completion_handshake(client) sentry_sdk.get_global_scope().set_client(client) if request.node.get_closest_marker("forked"): From 602a4129e51768161b490ce17e7d1cfa7433a2b2 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 08:42:56 +0200 Subject: [PATCH 12/21] clean up comments --- tests/conftest.py | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 1acf8af1b2..bccf6c8ad8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -297,9 +297,8 @@ def inner(identifier): def _install_flush_completion_handshake(client: "sentry_sdk.Client") -> None: """Make batcher.flush() wait for the flusher thread to drain. - Otherwise, test assertions can be run before envelopes are captured - despite `sentry_sdk.flush()`. The span batcher flushes pending items - asynchronously with the main thread. + Otherwise, test assertions can be run before envelopes are captured. + The span batcher flushes pending items asynchronously with the main thread. """ batcher = client.span_batcher if batcher is None: @@ -316,12 +315,7 @@ def _flush(*args: "Any", **kwargs: "Any") -> "Any": done.set() def flush() -> None: - # If a segment or threshold already woke the background flusher and - # it drained, `done` is set and we can skip waiting. Otherwise, - # poke the flusher so a drain is guaranteed to happen (otherwise - # done.wait() could hang for tests that rely on flush() itself). if not done.is_set(): - batcher._flush_event.set() done.wait() orig_flush() From bd5a977d0d5976d43e11e1dd42418db57905eb54 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 08:54:14 +0200 Subject: [PATCH 13/21] conftest update --- tests/conftest.py | 24 ++++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index bccf6c8ad8..d71f88553b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -299,6 +299,8 @@ def _install_flush_completion_handshake(client: "sentry_sdk.Client") -> None: Otherwise, test assertions can be run before envelopes are captured. The span batcher flushes pending items asynchronously with the main thread. + Flushes triggered by segments finishing are asynchronous, and can collect buckets + that would have otherwise been flushed synchronously by `sentry_sdk.flush()`. """ batcher = client.span_batcher if batcher is None: @@ -306,17 +308,31 @@ def _install_flush_completion_handshake(client: "sentry_sdk.Client") -> None: orig_flush_raw = batcher._flush orig_flush = batcher.flush - done = threading.Event() + lock = threading.Lock() + drained_count = 0 + wake = threading.Event() def _flush(*args: "Any", **kwargs: "Any") -> "Any": + nonlocal drained_count try: return orig_flush_raw(*args, **kwargs) finally: - done.set() + with lock: + drained_count += 1 + wake.set() def flush() -> None: - if not done.is_set(): - done.wait() + nonlocal drained_count + with lock: + target = drained_count + + batcher._flush_event.set() + while True: + with lock: + if drained_count > target: + break + wake.wait() + wake.clear() orig_flush() object.__setattr__(batcher, "_flush", _flush) From 02c0a678680245618e4c9f7f881fc5071fceac1c Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 09:17:31 +0200 Subject: [PATCH 14/21] work on fixture changes --- tests/conftest.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index d71f88553b..3d78cc1e95 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -323,6 +323,13 @@ def _flush(*args: "Any", **kwargs: "Any") -> "Any": def flush() -> None: nonlocal drained_count + # If the background flusher thread was never started (no spans have + # been added), there is no thread to drain and the counter will never + # advance. Fall back to the original synchronous flush. + if batcher._flusher is None or not batcher._flusher.is_alive(): + orig_flush() + return + with lock: target = drained_count From 8057cac2b0e9ca11f17e27922142f4191bd949fd Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 10:29:14 +0200 Subject: [PATCH 15/21] add re-entrancy guard --- tests/conftest.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index 3d78cc1e95..8019977178 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -323,6 +323,13 @@ def _flush(*args: "Any", **kwargs: "Any") -> "Any": def flush() -> None: nonlocal drained_count + # Re-entrancy guard: if this `flush()`` is invoked from within an + # in-progress drain (e.g. a custom transport), waiting on the background flusher + # would deadlock, because the flusher is blocked inside our own handler. + if getattr(getattr(batcher, "_active", None), "flag", False): + orig_flush() + return + # If the background flusher thread was never started (no spans have # been added), there is no thread to drain and the counter will never # advance. Fall back to the original synchronous flush. From 5653f6c0919b76c562b2e88027dff95caaad76ad Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 10:34:05 +0200 Subject: [PATCH 16/21] edit comment --- tests/conftest.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 8019977178..58a96aecbe 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -323,9 +323,9 @@ def _flush(*args: "Any", **kwargs: "Any") -> "Any": def flush() -> None: nonlocal drained_count - # Re-entrancy guard: if this `flush()`` is invoked from within an - # in-progress drain (e.g. a custom transport), waiting on the background flusher - # would deadlock, because the flusher is blocked inside our own handler. + # Re-entrancy guard: if `flush()` is invoked from within a drain (e.g. a + # custom transport), waiting on the flusher thread would deadlock, because + # the flusher is blocked inside our own handler. if getattr(getattr(batcher, "_active", None), "flag", False): orig_flush() return From 5ee9fb6325790f1c3bf9ded94a068e43d06ec415 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 12:49:37 +0200 Subject: [PATCH 17/21] resolve django test failures --- tests/integrations/django/test_basic.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/integrations/django/test_basic.py b/tests/integrations/django/test_basic.py index 93b5477010..d026eaa621 100644 --- a/tests/integrations/django/test_basic.py +++ b/tests/integrations/django/test_basic.py @@ -2454,8 +2454,14 @@ def test_transaction_http_method_custom( sentry_sdk.flush() spans = [item.payload for item in items] - assert spans[2]["attributes"][SPANDATA.HTTP_REQUEST_METHOD] == "OPTIONS" - assert spans[5]["attributes"][SPANDATA.HTTP_REQUEST_METHOD] == "HEAD" + http_methods = [ + span["attributes"][SPANDATA.HTTP_REQUEST_METHOD] + for span in spans + if span.get("parent_span_id") is None + ] + + assert "OPTIONS" in http_methods + assert "HEAD" in http_methods else: events = capture_events() From 40dfb01b5db345c8638f7f7496f1c33e7fee8bdd Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 13:15:21 +0200 Subject: [PATCH 18/21] second iteration django tests --- tests/integrations/django/test_basic.py | 15 +++++++-------- tests/integrations/django/test_cache_module.py | 6 +++++- 2 files changed, 12 insertions(+), 9 deletions(-) diff --git a/tests/integrations/django/test_basic.py b/tests/integrations/django/test_basic.py index d026eaa621..1430236aa1 100644 --- a/tests/integrations/django/test_basic.py +++ b/tests/integrations/django/test_basic.py @@ -2449,19 +2449,18 @@ def test_transaction_http_method_custom( client.get("/nomessage") client.options("/nomessage") - client.head("/nomessage") sentry_sdk.flush() spans = [item.payload for item in items] - http_methods = [ - span["attributes"][SPANDATA.HTTP_REQUEST_METHOD] - for span in spans - if span.get("parent_span_id") is None - ] + assert spans[2]["attributes"][SPANDATA.HTTP_REQUEST_METHOD] == "OPTIONS" + + client.head("/nomessage") + + sentry_sdk.flush() + spans = [item.payload for item in items] - assert "OPTIONS" in http_methods - assert "HEAD" in http_methods + assert spans[5]["attributes"][SPANDATA.HTTP_REQUEST_METHOD] == "HEAD" else: events = capture_events() diff --git a/tests/integrations/django/test_cache_module.py b/tests/integrations/django/test_cache_module.py index b00903104d..700c5e3a80 100644 --- a/tests/integrations/django/test_cache_module.py +++ b/tests/integrations/django/test_cache_module.py @@ -246,7 +246,6 @@ def test_cache_spans_middleware( if span_streaming: items = capture_items("span") - client.get(reverse("not_cached_view")) client.get(reverse("not_cached_view")) sentry_sdk.flush() @@ -269,6 +268,11 @@ def test_cache_spans_middleware( ) assert "cache.hit" not in spans[1]["attributes"] assert spans[1]["attributes"]["cache.item_size"] == 2 + + client.get(reverse("not_cached_view")) + + sentry_sdk.flush() + spans = [item.payload for item in items] # second_event - cache.get assert spans[4]["attributes"]["sentry.op"] == "cache.get" assert spans[4]["name"].startswith("views.decorators.cache.cache_header.") From c29c865b5ee0e9de276922797043a24303f6af56 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 13:24:11 +0200 Subject: [PATCH 19/21] resolve flask test failures --- tests/integrations/flask/test_flask.py | 25 ++++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/tests/integrations/flask/test_flask.py b/tests/integrations/flask/test_flask.py index cadf011b4c..1252673b34 100644 --- a/tests/integrations/flask/test_flask.py +++ b/tests/integrations/flask/test_flask.py @@ -1211,20 +1211,31 @@ def test_transaction_or_segment_http_method_custom( response = client.options("/nomessage") assert response.status_code == 200 - response = client.head("/nomessage") - assert response.status_code == 200 - if span_streaming: sentry_sdk.flush() spans = [i.payload for i in items] - assert len(spans) == 2 - (options_segment, head_segment) = spans + (options_segment,) = spans assert options_segment["attributes"]["http.request.method"] == "OPTIONS" + + response = client.head("/nomessage") + assert response.status_code == 200 + + sentry_sdk.flush() + spans = [i.payload for i in items] + assert len(spans) == 2 + (_, head_segment) = spans + assert head_segment["attributes"]["http.request.method"] == "HEAD" else: - assert len(events) == 2 - (event1, event2) = events + (event1,) = events assert event1["request"]["method"] == "OPTIONS" + + response = client.head("/nomessage") + assert response.status_code == 200 + + assert len(events) == 2 + (_, event2) = events + assert event2["request"]["method"] == "HEAD" From 14d342dce7e7da8deb9f3a6ec85b07119daeef58 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 13:32:45 +0200 Subject: [PATCH 20/21] resolve common test failures --- tests/tracing/test_span_streaming.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/tracing/test_span_streaming.py b/tests/tracing/test_span_streaming.py index 3800a4badc..821398eb6b 100644 --- a/tests/tracing/test_span_streaming.py +++ b/tests/tracing/test_span_streaming.py @@ -719,14 +719,13 @@ def test_sibling_segments_new_trace(sentry_init, capture_items): spans = [item.payload for item in items] assert len(spans) == 2 - segment1, segment2 = spans - assert segment1["name"] == "segment1" + (segment1,) = (span for span in spans if span["name"] == "segment1") assert segment1["attributes"]["sentry.segment.name"] == "segment1" assert segment1["is_segment"] is True assert "parent_span_id" not in segment1 - assert segment2["name"] == "segment2" + (segment2,) = (span for span in spans if span["name"] == "segment2") assert segment2["attributes"]["sentry.segment.name"] == "segment2" assert segment2["is_segment"] is True assert "parent_span_id" not in segment2 From 54334a33205b1f197b18ef487b1fc612ef6a16ad Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 12 Aug 2026 13:42:25 +0200 Subject: [PATCH 21/21] clean up comment --- tests/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 58a96aecbe..312c992e6b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -297,7 +297,7 @@ def inner(identifier): def _install_flush_completion_handshake(client: "sentry_sdk.Client") -> None: """Make batcher.flush() wait for the flusher thread to drain. - Otherwise, test assertions can be run before envelopes are captured. + Otherwise, test assertions can run before envelopes are captured. The span batcher flushes pending items asynchronously with the main thread. Flushes triggered by segments finishing are asynchronous, and can collect buckets that would have otherwise been flushed synchronously by `sentry_sdk.flush()`.