Add flashblock number filters to eth_sendBundle (#213)

* add min and max flashblock number fields to bundle

* add min and max flashblock number to FBPooledTransaction

* filter based on flashblock number

* wrap best txs iterator to consider flashblock number

* fix map structure and wrap better

* integration tests

Author: akundaz

crates/op-rbuilder/src/builders/flashblocks/best_txs.rs

@@ -0,0 +1,132 @@
+use std::{
+    collections::{BTreeMap, VecDeque},
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
+};
+
+use alloy_primitives::Address;
+use reth_payload_util::PayloadTransactions;
+use reth_transaction_pool::PoolTransaction;
+use tracing::debug;
+
+use crate::tx::MaybeFlashblockFilter;
+
+pub struct BestFlashblocksTxs<T, I>
+where
+    T: PoolTransaction,
+    I: PayloadTransactions<Transaction = T>,
+{
+    inner: I,
+    current_flashblock_number: Arc<AtomicU64>,
+    early_transactions: BTreeMap<u64, VecDeque<T>>,
+}
+
+impl<T, I> BestFlashblocksTxs<T, I>
+where
+    T: PoolTransaction,
+    I: PayloadTransactions<Transaction = T>,
+{
+    pub fn new(inner: I, current_flashblock_number: Arc<AtomicU64>) -> Self {
+        Self {
+            inner,
+            current_flashblock_number,
+            early_transactions: Default::default(),
+        }
+    }
+}
+
+impl<T, I> PayloadTransactions for BestFlashblocksTxs<T, I>
+where
+    T: PoolTransaction + MaybeFlashblockFilter,
+    I: PayloadTransactions<Transaction = T>,
+{
+    type Transaction = T;
+
+    fn next(&mut self, ctx: ()) -> Option<Self::Transaction> {
+        loop {
+            let flashblock_number = self.current_flashblock_number.load(Ordering::Relaxed);
+
+            // Check for new transactions that can be executed with the higher flashblock number
+            while let Some((&min_flashblock, _)) = self.early_transactions.first_key_value() {
+                if min_flashblock > flashblock_number {
+                    break;
+                }
+
+                if let Some(mut txs) = self.early_transactions.remove(&min_flashblock) {
+                    while let Some(tx) = txs.pop_front() {
+                        // Re-check max flashblock number just in case
+                        if let Some(max) = tx.flashblock_number_max() {
+                            if flashblock_number > max {
+                                debug!(
+                                    target: "payload_builder",
+                                    sender = ?tx.sender(),
+                                    nonce = tx.nonce(),
+                                    current_flashblock = flashblock_number,
+                                    max_flashblock = max,
+                                    "Bundle flashblock max exceeded"
+                                );
+                                self.mark_invalid(tx.sender(), tx.nonce());
+                                continue;
+                            }
+                        }
+
+                        // The vecdeque isn't modified in place so we need to replace it
+                        if !txs.is_empty() {
+                            self.early_transactions.insert(min_flashblock, txs);
+                        }
+
+                        return Some(tx);
+                    }
+                }
+            }
+
+            let tx = self.inner.next(ctx)?;
+
+            let flashblock_number_min = tx.flashblock_number_min();
+            let flashblock_number_max = tx.flashblock_number_max();
+
+            // Check min flashblock requirement
+            if let Some(min) = flashblock_number_min {
+                if flashblock_number < min {
+                    self.early_transactions
+                        .entry(min)
+                        .or_default()
+                        .push_back(tx);
+                    continue;
+                }
+            }
+
+            // Check max flashblock requirement
+            if let Some(max) = flashblock_number_max {
+                if flashblock_number > max {
+                    debug!(
+                        target: "payload_builder",
+                        sender = ?tx.sender(),
+                        nonce = tx.nonce(),
+                        current_flashblock = flashblock_number,
+                        max_flashblock = max,
+                        "Bundle flashblock max exceeded"
+                    );
+                    self.inner.mark_invalid(tx.sender(), tx.nonce());
+                    continue;
+                }
+            }
+
+            return Some(tx);
+        }
+    }
+
+    fn mark_invalid(&mut self, sender: Address, nonce: u64) {
+        self.inner.mark_invalid(sender, nonce);
+
+        // Clear early_transactions from this sender with a greater nonce as
+        // these transactions now will not execute because there would be a
+        // nonce gap
+        self.early_transactions.retain(|_, txs| {
+            txs.retain(|tx| !(tx.sender() == sender && tx.nonce() > nonce));
+            !txs.is_empty()
+        });
+    }
+}

crates/op-rbuilder/src/builders/flashblocks/mod.rs

@@ -5,6 +5,7 @@ use service::FlashblocksServiceBuilder;
 
 mod config;
 //mod context;
+mod best_txs;
 mod payload;
 mod service;
 mod wspub;

crates/op-rbuilder/src/builders/flashblocks/payload.rs

@@ -2,7 +2,7 @@ use super::{config::FlashblocksConfig, wspub::WebSocketPublisher};
 use crate::{
     builders::{
         context::{estimate_gas_for_builder_tx, OpPayloadBuilderCtx},
-        flashblocks::config::FlashBlocksConfigExt,
+        flashblocks::{best_txs::BestFlashblocksTxs, config::FlashBlocksConfigExt},
         generator::{BlockCell, BuildArguments},
         BuilderConfig, BuilderTx,
     },
@@ -42,7 +42,10 @@ use rollup_boost::{
 use serde::{Deserialize, Serialize};
 use std::{
     ops::{Div, Rem},
-    sync::Arc,
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
     time::Instant,
 };
 use tokio::sync::{
@@ -61,15 +64,15 @@ struct ExtraExecutionInfo {
 #[derive(Debug, Default)]
 struct FlashblocksExtraCtx {
     /// Current flashblock index
-    pub flashblock_index: u64,
+    pub flashblock_index: Arc<AtomicU64>,
     /// Target flashblock count
     pub target_flashblock_count: u64,
 }
 
 impl OpPayloadBuilderCtx<FlashblocksExtraCtx> {
     /// Returns the current flashblock index
     pub fn flashblock_index(&self) -> u64 {
-        self.extra_ctx.flashblock_index
+        self.extra_ctx.flashblock_index.load(Ordering::Relaxed)
     }
 
     /// Returns the target flashblock count
@@ -79,8 +82,10 @@ impl OpPayloadBuilderCtx<FlashblocksExtraCtx> {
 
     /// Increments the flashblock index
     pub fn increment_flashblock_index(&mut self) -> u64 {
-        self.extra_ctx.flashblock_index += 1;
-        self.extra_ctx.flashblock_index
+        self.extra_ctx
+            .flashblock_index
+            .fetch_add(1, Ordering::Relaxed);
+        self.flashblock_index()
     }
 
     /// Sets the target flashblock count
@@ -248,7 +253,7 @@ where
             builder_signer: self.config.builder_signer,
             metrics: Default::default(),
             extra_ctx: FlashblocksExtraCtx {
-                flashblock_index: 0,
+                flashblock_index: Arc::new(AtomicU64::new(0)),
                 target_flashblock_count: self.config.flashblocks_per_block(),
             },
         };
@@ -430,10 +435,13 @@ where
                         .build();
 
                     let best_txs_start_time = Instant::now();
-                    let best_txs = BestPayloadTransactions::new(
-                        // We are not using without_updates in here, so arriving transaction could target the current block
-                        self.pool
-                            .best_transactions_with_attributes(ctx.best_transaction_attributes()),
+                    let best_txs = BestFlashblocksTxs::new(
+                        BestPayloadTransactions::new(
+                            self.pool.best_transactions_with_attributes(
+                                ctx.best_transaction_attributes(),
+                            ),
+                        ),
+                        ctx.extra_ctx.flashblock_index.clone(),
                     );
                     let transaction_pool_fetch_time = best_txs_start_time.elapsed();
                     ctx.metrics