chore(flashblocks): modularize types (#254)

Author: refcell

crates/flashblocks/src/blocks.rs

@@ -0,0 +1,33 @@
+//! Contains the [`Flashblock`] and [`Metadata`] types used in Flashblocks.
+
+use alloy_primitives::{Address, B256, U256, map::foldhash::HashMap};
+use alloy_rpc_types_engine::PayloadId;
+use reth_optimism_primitives::OpReceipt;
+use rollup_boost::{ExecutionPayloadBaseV1, ExecutionPayloadFlashblockDeltaV1};
+use serde::{Deserialize, Serialize};
+
+/// Metadata associated with a flashblock.
+#[derive(Debug, Deserialize, Serialize, Clone, Default)]
+pub struct Metadata {
+    /// Transaction receipts indexed by hash.
+    pub receipts: HashMap<B256, OpReceipt>,
+    /// Updated account balances.
+    pub new_account_balances: HashMap<Address, U256>,
+    /// Block number this flashblock belongs to.
+    pub block_number: u64,
+}
+
+/// A flashblock containing partial block data.
+#[derive(Debug, Clone)]
+pub struct Flashblock {
+    /// Unique payload identifier.
+    pub payload_id: PayloadId,
+    /// Index of this flashblock within the block.
+    pub index: u64,
+    /// Base payload data (only present on first flashblock).
+    pub base: Option<ExecutionPayloadBaseV1>,
+    /// Delta containing transactions and state changes.
+    pub diff: ExecutionPayloadFlashblockDeltaV1,
+    /// Associated metadata.
+    pub metadata: Metadata,
+}

crates/flashblocks/src/lib.rs

@@ -4,13 +4,19 @@
 #![cfg_attr(not(test), warn(unused_crate_dependencies))]
 
 mod metrics;
-pub(crate) use metrics::Metrics;
+pub use metrics::Metrics;
 
 mod pending_blocks;
+pub use pending_blocks::{PendingBlocks, PendingBlocksBuilder};
 
-pub mod state;
+mod state;
+pub use state::FlashblocksState;
 
-pub mod subscription;
-pub mod traits;
+mod subscription;
+pub use subscription::{FlashblocksReceiver, FlashblocksSubscriber};
 
-pub use pending_blocks::PendingBlocks;
+mod traits;
+pub use traits::{FlashblocksAPI, PendingBlocksAPI};
+
+mod blocks;
+pub use blocks::{Flashblock, Metadata};

crates/flashblocks/src/metrics.rs

@@ -1,45 +1,59 @@
+//! Metrics for flashblocks.
+
 use metrics::{Counter, Gauge, Histogram};
 use metrics_derive::Metrics;
+
 /// Metrics for the `reth_flashblocks` component.
 /// Conventions:
 /// - Durations are recorded in seconds (histograms).
 /// - Counters are monotonic event counts.
 /// - Gauges reflect the current value/state.
 #[derive(Metrics, Clone)]
 #[metrics(scope = "reth_flashblocks")]
-pub(crate) struct Metrics {
+pub struct Metrics {
+    /// Count of times upstream receiver was closed/errored.
     #[metric(describe = "Count of times upstream receiver was closed/errored")]
     pub upstream_errors: Counter,
 
+    /// Count of messages received from the upstream source.
     #[metric(describe = "Count of messages received from the upstream source")]
     pub upstream_messages: Counter,
 
+    /// Time taken to process a message.
     #[metric(describe = "Time taken to process a message")]
     pub block_processing_duration: Histogram,
 
+    /// Number of Flashblocks that arrive in an unexpected order.
     #[metric(describe = "Number of Flashblocks that arrive in an unexpected order")]
     pub unexpected_block_order: Counter,
 
+    /// Number of flashblocks contained within a single block.
     #[metric(describe = "Number of flashblocks in a block")]
     pub flashblocks_in_block: Histogram,
 
+    /// Count of times flashblocks are unable to be converted to blocks.
     #[metric(describe = "Count of times flashblocks are unable to be converted to blocks")]
     pub block_processing_error: Counter,
 
+    /// Count of times pending snapshot was cleared because canonical caught up.
     #[metric(
         describe = "Number of times pending snapshot was cleared because canonical caught up"
     )]
     pub pending_clear_catchup: Counter,
 
+    /// Number of times pending snapshot was cleared because of reorg.
     #[metric(describe = "Number of times pending snapshot was cleared because of reorg")]
     pub pending_clear_reorg: Counter,
 
+    /// Pending snapshot flashblock index (current).
     #[metric(describe = "Pending snapshot flashblock index (current)")]
     pub pending_snapshot_fb_index: Gauge,
 
+    /// Pending snapshot block number (current).
     #[metric(describe = "Pending snapshot block number (current)")]
     pub pending_snapshot_height: Gauge,
 
+    /// Total number of WebSocket reconnection attempts.
     #[metric(describe = "Total number of WebSocket reconnection attempts")]
     pub reconnect_attempts: Counter,
 }

crates/flashblocks/src/pending_blocks.rs

@@ -17,9 +17,11 @@ use reth::revm::{db::Cache, state::EvmState};
 use reth_rpc_convert::RpcTransaction;
 use reth_rpc_eth_api::{RpcBlock, RpcReceipt};
 
-use crate::{subscription::Flashblock, traits::PendingBlocksAPI};
+use crate::{Flashblock, PendingBlocksAPI};
 
-pub(crate) struct PendingBlocksBuilder {
+/// Builder for [`PendingBlocks`].
+#[derive(Debug)]
+pub struct PendingBlocksBuilder {
     flashblocks: Vec<Flashblock>,
     headers: Vec<Sealed<Header>>,
 

crates/flashblocks/src/state.rs

@@ -43,11 +43,7 @@ use tokio::sync::{
 use tracing::{debug, error, info, warn};
 
 use crate::{
-    PendingBlocks,
-    metrics::Metrics,
-    pending_blocks::PendingBlocksBuilder,
-    subscription::{Flashblock, FlashblocksReceiver},
-    traits::FlashblocksAPI,
+    Flashblock, FlashblocksAPI, FlashblocksReceiver, Metrics, PendingBlocks, PendingBlocksBuilder,
 };
 
 // Buffer 4s of flashblocks for flashblock_sender

crates/flashblocks/src/subscription.rs

@@ -2,59 +2,21 @@
 
 use std::{io::Read, sync::Arc, time::Duration};
 
-use alloy_primitives::{Address, B256, U256, map::foldhash::HashMap};
-use alloy_rpc_types_engine::PayloadId;
 use futures_util::{SinkExt as _, StreamExt};
-use reth_optimism_primitives::OpReceipt;
-use rollup_boost::{
-    ExecutionPayloadBaseV1, ExecutionPayloadFlashblockDeltaV1, FlashblocksPayloadV1,
-};
-use serde::{Deserialize, Serialize};
+use rollup_boost::FlashblocksPayloadV1;
 use tokio::{sync::mpsc, time::interval};
 use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};
 use tracing::{error, info, trace, warn};
 use url::Url;
 
-use crate::Metrics;
-
-/// Interval of liveness check of upstream, in milliseconds.
-pub const PING_INTERVAL_MS: u64 = 500;
-
-/// Max duration of backoff before reconnecting to upstream.
-pub const MAX_BACKOFF: Duration = Duration::from_secs(10);
+use crate::{Flashblock, Metadata, Metrics};
 
 /// Trait for receiving flashblock updates.
 pub trait FlashblocksReceiver {
     /// Called when a new flashblock is received.
     fn on_flashblock_received(&self, flashblock: Flashblock);
 }
 
-/// Metadata associated with a flashblock.
-#[derive(Debug, Deserialize, Serialize, Clone, Default)]
-pub struct Metadata {
-    /// Transaction receipts indexed by hash.
-    pub receipts: HashMap<B256, OpReceipt>,
-    /// Updated account balances.
-    pub new_account_balances: HashMap<Address, U256>,
-    /// Block number this flashblock belongs to.
-    pub block_number: u64,
-}
-
-/// A flashblock containing partial block data.
-#[derive(Debug, Clone)]
-pub struct Flashblock {
-    /// Unique payload identifier.
-    pub payload_id: PayloadId,
-    /// Index of this flashblock within the block.
-    pub index: u64,
-    /// Base payload data (only present on first flashblock).
-    pub base: Option<ExecutionPayloadBaseV1>,
-    /// Delta containing transactions and state changes.
-    pub diff: ExecutionPayloadFlashblockDeltaV1,
-    /// Associated metadata.
-    pub metadata: Metadata,
-}
-
 // Simplify actor messages to just handle shutdown
 #[derive(Debug)]
 enum ActorMessage {
@@ -73,6 +35,12 @@ impl<Receiver> FlashblocksSubscriber<Receiver>
 where
     Receiver: FlashblocksReceiver + Send + Sync + 'static,
 {
+    /// Interval of liveness check of upstream, in milliseconds.
+    pub const PING_INTERVAL_MS: u64 = 500;
+
+    /// Max duration of backoff before reconnecting to upstream.
+    pub const MAX_BACKOFF: Duration = Duration::from_secs(10);
+
     /// Creates a new flashblocks subscriber.
     pub fn new(flashblocks_state: Arc<Receiver>, ws_url: Url) -> Self {
         Self { ws_url, flashblocks_state, metrics: Metrics::default() }
@@ -98,7 +66,8 @@ where
                     Ok((ws_stream, _)) => {
                         info!(message = "WebSocket connection established");
 
-                        let mut ping_interval = interval(Duration::from_millis(PING_INTERVAL_MS));
+                        let mut ping_interval =
+                            interval(Duration::from_millis(Self::PING_INTERVAL_MS));
                         let mut awaiting_pong_resp = false;
 
                         let (mut write, mut read) = ws_stream.split();
@@ -152,11 +121,11 @@ where
                                           warn!(
                                             target: "flashblocks_rpc::subscription",
                                             ?backoff,
-                                            timeout_ms = PING_INTERVAL_MS,
+                                            timeout_ms = Self::PING_INTERVAL_MS,
                                             "No pong response from upstream, reconnecting",
                                         );
 
-                                        backoff = sleep(&metrics, backoff).await;
+                                        backoff = Self::sleep(&metrics, backoff).await;
                                         break 'conn;
                                     }
 
@@ -172,7 +141,7 @@ where
                                             "WebSocket connection lost, reconnecting",
                                         );
 
-                                        backoff = sleep(&metrics, backoff).await;
+                                        backoff = Self::sleep(&metrics, backoff).await;
                                         break 'conn;
                                     }
                                     awaiting_pong_resp = true
@@ -187,7 +156,7 @@ where
                             error = %e
                         );
 
-                        backoff = sleep(&metrics, backoff).await;
+                        backoff = Self::sleep(&metrics, backoff).await;
                         continue;
                     }
                 }
@@ -205,13 +174,13 @@ where
             }
         });
     }
-}
 
-/// Sleeps for given backoff duration. Returns incremented backoff duration, capped at [`MAX_BACKOFF`].
-async fn sleep(metrics: &Metrics, backoff: Duration) -> Duration {
-    metrics.reconnect_attempts.increment(1);
-    tokio::time::sleep(backoff).await;
-    std::cmp::min(backoff * 2, MAX_BACKOFF)
+    /// Sleeps for given backoff duration. Returns incremented backoff duration, capped at [`MAX_BACKOFF`].
+    async fn sleep(metrics: &Metrics, backoff: Duration) -> Duration {
+        metrics.reconnect_attempts.increment(1);
+        tokio::time::sleep(backoff).await;
+        std::cmp::min(backoff * 2, Self::MAX_BACKOFF)
+    }
 }
 
 fn try_decode_message(bytes: &[u8]) -> eyre::Result<Flashblock> {

crates/flashblocks/tests/state.rs

@@ -9,9 +9,7 @@ use alloy_eips::{BlockHashOrNumber, Encodable2718};
 use alloy_primitives::{Address, B256, BlockNumber, Bytes, U256, hex, map::foldhash::HashMap};
 use alloy_rpc_types_engine::PayloadId;
 use base_reth_flashblocks::{
-    state::FlashblocksState,
-    subscription::{Flashblock, Metadata},
-    traits::{FlashblocksAPI, PendingBlocksAPI},
+    Flashblock, FlashblocksAPI, FlashblocksState, Metadata, PendingBlocksAPI,
 };
 use base_reth_test_utils::{
     accounts::TestAccounts, flashblocks_harness::FlashblocksHarness, node::LocalNodeProvider,

crates/rpc/src/base/pubsub.rs

@@ -2,7 +2,7 @@
 
 use std::sync::Arc;
 
-use base_reth_flashblocks::traits::FlashblocksAPI;
+use base_reth_flashblocks::FlashblocksAPI;
 use jsonrpsee::{
     PendingSubscriptionSink, SubscriptionSink,
     core::{SubscriptionResult, async_trait},

crates/rpc/src/eth/rpc.rs

@@ -13,7 +13,7 @@ use alloy_rpc_types::{
     state::{EvmOverrides, StateOverride, StateOverridesBuilder},
 };
 use alloy_rpc_types_eth::{Filter, Log};
-use base_reth_flashblocks::traits::{FlashblocksAPI, PendingBlocksAPI};
+use base_reth_flashblocks::{FlashblocksAPI, PendingBlocksAPI};
 use jsonrpsee::{
     core::{RpcResult, async_trait},
     proc_macros::rpc,

crates/rpc/tests/flashblocks_rpc.rs

@@ -14,7 +14,7 @@ use alloy_rpc_client::RpcClient;
 use alloy_rpc_types::simulate::{SimBlock, SimulatePayload};
 use alloy_rpc_types_engine::PayloadId;
 use alloy_rpc_types_eth::{TransactionInput, error::EthRpcErrorCode};
-use base_reth_flashblocks::subscription::{Flashblock, Metadata};
+use base_reth_flashblocks::{Flashblock, Metadata};
 use base_reth_test_utils::flashblocks_harness::FlashblocksHarness;
 use common::{BLOCK_INFO_TXN, BLOCK_INFO_TXN_HASH};
 use eyre::Result;