FIX: Fix Live.stop() disconnection with cleanup

Author: nmacholl

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.62.1 - TBD
+
+#### Bug fixes
+- Fixed an issue where calling `Live.stop()` would not clean up the client state once the socket is closed
+
 ## 0.62.0 - 2025-08-19
 
 This release delivers a number of breaking changes to the Python interface for DBN records to provide a cleaner and more consistent API.

databento/live/session.py

@@ -439,9 +439,9 @@ def stop(self) -> None:
         with self._lock:
             if self._transport is None:
                 return
-            if self._transport.can_write_eof():
-                self._transport.write_eof()
-            self._transport.close()
+            if self._protocol is not None:
+                self._protocol.disconnected.add_done_callback(lambda _: self._cleanup())
+            self._loop.call_soon_threadsafe(self._transport.close)
 
     def start(self) -> None:
         """
@@ -516,8 +516,6 @@ def terminate(self) -> None:
         with self._lock:
             if self._transport is None:
                 return
-            if self._transport.can_write_eof():
-                self._transport.write_eof()
             self._transport.abort()
             self._cleanup()
 

tests/test_live_client.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import pathlib
 import platform
 import random
@@ -255,6 +256,42 @@ async def test_live_connect_auth(
     assert message.encoding == Encoding.DBN
 
 
+async def test_live_client_reuse(
+    mock_live_server: MockLiveServerInterface,
+    live_client: client.Live,
+) -> None:
+    """
+    Test that calling stop will *eventually* close a the connection and trigger
+    a cleanup of the client state.
+    """
+    live_client.subscribe(
+        dataset=Dataset.GLBX_MDP3,
+        schema=Schema.MBO,
+    )
+
+    await mock_live_server.wait_for_message_of_type(
+        message_type=gateway.AuthenticationRequest,
+    )
+
+    live_client.start()
+    live_client.stop()
+
+    await asyncio.sleep(1)
+
+    live_client.subscribe(
+        dataset=Dataset.GLBX_MDP3,
+        schema=Schema.MBP_1,
+    )
+
+    await mock_live_server.wait_for_message_of_type(
+        message_type=gateway.AuthenticationRequest,
+    )
+
+    live_client.start()
+    live_client.stop()
+    await live_client.wait_for_close()
+
+
 async def test_live_connect_auth_with_heartbeat_interval(
     mock_live_server: MockLiveServerInterface,
     test_live_api_key: str,