feat: FlashblocksPubSubManager

crates/flashblocks-rpc/src/cache.rs

@@ -17,9 +17,8 @@ use reth_primitives::Recovered;
 use reth_primitives_traits::block::body::BlockBody;
 
 use reth_rpc_eth_api::{RpcBlock, RpcReceipt};
-use rollup_boost::{
-    FlashblockBuilder, FlashblocksPayloadV1, OpExecutionPayloadEnvelope, PayloadVersion,
-};
+use rollup_boost::provider::FlashblockBuilder;
+use rollup_boost::{FlashblocksPayloadV1, OpExecutionPayloadEnvelope, PayloadVersion};
 use serde::{Deserialize, Serialize};
 use std::{collections::HashMap, str::FromStr, sync::Arc};
 

crates/flashblocks-rpc/src/tests/mod.rs

@@ -176,7 +176,8 @@ mod tests {
         let tx2 = Bytes::from_str("0xf8cd82016d8316e5708302c01c94f39635f2adf40608255779ff742afe13de31f57780b8646e530e9700000000000000000000000000000000000000000000000000000000000000010000000000000000000000000000000000000000000000001bc16d674ec8000000000000000000000000000000000000000000000000000156ddc81eed2a36d68302948ba0a608703e79b22164f74523d188a11f81c25a65dd59535bab1cd1d8b30d115f3ea07f4cfbbad77a139c9209d3bded89091867ff6b548dd714109c61d1f8e7a84d14").unwrap();
 
         // Send another test flashblock payload
-        let payload = FlashblocksPayloadV1 {
+
+        FlashblocksPayloadV1 {
             payload_id: PayloadId::new([0; 8]),
             index: 1,
             base: None,
@@ -222,9 +223,7 @@ mod tests {
                 },
             })
             .unwrap(),
-        };
-
-        payload
+        }
     }
 
     #[tokio::test]
@@ -243,7 +242,7 @@ mod tests {
         let pending_block = provider
             .get_block_by_number(alloy_eips::BlockNumberOrTag::Pending)
             .await?;
-        assert_eq!(pending_block.is_none(), true);
+        assert!(pending_block.is_none());
 
         let base_payload = create_first_payload();
         node.send_payload(base_payload).await?;
@@ -296,7 +295,7 @@ mod tests {
         let provider = node.provider().await?;
 
         let receipt = provider.get_transaction_receipt(TX1_HASH).await?;
-        assert_eq!(receipt.is_none(), true);
+        assert!(receipt.is_none());
 
         node.send_test_payloads().await?;
 

crates/rollup-boost/src/cli.rs

@@ -1,3 +1,13 @@
+use crate::{
+    BlockSelectionPolicy, FlashblocksArgs, ProxyLayer, RollupBoostServer, RpcClient,
+    client::rpc::{BuilderArgs, L2ClientArgs},
+    debug_api::ExecutionMode,
+    get_version, init_metrics,
+    payload::PayloadSource,
+    probe::ProbeLayer,
+    provider::FlashblocksProvider,
+    pubsub::FlashblocksPubSubManager,
+};
 use alloy_rpc_types_engine::JwtSecret;
 use clap::Parser;
 use eyre::bail;
@@ -8,18 +18,13 @@ use std::{
     path::PathBuf,
     str::FromStr,
     sync::Arc,
+    time::Duration,
 };
-use tokio::signal::unix::{SignalKind, signal as unix_signal};
-use tracing::{Level, info};
-
-use crate::{
-    BlockSelectionPolicy, Flashblocks, FlashblocksArgs, ProxyLayer, RollupBoostServer, RpcClient,
-    client::rpc::{BuilderArgs, L2ClientArgs},
-    debug_api::ExecutionMode,
-    get_version, init_metrics,
-    payload::PayloadSource,
-    probe::ProbeLayer,
+use tokio::{
+    net::TcpListener,
+    signal::unix::{SignalKind, signal as unix_signal},
 };
+use tracing::{Level, info};
 
 #[derive(Clone, Parser, Debug)]
 #[clap(author, version = get_version(), about)]
@@ -140,23 +145,24 @@ impl RollupBoostArgs {
         let execution_mode = Arc::new(Mutex::new(self.execution_mode));
 
         let (rpc_module, health_handle): (RpcModule<()>, _) = if self.flashblocks.flashblocks {
-            let flashblocks_args = self.flashblocks;
-            let inbound_url = flashblocks_args.flashblocks_builder_url;
-            let outbound_addr = SocketAddr::new(
-                IpAddr::from_str(&flashblocks_args.flashblocks_host)?,
-                flashblocks_args.flashblocks_port,
+            let builder_ws_url = self.flashblocks.flashblocks_builder_url;
+            let listener_addr = SocketAddr::new(
+                IpAddr::from_str(&self.flashblocks.flashblocks_host)?,
+                self.flashblocks.flashblocks_port,
             );
 
-            let builder_client = Arc::new(Flashblocks::run(
-                builder_client.clone(),
-                inbound_url,
-                outbound_addr,
-                flashblocks_args.flashblock_builder_ws_reconnect_ms,
-            )?);
+            let listener = TcpListener::bind(listener_addr).await?;
+            let flashblocks_provider = Arc::new(FlashblocksProvider::new(builder_client));
+            FlashblocksPubSubManager::spawn(
+                builder_ws_url,
+                listener,
+                flashblocks_provider.clone(),
+                Duration::from_millis(self.flashblocks.flashblock_builder_ws_reconnect_ms),
+            );
 
             let rollup_boost = RollupBoostServer::new(
                 l2_client,
-                builder_client,
+                flashblocks_provider,
                 execution_mode.clone(),
                 self.block_selection_policy,
                 probes.clone(),