Add logic

Author: SozinM

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

@@ -24,7 +24,7 @@ use reth_node_api::{Block, NodePrimitives, PayloadBuilderError};
 use reth_optimism_consensus::{calculate_receipt_root_no_memo_optimism, isthmus};
 use reth_optimism_evm::{OpEvmConfig, OpNextBlockEnvAttributes};
 use reth_optimism_forks::OpHardforks;
-use reth_optimism_node::{OpBuiltPayload, OpPayloadBuilderAttributes};
+use reth_optimism_node::{OpBuiltPayload, OpEngineTypes, OpPayloadBuilderAttributes};
 use reth_optimism_primitives::{OpPrimitives, OpReceipt, OpTransactionSigned};
 use reth_payload_util::BestPayloadTransactions;
 use reth_primitives_traits::RecoveredBlock;
@@ -48,6 +48,8 @@ use std::{
     },
     time::Instant,
 };
+use std::sync::Mutex;
+use reth_payload_builder_primitives::Events;
 use tokio::sync::{
     mpsc,
     mpsc::{error::SendError, Sender},
@@ -119,6 +121,8 @@ pub struct OpPayloadBuilder<Pool, Client, BT> {
     /// The end of builder transaction type
     #[allow(dead_code)]
     pub builder_tx: BT,
+    /// Builder events handle to send BuiltPayload events
+    payload_builder_handle: Arc<Mutex<Option<tokio::sync::broadcast::Sender<Events<OpEngineTypes>>>>>
 }
 
 impl<Pool, Client, BT> OpPayloadBuilder<Pool, Client, BT> {
@@ -129,6 +133,7 @@ impl<Pool, Client, BT> OpPayloadBuilder<Pool, Client, BT> {
         client: Client,
         config: BuilderConfig<FlashblocksConfig>,
         builder_tx: BT,
+        payload_builder_handle: Arc<Mutex<Option<tokio::sync::broadcast::Sender<Events<OpEngineTypes>>>>>
     ) -> eyre::Result<Self> {
         let metrics = Arc::new(OpRBuilderMetrics::default());
         let ws_pub = WebSocketPublisher::new(config.specific.ws_addr, Arc::clone(&metrics))?.into();
@@ -140,6 +145,7 @@ impl<Pool, Client, BT> OpPayloadBuilder<Pool, Client, BT> {
             config,
             metrics,
             builder_tx,
+            payload_builder_handle,
         })
     }
 }
@@ -290,6 +296,8 @@ where
         let (payload, fb_payload) = build_block(&mut state, &ctx, &mut info)?;
 
         best_payload.set(payload.clone());
+        self.send_payload_to_engine(payload);
+
         self.ws_pub
             .publish(&fb_payload)
             .map_err(PayloadBuilderError::other)?;
@@ -536,6 +544,7 @@ where
                                 .record(info.executed_transactions.len() as f64);
 
                             best_payload.set(new_payload.clone());
+                            self.send_payload_to_engine(new_payload);
                             // Update bundle_state for next iteration
                             total_gas_per_batch += gas_per_batch;
                             if let Some(da_limit) = da_per_batch {
@@ -605,6 +614,27 @@ where
         span.record("flashblock_count", ctx.flashblock_index());
     }
 
+    /// Sends built payload via payload builder handle broadcast channel to the engine
+    pub fn send_payload_to_engine(&self, payload: OpBuiltPayload) {
+        // Send built payload as create one
+        match self.payload_builder_handle.lock().as_deref() {
+            Ok(Some(handle)) => {
+                let res = handle.send(Events::BuiltPayload(payload.clone().into()));
+                if let Err(e) = res {
+                    error!(
+                        message = "Failed to send payload via payload builder handle",
+                        error = ?e,
+                    );
+                }
+            },
+            Ok(None) => error!(message = "Payload builder handle is not setup, skipping sending payload"),
+            Err(e) => error!(
+                message = "Failed to get access to payload builder handle",
+                error = ?e.
+            ),
+        }
+    }
+
     /// Spawn task that will send new flashblock level cancel token in steady intervals (first interval
     /// may vary if --flashblocks.dynamic enabled)
     pub fn spawn_timer_task(

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

@@ -1,3 +1,4 @@
+use std::sync::{Arc, Mutex};
 use super::{payload::OpPayloadBuilder, FlashblocksConfig};
 use crate::{
     builders::{
@@ -28,12 +29,17 @@ impl FlashblocksServiceBuilder {
         Pool: PoolBounds,
         BT: BuilderTx + Unpin + Clone + Send + Sync + 'static,
     {
+
+        let handle = Arc::new(Mutex::new(None));
+        let payload_builder_handle = handle.clone();
+
         let payload_builder = OpPayloadBuilder::new(
             OpEvmConfig::optimism(ctx.chain_spec()),
             pool,
             ctx.provider().clone(),
             self.0.clone(),
             builder_tx,
+            handle,
         )?;
 
         let payload_job_config = BasicPayloadJobGeneratorConfig::default();
@@ -50,6 +56,8 @@ impl FlashblocksServiceBuilder {
         let (payload_service, payload_builder) =
             PayloadBuilderService::new(payload_generator, ctx.provider().canonical_state_stream());
 
+        payload_builder_handle.lock()?.replace(payload_service.payload_events_handle());
+
         ctx.task_executor()
             .spawn_critical("custom payload builder service", Box::pin(payload_service));
 

crates/op-rbuilder/src/builders/generator.rs

@@ -8,9 +8,7 @@ use reth_basic_payload_builder::{
     BasicPayloadJobGeneratorConfig, HeaderForPayload, PayloadConfig, PrecachedState,
 };
 use reth_node_api::{NodePrimitives, PayloadBuilderAttributes, PayloadKind};
-use reth_payload_builder::{
-    KeepPayloadJobAlive, PayloadBuilderError, PayloadJob, PayloadJobGenerator,
-};
+use reth_payload_builder::{KeepPayloadJobAlive, PayloadBuilderError, PayloadBuilderHandle, PayloadJob, PayloadJobGenerator};
 use reth_payload_primitives::BuiltPayload;
 use reth_primitives_traits::HeaderTy;
 use reth_provider::CanonStateNotification;
@@ -199,6 +197,7 @@ where
             deadline,
             build_complete: None,
             cached_reads: self.maybe_pre_cached(parent_header.hash()),
+            payload_builder_handle: self.payload_builder_handle.clone(),
         };
 
         job.spawn_build_job();
@@ -236,6 +235,9 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
+use tokio::sync::broadcast::Sender;
+use reth_optimism_node::OpEngineTypes;
+use reth_payload_builder_primitives::Events;
 
 /// A [PayloadJob] that builds empty blocks.
 pub struct BlockPayloadJob<Tasks, Builder>
@@ -261,6 +263,7 @@ where
     /// This is used to avoid reading the same state over and over again when new attempts are
     /// triggered, because during the building process we'll repeatedly execute the transactions.
     pub(crate) cached_reads: Option<CachedReads>,
+    pub(crate) payload_builder_handle: Arc<Mutex<Option<Sender<Events<OpEngineTypes>>>>>,
 }
 
 impl<Tasks, Builder> PayloadJob for BlockPayloadJob<Tasks, Builder>