fix(batcher): Do not let a failed flush kill the flusher thread (#7138)

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]