chore: payload decoding tests (#287)

crates/derive/Cargo.toml

@@ -32,6 +32,8 @@ alloy-sol-types = { version = "0.7.6", default-features = false }
 hashbrown = "0.14.5"
 unsigned-varint = "0.8.0"
 miniz_oxide = "0.7.3"
+brotli = { version = "6.0.0", default-features = false }
+alloc-no-stdlib = "2.0.4"
 
 # `serde` feature dependencies
 serde = { version = "1.0.203", default-features = false, features = ["derive"], optional = true }

crates/derive/src/stages/channel_reader.rs

@@ -1,7 +1,7 @@
 //! This module contains the `ChannelReader` struct.
 
 use crate::{
-    stages::BatchQueueProvider,
+    stages::{decompress_brotli, BatchQueueProvider},
     traits::{OriginAdvancer, OriginProvider, PreviousStage, ResettableStage},
     types::{Batch, BlockInfo, RollupConfig, StageError, StageResult, SystemConfig},
 };
@@ -14,6 +14,15 @@ use core::fmt::Debug;
 use miniz_oxide::inflate::decompress_to_vec_zlib;
 use tracing::warn;
 
+/// 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;
+
 /// The [ChannelReader] provider trait.
 #[async_trait]
 pub trait ChannelReaderProvider {
@@ -148,16 +157,38 @@ impl BatchReader {
     /// 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() {
-            let decompressed_data = decompress_to_vec_zlib(&data).ok()?;
-            self.decompressed = decompressed_data;
+            // Peek at the data to determine the compression type.
+            if data.is_empty() {
+                tracing::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()?;
+            } else if compression_type == CHANNEL_VERSION_BROTLI {
+                brotli_used = true;
+                self.decompressed = decompress_brotli(&data).ok()?;
+            } else {
+                tracing::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 batch = Batch::decode(&mut bytes.as_ref(), cfg).unwrap();
 
+        // Confirm that brotli decompression was performed *after* the Fjord hardfork.
+        if brotli_used && !cfg.is_fjord_active(batch.timestamp()) {
+            tracing::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();
 

crates/derive/src/stages/mod.rs

@@ -36,5 +36,8 @@ pub use attributes_queue::{
     AttributesBuilder, AttributesProvider, AttributesQueue, StatefulAttributesBuilder,
 };
 
+mod utils;
+pub use utils::decompress_brotli;
+
 #[cfg(any(test, feature = "test-utils"))]
 pub mod test_utils;