Relay submit improvements (#223)

## 📝 Summary

Removed the sleep at the beginning of the relay submit loop since this
is the responsibility of the bidding module.
Replaced a 5ms polling for an await to improve bidding latency.
 
## 💡 Motivation and Context

New root hash speed makes this 5ms latency important.

---

## ✅ I have completed the following steps:

* [X] Run `make lint`
* [X] Run `make test`
* [ ] Added tests (if applicable)

Author: ZanCorDX

config-live-example.toml

@@ -32,8 +32,6 @@ max_concurrent_seals = 4
 # genesis_fork_version = "0x00112233"
 
 sbundle_mergeabe_signers = []
-# slot_delta_to_start_submits_ms is usually negative since we start bidding BEFORE the slot start
-# slot_delta_to_start_submits_ms = -5000
 live_builders = ["mp-ordering", "mgp-ordering", "merging"]
 
 [[relays]]

config-playground.toml

@@ -25,8 +25,6 @@ dry_run_validation_url = "http://localhost:8545"
 ignore_cancellable_orders = true
 
 sbundle_mergeabe_signers = []
-# slot_delta_to_start_submits_ms is usually negative since we start bidding BEFORE the slot start
-# slot_delta_to_start_submits_ms = -5000
 live_builders = ["mp-ordering"]
 
 [[relays]]

crates/rbuilder/src/live_builder/block_output/relay_submit.rs

@@ -19,11 +19,9 @@ use alloy_primitives::{utils::format_ether, U256};
 use mockall::automock;
 use reth_chainspec::ChainSpec;
 use reth_primitives::SealedBlock;
-use std::{
-    sync::{Arc, Mutex},
-    time::Duration,
-};
-use tokio::time::{sleep, Instant};
+use std::sync::{Arc, Mutex};
+use tokio::sync::Notify;
+use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, event, info_span, trace, warn, Instrument, Level};
 
@@ -36,40 +34,45 @@ const SIM_ERROR_CATEGORY: &str = "submit_block_simulation";
 const VALIDATION_ERROR_CATEGORY: &str = "validate_block_simulation";
 
 /// Contains the best block so far.
-/// Building updates via compare_and_update while relay submitter polls via take_best_block
-#[derive(Debug, Clone)]
+/// Building updates via compare_and_update while relay submitter polls via take_best_block.
+/// A new block can be waited without polling via wait_for_change.
+#[derive(Debug, Default)]
 pub struct BestBlockCell {
-    val: Arc<Mutex<Option<Block>>>,
-}
-
-impl Default for BestBlockCell {
-    fn default() -> Self {
-        Self {
-            val: Arc::new(Mutex::new(None)),
-        }
-    }
-}
-
-impl BlockBuildingSink for BestBlockCell {
-    fn new_block(&self, block: Block) {
-        self.compare_and_update(block);
-    }
+    block: Mutex<Option<Block>>,
+    block_notify: Notify,
 }
 
 impl BestBlockCell {
     pub fn compare_and_update(&self, block: Block) {
-        let mut best_block = self.val.lock().unwrap();
+        let mut best_block = self.block.lock().unwrap();
         let old_value = best_block
             .as_ref()
             .map(|b| b.trace.bid_value)
             .unwrap_or_default();
         if block.trace.bid_value > old_value {
             *best_block = Some(block);
+            self.block_notify.notify_one();
         }
     }
 
     pub fn take_best_block(&self) -> Option<Block> {
-        self.val.lock().unwrap().take()
+        self.block.lock().unwrap().take()
+    }
+
+    pub async fn wait_for_change(&self) {
+        self.block_notify.notified().await
+    }
+}
+
+/// Adapts BestBlockCell to BlockBuildingSink by calling compare_and_update on new_block.
+#[derive(Debug)]
+struct BestBlockCellToBlockBuildingSink {
+    best_block_cell: Arc<BestBlockCell>,
+}
+
+impl BlockBuildingSink for BestBlockCellToBlockBuildingSink {
+    fn new_block(&self, block: Block) {
+        self.best_block_cell.compare_and_update(block);
     }
 }
 
@@ -105,13 +108,8 @@ pub struct SubmissionConfig {
     pub optimistic_prevalidate_optimistic_blocks: bool,
 
     pub bid_observer: Box<dyn BidObserver + Send + Sync>,
-    /// Delta relative to slot_time at which we start to submit blocks. Usually negative since we need to start submitting BEFORE the slot time.
-    pub slot_delta_to_start_submits: time::Duration,
 }
 
-/// run_submit_to_relays_job waits at least MIN_TIME_BETWEEN_BLOCK_CHECK between new block polls to avoid 100% CPU
-const MIN_TIME_BETWEEN_BLOCK_CHECK: Duration = Duration::from_millis(5);
-
 /// Values from [`BuiltBlockTrace`]
 struct BuiltBlockInfo {
     pub bid_value: U256,
@@ -130,7 +128,7 @@ struct BuiltBlockInfo {
 ///    returns the best bid made
 #[allow(clippy::too_many_arguments)]
 async fn run_submit_to_relays_job(
-    best_bid: BestBlockCell,
+    best_bid: Arc<BestBlockCell>,
     slot_data: MevBoostSlotData,
     relays: Vec<MevBoostRelay>,
     config: Arc<SubmissionConfig>,
@@ -143,14 +141,6 @@ async fn run_submit_to_relays_job(
         slot_data.slot(),
     );
     let mut res = None;
-    // first, sleep to slot time - slot_delta_to_start_submits
-    {
-        let submit_start_time = slot_data.timestamp() + config.slot_delta_to_start_submits;
-        let sleep_duration = submit_start_time - time::OffsetDateTime::now_utc();
-        if sleep_duration.is_positive() {
-            sleep(sleep_duration.try_into().unwrap()).await;
-        }
-    }
 
     let (normal_relays, optimistic_relays) = {
         let mut normal_relays = Vec::new();
@@ -166,18 +156,12 @@ async fn run_submit_to_relays_job(
     };
 
     let mut last_bid_value = U256::from(0);
-    let mut last_submit_time = Instant::now();
     'submit: loop {
         if cancel.is_cancelled() {
             break 'submit res;
         }
 
-        let time_since_submit = last_submit_time.elapsed();
-        if time_since_submit < MIN_TIME_BETWEEN_BLOCK_CHECK {
-            sleep(MIN_TIME_BETWEEN_BLOCK_CHECK - time_since_submit).await;
-        }
-        last_submit_time = Instant::now();
-
+        best_bid.wait_for_change().await;
         let block = if let Some(new_block) = best_bid.take_best_block() {
             if new_block.trace.bid_value > last_bid_value {
                 last_bid_value = new_block.trace.bid_value;
@@ -348,26 +332,29 @@ async fn run_submit_to_relays_job(
 }
 
 pub async fn run_submit_to_relays_job_and_metrics(
-    best_bid: BestBlockCell,
+    best_bid: Arc<BestBlockCell>,
     slot_data: MevBoostSlotData,
     relays: Vec<MevBoostRelay>,
     config: Arc<SubmissionConfig>,
     cancel: CancellationToken,
     competition_bid_value_source: Arc<dyn BidValueSource + Send + Sync>,
 ) {
-    let best_bid = run_submit_to_relays_job(
-        best_bid.clone(),
+    let last_build_block_info = run_submit_to_relays_job(
+        best_bid,
         slot_data,
         relays,
         config,
         cancel,
         competition_bid_value_source,
     )
     .await;
-    if let Some(best_bid) = best_bid {
-        if best_bid.bid_value > best_bid.true_bid_value {
+    if let Some(last_build_block_info) = last_build_block_info {
+        if last_build_block_info.bid_value > last_build_block_info.true_bid_value {
             inc_subsidized_blocks(false);
-            add_subsidy_value(best_bid.bid_value - best_bid.true_bid_value, false);
+            add_subsidy_value(
+                last_build_block_info.bid_value - last_build_block_info.true_bid_value,
+                false,
+            );
         }
     }
 }
@@ -540,7 +527,7 @@ impl BuilderSinkFactory for RelaySubmitSinkFactory {
         competition_bid_value_source: Arc<dyn BidValueSource + Send + Sync>,
         cancel: CancellationToken,
     ) -> Box<dyn BlockBuildingSink> {
-        let best_bid = BestBlockCell::default();
+        let best_block_cell = Arc::new(BestBlockCell::default());
 
         let relays = slot_data
             .relays
@@ -553,13 +540,13 @@ impl BuilderSinkFactory for RelaySubmitSinkFactory {
             })
             .collect();
         tokio::spawn(run_submit_to_relays_job_and_metrics(
-            best_bid.clone(),
+            best_block_cell.clone(),
             slot_data,
             relays,
             self.submission_config.clone(),
             cancel,
             competition_bid_value_source,
         ));
-        Box::new(best_bid)
+        Box::new(BestBlockCellToBlockBuildingSink { best_block_cell })
     }
 }

crates/rbuilder/src/live_builder/config.rs

@@ -63,8 +63,6 @@ use std::{
 use tracing::info;
 use url::Url;
 
-/// From experience (Vitaly's) all generated blocks before slot_time-8sec end loosing (due to last moment orders?)
-const DEFAULT_SLOT_DELTA_TO_START_SUBMITS: time::Duration = time::Duration::milliseconds(-8000);
 /// We initialize the wallet with the last full day. This should be enough for any bidder.
 /// On debug I measured this to be < 300ms so it's not big deal.
 pub const WALLET_INIT_HISTORY_SIZE: Duration = Duration::from_secs(60 * 60 * 24);
@@ -120,9 +118,6 @@ pub struct L1Config {
     /// If true all optimistic submissions will be validated on nodes specified in `dry_run_validation_url`
     pub optimistic_prevalidate_optimistic_blocks: bool,
 
-    /// See [`SubmissionConfig`]
-    slot_delta_to_start_submits_ms: Option<i64>,
-
     /// How many seals we are going to be doing in parallel.
     /// Optimal value may change depending on the roothash computation caching strategies.
     pub max_concurrent_seals: u64,
@@ -146,7 +141,6 @@ impl Default for L1Config {
             optimistic_enabled: false,
             optimistic_max_bid_value_eth: "0.0".to_string(),
             optimistic_prevalidate_optimistic_blocks: false,
-            slot_delta_to_start_submits_ms: None,
             cl_node_url: vec![EnvOrValue::from("http://127.0.0.1:3500")],
             max_concurrent_seals: DEFAULT_MAX_CONCURRENT_SEALS,
             genesis_fork_version: None,
@@ -204,12 +198,6 @@ impl L1Config {
         BLSBlockSigner::new(secret_key, signing_domain)
     }
 
-    pub fn slot_delta_to_start_submits(&self) -> time::Duration {
-        self.slot_delta_to_start_submits_ms
-            .map(time::Duration::milliseconds)
-            .unwrap_or(DEFAULT_SLOT_DELTA_TO_START_SUBMITS)
-    }
-
     fn submission_config(
         &self,
         chain_spec: Arc<ChainSpec>,
@@ -258,7 +246,6 @@ impl L1Config {
             optimistic_signer,
             optimistic_max_bid_value: parse_ether(&self.optimistic_max_bid_value_eth)?,
             optimistic_prevalidate_optimistic_blocks: self.optimistic_prevalidate_optimistic_blocks,
-            slot_delta_to_start_submits: self.slot_delta_to_start_submits(),
             bid_observer,
         })
     }

crates/rbuilder/src/primitives/serialize.rs

@@ -262,6 +262,7 @@ pub struct CancelShareBundle {
     pub key: ShareBundleReplacementKey,
 }
 /// Since we use the same API (mev_sendBundle) to get new bundles and also to cancel them we need this struct
+#[allow(clippy::large_enum_variant)]
 pub enum RawShareBundleDecodeResult {
     NewShareBundle(ShareBundle),
     CancelShareBundle(CancelShareBundle),