VER: Release 0.35.1

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## 0.35.1 - 2025-06-03
+
+### Enhancements
+- Documented `AsyncMetadataDecoder::decode`, `AsyncDecoder::new`, and
+  `AsyncDecoder::with_upgrade_policy()` as now being cancellation safe
+- Added `DbnFsm::reset()` method that resets state to facilitate reuse
+- Added `DbnFsm::has_decoded_metadata()` method to check the internal
+  state
+
+### Bug fixes
+- Fixed behavior where encoding metadata could lower the `version`
+- Changed `DbnFsm::data()` to exclude all processed data
+- Fixed `Metadata::upgrade()` behavior with `UpgradeToV2`
+
 ## 0.35.0 - 2025-05-28
 This version marks the release of DBN version 3 (DBNv3), which is the new default.
 Decoders support decoding all versions of DBN and the DBN encoders default to

Cargo.lock

@@ -11,7 +11,7 @@ resolver = "2"
 [workspace.package]
 authors = ["Databento <[email protected]>"]
 edition = "2021"
-version = "0.35.0"
+version = "0.35.1"
 documentation = "https://databento.com/docs"
 repository = "https://github.com/databento/dbn"
 license = "Apache-2.0"

Cargo.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "databento-dbn"
-version = "0.35.0"
+version = "0.35.1"
 description = "Python bindings for encoding and decoding Databento Binary Encoding (DBN)"
 authors = ["Databento <[email protected]>"]
 license = "Apache-2.0"
@@ -17,7 +17,7 @@ build-backend = "maturin"
 
 [project]
 name = "databento-dbn"
-version = "0.35.0"
+version = "0.35.1"
 authors = [
     { name = "Databento", email = "[email protected]" }
 ]

python/pyproject.toml

@@ -16,7 +16,7 @@ name = "dbn"
 path = "src/main.rs"
 
 [dependencies]
-dbn = { path = "../dbn", version = "=0.35.0", default-features = false }
+dbn = { path = "../dbn", version = "=0.35.1", default-features = false }
 
 anyhow = { workspace = true }
 clap = { version = "4.5", features = ["derive", "wrap_help"] }

rust/dbn-cli/Cargo.toml

@@ -25,7 +25,7 @@ serde = ["dep:serde", "time/parsing", "time/serde"]
 trivial_copy = []
 
 [dependencies]
-dbn-macros = { version = "=0.35.0", path = "../dbn-macros" }
+dbn-macros = { version = "=0.35.1", path = "../dbn-macros" }
 
 async-compression = { version = "0.4.23", features = ["tokio", "zstd"], optional = true }
 csv = { workspace = true }

rust/dbn/Cargo.toml

@@ -16,6 +16,7 @@ pub mod fsm;
 mod r#async;
 #[cfg(feature = "async")]
 pub use r#async::{
-    Decoder as AsyncDecoder, MetadataDecoder as AsyncMetadataDecoder,
-    RecordDecoder as AsyncRecordDecoder,
+    decode_metadata_with_fsm as async_decode_metadata_with_fsm,
+    decode_record_ref_with_fsm as async_decode_record_ref_with_fsm, Decoder as AsyncDecoder,
+    MetadataDecoder as AsyncMetadataDecoder, RecordDecoder as AsyncRecordDecoder,
 };

rust/dbn/src/decode/dbn.rs

@@ -37,9 +37,8 @@ where
     /// `reader` or the input is encoded in a newer version of DBN.
     ///
     /// # Cancel safety
-    /// This method is not cancellation safe. If this method is used in a
-    /// `tokio::select!` statement and another branch completes first, the metadata
-    /// may have been partially read, corrupting the stream.
+    /// This method is cancel safe. It can be used within a `tokio::select!` statement
+    /// without the potential for corrupting the input stream.
     pub async fn new(reader: R) -> crate::Result<Self> {
         let mut metadata_decoder = MetadataDecoder::new(reader);
         let mut metadata = metadata_decoder.decode().await?;
@@ -58,9 +57,8 @@ where
     /// `reader` or the input is encoded in a newer version of DBN.
     ///
     /// # Cancel safety
-    /// This method is not cancellation safe. If this method is used in a
-    /// `tokio::select!` statement and another branch completes first, the metadata
-    /// may have been partially read, corrupting the stream.
+    /// This method is cancel safe. It can be used within a `tokio::select!` statement
+    /// without the potential for corrupting the input stream.
     pub async fn with_upgrade_policy(
         reader: R,
         upgrade_policy: VersionUpgradePolicy,
@@ -102,7 +100,7 @@ where
     /// incompatible.
     pub fn set_upgrade_policy(&mut self, upgrade_policy: VersionUpgradePolicy) -> Result<()> {
         self.decoder.set_upgrade_policy(upgrade_policy)?;
-        self.metadata.set_version(upgrade_policy);
+        self.metadata.upgrade(upgrade_policy);
         Ok(())
     }
 
@@ -293,6 +291,12 @@ where
         Ok(Self { reader, fsm })
     }
 
+    /// Creates a new `RecordDecoder` that will decode from `reader` and use the state
+    /// and buffered data already in `fsm`.
+    pub fn with_fsm(reader: R, fsm: DbnFsm) -> Self {
+        Self { reader, fsm }
+    }
+
     /// Sets the DBN version to expect when decoding.
     ///
     /// # Errors
@@ -496,45 +500,23 @@ where
         Self { reader, fsm }
     }
 
+    /// Creates a new `MetadataDecoder` that will decode from `reader` and use the state
+    /// and buffered data already in `fsm`.
+    pub fn with_fsm(reader: R, fsm: DbnFsm) -> Self {
+        Self { reader, fsm }
+    }
+
     /// Decodes and returns a DBN [`Metadata`].
     ///
     /// # Errors
     /// This function will return an error if it is unable to parse the metadata or the
     /// input is encoded in a newere version of DBN.
     ///
     /// # Cancel safety
-    /// This method is not cancellation safe. If this method is used in a
-    /// `tokio::select!` statement and another branch completes first, the metadata
-    /// may have been partially read, corrupting the stream.
+    /// This method is cancel safe. It can be used within a `tokio::select!` statement
+    /// without the potential for corrupting the input stream.
     pub async fn decode(&mut self) -> Result<Metadata> {
-        let io_err = |err| crate::Error::io(err, "decoding metadata");
-        let nbytes = self.reader.read(self.fsm.space()).await.map_err(io_err)?;
-        self.fsm.fill(nbytes);
-        loop {
-            match self.fsm.process() {
-                ProcessResult::ReadMore(n) => {
-                    // asm guarantees there's at least `n` bytes available in `space()`
-                    let mut total_read = 0;
-                    loop {
-                        let read = self.reader.read(self.fsm.space()).await.map_err(io_err)?;
-                        if read == 0 {
-                            return Err(crate::Error::io(
-                                io::Error::from(io::ErrorKind::UnexpectedEof),
-                                "decoding metadata",
-                            ));
-                        }
-                        self.fsm.fill(read);
-                        total_read += read;
-                        if total_read >= n {
-                            break;
-                        }
-                    }
-                }
-                ProcessResult::Metadata(metadata) => return Ok(metadata),
-                ProcessResult::Record(_) => unreachable!("metadata precedes records"),
-                ProcessResult::Err(error) => return Err(error),
-            }
-        }
+        decode_metadata_with_fsm(&mut self.reader, &mut self.fsm).await
     }
 
     /// Returns a mutable reference to the inner reader.
@@ -568,6 +550,99 @@ where
     }
 }
 
+/// A low-level function for decoding DBN [`Metadata`]. Generally, the [`Decoder`] or
+/// [`MetadataDecoder`] should be used instead of this function.
+///
+/// # Errors
+/// This function will return an error if it is unable to parse the metadata or the
+/// input is encoded in a newere version of DBN.
+///
+/// # Cancel safety
+/// This method is cancel safe. It can be used within a `tokio::select!` statement
+/// without the potential for corrupting the input stream.
+///
+/// # Panics
+/// This function will panic if it encounters DBN records. The caller the initial data
+/// from `reader` contains DBN metadata.
+pub async fn decode_metadata_with_fsm<R>(mut reader: R, fsm: &mut DbnFsm) -> crate::Result<Metadata>
+where
+    R: io::AsyncReadExt + Unpin,
+{
+    let io_err = |err| crate::Error::io(err, "decoding metadata");
+    let nbytes = reader.read(fsm.space()).await.map_err(io_err)?;
+    fsm.fill(nbytes);
+    loop {
+        match fsm.process() {
+            ProcessResult::ReadMore(n) => {
+                // asm guarantees there's at least `n` bytes available in `space()`
+                let mut total_read = 0;
+                loop {
+                    let read = reader.read(fsm.space()).await.map_err(io_err)?;
+                    if read == 0 {
+                        return Err(crate::Error::io(
+                            io::Error::from(io::ErrorKind::UnexpectedEof),
+                            "decoding metadata",
+                        ));
+                    }
+                    fsm.fill(read);
+                    total_read += read;
+                    if total_read >= n {
+                        break;
+                    }
+                }
+            }
+            ProcessResult::Metadata(metadata) => return Ok(metadata),
+            ProcessResult::Record(_) => panic!("DBN metadata should precede records"),
+            ProcessResult::Err(error) => return Err(error),
+        }
+    }
+}
+
+/// A low-level function for decoding the next [`RecordRef`]. Returns `None` if `reader`
+/// has been exhausted.
+///
+/// Generally [`Decoder`] and [`RecordDecoder`] should be used instead of this function.
+///
+/// # Errors
+/// This function returns an error if the underlying reader returns an
+/// error of a kind other than `io::ErrorKind::UnexpectedEof` upon reading.
+/// It will also return an error if it encounters an invalid record.
+///
+/// # Cancel safety
+/// This method is cancel safe. It can be used within a `tokio::select!` statement
+/// without the potential for corrupting the input stream.
+///
+/// # Panics
+/// This function will panic if it encounters DBN metadata. The caller must ensure
+/// the metadata has already been decoded.
+pub async fn decode_record_ref_with_fsm<'a, R>(
+    mut reader: R,
+    fsm: &'a mut DbnFsm,
+) -> Result<Option<RecordRef<'a>>>
+where
+    R: io::AsyncReadExt + Unpin,
+{
+    loop {
+        match fsm.process() {
+            ProcessResult::ReadMore(_) => match reader.read(fsm.space()).await {
+                Ok(0) => return Ok(None),
+                Ok(nbytes) => {
+                    fsm.fill(nbytes);
+                }
+                Err(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => {
+                    return Ok(None);
+                }
+                Err(err) => {
+                    return Err(crate::Error::io(err, "decoding record reference"));
+                }
+            },
+            ProcessResult::Record(_) => return Ok(fsm.last_record()),
+            ProcessResult::Err(error) => return Err(error),
+            ProcessResult::Metadata(_) => panic!("found DBN metadata when expected records"),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     #![allow(clippy::clone_on_copy)]
@@ -770,7 +845,7 @@ mod tests {
             VersionUpgradePolicy::UpgradeToV2,
         )
         .await?;
-        assert_eq!(decoder.metadata().version, crate::DBN_VERSION);
+        assert_eq!(decoder.metadata().version, 2);
         assert_eq!(decoder.metadata().symbol_cstr_len, crate::SYMBOL_CSTR_LEN);
         let mut has_decoded = false;
         while let Some(_rec) = decoder.decode_record::<v2::InstrumentDefMsg>().await? {