Better replacement latancey (#908)

ZanCorDX · claude[bot] · web-flow · commit 55bbd32a1e4c · 2026-03-27T11:59:23.000-03:00
## 📝 Summary

Increassed the ORDERS_CONSUMED_PER_BATCH from 1024 to 4096.
Implemented cancellation in orders queued for simlulation.

## ✅ I have completed the following steps:

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

---------

Co-authored-by: claude[bot] &lt;209825114+claude[bot]@users.noreply.github.com&gt;
diff --git a/crates/rbuilder/src/building/builders/mod.rs b/crates/rbuilder/src/building/builders/mod.rs
@@ -41,6 +41,11 @@ use tracing::{error, info, warn};
 
 use super::{simulated_order_command_to_sink, OrderPriority, PrioritizedOrderStore};
 
+/// Orders that blocking_consume_next_commands will consume.
+/// A slow algorithm would check approx every 200ms, to fill this batch size it would take
+/// 8192/.2 = 40960 order/sec which is even more than what we see in the whole slot for a busy block.
+const ORDERS_CONSUMED_PER_BATCH: usize = 8192;
+
 /// Block we built
 #[derive(Debug, Clone)]
 pub struct Block {
@@ -137,6 +142,7 @@ impl OrderConsumer {
     /// New commands are accumulatd in self.new_commands
     /// Call apply_new_commands to easily consume them.
     /// This method will block until the first command is received
+    /// @Pending: This method consumes a fixed (ORDERS_CONSUMED_PER_BATCH) number of pending orders. We should reconsider doing something depending on the age of the orders?
     pub fn blocking_consume_next_commands(
         &mut self,
     ) -> eyre::Result<Option<JournalSequenceNumber>> {
@@ -149,7 +155,7 @@ impl OrderConsumer {
                 warn!(msg, "Builder thread lagging on sim orders channel");
             }
         }
-        for _ in 0..1024 {
+        for _ in 0..ORDERS_CONSUMED_PER_BATCH {
             match self.orders.try_recv() {
                 Ok(order) => self.add_command(order),
                 Err(TryRecvError::Empty) => {
diff --git a/crates/rbuilder/src/building/sim.rs b/crates/rbuilder/src/building/sim.rs
@@ -22,7 +22,7 @@ use reth_provider::StateProvider;
 use std::{
     cmp::{max, min, Ordering},
     collections::hash_map::Entry,
-    sync::Arc,
+    sync::{atomic::AtomicBool, Arc},
     time::{Duration, Instant},
 };
 use tracing::{error, trace};
@@ -55,13 +55,36 @@ struct PendingOrder {
 
 pub type SimulationId = u64;
 
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Clone)]
 pub struct SimulationRequest {
     pub id: SimulationId,
     pub order: Arc<Order>,
     pub parents: Vec<Arc<Order>>,
 }
 
+/// SimulationRequest with an extra bool to be able to cancel it externally.
+#[derive(Debug, Clone)]
+pub struct CancellableSimulationRequest {
+    request: SimulationRequest,
+    // If cancelled we don't simulate.
+    cancelled: Arc<AtomicBool>,
+}
+
+impl CancellableSimulationRequest {
+    pub fn new(request: SimulationRequest, cancelled: Arc<AtomicBool>) -> Self {
+        Self { request, cancelled }
+    }
+
+    /// If cancelled returns none.
+    pub fn into_request(self) -> Option<SimulationRequest> {
+        if self.cancelled.load(std::sync::atomic::Ordering::Relaxed) {
+            None
+        } else {
+            Some(self.request)
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SimulatedResult {
     pub id: SimulationId,
diff --git a/crates/rbuilder/src/live_builder/simulation/mod.rs b/crates/rbuilder/src/live_builder/simulation/mod.rs
@@ -4,7 +4,7 @@ pub mod simulation_job_tracer;
 
 use crate::{
     building::{
-        sim::{SimTree, SimulatedResult, SimulationRequest},
+        sim::{CancellableSimulationRequest, SimTree, SimulatedResult},
         tx_sim_cache::TxExecutionCache,
         BlockBuildingContext,
     },
@@ -36,7 +36,7 @@ type BlockContextId = u64;
 pub struct SimulationContext {
     pub block_ctx: BlockBuildingContext,
     /// Simulation requests come in through this channel.
-    pub requests: flume::Receiver<SimulationRequest>,
+    pub requests: flume::Receiver<CancellableSimulationRequest>,
     /// Simulation results go out through this channel.
     pub results: mpsc::Sender<SimulatedResult>,
 }
diff --git a/crates/rbuilder/src/live_builder/simulation/sim_worker.rs b/crates/rbuilder/src/live_builder/simulation/sim_worker.rs
@@ -57,61 +57,63 @@ pub fn run_sim_worker<P>(
                     continue 'main;
                 }
             };
-        while let Ok(task) = current_sim_context.requests.recv() {
-            let sim_thread_wait_time = last_sim_finished.elapsed();
-            let sim_start = Instant::now();
+        while let Ok(cancellable_task) = current_sim_context.requests.recv() {
+            if let Some(task) = cancellable_task.into_request() {
+                let sim_thread_wait_time = last_sim_finished.elapsed();
+                let sim_start = Instant::now();
 
-            let order_id = task.order.id();
-            let start_time = Instant::now();
-            let mut block_state = BlockState::new_arc(state_provider.clone());
-            let sim_result = simulate_order(
-                task.parents.clone(),
-                task.order,
-                &current_sim_context.block_ctx,
-                &mut local_ctx,
-                &mut block_state,
-            );
-            let sim_ok = match sim_result {
-                Ok(sim_result) => {
-                    let sim_ok = match sim_result.result {
-                        OrderSimResult::Success(simulated_order, nonces_after) => {
-                            let result = SimulatedResult {
-                                id: task.id,
-                                simulated_order,
-                                previous_orders: task.parents,
-                                nonces_after: nonces_after
-                                    .into_iter()
-                                    .map(|(address, nonce)| NonceKey { address, nonce })
-                                    .collect(),
-                                simulation_time: start_time.elapsed(),
-                            };
-                            current_sim_context
-                                .results
-                                .try_send(result)
-                                .unwrap_or_default();
-                            true
-                        }
-                        OrderSimResult::Failed(_) => false,
-                    };
-                    telemetry::inc_simulated_orders(sim_ok);
-                    telemetry::inc_simulation_gas_used(sim_result.gas_used);
-                    sim_ok
-                }
-                Err(err) => {
-                    error!(?err, ?order_id, "Critical error while simulating order");
-                    // @Metric
-                    break;
-                }
-            };
+                let order_id = task.order.id();
+                let start_time = Instant::now();
+                let mut block_state = BlockState::new_arc(state_provider.clone());
+                let sim_result = simulate_order(
+                    task.parents.clone(),
+                    task.order,
+                    &current_sim_context.block_ctx,
+                    &mut local_ctx,
+                    &mut block_state,
+                );
+                let sim_ok = match sim_result {
+                    Ok(sim_result) => {
+                        let sim_ok = match sim_result.result {
+                            OrderSimResult::Success(simulated_order, nonces_after) => {
+                                let result = SimulatedResult {
+                                    id: task.id,
+                                    simulated_order,
+                                    previous_orders: task.parents,
+                                    nonces_after: nonces_after
+                                        .into_iter()
+                                        .map(|(address, nonce)| NonceKey { address, nonce })
+                                        .collect(),
+                                    simulation_time: start_time.elapsed(),
+                                };
+                                current_sim_context
+                                    .results
+                                    .try_send(result)
+                                    .unwrap_or_default();
+                                true
+                            }
+                            OrderSimResult::Failed(_) => false,
+                        };
+                        telemetry::inc_simulated_orders(sim_ok);
+                        telemetry::inc_simulation_gas_used(sim_result.gas_used);
+                        sim_ok
+                    }
+                    Err(err) => {
+                        error!(?err, ?order_id, "Critical error while simulating order");
+                        // @Metric
+                        break;
+                    }
+                };
 
-            mark_order_simulation_end(order_id, sim_ok);
-            last_sim_finished = Instant::now();
-            let sim_thread_work_time = sim_start.elapsed();
-            add_sim_thread_utilisation_timings(
-                sim_thread_work_time,
-                sim_thread_wait_time,
-                worker_id,
-            );
+                mark_order_simulation_end(order_id, sim_ok);
+                last_sim_finished = Instant::now();
+                let sim_thread_work_time = sim_start.elapsed();
+                add_sim_thread_utilisation_timings(
+                    sim_thread_work_time,
+                    sim_thread_wait_time,
+                    worker_id,
+                );
+            }
         }
     }
 }
diff --git a/crates/rbuilder/src/live_builder/simulation/simulation_job.rs b/crates/rbuilder/src/live_builder/simulation/simulation_job.rs
@@ -1,13 +1,19 @@
-use std::{fmt, sync::Arc};
+use std::{
+    fmt,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+};
 
 use crate::{
-    building::sim::{SimTree, SimulatedResult, SimulationRequest},
+    building::sim::{CancellableSimulationRequest, SimTree, SimulatedResult},
     live_builder::{
         order_input::order_sink::OrderPoolCommand,
         simulation::simulation_job_tracer::SimulationJobTracer,
     },
 };
-use ahash::HashSet;
+use ahash::{HashMap, HashSet};
 use alloy_primitives::utils::format_ether;
 use rbuilder_primitives::{BundleReplacementKey, Order, OrderId};
 use tokio::sync::mpsc;
@@ -32,7 +38,7 @@ pub struct SimulationJob {
     /// Input orders to be simulated
     new_order_sub: mpsc::UnboundedReceiver<OrderPoolCommand>,
     /// Here we send requests to the simulator pool
-    sim_req_sender: flume::Sender<SimulationRequest>,
+    sim_req_sender: flume::Sender<CancellableSimulationRequest>,
     /// Here we receive the results we asked to sim_req_sender
     sim_results_receiver: mpsc::Receiver<SimulatedResult>,
     /// Output of the simulations
@@ -50,7 +56,8 @@ pub struct SimulationJob {
 
     /// Orders we got via new_order_sub and are still being processed (they could be inside the SimTree or in the sim queue)
     /// and were not cancelled.
-    in_flight_orders: HashSet<OrderId>,
+    /// When we remove the order we set the associated bool to false (which is checked before simulating)
+    in_flight_orders: HashMap<OrderId, Arc<AtomicBool>>,
 
     /// Orders for which we sent downstream SimulatedOrderCommand::Simulation but not SimulatedOrderCommand::Cancellation.
     /// We store them to avoid generating SimulatedOrderCommand::Cancellation for failed orders since they never generated
@@ -71,7 +78,7 @@ impl SimulationJob {
     pub fn new(
         block_cancellation: CancellationToken,
         new_order_sub: mpsc::UnboundedReceiver<OrderPoolCommand>,
-        sim_req_sender: flume::Sender<SimulationRequest>,
+        sim_req_sender: flume::Sender<CancellableSimulationRequest>,
         sim_results_receiver: mpsc::Receiver<SimulatedResult>,
         slot_sim_results_sender: mpsc::Sender<SimulatedOrderCommand>,
         sim_tree: SimTree,
@@ -144,40 +151,41 @@ impl SimulationJob {
         }
     }
 
-    /// Cancelled orders will return false
-    fn order_still_valid(&self, order_id: &OrderId) -> bool {
-        self.in_flight_orders.contains(order_id)
-    }
-
     /// Pops tasks from SimTree and sends them for simulation
     fn send_new_tasks_for_simulation(&mut self) {
         // submit sim tasks loop
         loop {
-            let mut new_sim_request = self.sim_tree.pop_simulation_tasks(1024);
+            let new_sim_request = self.sim_tree.pop_simulation_tasks(1024);
             if new_sim_request.is_empty() {
                 break;
             }
-            // filter out cancelled orders
-            new_sim_request.retain(|s| self.order_still_valid(&s.order.id()));
 
             for sim_request in new_sim_request {
                 let order_id = sim_request.order.id();
-                let delivered = match self.sim_req_sender.try_send(sim_request) {
-                    Ok(()) => true,
-                    Err(flume::TrySendError::Full(_)) => {
-                        warn!("Sim channel is full, dropping order");
-                        false
-                        // @Metric
-                    }
-                    Err(flume::TrySendError::Disconnected(_)) => {
-                        error!("Sim channel is closed, dropping order");
-                        false
-                        // @Metric
+                if let Some(cancel_handle) = self.in_flight_orders.get(&order_id) {
+                    let delivered =
+                        match self
+                            .sim_req_sender
+                            .try_send(CancellableSimulationRequest::new(
+                                sim_request,
+                                cancel_handle.clone(),
+                            )) {
+                            Ok(()) => true,
+                            Err(flume::TrySendError::Full(_)) => {
+                                warn!("Sim channel is full, dropping order");
+                                false
+                                // @Metric
+                            }
+                            Err(flume::TrySendError::Disconnected(_)) => {
+                                error!("Sim channel is closed, dropping order");
+                                false
+                                // @Metric
+                            }
+                        };
+                    if !delivered {
+                        // Small bug, if a cancel arrives we are going to propagate it.
+                        self.in_flight_orders.remove(&order_id);
                     }
-                };
-                if !delivered {
-                    // Small bug, if a cancel arrives we are going to propagate it.
-                    self.in_flight_orders.remove(&order_id);
                 }
             }
         }
@@ -206,6 +214,7 @@ impl SimulationJob {
             if self
                 .in_flight_orders
                 .remove(&sim_result.simulated_order.id())
+                .is_some()
             {
                 valid_simulated_orders.push(sim_result.clone());
                 // Only send if it's the first time.
@@ -260,8 +269,10 @@ impl SimulationJob {
 
     /// return if everything went OK
     async fn process_order_cancellation(&mut self, cancellation_id: &OrderId) -> bool {
-        if !self.in_flight_orders.remove(cancellation_id) {
-            // if we removed from in_flight_orders it was never sent so there is no need to cancel
+        if let Some(cancel_handle) = self.in_flight_orders.remove(cancellation_id) {
+            cancel_handle.store(true, Ordering::Relaxed);
+        } else {
+            // Order was not in in_flight_orders (already simulated/sent), so forward the cancellation downstream.
             return self.send_cancel(cancellation_id).await;
         }
         true
@@ -280,7 +291,8 @@ impl SimulationJob {
             // @Metric
             return false;
         }
-        self.in_flight_orders.insert(order_id);
+        self.in_flight_orders
+            .insert(order_id, Arc::new(AtomicBool::new(false)));
         true
     }
 
