Refactored Task trait

have an error with lifetime of &self with executor.spawn_blocking

Author: SozinM

crates/op-rbuilder/src/generator.rs

@@ -1,7 +1,7 @@
 use futures_util::Future;
 use futures_util::FutureExt;
 use reth::providers::BlockReaderIdExt;
-use reth::{providers::StateProviderFactory, tasks::TaskSpawner};
+use reth::providers::StateProviderFactory;
 use reth_basic_payload_builder::HeaderForPayload;
 use reth_basic_payload_builder::{BasicPayloadJobGeneratorConfig, PayloadConfig};
 use reth_node_api::PayloadBuilderAttributes;
@@ -56,11 +56,10 @@ pub trait PayloadBuilder: Send + Sync + Clone {
 
 /// The generator type that creates new jobs that builds empty blocks.
 #[derive(Debug)]
-pub struct BlockPayloadJobGenerator<Client, Tasks, Builder> {
+pub struct BlockPayloadJobGenerator<Client, Builder> {
     /// The client that can interact with the chain.
     client: Client,
-    /// How to spawn building tasks
-    executor: Tasks,
+
     /// The configuration for the job generator.
     _config: BasicPayloadJobGeneratorConfig,
     /// The type responsible for building payloads.
@@ -75,19 +74,17 @@ pub struct BlockPayloadJobGenerator<Client, Tasks, Builder> {
 
 // === impl EmptyBlockPayloadJobGenerator ===
 
-impl<Client, Tasks, Builder> BlockPayloadJobGenerator<Client, Tasks, Builder> {
+impl<Client, Builder> BlockPayloadJobGenerator<Client, Builder> {
     /// Creates a new [EmptyBlockPayloadJobGenerator] with the given config and custom
     /// [PayloadBuilder]
     pub fn with_builder(
         client: Client,
-        executor: Tasks,
         config: BasicPayloadJobGeneratorConfig,
         builder: Builder,
         ensure_only_one_payload: bool,
     ) -> Self {
         Self {
             client,
-            executor,
             _config: config,
             builder,
             ensure_only_one_payload,
@@ -96,20 +93,18 @@ impl<Client, Tasks, Builder> BlockPayloadJobGenerator<Client, Tasks, Builder> {
     }
 }
 
-impl<Client, Tasks, Builder> PayloadJobGenerator
-    for BlockPayloadJobGenerator<Client, Tasks, Builder>
+impl<Client, Builder> PayloadJobGenerator for BlockPayloadJobGenerator<Client, Builder>
 where
     Client: StateProviderFactory
         + BlockReaderIdExt<Header = HeaderForPayload<Builder::BuiltPayload>>
         + Clone
         + Unpin
         + 'static,
-    Tasks: TaskSpawner + Clone + Unpin + 'static,
     Builder: PayloadBuilder + Unpin + 'static,
     Builder::Attributes: Unpin + Clone,
     Builder::BuiltPayload: Unpin + Clone,
 {
-    type Job = BlockPayloadJob<Tasks, Builder>;
+    type Job = BlockPayloadJob<Builder>;
 
     /// This is invoked when the node receives payload attributes from the beacon node via
     /// `engine_forkchoiceUpdatedV1`
@@ -161,7 +156,6 @@ where
         let config = PayloadConfig::new(Arc::new(parent_header.clone()), attributes);
 
         let mut job = BlockPayloadJob {
-            executor: self.executor.clone(),
             builder: self.builder.clone(),
             config,
             cell: BlockCell::new(),
@@ -182,14 +176,12 @@ use std::{
 };
 
 /// A [PayloadJob] that builds empty blocks.
-pub struct BlockPayloadJob<Tasks, Builder>
+pub struct BlockPayloadJob<Builder>
 where
     Builder: PayloadBuilder,
 {
     /// The configuration for how the payload will be created.
     pub(crate) config: PayloadConfig<Builder::Attributes, HeaderForPayload<Builder::BuiltPayload>>,
-    /// How to spawn building tasks
-    pub(crate) executor: Tasks,
     /// The type responsible for building payloads.
     ///
     /// See [PayloadBuilder]
@@ -202,9 +194,8 @@ where
     pub(crate) build_complete: Option<oneshot::Receiver<Result<(), PayloadBuilderError>>>,
 }
 
-impl<Tasks, Builder> PayloadJob for BlockPayloadJob<Tasks, Builder>
+impl<Builder> PayloadJob for BlockPayloadJob<Builder>
 where
-    Tasks: TaskSpawner + Clone + 'static,
     Builder: PayloadBuilder + Unpin + 'static,
     Builder::Attributes: Unpin + Clone,
     Builder::BuiltPayload: Unpin + Clone,
@@ -245,9 +236,8 @@ pub struct BuildArguments<Attributes, Payload: BuiltPayload> {
 }
 
 /// A [PayloadJob] is a future that's being polled by the `PayloadBuilderService`
-impl<Tasks, Builder> BlockPayloadJob<Tasks, Builder>
+impl<Builder> BlockPayloadJob<Builder>
 where
-    Tasks: TaskSpawner + Clone + 'static,
     Builder: PayloadBuilder + Unpin + 'static,
     Builder::Attributes: Unpin + Clone,
     Builder::BuiltPayload: Unpin + Clone,
@@ -260,24 +250,20 @@ where
 
         let (tx, rx) = oneshot::channel();
         self.build_complete = Some(rx);
+        let args = BuildArguments {
+            cached_reads: Default::default(),
+            config: payload_config,
+            cancel,
+        };
 
-        self.executor.spawn_blocking(Box::pin(async move {
-            let args = BuildArguments {
-                cached_reads: Default::default(),
-                config: payload_config,
-                cancel,
-            };
-
-            let result = builder.try_build(args, cell);
-            let _ = tx.send(result);
-        }));
+        let result = builder.try_build(args, cell);
+        let _ = tx.send(result);
     }
 }
 
 /// A [PayloadJob] is a a future that's being polled by the `PayloadBuilderService`
-impl<Tasks, Builder> Future for BlockPayloadJob<Tasks, Builder>
+impl<Builder> Future for BlockPayloadJob<Builder>
 where
-    Tasks: TaskSpawner + Clone + 'static,
     Builder: PayloadBuilder + Unpin + 'static,
     Builder::Attributes: Unpin + Clone,
     Builder::BuiltPayload: Unpin + Clone,
@@ -601,7 +587,6 @@ mod tests {
         let mut rng = thread_rng();
 
         let client = MockEthProvider::default();
-        let executor = TokioTaskExecutor::default();
         let config = BasicPayloadJobGeneratorConfig::default();
         let builder = MockBuilder::<OpPrimitives>::new();
 
@@ -617,13 +602,8 @@ mod tests {
 
         client.extend_blocks(blocks.iter().cloned().map(|b| (b.hash(), b.unseal())));
 
-        let generator = BlockPayloadJobGenerator::with_builder(
-            client.clone(),
-            executor,
-            config,
-            builder.clone(),
-            false,
-        );
+        let generator =
+            BlockPayloadJobGenerator::with_builder(client.clone(), config, builder.clone(), false);
 
         // this is not nice but necessary
         let mut attr = OpPayloadBuilderAttributes::default();

crates/op-rbuilder/src/payload_builder_vanilla.rs

@@ -19,6 +19,7 @@ use op_alloy_consensus::{OpDepositReceipt, OpTypedTransaction};
 use reth::builder::{components::PayloadServiceBuilder, node::FullNodeTypes, BuilderContext};
 use reth::core::primitives::InMemorySize;
 use reth::payload::PayloadBuilderHandle;
+use reth::tasks::{TaskExecutor, TaskSpawner};
 use reth_basic_payload_builder::{
     BasicPayloadJobGeneratorConfig, BuildOutcome, BuildOutcomeKind, PayloadConfig,
 };
@@ -73,6 +74,7 @@ use revm::{
 };
 use std::error::Error as StdError;
 use std::{fmt::Display, sync::Arc, time::Instant};
+use tokio::sync::oneshot;
 use tokio_util::sync::CancellationToken;
 use tracing::{info, trace, warn};
 
@@ -101,7 +103,8 @@ where
         + Unpin
         + 'static,
 {
-    type PayloadBuilder = OpPayloadBuilderVanilla<Pool, Node::Provider, OpEvmConfig, OpPrimitives>;
+    type PayloadBuilder =
+        OpPayloadBuilderVanilla<Pool, Node::Provider, OpEvmConfig, OpPrimitives, TaskExecutor>;
 
     async fn build_payload_builder(
         &self,
@@ -114,6 +117,7 @@ where
             pool,
             ctx.provider().clone(),
             Arc::new(BasicOpReceiptBuilder::default()),
+            ctx.task_executor().clone(),
         ))
     }
 
@@ -127,7 +131,6 @@ where
 
         let payload_generator = BlockPayloadJobGenerator::with_builder(
             ctx.provider().clone(),
-            ctx.task_executor().clone(),
             payload_job_config,
             payload_builder,
             false,
@@ -145,14 +148,15 @@ where
     }
 }
 
-impl<Pool, Client, EvmConfig, N, Txs> reth_basic_payload_builder::PayloadBuilder
-    for OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N, Txs>
+impl<Pool, Client, EvmConfig, N, Tasks, Txs> reth_basic_payload_builder::PayloadBuilder
+    for OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N, Tasks, Txs>
 where
     Pool: Clone + Send + Sync,
     Client: Clone + Send + Sync,
     EvmConfig: Clone + Send + Sync,
     N: NodePrimitives,
     Txs: Clone + Send + Sync,
+    Tasks: TaskSpawner + Clone + Unpin + 'static,
 {
     type Attributes = OpPayloadBuilderAttributes<N::SignedTx>;
     type BuiltPayload = OpBuiltPayload<N>;
@@ -177,7 +181,10 @@ where
 
 /// Optimism's payload builder
 #[derive(Debug, Clone)]
-pub struct OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N: NodePrimitives, Txs = ()> {
+pub struct OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N: NodePrimitives, Tasks, Txs = ()>
+where
+    Tasks: TaskSpawner + Clone + Unpin + 'static,
+{
     /// The type responsible for creating the evm.
     pub evm_config: EvmConfig,
     /// The builder's signer key to use for an end of block tx
@@ -195,10 +202,14 @@ pub struct OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N: NodePrimitives, T
     pub metrics: OpRBuilderMetrics,
     /// Node primitive types.
     pub receipt_builder: Arc<dyn OpReceiptBuilder<N::SignedTx, Receipt = N::Receipt>>,
+    /// Executor to spawn tasks
+    pub executor: Tasks,
 }
 
-impl<Pool, Client, EvmConfig, N: NodePrimitives>
-    OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N>
+impl<Pool, Client, EvmConfig, N: NodePrimitives, Tasks>
+    OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N, Tasks>
+where
+    Tasks: TaskSpawner + Clone + Unpin + 'static,
 {
     /// `OpPayloadBuilder` constructor.
     pub fn new(
@@ -207,13 +218,15 @@ impl<Pool, Client, EvmConfig, N: NodePrimitives>
         pool: Pool,
         client: Client,
         receipt_builder: Arc<dyn OpReceiptBuilder<N::SignedTx, Receipt = N::Receipt>>,
+        executor: Tasks,
     ) -> Self {
         Self::with_builder_config(
             evm_config,
             builder_signer,
             pool,
             client,
             receipt_builder,
+            executor,
             Default::default(),
         )
     }
@@ -224,6 +237,7 @@ impl<Pool, Client, EvmConfig, N: NodePrimitives>
         pool: Pool,
         client: Client,
         receipt_builder: Arc<dyn OpReceiptBuilder<N::SignedTx, Receipt = N::Receipt>>,
+        executor: Tasks,
         config: OpBuilderConfig,
     ) -> Self {
         Self {
@@ -235,18 +249,23 @@ impl<Pool, Client, EvmConfig, N: NodePrimitives>
             best_transactions: (),
             metrics: Default::default(),
             builder_signer,
+            executor,
         }
     }
 }
 
-impl<EvmConfig, Pool, Client, N, Txs> PayloadBuilder
-    for OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N, Txs>
+impl<EvmConfig, Pool, Client, N, Tasks, Txs> PayloadBuilder
+    for OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N, Tasks, Txs>
 where
-    Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks> + Clone,
+    Client: StateProviderFactory
+        + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks>
+        + Clone
+        + 'static,
     N: OpPayloadPrimitives<_TX = OpTransactionSigned>,
-    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = N::SignedTx>>,
+    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = N::SignedTx>> + 'static,
     EvmConfig: ConfigureEvmFor<N>,
     Txs: OpPayloadTransactions<Pool::Transaction>,
+    Tasks: TaskSpawner + Clone + Unpin + 'static,
 {
     type Attributes = OpPayloadBuilderAttributes<N::SignedTx>;
     type BuiltPayload = OpBuiltPayload<N>;
@@ -258,44 +277,56 @@ where
     ) -> Result<(), PayloadBuilderError> {
         let pool = self.pool.clone();
         let block_build_start_time = Instant::now();
-
-        match self.build_payload(args, |attrs| {
-            #[allow(clippy::unit_arg)]
-            self.best_transactions.best_transactions(pool, attrs)
-        })? {
-            BuildOutcome::Better { payload, .. } => {
-                best_payload.set(payload);
-                self.metrics
-                    .total_block_built_duration
-                    .record(block_build_start_time.elapsed());
-                self.metrics.block_built_success.increment(1);
-                Ok(())
-            }
-            BuildOutcome::Freeze(payload) => {
-                best_payload.set(payload);
-                self.metrics
-                    .total_block_built_duration
-                    .record(block_build_start_time.elapsed());
-                Ok(())
-            }
-            BuildOutcome::Cancelled => {
-                tracing::warn!("Payload build cancelled");
-                Err(PayloadBuilderError::MissingPayload)
-            }
-            _ => {
-                tracing::warn!("No better payload found");
-                Err(PayloadBuilderError::MissingPayload)
+        let (tx, rx) = oneshot::channel();
+        let ctx = self.clone();
+        self.executor.spawn_blocking(Box::pin(async move {
+            match ctx.build_payload(args, |attrs| {
+                #[allow(clippy::unit_arg)]
+                ctx.best_transactions.best_transactions(pool, attrs)
+            }) {
+                Ok(BuildOutcome::Better { payload, .. }) => {
+                    best_payload.set(payload);
+                    ctx.metrics
+                        .total_block_built_duration
+                        .record(block_build_start_time.elapsed());
+                    ctx.metrics.block_built_success.increment(1);
+                    let _ = tx.send(Ok(()));
+                }
+                Ok(BuildOutcome::Freeze(payload)) => {
+                    best_payload.set(payload);
+                    ctx.metrics
+                        .total_block_built_duration
+                        .record(block_build_start_time.elapsed());
+                    let _ = tx.send(Ok(()));
+                }
+                Ok(BuildOutcome::Cancelled) => {
+                    tracing::warn!("Payload build cancelled");
+                    let _ = tx.send(Err(PayloadBuilderError::MissingPayload));
+                }
+                Ok(_) => {
+                    tracing::warn!("No better payload found");
+                    let _ = tx.send(Err(PayloadBuilderError::MissingPayload));
+                }
+                Err(err) => {
+                    tracing::warn!("Build payload error {}", err);
+                    let _ = tx.send(Err(err));
+                }
             }
-        }
+        }));
+        rx.blocking_recv()
+            .map_err(|err| PayloadBuilderError::from(err))?
     }
 }
 
-impl<Pool, Client, EvmConfig, N, T> OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N, T>
+impl<Pool, Client, EvmConfig, N, Tasks, T>
+    OpPayloadBuilderVanilla<Pool, Client, EvmConfig, N, Tasks, T>
 where
-    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = N::SignedTx>>,
-    Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks>,
+    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = N::SignedTx>> + 'static,
+    Client:
+        StateProviderFactory + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks> + 'static,
     N: OpPayloadPrimitives<_TX = OpTransactionSigned>,
     EvmConfig: ConfigureEvmFor<N>,
+    Tasks: TaskSpawner + Clone + Unpin + 'static,
 {
     /// Constructs an Optimism payload from the transactions sent via the
     /// Payload attributes by the sequencer. If the `no_tx_pool` argument is passed in