VER: Release 0.31.1

See release notes.

Author: nmacholl

.github/PULL_REQUEST_TEMPLATE.md

@@ -1,6 +1,8 @@
 ## Description
 
-Please include a summary of the change and which issue is fixed. Please also include relevant motivation and context. List any dependencies that are required for this change.
+Please include a summary of the change and which issue is fixed.
+Please also include relevant motivation and context.
+List any dependencies that are required for this change.
 
 Fixes # (issue)
 
@@ -13,18 +15,24 @@ Please delete options that are not relevant.
 - [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
 - [ ] This change requires a documentation update
 
-### How Has This Been Tested?
+### How has this been tested?
 
-Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration
+Please describe the tests that you ran to verify your changes.
+Provide instructions so we can reproduce.
+Please also list any relevant details for your test configuration.
 
 - [ ] Test A
 - [ ] Test B
 
+## Checklist
 
-### Checklist:
-
-- [ ] My code follows the style guidelines of this project
+- [ ] My code builds locally with no new warnings (`scripts/build.sh`)
+- [ ] My code follows the style guidelines (`scripts/lint.sh`)
+- [ ] New and existing unit tests pass locally with my changes (`scripts/test.sh`)
 - [ ] I have made corresponding changes to the documentation
-- [ ] My changes generate no new warnings
 - [ ] I have added tests that prove my fix is effective or that my feature works
-- [ ] New and existing unit tests pass locally with my changes
+
+### Declaration
+
+I confirm this contribution is made under an Apache 2.0 license and that I have the authority
+necessary to make this contribution on behalf of its copyright owner.

.github/workflows/test.yml

@@ -1,8 +1,8 @@
 name: test
 
 on:
+  pull_request:
   push:
-    branches: [main, dev]
 
 jobs:
   test:
@@ -30,13 +30,13 @@ jobs:
         shell: bash
 
       - name: Test (release)
-        timeout-minutes: 5
+        timeout-minutes: 2
         if: ${{ github.ref == 'refs/heads/main' }}
         run: scripts/test.sh --release
         shell: bash
 
       - name: Test
-        timeout-minutes: 5
+        timeout-minutes: 2
         if: ${{ github.ref != 'refs/heads/main' }}
         run: scripts/test.sh
         shell: bash

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## 0.31.1 - 2024-03-20
+
+#### Enhancements
+- Increase `Live` session connection and authentication timeouts
+- Added new `F_TOB` and `F_MAYBE_BAD_BOOK` variants to `RecordFlags`
+
+#### Bug fixes
+- Fixed an issue where calling `Live.subscribe` from a `Live` client callback would cause a deadlock
+
 ## 0.31.0 - 2024-03-05
 
 #### Enhancements

CONTRIBUTING.md

@@ -1,3 +1,4 @@
-We welcome feedback through discussions and issues on GitHub, however we don't currently accept pull requests due to the open-soruce repository being a downstream mirror of our internal Python client library codebase.
-
-Please direct email feedback to [email protected] or [email protected].
+Thank you for taking the time to contribute to our project.
+We welcome feedback through discussions and issues on GitHub, as well as our [community Slack](https://databento.com/support).
+While we don't merge pull requests directly due to the open-source repository being a downstream
+mirror of our internal codebase, we can commit the changes upstream with the original author.

databento/common/enums.py

@@ -199,18 +199,24 @@ class RecordFlags(StringyMixin, IntFlag):  # type: ignore
 
     F_LAST
         Last message in the packet from the venue for a given `instrument_id`.
+    F_TOB
+        Indicates a top-of-book message, not an individual order.
     F_SNAPSHOT
         Message sourced from a replay, such as a snapshot server.
     F_MBP
         Aggregated price level message, not an individual order.
     F_BAD_TS_RECV
         The `ts_recv` value is inaccurate (clock issues or reordering).
+    F_MAYBE_BAD_BOOK
+        Indicates an unrecoverable gap was detected in the channel.
 
     Other bits are reserved and have no current meaning.
 
     """
 
     F_LAST = 128
+    F_TOB = 64
     F_SNAPSHOT = 32
     F_MBP = 16
     F_BAD_TS_RECV = 8
+    F_MAYBE_BAD_BOOK = 4

databento/live/session.py

@@ -25,8 +25,8 @@
 
 logger = logging.getLogger(__name__)
 
-AUTH_TIMEOUT_SECONDS: Final = 2.0
-CONNECT_TIMEOUT_SECONDS: Final = 5.0
+AUTH_TIMEOUT_SECONDS: Final = 30.0
+CONNECT_TIMEOUT_SECONDS: Final = 10.0
 DBN_QUEUE_CAPACITY: Final = 2**20
 DEFAULT_REMOTE_PORT: Final = 13000
 
@@ -434,15 +434,12 @@ def subscribe(
                     loop=self._loop,
                 )
 
-        asyncio.run_coroutine_threadsafe(
-            self._subscribe_task(
+            self._protocol.subscribe(
                 schema=schema,
                 symbols=symbols,
                 stype_in=stype_in,
                 start=start,
-            ),
-            loop=self._loop,
-        ).result()
+            )
 
     def resume_reading(self) -> None:
         """
@@ -565,21 +562,3 @@ async def _connect_task(
         )
 
         return transport, protocol
-
-    async def _subscribe_task(
-        self,
-        schema: Schema | str,
-        symbols: Iterable[str | int] | str | int = ALL_SYMBOLS,
-        stype_in: SType | str = SType.RAW_SYMBOL,
-        start: str | int | None = None,
-    ) -> None:
-        with self._lock:
-            if self._protocol is None:
-                return
-
-            self._protocol.subscribe(
-                schema=schema,
-                symbols=symbols,
-                stype_in=stype_in,
-                start=start,
-            )

databento/version.py

@@ -1 +1 @@
-__version__ = "0.31.0"
+__version__ = "0.31.1"

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "databento"
-version = "0.31.0"
+version = "0.31.1"
 description = "Official Python client library for Databento"
 authors = [
     "Databento <[email protected]>",

tests/mock_live_server.py

@@ -16,6 +16,7 @@
 from functools import singledispatchmethod
 from io import BytesIO
 from os import PathLike
+from typing import Any
 from typing import Callable
 from typing import NewType
 from typing import TypeVar
@@ -32,9 +33,7 @@
 from databento.live.gateway import SessionStart
 from databento.live.gateway import SubscriptionRequest
 from databento.live.gateway import parse_gateway_message
-from databento_dbn import Metadata
 from databento_dbn import Schema
-from databento_dbn import SType
 
 
 LIVE_SERVER_VERSION: str = "1.0.0"
@@ -100,7 +99,8 @@ def __init__(
         self._version: str = version
         self._is_authenticated: bool = False
         self._is_streaming: bool = False
-        self._repeater_tasks: set[asyncio.Task[None]] = set()
+        self._subscriptions: list[SubscriptionRequest] = []
+        self._tasks: set[asyncio.Task[None]] = set()
 
         self._dbn_path = pathlib.Path(dbn_path)
         self._user_api_keys = user_api_keys
@@ -155,6 +155,18 @@ def is_streaming(self) -> bool:
         """
         return self._is_streaming
 
+    @property
+    def dataset_path(self) -> pathlib.Path:
+        """
+        The path to the DBN files for serving.
+
+        Returns
+        -------
+        Path
+
+        """
+        return self._dbn_path / (self._dataset or "")
+
     @property
     def mode(self) -> MockLiveMode:
         """
@@ -205,6 +217,18 @@ def session_id(self) -> str:
         """
         return str(hash(self))
 
+    @property
+    def subscriptions(self) -> tuple[SubscriptionRequest, ...]:
+        """
+        The received subscriptions.
+
+        Returns
+        -------
+        tuple[SubscriptionRequest, ...]
+
+        """
+        return tuple(self._subscriptions)
+
     @property
     def version(self) -> str:
         """
@@ -353,11 +377,11 @@ def _(self, message: AuthenticationRequest) -> None:
         logger.info("received CRAM response: %s", message.auth)
         if self.is_authenticated:
             logger.error("authentication request sent when already authenticated")
-            self.__transport.close()
+            self.__transport.write_eof()
             return
         if self.is_streaming:
             logger.error("authentication request sent while streaming")
-            self.__transport.close()
+            self.__transport.write_eof()
             return
 
         _, bucket_id = message.auth.split("-")
@@ -400,54 +424,49 @@ def _(self, message: SubscriptionRequest) -> None:
         logger.info("received subscription request: %s", str(message).strip())
         if not self.is_authenticated:
             logger.error("subscription request sent while unauthenticated")
-            self.__transport.close()
+            self.__transport.write_eof()
 
-        if self.is_streaming:
-            logger.error("subscription request sent while streaming")
-            self.__transport.close()
+        self._subscriptions.append(message)
 
-        self._schemas.append(Schema(message.schema))
+        if self.is_streaming:
+            self.create_server_task(message)
 
     @handle_client_message.register(SessionStart)
     def _(self, message: SessionStart) -> None:
         logger.info("received session start request: %s", str(message).strip())
         self._is_streaming = True
 
-        dataset_path = self._dbn_path / (self._dataset or "")
+        for sub in self.subscriptions:
+            self.create_server_task(sub)
+
+    def create_server_task(self, message: SubscriptionRequest) -> None:
         if self.mode is MockLiveMode.REPLAY:
-            for schema in self._schemas:
-                for test_data_path in dataset_path.glob(f"*{schema}.dbn.zst"):
-                    decompressor = zstandard.ZstdDecompressor().stream_reader(
-                        test_data_path.read_bytes(),
-                    )
-                    logger.info(
-                        "streaming %s for %s schema",
-                        test_data_path.name,
-                        schema,
-                    )
-                    self.__transport.write(decompressor.readall())
+            task = asyncio.create_task(self.replay_task(schema=Schema(message.schema)))
+        else:
+            task = asyncio.create_task(self.repeater_task(schema=Schema(message.schema)))
 
-            logger.info(
-                "data streaming for %d schema(s) completed",
-                len(self._schemas),
-            )
+        self._tasks.add(task)
+        task.add_done_callback(self._tasks.remove)
+        task.add_done_callback(self.check_done)
 
+    def check_done(self, _: Any) -> None:
+        if not self._tasks:
+            logger.info("streaming tasks completed")
             self.__transport.write_eof()
-            self.__transport.close()
 
-        elif self.mode is MockLiveMode.REPEAT:
-            metadata = Metadata("UNIT.TEST", 0, SType.RAW_SYMBOL, [], [], [], [])
-            self.__transport.write(bytes(metadata))
-
-            loop = asyncio.get_event_loop()
-            for schema in self._schemas:
-                task = loop.create_task(self.repeater(schema))
-                self._repeater_tasks.add(task)
-                task.add_done_callback(self._repeater_tasks.remove)
-        else:
-            raise ValueError(f"unsupported mode {MockLiveMode.REPEAT}")
+    async def replay_task(self, schema: Schema) -> None:
+        for test_data_path in self.dataset_path.glob(f"*{schema}.dbn.zst"):
+            decompressor = zstandard.ZstdDecompressor().stream_reader(
+                test_data_path.read_bytes(),
+            )
+            logger.info(
+                "streaming %s for %s schema",
+                test_data_path.name,
+                schema,
+            )
+            self.__transport.write(decompressor.readall())
 
-    async def repeater(self, schema: Schema) -> None:
+    async def repeater_task(self, schema: Schema) -> None:
         struct = SCHEMA_STRUCT_MAP[schema]
         repeated = bytes(struct(*[0] * 12))  # for now we only support MBP_1