FIX: Fix reconnect not spawning a new monitor

Author: nmacholl

CHANGELOG.md

@@ -6,6 +6,10 @@
 - Added `CorporateActionsHttpAPI.list_events()` and `list_enums()` for fetching
   documentation on supported corporate action event types and enum values
 
+#### Bug fixes
+- Fixed an issue where the `Live` client's monitor task would not get recreated
+for the new session after a reconnection
+
 ## 0.82.0 - 2026-07-21
 
 #### Enhancements

databento/live/session.py

@@ -519,7 +519,7 @@ def start(self) -> None:
                 raise ValueError("session is not connected")
             self._loop.call_soon_threadsafe(self._protocol.start)
             self._heartbeat_monitor_task = self._loop.create_task(
-                self._heartbeat_monitor(),
+                self._heartbeat_monitor(self._protocol),
             )
 
     def subscribe(
@@ -719,20 +719,22 @@ async def _connect_task(
 
         return transport, protocol
 
-    async def _heartbeat_monitor(self) -> None:
-        while not self._protocol.disconnected.done():
+    async def _heartbeat_monitor(self, protocol: _SessionProtocol) -> None:
+        while not protocol.disconnected.done():
             await asyncio.sleep(1)
-            gap = self._loop.time() - self._protocol._last_msg_loop_time
+            gap = self._loop.time() - protocol._last_msg_loop_time
             if gap > (self._heartbeat_interval_s + CLIENT_TIMEOUT_MARGIN_SECONDS):
                 logger.error(
                     "disconnecting client due to timeout, no data received for %d second(s)",
                     int(gap),
                 )
-                self._protocol.disconnected.set_exception(
-                    BentoError(
-                        f"Gateway timeout: {gap:.0f} second(s) since last message",
-                    ),
-                )
+                if not protocol.disconnected.done():
+                    protocol.disconnected.set_exception(
+                        BentoError(
+                            f"Gateway timeout: {gap:.0f} second(s) since last message",
+                        ),
+                    )
+                return
 
     async def _reconnect(self) -> None:
         while True:
@@ -770,6 +772,11 @@ async def _reconnect(self) -> None:
                         )
 
                     if should_restart:
+                        if self._heartbeat_monitor_task is not None:
+                            self._heartbeat_monitor_task.cancel()
+                        self._heartbeat_monitor_task = self._loop.create_task(
+                            self._heartbeat_monitor(self._protocol),
+                        )
                         self._protocol.start()
                         metadata = await self._protocol._metadata_received
                         gap_end = pd.Timestamp(metadata.start, tz="UTC")

tests/test_live_client_reconnect.py

@@ -13,9 +13,11 @@
 from databento import SType
 from databento.common.enums import ReconnectPolicy
 from databento.live import client
+from databento.live import session
 from databento.live.gateway import AuthenticationRequest
 from databento.live.gateway import SessionStart
 from databento.live.gateway import SubscriptionRequest
+from databento.live.protocol import DatabentoLiveProtocol
 from tests.mockliveserver.fixture import MockLiveServerInterface
 
 
@@ -241,3 +243,54 @@ async def test_reconnect_callback(
     gap_start, gap_end = args
     assert isinstance(gap_start, pd.Timestamp)
     assert isinstance(gap_end, pd.Timestamp)
+
+
+async def test_reconnect_after_heartbeat_timeout(
+    test_live_api_key: str,
+    mock_live_server: MockLiveServerInterface,
+    monkeypatch: pytest.MonkeyPatch,
+    reconnect_policy: ReconnectPolicy = ReconnectPolicy.RECONNECT,
+) -> None:
+    """
+    Test that the heartbeat monitor reconnects the client every time the
+    gateway times out, not just the first time.
+    """
+    # Arrange
+    # this mock will make the connection go silent instead of disconnecting
+    monkeypatch.setattr(DatabentoLiveProtocol, "eof_received", lambda _: True)
+    monkeypatch.setattr(session, "CLIENT_TIMEOUT_MARGIN_SECONDS", 0)
+
+    live_client = client.Live(
+        key=test_live_api_key,
+        gateway=mock_live_server.host,
+        port=mock_live_server.port,
+        heartbeat_interval_s=1,
+        reconnect_policy=reconnect_policy,
+    )
+
+    live_client.subscribe(
+        dataset=Dataset.GLBX_MDP3,
+        schema=Schema.MBO,
+        stype_in=SType.RAW_SYMBOL,
+        symbols="TEST",
+    )
+
+    await mock_live_server.wait_for_message_of_type(AuthenticationRequest)
+
+    # Act
+    live_client.start()
+
+    await mock_live_server.wait_for_message_of_type(SessionStart)
+
+    # Assert
+    for _ in range(2):
+        await mock_live_server.wait_for_message_of_type(
+            AuthenticationRequest,
+            timeout=10,
+        )
+        await mock_live_server.wait_for_message_of_type(
+            SessionStart,
+            timeout=10,
+        )
+
+    live_client.stop()