VER: Release 0.26.2

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## 0.26.2 - 2025-06-03
+
+### Enhancements
+- Improved performance of live client by removing redundant state
+- Upgraded DBN version to 0.35.1
+
+### Bug fixes
+- Fixed handling of `VersionUpgradePolicy` in `timeseries().get_range()` and
+  `get_range_to_file()`
+- Bug fixes from DBN:
+  - Fixed behavior where encoding metadata could lower the `version`
+  - Changed `DbnFsm::data()` to exclude all processed data
+  - Fixed `Metadata::upgrade()` behavior with `UpgradeToV2`
+
 ## 0.26.1 - 2025-05-30
 
 ### Bug fixes

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "databento"
 authors = ["Databento <support@databento.com>"]
-version = "0.26.1"
+version = "0.26.2"
 edition = "2021"
 repository = "https://github.com/databento/databento-rs"
 description = "Official Databento client library"
@@ -19,11 +19,21 @@ rustdoc-args = ["--cfg", "docsrs"]
 
 [features]
 default = ["historical", "live"]
-historical = ["dep:futures", "dep:reqwest", "dep:serde", "dep:tokio-util", "dep:serde_json", "tokio/fs"]
+historical = [
+  "dep:async-compression",
+  "dep:futures",
+  "dep:reqwest",
+  "dep:serde",
+  "dep:tokio-util",
+  "dep:serde_json",
+  "tokio/fs"
+]
 live = ["dep:hex", "dep:sha2", "tokio/net"]
 
 [dependencies]
-dbn = { version = "0.35.0", features = ["async", "serde"] }
+dbn = { version = "0.35.1", features = ["async", "serde"] }
+
+async-compression = { version = "0.4", features = ["tokio", "zstd"], optional = true }
 # Async stream trait
 futures = { version = "0.3", optional = true }
 # Used for Live authentication
@@ -43,8 +53,9 @@ typed-builder = "0.21"
 
 [dev-dependencies]
 anyhow = "1.0.98"
-async-compression = { version = "0.4.23", features = ["tokio", "zstd"] }
+async-compression = { version = "0.4", features = ["tokio", "zstd"] }
 clap = { version = "4.5.37", features = ["derive"] }
+rstest = "0.25.0"
 tempfile = "3.19.1"
 tokio = { version = "1.44", features = ["full"] }
 tracing-subscriber = "0.3.19"

src/historical/timeseries.rs

@@ -7,7 +7,7 @@ use futures::{Stream, TryStreamExt};
 use reqwest::{header::ACCEPT, RequestBuilder};
 use tokio::{
     fs::File,
-    io::{AsyncReadExt, AsyncWriteExt, BufWriter},
+    io::{AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},
 };
 use tokio_util::{bytes::Bytes, io::StreamReader};
 use typed_builder::TypedBuilder;
@@ -53,9 +53,10 @@ impl TimeseriesClient<'_> {
                 params.limit,
             )
             .await?;
-        let mut decoder: AsyncDbnDecoder<_> = AsyncDbnDecoder::with_zstd_buffer(reader).await?;
-        decoder.set_upgrade_policy(params.upgrade_policy)?;
-        Ok(decoder)
+        Ok(
+            AsyncDbnDecoder::with_upgrade_policy(zstd_decoder(reader), params.upgrade_policy)
+                .await?,
+        )
     }
 
     /// Makes a streaming request for timeseries data from Databento.
@@ -86,15 +87,21 @@ impl TimeseriesClient<'_> {
                 params.limit,
             )
             .await?;
-        let mut http_decoder = AsyncDbnDecoder::with_zstd_buffer(reader).await?;
-        http_decoder.set_upgrade_policy(params.upgrade_policy)?;
+        let mut http_decoder =
+            AsyncDbnDecoder::with_upgrade_policy(zstd_decoder(reader), params.upgrade_policy)
+                .await?;
         let file = BufWriter::new(File::create(&params.path).await?);
         let mut encoder = AsyncDbnEncoder::with_zstd(file, http_decoder.metadata()).await?;
         while let Some(rec_ref) = http_decoder.decode_record_ref().await? {
             encoder.encode_record_ref(rec_ref).await?;
         }
         encoder.get_mut().shutdown().await?;
-        Ok(AsyncDbnDecoder::from_zstd_file(&params.path).await?)
+        Ok(AsyncDbnDecoder::with_upgrade_policy(
+            zstd_decoder(BufReader::new(File::open(&params.path).await?)),
+            // Applied upgrade policy during initial decoding
+            VersionUpgradePolicy::AsIs,
+        )
+        .await?)
     }
 
     #[allow(clippy::too_many_arguments)] // private method
@@ -241,10 +248,21 @@ impl GetRangeParams {
     }
 }
 
+fn zstd_decoder<R>(reader: R) -> async_compression::tokio::bufread::ZstdDecoder<R>
+where
+    R: tokio::io::AsyncBufReadExt + Unpin,
+{
+    let mut zstd_decoder = async_compression::tokio::bufread::ZstdDecoder::new(reader);
+    // explicitly enable decoding multiple frames
+    zstd_decoder.multiple_members(true);
+    zstd_decoder
+}
+
 #[cfg(test)]
 mod tests {
     use dbn::{record::TradeMsg, Dataset};
     use reqwest::StatusCode;
+    use rstest::*;
     use time::macros::datetime;
     use wiremock::{
         matchers::{basic_auth, method, path},
@@ -260,8 +278,12 @@ mod tests {
 
     const API_KEY: &str = "test-API";
 
+    #[rstest]
+    #[case(VersionUpgradePolicy::AsIs, 1)]
+    #[case(VersionUpgradePolicy::UpgradeToV2, 2)]
+    #[case(VersionUpgradePolicy::UpgradeToV3, 3)]
     #[tokio::test]
-    async fn test_get_range() {
+    async fn test_get_range(#[case] upgrade_policy: VersionUpgradePolicy, #[case] exp_version: u8) {
         const START: time::OffsetDateTime = datetime!(2023 - 06 - 14 00:00 UTC);
         const END: time::OffsetDateTime = datetime!(2023 - 06 - 17 00:00 UTC);
         const SCHEMA: Schema = Schema::Trades;
@@ -299,19 +321,29 @@ mod tests {
                     .schema(SCHEMA)
                     .symbols(vec!["SPOT", "AAPL"])
                     .date_time_range((START, END))
+                    .upgrade_policy(upgrade_policy)
                     .build(),
             )
             .await
             .unwrap();
-        assert_eq!(decoder.metadata().schema.unwrap(), SCHEMA);
+        let metadata = decoder.metadata();
+        assert_eq!(metadata.schema.unwrap(), SCHEMA);
+        assert_eq!(metadata.version, exp_version);
         // Two records
         decoder.decode_record::<TradeMsg>().await.unwrap().unwrap();
         decoder.decode_record::<TradeMsg>().await.unwrap().unwrap();
         assert!(decoder.decode_record::<TradeMsg>().await.unwrap().is_none());
     }
 
+    #[rstest]
+    #[case(VersionUpgradePolicy::AsIs, 1)]
+    #[case(VersionUpgradePolicy::UpgradeToV2, 2)]
+    #[case(VersionUpgradePolicy::UpgradeToV3, 3)]
     #[tokio::test]
-    async fn test_get_range_to_file() {
+    async fn test_get_range_to_file(
+        #[case] upgrade_policy: VersionUpgradePolicy,
+        #[case] exp_version: u8,
+    ) {
         const START: time::OffsetDateTime = datetime!(2024 - 05 - 17 00:00 UTC);
         const END: time::OffsetDateTime = datetime!(2024 - 05 - 18 00:00 UTC);
         const SCHEMA: Schema = Schema::Trades;
@@ -354,11 +386,14 @@ mod tests {
                     .stype_in(SType::Parent)
                     .date_time_range((START, END))
                     .path(path.clone())
+                    .upgrade_policy(upgrade_policy)
                     .build(),
             )
             .await
             .unwrap();
-        assert_eq!(decoder.metadata().schema.unwrap(), SCHEMA);
+        let metadata = decoder.metadata();
+        assert_eq!(metadata.schema.unwrap(), SCHEMA);
+        assert_eq!(metadata.version, exp_version);
         // Two records
         decoder.decode_record::<TradeMsg>().await.unwrap().unwrap();
         decoder.decode_record::<TradeMsg>().await.unwrap().unwrap();

src/live/client.rs

@@ -1,7 +1,7 @@
 use std::{fmt, net::SocketAddr};
 
 use dbn::{
-    decode::dbn::{AsyncMetadataDecoder, AsyncRecordDecoder},
+    decode::dbn::{async_decode_metadata_with_fsm, async_decode_record_ref_with_fsm, fsm::DbnFsm},
     Metadata, RecordRef, VersionUpgradePolicy,
 };
 use time::Duration;
@@ -32,17 +32,12 @@ pub struct Client {
     peer_addr: SocketAddr,
     sub_counter: u32,
     subscriptions: Vec<Subscription>,
-    decoder: Decoder,
+    reader: ReadHalf<TcpStream>,
+    fsm: DbnFsm,
     session_id: String,
     span: Span,
 }
 
-enum Decoder {
-    Metadata(AsyncMetadataDecoder<ReadHalf<TcpStream>>),
-    Record(AsyncRecordDecoder<ReadHalf<TcpStream>>),
-    Empty,
-}
-
 impl Client {
     /// Creates a new client connected to a Live gateway.
     ///
@@ -107,10 +102,12 @@ impl Client {
             heartbeat_interval,
             protocol,
             peer_addr,
-            decoder: Decoder::Metadata(AsyncMetadataDecoder::with_upgrade_policy(
-                recver.into_inner(),
-                upgrade_policy,
-            )),
+            reader: recver.into_inner(),
+            fsm: DbnFsm::builder()
+                .upgrade_policy(upgrade_policy)
+                .build()
+                // Not setting input version so it's infallible
+                .unwrap(),
             session_id,
             span,
             sub_counter: 0,
@@ -220,20 +217,15 @@ impl Client {
     /// closing the connection.
     #[instrument(parent = &self.span, skip_all)]
     pub async fn start(&mut self) -> crate::Result<Metadata> {
-        let decoder @ Decoder::Metadata(_) = &mut self.decoder else {
+        if self.fsm.has_decoded_metadata() {
             return Err(crate::Error::BadArgument {
                 param_name: "self".to_owned(),
                 desc: "ignored request to start session that has already been started".to_owned(),
             });
         };
-        let Decoder::Metadata(mut decoder) = std::mem::replace(decoder, Decoder::Empty) else {
-            unreachable!("previously checked decoder type");
-        };
         info!("Starting session");
         self.protocol.start_session().await?;
-        let metadata = decoder.decode().await?;
-        self.decoder = Decoder::Record(AsyncRecordDecoder::from(decoder));
-        Ok(metadata)
+        Ok(async_decode_metadata_with_fsm(&mut self.reader, &mut self.fsm).await?)
     }
 
     /// Fetches the next record. This method should only be called after the session has
@@ -252,13 +244,13 @@ impl Client {
     /// without the potential for corrupting the input stream.
     #[instrument(parent = &self.span, level = "debug", skip_all)]
     pub async fn next_record(&mut self) -> crate::Result<Option<RecordRef>> {
-        let Decoder::Record(decoder) = &mut self.decoder else {
+        if !self.fsm.has_decoded_metadata() {
             return Err(crate::Error::BadArgument {
                 param_name: "self".to_owned(),
                 desc: "Can't call LiveClient::next_record before starting session".to_owned(),
             });
         };
-        Ok(decoder.decode_ref().await?)
+        Ok(async_decode_record_ref_with_fsm(&mut self.reader, &mut self.fsm).await?)
     }
 
     /// Closes the current connection, then reopens the connection and authenticates
@@ -295,10 +287,8 @@ impl Client {
                 self.heartbeat_interval.map(|i| i.whole_seconds()),
             )
             .await?;
-        self.decoder = Decoder::Metadata(AsyncMetadataDecoder::with_upgrade_policy(
-            recver.into_inner(),
-            self.upgrade_policy,
-        ));
+        self.reader = recver.into_inner();
+        self.fsm.reset();
         self.span = info_span!("LiveClient", dataset = %self.dataset, session_id = self.session_id);
         Ok(())
     }