Have the option to use the reth payload generator

Cargo.toml

@@ -73,6 +73,7 @@ reth-payload-builder-primitives = { git = "https://github.com/paradigmxyz/reth",
 reth-payload-util = { git = "https://github.com/paradigmxyz/reth", tag = "v1.3.12" }
 reth-rpc-layer = { git = "https://github.com/paradigmxyz/reth", tag = "v1.3.12" }
 reth-testing-utils = { git = "https://github.com/paradigmxyz/reth", tag = "v1.3.12" }
+reth-node-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.3.12" }
 
 # reth optimism
 reth-optimism-primitives = { git = "https://github.com/paradigmxyz/reth", tag = "v1.3.12" }

crates/op-rbuilder/Cargo.toml

@@ -40,6 +40,7 @@ reth-payload-util.workspace = true
 reth-transaction-pool.workspace = true
 reth-testing-utils.workspace = true
 reth-optimism-forks.workspace = true
+reth-node-builder.workspace = true
 
 alloy-primitives.workspace = true
 alloy-consensus.workspace = true

crates/op-rbuilder/src/integration/integration_test.rs

@@ -461,6 +461,40 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    #[cfg(not(feature = "flashblocks"))]
+    async fn integration_test_transaction_flood_no_sleep() -> eyre::Result<()> {
+        // This test validates that if we flood the builder with many transactions
+        // and we request short block times, the builder can still eventually resolve all the transactions
+
+        let test_harness = TestHarnessBuilder::new("integration_test_transaction_flood_no_sleep")
+            .build()
+            .await?;
+
+        let mut block_generator = test_harness.block_generator().await?;
+        let provider = test_harness.provider()?;
+
+        // Send 200 valid transactions to the builder
+        // More than this and there is an issue with the RPC endpoint not being able to handle the load
+        let mut transactions = vec![];
+        for _ in 0..200 {
+            let tx = test_harness.send_valid_transaction().await?;
+            let tx_hash = *tx.tx_hash();
+            transactions.push(tx_hash);
+        }
+
+        // After a 10 blocks all the transactions should be included in a block
+        for _ in 0..10 {
+            block_generator.submit_payload(None, 0, true).await.unwrap();
+        }
+
+        for tx in transactions {
+            provider.get_transaction_receipt(tx).await?;
+        }
+
+        Ok(())
+    }
+
     #[tokio::test]
     #[cfg(feature = "flashblocks")]
     async fn integration_test_chain_produces_blocks() -> eyre::Result<()> {

crates/op-rbuilder/src/integration/op_rbuilder.rs

@@ -121,7 +121,8 @@ impl Service for OpRbuilderConfig {
             .arg("--builder.log-pool-transactions")
             .arg("--port")
             .arg(self.network_port.expect("network_port not set").to_string())
-            .arg("--ipcdisable");
+            .arg("--ipcdisable")
+            .arg("-vvvv");
 
         if let Some(revert_protection) = self.with_revert_protection {
             if revert_protection {

crates/op-rbuilder/src/payload_builder_vanilla.rs

@@ -25,7 +25,8 @@ use reth::{
     payload::PayloadBuilderHandle,
 };
 use reth_basic_payload_builder::{
-    BasicPayloadJobGeneratorConfig, BuildOutcome, BuildOutcomeKind, PayloadConfig,
+    BasicPayloadJobGeneratorConfig, BuildOutcome, BuildOutcomeKind, MissingPayloadBehaviour,
+    PayloadConfig,
 };
 use reth_chain_state::{ExecutedBlock, ExecutedBlockWithTrieUpdates};
 use reth_chainspec::{ChainSpecProvider, EthChainSpec, EthereumHardforks};
@@ -35,6 +36,7 @@ use reth_evm::{
 };
 use reth_execution_types::ExecutionOutcome;
 use reth_node_api::{NodePrimitives, NodeTypes, TxTy};
+use reth_node_builder::components::BasicPayloadServiceBuilder;
 use reth_optimism_chainspec::OpChainSpec;
 use reth_optimism_consensus::{calculate_receipt_root_no_memo_optimism, isthmus};
 use reth_optimism_evm::{OpEvmConfig, OpNextBlockEnvAttributes};
@@ -51,7 +53,7 @@ use reth_optimism_txpool::OpPooledTx;
 use reth_payload_builder::PayloadBuilderService;
 use reth_payload_builder_primitives::PayloadBuilderError;
 use reth_payload_primitives::PayloadBuilderAttributes;
-use reth_payload_util::{BestPayloadTransactions, PayloadTransactions};
+use reth_payload_util::{BestPayloadTransactions, NoopPayloadTransactions, PayloadTransactions};
 use reth_primitives::{BlockBody, SealedHeader};
 use reth_primitives_traits::{proofs, Block as _, RecoveredBlock, SignedTransaction};
 use reth_provider::{
@@ -110,12 +112,12 @@ impl CustomOpPayloadBuilder {
         _flashblocks_ws_url: String,
         _chain_block_time: u64,
         _flashblock_block_time: u64,
-    ) -> Self {
-        Self {
+    ) -> BasicPayloadServiceBuilder<CustomOpPayloadBuilder> {
+        BasicPayloadServiceBuilder::new(CustomOpPayloadBuilder {
             builder_signer,
             extra_block_deadline,
             enable_revert_protection,
-        }
+        })
     }
 }
 
@@ -145,6 +147,7 @@ where
             self.builder_signer,
             pool,
             ctx.provider().clone(),
+            self.enable_revert_protection,
         ))
     }
 }
@@ -199,28 +202,74 @@ where
 impl<Pool, Client, Txs> reth_basic_payload_builder::PayloadBuilder
     for OpPayloadBuilderVanilla<Pool, Client, Txs>
 where
-    Pool: Clone + Send + Sync,
-    Client: Clone + Send + Sync,
-    Txs: Clone + Send + Sync,
+    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = OpTransactionSigned>>,
+    Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks> + Clone,
+    Txs: OpPayloadTransactions<Pool::Transaction>,
 {
     type Attributes = OpPayloadBuilderAttributes<OpTransactionSigned>;
     type BuiltPayload = OpBuiltPayload;
 
     fn try_build(
         &self,
-        _args: reth_basic_payload_builder::BuildArguments<Self::Attributes, Self::BuiltPayload>,
+        args: reth_basic_payload_builder::BuildArguments<Self::Attributes, Self::BuiltPayload>,
     ) -> Result<BuildOutcome<Self::BuiltPayload>, PayloadBuilderError> {
-        unimplemented!()
+        let pool = self.pool.clone();
+
+        let reth_basic_payload_builder::BuildArguments {
+            cached_reads,
+            config,
+            cancel: _, // TODO
+            best_payload: _,
+        } = args;
+
+        let args = BuildArguments {
+            cached_reads,
+            config,
+            enable_revert_protection: self.enable_revert_protection,
+            cancel: CancellationToken::new(),
+        };
+
+        self.build_payload(
+            args,
+            |attrs| {
+                #[allow(clippy::unit_arg)]
+                self.best_transactions
+                    .best_transactions(pool.clone(), attrs)
+            },
+            |hashes| {
+                #[allow(clippy::unit_arg)]
+                self.best_transactions.remove_invalid(pool.clone(), hashes)
+            },
+        )
+    }
+
+    fn on_missing_payload(
+        &self,
+        _args: reth_basic_payload_builder::BuildArguments<Self::Attributes, Self::BuiltPayload>,
+    ) -> MissingPayloadBehaviour<Self::BuiltPayload> {
+        MissingPayloadBehaviour::AwaitInProgress
     }
 
     fn build_empty_payload(
         &self,
-        _config: reth_basic_payload_builder::PayloadConfig<
+        config: reth_basic_payload_builder::PayloadConfig<
             Self::Attributes,
             reth_basic_payload_builder::HeaderForPayload<Self::BuiltPayload>,
         >,
     ) -> Result<Self::BuiltPayload, PayloadBuilderError> {
-        unimplemented!()
+        let args = BuildArguments {
+            config,
+            cached_reads: Default::default(),
+            cancel: Default::default(),
+            enable_revert_protection: false,
+        };
+        self.build_payload(
+            args,
+            |_| NoopPayloadTransactions::<Pool::Transaction>::default(),
+            |_| {},
+        )?
+        .into_payload()
+        .ok_or_else(|| PayloadBuilderError::MissingPayload)
     }
 }
 
@@ -242,6 +291,8 @@ pub struct OpPayloadBuilderVanilla<Pool, Client, Txs = ()> {
     pub best_transactions: Txs,
     /// The metrics for the builder
     pub metrics: OpRBuilderMetrics,
+    /// Whether we enable revert protection
+    pub enable_revert_protection: bool,
 }
 
 impl<Pool, Client> OpPayloadBuilderVanilla<Pool, Client> {
@@ -251,8 +302,16 @@ impl<Pool, Client> OpPayloadBuilderVanilla<Pool, Client> {
         builder_signer: Option<Signer>,
         pool: Pool,
         client: Client,
+        enable_revert_protection: bool,
     ) -> Self {
-        Self::with_builder_config(evm_config, builder_signer, pool, client, Default::default())
+        Self::with_builder_config(
+            evm_config,
+            builder_signer,
+            pool,
+            client,
+            Default::default(),
+            enable_revert_protection,
+        )
     }
 
     pub fn with_builder_config(
@@ -261,6 +320,7 @@ impl<Pool, Client> OpPayloadBuilderVanilla<Pool, Client> {
         pool: Pool,
         client: Client,
         config: OpBuilderConfig,
+        enable_revert_protection: bool,
     ) -> Self {
         Self {
             pool,
@@ -270,6 +330,7 @@ impl<Pool, Client> OpPayloadBuilderVanilla<Pool, Client> {
             best_transactions: (),
             metrics: Default::default(),
             builder_signer,
+            enable_revert_protection,
         }
     }
 }
@@ -528,13 +589,18 @@ impl<Txs> OpBuilder<'_, Txs> {
             ctx.metrics
                 .transaction_pool_fetch_duration
                 .record(best_txs_start_time.elapsed());
-            ctx.execute_best_transactions(
-                &mut info,
-                state,
-                best_txs,
-                block_gas_limit,
-                block_da_limit,
-            )?;
+            if ctx
+                .execute_best_transactions(
+                    &mut info,
+                    state,
+                    best_txs,
+                    block_gas_limit,
+                    block_da_limit,
+                )?
+                .is_some()
+            {
+                return Ok(BuildOutcomeKind::Cancelled);
+            }
         }
 
         // Add builder tx to the block

crates/op-rbuilder/src/tester/main.rs

@@ -30,6 +30,9 @@ enum Commands {
 
         #[clap(long, short, action)]
         flashblocks_endpoint: Option<String>,
+
+        #[clap(long, action, default_value = "false")]
+        no_sleep: bool,
     },
     /// Deposit funds to the system
     Deposit {
@@ -51,12 +54,14 @@ async fn main() -> eyre::Result<()> {
             no_tx_pool,
             block_time_secs,
             flashblocks_endpoint,
+            no_sleep,
         } => {
             run_system(
                 validation,
                 no_tx_pool,
                 block_time_secs,
                 flashblocks_endpoint,
+                no_sleep,
             )
             .await
         }

crates/op-rbuilder/src/tester/mod.rs

@@ -20,10 +20,7 @@ use reth_payload_builder::PayloadId;
 use reth_rpc_layer::{AuthClientLayer, AuthClientService, JwtSecret};
 use rollup_boost::{Flashblocks, FlashblocksService};
 use serde_json::Value;
-use std::{
-    str::FromStr,
-    time::{SystemTime, UNIX_EPOCH},
-};
+use std::str::FromStr;
 
 /// Helper for engine api operations
 pub struct EngineApi {
@@ -205,6 +202,7 @@ pub struct BlockGenerator {
     latest_hash: B256,
     no_tx_pool: bool,
     block_time_secs: u64,
+    timestamp: u64,
     // flashblocks service
     flashblocks_endpoint: Option<String>,
     flashblocks_service: Option<FlashblocksService>,
@@ -223,6 +221,7 @@ impl BlockGenerator {
             validation_api,
             latest_hash: B256::ZERO, // temporary value
             no_tx_pool,
+            timestamp: 0,
             block_time_secs,
             flashblocks_endpoint,
             flashblocks_service: None,
@@ -233,6 +232,7 @@ impl BlockGenerator {
     pub async fn init(&mut self) -> eyre::Result<Block> {
         let latest_block = self.engine_api.latest().await?.expect("block not found");
         self.latest_hash = latest_block.header.hash;
+        self.timestamp = latest_block.header.timestamp;
 
         // Sync validation node if it exists
         if let Some(validation_api) = &self.validation_api {
@@ -340,11 +340,7 @@ impl BlockGenerator {
         block_building_delay_secs: u64,
         no_sleep: bool, // TODO: Change this, too many parameters we can tweak here to put as a function arguments
     ) -> eyre::Result<B256> {
-        let timestamp = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .unwrap()
-            .as_secs();
-        let timestamp = timestamp + self.block_time_secs;
+        let timestamp = self.timestamp + self.block_time_secs;
 
         // Add L1 block info as the first transaction in every L2 block
         // This deposit transaction contains L1 block metadata required by the L2 chain
@@ -464,6 +460,8 @@ impl BlockGenerator {
 
         // Update internal state
         self.latest_hash = new_block_hash;
+        self.timestamp = payload.execution_payload.timestamp();
+
         Ok(new_block_hash)
     }
 
@@ -508,6 +506,7 @@ pub async fn run_system(
     no_tx_pool: bool,
     block_time_secs: u64,
     flashblocks_endpoint: Option<String>,
+    no_sleep: bool,
 ) -> eyre::Result<()> {
     println!("Validation: {validation}");
 
@@ -531,7 +530,7 @@ pub async fn run_system(
     // Infinite loop generating blocks
     loop {
         println!("Generating new block...");
-        let block_hash = generator.generate_block().await?;
+        let block_hash = generator.submit_payload(None, 0, no_sleep).await?;
         println!("Generated block: {block_hash}");
     }
 }