make pr

Author: rezbera

crates/optimism/flashblocks/src/cache.rs

@@ -17,6 +17,16 @@ use ringbuffer::{AllocRingBuffer, RingBuffer};
 use tokio::sync::broadcast;
 use tracing::*;
 
+type CachedSequenceEntry<P> = (
+    FlashBlockCompleteSequence<P>,
+    Vec<WithEncoded<Recovered<<P as FlashblockPayload>::SignedTx>>>,
+);
+
+type SequenceBuildArgs<P> = BuildArgs<
+    Vec<WithEncoded<Recovered<<P as FlashblockPayload>::SignedTx>>>,
+    <P as FlashblockPayload>::Base,
+>;
+
 /// Maximum number of cached sequences in the ring buffer.
 const CACHE_SIZE: usize = 3;
 /// 200 ms flashblock time.
@@ -37,7 +47,7 @@ pub(crate) struct SequenceManager<P: FlashblockPayload> {
     pending_transactions: Vec<WithEncoded<Recovered<P::SignedTx>>>,
     /// Ring buffer of recently completed sequences bundled with their decoded transactions (FIFO,
     /// size 3)
-    completed_cache: AllocRingBuffer<(FlashBlockCompleteSequence<P>, Vec<WithEncoded<Recovered<P::SignedTx>>>)>,
+    completed_cache: AllocRingBuffer<CachedSequenceEntry<P>>,
     /// Broadcast channel for completed sequences
     block_broadcaster: broadcast::Sender<FlashBlockCompleteSequence<P>>,
     /// Whether to compute state roots when building blocks
@@ -65,7 +75,9 @@ impl<P: FlashblockPayload> SequenceManager<P> {
     }
 
     /// Gets a subscriber to the flashblock sequences produced.
-    pub(crate) fn subscribe_block_sequence(&self) -> broadcast::Receiver<FlashBlockCompleteSequence<P>> {
+    pub(crate) fn subscribe_block_sequence(
+        &self,
+    ) -> broadcast::Receiver<FlashBlockCompleteSequence<P>> {
         self.block_broadcaster.subscribe()
     }
 
@@ -130,7 +142,7 @@ impl<P: FlashblockPayload> SequenceManager<P> {
         &mut self,
         local_tip_hash: B256,
         local_tip_timestamp: u64,
-    ) -> Option<BuildArgs<Vec<WithEncoded<Recovered<P::SignedTx>>>, P::Base>> {
+    ) -> Option<SequenceBuildArgs<P>> {
         // Try to find a buildable sequence: (base, last_fb, transactions, cached_state,
         // source_name)
         let (base, last_flashblock, transactions, cached_state, source_name) =
@@ -143,7 +155,7 @@ impl<P: FlashblockPayload> SequenceManager<P> {
             }
             // Priority 2: Try cached sequence with exact parent match
             else if let Some((cached, txs)) = self.completed_cache.iter().find(|(c, _)| c.payload_base().parent_hash() == local_tip_hash) {
-                let base = cached.payload_base().clone();
+                let base = cached.payload_base();
                 let last_fb = cached.last().clone();
                 let transactions = txs.clone();
                 let cached_state = None;
@@ -267,8 +279,7 @@ impl<P: FlashblockPayload> SequenceManager<P> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::test_utils::TestFlashBlockFactory;
-    use crate::FlashBlock;
+    use crate::{test_utils::TestFlashBlockFactory, FlashBlock};
     use alloy_primitives::B256;
 
     #[test]

crates/optimism/flashblocks/src/consensus.rs

@@ -1,5 +1,6 @@
 use crate::{
-    traits::FlashblockPayloadBase, FlashBlock, FlashBlockCompleteSequence, FlashBlockCompleteSequenceRx,
+    traits::FlashblockPayloadBase, FlashBlock, FlashBlockCompleteSequence,
+    FlashBlockCompleteSequenceRx,
 };
 use alloy_primitives::B256;
 use alloy_rpc_types_engine::PayloadStatusEnum;
@@ -30,7 +31,8 @@ where
 impl<P> FlashBlockConsensusClient<P>
 where
     P: PayloadTypes,
-    P::ExecutionData: for<'a> TryFrom<&'a FlashBlockCompleteSequence<FlashBlock>, Error: std::fmt::Display>,
+    P::ExecutionData:
+        for<'a> TryFrom<&'a FlashBlockCompleteSequence<FlashBlock>, Error: std::fmt::Display>,
 {
     /// Create a new `FlashBlockConsensusClient` with the given Op engine and sequence receiver.
     pub const fn new(

crates/optimism/flashblocks/src/lib.rs

@@ -26,7 +26,9 @@ mod payload;
 pub use payload::{FlashBlock, PendingFlashBlock};
 
 mod sequence;
-pub use sequence::{FlashBlockCompleteSequence, FlashBlockPendingSequence, SequenceExecutionOutcome};
+pub use sequence::{
+    FlashBlockCompleteSequence, FlashBlockPendingSequence, SequenceExecutionOutcome,
+};
 
 mod service;
 pub use service::{FlashBlockBuildInfo, FlashBlockService};

crates/optimism/flashblocks/src/op_impl.rs

@@ -51,8 +51,7 @@ impl FlashblockDiff for OpFlashblockPayloadDelta {
     }
 
     fn withdrawals(&self) -> Option<&Withdrawals> {
-        // OpFlashblockPayloadDelta stores Vec<Withdrawal>, not Withdrawals newtype
-        // This method isn't currently used in the flashblocks infrastructure
+        // TODO: Might not be needed as withdrawals aren't processed in a block except if at start or end
         None
     }
 
@@ -83,12 +82,12 @@ impl FlashblockPayload for OpFlashblockPayload {
     }
 
     fn block_number(&self) -> u64 {
-        OpFlashblockPayload::block_number(self)
+        Self::block_number(self)
     }
 
     fn recover_transactions(
         &self,
     ) -> impl Iterator<Item = Result<WithEncoded<Recovered<Self::SignedTx>>, RecoveryError>> {
-        OpFlashblockPayload::recover_transactions::<OpTxEnvelope>(self)
+        Self::recover_transactions::<OpTxEnvelope>(self)
     }
 }

crates/optimism/flashblocks/src/sequence.rs

@@ -210,7 +210,7 @@ impl<P: FlashblockPayload> FlashBlockCompleteSequence<P> {
     }
 
     /// Returns the number of flashblocks in the sequence.
-    pub fn count(&self) -> usize {
+    pub const fn count(&self) -> usize {
         self.inner.len()
     }
 
@@ -235,10 +235,7 @@ impl<P: FlashblockPayload> FlashBlockCompleteSequence<P> {
     /// Returns all transactions from all flashblocks in the sequence
     pub fn all_transactions(&self) -> Vec<Bytes> {
         use crate::traits::FlashblockDiff;
-        self.inner
-            .iter()
-            .flat_map(|fb| fb.diff().transactions_raw().iter().cloned())
-            .collect()
+        self.inner.iter().flat_map(|fb| fb.diff().transactions_raw().iter().cloned()).collect()
     }
 
     /// Returns an iterator over all flashblocks in the sequence.
@@ -306,7 +303,8 @@ mod tests {
 
         #[test]
         fn test_insert_ignores_different_block_number() {
-            let mut sequence: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut sequence: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             let fb0 = factory.flashblock_at(0).build();
@@ -322,7 +320,8 @@ mod tests {
 
         #[test]
         fn test_insert_ignores_different_payload_id() {
-            let mut sequence: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut sequence: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             let fb0 = factory.flashblock_at(0).build();
@@ -340,7 +339,8 @@ mod tests {
 
         #[test]
         fn test_insert_maintains_btree_order() {
-            let mut sequence: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut sequence: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             let fb0 = factory.flashblock_at(0).build();
@@ -362,7 +362,8 @@ mod tests {
 
         #[test]
         fn test_finalize_empty_sequence_fails() {
-            let mut sequence: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut sequence: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let result = sequence.finalize();
 
             assert!(result.is_err());
@@ -374,7 +375,8 @@ mod tests {
 
         #[test]
         fn test_finalize_clears_pending_state() {
-            let mut sequence: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut sequence: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             let fb0 = factory.flashblock_at(0).build();
@@ -391,7 +393,8 @@ mod tests {
 
         #[test]
         fn test_finalize_preserves_execution_outcome() {
-            let mut sequence: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut sequence: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             let fb0 = factory.flashblock_at(0).build();
@@ -408,7 +411,8 @@ mod tests {
 
         #[test]
         fn test_finalize_clears_cached_reads() {
-            let mut sequence: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut sequence: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             let fb0 = factory.flashblock_at(0).build();
@@ -426,7 +430,8 @@ mod tests {
 
         #[test]
         fn test_finalize_multiple_times_after_refill() {
-            let mut sequence: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut sequence: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             // First sequence
@@ -616,7 +621,8 @@ mod tests {
 
         #[test]
         fn test_try_from_pending_to_complete_valid() {
-            let mut pending: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut pending: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             let fb0 = factory.flashblock_at(0).build();
@@ -637,7 +643,8 @@ mod tests {
 
         #[test]
         fn test_try_from_preserves_execution_outcome() {
-            let mut pending: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut pending: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             let fb0 = factory.flashblock_at(0).build();
@@ -657,7 +664,8 @@ mod tests {
 
         #[test]
         fn test_last_flashblock_returns_highest_index() {
-            let mut sequence: FlashBlockPendingSequence<FlashBlock> = FlashBlockPendingSequence::new();
+            let mut sequence: FlashBlockPendingSequence<FlashBlock> =
+                FlashBlockPendingSequence::new();
             let factory = TestFlashBlockFactory::new();
 
             let fb0 = factory.flashblock_at(0).build();

crates/optimism/flashblocks/src/service.rs

@@ -51,9 +51,8 @@ where
     N: NodePrimitives,
     P: FlashblockPayload<SignedTx = N::SignedTx>,
     S: Stream<Item = eyre::Result<P>> + Unpin + 'static,
-    EvmConfig: ConfigureEvm<Primitives = N, NextBlockEnvCtx: From<P::Base> + Unpin>
-        + Clone
-        + 'static,
+    EvmConfig:
+        ConfigureEvm<Primitives = N, NextBlockEnvCtx: From<P::Base> + Unpin> + Clone + 'static,
     Provider: StateProviderFactory
         + BlockReaderIdExt<
             Header = HeaderTy<N>,
@@ -92,7 +91,9 @@ where
     }
 
     /// Returns the sender half to the flashblock sequence.
-    pub const fn block_sequence_broadcaster(&self) -> &broadcast::Sender<FlashBlockCompleteSequence<P>> {
+    pub const fn block_sequence_broadcaster(
+        &self,
+    ) -> &broadcast::Sender<FlashBlockCompleteSequence<P>> {
         self.sequences.block_sequence_broadcaster()
     }
 

crates/optimism/flashblocks/src/traits.rs

@@ -86,5 +86,10 @@ pub trait FlashblockPayload:
     /// or an error if decoding/recovery failed.
     fn recover_transactions(
         &self,
-    ) -> impl Iterator<Item = Result<alloy_eips::eip2718::WithEncoded<reth_primitives_traits::Recovered<Self::SignedTx>>, RecoveryError>>;
+    ) -> impl Iterator<
+        Item = Result<
+            alloy_eips::eip2718::WithEncoded<reth_primitives_traits::Recovered<Self::SignedTx>>,
+            RecoveryError,
+        >,
+    >;
 }

crates/optimism/flashblocks/src/worker.rs

@@ -144,8 +144,14 @@ where
     }
 }
 
-impl<EvmConfig: Clone, Provider: Clone, Base> Clone for FlashBlockBuilder<EvmConfig, Provider, Base> {
+impl<EvmConfig: Clone, Provider: Clone, Base> Clone
+    for FlashBlockBuilder<EvmConfig, Provider, Base>
+{
     fn clone(&self) -> Self {
-        Self { evm_config: self.evm_config.clone(), provider: self.provider.clone(), _base: PhantomData }
+        Self {
+            evm_config: self.evm_config.clone(),
+            provider: self.provider.clone(),
+            _base: PhantomData,
+        }
     }
 }

crates/optimism/flashblocks/tests/it/stream.rs

@@ -1,11 +1,12 @@
 use futures_util::stream::StreamExt;
+use op_alloy_rpc_types_engine::OpFlashblockPayload;
 use reth_optimism_flashblocks::WsFlashBlockStream;
 
 #[tokio::test]
 async fn test_streaming_flashblocks_from_remote_source_is_successful() {
     let items = 3;
     let ws_url = "wss://sepolia.flashblocks.base.org/ws".parse().unwrap();
-    let stream = WsFlashBlockStream::new(ws_url);
+    let stream: WsFlashBlockStream<_, _, _, OpFlashblockPayload> = WsFlashBlockStream::new(ws_url);
 
     let blocks: Vec<_> = stream.take(items).collect().await;
 

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

@@ -16,19 +16,19 @@ use alloy_consensus::BlockHeader;
 use alloy_primitives::{B256, U256};
 use eyre::WrapErr;
 use op_alloy_network::Optimism;
-use op_alloy_rpc_types_engine::{OpFlashblockPayload};
+use op_alloy_rpc_types_engine::OpFlashblockPayload;
 pub use receipt::{OpReceiptBuilder, OpReceiptFieldsBuilder};
 use reqwest::Url;
 use reth_chainspec::{EthereumHardforks, Hardforks};
 use reth_evm::ConfigureEvm;
 use reth_node_api::{FullNodeComponents, FullNodeTypes, HeaderTy, NodeTypes};
-use reth_primitives_traits::NodePrimitives;
 use reth_node_builder::rpc::{EthApiBuilder, EthApiCtx};
 use reth_optimism_flashblocks::{
     FlashBlockBuildInfo, FlashBlockCompleteSequence, FlashBlockCompleteSequenceRx,
-    FlashBlockService, FlashblocksListeners, FlashblockPayload, PendingBlockRx, PendingFlashBlock,
+    FlashBlockService, FlashblockPayload, FlashblocksListeners, PendingBlockRx, PendingFlashBlock,
     WsFlashBlockStream,
 };
+use reth_primitives_traits::NodePrimitives;
 use reth_rpc::eth::core::EthApiInner;
 use reth_rpc_eth_api::{
     helpers::{
@@ -366,8 +366,11 @@ impl<N: RpcNodeCore, Rpc: RpcConvert, F: FlashblockPayload> fmt::Debug for OpEth
 }
 
 /// Container type `OpEthApi`
-pub struct OpEthApiInner<N: RpcNodeCore, Rpc: RpcConvert, F: FlashblockPayload = OpFlashblockPayload>
-{
+pub struct OpEthApiInner<
+    N: RpcNodeCore,
+    Rpc: RpcConvert,
+    F: FlashblockPayload = OpFlashblockPayload,
+> {
     /// Gateway to node's core components.
     eth_api: EthApiNodeBackend<N, Rpc>,
     /// Sequencer client, configured to forward submitted transactions to sequencer of given OP
@@ -383,7 +386,9 @@ pub struct OpEthApiInner<N: RpcNodeCore, Rpc: RpcConvert, F: FlashblockPayload =
     flashblocks: Option<FlashblocksListeners<N::Primitives, F>>,
 }
 
-impl<N: RpcNodeCore, Rpc: RpcConvert, F: FlashblockPayload> fmt::Debug for OpEthApiInner<N, Rpc, F> {
+impl<N: RpcNodeCore, Rpc: RpcConvert, F: FlashblockPayload> fmt::Debug
+    for OpEthApiInner<N, Rpc, F>
+{
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         f.debug_struct("OpEthApiInner").finish()
     }
@@ -562,8 +567,9 @@ where
             let in_progress_rx = service.subscribe_in_progress();
             ctx.components.task_executor().spawn(Box::pin(service.run(tx)));
 
-
-            if flashblock_consensus { todo!("Modularize FlashBlockConsensusClient?") }
+            if flashblock_consensus {
+                todo!("Modularize FlashBlockConsensusClient?")
+            }
             // if flashblock_consensus {
             //     info!(target: "reth::cli", "Launching FlashBlockConsensusClient");
             //     let flashblock_client = FlashBlockConsensusClient::new(