FIX: Fix threadsafety in Live client

Author: nmacholl

CHANGELOG.md

@@ -8,6 +8,10 @@
   downloading a batch job as a ZIP archive
 - Added method for `batch.get_job_details` to access the full details of a batch job
 
+#### Bug fixes
+- Fixed some thread-unsafe behavior in `Live.start()` and `Live.terminate()` which would
+  call methods from the client's event loop in the main thread
+
 ## 0.78.0 - 2026-05-12
 
 #### Enhancements

databento/live/session.py

@@ -517,7 +517,7 @@ def start(self) -> None:
         with self._lock:
             if self._protocol is None:
                 raise ValueError("session is not connected")
-            self._protocol.start()
+            self._loop.call_soon_threadsafe(self._protocol.start)
             self._heartbeat_monitor_task = self._loop.create_task(
                 self._heartbeat_monitor(),
             )
@@ -585,7 +585,7 @@ def terminate(self) -> None:
         with self._lock:
             if self._transport is None:
                 return
-            self._transport.abort()
+            self._loop.call_soon_threadsafe(self._transport.abort)
             self._cleanup()
 
     async def wait_for_close(self) -> None:

tests/test_live_client.py

@@ -465,6 +465,7 @@ async def test_live_start_twice(
 
     # Act, Assert
     live_client.start()
+    await mock_live_server.wait_for_message_of_type(message_type=gateway.SessionStart)
     with pytest.raises(ValueError):
         live_client.start()
 

tests/test_live_client_reconnect.py

@@ -45,7 +45,10 @@ async def test_reconnect_policy_none(
         stype_in=SType.RAW_SYMBOL,
         symbols="TEST",
     )
-    live_client._session._protocol.disconnected.set_exception(ConnectionResetError)
+    live_client._loop.call_soon_threadsafe(
+        live_client._session._protocol.disconnected.set_exception,
+        ConnectionResetError,
+    )
 
     await mock_live_server.wait_for_message_of_type(AuthenticationRequest)
 
@@ -90,7 +93,10 @@ async def test_reconnect_before_start(
     live_client.add_callback(records.append)
 
     # Act
-    live_client._session._protocol.disconnected.set_exception(ConnectionResetError)
+    live_client._loop.call_soon_threadsafe(
+        live_client._session._protocol.disconnected.set_exception,
+        ConnectionResetError,
+    )
     await mock_live_server.disconnect(
         session_id=live_client._session.session_id,
     )
@@ -162,7 +168,10 @@ async def test_reconnect_subscriptions(
     await mock_live_server.wait_for_message_of_type(AuthenticationRequest)
 
     # Act
-    live_client._session._protocol.disconnected.set_exception(ConnectionResetError)
+    live_client._loop.call_soon_threadsafe(
+        live_client._session._protocol.disconnected.set_exception,
+        ConnectionResetError,
+    )
     await mock_live_server.disconnect(
         session_id=live_client._session.session_id,
     )
@@ -217,8 +226,11 @@ async def test_reconnect_callback(
 
     # Act
     live_client.start()
-    live_client._session._protocol.disconnected.set_exception(ConnectionResetError)
 
+    live_client._loop.call_soon_threadsafe(
+        live_client._session._protocol.disconnected.set_exception,
+        ConnectionResetError,
+    )
     await mock_live_server.wait_for_message_of_type(SessionStart)
 
     await live_client.wait_for_close()