apply review

Author: lightsing

crates/codec/Cargo.toml

@@ -20,7 +20,7 @@ derive_more = { version = "2.0", default-features = false }
 eyre = { workspace = true, optional = true }
 thiserror = { version = "2.0", default-features = false }
 zstd = { version = "=0.13.3", optional = true }
-ruzstd = "0.8"
+ruzstd = { version = "0.8", optional = true }
 
 [dev-dependencies]
 eyre.workspace = true
@@ -30,3 +30,4 @@ serde_json = "1.0"
 default = ["zstd"]
 test-utils = ["dep:eyre", "scroll-l1/test-utils"]
 zstd = ["dep:zstd"]
+ruzstd = ["dep:ruzstd"]

crates/codec/src/decoding/v2/mod.rs

@@ -35,7 +35,7 @@ pub fn decode_v2(calldata: &[u8], blob: &[u8]) -> Result<Batch, DecodingError> {
 
     // get blob iterator and collect, skipping unused bytes.
     let compressed_heap_blob = BlobSliceIter::from_blob_slice(blob).copied().collect::<Vec<_>>();
-    let uncompressed_heap_blob = decompress_blob_data(&compressed_heap_blob);
+    let uncompressed_heap_blob = decompress_blob_data(&compressed_heap_blob)?;
     let buf = &mut (uncompressed_heap_blob.as_slice());
 
     // check buf len.

crates/codec/src/decoding/v2/zstd.rs

@@ -2,22 +2,46 @@
 
 use std::io::Read;
 
-#[cfg(feature = "zstd")]
-use ruzstd as _;
+#[cfg(not(any(feature = "zstd", feature = "ruzstd")))]
+compile_error!("You must enable exactly one of the `zstd` or `ruzstd` features");
+#[cfg(all(feature = "zstd", feature = "ruzstd"))]
+compile_error!("Features `zstd` and `ruzstd` are mutually exclusive");
 
 /// The ZSTD magic number for zstd compressed data header.
 const ZSTD_MAGIC_NUMBER: [u8; 4] = [0x28, 0xb5, 0x2f, 0xfd];
 
+/// Result type for Zstd operations.
+type Result<T> = std::result::Result<T, ZstdError>;
+
+/// Zstd error type.
+#[derive(Debug)]
+pub struct ZstdError(Box<dyn std::error::Error + Send + Sync>);
+
+impl std::fmt::Display for ZstdError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "ZstdError: {}", self.0)
+    }
+}
+
+impl std::error::Error for ZstdError {}
+
+impl ZstdError {
+    /// Consumes the error and returns the inner error.
+    pub fn into_inner(self) -> Box<dyn std::error::Error + Send + Sync> {
+        self.0
+    }
+}
+
 /// Uncompress the provided data.
 #[cfg(feature = "zstd")]
-pub fn decompress_blob_data(data: &[u8]) -> Vec<u8> {
+pub fn decompress_blob_data(data: &[u8]) -> Result<Vec<u8>> {
     use zstd::Decoder;
     let mut header_data = ZSTD_MAGIC_NUMBER.to_vec();
 
     header_data.extend_from_slice(data);
 
     // init decoder and owned output data.
-    let mut decoder = Decoder::new(header_data.as_slice()).unwrap();
+    let mut decoder = Decoder::new(header_data.as_slice()).map_err(|e| ZstdError(Box::new(e)))?;
     // heuristic: use data length as the allocated output capacity.
     let mut output = Vec::with_capacity(header_data.len());
 
@@ -33,22 +57,23 @@ pub fn decompress_blob_data(data: &[u8]) -> Vec<u8> {
         output.extend_from_slice(&dst[..size]);
     }
 
-    output
+    Ok(output)
 }
 
 /// Uncompress the provided data.
-#[cfg(not(feature = "zstd"))]
-pub fn decompress_blob_data(data: &[u8]) -> Vec<u8> {
+#[cfg(feature = "ruzstd")]
+pub fn decompress_blob_data(data: &[u8]) -> Result<Vec<u8>> {
     use ruzstd::decoding::StreamingDecoder;
 
     let mut header_data = ZSTD_MAGIC_NUMBER.to_vec();
     header_data.extend_from_slice(data);
 
     // init decoder and owned output data.
-    let mut decoder = StreamingDecoder::new(header_data.as_slice()).unwrap();
+    let mut decoder =
+        StreamingDecoder::new(header_data.as_slice()).map_err(|e| ZstdError(Box::new(e)))?;
     // heuristic: use data length as the allocated output capacity.
     let mut output = Vec::with_capacity(header_data.len());
-    decoder.read_to_end(&mut output).unwrap();
+    decoder.read_to_end(&mut output).map_err(|e| ZstdError(Box::new(e)))?;
 
-    output
+    Ok(output)
 }

crates/codec/src/decoding/v4/mod.rs

@@ -31,7 +31,7 @@ pub fn decode_v4(calldata: &[u8], blob: &[u8]) -> Result<Batch, DecodingError> {
     debug_assert!(is_compressed == 1 || is_compressed == 0, "incorrect compressed byte flag");
 
     let buf = if is_compressed == 1 {
-        heap_blob = decompress_blob_data(&heap_blob[1..]);
+        heap_blob = decompress_blob_data(&heap_blob[1..])?;
         &mut heap_blob.as_slice()
     } else {
         &mut (&heap_blob[1..])

crates/codec/src/decoding/v7/mod.rs

@@ -49,7 +49,7 @@ pub fn decode_v7(blob: &[u8]) -> Result<Batch, DecodingError> {
 
     // uncompress if necessary.
     let buf = if is_compressed == 1 {
-        heap_blob = decompress_blob_data(buf);
+        heap_blob = decompress_blob_data(buf)?;
         &mut heap_blob.as_slice()
     } else {
         buf

crates/codec/src/error.rs

@@ -30,6 +30,8 @@ pub enum DecodingError {
     InvalidCommitBatchCall(#[from] InvalidCommitBatchCall),
     #[error("end of file")]
     Eof,
+    #[error("zstd decompression error occurred: {0}")]
+    ZstdDecompression(Box<dyn std::error::Error + Send + Sync + 'static>),
     #[error("decoding error occurred: {0}")]
     Other(Box<dyn std::error::Error + Send + Sync + 'static>),
 }
@@ -46,3 +48,9 @@ impl From<String> for DecodingError {
         DecodingError::Other(value.into())
     }
 }
+
+impl From<crate::decoding::v2::zstd::ZstdError> for DecodingError {
+    fn from(e: crate::decoding::v2::zstd::ZstdError) -> Self {
+        DecodingError::ZstdDecompression(e.into_inner())
+    }
+}