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.

Author: zkasuran

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.

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.

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]

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]