MOD: Reduce Python client buffering

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.78.0 - TBD
+
+#### Enhancements
+- Added time-based backpressure to the live client: pauses reading records from the live
+  gateway when the internal queue spans more than 1 second of data by `ts_index`
+  (`ts_recv` when present, otherwise `ts_event`)
+
 ## 0.77.0 - 2026-04-28
 
 #### Enhancements

databento/live/session.py

@@ -36,6 +36,9 @@
 AUTH_TIMEOUT_SECONDS: Final = 30.0
 CONNECT_TIMEOUT_SECONDS: Final = 10.0
 DBN_QUEUE_CAPACITY: Final = 2**20
+DBN_QUEUE_LAG_THRESHOLD: Final = 128
+DBN_QUEUE_MAX_LAG_NS: Final = 1_000_000_000
+DBN_QUEUE_FULL_WARNING_INTERVAL_S: Final = 60.0
 DEFAULT_REMOTE_PORT: Final = 13000
 CLIENT_TIMEOUT_MARGIN_SECONDS: Final = 10
 
@@ -48,6 +51,8 @@ class DBNQueue(queue.SimpleQueue):  # type: ignore [type-arg]
     def __init__(self) -> None:
         super().__init__()
         self._enabled = threading.Event()
+        self._front_ts_index: int | None = None
+        self._back_ts_index: int | None = None
 
     def is_enabled(self) -> bool:
         """
@@ -62,7 +67,16 @@ def is_full(self) -> bool:
         """
         Return True when the queue has reached capacity; False otherwise.
         """
-        return self.qsize() > DBN_QUEUE_CAPACITY
+        if self.qsize() > DBN_QUEUE_CAPACITY:
+            return True
+        if (
+            self.qsize() > DBN_QUEUE_LAG_THRESHOLD
+            and self._front_ts_index is not None
+            and self._back_ts_index is not None
+            and self._back_ts_index - self._front_ts_index > DBN_QUEUE_MAX_LAG_NS
+        ):
+            return True
+        return False
 
     def enable(self) -> None:
         """
@@ -106,6 +120,9 @@ def put(
 
         """
         if self._enabled.wait(timeout):
+            if self._front_ts_index is None:
+                self._front_ts_index = item.ts_index
+            self._back_ts_index = item.ts_index
             return super().put(item, block, timeout)
         if timeout is not None:
             raise BentoError(f"queue is not enabled after {timeout} second(s)")
@@ -131,9 +148,34 @@ def put_nowait(self, item: DBNRecord) -> None:
 
         """
         if self.is_enabled():
+            if self._front_ts_index is None:
+                self._front_ts_index = item.ts_index
+            self._back_ts_index = item.ts_index
             return super().put_nowait(item)
         raise BentoError("queue is not enabled")
 
+    def get(
+        self,
+        block: bool = True,
+        timeout: float | None = None,
+    ) -> DBNRecord:
+        record = super().get(block, timeout)
+        if self.empty():
+            self._front_ts_index = None
+            self._back_ts_index = None
+        else:
+            self._front_ts_index = record.ts_index
+        return record
+
+    def get_nowait(self) -> DBNRecord:
+        record = super().get_nowait()
+        if self.empty():
+            self._front_ts_index = None
+            self._back_ts_index = None
+        else:
+            self._front_ts_index = record.ts_index
+        return record
+
 
 @dataclasses.dataclass
 class SessionMetadata:
@@ -225,6 +267,7 @@ def __init__(
         self._user_streams = user_streams
         self._last_ts_event: int | None = None
         self._last_msg_loop_time: float = math.inf
+        self._last_queue_full_warning_t: float = -math.inf
 
     def received_metadata(self, metadata: databento_dbn.Metadata) -> None:
         if self._metadata:
@@ -282,10 +325,13 @@ def _queue_for_iteration(self, record: DBNRecord) -> None:
         self._dbn_queue.put(record)
         # DBNQueue has no max size; so check if it's above capacity, and if so, pause reading
         if self._dbn_queue.is_full():
-            logger.warning(
-                "record queue is full; %d record(s) to be processed",
-                self._dbn_queue.qsize(),
-            )
+            now = self._loop.time()
+            if now - self._last_queue_full_warning_t >= DBN_QUEUE_FULL_WARNING_INTERVAL_S:
+                logger.warning(
+                    "record queue is full; %d record(s) to be processed",
+                    self._dbn_queue.qsize(),
+                )
+                self._last_queue_full_warning_t = now
             self.transport.pause_reading()
 
 

tests/test_live_session.py

@@ -1,9 +1,41 @@
 import pytest
+from databento_dbn import CBBOMsg
+from databento_dbn import OHLCVMsg
+from databento_dbn import Side
 
 from databento.common.error import BentoError
+from databento.live.session import DBN_QUEUE_LAG_THRESHOLD
+from databento.live.session import DBN_QUEUE_MAX_LAG_NS
 from databento.live.session import DBNQueue
 
 
+def _record(ts_event: int = 0) -> OHLCVMsg:
+    return OHLCVMsg(
+        rtype=1,
+        publisher_id=1,
+        instrument_id=0,
+        ts_event=ts_event,
+        open=100,
+        high=110,
+        low=90,
+        close=105,
+        volume=1000,
+    )
+
+
+def _cbbo_record(ts_event: int, ts_recv: int) -> CBBOMsg:
+    return CBBOMsg(
+        rtype=0xC1,
+        publisher_id=1,
+        instrument_id=0,
+        ts_event=ts_event,
+        price=1_000_000_000,
+        size=1,
+        side=Side.NONE,
+        ts_recv=ts_recv,
+    )
+
+
 def test_dbn_queue_put(
     timeout: float = 0.01,
 ) -> None:
@@ -15,17 +47,18 @@ def test_dbn_queue_put(
     """
     # Arrange
     queue = DBNQueue()
+    record = _record()
 
     # Act, Assert
     with pytest.raises(BentoError):
-        queue.put(None, timeout=timeout)
+        queue.put(record, timeout=timeout)
 
     queue.enable()
-    queue.put(None, timeout=timeout)
+    queue.put(record, timeout=timeout)
 
     queue.disable()
     with pytest.raises(BentoError):
-        queue.put(None, timeout=timeout)
+        queue.put(record, timeout=timeout)
 
 
 def test_dbn_queue_put_nowait() -> None:
@@ -34,14 +67,55 @@ def test_dbn_queue_put_nowait() -> None:
     """
     # Arrange
     queue = DBNQueue()
+    record = _record()
 
     # Act, Assert
     with pytest.raises(BentoError):
-        queue.put_nowait(None)
+        queue.put_nowait(record)
 
     queue.enable()
-    queue.put_nowait(None)
+    queue.put_nowait(record)
 
     queue.disable()
     with pytest.raises(BentoError):
-        queue.put_nowait(None)
+        queue.put_nowait(record)
+
+
+def test_dbn_queue_is_full_uses_ts_index() -> None:
+    queue = DBNQueue()
+    queue.enable()
+
+    base_recv = 1_700_000_000_000_000_000
+    stale_event_start = 1_600_000_000_000_000_000
+    ten_ms = 10_000_000
+    for i in range(DBN_QUEUE_LAG_THRESHOLD + 2):
+        queue.put_nowait(
+            _cbbo_record(
+                ts_event=stale_event_start + i * ten_ms,
+                ts_recv=base_recv + i,
+            ),
+        )
+
+    assert queue.qsize() > DBN_QUEUE_LAG_THRESHOLD
+    assert not queue.is_full()
+
+
+def test_dbn_queue_is_full_triggers_on_ts_recv_lag() -> None:
+    """
+    `ts_recv` lag past `DBN_QUEUE_MAX_LAG_NS` should trip backpressure even
+    when `ts_event` is constant.
+    """
+    queue = DBNQueue()
+    queue.enable()
+
+    base_recv = 1_700_000_000_000_000_000
+    step = (DBN_QUEUE_MAX_LAG_NS // DBN_QUEUE_LAG_THRESHOLD) + 1
+    for i in range(DBN_QUEUE_LAG_THRESHOLD + 2):
+        queue.put_nowait(
+            _cbbo_record(
+                ts_event=0,
+                ts_recv=base_recv + i * step,
+            ),
+        )
+
+    assert queue.is_full()