Stop re-formatting the same stream exception on every read-loop iteration

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## [Unreleased]
+
+**Fixed bugs:**
+
+- Log the full traceback of an unexpected read error only on its first occurrence: asyncio streams re-raise the same stored exception object on every read of a failed stream, growing its traceback on each raise, and re-formatting it every loop iteration could starve the event loop (CPU pinned at 100%). Repeats are now summarised in a rate-limited single line and the listen loop yields for a second before retrying.
+
+
 ## [0.4.5](https://github.com/sdb9696/firebase-messaging/tree/0.4.5) (2025-05-10)
 
 [Full Changelog](https://github.com/sdb9696/firebase-messaging/compare/0.4.4...0.4.5)

firebase_messaging/fcmpushclient.py

@@ -171,6 +171,7 @@ def __init__(
         self.do_listen = False
         self.sequential_error_counters: dict[ErrorType, int] = {}
         self.log_warn_counters: dict[str, int] = {}
+        self._last_listen_exception: BaseException | None = None
 
         # reset variables
         self.input_stream_id = 0
@@ -751,7 +752,27 @@ async def _listen(self) -> None:
                                 "Expected read error during reset: %s",
                                 type(osex).__name__,
                             )
+                    elif osex is self._last_listen_exception:
+                        # asyncio streams store a failed reader's exception
+                        # and re-raise the SAME object on every subsequent
+                        # read, growing its traceback by a few frames per
+                        # raise. Re-formatting it on every iteration is
+                        # quadratically expensive — enough to starve the
+                        # event loop (observed pegging a core for >10 min
+                        # on Python 3.14, where traceback formatting also
+                        # runs ast.parse per frame for caret anchors). Log
+                        # a single line instead and yield so the loop can
+                        # never spin hot on a poisoned reader.
+                        self._log_warn_with_limit(
+                            "Repeated unexpected read error: %s: %s",
+                            type(osex).__name__,
+                            osex,
+                        )
+                        await asyncio.sleep(1)
+                        if self._try_increment_error_count(ErrorType.CONNECTION):
+                            await self._reset()
                     else:
+                        self._last_listen_exception = osex
                         _logger.exception("Unexpected exception during read\n")
                         if self._try_increment_error_count(ErrorType.CONNECTION):
                             await self._reset()

tests/fakes.py

@@ -43,8 +43,17 @@ class FakeReader:
         def __init__(self):
             self.queue = asyncio.Queue()
             self.lock = asyncio.Lock()
+            self.sticky_exception = None
+
+        def set_exception(self, error):
+            # Mirrors asyncio.StreamReader behaviour: once the stream has
+            # failed, every subsequent read re-raises the SAME exception
+            # object (asyncio.streams.StreamReader._exception).
+            self.sticky_exception = error
 
         async def readexactly(self, size):
+            if self.sticky_exception is not None:
+                raise self.sticky_exception
             if size == 0:
                 return b""
             val = await self.queue.get()

tests/test_fcmpushclient.py

@@ -251,3 +251,27 @@ def set_app_data_by_key(msg, key, value):
     )
 
     assert raw_data_decrypted == raw_data
+
+
+async def test_repeated_read_exception_logged_once_without_spinning(
+    logged_in_push_client, fake_mcs_endpoint, mocker, caplog
+):
+    await logged_in_push_client(None, None, reset_interval=0.1)
+
+    # asyncio.StreamReader re-raises the SAME stored exception object on
+    # every read of a failed stream; reproduce that with the sticky
+    # exception so each listen iteration sees the identical object.
+    error = ConnectionResetError("Connection lost")
+    fake_mcs_endpoint.client_reader.set_exception(error)
+    await fake_mcs_endpoint.put_error(error)
+
+    await asyncio.sleep(0.5)
+
+    # The full traceback must be logged exactly once; repeats of the same
+    # exception object are summarised in a single rate-limited line so the
+    # (ever-growing) traceback is never re-formatted inside the loop.
+    exc_records = [r for r in caplog.records if r.exc_info]
+    assert len(exc_records) == 1
+    assert any(
+        "Repeated unexpected read error" in r.getMessage() for r in caplog.records
+    )