VER: Release 0.26.0

Author: threecgreen

CHANGELOG.md

@@ -1,6 +1,39 @@
 # Changelog
 
-## 0.25.0 - TBD
+
+## 0.26.0 - 2025-05-28
+
+This version marks the release of DBN version 3 (DBNv3), which is the new default.
+API methods and decoders support decoding all versions of DBN, but now default to
+upgrading data to version 3.
+
+### Enhancements - Added `From<DatasetRange>` conversion for `DateTimeRange`
+- Added `is_last` field to live subscription requests which will be used to improve the
+  handling of split subscription requests
+- Upgraded DBN version to 0.35.0:
+  - Version 1 and 2 structs can be converted to version 3 structs with the `From` trait
+  - Implemented conversion from `RecordRef` to `IoSlice` for use with
+    `Write::write_vectored`
+
+### Breaking changes
+- Breaking changes from DBN:
+  - Definition schema:
+    - Updated `InstrumentDefMsg` with new `leg_` fields to support multi-leg strategy
+      definitions.
+    - Expanded `asset` to 11 bytes and `ASSET_CSTR_LEN` to match
+    - Expanded `raw_instrument_id` to 64 bits to support more venues. Like other 64-bit
+      integer fields, its value will now be quoted in JSON
+    - Removed `trading_reference_date`, `trading_reference_price`, and
+      `settl_price_type` fields which will be normalized in the statistics schema
+    - Removed `md_security_trading_status` better served by the status schema
+  - Statistics schema:
+    - Updated `StatMsg` has an expanded 64-bit `quantity` field. Like other 64-bit
+      integer fields, its value will now be quoted in JSON
+    - The previous `StatMsg` has been moved to `v2::StatMsg` or `StatMsgV2`
+  - Changed the default `VersionUpgradePolicy` to `UpgradeToV3`
+  - Updated the minimum supported `tokio` version to 1.38, which was released one year ago
+
+## 0.25.0 - 2025-05-13
 
 ### Enhancements
 - Increased live subscription symbol chunking size

Cargo.toml

@@ -23,7 +23,7 @@ historical = ["dep:futures", "dep:reqwest", "dep:serde", "dep:tokio-util", "dep:
 live = ["dep:hex", "dep:sha2", "tokio/net"]
 
 [dependencies]
-dbn = { version = "0.34.0", features = ["async", "serde"] }
+dbn = { version = "0.35.0", features = ["async", "serde"] }
 # Async stream trait
 futures = { version = "0.3", optional = true }
 # Used for Live authentication
@@ -44,8 +44,8 @@ typed-builder = "0.21"
 [dev-dependencies]
 anyhow = "1.0.98"
 async-compression = { version = "0.4.23", features = ["tokio", "zstd"] }
-clap = { version = "4.5.37", features = ["derive"] }
-tempfile = "3.19.1"
-tokio = { version = "1.44", features = ["full"] }
+clap = { version = "4.5.39", features = ["derive"] }
+tempfile = "3.20.0"
+tokio = { version = "1.45", features = ["full"] }
 tracing-subscriber = "0.3.19"
 wiremock = "0.6"

src/historical/metadata.rs

@@ -159,7 +159,7 @@ impl MetadataClient<'_> {
     }
 
     /// Gets the cost in US dollars for a historical streaming or batch download
-    /// request.
+    /// request. This cost respects any discounts provided by flat rate plans.
     ///
     /// # Errors
     /// This function returns an error when it fails to communicate with the Databento API
@@ -286,6 +286,12 @@ pub struct DatasetRange {
     pub end: time::OffsetDateTime,
 }
 
+impl From<DatasetRange> for DateTimeRange {
+    fn from(DatasetRange { start, end }: DatasetRange) -> Self {
+        Self { start, end }
+    }
+}
+
 impl<'de> Deserialize<'de> for DatasetRange {
     fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
     where

src/historical/timeseries.rs

@@ -54,7 +54,7 @@ impl TimeseriesClient<'_> {
             )
             .await?;
         let mut decoder: AsyncDbnDecoder<_> = AsyncDbnDecoder::with_zstd_buffer(reader).await?;
-        decoder.set_upgrade_policy(params.upgrade_policy);
+        decoder.set_upgrade_policy(params.upgrade_policy)?;
         Ok(decoder)
     }
 
@@ -87,7 +87,7 @@ impl TimeseriesClient<'_> {
             )
             .await?;
         let mut http_decoder = AsyncDbnDecoder::with_zstd_buffer(reader).await?;
-        http_decoder.set_upgrade_policy(params.upgrade_policy);
+        http_decoder.set_upgrade_policy(params.upgrade_policy)?;
         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? {
@@ -132,7 +132,7 @@ impl TimeseriesClient<'_> {
             .await?
             .error_for_status()?
             .bytes_stream()
-            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e));
+            .map_err(std::io::Error::other);
         Ok(tokio_util::io::StreamReader::new(stream))
     }
 

src/live/client.rs

@@ -38,8 +38,8 @@ pub struct Client {
 }
 
 enum Decoder {
-    Metadata(AsyncMetadataDecoder<BufReader<ReadHalf<TcpStream>>>),
-    Record(AsyncRecordDecoder<BufReader<ReadHalf<TcpStream>>>),
+    Metadata(AsyncMetadataDecoder<ReadHalf<TcpStream>>),
+    Record(AsyncRecordDecoder<ReadHalf<TcpStream>>),
     Empty,
 }
 
@@ -107,7 +107,7 @@ impl Client {
             heartbeat_interval,
             protocol,
             peer_addr,
-            decoder: Decoder::Metadata(AsyncMetadataDecoder::new(recver)),
+            decoder: Decoder::Metadata(AsyncMetadataDecoder::new(recver.into_inner())),
             session_id,
             span,
             sub_counter: 0,
@@ -229,12 +229,7 @@ impl Client {
         info!("Starting session");
         self.protocol.start_session().await?;
         let mut metadata = decoder.decode().await?;
-        self.decoder = Decoder::Record(AsyncRecordDecoder::with_version(
-            decoder.into_inner(),
-            metadata.version,
-            self.upgrade_policy,
-            metadata.ts_out,
-        )?);
+        self.decoder = Decoder::Record(AsyncRecordDecoder::from(decoder));
         // Should match `send_ts_out` but set again here for safety
         metadata.upgrade(self.upgrade_policy);
         Ok(metadata)
@@ -299,7 +294,7 @@ impl Client {
                 self.heartbeat_interval.map(|i| i.whole_seconds()),
             )
             .await?;
-        self.decoder = Decoder::Metadata(AsyncMetadataDecoder::new(recver));
+        self.decoder = Decoder::Metadata(AsyncMetadataDecoder::new(recver.into_inner()));
         self.span = info_span!("LiveClient", dataset = %self.dataset, session_id = self.session_id);
         Ok(())
     }
@@ -413,7 +408,7 @@ mod tests {
             self.send("success=1|session_id=5\n").await;
         }
 
-        async fn subscribe(&mut self, subscription: Subscription) {
+        async fn subscribe(&mut self, subscription: Subscription, is_last: bool) {
             let sub_line = self.read_line().await;
             assert!(sub_line.contains(&format!("symbols={}", subscription.symbols.to_api_string())));
             assert!(sub_line.contains(&format!("schema={}", subscription.schema)));
@@ -423,6 +418,7 @@ mod tests {
                 assert!(sub_line.contains(&format!("start={}", start.unix_timestamp_nanos())))
             }
             assert!(sub_line.contains(&format!("snapshot={}", subscription.use_snapshot as u8)));
+            assert!(sub_line.contains(&format!("is_last={}", is_last as u8)));
         }
 
         async fn start(&mut self) {
@@ -489,7 +485,7 @@ mod tests {
         Accept,
         Authenticate(Option<Duration>),
         Send(String),
-        Subscribe(Subscription),
+        Subscribe(Subscription, bool),
         Start,
         SendRecord(Box<dyn AsRef<[u8]> + Send>),
         Disconnect,
@@ -502,7 +498,7 @@ mod tests {
                 Event::Accept => write!(f, "Accept"),
                 Event::Authenticate(hb_int) => write!(f, "Authenticate({hb_int:?})"),
                 Event::Send(msg) => write!(f, "Send({msg:?})"),
-                Event::Subscribe(sub) => write!(f, "Subscribe({sub:?})"),
+                Event::Subscribe(sub, is_last) => write!(f, "Subscribe({sub:?}, {is_last:?})"),
                 Event::Start => write!(f, "Start"),
                 Event::SendRecord(_) => write!(f, "SendRecord"),
                 Event::Disconnect => write!(f, "Disconnect"),
@@ -521,7 +517,7 @@ mod tests {
                         Some(Event::Authenticate(hb_int)) => mock.authenticate(hb_int).await,
                         Some(Event::Accept) => mock.accept().await,
                         Some(Event::Send(msg)) => mock.send(&msg).await,
-                        Some(Event::Subscribe(sub)) => mock.subscribe(sub).await,
+                        Some(Event::Subscribe(sub, is_last)) => mock.subscribe(sub, is_last).await,
                         Some(Event::Start) => mock.start().await,
                         Some(Event::SendRecord(rec)) => mock.send_record(rec).await,
                         Some(Event::Disconnect) => mock.close().await,
@@ -544,8 +540,10 @@ mod tests {
                 .unwrap();
         }
 
-        pub fn expect_subscribe(&mut self, subscription: Subscription) {
-            self.send.send(Event::Subscribe(subscription)).unwrap();
+        pub fn expect_subscribe(&mut self, subscription: Subscription, is_last: bool) {
+            self.send
+                .send(Event::Subscribe(subscription, is_last))
+                .unwrap();
         }
 
         pub fn start(&mut self) {
@@ -613,7 +611,7 @@ mod tests {
             .schema(Schema::Ohlcv1M)
             .stype_in(SType::RawSymbol)
             .build();
-        fixture.expect_subscribe(subscription.clone());
+        fixture.expect_subscribe(subscription.clone(), true);
         client.subscribe(subscription).await.unwrap();
         fixture.stop().await;
     }
@@ -628,7 +626,7 @@ mod tests {
             .stype_in(SType::RawSymbol)
             .use_snapshot()
             .build();
-        fixture.expect_subscribe(subscription.clone());
+        fixture.expect_subscribe(subscription.clone(), true);
         client.subscribe(subscription).await.unwrap();
         fixture.stop().await;
     }
@@ -670,7 +668,10 @@ mod tests {
         let mut i = 0;
         while i < SYMBOL_COUNT {
             let chunk_size = 500.min(SYMBOL_COUNT - i);
-            fixture.expect_subscribe(sub_base.clone().symbols(vec![SYMBOL; chunk_size]).build());
+            fixture.expect_subscribe(
+                sub_base.clone().symbols(vec![SYMBOL; chunk_size]).build(),
+                i + chunk_size == SYMBOL_COUNT,
+            );
             i += chunk_size;
         }
         fixture.stop().await;
@@ -823,7 +824,7 @@ mod tests {
             .schema(Schema::Trades)
             .start(OffsetDateTime::UNIX_EPOCH)
             .build();
-        fixture.expect_subscribe(sub.clone());
+        fixture.expect_subscribe(sub.clone(), true);
         client.subscribe(sub.clone()).await.unwrap();
         fixture.start();
         let metadata = client.start().await.unwrap();
@@ -863,7 +864,7 @@ mod tests {
 
         let mut resub = sub.clone();
         resub.start = None;
-        fixture.expect_subscribe(resub);
+        fixture.expect_subscribe(resub, true);
         client.resubscribe().await.unwrap();
         fixture.start();
         client.start().await.unwrap();

src/live/protocol.rs

@@ -134,14 +134,16 @@ where
         let start_nanos = sub.start.as_ref().map(|start| start.unix_timestamp_nanos());
 
         let symbol_chunks = sub.symbols.to_chunked_api_string();
-        for sym_str in symbol_chunks {
+        let last_chunk_idx = symbol_chunks.len() - 1;
+        for (i, sym_str) in symbol_chunks.into_iter().enumerate() {
             let sub_req = SubRequest::new(
                 *schema,
                 *stype_in,
                 start_nanos,
                 *use_snapshot,
                 sub.id,
                 &sym_str,
+                i == last_chunk_idx,
             );
             debug!(?sub_req, "Sending subscription request");
             self.sender.write_all(sub_req.as_bytes()).await?;
@@ -309,10 +311,12 @@ impl SubRequest {
         use_snapshot: bool,
         id: Option<u32>,
         symbols: &str,
+        is_last: bool,
     ) -> Self {
         let use_snapshot = use_snapshot as u8;
+        let is_last = is_last as u8;
         let mut args = format!(
-            "schema={schema}|stype_in={stype_in}|symbols={symbols}|snapshot={use_snapshot}"
+            "schema={schema}|stype_in={stype_in}|symbols={symbols}|snapshot={use_snapshot}|is_last={is_last}"
         );
 
         if let Some(start) = start_nanos {