ADD: Add support for slow_reader_behavior

Author: nmacholl

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.71.0 - TBD
+
+#### Enhancements
+- Added `slow_reader_behavior` field to `AuthenticationRequest` message
+- Added `SlowReadBehavior` enum
+
 ## 0.70.0 - 2026-01-27
 
 #### Enhancements

databento/common/enums.py

@@ -257,3 +257,14 @@ class JobState(StringyMixin, str, Enum):
     PROCESSING = "processing"
     DONE = "done"
     EXPIRED = "expired"
+
+
+@unique
+class SlowReadBehavior(StringyMixin, str, Enum):
+    """
+    Live session parameter which controls gateway behavior when the client
+    falls behind real time.
+    """
+
+    SKIP = "skip"
+    WARN = "warn"

databento/live/client.py

@@ -21,6 +21,7 @@
 from databento.common.constants import ALL_SYMBOLS
 from databento.common.cram import BUCKET_ID_LENGTH
 from databento.common.enums import ReconnectPolicy
+from databento.common.enums import SlowReadBehavior
 from databento.common.error import BentoError
 from databento.common.parsing import optional_datetime_to_unix_nanoseconds
 from databento.common.publishers import Dataset
@@ -63,6 +64,10 @@ class Live:
         The reconnect policy for the live session.
             - "none": the client will not reconnect (default)
             - "reconnect": the client will reconnect automatically
+    slow_reader_behavior: SlowReadBehavior | str, optional
+        The live gateway behavior when the client falls behind real time.
+            - "skip": skip records to immediately catch up
+            - "warn": send a slow reader warning `SystemMsg` but continue reading every record
 
     """
 
@@ -82,6 +87,7 @@ def __init__(
         ts_out: bool = False,
         heartbeat_interval_s: int | None = None,
         reconnect_policy: ReconnectPolicy | str = ReconnectPolicy.NONE,
+        slow_reader_behavior: SlowReadBehavior | str | None = None,
     ) -> None:
         if key is None:
             key = os.environ.get("DATABENTO_API_KEY")
@@ -112,6 +118,7 @@ def __init__(
             user_gateway=self._gateway,
             user_port=port,
             reconnect_policy=reconnect_policy,
+            slow_reader_behavior=slow_reader_behavior,
         )
 
         self._session._user_callbacks.append(ClientRecordCallback(self._map_symbol))

databento/live/gateway.py

@@ -11,6 +11,7 @@
 from databento_dbn import Schema
 from databento_dbn import SType
 
+from databento.common.enums import SlowReadBehavior
 from databento.common.publishers import Dataset
 from databento.common.system import USER_AGENT
 
@@ -118,8 +119,14 @@ class AuthenticationRequest(GatewayControl):
     details: str | None = None
     ts_out: str = "0"
     heartbeat_interval_s: int | None = None
+    slow_reader_behavior: SlowReadBehavior | str | None = None
     client: str = USER_AGENT
 
+    def __post_init__(self) -> None:
+        # Temporary work around for LSG support
+        if self.slow_reader_behavior in [SlowReadBehavior.SKIP, "skip"]:
+            self.slow_reader_behavior = "drop"
+
 
 @dataclasses.dataclass
 class SubscriptionRequest(GatewayControl):

databento/live/protocol.py

@@ -16,6 +16,7 @@
 
 from databento.common import cram
 from databento.common.constants import ALL_SYMBOLS
+from databento.common.enums import SlowReadBehavior
 from databento.common.error import BentoError
 from databento.common.iterator import chunk
 from databento.common.parsing import optional_datetime_to_unix_nanoseconds
@@ -73,6 +74,7 @@ def __init__(
         dataset: Dataset | str,
         ts_out: bool = False,
         heartbeat_interval_s: int | None = None,
+        slow_reader_behavior: SlowReadBehavior | str | None = None,
     ) -> None:
         self.__api_key = api_key
         self.__transport: asyncio.Transport | None = None
@@ -81,6 +83,7 @@ def __init__(
         self._dataset = validate_semantic_string(dataset, "dataset")
         self._ts_out = ts_out
         self._heartbeat_interval_s = heartbeat_interval_s
+        self._slow_reader_behavior: SlowReadBehavior | str | None = slow_reader_behavior
 
         self._dbn_decoder = databento_dbn.DBNDecoder(
             upgrade_policy=VersionUpgradePolicy.UPGRADE_TO_V3,
@@ -441,6 +444,7 @@ def _(self, message: ChallengeRequest) -> None:
             dataset=self._dataset,
             ts_out=str(int(self._ts_out)),
             heartbeat_interval_s=self._heartbeat_interval_s,
+            slow_reader_behavior=self._slow_reader_behavior,
         )
         logger.debug(
             "sending CRAM challenge response auth='%s' dataset=%s encoding=%s ts_out=%s heartbeat_interval_s=%s client='%s'",

databento/live/session.py

@@ -20,6 +20,7 @@
 
 from databento.common.constants import ALL_SYMBOLS
 from databento.common.enums import ReconnectPolicy
+from databento.common.enums import SlowReadBehavior
 from databento.common.error import BentoError
 from databento.common.publishers import Dataset
 from databento.common.types import ClientRecordCallback
@@ -205,8 +206,9 @@ def __init__(
         metadata: SessionMetadata,
         ts_out: bool = False,
         heartbeat_interval_s: int | None = None,
+        slow_reader_behavior: SlowReadBehavior | str | None = None,
     ):
-        super().__init__(api_key, dataset, ts_out, heartbeat_interval_s)
+        super().__init__(api_key, dataset, ts_out, heartbeat_interval_s, slow_reader_behavior)
 
         self._dbn_queue = dbn_queue
         self._loop = loop
@@ -313,6 +315,7 @@ def __init__(
         user_gateway: str | None = None,
         user_port: int = DEFAULT_REMOTE_PORT,
         reconnect_policy: ReconnectPolicy | str = ReconnectPolicy.NONE,
+        slow_reader_behavior: SlowReadBehavior | str | None = None,
     ) -> None:
         self._dbn_queue = DBNQueue()
         self._lock = threading.RLock()
@@ -329,6 +332,7 @@ def __init__(
         self._api_key = api_key
         self._ts_out = ts_out
         self._heartbeat_interval_s = heartbeat_interval_s or 30
+        self._slow_reader_behavior = slow_reader_behavior
 
         self._protocol: _SessionProtocol | None = None
         self._transport: asyncio.Transport | None = None
@@ -579,6 +583,7 @@ def _create_protocol(self, dataset: Dataset | str) -> _SessionProtocol:
             metadata=self._metadata,
             ts_out=self.ts_out,
             heartbeat_interval_s=self.heartbeat_interval_s,
+            slow_reader_behavior=self._slow_reader_behavior,
         )
 
     def _connect(

tests/test_live_client.py

@@ -24,6 +24,7 @@
 from databento.common.constants import SCHEMA_STRUCT_MAP
 from databento.common.cram import BUCKET_ID_LENGTH
 from databento.common.dbnstore import DBNStore
+from databento.common.enums import SlowReadBehavior
 from databento.common.error import BentoError
 from databento.common.publishers import Dataset
 from databento.live import client
@@ -331,6 +332,50 @@ async def test_live_connect_auth_with_heartbeat_interval(
     assert message.heartbeat_interval_s == "10"
 
 
+@pytest.mark.parametrize(
+    "slow_reader_behavior",
+    [b for b in SlowReadBehavior],
+)
+async def test_live_connect_auth_with_slow_reader_behavior(
+    mock_live_server: MockLiveServerInterface,
+    test_live_api_key: str,
+    slow_reader_behavior: SlowReadBehavior,
+) -> None:
+    """
+    Test that setting `slow_reader_behavior` on a Live client sends that field
+    to the gateway.
+    """
+    # Arrange
+    live_client = client.Live(
+        key=test_live_api_key,
+        gateway=mock_live_server.host,
+        port=mock_live_server.port,
+        heartbeat_interval_s=10,
+        slow_reader_behavior=slow_reader_behavior,
+    )
+
+    live_client.subscribe(
+        dataset=Dataset.GLBX_MDP3,
+        schema=Schema.MBO,
+    )
+
+    # Act
+    message = await mock_live_server.wait_for_message_of_type(
+        message_type=gateway.AuthenticationRequest,
+    )
+
+    # Assert
+    assert message.auth.endswith(live_client.key[-BUCKET_ID_LENGTH:])
+    assert message.dataset == live_client.dataset
+    assert message.encoding == Encoding.DBN
+
+    # Temporary handling of renamed variant
+    if slow_reader_behavior == SlowReadBehavior.SKIP:
+        assert message.slow_reader_behavior == "drop"
+    else:
+        assert message.slow_reader_behavior == slow_reader_behavior
+
+
 async def test_live_connect_auth_two_clients(
     mock_live_server: MockLiveServerInterface,
     test_live_api_key: str,