Metrics improvements. (#398)

## 📝 Summary

* Reset histograms periodically, otherwise recent changes do not affect
them in the long running process.
* Make fill_time less biased as we now record every fill instead of the
finalized block fill (not every block is finalized)
* Add counter for received orderpool commands by type
(orderpool_commands_received)
* Add histogram of order simualtion time (order_simulation_time)
* Remove flawed e2e latency metric.
* Remove gas per second metric from simulations.
* Adds E2E global latency metrics

Global E2E latency metrics are constructed in a way to cover the whole
path of the bundle without gaps.
For this we record events such as: bundle was received, bundle was
simulated, bundle is considered for inclusion by this builder, etc. This
is similar to tracing but expose aggregated metrics for each "bundle
history".

There are multiple steps that are needed to prevent bias: only consider
bundle that arrive when we actively submit blocks, exclude bundles that
can't be landed because of nonce or other bundle with the same nonce,
etc.

One of the findings from this is that: `2*fill time + finalize time` is
a good proxy for p99 latency. (2x fill time because while builder
building a block it does not take new bundle into account) and there are
no major gaps in the builder that can't be explained.

## ✅ I have completed the following steps:

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

---------

Co-authored-by: Daniel Xifra <[email protected]>

crates/rbuilder/src/building/block_orders/prioritized_order_store.rs

@@ -7,6 +7,7 @@ use priority_queue::PriorityQueue;
 use crate::{
     building::Sorting,
     primitives::{AccountNonce, Nonce, OrderId, SimulatedOrder},
+    telemetry::mark_order_not_ready_for_immediate_inclusion,
 };
 
 use super::SimulatedOrderSink;
@@ -125,7 +126,6 @@ impl PrioritizedOrderStore {
 
         for order_id in invalidated_orders {
             // check if order can still be valid because of optional nonces
-
             self.main_queue.remove(&order_id);
             let order = self
                 .remove_poped_order(&order_id)
@@ -158,6 +158,8 @@ impl PrioritizedOrderStore {
             );
             if retain_order {
                 self.insert_order(order);
+            } else {
+                mark_order_not_ready_for_immediate_inclusion(&order_id);
             }
         }
 

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

@@ -17,7 +17,7 @@ use crate::{
     },
     primitives::SimulatedOrder,
     provider::StateProviderFactory,
-    telemetry,
+    telemetry::{self, add_block_fill_time, add_order_simulation_time},
     utils::{check_block_hash_reader_health, HistoricalBlockError},
 };
 
@@ -218,10 +218,8 @@ where
         let gas_used = finalized_block.sealed_block.gas_used;
         let blobs = finalized_block.txs_blob_sidecars.len();
 
-        telemetry::add_built_block_metrics(
-            built_block_trace.fill_time,
-            built_block_trace.finalize_time,
-            built_block_trace.root_hash_time,
+        telemetry::add_finalized_block_metrics(
+            built_block_trace,
             txs,
             blobs,
             gas_used,
@@ -234,6 +232,7 @@ where
             block = building_ctx.block_env.number.to::<u64>(),
             build_time_mus = built_block_trace.fill_time.as_micros(),
             finalize_time_mus = built_block_trace.finalize_time.as_micros(),
+            root_hash_time_mus = built_block_trace.root_hash_time.as_micros(),
             profit = format_ether(built_block_trace.bid_value),
             builder_name = builder_name,
             txs,
@@ -293,27 +292,35 @@ where
         &mut self,
         order: &SimulatedOrder,
     ) -> Result<Result<&ExecutionResult, ExecutionError>, CriticalCommitOrderError> {
+        let start = Instant::now();
         let result =
             self.partial_block
                 .commit_order(order, &self.building_ctx, &mut self.block_state);
-        match result {
+        let sim_time = start.elapsed();
+        let (result, sim_ok) = match result {
             Ok(ok_result) => match ok_result {
                 Ok(res) => {
                     self.built_block_trace.add_included_order(res);
-                    Ok(Ok(self.built_block_trace.included_orders.last().unwrap()))
+                    (
+                        Ok(Ok(self.built_block_trace.included_orders.last().unwrap())),
+                        true,
+                    )
                 }
                 Err(err) => {
                     self.built_block_trace
                         .modify_payment_when_no_signer_error(&err);
-                    Ok(Err(err))
+                    (Ok(Err(err)), false)
                 }
             },
-            Err(e) => Err(e),
-        }
+            Err(e) => (Err(e), false),
+        };
+        add_order_simulation_time(sim_time, &self.builder_name, sim_ok);
+        result
     }
 
     fn set_trace_fill_time(&mut self, time: Duration) {
         self.built_block_trace.fill_time = time;
+        add_block_fill_time(time, &self.builder_name, self.building_ctx.timestamp())
     }
 
     fn set_trace_orders_closed_at(&mut self, orders_closed_at: OffsetDateTime) {
@@ -369,7 +376,6 @@ where
         };
         self.built_block_trace.update_orders_sealed_at();
         self.built_block_trace.root_hash_time = finalized_block.root_hash_time;
-
         self.built_block_trace.finalize_time = start_time.elapsed();
 
         Self::trace_finalized_block(

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

@@ -15,6 +15,7 @@ use crate::{
     },
     primitives::{AccountNonce, OrderId},
     provider::StateProviderFactory,
+    telemetry::mark_builder_considers_order,
 };
 use ahash::{HashMap, HashSet};
 use reth::revm::cached::CachedReads;
@@ -258,6 +259,11 @@ where
                     break;
                 }
             }
+            mark_builder_considers_order(
+                sim_order.id(),
+                &block_building_helper.built_block_trace().orders_closed_at,
+                block_building_helper.builder_name(),
+            );
             let start_time = Instant::now();
             let commit_result = block_building_helper.commit_order(&sim_order)?;
             let order_commit_time = start_time.elapsed();

crates/rbuilder/src/building/builders/parallel_builder/block_building_result_assembler.rs

@@ -19,6 +19,7 @@ use crate::{
         BlockBuildingContext,
     },
     provider::StateProviderFactory,
+    telemetry::mark_builder_considers_order,
 };
 
 /// Assembles block building results from the best orderings of order groups.
@@ -223,6 +224,11 @@ where
                 let (order_idx, _) = sequence_of_orders.sequence_of_orders.remove(0);
                 let sim_order = &order_group.orders[order_idx];
 
+                mark_builder_considers_order(
+                    sim_order.id(),
+                    &block_building_helper.built_block_trace().orders_closed_at,
+                    block_building_helper.builder_name(),
+                );
                 let start_time = Instant::now();
                 let commit_result = block_building_helper.commit_order(sim_order)?;
                 let order_commit_time = start_time.elapsed();

crates/rbuilder/src/building/built_block_trace.rs

@@ -19,7 +19,9 @@ pub struct BuiltBlockTrace {
     pub true_bid_value: U256,
     /// Some bundle failed with BundleErr::NoSigner, we might want to switch to !use_suggested_fee_recipient_as_coinbase
     pub got_no_signer_error: bool,
+    /// Timestamp of the moment we stopped considering new orders for this block.
     pub orders_closed_at: OffsetDateTime,
+    /// Timestamp when this block was fully sealed and ready for submission.
     pub orders_sealed_at: OffsetDateTime,
     pub fill_time: Duration,
     pub finalize_time: Duration,

crates/rbuilder/src/building/sim.rs

@@ -6,6 +6,7 @@ use crate::{
     building::{BlockBuildingContext, BlockState, CriticalCommitOrderError},
     primitives::{Order, OrderId, SimValue, SimulatedOrder},
     provider::StateProviderFactory,
+    telemetry::{add_order_simulation_time, mark_order_pending_nonce},
     utils::{NonceCache, NonceCacheRef},
 };
 use ahash::{HashMap, HashSet};
@@ -110,11 +111,14 @@ where
 
         let order_nonce_state = self.get_order_nonce_state(&order, nonces)?;
 
+        let order_id = order.id();
+
         match order_nonce_state {
             OrderNonceState::Invalid => {
                 return Ok(());
             }
             OrderNonceState::PendingNonces(pending_nonces) => {
+                mark_order_pending_nonce(order_id);
                 let unsatisfied_nonces = pending_nonces.len();
                 for nonce in pending_nonces {
                     self.pending_nonces
@@ -429,6 +433,7 @@ pub fn simulate_order_using_fork<Tracer: SimulationTracer>(
     ctx: &BlockBuildingContext,
     fork: &mut PartialBlockFork<'_, '_, Tracer>,
 ) -> Result<OrderSimResult, CriticalCommitOrderError> {
+    let start = Instant::now();
     // simulate parents
     let mut gas_used = 0;
     let mut blob_gas_used = 0;
@@ -452,6 +457,9 @@ pub fn simulate_order_using_fork<Tracer: SimulationTracer>(
 
     // simulate
     let result = fork.commit_order(&order, ctx, gas_used, 0, blob_gas_used, true)?;
+    let sim_time = start.elapsed();
+    add_order_simulation_time(sim_time, "sim", result.is_ok()); // we count parent sim time + order sim time time here
+
     match result {
         Ok(res) => {
             let sim_value = SimValue::new(

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

@@ -13,7 +13,7 @@ use crate::{
         add_relay_submit_time, add_subsidy_value, inc_conn_relay_errors,
         inc_failed_block_simulations, inc_initiated_submissions, inc_other_relay_errors,
         inc_relay_accepted_submissions, inc_subsidized_blocks, inc_too_many_req_relay_errors,
-        measure_block_e2e_latency,
+        mark_submission_start_time,
     },
     utils::{error_storage::store_error_event, tracing::dynamic_event},
     validation_api_client::{ValidationAPIClient, ValidationError},
@@ -291,22 +291,33 @@ async fn run_submit_to_relays_job(
             (normal_signed_submission, optimistic_signed_submission)
         };
 
+        mark_submission_start_time(block.trace.orders_sealed_at);
+
         if config.dry_run {
-            validate_block(
-                &slot_data,
-                &normal_signed_submission.submission,
-                block.sealed_block.clone(),
-                &config,
-                cancel.clone(),
-                "Dry run",
-            )
-            .instrument(submission_span)
-            .await;
+            tokio::spawn({
+                let slot_data = slot_data.clone();
+                let submission = normal_signed_submission.submission.clone();
+                let sealed_block = block.sealed_block.clone();
+                let config = config.clone();
+                let cancel = cancel.clone();
+                let submission_span = submission_span.clone();
+
+                async move {
+                    validate_block(
+                        &slot_data,
+                        &submission,
+                        sealed_block,
+                        &config,
+                        cancel,
+                        "Dry run",
+                    )
+                    .instrument(submission_span)
+                    .await
+                }
+            });
             continue 'submit;
         }
 
-        measure_block_e2e_latency(&block.trace.included_orders);
-
         for relay in &normal_relays {
             let span = info_span!(parent: &submission_span, "relay_submit", relay = &relay.id(), optimistic = false);
             let relay = relay.clone();

crates/rbuilder/src/live_builder/mod.rs

@@ -21,7 +21,7 @@ use crate::{
     },
     primitives::{MempoolTx, Order, TransactionSignedEcRecoveredWithBlobs},
     provider::StateProviderFactory,
-    telemetry::inc_active_slots,
+    telemetry::{inc_active_slots, mark_building_started, reset_histogram_metrics},
     utils::{
         error_storage::spawn_error_storage_writer, provider_head_state::ProviderHeadState, Signer,
     },
@@ -204,6 +204,8 @@ where
         };
 
         while let Some(payload) = payload_events_channel.recv().await {
+            reset_histogram_metrics();
+
             let blocklist = self.blocklist_provider.get_blocklist()?;
             if blocklist.contains(&payload.fee_recipient()) {
                 warn!(
@@ -278,13 +280,13 @@ where
                 None,
                 root_hasher,
             ) {
+                mark_building_started(block_ctx.timestamp());
                 builder_pool.start_block_building(
                     payload,
                     block_ctx,
                     self.global_cancellation.clone(),
                     time_until_slot_end.try_into().unwrap_or_default(),
                 );
-
                 if let Some(watchdog_sender) = watchdog_sender.as_ref() {
                     watchdog_sender.try_send(()).unwrap_or_default();
                 };