1use alloy_consensus::TxEnvelope;
2use alloy_eips::BlockNumberOrTag;
3use alloy_eips::eip2718::Encodable2718;
4use alloy_evm::EvmEnv;
5use alloy_network::{Ethereum, Network};
6use alloy_primitives::{Bytes, TxKind, U256};
7use alloy_provider::Provider;
8use alloy_rpc_types::{TransactionInput, TransactionRequest};
9use eyre::{Result, eyre};
10use influxdb::{Timestamp, WriteQuery};
11use std::marker::PhantomData;
12use tokio::sync::{broadcast, broadcast::error::RecvError};
13use tracing::{debug, error, info, warn};
14
15use kabu_evm_utils::{NWETH, evm_access_list};
16use kabu_types_swap::{EstimationError, Swap, SwapEncoder};
17
18use kabu_core_components::Component;
19use kabu_evm_db::{AlloyDB, DatabaseKabuExt, KabuDBError};
20use kabu_evm_utils::evm_env::tx_req_to_env;
21use kabu_types_events::{HealthEvent, MessageHealthEvent, MessageSwapCompose, SwapComposeData, SwapComposeMessage, TxComposeData, TxState};
22use reth_tasks::TaskExecutor;
23use revm::context::BlockEnv;
24use revm::context_interface::block::BlobExcessGasAndPrice;
25use revm::{Database, DatabaseCommit, DatabaseRef};
26
27async fn estimator_task<N, DB>(
28 client: Option<impl Provider<N> + 'static>,
29 swap_encoder: impl SwapEncoder,
30 estimate_request: SwapComposeData<DB>,
31 compose_channel_tx: broadcast::Sender<MessageSwapCompose<DB>>,
32 health_monitor_channel_tx: Option<broadcast::Sender<MessageHealthEvent>>,
33 influxdb_write_channel_tx: Option<broadcast::Sender<WriteQuery>>,
34) -> Result<()>
35where
36 N: Network,
37 DB: DatabaseRef<Error = KabuDBError> + Database<Error = KabuDBError> + DatabaseCommit + DatabaseKabuExt + Send + Sync + Clone + 'static,
38{
39 info!("Starting estimator_task for swap: {}", estimate_request.swap);
40
41 debug!(
42 gas_limit = estimate_request.tx_compose.gas,
43 base_fee = NWETH::to_float_gwei(estimate_request.tx_compose.next_block_base_fee as u128),
44 gas_cost = NWETH::to_float_wei(estimate_request.gas_cost()),
45 stuffing_txs_len = estimate_request.tx_compose.stuffing_txs_hashes.len(),
46 "EVM estimation",
47 );
48
49 let start_time = chrono::Utc::now();
50
51 let tx_signer = estimate_request.tx_compose.signer.clone().ok_or(eyre!("NO_SIGNER"))?;
52 info!("Got signer: {:?}", tx_signer.address());
53 let gas_price = estimate_request.tx_compose.priority_gas_fee + estimate_request.tx_compose.next_block_base_fee;
54
55 info!("Encoding swap...");
56 let (to, call_value, call_data, _) = swap_encoder.encode(
57 estimate_request.swap.clone(),
58 estimate_request.tips_pct,
59 Some(estimate_request.tx_compose.next_block_number),
60 None,
61 Some(tx_signer.address()),
62 Some(estimate_request.tx_compose.eth_balance),
63 )?;
64
65 info!("Swap encoded successfully, creating transaction request with nonce {}", estimate_request.tx_compose.nonce);
66
67 let tx_request = TransactionRequest {
68 transaction_type: Some(2),
69 chain_id: Some(1),
70 from: Some(tx_signer.address()),
71 to: Some(TxKind::Call(to)),
72 gas: Some(estimate_request.tx_compose.gas),
73 value: call_value,
74 input: TransactionInput::new(call_data.clone()),
75 nonce: Some(estimate_request.tx_compose.nonce),
76 max_priority_fee_per_gas: Some(estimate_request.tx_compose.priority_gas_fee as u128),
77 max_fee_per_gas: Some(
78 estimate_request.tx_compose.next_block_base_fee as u128 + estimate_request.tx_compose.priority_gas_fee as u128,
79 ),
80 ..TransactionRequest::default()
81 };
82
83 info!("Checking for poststate...");
84 let Some(mut db) = estimate_request.poststate else {
85 error!("StateDB is None - poststate not set!");
86 return Err(eyre!("STATE_DB_IS_NONE"));
87 };
88 info!("Got poststate DB");
89
90 if let Some(client) = client {
91 let ext_db = AlloyDB::new(client, BlockNumberOrTag::Latest.into());
92 if let Some(ext_db) = ext_db {
93 db.with_ext_db(ext_db)
94 } else {
95 error!("AlloyDB is None");
96 }
97 }
98
99 let evm_env = EvmEnv {
100 block_env: BlockEnv {
101 timestamp: U256::from(estimate_request.tx_compose.next_block_timestamp),
102 number: U256::from(estimate_request.tx_compose.next_block_number),
103 basefee: estimate_request.tx_compose.next_block_base_fee,
104 blob_excess_gas_and_price: Some(BlobExcessGasAndPrice { excess_blob_gas: 0, blob_gasprice: 0 }),
105 ..Default::default()
106 },
107 ..Default::default()
108 };
109
110 let tx_env = tx_req_to_env(tx_request);
111
112 info!("Calling evm_access_list...");
113 let (gas_used, access_list) = match evm_access_list(&db, &evm_env, tx_env) {
114 Ok((gas_used, access_list)) => {
115 let pool_id_vec = estimate_request.swap.get_pool_id_vec();
116
117 tokio::task::spawn(async move {
118 for pool_id in pool_id_vec {
119 let pool_id_string = format!("{pool_id}");
120 let write_query = WriteQuery::new(Timestamp::from(start_time), "estimation")
121 .add_field("success", 1i64)
122 .add_tag("pool", pool_id_string);
123
124 if let Some(influxdb_write_channel_tx) = &influxdb_write_channel_tx
125 && let Err(e) = influxdb_write_channel_tx.send(write_query)
126 {
127 error!("Failed to send successful estimation latency to influxdb: {:?}", e);
128 }
129 }
130 });
131
132 (gas_used, access_list)
133 }
134 Err(e) => {
135 warn!(
136 "evm_access_list error for block_number={}, block_timestamp={}, swap={}, err={e}",
137 estimate_request.tx_compose.next_block_number, estimate_request.tx_compose.next_block_timestamp, estimate_request.swap
138 );
139 debug!("evm_access_list error calldata : {} {}", to, call_data);
141
142 if let Some(health_monitor_channel_tx) = &health_monitor_channel_tx
143 && let Swap::BackrunSwapLine(swap_line) = estimate_request.swap
144 && let Err(e) =
145 health_monitor_channel_tx.send(MessageHealthEvent::new(HealthEvent::SwapLineEstimationError(EstimationError {
146 swap_path: swap_line.path,
147 msg: e.to_string(),
148 })))
149 {
150 error!("Failed to send message to health monitor channel: {:?}", e);
151 }
152
153 info!("Simulation failed, returning early without sending Ready");
154 return Ok(());
155 }
156 };
157 let swap = estimate_request.swap.clone();
158
159 if gas_used < 60_000 {
160 error!(gas_used, %swap, "Incorrect transaction estimation");
161 return Err(eyre!("TRANSACTION_ESTIMATED_INCORRECTLY"));
162 }
163
164 let gas_cost = U256::from(gas_used as u128 * gas_price as u128);
165
166 debug!(
167 "Swap encode swap={}, tips_pct={:?}, next_block_number={}, gas_cost={}, signer={}",
168 estimate_request.swap,
169 estimate_request.tips_pct,
170 estimate_request.tx_compose.next_block_number,
171 gas_cost,
172 tx_signer.address()
173 );
174
175 let (to, call_value, call_data, tips_vec) = match swap_encoder.encode(
176 estimate_request.swap.clone(),
177 estimate_request.tips_pct,
178 Some(estimate_request.tx_compose.next_block_number),
179 Some(gas_cost),
180 Some(tx_signer.address()),
181 Some(estimate_request.tx_compose.eth_balance),
182 ) {
183 Ok((to, call_value, call_data, tips_vec)) => (to, call_value, call_data, tips_vec),
184 Err(error) => {
185 error!(%error, %swap, "swap_encoder.encode");
186 return Err(error);
187 }
188 };
189
190 let tx_request = TransactionRequest {
191 transaction_type: Some(2),
192 chain_id: Some(1),
193 from: Some(tx_signer.address()),
194 to: Some(TxKind::Call(to)),
195 gas: Some((gas_used * 1500) / 1000),
196 value: call_value,
197 input: TransactionInput::new(call_data),
198 nonce: Some(estimate_request.tx_compose.nonce),
199 access_list: Some(access_list),
200 max_priority_fee_per_gas: Some(estimate_request.tx_compose.priority_gas_fee as u128),
201 max_fee_per_gas: Some(
202 estimate_request.tx_compose.priority_gas_fee as u128 + estimate_request.tx_compose.next_block_base_fee as u128,
203 ),
204 ..TransactionRequest::default()
205 };
206
207 let encoded_txes: Vec<TxEnvelope> =
208 estimate_request.tx_compose.stuffing_txs.iter().map(|item| TxEnvelope::from(item.clone())).collect();
209
210 let stuffing_txs_rlp: Vec<Bytes> = encoded_txes.into_iter().map(|x| Bytes::from(x.encoded_2718())).collect();
211
212 let mut tx_with_state: Vec<TxState> = stuffing_txs_rlp.into_iter().map(TxState::ReadyForBroadcastStuffing).collect();
213
214 tx_with_state.push(TxState::SignatureRequired(Box::new(tx_request)));
215
216 let total_tips = tips_vec.into_iter().map(|v| v.tips).sum();
217 let profit_eth = estimate_request.swap.arb_profit_eth();
218 let gas_cost_f64 = NWETH::to_float(gas_cost);
219 let tips_f64 = NWETH::to_float(total_tips);
220 let profit_eth_f64 = NWETH::to_float(profit_eth);
221 let profit_f64 = match estimate_request.swap.get_first_token() {
222 Some(token_in) => token_in.to_float(estimate_request.swap.arb_profit()),
223 None => profit_eth_f64,
224 };
225
226 let swap_for_log = estimate_request.swap.clone();
227 let sign_request = MessageSwapCompose::ready(SwapComposeData {
228 tx_compose: TxComposeData { tx_bundle: Some(tx_with_state), ..estimate_request.tx_compose },
229 poststate: Some(db),
230 tips: Some(total_tips + gas_cost),
231 ..estimate_request
232 });
233
234 info!("Sending Ready message for swap: {}", swap_for_log);
235 let result = match compose_channel_tx.send(sign_request) {
236 Err(error) => {
237 error!(%error, "compose_channel_tx.send");
238 Err(eyre!("COMPOSE_CHANNEL_SEND_ERROR"))
239 }
240 _ => Ok(()),
241 };
242
243 let sim_duration = chrono::Utc::now() - start_time;
244
245 info!(
246 cost=gas_cost_f64,
247 profit=profit_f64,
248 tips=tips_f64,
249 gas_used,
250 %swap,
251 duration=sim_duration.num_microseconds().unwrap_or_default(),
252 " +++ Simulation successful",
253 );
254
255 result
256}
257
258async fn estimator_worker<N, DB>(
259 client: Option<impl Provider<N> + Clone + 'static>,
260 encoder: impl SwapEncoder + Send + Sync + Clone + 'static,
261 mut compose_channel_rx: broadcast::Receiver<MessageSwapCompose<DB>>,
262 compose_channel_tx: broadcast::Sender<MessageSwapCompose<DB>>,
263 health_monitor_channel_tx: Option<broadcast::Sender<MessageHealthEvent>>,
264 influxdb_write_channel_tx: Option<broadcast::Sender<WriteQuery>>,
265) where
266 N: Network,
267 DB: DatabaseRef<Error = KabuDBError> + Database<Error = KabuDBError> + DatabaseCommit + DatabaseKabuExt + Send + Sync + Clone + 'static,
268{
269 loop {
270 tokio::select! {
271 msg = compose_channel_rx.recv() => {
272 let compose_request_msg : Result<MessageSwapCompose<DB>, RecvError> = msg;
273 match compose_request_msg {
274 Ok(compose_request) =>{
275 if let SwapComposeMessage::Estimate(estimate_request) = compose_request.inner {
276 info!("Estimator processing Estimate request for swap: {}", estimate_request.swap);
277 let compose_channel_tx_cloned = compose_channel_tx.clone();
278 let encoder_cloned = encoder.clone();
279 let client_cloned = client.clone();
280 let influxdb_channel_tx_cloned = influxdb_write_channel_tx.clone();
281 let health_monitor_channel_tx_cloned = health_monitor_channel_tx.clone();
282 tokio::task::spawn(
283 async move {
284 match estimator_task(
285 client_cloned,
286 encoder_cloned,
287 estimate_request.clone(),
288 compose_channel_tx_cloned,
289 health_monitor_channel_tx_cloned,
290 influxdb_channel_tx_cloned,
291 ).await {
292 Ok(_) => info!("Estimator task completed successfully"),
293 Err(e) => error!("Error in EVM estimator_task: {:?}", e),
294 }
295 }
296 );
297 }
298 }
299 Err(e)=>{error!("{e}")}
300 }
301 }
302 }
303 }
304}
305
306pub struct EvmEstimatorComponent<P, N, E, DB: Clone + Send + Sync + 'static> {
307 encoder: E,
308 client: Option<P>,
309 compose_channel_rx: Option<broadcast::Sender<MessageSwapCompose<DB>>>,
310 compose_channel_tx: Option<broadcast::Sender<MessageSwapCompose<DB>>>,
311 health_monitor_channel_tx: Option<broadcast::Sender<MessageHealthEvent>>,
312 influxdb_write_channel_tx: Option<broadcast::Sender<WriteQuery>>,
313 _n: PhantomData<N>,
314}
315
316impl<P, N, E, DB> EvmEstimatorComponent<P, N, E, DB>
317where
318 N: Network,
319 P: Provider<Ethereum>,
320 E: SwapEncoder + Send + Sync + Clone + 'static,
321 DB: DatabaseRef + DatabaseKabuExt + Send + Sync + Clone + 'static,
322{
323 pub fn new(encoder: E) -> Self {
324 Self {
325 encoder,
326 client: None,
327 compose_channel_tx: None,
328 compose_channel_rx: None,
329 health_monitor_channel_tx: None,
330 influxdb_write_channel_tx: None,
331 _n: PhantomData::<N>,
332 }
333 }
334
335 pub fn new_with_provider(encoder: E, client: Option<P>) -> Self {
336 Self {
337 encoder,
338 client,
339 compose_channel_tx: None,
340 compose_channel_rx: None,
341 health_monitor_channel_tx: None,
342 influxdb_write_channel_tx: None,
343 _n: PhantomData::<N>,
344 }
345 }
346
347 pub fn with_swap_compose_channel(self, channel: broadcast::Sender<MessageSwapCompose<DB>>) -> Self {
348 Self { compose_channel_tx: Some(channel.clone()), compose_channel_rx: Some(channel), ..self }
349 }
350}
351
352impl<P, N, E, DB> Component for EvmEstimatorComponent<P, N, E, DB>
353where
354 N: Network,
355 P: Provider<N> + Send + Sync + Clone + 'static,
356 E: SwapEncoder + Clone + Send + Sync + 'static,
357 DB: DatabaseRef<Error = KabuDBError> + Database<Error = KabuDBError> + DatabaseCommit + DatabaseKabuExt + Send + Sync + Clone,
358{
359 fn spawn(self, executor: TaskExecutor) -> Result<()> {
360 let name = self.name();
361
362 let compose_channel_rx = self.compose_channel_rx.ok_or_else(|| eyre!("compose_channel_rx not set"))?.subscribe();
363 let compose_channel_tx = self.compose_channel_tx.ok_or_else(|| eyre!("compose_channel_tx not set"))?;
364
365 executor.spawn_critical(
366 name,
367 estimator_worker(
368 self.client.clone(),
369 self.encoder.clone(),
370 compose_channel_rx,
371 compose_channel_tx,
372 self.health_monitor_channel_tx.clone(),
373 self.influxdb_write_channel_tx.clone(),
374 ),
375 );
376
377 Ok(())
378 }
379 fn name(&self) -> &'static str {
380 "EvmEstimatorComponent"
381 }
382}