fix(ingress): max txs per bundle in validation (#114)

Closes #113

Author: thedevbirb

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "buildernet-orderflow-proxy"
-version = "1.1.1"
+version = "1.1.2"
 edition = "2021"
 publish = false
 build = "build.rs"

Cargo.toml

@@ -6,7 +6,7 @@ use alloy_signer_local::PrivateKeySigner;
 use buildernet_orderflow_proxy::{
     consts::FLASHBOTS_SIGNATURE_HEADER,
     ingress::maybe_verify_signature,
-    primitives::{SystemBundle, SystemBundleMetadata, UtcInstant},
+    primitives::{SystemBundleDecoder, SystemBundleMetadata, UtcInstant},
     priority::Priority,
     utils::testutils::Random,
 };
@@ -81,6 +81,8 @@ pub fn bench_validation(c: &mut Criterion) {
     group.bench_function(BenchmarkId::from_parameter(size), |b| {
         let mut rng = StdRng::seed_from_u64(12);
 
+        let decoder = SystemBundleDecoder::default();
+
         // We use iter_batched here so we have an owned value for the benchmarked function (second
         // closure)
         b.iter_batched(
@@ -92,7 +94,7 @@ pub fn bench_validation(c: &mut Criterion) {
                         signer: input.signer,
                         priority: Priority::Medium,
                     };
-                    let result = SystemBundle::try_decode(input.raw_bundle, metadata).unwrap();
+                    let result = decoder.try_decode(input.raw_bundle, metadata).unwrap();
                     black_box(result);
                 }
             },

benches/validation.rs

@@ -4,11 +4,15 @@ use alloy_primitives::Address;
 use alloy_signer_local::PrivateKeySigner;
 use clap::{Args, Parser, ValueHint};
 
-use crate::indexer::{
-    click::{
-        default_disk_backup_database_path, MAX_DISK_BACKUP_SIZE_BYTES, MAX_MEMORY_BACKUP_SIZE_BYTES,
+use crate::{
+    indexer::{
+        click::{
+            default_disk_backup_database_path, MAX_DISK_BACKUP_SIZE_BYTES,
+            MAX_MEMORY_BACKUP_SIZE_BYTES,
+        },
+        BUNDLE_RECEIPTS_TABLE_NAME, BUNDLE_TABLE_NAME,
     },
-    BUNDLE_RECEIPTS_TABLE_NAME, BUNDLE_TABLE_NAME,
+    SystemBundleDecoder,
 };
 
 /// The maximum request size in bytes (10 MiB).
@@ -199,6 +203,10 @@ pub struct OrderflowIngressArgs {
     #[clap(long, default_value_t = MAX_REQUEST_SIZE_BYTES)]
     pub max_request_size: usize,
 
+    /// The maximum number of raw transactions per bundle.
+    #[clap(long, default_value_t = SystemBundleDecoder::DEFAULT_MAX_TXS_PER_BUNDLE)]
+    pub max_txs_per_bundle: usize,
+
     /// Enable rate limiting.
     #[clap(long, default_value_t = false)]
     pub enable_rate_limiting: bool,
@@ -249,6 +257,7 @@ impl Default for OrderflowIngressArgs {
             builder_name: String::from("buildernet"),
             builder_hub_url: None,
             flashbots_signer: None,
+            max_txs_per_bundle: 100,
             enable_rate_limiting: false,
             metrics: None,
             orderflow_signer: None,

src/cli.rs

@@ -383,7 +383,7 @@ pub(crate) mod tests {
             click::models::{BundleReceiptRow, BundleRow},
             tests::bundle_receipt_example,
         },
-        primitives::{BundleReceipt, SystemBundle},
+        primitives::{BundleReceipt, SystemBundleDecoder},
         priority::Priority,
     };
 
@@ -458,9 +458,9 @@ pub(crate) mod tests {
 
         assert_eq!(system_bundle.raw_bundle, Arc::new(raw_bundle_round_trip.clone()));
 
+        let decoder = SystemBundleDecoder::default();
         let system_bundle_round_trip =
-            SystemBundle::try_decode(raw_bundle_round_trip, system_bundle.metadata.clone())
-                .unwrap();
+            decoder.try_decode(raw_bundle_round_trip, system_bundle.metadata.clone()).unwrap();
 
         assert_eq!(system_bundle, system_bundle_round_trip);
     }
@@ -473,9 +473,10 @@ pub(crate) mod tests {
         let raw_bundle_round_trip: RawBundle = bundle_row.into();
 
         assert_eq!(system_bundle.raw_bundle, Arc::new(raw_bundle_round_trip.clone()));
+
+        let decoder = SystemBundleDecoder::default();
         let system_bundle_round_trip =
-            SystemBundle::try_decode(raw_bundle_round_trip, system_bundle.metadata.clone())
-                .unwrap();
+            decoder.try_decode(raw_bundle_round_trip, system_bundle.metadata.clone()).unwrap();
 
         assert_eq!(system_bundle, system_bundle_round_trip);
     }

src/indexer/click/models.rs

@@ -169,7 +169,9 @@ pub(crate) mod tests {
     use time::UtcDateTime;
 
     use crate::{
-        primitives::{BundleReceipt, SystemBundle, SystemBundleMetadata, UtcInstant},
+        primitives::{
+            BundleReceipt, SystemBundle, SystemBundleDecoder, SystemBundleMetadata, UtcInstant,
+        },
         priority::Priority,
     };
 
@@ -231,7 +233,8 @@ pub(crate) mod tests {
 
         let metadata = SystemBundleMetadata { signer, received_at, priority: Priority::Medium };
 
-        SystemBundle::try_decode(bundle, metadata).unwrap()
+        let decoder = SystemBundleDecoder::default();
+        decoder.try_decode(bundle, metadata).unwrap()
     }
 
     /// An example cancel bundle to use for testing.
@@ -241,7 +244,8 @@ pub(crate) mod tests {
         let received_at = UtcInstant::now();
 
         let metadata = SystemBundleMetadata { signer, received_at, priority: Priority::Medium };
-        SystemBundle::try_decode(bundle, metadata).unwrap()
+        let decoder = SystemBundleDecoder::default();
+        decoder.try_decode(bundle, metadata).unwrap()
     }
 
     pub(crate) fn bundle_receipt_example() -> BundleReceipt {

src/indexer/mod.rs

@@ -1,7 +1,9 @@
-use crate::{jsonrpc::JsonRpcError, validation::ValidationError};
+use crate::{
+    jsonrpc::JsonRpcError, primitives::SystemBundleDecodingError, validation::ValidationError,
+};
 use alloy_consensus::crypto::RecoveryError;
 use alloy_eips::eip2718::Eip2718Error;
-use rbuilder_primitives::serialize::{RawBundleConvertError, RawShareBundleConvertError};
+use rbuilder_primitives::serialize::RawShareBundleConvertError;
 
 #[derive(Debug, thiserror::Error)]
 pub enum IngressError {
@@ -16,7 +18,7 @@ pub enum IngressError {
     Decode2718(#[from] Eip2718Error),
     /// Bundle decoding error.
     #[error(transparent)]
-    BundleDecode(#[from] RawBundleConvertError),
+    SystemBundleDecoding(#[from] SystemBundleDecodingError),
     /// MEV Share bundle decoding error.
     #[error(transparent)]
     ShareBundleDecode(#[from] RawShareBundleConvertError),
@@ -26,6 +28,8 @@ pub enum IngressError {
     /// Serde error.
     #[error(transparent)]
     Serde(#[from] serde_json::Error),
+    #[error("too many transactions in bundle")]
+    TooManyTransactions,
 }
 
 impl IngressError {
@@ -35,8 +39,9 @@ impl IngressError {
             Self::EmptyRawTransaction |
             Self::Validation(_) |
             Self::Decode2718(_) |
-            Self::BundleDecode(_) |
+            Self::SystemBundleDecoding(_) |
             Self::ShareBundleDecode(_) |
+            Self::TooManyTransactions |
             Self::Recovery(_) => JsonRpcError::InvalidParams,
             Self::Serde(_) => JsonRpcError::ParseError,
         }

src/ingress/error.rs

@@ -15,8 +15,8 @@ use crate::{
     },
     primitives::{
         decode_transaction, BundleHash as _, BundleReceipt, DecodedBundle, DecodedShareBundle,
-        EthResponse, EthereumTransaction, Samplable, SystemBundle, SystemBundleMetadata,
-        SystemMevShareBundle, SystemTransaction, UtcInstant,
+        EthResponse, EthereumTransaction, Samplable, SystemBundle, SystemBundleDecoder,
+        SystemBundleMetadata, SystemMevShareBundle, SystemTransaction, UtcInstant,
     },
     priority::{pqueue::PriorityQueues, Priority},
     rate_limit::CounterOverTime,
@@ -62,6 +62,7 @@ pub struct OrderflowIngress {
     pub rate_limit_count: u64,
     pub score_lookback_s: u64,
     pub score_bucket_s: u64,
+    pub system_bundle_decoder: SystemBundleDecoder,
     pub spam_thresholds: SpamThresholds,
     pub pqueues: PriorityQueues,
     pub entities: DashMap<Entity, EntityData>,
@@ -605,11 +606,12 @@ impl OrderflowIngress {
         self.record_queue_capacity_metrics(priority);
 
         // Decode and validate the bundle.
+        let decoder = self.system_bundle_decoder;
         let bundle = self
             .pqueues
             .spawn_with_priority(priority, move || {
                 let metadata = SystemBundleMetadata { signer, received_at, priority };
-                SystemBundle::try_decode_with_lookup(bundle, metadata, lookup)
+                decoder.try_decode_with_lookup(bundle, metadata, lookup)
             })
             .await
             .inspect_err(|e| {

src/ingress/mod.rs

@@ -9,6 +9,7 @@ use crate::{
     metrics::{
         BuilderHubMetrics, IngressHandlerMetricsExt, IngressSystemMetrics, IngressUserMetrics,
     },
+    primitives::SystemBundleDecoder,
     runner::CliContext,
     statics::LOCAL_PEER_STORE,
     tasks::TaskExecutor,
@@ -205,6 +206,7 @@ pub async fn run_with_listeners(
         rate_limit_count: args.rate_limit_count,
         score_lookback_s: args.score_lookback_s,
         score_bucket_s: args.score_bucket_s,
+        system_bundle_decoder: SystemBundleDecoder { max_txs_per_bundle: args.max_txs_per_bundle },
         spam_thresholds: SpamThresholds::default(),
         flashbots_signer: args.flashbots_signer,
         pqueues: Default::default(),

src/lib.rs

@@ -59,6 +59,14 @@ pub struct SystemBundle {
     pub metadata: SystemBundleMetadata,
 }
 
+#[derive(Debug, thiserror::Error)]
+pub enum SystemBundleDecodingError {
+    #[error(transparent)]
+    RawBundleConvertError(#[from] RawBundleConvertError),
+    #[error("bundle contains too many transactions")]
+    TooManyTransactions,
+}
+
 /// Decoded bundle type. Either a new, full bundle or an empty replacement bundle.
 #[allow(clippy::large_enum_variant)]
 #[derive(PartialEq, Eq, Clone, Debug)]
@@ -241,33 +249,56 @@ impl BundleHash for RawShareBundle {
     }
 }
 
-impl SystemBundle {
+/// Decoder for system bundles with additional constraints.
+#[derive(Debug, Clone, Copy)]
+pub struct SystemBundleDecoder {
+    /// Maximum number of transactions allowed in a bundle.
+    pub max_txs_per_bundle: usize,
+}
+
+impl Default for SystemBundleDecoder {
+    fn default() -> Self {
+        Self { max_txs_per_bundle: Self::DEFAULT_MAX_TXS_PER_BUNDLE }
+    }
+}
+
+impl SystemBundleDecoder {
+    /// The maximum number of transactions allowed in a bundle received via `eth_sendBundle`.
+    pub const DEFAULT_MAX_TXS_PER_BUNDLE: usize = 100;
+
     /// Create a new system bundle from a raw bundle and additional data.
     /// Returns an error if the raw bundle fails to decode.
     pub fn try_decode(
+        &self,
         bundle: RawBundle,
         metadata: SystemBundleMetadata,
-    ) -> Result<Self, RawBundleConvertError> {
-        Self::try_decode_inner(bundle, metadata, None::<fn(B256) -> Option<Address>>)
+    ) -> Result<SystemBundle, SystemBundleDecodingError> {
+        self.try_decode_inner(bundle, metadata, None::<fn(B256) -> Option<Address>>)
     }
 
     /// Create a new system bundle from a raw bundle and additional data, using a signer lookup
     /// function for the transaction signers.
     pub fn try_decode_with_lookup(
+        &self,
         bundle: RawBundle,
         metadata: SystemBundleMetadata,
         lookup: impl Fn(B256) -> Option<Address>,
-    ) -> Result<Self, RawBundleConvertError> {
-        Self::try_decode_inner(bundle, metadata, Some(lookup))
+    ) -> Result<SystemBundle, SystemBundleDecodingError> {
+        self.try_decode_inner(bundle, metadata, Some(lookup))
     }
 
     /// Create a new system bundle from a raw bundle and additional data, using a signer lookup
     /// function for the transaction signers. Returns an error if the raw bundle fails to decode.
     fn try_decode_inner(
+        &self,
         mut bundle: RawBundle,
         metadata: SystemBundleMetadata,
         lookup: Option<impl Fn(B256) -> Option<Address>>,
-    ) -> Result<Self, RawBundleConvertError> {
+    ) -> Result<SystemBundle, SystemBundleDecodingError> {
+        if bundle.txs.len() > self.max_txs_per_bundle {
+            return Err(SystemBundleDecodingError::TooManyTransactions);
+        }
+
         let raw_bundle_hash = bundle.bundle_hash();
         // Set the bundle hash in the metadata.
         bundle.metadata.bundle_hash = Some(raw_bundle_hash);
@@ -282,14 +313,16 @@ impl SystemBundle {
             bundle.signer = Some(metadata.signer);
         }
 
-        Ok(Self {
+        Ok(SystemBundle {
             raw_bundle: Arc::new(bundle),
             decoded_bundle: Arc::new(decoded),
             bundle_hash: raw_bundle_hash,
             metadata,
         })
     }
+}
 
+impl SystemBundle {
     /// Returns `true` if the bundle is a replacement.
     pub fn is_replacement(&self) -> bool {
         matches!(self.decoded_bundle.as_ref(), DecodedBundle::EmptyReplacement(_))
@@ -761,4 +794,39 @@ mod tests {
         let json = serde_json::to_value(response).unwrap();
         assert_eq!(json, json!(hash));
     }
+
+    #[test]
+    fn too_many_txs_error() {
+        let decoder = SystemBundleDecoder::default();
+        let raw_bundle = RawBundle {
+            txs: vec![Bytes::from(vec![0u8; 8]); decoder.max_txs_per_bundle + 1],
+            metadata: RawBundleMetadata {
+                version: None,
+                block_number: None,
+                reverting_tx_hashes: vec![],
+                dropping_tx_hashes: vec![],
+                replacement_uuid: None,
+                uuid: None,
+                signing_address: None,
+                refund_identity: None,
+                min_timestamp: None,
+                max_timestamp: None,
+                replacement_nonce: None,
+                refund_percent: None,
+                refund_recipient: None,
+                refund_tx_hashes: None,
+                delayed_refund: None,
+                bundle_hash: None,
+            },
+        };
+        let metadata = SystemBundleMetadata {
+            signer: Address::ZERO,
+            received_at: UtcInstant::now(),
+            priority: Priority::Medium,
+        };
+
+        // This should be the first reason decoding of such garbage data fails.
+        let result = decoder.try_decode(raw_bundle.clone(), metadata.clone());
+        assert!(matches!(result, Err(SystemBundleDecodingError::TooManyTransactions)));
+    }
 }