From f6ab797ec24375e68769edef0ded0f3c4c12784e Mon Sep 17 00:00:00 2001 From: zkasuran Date: Wed, 12 Aug 2026 14:05:11 +0530 Subject: [PATCH] fix(batcher): Do not let a failed flush kill the flusher thread (#7138) An unhandled exception inside `_flush_loop` terminated the batcher's daemon flusher thread. After that the buffer kept filling with nothing draining it. Every later log, metric or span was then dropped for the rest of the process lifetime once the queue hit its cap. Wrap the flush call in each loop (`Batcher._flush_loop` and `SpanBatcher._flush_loop`) in `capture_internal_exceptions()`, the SDK's own helper for errors that should be logged rather than propagated. A single bad batch is now swallowed and logged so the loop keeps running. Adds a regression test for each loop that drives one iteration where the flush raises then asserts the loop returns instead of propagating. --- sentry_sdk/_batcher.py | 10 +++++++-- sentry_sdk/_span_batcher.py | 25 ++++++++++++++-------- tests/test_logs.py | 33 ++++++++++++++++++++++++++++++ tests/tracing/test_span_batcher.py | 31 ++++++++++++++++++++++++++++ 4 files changed, 89 insertions(+), 10 deletions(-) diff --git a/sentry_sdk/_batcher.py b/sentry_sdk/_batcher.py index 565fac2a2d..717129e16c 100644 --- a/sentry_sdk/_batcher.py +++ b/sentry_sdk/_batcher.py @@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, Generic, TypeVar from sentry_sdk.envelope import Envelope, Item, PayloadRef -from sentry_sdk.utils import format_timestamp +from sentry_sdk.utils import capture_internal_exceptions, format_timestamp if TYPE_CHECKING: from typing import Any, Callable, Optional @@ -100,7 +100,13 @@ def _flush_loop(self) -> None: while self._running: self._flush_event.wait(self.FLUSH_WAIT_TIME + random.random()) self._flush_event.clear() - self._flush() + # A failure to serialize or send one batch must not kill the + # flusher thread. If it did, the buffer would keep filling with + # nothing draining it, and every later log or metric would be + # dropped for the rest of the process lifetime. Swallow and log + # the error instead so the loop keeps running. + with capture_internal_exceptions(): + self._flush() def add(self, item: "T") -> None: # Bail out if the current thread is already executing batcher code. diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index c343b533eb..ea67ff6e01 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -9,7 +9,11 @@ from sentry_sdk._batcher import Batcher from sentry_sdk.envelope import Envelope, Item, PayloadRef -from sentry_sdk.utils import format_timestamp, serialize_attribute +from sentry_sdk.utils import ( + capture_internal_exceptions, + format_timestamp, + serialize_attribute, +) if TYPE_CHECKING: from typing import Any, Callable, Optional @@ -91,14 +95,19 @@ def _flush_loop(self) -> None: self._flush_event.wait(timeout=self.FLUSH_WAIT_TIME + jitter) self._flush_event.clear() - self._flush(only_pending=True) + # A failure in one flush must not kill the flusher thread, or the + # span buffer would keep filling with nothing draining it and every + # later span would be dropped for the rest of the process lifetime. + # Swallow and log the error instead so the loop keeps running. + with capture_internal_exceptions(): + self._flush(only_pending=True) - if ( - time.monotonic() - self._last_full_flush - >= self.FLUSH_WAIT_TIME + jitter - ): - self._flush() - self._last_full_flush = time.monotonic() + if ( + time.monotonic() - self._last_full_flush + >= self.FLUSH_WAIT_TIME + jitter + ): + self._flush() + self._last_full_flush = time.monotonic() def add(self, span: "SpanJSON") -> None: # Bail out if the current thread is already executing batcher code. diff --git a/tests/test_logs.py b/tests/test_logs.py index 9147e594f8..e8a098ff44 100644 --- a/tests/test_logs.py +++ b/tests/test_logs.py @@ -922,3 +922,36 @@ def test_log_batcher_lock_reset_in_child_after_fork(sentry_init): original_lock.release() _, status = os.waitpid(pid, 0) assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0 + + +@pytest.mark.tests_internal_exceptions +def test_flush_loop_swallows_flush_exception(): + """The flush loop must not let one failed flush kill the flusher thread. + + Regression test for #7138: an unhandled exception inside _flush_loop + terminated the daemon flusher thread. After that logs silently stopped + being delivered and eventually got dropped at the queue cap. The loop must + swallow the error and keep running. + + Driven synchronously on a bare batcher: _flush raises once and then stops + the loop, so _flush_loop returns cleanly on fixed code and propagates the + exception on unfixed code. + """ + from sentry_sdk._batcher import Batcher + + calls = [] + + class ExplodingBatcher(Batcher): + def _flush(self): + calls.append(1) + self._running = False # exit the loop after this one iteration + raise RuntimeError("boom in flush") + + batcher = ExplodingBatcher( + capture_func=lambda envelope: None, + record_lost_func=lambda *a, **k: None, + ) + batcher._flush_event.set() # so the loop's wait() returns at once + batcher._flush_loop() + + assert calls == [1] diff --git a/tests/tracing/test_span_batcher.py b/tests/tracing/test_span_batcher.py index 679a2be4a7..8eae824ce0 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -541,3 +541,34 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): original_lock.release() _, status = os.waitpid(pid, 0) assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0 + + +@pytest.mark.tests_internal_exceptions +def test_flush_loop_swallows_flush_exception(): + """The flush loop must not let one failed flush kill the flusher thread. + + Regression test for #7138: an unhandled exception inside _flush_loop + terminated the daemon flusher thread. After that the span buffer filled up + with nothing draining it, and every later span was dropped for the rest of + the process lifetime. The loop must swallow the error and keep running. + + Driven synchronously on a bare batcher: _flush raises once and then stops + the loop, so _flush_loop returns cleanly on fixed code and propagates the + exception on unfixed code. + """ + calls = [] + + class ExplodingSpanBatcher(SpanBatcher): + def _flush(self, only_pending=False): + calls.append(1) + self._running = False # exit the loop after this one iteration + raise RuntimeError("boom in flush") + + batcher = ExplodingSpanBatcher( + capture_func=lambda envelope: None, + record_lost_func=lambda *a, **k: None, + ) + batcher._flush_event.set() # so the loop's wait() returns at once + batcher._flush_loop() + + assert calls == [1]