chore(derive): remove batch reader (op-rs#826)

Author: refcell

Cargo.lock

@@ -106,7 +106,6 @@ os_pipe = "1.2.1"
 reqwest = "0.12.8"
 command-fds = "0.3.0"
 async-trait = "0.1.83"
-alloc-no-stdlib = "2.0.4"
 linked_list_allocator = "0.10.5"
 
 # General
@@ -121,10 +120,6 @@ tracing-loki = "0.2.5"
 tracing-subscriber = "0.3.18"
 tracing = { version = "0.1.40", default-features = false }
 
-# Encoding
-miniz_oxide = "0.8.0"
-brotli = { version = "7.0.0", default-features = false }
-
 # Testing
 pprof = "0.13.0"
 proptest = "1.5"

Cargo.toml

@@ -26,11 +26,8 @@ op-alloy-rpc-types-engine.workspace = true
 op-alloy-consensus = { workspace = true, features = ["k256"] }
 
 # General
-brotli.workspace = true
 tracing.workspace = true
-miniz_oxide.workspace = true
 async-trait.workspace = true
-alloc-no-stdlib.workspace = true
 thiserror.workspace = true
 
 # `serde` feature dependencies

crates/derive/Cargo.toml

@@ -460,9 +460,8 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{
-        stages::channel::channel_reader::BatchReader,
-        test_utils::{CollectingLayer, TestL2ChainProvider, TestNextBatchProvider, TraceStorage},
+    use crate::test_utils::{
+        CollectingLayer, TestL2ChainProvider, TestNextBatchProvider, TraceStorage,
     };
     use alloc::vec;
     use alloy_consensus::Header;
@@ -471,7 +470,7 @@ mod tests {
     use alloy_rlp::{BytesMut, Encodable};
     use op_alloy_consensus::{OpBlock, OpTxEnvelope, OpTxType, TxDeposit};
     use op_alloy_genesis::{ChainGenesis, MAX_RLP_BYTES_PER_CHANNEL_FJORD};
-    use op_alloy_protocol::{L1BlockInfoBedrock, L1BlockInfoTx};
+    use op_alloy_protocol::{BatchReader, L1BlockInfoBedrock, L1BlockInfoTx};
     use tracing::Level;
     use tracing_subscriber::layer::SubscriberExt;
 

crates/derive/src/stages/batch/batch_queue.rs

@@ -2,29 +2,19 @@
 
 use crate::{
     errors::PipelineError,
-    stages::{decompress_brotli, BatchStreamProvider},
+    stages::BatchStreamProvider,
     traits::{OriginAdvancer, OriginProvider, SignalReceiver},
     types::{PipelineResult, Signal},
 };
-use alloc::{boxed::Box, sync::Arc, vec::Vec};
+use alloc::{boxed::Box, sync::Arc};
 use alloy_primitives::Bytes;
-use alloy_rlp::Decodable;
 use async_trait::async_trait;
 use core::fmt::Debug;
-use miniz_oxide::inflate::decompress_to_vec_zlib;
 use op_alloy_genesis::{
     RollupConfig, MAX_RLP_BYTES_PER_CHANNEL_BEDROCK, MAX_RLP_BYTES_PER_CHANNEL_FJORD,
 };
-use op_alloy_protocol::{Batch, BlockInfo};
-
-/// ZLIB Deflate Compression Method.
-pub(crate) const ZLIB_DEFLATE_COMPRESSION_METHOD: u8 = 8;
-
-/// ZLIB Reserved Compression Info.
-pub(crate) const ZLIB_RESERVED_COMPRESSION_METHOD: u8 = 15;
-
-/// Brotili Compression Channel Version.
-pub(crate) const CHANNEL_VERSION_BROTLI: u8 = 1;
+use op_alloy_protocol::{Batch, BatchReader, BlockInfo};
+use tracing::{debug, warn};
 
 /// The [ChannelReader] provider trait.
 #[async_trait]
@@ -171,88 +161,6 @@ where
     }
 }
 
-/// Batch Reader provides a function that iteratively consumes batches from the reader.
-/// The L1Inclusion block is also provided at creation time.
-/// Warning: the batch reader can read every batch-type.
-/// The caller of the batch-reader should filter the results.
-#[derive(Debug)]
-pub(crate) struct BatchReader {
-    /// The raw data to decode.
-    data: Option<Vec<u8>>,
-    /// Decompressed data.
-    decompressed: Vec<u8>,
-    /// The current cursor in the `decompressed` data.
-    cursor: usize,
-    /// The maximum RLP bytes per channel.
-    max_rlp_bytes_per_channel: usize,
-}
-
-impl BatchReader {
-    /// Creates a new [BatchReader] from the given data and max decompressed RLP bytes per channel.
-    pub(crate) fn new<T>(data: T, max_rlp_bytes_per_channel: usize) -> Self
-    where
-        T: Into<Vec<u8>>,
-    {
-        Self {
-            data: Some(data.into()),
-            decompressed: Vec::new(),
-            cursor: 0,
-            max_rlp_bytes_per_channel,
-        }
-    }
-
-    /// Pulls out the next batch from the reader.
-    pub(crate) fn next_batch(&mut self, cfg: &RollupConfig) -> Option<Batch> {
-        // If the data is not already decompressed, decompress it.
-        let mut brotli_used = false;
-
-        if let Some(data) = self.data.take() {
-            // Peek at the data to determine the compression type.
-            if data.is_empty() {
-                warn!(target: "batch-reader", "Data is too short to determine compression type, skipping batch");
-                return None;
-            }
-
-            let compression_type = data[0];
-            if (compression_type & 0x0F) == ZLIB_DEFLATE_COMPRESSION_METHOD ||
-                (compression_type & 0x0F) == ZLIB_RESERVED_COMPRESSION_METHOD
-            {
-                self.decompressed = decompress_to_vec_zlib(&data).ok()?;
-
-                // Check the size of the decompressed channel RLP.
-                if self.decompressed.len() > self.max_rlp_bytes_per_channel {
-                    return None;
-                }
-            } else if compression_type == CHANNEL_VERSION_BROTLI {
-                brotli_used = true;
-                self.decompressed =
-                    decompress_brotli(&data[1..], self.max_rlp_bytes_per_channel).ok()?;
-            } else {
-                error!(target: "batch-reader", "Unsupported compression type: {:x}, skipping batch", compression_type);
-                return None;
-            }
-        }
-
-        // Decompress and RLP decode the batch data, before finally decoding the batch itself.
-        let decompressed_reader = &mut self.decompressed.as_slice()[self.cursor..].as_ref();
-        let bytes = Bytes::decode(decompressed_reader).ok()?;
-        let Ok(batch) = Batch::decode(&mut bytes.as_ref(), cfg) else {
-            error!(target: "batch-reader", "Failed to decode batch, skipping batch");
-            return None;
-        };
-
-        // Confirm that brotli decompression was performed *after* the Fjord hardfork.
-        if brotli_used && !cfg.is_fjord_active(batch.timestamp()) {
-            warn!(target: "batch-reader", "Brotli compression used before Fjord hardfork, skipping batch");
-            return None;
-        }
-
-        // Advance the cursor on the reader.
-        self.cursor = self.decompressed.len() - decompressed_reader.len();
-        Some(batch)
-    }
-}
-
 #[cfg(test)]
 mod test {
     use super::*;
@@ -334,24 +242,6 @@ mod test {
         assert!(reader.next_batch.is_some());
     }
 
-    #[test]
-    fn test_batch_reader() {
-        let raw = new_compressed_batch_data();
-        let decompressed_len = decompress_to_vec_zlib(&raw).unwrap().len();
-        let mut reader = BatchReader::new(raw, MAX_RLP_BYTES_PER_CHANNEL_BEDROCK as usize);
-        reader.next_batch(&RollupConfig::default()).unwrap();
-        assert_eq!(reader.cursor, decompressed_len);
-    }
-
-    #[test]
-    fn test_batch_reader_fjord() {
-        let raw = new_compressed_batch_data();
-        let decompressed_len = decompress_to_vec_zlib(&raw).unwrap().len();
-        let mut reader = BatchReader::new(raw, MAX_RLP_BYTES_PER_CHANNEL_FJORD as usize);
-        reader.next_batch(&RollupConfig { fjord_time: Some(0), ..Default::default() }).unwrap();
-        assert_eq!(reader.cursor, decompressed_len);
-    }
-
     #[tokio::test]
     async fn test_flush_post_holocene() {
         let raw = new_compressed_batch_data();

crates/derive/src/stages/channel/channel_reader.rs

@@ -37,6 +37,3 @@ pub use batch::{
 
 mod attributes_queue;
 pub use attributes_queue::AttributesQueue;
-
-mod utils;
-pub use utils::decompress_brotli;