feat: wait for new blocks when build is in progress (#18831)

Co-authored-by: Roman Hodulák <roman.hodulak@polyglot-software.com>

Author: lean-apple

crates/optimism/flashblocks/src/lib.rs

@@ -4,7 +4,7 @@ pub use payload::{
     ExecutionPayloadBaseV1, ExecutionPayloadFlashblockDeltaV1, FlashBlock, FlashBlockDecoder,
     Metadata,
 };
-pub use service::FlashBlockService;
+pub use service::{FlashBlockBuildInfo, FlashBlockService};
 pub use ws::{WsConnect, WsFlashBlockStream};
 
 mod consensus;
@@ -28,3 +28,6 @@ pub type PendingBlockRx<N> = tokio::sync::watch::Receiver<Option<PendingFlashBlo
 /// [`FlashBlock`]: crate::FlashBlock
 pub type FlashBlockCompleteSequenceRx =
     tokio::sync::broadcast::Receiver<FlashBlockCompleteSequence>;
+
+/// Receiver that signals whether a [`FlashBlock`] is currently being built.
+pub type InProgressFlashBlockRx = tokio::sync::watch::Receiver<Option<FlashBlockBuildInfo>>;

crates/optimism/flashblocks/src/service.rs

@@ -1,7 +1,8 @@
 use crate::{
     sequence::FlashBlockPendingSequence,
     worker::{BuildArgs, FlashBlockBuilder},
-    ExecutionPayloadBaseV1, FlashBlock, FlashBlockCompleteSequenceRx, PendingFlashBlock,
+    ExecutionPayloadBaseV1, FlashBlock, FlashBlockCompleteSequenceRx, InProgressFlashBlockRx,
+    PendingFlashBlock,
 };
 use alloy_eips::eip2718::WithEncoded;
 use alloy_primitives::B256;
@@ -21,7 +22,10 @@ use std::{
     task::{ready, Context, Poll},
     time::Instant,
 };
-use tokio::{pin, sync::oneshot};
+use tokio::{
+    pin,
+    sync::{oneshot, watch},
+};
 use tracing::{debug, trace, warn};
 
 pub(crate) const FB_STATE_ROOT_FROM_INDEX: usize = 9;
@@ -48,11 +52,25 @@ pub struct FlashBlockService<
     /// when fb received on top of the same block. Avoid redundant I/O across multiple
     /// executions within the same block.
     cached_state: Option<(B256, CachedReads)>,
+    /// Signals when a block build is in progress
+    in_progress_tx: watch::Sender<Option<FlashBlockBuildInfo>>,
+    /// `FlashBlock` service's metrics
     metrics: FlashBlockServiceMetrics,
     /// Enable state root calculation from flashblock with index [`FB_STATE_ROOT_FROM_INDEX`]
     compute_state_root: bool,
 }
 
+/// Information for a flashblock currently built
+#[derive(Debug, Clone, Copy)]
+pub struct FlashBlockBuildInfo {
+    /// Parent block hash
+    pub parent_hash: B256,
+    /// Flashblock index within the current block's sequence
+    pub index: u64,
+    /// Block number of the flashblock being built.
+    pub block_number: u64,
+}
+
 impl<N, S, EvmConfig, Provider> FlashBlockService<N, S, EvmConfig, Provider>
 where
     N: NodePrimitives,
@@ -73,6 +91,7 @@ where
 {
     /// Constructs a new `FlashBlockService` that receives [`FlashBlock`]s from `rx` stream.
     pub fn new(rx: S, evm_config: EvmConfig, provider: Provider, spawner: TaskExecutor) -> Self {
+        let (in_progress_tx, _) = watch::channel(None);
         Self {
             rx,
             current: None,
@@ -83,6 +102,7 @@ where
             spawner,
             job: None,
             cached_state: None,
+            in_progress_tx,
             metrics: FlashBlockServiceMetrics::default(),
             compute_state_root: false,
         }
@@ -99,6 +119,11 @@ where
         self.blocks.subscribe_block_sequence()
     }
 
+    /// Returns a receiver that signals when a flashblock is being built.
+    pub fn subscribe_in_progress(&self) -> InProgressFlashBlockRx {
+        self.in_progress_tx.subscribe()
+    }
+
     /// Drives the services and sends new blocks to the receiver
     ///
     /// Note: this should be spawned
@@ -218,6 +243,8 @@ where
             };
             // reset job
             this.job.take();
+            // No build in progress
+            let _ = this.in_progress_tx.send(None);
 
             if let Some((now, result)) = result {
                 match result {
@@ -293,6 +320,13 @@ where
             if let Some(args) = this.build_args() {
                 let now = Instant::now();
 
+                let fb_info = FlashBlockBuildInfo {
+                    parent_hash: args.base.parent_hash,
+                    index: args.last_flashblock_index,
+                    block_number: args.base.block_number,
+                };
+                // Signal that a flashblock build has started with build metadata
+                let _ = this.in_progress_tx.send(Some(fb_info));
                 let (tx, rx) = oneshot::channel();
                 let builder = this.builder.clone();
 

crates/optimism/rpc/src/eth/mod.rs

@@ -13,7 +13,7 @@ use crate::{
     OpEthApiError, SequencerClient,
 };
 use alloy_consensus::BlockHeader;
-use alloy_primitives::U256;
+use alloy_primitives::{B256, U256};
 use eyre::WrapErr;
 use op_alloy_network::Optimism;
 pub use receipt::{OpReceiptBuilder, OpReceiptFieldsBuilder};
@@ -23,8 +23,8 @@ use reth_evm::ConfigureEvm;
 use reth_node_api::{FullNodeComponents, FullNodeTypes, HeaderTy, NodeTypes};
 use reth_node_builder::rpc::{EthApiBuilder, EthApiCtx};
 use reth_optimism_flashblocks::{
-    ExecutionPayloadBaseV1, FlashBlockCompleteSequenceRx, FlashBlockService, PendingBlockRx,
-    WsFlashBlockStream,
+    ExecutionPayloadBaseV1, FlashBlockBuildInfo, FlashBlockCompleteSequenceRx, FlashBlockService,
+    InProgressFlashBlockRx, PendingBlockRx, PendingFlashBlock, WsFlashBlockStream,
 };
 use reth_rpc::eth::{core::EthApiInner, DevSigner};
 use reth_rpc_eth_api::{
@@ -43,10 +43,18 @@ use reth_tasks::{
     pool::{BlockingTaskGuard, BlockingTaskPool},
     TaskSpawner,
 };
-use std::{fmt, fmt::Formatter, marker::PhantomData, sync::Arc, time::Instant};
-use tokio::sync::watch;
+use std::{
+    fmt::{self, Formatter},
+    marker::PhantomData,
+    sync::Arc,
+    time::Duration,
+};
+use tokio::{sync::watch, time};
 use tracing::info;
 
+/// Maximum duration to wait for a fresh flashblock when one is being built.
+const MAX_FLASHBLOCK_WAIT_DURATION: Duration = Duration::from_millis(50);
+
 /// Adapter for [`EthApiInner`], which holds all the data required to serve core `eth_` API.
 pub type EthApiNodeBackend<N, Rpc> = EthApiInner<N, Rpc>;
 
@@ -79,13 +87,15 @@ impl<N: RpcNodeCore, Rpc: RpcConvert> OpEthApi<N, Rpc> {
         min_suggested_priority_fee: U256,
         pending_block_rx: Option<PendingBlockRx<N::Primitives>>,
         flashblock_rx: Option<FlashBlockCompleteSequenceRx>,
+        in_progress_rx: Option<InProgressFlashBlockRx>,
     ) -> Self {
         let inner = Arc::new(OpEthApiInner {
             eth_api,
             sequencer_client,
             min_suggested_priority_fee,
             pending_block_rx,
             flashblock_rx,
+            in_progress_rx,
         });
         Self { inner }
     }
@@ -109,15 +119,57 @@ impl<N: RpcNodeCore, Rpc: RpcConvert> OpEthApi<N, Rpc> {
         self.inner.flashblock_rx.as_ref().map(|rx| rx.resubscribe())
     }
 
+    /// Returns information about the flashblock currently being built, if any.
+    fn flashblock_build_info(&self) -> Option<FlashBlockBuildInfo> {
+        self.inner.in_progress_rx.as_ref().and_then(|rx| *rx.borrow())
+    }
+
+    /// Extracts pending block if it matches the expected parent hash.
+    fn extract_matching_block(
+        &self,
+        block: Option<&PendingFlashBlock<N::Primitives>>,
+        parent_hash: B256,
+    ) -> Option<PendingBlock<N::Primitives>> {
+        block.filter(|b| b.block().parent_hash() == parent_hash).map(|b| b.pending.clone())
+    }
+
     /// Build a [`OpEthApi`] using [`OpEthApiBuilder`].
     pub const fn builder() -> OpEthApiBuilder<Rpc> {
         OpEthApiBuilder::new()
     }
 
+    /// Awaits a fresh flashblock if one is being built, otherwise returns current.
+    async fn flashblock(
+        &self,
+        parent_hash: B256,
+    ) -> eyre::Result<Option<PendingBlock<N::Primitives>>> {
+        let Some(rx) = self.inner.pending_block_rx.as_ref() else { return Ok(None) };
+
+        // Check if a flashblock is being built
+        if let Some(build_info) = self.flashblock_build_info() {
+            let current_index = rx.borrow().as_ref().map(|b| b.last_flashblock_index);
+
+            // Check if this is the first flashblock or the next consecutive index
+            let is_next_index = current_index.is_none_or(|idx| build_info.index == idx + 1);
+
+            // Wait only for relevant flashblocks: matching parent and next in sequence
+            if build_info.parent_hash == parent_hash && is_next_index {
+                let mut rx_clone = rx.clone();
+                // Wait up to MAX_FLASHBLOCK_WAIT_DURATION for a new flashblock to arrive
+                let _ = time::timeout(MAX_FLASHBLOCK_WAIT_DURATION, rx_clone.changed()).await;
+            }
+        }
+
+        // Fall back to current block
+        Ok(self.extract_matching_block(rx.borrow().as_ref(), parent_hash))
+    }
+
     /// Returns a [`PendingBlock`] that is built out of flashblocks.
     ///
     /// If flashblocks receiver is not set, then it always returns `None`.
-    pub fn pending_flashblock(&self) -> eyre::Result<Option<PendingBlock<N::Primitives>>>
+    ///
+    /// It may wait up to 50ms for a fresh flashblock if one is currently being built.
+    pub async fn pending_flashblock(&self) -> eyre::Result<Option<PendingBlock<N::Primitives>>>
     where
         OpEthApiError: FromEvmError<N::Evm>,
         Rpc: RpcConvert<Primitives = N::Primitives>,
@@ -128,21 +180,7 @@ impl<N: RpcNodeCore, Rpc: RpcConvert> OpEthApi<N, Rpc> {
             PendingBlockEnvOrigin::DerivedFromLatest(parent) => parent,
         };
 
-        let Some(rx) = self.inner.pending_block_rx.as_ref() else { return Ok(None) };
-        let pending_block = rx.borrow();
-        let Some(pending_block) = pending_block.as_ref() else { return Ok(None) };
-
-        let now = Instant::now();
-
-        // Is the pending block not expired and latest is its parent?
-        if pending.evm_env.block_env.number == U256::from(pending_block.block().number()) &&
-            parent.hash() == pending_block.block().parent_hash() &&
-            now <= pending_block.expires_at
-        {
-            return Ok(Some(pending_block.pending.clone()));
-        }
-
-        Ok(None)
+        self.flashblock(parent.hash()).await
     }
 }
 
@@ -330,6 +368,8 @@ pub struct OpEthApiInner<N: RpcNodeCore, Rpc: RpcConvert> {
     ///
     /// If set, then it provides sequences of flashblock built.
     flashblock_rx: Option<FlashBlockCompleteSequenceRx>,
+    /// Receiver that signals when a flashblock is being built
+    in_progress_rx: Option<InProgressFlashBlockRx>,
 }
 
 impl<N: RpcNodeCore, Rpc: RpcConvert> fmt::Debug for OpEthApiInner<N, Rpc> {
@@ -465,24 +505,28 @@ where
             None
         };
 
-        let rxs = if let Some(ws_url) = flashblocks_url {
-            info!(target: "reth:cli", %ws_url,  "Launching flashblocks service");
-            let (tx, pending_block_rx) = watch::channel(None);
-            let stream = WsFlashBlockStream::new(ws_url);
-            let service = FlashBlockService::new(
-                stream,
-                ctx.components.evm_config().clone(),
-                ctx.components.provider().clone(),
-                ctx.components.task_executor().clone(),
-            );
-            let flashblock_rx = service.subscribe_block_sequence();
-            ctx.components.task_executor().spawn(Box::pin(service.run(tx)));
-            Some((pending_block_rx, flashblock_rx))
-        } else {
-            None
-        };
+        let (pending_block_rx, flashblock_rx, in_progress_rx) =
+            if let Some(ws_url) = flashblocks_url {
+                info!(target: "reth:cli", %ws_url, "Launching flashblocks service");
+
+                let (tx, pending_rx) = watch::channel(None);
+                let stream = WsFlashBlockStream::new(ws_url);
+                let service = FlashBlockService::new(
+                    stream,
+                    ctx.components.evm_config().clone(),
+                    ctx.components.provider().clone(),
+                    ctx.components.task_executor().clone(),
+                );
+
+                let flashblock_rx = service.subscribe_block_sequence();
+                let in_progress_rx = service.subscribe_in_progress();
+
+                ctx.components.task_executor().spawn(Box::pin(service.run(tx)));
 
-        let (pending_block_rx, flashblock_rx) = rxs.unzip();
+                (Some(pending_rx), Some(flashblock_rx), Some(in_progress_rx))
+            } else {
+                (None, None, None)
+            };
 
         let eth_api = ctx.eth_api_builder().with_rpc_converter(rpc_converter).build_inner();
 
@@ -492,6 +536,7 @@ where
             U256::from(min_suggested_priority_fee),
             pending_block_rx,
             flashblock_rx,
+            in_progress_rx,
         ))
     }
 }

crates/optimism/rpc/src/eth/pending_block.rs

@@ -42,7 +42,7 @@ where
     async fn local_pending_block(
         &self,
     ) -> Result<Option<BlockAndReceipts<Self::Primitives>>, Self::Error> {
-        if let Ok(Some(pending)) = self.pending_flashblock() {
+        if let Ok(Some(pending)) = self.pending_flashblock().await {
             return Ok(Some(pending.into_block_and_receipts()));
         }
 
@@ -70,7 +70,7 @@ where
     where
         Self: SpawnBlocking,
     {
-        let Ok(Some(pending_block)) = self.pending_flashblock() else {
+        let Ok(Some(pending_block)) = self.pending_flashblock().await else {
             return Ok(None);
         };
 

crates/optimism/rpc/src/eth/transaction.rs

@@ -127,7 +127,7 @@ where
                             }
                         } => {
                             // Check flashblocks for faster confirmation (Optimism-specific)
-                            if let Ok(Some(pending_block)) = this.pending_flashblock() {
+                            if let Ok(Some(pending_block)) = this.pending_flashblock().await {
                                 let block_and_receipts = pending_block.into_block_and_receipts();
                                 if block_and_receipts.block.body().contains_transaction(&hash)
                                     && let Some(receipt) = this.transaction_receipt(hash).await? {
@@ -168,7 +168,7 @@ where
 
             if tx_receipt.is_none() {
                 // if flashblocks are supported, attempt to find id from the pending block
-                if let Ok(Some(pending_block)) = this.pending_flashblock() {
+                if let Ok(Some(pending_block)) = this.pending_flashblock().await {
                     let block_and_receipts = pending_block.into_block_and_receipts();
                     if let Some((tx, receipt)) =
                         block_and_receipts.find_transaction_and_receipt_by_hash(hash)