feat(protocol): Batch

crates/protocol/src/batch/bits.rs

@@ -16,6 +16,11 @@ impl AsRef<[u8]> for SpanBatchBits {
 }
 
 impl SpanBatchBits {
+    /// Creates a new span batch bits.
+    pub const fn new(inner: Vec<u8>) -> Self {
+        Self(inner)
+    }
+
     /// Decodes a standard span-batch bitlist from a reader.
     /// The bitlist is encoded as big-endian integer, left-padded with zeroes to a multiple of 8
     /// bits. The encoded bitlist cannot be longer than `bit_length`.

crates/protocol/src/batch/core.rs

@@ -0,0 +1,51 @@
+//! Module containing the core [Batch] enum.
+
+use crate::{BatchDecodingError, BatchType, RawSpanBatch, SingleBatch, SpanBatch};
+use alloy_rlp::{Buf, Decodable};
+use op_alloy_genesis::RollupConfig;
+
+/// A Batch.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[allow(clippy::large_enum_variant)]
+pub enum Batch {
+    /// A single batch
+    Single(SingleBatch),
+    /// Span Batches
+    Span(SpanBatch),
+}
+
+impl Batch {
+    /// Returns the timestamp for the batch.
+    pub fn timestamp(&self) -> u64 {
+        match self {
+            Self::Single(sb) => sb.timestamp,
+            Self::Span(sb) => sb.starting_timestamp(),
+        }
+    }
+
+    /// Attempts to decode a batch from a reader.
+    pub fn decode(r: &mut &[u8], cfg: &RollupConfig) -> Result<Self, BatchDecodingError> {
+        if r.is_empty() {
+            return Err(BatchDecodingError::EmptyBuffer);
+        }
+
+        // Read the batch type
+        let batch_type = BatchType::from(r[0]);
+        r.advance(1);
+
+        match batch_type {
+            BatchType::Single => {
+                let single_batch =
+                    SingleBatch::decode(r).map_err(BatchDecodingError::AlloyRlpError)?;
+                Ok(Self::Single(single_batch))
+            }
+            BatchType::Span => {
+                let mut raw_span_batch = RawSpanBatch::decode(r)?;
+                let span_batch = raw_span_batch
+                    .derive(cfg.block_time, cfg.genesis.l2_time, cfg.l2_chain_id)
+                    .map_err(BatchDecodingError::SpanBatchError)?;
+                Ok(Self::Span(span_batch))
+            }
+        }
+    }
+}

crates/protocol/src/batch/errors.rs

@@ -35,6 +35,41 @@ impl core::error::Error for SpanBatchError {
     }
 }
 
+/// An error decoding a batch.
+#[derive(Debug, derive_more::Display, Clone, PartialEq, Eq)]
+pub enum BatchDecodingError {
+    /// Empty buffer
+    #[display("Empty buffer")]
+    EmptyBuffer,
+    /// Error decoding an Alloy RLP
+    #[display("Error decoding an Alloy RLP: {_0}")]
+    AlloyRlpError(alloy_rlp::Error),
+    /// Error decoding a span batch
+    #[display("Error decoding a span batch: {_0}")]
+    SpanBatchError(SpanBatchError),
+}
+
+impl From<alloy_rlp::Error> for BatchDecodingError {
+    fn from(err: alloy_rlp::Error) -> Self {
+        Self::AlloyRlpError(err)
+    }
+}
+
+impl From<SpanBatchError> for BatchDecodingError {
+    fn from(err: SpanBatchError) -> Self {
+        Self::SpanBatchError(err)
+    }
+}
+
+impl core::error::Error for BatchDecodingError {
+    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
+        match self {
+            Self::SpanBatchError(err) => Some(err),
+            _ => None,
+        }
+    }
+}
+
 /// Decoding Error
 #[derive(Debug, derive_more::Display, Clone, PartialEq, Eq)]
 pub enum SpanDecodingError {

crates/protocol/src/batch/inclusion.rs

@@ -0,0 +1,44 @@
+//! Module containing the [BatchWithInclusionBlock] struct.
+
+use crate::{Batch, BatchValidationProvider, BatchValidity, BlockInfo, L2BlockInfo};
+use op_alloy_genesis::RollupConfig;
+
+/// A batch with its inclusion block.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BatchWithInclusionBlock {
+    /// The inclusion block
+    pub inclusion_block: BlockInfo,
+    /// The batch
+    pub batch: Batch,
+}
+
+impl BatchWithInclusionBlock {
+    /// Creates a new batch with inclusion block.
+    pub const fn new(inclusion_block: BlockInfo, batch: Batch) -> Self {
+        Self { inclusion_block, batch }
+    }
+
+    /// Validates the batch can be applied on top of the specified L2 safe head.
+    /// The first entry of the l1_blocks should match the origin of the l2_safe_head.
+    /// One or more consecutive l1_blocks should be provided.
+    /// In case of only a single L1 block, the decision whether a batch is valid may have to stay
+    /// undecided.
+    pub async fn check_batch<BF: BatchValidationProvider>(
+        &self,
+        cfg: &RollupConfig,
+        l1_blocks: &[BlockInfo],
+        l2_safe_head: L2BlockInfo,
+        fetcher: &mut BF,
+    ) -> BatchValidity {
+        match &self.batch {
+            Batch::Single(single_batch) => {
+                single_batch.check_batch(cfg, l1_blocks, l2_safe_head, &self.inclusion_block)
+            }
+            Batch::Span(span_batch) => {
+                span_batch
+                    .check_batch(cfg, l1_blocks, l2_safe_head, &self.inclusion_block, fetcher)
+                    .await
+            }
+        }
+    }
+}

crates/protocol/src/batch/mod.rs

@@ -3,8 +3,23 @@
 mod r#type;
 pub use r#type::*;
 
+mod core;
+pub use core::Batch;
+
+mod raw;
+pub use raw::RawSpanBatch;
+
+mod payload;
+pub use payload::SpanBatchPayload;
+
+mod prefix;
+pub use prefix::SpanBatchPrefix;
+
+mod inclusion;
+pub use inclusion::BatchWithInclusionBlock;
+
 mod errors;
-pub use errors::{SpanBatchError, SpanDecodingError};
+pub use errors::{BatchDecodingError, SpanBatchError, SpanDecodingError};
 
 mod bits;
 pub use bits::SpanBatchBits;

crates/protocol/src/batch/payload.rs

@@ -0,0 +1,196 @@
+//! Raw Span Batch Payload
+
+use super::MAX_SPAN_BATCH_ELEMENTS;
+use crate::{SpanBatchBits, SpanBatchError, SpanBatchTransactions, SpanDecodingError};
+use alloc::vec::Vec;
+
+/// Span Batch Payload
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct SpanBatchPayload {
+    /// Number of L2 block in the span
+    pub block_count: u64,
+    /// Standard span-batch bitlist of blockCount bits. Each bit indicates if the L1 origin is
+    /// changed at the L2 block.
+    pub origin_bits: SpanBatchBits,
+    /// List of transaction counts for each L2 block
+    pub block_tx_counts: Vec<u64>,
+    /// Transactions encoded in SpanBatch specs
+    pub txs: SpanBatchTransactions,
+}
+
+impl SpanBatchPayload {
+    /// Decodes a [SpanBatchPayload] from a reader.
+    pub fn decode_payload(r: &mut &[u8]) -> Result<Self, SpanBatchError> {
+        let mut payload = Self::default();
+        payload.decode_block_count(r)?;
+        payload.decode_origin_bits(r)?;
+        payload.decode_block_tx_counts(r)?;
+        payload.decode_txs(r)?;
+        Ok(payload)
+    }
+
+    /// Encodes a [SpanBatchPayload] into a writer.
+    pub fn encode_payload(&self, w: &mut Vec<u8>) -> Result<(), SpanBatchError> {
+        self.encode_block_count(w);
+        self.encode_origin_bits(w)?;
+        self.encode_block_tx_counts(w);
+        self.encode_txs(w)
+    }
+
+    /// Decodes the origin bits from a reader.
+    pub fn decode_origin_bits(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        if self.block_count > MAX_SPAN_BATCH_ELEMENTS {
+            return Err(SpanBatchError::TooBigSpanBatchSize);
+        }
+
+        self.origin_bits = SpanBatchBits::decode(r, self.block_count as usize)?;
+        Ok(())
+    }
+
+    /// Decode a block count from a reader.
+    pub fn decode_block_count(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        let (block_count, remaining) = unsigned_varint::decode::u64(r)
+            .map_err(|_| SpanBatchError::Decoding(SpanDecodingError::BlockCount))?;
+        // The number of transactions in a single L2 block cannot be greater than
+        // [MAX_SPAN_BATCH_ELEMENTS].
+        if block_count > MAX_SPAN_BATCH_ELEMENTS {
+            return Err(SpanBatchError::TooBigSpanBatchSize);
+        }
+        if block_count == 0 {
+            return Err(SpanBatchError::EmptySpanBatch);
+        }
+        self.block_count = block_count;
+        *r = remaining;
+        Ok(())
+    }
+
+    /// Decode block transaction counts from a reader.
+    pub fn decode_block_tx_counts(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        // Initially allocate the vec with the block count, to reduce re-allocations in the first
+        // few blocks.
+        let mut block_tx_counts = Vec::with_capacity(self.block_count as usize);
+
+        for _ in 0..self.block_count {
+            let (block_tx_count, remaining) = unsigned_varint::decode::u64(r)
+                .map_err(|_| SpanBatchError::Decoding(SpanDecodingError::BlockTxCounts))?;
+
+            // The number of transactions in a single L2 block cannot be greater than
+            // [MAX_SPAN_BATCH_ELEMENTS].
+            if block_tx_count > MAX_SPAN_BATCH_ELEMENTS {
+                return Err(SpanBatchError::TooBigSpanBatchSize);
+            }
+            block_tx_counts.push(block_tx_count);
+            *r = remaining;
+        }
+        self.block_tx_counts = block_tx_counts;
+        Ok(())
+    }
+
+    /// Decode transactions from a reader.
+    pub fn decode_txs(&mut self, r: &mut &[u8]) -> Result<(), SpanBatchError> {
+        if self.block_tx_counts.is_empty() {
+            return Err(SpanBatchError::EmptySpanBatch);
+        }
+
+        let total_block_tx_count =
+            self.block_tx_counts.iter().try_fold(0u64, |acc, block_tx_count| {
+                acc.checked_add(*block_tx_count).ok_or(SpanBatchError::TooBigSpanBatchSize)
+            })?;
+
+        // The total number of transactions in a span batch cannot be greater than
+        // [MAX_SPAN_BATCH_ELEMENTS].
+        if total_block_tx_count > MAX_SPAN_BATCH_ELEMENTS {
+            return Err(SpanBatchError::TooBigSpanBatchSize);
+        }
+        self.txs.total_block_tx_count = total_block_tx_count;
+        self.txs.decode(r)?;
+        Ok(())
+    }
+
+    /// Encode the origin bits into a writer.
+    pub fn encode_origin_bits(&self, w: &mut Vec<u8>) -> Result<(), SpanBatchError> {
+        SpanBatchBits::encode(w, self.block_count as usize, &self.origin_bits)
+    }
+
+    /// Encode the block count into a writer.
+    pub fn encode_block_count(&self, w: &mut Vec<u8>) {
+        let mut u64_varint_buf = [0u8; 10];
+        w.extend_from_slice(unsigned_varint::encode::u64(self.block_count, &mut u64_varint_buf));
+    }
+
+    /// Encode the block transaction counts into a writer.
+    pub fn encode_block_tx_counts(&self, w: &mut Vec<u8>) {
+        let mut u64_varint_buf = [0u8; 10];
+        for block_tx_count in &self.block_tx_counts {
+            u64_varint_buf.fill(0);
+            w.extend_from_slice(unsigned_varint::encode::u64(*block_tx_count, &mut u64_varint_buf));
+        }
+    }
+
+    /// Encode the transactions into a writer.
+    pub fn encode_txs(&self, w: &mut Vec<u8>) -> Result<(), SpanBatchError> {
+        self.txs.encode(w)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloc::vec;
+
+    #[test]
+    fn test_decode_origin_bits() {
+        let block_count = 10;
+        let encoded = vec![2; block_count / 8 + 1];
+        let mut payload =
+            SpanBatchPayload { block_count: block_count as u64, ..Default::default() };
+        payload.decode_origin_bits(&mut encoded.as_slice()).unwrap();
+        assert_eq!(payload.origin_bits, SpanBatchBits::new(vec![2; block_count / 8 + 1]));
+    }
+
+    #[test]
+    fn test_zero_block_count() {
+        let mut u64_varint_buf = [0; 10];
+        let mut encoded = unsigned_varint::encode::u64(0, &mut u64_varint_buf);
+        let mut payload = SpanBatchPayload::default();
+        let err = payload.decode_block_count(&mut encoded).unwrap_err();
+        assert_eq!(err, SpanBatchError::EmptySpanBatch);
+    }
+
+    #[test]
+    fn test_decode_block_count() {
+        let block_count = MAX_SPAN_BATCH_ELEMENTS;
+        let mut u64_varint_buf = [0; 10];
+        let mut encoded = unsigned_varint::encode::u64(block_count, &mut u64_varint_buf);
+        let mut payload = SpanBatchPayload::default();
+        payload.decode_block_count(&mut encoded).unwrap();
+        assert_eq!(payload.block_count, block_count);
+    }
+
+    #[test]
+    fn test_decode_block_count_errors() {
+        let block_count = MAX_SPAN_BATCH_ELEMENTS + 1;
+        let mut u64_varint_buf = [0; 10];
+        let mut encoded = unsigned_varint::encode::u64(block_count, &mut u64_varint_buf);
+        let mut payload = SpanBatchPayload::default();
+        let err = payload.decode_block_count(&mut encoded).unwrap_err();
+        assert_eq!(err, SpanBatchError::TooBigSpanBatchSize);
+    }
+
+    #[test]
+    fn test_decode_block_tx_counts() {
+        let block_count = 2;
+        let mut u64_varint_buf = [0; 10];
+        let mut encoded = unsigned_varint::encode::u64(block_count, &mut u64_varint_buf);
+        let mut payload = SpanBatchPayload::default();
+        payload.decode_block_count(&mut encoded).unwrap();
+        let mut r: Vec<u8> = Vec::new();
+        for _ in 0..2 {
+            let mut buf = [0u8; 10];
+            let encoded = unsigned_varint::encode::u64(2, &mut buf);
+            r.append(&mut encoded.to_vec());
+        }
+        payload.decode_block_tx_counts(&mut r.as_slice()).unwrap();
+        assert_eq!(payload.block_tx_counts, vec![2, 2]);
+    }
+}