proper StateProvider life Arc->Box

Author: ZanCorDX

crates/rbuilder/src/backtest/restore_landed_orders/resim_landed_block.rs

@@ -16,7 +16,6 @@ use eyre::Context;
 use rbuilder_primitives::evm_inspector::SlotKey;
 use reth_chainspec::ChainSpec;
 use reth_primitives::{Receipt, Recovered, TransactionSigned};
-use reth_provider::StateProvider;
 use std::sync::Arc;
 
 #[derive(Debug)]
@@ -69,11 +68,10 @@ where
 
     let mut local_ctx = ThreadBlockBuildingContext::default();
 
-    let state_provider: Arc<dyn StateProvider + Send + Sync> =
-        crate::provider::shared_state_provider(
-            provider.history_by_block_hash(ctx.attributes.parent)?,
-        );
-    let cached = CachedDB::new(state_provider, ctx.shared_cached_reads.clone());
+    let cached = CachedDB::new(
+        provider.history_by_block_hash(ctx.attributes.parent)?,
+        ctx.shared_cached_reads.clone(),
+    );
     let mut partial_block = PartialBlock::new(true);
     let mut state = BlockState::new(cached);
 

crates/rbuilder/src/bin/debug-bench-machine.rs

@@ -103,10 +103,9 @@ async fn main() -> eyre::Result<()> {
         mev_blocker_price,
     );
 
-    let state_provider = rbuilder::provider::shared_state_provider(
-        provider_factory
-            .provider_factory_unchecked()
-            .history_by_block_number(last_block)?,
+    let source = rbuilder::provider::StateProviderSource::new(
+        Arc::new(provider_factory.clone()),
+        parent_num_hash.hash,
     );
 
     let mut build_times_ms = Vec::new();
@@ -121,11 +120,12 @@ async fn main() -> eyre::Result<()> {
         .into_iter()
         .collect();
         let txs = txs.clone();
-        let state_provider = state_provider.clone();
+        let source = source.clone();
         let (build_time, finalize_time) =
             tokio::task::spawn_blocking(move || -> eyre::Result<_> {
                 let mut partial_block = PartialBlock::new(true);
-                let cached = CachedDB::new(state_provider, ctx.shared_cached_reads.clone());
+                let cached =
+                    CachedDB::new(source.state_provider()?, ctx.shared_cached_reads.clone());
                 let mut state = BlockState::new(cached);
                 let mut local_ctx = ThreadBlockBuildingContext::default();
 

crates/rbuilder/src/bin/run-bundle-on-prefix.rs

@@ -20,7 +20,6 @@ use rbuilder::{
         OrderPriority, ThreadBlockBuildingContext,
     },
     live_builder::{cli::LiveBuilderConfig, config::Config},
-    provider::StateProviderFactory,
     utils::{extract_onchain_block_txs, find_suggested_fee_recipient},
 };
 use rbuilder_config::load_toml_config;
@@ -156,14 +155,13 @@ impl LandedBlockInfo {
             .config
             .base_config()
             .create_reth_provider_factory(true)?;
-        let block_state = rbuilder::provider::shared_state_provider(
-            provider.history_by_block_hash(ctx.attributes.parent)?,
-        );
+        let source =
+            rbuilder::provider::StateProviderSource::new(Arc::new(provider), ctx.attributes.parent);
         let order_statistics = OrderStatistics::new();
         Ok(BlockBuildingHelperFromProvider::new(
             BuiltBlockId::ZERO,
             0,
-            block_state,
+            source,
             ctx,
             "TEST".to_string(),
             false,
@@ -225,7 +223,7 @@ async fn main() -> eyre::Result<()> {
     // Test backruns after prefix.
     println!("Backruns after prefix");
     for sim_order in sim_orders {
-        let mut builder = builder.box_clone();
+        let mut builder = builder.try_clone()?;
         let res = builder.commit_order(&mut block_info.local_ctx, &sim_order, &|sim_result| {
             simulation_too_low::<OrderMaxProfitPriority<FullProfitInfoGetter>>(
                 &sim_order.sim_value,

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

@@ -1,8 +1,6 @@
 use alloy_primitives::{utils::format_ether, Address, TxHash, I256, U256};
-use reth_provider::StateProvider;
 use std::{
     cmp::max,
-    sync::Arc,
     time::{Duration, Instant},
 };
 use time::OffsetDateTime;
@@ -20,6 +18,7 @@ use crate::{
         ThreadBlockBuildingContext,
     },
     live_builder::block_output::bidding_service_interface::CompetitionBidContext,
+    provider::StateProviderSource,
     telemetry::{self, add_block_fill_time, add_order_simulation_time},
     utils::{check_block_hash_reader_health, elapsed_ms, HistoricalBlockError},
 };
@@ -37,8 +36,9 @@ 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 + Sync {
-    fn box_clone(&self) -> Box<dyn BlockBuildingHelper>;
+pub trait BlockBuildingHelper: Send {
+    /// Clones the helper, reopening its state provider. Fallible because reopening can fail.
+    fn try_clone(&self) -> Result<Box<dyn BlockBuildingHelper>, BlockBuildingHelperError>;
 
     /// Tries to add an order to the end of the block.
     /// Block state changes only on Ok(Ok)
@@ -112,16 +112,6 @@ pub struct BiddableUnfinishedBlock {
     pub chosen_as_best_at: OffsetDateTime,
 }
 
-impl Clone for BiddableUnfinishedBlock {
-    fn clone(&self) -> Self {
-        Self {
-            block: self.block.box_clone(),
-            true_block_value: self.true_block_value,
-            chosen_as_best_at: self.chosen_as_best_at,
-        }
-    }
-}
-
 impl BiddableUnfinishedBlock {
     pub fn new(block: Box<dyn BlockBuildingHelper>) -> Result<Self, BlockBuildingHelperError> {
         let true_block_value = block.true_block_value()?;
@@ -131,6 +121,16 @@ impl BiddableUnfinishedBlock {
             chosen_as_best_at: OffsetDateTime::now_utc(),
         })
     }
+
+    /// Clones the block by reopening the underlying state provider. Fallible because reopening the
+    /// provider can fail.
+    pub fn try_clone(&self) -> Result<Self, BlockBuildingHelperError> {
+        Ok(Self {
+            block: self.block.try_clone()?,
+            true_block_value: self.true_block_value,
+            chosen_as_best_at: self.chosen_as_best_at,
+        })
+    }
     pub fn id(&self) -> BuiltBlockId {
         self.block.id()
     }
@@ -145,12 +145,13 @@ impl BiddableUnfinishedBlock {
 }
 
 /// Implementation of BlockBuildingHelper based on a generic Provider
-#[derive(Clone)]
 pub struct BlockBuildingHelperFromProvider<
     PartialBlockExecutionTracerType: PartialBlockExecutionTracer + Clone + Send + Sync + 'static,
 > {
     /// Balance of fee recipient before we stared building.
     _fee_recipient_balance_start: U256,
+    /// Reopens a fresh state provider when the helper is cloned for bidding/finalization.
+    source: StateProviderSource,
     /// Accumulated changes for the block (due to commit_order calls).
     block_state: BlockState<CachedDB>,
     partial_block: PartialBlock<GasUsedSimulationTracer, PartialBlockExecutionTracerType>,
@@ -215,7 +216,7 @@ impl BlockBuildingHelperFromProvider<NullPartialBlockExecutionTracer> {
     pub fn new(
         built_block_id: BuiltBlockId,
         next_journal_sequence_number: JournalSequenceNumber,
-        state_provider: Arc<dyn StateProvider + Send + Sync>,
+        state_provider_source: StateProviderSource,
         building_ctx: BlockBuildingContext,
         builder_name: String,
         discard_txs: bool,
@@ -226,7 +227,7 @@ impl BlockBuildingHelperFromProvider<NullPartialBlockExecutionTracer> {
         BlockBuildingHelperFromProvider::new_with_execution_tracer(
             built_block_id,
             next_journal_sequence_number,
-            state_provider,
+            state_provider_source,
             building_ctx,
             builder_name,
             discard_txs,
@@ -251,7 +252,7 @@ impl<
     pub fn new_with_execution_tracer(
         built_block_id: BuiltBlockId,
         next_journal_sequence_number: JournalSequenceNumber,
-        state_provider: Arc<dyn StateProvider + Send + Sync>,
+        state_provider_source: StateProviderSource,
         building_ctx: BlockBuildingContext,
         builder_name: String,
         discard_txs: bool,
@@ -260,6 +261,7 @@ impl<
         partial_block_execution_tracer: PartialBlockExecutionTracerType,
         max_order_execution_duration_warning: Option<Duration>,
     ) -> Result<Self, BlockBuildingHelperError> {
+        let state_provider = state_provider_source.state_provider()?;
         let last_committed_block = building_ctx.block() - 1;
         check_block_hash_reader_health(last_committed_block, &state_provider)?;
 
@@ -288,6 +290,7 @@ impl<
         built_block_trace.available_orders_statistics = available_orders_statistics;
         Ok(Self {
             _fee_recipient_balance_start: fee_recipient_balance_start,
+            source: state_provider_source,
             block_state,
             partial_block,
             payout_tx_gas,
@@ -621,8 +624,27 @@ impl<
         &self.building_ctx
     }
 
-    fn box_clone(&self) -> Box<dyn BlockBuildingHelper> {
-        Box::new(self.clone())
+    fn try_clone(&self) -> Result<Box<dyn BlockBuildingHelper>, BlockBuildingHelperError> {
+        let state_provider = self.source.state_provider()?;
+        let cached = CachedDB::new(
+            state_provider,
+            self.building_ctx.shared_cached_reads.clone(),
+        );
+        let block_state =
+            BlockState::new(cached).with_bundle_state(self.block_state.clone_bundle());
+        Ok(Box::new(Self {
+            _fee_recipient_balance_start: self._fee_recipient_balance_start,
+            source: self.source.clone(),
+            block_state,
+            partial_block: self.partial_block.clone(),
+            payout_tx_gas: self.payout_tx_gas,
+            builder_name: self.builder_name.clone(),
+            building_ctx: self.building_ctx.clone(),
+            built_block_trace: self.built_block_trace.clone(),
+            cancel_on_fatal_error: self.cancel_on_fatal_error.clone(),
+            finalize_adjustment_state: self.finalize_adjustment_state.clone(),
+            max_order_execution_duration_warning: self.max_order_execution_duration_warning,
+        }))
     }
 
     fn builder_name(&self) -> &str {

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

@@ -160,8 +160,8 @@ impl<'a> BlockBuildingHelperStatsLogger<'a> {
 
 impl BlockBuildingHelper for BlockBuildingHelperStatsLogger<'_> {
     /// logging is not cloned
-    fn box_clone(&self) -> Box<dyn BlockBuildingHelper> {
-        self.block_building_helper.box_clone()
+    fn try_clone(&self) -> Result<Box<dyn BlockBuildingHelper>, BlockBuildingHelperError> {
+        self.block_building_helper.try_clone()
     }
 
     fn commit_order(

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

@@ -59,8 +59,8 @@ impl MockBlockBuildingHelper {
 }
 
 impl BlockBuildingHelper for MockBlockBuildingHelper {
-    fn box_clone(&self) -> Box<dyn BlockBuildingHelper> {
-        Box::new(self.clone())
+    fn try_clone(&self) -> Result<Box<dyn BlockBuildingHelper>, BlockBuildingHelperError> {
+        Ok(Box::new(self.clone()))
     }
 
     fn commit_order(

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

@@ -21,7 +21,7 @@ use crate::{
         block_output::bidding_service_interface::CompetitionBidContext,
         building::built_block_cache::BuiltBlockCache,
     },
-    provider::StateProviderFactory,
+    provider::{StateProviderFactory, StateProviderSource},
     telemetry::{
         add_ordering_builder_base_stage_stats, add_ordering_builder_pre_filtered_stage_stats,
         mark_builder_considers_order, OrderInclusionRatio,
@@ -32,7 +32,6 @@ use ahash::{HashMap, HashSet};
 use alloy_primitives::I256;
 use derivative::Derivative;
 use rbuilder_primitives::{AccountNonce, OrderId, SimValue, SimulatedOrder};
-use reth_provider::StateProvider;
 use serde::Deserialize;
 use std::{
     marker::PhantomData,
@@ -95,11 +94,12 @@ pub fn run_ordering_builder<P, OrderPriorityType>(
 {
     let payload_id = input.ctx.payload_id;
 
-    let block_state: Arc<dyn StateProvider + Send + Sync> = match input
-        .provider
-        .history_by_block_hash(input.ctx.attributes.parent)
-    {
-        Ok(state) => crate::provider::shared_state_provider(state),
+    let source = StateProviderSource::new(
+        Arc::new(input.provider.clone()),
+        input.ctx.attributes.parent,
+    );
+    let state_provider = match source.state_provider() {
+        Ok(state_provider) => state_provider,
         Err(err) => {
             error!(
                 ?err,
@@ -111,13 +111,13 @@ pub fn run_ordering_builder<P, OrderPriorityType>(
         }
     };
 
-    let nonces = NonceCache::new(block_state.clone());
+    let nonces = NonceCache::new(state_provider);
 
     let mut order_intake_consumer =
         OrderIntakeConsumer::<OrderPriorityType>::new(nonces, input.input);
 
     let mut builder = OrderingBuilderContext::new(
-        block_state.clone(),
+        source,
         input.builder_name,
         input.ctx,
         config.clone(),
@@ -156,7 +156,9 @@ pub fn run_ordering_builder<P, OrderPriorityType>(
         ) {
             Ok(block) => {
                 if let Ok(block) = BiddableUnfinishedBlock::new(block) {
-                    input.sink.new_block(block);
+                    if let Err(err) = input.sink.new_block(block) {
+                        error!(?err, payload_id, "Failed to submit unfinished block");
+                    }
                 }
             }
             Err(err) => {
@@ -190,8 +192,12 @@ where
     let block_orders =
         block_orders_from_sim_orders::<OrderPriorityType>(input.sim_orders, &state_provider)?;
     let mut local_ctx = ThreadBlockBuildingContext::default();
+    let source = StateProviderSource::new(
+        Arc::new(input.provider.clone()),
+        input.ctx.attributes.parent,
+    );
     let mut builder = OrderingBuilderContext::new(
-        crate::provider::shared_state_provider(state_provider),
+        source,
         input.builder_name,
         input.ctx.clone(),
         ordering_config,
@@ -220,7 +226,7 @@ where
 #[derivative(Debug)]
 pub struct OrderingBuilderContext {
     #[derivative(Debug = "ignore")]
-    state: Arc<dyn StateProvider + Send + Sync>,
+    source: StateProviderSource,
     builder_name: String,
     ctx: BlockBuildingContext,
     config: OrderingBuilderConfig,
@@ -238,15 +244,15 @@ pub struct OrderingBuilderContext {
 
 impl OrderingBuilderContext {
     pub fn new(
-        state: Arc<dyn StateProvider + Send + Sync>,
+        source: StateProviderSource,
         builder_name: String,
         ctx: BlockBuildingContext,
         config: OrderingBuilderConfig,
         max_order_execution_duration_warning: Option<Duration>,
         built_block_cache: Arc<BuiltBlockCache>,
     ) -> Self {
         Self {
-            state,
+            source,
             builder_name,
             ctx,
             local_ctx: Default::default(),
@@ -301,7 +307,7 @@ impl OrderingBuilderContext {
         let mut block_building_helper = BlockBuildingHelperFromProvider::new_with_execution_tracer(
             built_block_id,
             next_journal_sequence_number,
-            self.state.clone(),
+            self.source.clone(),
             self.ctx.clone(),
             self.builder_name.clone(),
             self.config.discard_txs,