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]