V3 wait handle

Author: ZanCorDX

crates/rbuilder-operator/src/flashbots_config.rs

@@ -177,18 +177,22 @@ impl LiveBuilderConfig for FlashbotsConfig {
             .create_bid_observer_and_submission_policy(&cancellation_token, &abort_token)
             .await?;
 
-        let (sink_factory, slot_info_provider, adjustment_fee_payers) =
-            create_sink_factory_and_relays(
-                &self.base_config,
-                &self.l1_config,
-                bidding_service.relay_sets().to_vec(),
-                wallet_balance_watcher,
-                bid_observer,
-                submission_policy,
-                bidding_service.clone(),
-                cancellation_token.clone(),
-            )
-            .await?;
+        let (
+            sink_factory,
+            slot_info_provider,
+            adjustment_fee_payers,
+            optimistic_v3_server_join_handle,
+        ) = create_sink_factory_and_relays(
+            &self.base_config,
+            &self.l1_config,
+            bidding_service.relay_sets().to_vec(),
+            wallet_balance_watcher,
+            bid_observer,
+            submission_policy,
+            bidding_service.clone(),
+            cancellation_token.clone(),
+        )
+        .await?;
 
         let mut live_builder = create_builder_from_sink(
             &self.base_config,
@@ -205,7 +209,9 @@ impl LiveBuilderConfig for FlashbotsConfig {
         if let Some(handle) = clickhouse_shutdown_handle {
             live_builder.add_critical_task(handle);
         }
-
+        if let Some(optimistic_v3_server_join_handle) = optimistic_v3_server_join_handle {
+            live_builder.add_critical_task(optimistic_v3_server_join_handle);
+        }
         let mut module = RpcModule::new(());
         module.register_async_method("bid_subsidiseBlock", move |params, _| {
             handle_subsidise_block(bidding_service.clone(), params)

crates/rbuilder/src/live_builder/cli.rs

@@ -17,10 +17,7 @@ use crate::{
         builders::{BacktestSimulateBlockInput, Block},
         PartialBlockExecutionTracer,
     },
-    live_builder::{
-        process_killer::{ProcessKiller, MAX_WAIT_TIME},
-        watchdog::spawn_watchdog_thread,
-    },
+    live_builder::{process_killer::ProcessKiller, watchdog::spawn_watchdog_thread},
     provider::StateProviderFactory,
     telemetry,
     utils::bls::generate_random_bls_address,

crates/rbuilder/src/live_builder/config.rs

@@ -78,7 +78,7 @@ use std::{
     sync::Arc,
     time::Duration,
 };
-use tokio::sync::Mutex as TokioMutex;
+use tokio::{sync::Mutex as TokioMutex, task::JoinHandle};
 use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};
 use url::Url;
@@ -395,7 +395,11 @@ impl L1Config {
         })
     }
 
-    /// Creates the RelaySubmitSinkFactory and also returns the associated relays (MevBoostRelaySlotInfoProvider).
+    /// Creates:
+    /// - RelaySubmitSinkFactory
+    /// - The associated relays (MevBoostRelaySlotInfoProvider).
+    /// - A hashmap of adjustment fee payers.
+    /// - JoinHandle for the optimistic V3 server.
     #[allow(clippy::type_complexity)]
     pub fn create_relays_sealed_sink_factory(
         &self,
@@ -408,6 +412,7 @@ impl L1Config {
         RelaySubmitSinkFactory,
         Vec<MevBoostRelaySlotInfoProvider>,
         ahash::HashMap<MevBoostRelayID, Address>,
+        Option<JoinHandle<()>>,
     )> {
         let signing_domain = get_signing_domain(
             chain_spec.chain,
@@ -416,7 +421,7 @@ impl L1Config {
         )?;
 
         let mut optimistic_v3_config = None;
-        if self
+        let optimistic_v3_server_join_handle = if self
             .relays
             .iter()
             .any(|r| r.submit_config.as_ref().is_some_and(|c| c.optimistic_v3))
@@ -433,7 +438,7 @@ impl L1Config {
             }
 
             let optimistic_v3_cache = OptimisticV3BlockCache::default();
-            optimistic_v3::spawn_server(
+            let optimistic_v3_server_join_handle = optimistic_v3::spawn_server(
                 address,
                 signing_domain,
                 self.optimistic_v3_relay_pubkeys.clone(),
@@ -444,8 +449,11 @@ impl L1Config {
             optimistic_v3_config = Some(OptimisticV3Config {
                 builder_url: builder_url.into_bytes(),
                 cache: optimistic_v3_cache,
-            })
-        }
+            });
+            Some(optimistic_v3_server_join_handle)
+        } else {
+            None
+        };
 
         let submission_config = self.submission_config(
             chain_spec,
@@ -479,7 +487,12 @@ impl L1Config {
             })
             .collect();
 
-        Ok((sink_factory, slot_info_providers, adjustment_fee_payers))
+        Ok((
+            sink_factory,
+            slot_info_providers,
+            adjustment_fee_payers,
+            optimistic_v3_server_join_handle,
+        ))
     }
 
     pub fn registration_update_interval(&self) -> Duration {
@@ -516,20 +529,24 @@ impl LiveBuilderConfig for Config {
         let (wallet_balance_watcher, _) =
             create_wallet_balance_watcher(provider.clone(), &self.base_config).await?;
 
-        let (sink_factory, slot_info_provider, adjustment_fee_payers) =
-            create_sink_factory_and_relays(
-                &self.base_config,
-                &self.l1_config,
-                bidding_service.relay_sets(),
-                wallet_balance_watcher,
-                Box::new(NullBidObserver {}),
-                Box::new(AlwaysSubmitPolicy {}),
-                bidding_service,
-                cancellation_token.clone(),
-            )
-            .await?;
-
-        let live_builder = create_builder_from_sink(
+        let (
+            sink_factory,
+            slot_info_provider,
+            adjustment_fee_payers,
+            optimistic_v3_server_join_handle,
+        ) = create_sink_factory_and_relays(
+            &self.base_config,
+            &self.l1_config,
+            bidding_service.relay_sets(),
+            wallet_balance_watcher,
+            Box::new(NullBidObserver {}),
+            Box::new(AlwaysSubmitPolicy {}),
+            bidding_service,
+            cancellation_token.clone(),
+        )
+        .await?;
+
+        let mut live_builder = create_builder_from_sink(
             &self.base_config,
             &self.l1_config,
             provider,
@@ -544,6 +561,9 @@ impl LiveBuilderConfig for Config {
             self.live_builders()?,
             self.base_config.max_order_execution_duration_warning(),
         );
+        if let Some(optimistic_v3_server_join_handle) = optimistic_v3_server_join_handle {
+            live_builder.add_critical_task(optimistic_v3_server_join_handle);
+        }
         Ok(live_builder.with_builders(builders))
     }
 
@@ -1082,6 +1102,11 @@ where
     .await??)
 }
 
+/// Returns:
+/// - UnfinishedBuiltBlocksInputFactory
+/// - The associated relays (MevBoostRelaySlotInfoProvider).
+/// - A hashmap of adjustment fee payers.
+/// - JoinHandle for the optimistic V3 server.
 #[allow(clippy::too_many_arguments)]
 pub async fn create_sink_factory_and_relays<P>(
     base_config: &BaseConfig,
@@ -1096,18 +1121,23 @@ pub async fn create_sink_factory_and_relays<P>(
     UnfinishedBuiltBlocksInputFactory<P>,
     Vec<MevBoostRelaySlotInfoProvider>,
     ahash::HashMap<MevBoostRelayID, Address>,
+    Option<JoinHandle<()>>,
 )>
 where
     P: StateProviderFactory + Clone + 'static,
 {
-    let (sink_sealed_factory, slot_info_provider, adjustment_fee_payers) = l1_config
-        .create_relays_sealed_sink_factory(
-            base_config.chain_spec()?,
-            relay_sets.clone(),
-            bid_observer,
-            submission_policy,
-            cancellation_token.clone(),
-        )?;
+    let (
+        sink_sealed_factory,
+        slot_info_provider,
+        adjustment_fee_payers,
+        optimistic_v3_server_join_handle,
+    ) = l1_config.create_relays_sealed_sink_factory(
+        base_config.chain_spec()?,
+        relay_sets.clone(),
+        bid_observer,
+        submission_policy,
+        cancellation_token.clone(),
+    )?;
 
     if !l1_config.relay_bid_scrapers.is_empty() {
         let sender = Arc::new(BiddingService2BidSender::new(bidding_service.clone()));
@@ -1126,7 +1156,12 @@ where
         relay_sets,
     );
 
-    Ok((sink_factory, slot_info_provider, adjustment_fee_payers))
+    Ok((
+        sink_factory,
+        slot_info_provider,
+        adjustment_fee_payers,
+        optimistic_v3_server_join_handle,
+    ))
 }
 
 /// Take the end of the pipeline (sink_factory) + pre-created slot_info_provider and creates an empty builder (it still needs the with_builders to be called)

crates/rbuilder/src/mev_boost/optimistic_v3.rs

@@ -1,4 +1,5 @@
 use crate::{
+    live_builder::process_killer::OPTIMISTIC_V3_CLOSE_TIME_SECONDS,
     telemetry::{exponential_buckets_range, REGISTRY},
     utils,
 };
@@ -19,6 +20,7 @@ use std::{
     sync::Arc,
     time::{Duration, Instant, SystemTime},
 };
+use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 use tracing::*;
 use warp::{
@@ -88,7 +90,7 @@ pub fn spawn_server(
     relay_pubkeys: HashSet<BlsPublicKey>,
     blocks: OptimisticV3BlockCache,
     cancellation: CancellationToken,
-) -> eyre::Result<()> {
+) -> eyre::Result<JoinHandle<()>> {
     // Spawn relay server.
     let handler = Handler {
         domain,
@@ -111,13 +113,13 @@ pub fn spawn_server(
         let now = std::time::Instant::now();
         info!(target: "relay_server", "Received cancellation, initiating graceful shutdown");
         // Sleep for 12 seconds to avoid being demoted for the current slot.
-        tokio::time::sleep(Duration::from_secs(12)).await;
+        tokio::time::sleep(Duration::from_secs(OPTIMISTIC_V3_CLOSE_TIME_SECONDS)).await;
         info!(target: "relay_server", elapsed = ?now.elapsed(), "Graceful shutdown complete");
     });
-    tokio::spawn(server);
+    let res = tokio::spawn(server);
     info!(target: "relay_server", %address, "Relay server listening");
 
-    Ok(())
+    Ok(res)
 }
 
 #[derive(Clone, Debug)]