New improved bidding core. (#182)

## 📝 Summary

This new bidding core is designed to encapsulate the essence of the
bidding process. The `SlotBidder` now focuses solely on mathematical
calculations and bidding activities, operating without any inherent
subsystems since all necessary information is provided externally. The
`SlotBidder` simply listens for new blocks and incoming bids from
competitors.

```rust
pub trait SlotBidder: UnfinishedBlockBuildingSink + BidValueObs {}
```

The `BiddingService` has been enhanced to gather information on landed
blocks, enabling it to fine-tune bidding strategies and subsidies.
Additionally, it can be instructed to pursue a block with heightened
intensity:

```rust
pub trait BiddingService: std::fmt::Debug + Send + Sync {
    fn create_slot_bidder(
        &mut self,
        block: u64,
        slot: u64,
        slot_end_timestamp: u64,
        bid_maker: Box<dyn BidMaker + Send + Sync>,
    ) -> Arc<dyn SlotBidder>;

    /// When invoked, this method instructs all current or future `SlotBidder` instances working on the specified block to bid more aggressively in order to secure it.
    fn must_win_block(&self, block: u64);

    /// This method notifies the service of blocks that we have successfully landed.
    fn update_new_landed_blocks_detected(
        &self,
        landed_block_interval_info: LandedBlockIntervalInfo,
    );

    /// We inform the `BiddingService` of any issues encountered while reading landed blocks, which may prompt a strategy adjustment (e.g., pausing bidding until the next update).
    fn update_failed_reading_new_landed_blocks(&self);
}
```
The `bid_maker` is provided to the created `SlotBidder`, enabling it to
bid autonomously whenever it sees fit.
## 💡 Motivation and Context

This new core is the first step to allow having secret sandboxed code in
a TEE environment.

## ✅ I have completed the following steps:

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

crates/rbuilder/src/building/builders/block_building_helper.rs

@@ -34,7 +34,7 @@ use super::Block;
 /// 2 - Call lots of commit_order.
 /// 3 - Call set_trace_fill_time when you are done calling commit_order (we still have to review this step).
 /// 4 - Call finalize_block.
-pub trait BlockBuildingHelper: Send {
+pub trait BlockBuildingHelper: Send + Sync {
     fn box_clone(&self) -> Box<dyn BlockBuildingHelper>;
 
     /// Tries to add an order to the end of the block.

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

@@ -3,7 +3,7 @@ use reth_primitives::SealedBlock;
 
 use crate::{building::BuiltBlockTrace, mev_boost::SubmitBlockRequest};
 
-/// Trait that receives every bid made to the relays.
+/// Trait that receives every bid made by us to the relays.
 pub trait BidObserver: std::fmt::Debug {
     /// This callback is executed after the bid was made so it gives away ownership of the data.
     /// This should NOT block since it's executed in the submitting thread.

crates/rbuilder/src/live_builder/block_output/bid_value_source/best_bid_sync_source.rs

@@ -0,0 +1,49 @@
+use super::interfaces::{BidValueObs, BidValueSource};
+use alloy_primitives::U256;
+use std::sync::{Arc, Mutex};
+
+/// Simple struct tracking the last best bid and asking it in a sync way via best_bid_value.
+pub struct BestBidSyncSource {
+    best_bid_source_inner: Arc<BestBidSyncSourceInner>,
+    bid_value_source: Arc<dyn BidValueSource + Send + Sync>,
+}
+
+impl Drop for BestBidSyncSource {
+    fn drop(&mut self) {
+        self.bid_value_source
+            .unsubscribe(self.best_bid_source_inner.clone());
+    }
+}
+
+impl BestBidSyncSource {
+    pub fn new(
+        bid_value_source: Arc<dyn BidValueSource + Send + Sync>,
+        block_number: u64,
+        slot_number: u64,
+    ) -> Self {
+        let best_bid_source_inner = Arc::new(BestBidSyncSourceInner::default());
+        bid_value_source.subscribe(block_number, slot_number, best_bid_source_inner.clone());
+        Self {
+            best_bid_source_inner,
+            bid_value_source,
+        }
+    }
+
+    pub fn best_bid_value(&self) -> Option<U256> {
+        *self.best_bid_source_inner.best_bid.lock().unwrap()
+    }
+}
+
+#[derive(Debug, Default)]
+struct BestBidSyncSourceInner {
+    best_bid: Mutex<Option<U256>>,
+}
+
+impl BidValueObs for BestBidSyncSourceInner {
+    fn update_new_bid(&self, bid: U256) {
+        let mut best_bid = self.best_bid.lock().unwrap();
+        if best_bid.map_or(true, |old_bid| old_bid < bid) {
+            *best_bid = Some(bid);
+        }
+    }
+}

crates/rbuilder/src/live_builder/block_output/bid_value_source/interfaces.rs

@@ -0,0 +1,15 @@
+use alloy_primitives::U256;
+use std::sync::Arc;
+
+/// Sync + Send to allow to be called from another thread.
+pub trait BidValueObs: std::fmt::Debug + Sync + Send {
+    /// @Pending: add source of the bid.
+    fn update_new_bid(&self, bid: U256);
+}
+
+/// Object watching a stream af the bids made.
+/// Allows us to subscribe to notifications for particular blocks/slots.
+pub trait BidValueSource: std::fmt::Debug {
+    fn subscribe(&self, block_number: u64, slot_number: u64, obs: Arc<dyn BidValueObs>);
+    fn unsubscribe(&self, obs: Arc<dyn BidValueObs>);
+}

crates/rbuilder/src/live_builder/block_output/bid_value_source/mod.rs

@@ -0,0 +1,4 @@
+//! This module handles all objects needed to get feedback from the bids made by the competition.
+pub mod best_bid_sync_source;
+pub mod interfaces;
+pub mod null_bid_value_source;

crates/rbuilder/src/live_builder/block_output/bid_value_source/null_bid_value_source.rs

@@ -0,0 +1,12 @@
+use std::sync::Arc;
+
+use super::interfaces::{BidValueObs, BidValueSource};
+
+/// BidValueSource that will NOT report anything.
+#[derive(Debug)]
+pub struct NullBidValueSource {}
+
+impl BidValueSource for NullBidValueSource {
+    fn subscribe(&self, _block_number: u64, _slot_number: u64, _obs: Arc<dyn BidValueObs>) {}
+    fn unsubscribe(&self, _obs: Arc<dyn BidValueObs>) {}
+}

crates/rbuilder/src/live_builder/block_output/bidding/interfaces.rs

@@ -0,0 +1,101 @@
+use std::sync::Arc;
+
+use crate::{
+    building::builders::{block_building_helper::BlockBuildingHelper, UnfinishedBlockBuildingSink},
+    live_builder::block_output::bid_value_source::interfaces::BidValueObs,
+};
+use alloy_primitives::U256;
+use reth_primitives::BlockNumber;
+use time::OffsetDateTime;
+use tokio_util::sync::CancellationToken;
+
+/// Trait in charge of bidding blocks.
+/// It is created for each block / slot.
+/// Via UnfinishedBlockBuildingSink it gets the new biddable blocks.
+/// Via BidValueObs it gets the competition bids that it should improve when possible.
+/// On creation the concrete SlotBidder will get a BidMaker to make the bids.
+pub trait SlotBidder: UnfinishedBlockBuildingSink + BidValueObs {}
+
+/// Bid we want to make.
+pub struct Bid {
+    /// Block we should seal with payout tx of payout_tx_value.
+    block: Box<dyn BlockBuildingHelper>,
+    /// payout_tx_value should be Some <=> block.can_add_payout_tx()
+    payout_tx_value: Option<U256>,
+}
+
+impl std::fmt::Debug for Bid {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Bid")
+            .field("payout_tx_value", &self.payout_tx_value)
+            .finish_non_exhaustive()
+    }
+}
+
+impl Bid {
+    /// Creates a new Bid instance.
+    pub fn new(block: Box<dyn BlockBuildingHelper>, payout_tx_value: Option<U256>) -> Self {
+        Self {
+            block,
+            payout_tx_value,
+        }
+    }
+
+    pub fn block(self) -> Box<dyn BlockBuildingHelper> {
+        self.block
+    }
+
+    /// Returns the payout transaction value.
+    pub fn payout_tx_value(&self) -> Option<U256> {
+        self.payout_tx_value
+    }
+}
+
+/// Makes the actual bid (send it to the relay)
+pub trait BidMaker: std::fmt::Debug {
+    fn send_bid(&self, bid: Bid);
+}
+
+/// Info about a onchain block from reth.
+pub struct LandedBlockInfo {
+    pub block_number: BlockNumber,
+    pub block_timestamp: OffsetDateTime,
+    pub builder_balance: U256,
+    /// true -> we landed this block.
+    /// If false we could have landed it in coinbase == fee recipient mode but balance wouldn't change so we don't care.
+    pub beneficiary_is_builder: bool,
+}
+
+/// Trait in charge of bidding.
+/// We use one for the whole execution and ask for a [SlotBidder] for each particular slot.
+/// After BiddingService creation the builder will try to feed it all the needed update_new_landed_block_detected from the DB history.
+/// To avoid exposing how much info the BiddingService uses we don't ask it anything and feed it the max history we are willing to read.
+/// After that the builder will update each block via update_new_landed_block_detected.
+pub trait BiddingService: std::fmt::Debug + Send + Sync {
+    fn create_slot_bidder(
+        &mut self,
+        block: u64,
+        slot: u64,
+        slot_timestamp: OffsetDateTime,
+        bid_maker: Box<dyn BidMaker + Send + Sync>,
+        cancel: CancellationToken,
+    ) -> Arc<dyn SlotBidder>;
+
+    /// Access to BiddingServiceWinControl::must_win_block.
+    fn win_control(&self) -> Arc<dyn BiddingServiceWinControl>;
+
+    /// We are notified about some landed blocks.
+    /// They are sorted in ascending order.
+    /// Consecutive calls will have consecutive block numbers.
+    fn update_new_landed_blocks_detected(&mut self, landed_blocks: &[LandedBlockInfo]);
+
+    /// We let the BiddingService know we had some problem reading landed blocks just in case we wants to change his strategy (eg: stop bidding until next update_new_landed_blocks_detected)
+    fn update_failed_reading_new_landed_blocks(&mut self);
+}
+
+/// Trait to control the must_win_block feature of the BiddingService.
+/// It allows to use BiddingService as a Box (single threaded mutable access) but be able to call must_win_block from another thread.
+pub trait BiddingServiceWinControl: Send + Sync + std::fmt::Debug {
+    /// If called, any current or future SlotBidder working on that block will bid more aggressively to win the block.
+    fn must_win_block(&self, block: u64);
+}

crates/rbuilder/src/live_builder/block_output/bidding/mod.rs

@@ -1,57 +1,4 @@
-use std::sync::Arc;
-
-use alloy_primitives::U256;
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum SealInstruction {
-    /// Don't waste cycles sealing block that has no chances
-    Skip,
-    /// Set this value in the last tx before sealing block
-    Value(U256),
-}
-
-/// Slot bidder is used by builder to decide what value should be put into the last tx.
-/// It is created for each block / slot.
-pub trait SlotBidder: Send + Sync + std::fmt::Debug {
-    /// Returns true if payment for the slot can go directly to fee recipient through coinbase.
-    fn is_pay_to_coinbase_allowed(&self) -> bool;
-
-    /// Returns what value needs to be sent to the fee recipient or if block should be skipped.
-    fn seal_instruction(
-        &self,
-        unsealed_block_profit: U256,
-        slot_timestamp: time::OffsetDateTime,
-    ) -> SealInstruction;
-}
-
-impl SlotBidder for () {
-    fn is_pay_to_coinbase_allowed(&self) -> bool {
-        true
-    }
-
-    fn seal_instruction(
-        &self,
-        unsealed_block_profit: U256,
-        _slot_timestamp: time::OffsetDateTime,
-    ) -> SealInstruction {
-        SealInstruction::Value(unsealed_block_profit)
-    }
-}
-
-pub trait BiddingService: std::fmt::Debug + Send + Sync {
-    fn create_slot_bidder(
-        &mut self,
-        block: u64,
-        slot: u64,
-        slot_end_timestamp: u64,
-    ) -> Arc<dyn SlotBidder>;
-}
-
-/// Creates () which implements the dummy SlotBidder which bids all true value
-#[derive(Debug)]
-pub struct DummyBiddingService {}
-impl BiddingService for DummyBiddingService {
-    fn create_slot_bidder(&mut self, _: u64, _: u64, _: u64) -> Arc<dyn SlotBidder> {
-        Arc::new(())
-    }
-}
+pub mod interfaces;
+pub mod sequential_sealer_bid_maker;
+pub mod true_block_value_bidder;
+pub mod wallet_balance_watcher;

crates/rbuilder/src/live_builder/block_output/bidding/sequential_sealer_bid_maker.rs

@@ -0,0 +1,112 @@
+use std::sync::{Arc, Mutex};
+use tokio::sync::Notify;
+use tokio_util::sync::CancellationToken;
+use tracing::error;
+
+use crate::live_builder::block_output::relay_submit::BlockBuildingSink;
+
+use super::interfaces::{Bid, BidMaker};
+
+/// BidMaker with a background task sealing only one bid at a time.
+/// If several bids arrive while sealing another one we keep only the last one since we assume new is better.
+#[derive(Debug)]
+pub struct SequentialSealerBidMaker {
+    pending_bid: Arc<PendingBid>,
+}
+
+impl BidMaker for SequentialSealerBidMaker {
+    fn send_bid(&self, bid: Bid) {
+        self.pending_bid.update(bid);
+    }
+}
+
+/// Object used to send new bids to the [SequentialSealerBidMakerProcess].
+#[derive(Debug)]
+struct PendingBid {
+    /// Next bid to send.
+    bid: Mutex<Option<Bid>>,
+    /// Signaled when we set a new bid.
+    bid_notify: Notify,
+}
+
+impl PendingBid {
+    fn new() -> Self {
+        Self {
+            bid: Default::default(),
+            bid_notify: Notify::new(),
+        }
+    }
+    pub async fn wait_for_change(&self) {
+        self.bid_notify.notified().await
+    }
+    /// Updates bid, replacing  on current (we assume they are always increasing but we don't check it).
+    fn update(&self, bid: Bid) {
+        let mut current_bid = self.bid.lock().unwrap();
+        *current_bid = Some(bid);
+        self.bid_notify.notify_one();
+    }
+
+    fn consume_bid(&self) -> Option<Bid> {
+        let mut current_bid = self.bid.lock().unwrap();
+        current_bid.take()
+    }
+}
+
+impl SequentialSealerBidMaker {
+    pub fn new(sink: Arc<dyn BlockBuildingSink>, cancel: CancellationToken) -> Self {
+        let pending_bid = Arc::new(PendingBid::new());
+        let mut sealing_process = SequentialSealerBidMakerProcess {
+            sink,
+            cancel,
+            pending_bid: pending_bid.clone(),
+        };
+
+        tokio::task::spawn(async move {
+            sealing_process.run().await;
+        });
+        Self { pending_bid }
+    }
+}
+
+/// Background task waiting for new bids to seal.
+struct SequentialSealerBidMakerProcess {
+    /// Destination of the finished blocks.
+    sink: Arc<dyn BlockBuildingSink>,
+    cancel: CancellationToken,
+    pending_bid: Arc<PendingBid>,
+}
+
+impl SequentialSealerBidMakerProcess {
+    async fn run(&mut self) {
+        loop {
+            tokio::select! {
+                _ = self.pending_bid.wait_for_change() => self.check_for_new_bid().await,
+                _ = self.cancel.cancelled() => return
+            }
+        }
+    }
+
+    /// block.finalize_block + self.sink.new_block inside spawn_blocking.
+    async fn check_for_new_bid(&mut self) {
+        if let Some(bid) = self.pending_bid.consume_bid() {
+            let payout_tx_val = bid.payout_tx_value();
+            let block = bid.block();
+            let block_number = block.building_context().block();
+            match tokio::task::spawn_blocking(move || block.finalize_block(payout_tx_val)).await {
+                Ok(finalize_res) => match finalize_res {
+                    Ok(res) => self.sink.new_block(res.block),
+                    Err(error) => error!(
+                        block_number,
+                        ?error,
+                        "Error on finalize_block on SequentialSealerBidMaker"
+                    ),
+                },
+                Err(error) => error!(
+                    block_number,
+                    ?error,
+                    "Error on join finalize_block on on SequentialSealerBidMaker"
+                ),
+            }
+        }
+    }
+}