kabu_defi_health_monitor/
stuffing_tx_monitor.rs

1use alloy_consensus::transaction::Transaction;
2use alloy_eips::BlockNumberOrTag;
3use alloy_network::{BlockResponse, Ethereum, TransactionResponse};
4use alloy_primitives::{Address, TxHash, U256};
5use alloy_provider::Provider;
6use alloy_rpc_types::BlockTransactions;
7use eyre::{Result, eyre};
8use influxdb::{Timestamp, WriteQuery};
9use kabu_core_components::Component;
10use reth_tasks::TaskExecutor;
11use std::collections::HashMap;
12use std::sync::Arc;
13use tokio::sync::broadcast;
14use tokio::sync::broadcast::Receiver;
15use tokio::sync::broadcast::error::RecvError;
16use tracing::{error, info};
17
18use kabu_core_blockchain::Blockchain;
19use kabu_evm_utils::NWETH;
20use kabu_types_market::Token;
21use kabu_types_swap::Swap;
22
23use kabu_types_blockchain::debug_trace_transaction;
24use kabu_types_events::{MarketEvents, MessageTxCompose, TxComposeMessageType};
25
26#[derive(Clone, Debug)]
27struct TxToCheck {
28    block: u64,
29    token_in: Token,
30    profit: U256,
31    cost: U256,
32    tips: U256,
33    swap: Swap,
34}
35
36async fn calc_coinbase_diff<P: Provider<Ethereum> + 'static>(client: P, tx_hash: TxHash, coinbase: Address) -> Result<U256> {
37    let (pre, post) = debug_trace_transaction(client, tx_hash, true).await?;
38
39    let coinbase_pre = pre.get(&coinbase).ok_or(eyre!("COINBASE_NOT_FOUND_IN_PRE"))?;
40    let coinbase_post = post.get(&coinbase).ok_or(eyre!("COINBASE_NOT_FOUND_IN_POST"))?;
41
42    let balance_diff = coinbase_post.balance.unwrap_or_default().checked_sub(coinbase_pre.balance.unwrap_or_default()).unwrap_or_default();
43    info!("Stuffing tx mined MF tx: {:?} sent to coinbase: {}", tx_hash, NWETH::to_float(balance_diff));
44
45    Ok(balance_diff)
46}
47
48pub async fn stuffing_tx_monitor_worker<P: Provider<Ethereum> + Clone + 'static>(
49    client: P,
50    tx_compose_channel_rx: broadcast::Sender<MessageTxCompose>,
51    market_events_rx: broadcast::Sender<MarketEvents>,
52    influxdb_write_channel_tx: Option<broadcast::Sender<WriteQuery>>,
53) -> Result<()> {
54    let mut tx_compose_channel_rx: Receiver<MessageTxCompose> = tx_compose_channel_rx.subscribe();
55    let mut market_events_rx: Receiver<MarketEvents> = market_events_rx.subscribe();
56
57    let mut txs_to_check: HashMap<TxHash, TxToCheck> = HashMap::new();
58
59    loop {
60        tokio::select! {
61            msg = market_events_rx.recv() => {
62                let market_event_msg : Result<MarketEvents, RecvError> = msg;
63                match market_event_msg {
64                    Ok(market_event)=>{
65                        if let MarketEvents::BlockTxUpdate{ block_number,..} = market_event {
66                            // Get block from provider to get coinbase and transactions
67                            let block = match client.get_block_by_number(BlockNumberOrTag::Number(block_number)).await {
68                                Ok(Some(block)) => block,
69                                Ok(None) => {
70                                    error!("Block {} not found", block_number);
71                                    continue;
72                                }
73                                Err(e) => {
74                                    error!("Failed to get block {}: {}", block_number, e);
75                                    continue;
76                                }
77                            };
78                            let coinbase = block.header().beneficiary;
79                            // Get transactions from the block
80                            let txs = match block.transactions() {
81                                BlockTransactions::Full(txs) => txs.clone(),
82                                _ => {
83                                    // If we don't have full transactions, skip
84                                    continue;
85                                }
86                            };
87                            if !txs.is_empty() {
88                                for (idx, tx) in txs.iter().enumerate() {
89                                    let tx_hash = tx.tx_hash();
90                                    if let Some(tx_to_check) = txs_to_check.get(&tx_hash).cloned(){
91                                        info!("Stuffing tx found mined {:?} block: {} -> {} idx: {} profit: {} tips: {} token: {} to: {:?} {}", tx.tx_hash(), tx_to_check.block, block_number, idx, NWETH::to_float(tx_to_check.profit), NWETH::to_float(tx_to_check.tips), tx_to_check.token_in.get_symbol(), tx.to().unwrap_or_default(), tx_to_check.swap );
92                                        if idx < txs.len() - 1 {
93                                            let others_tx = &txs[idx+1];
94                                            let others_tx_hash = others_tx.tx_hash();
95                                            let client_clone = client.clone();
96                                            let influx_channel_clone = influxdb_write_channel_tx.clone();
97                                            info!("Stuffing tx mined {:?} MF tx: {:?} to: {:?}", tx.tx_hash(), others_tx.tx_hash(), others_tx.to().unwrap_or_default() );
98                                            tokio::task::spawn( async move {
99                                                if let Ok(coinbase_diff)  = calc_coinbase_diff(client_clone, others_tx_hash, coinbase).await {
100                                                    let start_time_utc =   chrono::Utc::now();
101                                                    let bribe = NWETH::to_float(tx_to_check.tips);
102                                                    let others_bribe = NWETH::to_float(coinbase_diff);
103                                                    let cost = NWETH::to_float(tx_to_check.cost);
104
105                                                    let write_query = WriteQuery::new(Timestamp::from(start_time_utc), "stuffing_mined")
106                                                        .add_field("our_bribe", bribe)
107                                                        .add_field("our_cost", cost)
108                                                        .add_field("others_bribe", others_bribe)
109                                                        .add_tag("tx_block", tx_to_check.block)
110                                                        .add_tag("block", block_number)
111                                                        .add_tag("block_idx", idx as u64)
112                                                        .add_tag("stuffing_tx", tx_hash.to_string())
113                                                        .add_tag("other_tx", others_tx_hash.to_string());
114                                                    if let Some(tx) = influx_channel_clone
115                                                        && let Err(e) = tx.send(write_query) {
116                                                            error!("Failed to send block latency to influxdb: {:?}", e);
117                                                    }
118                                                };
119                                            });
120                                        }
121                                        txs_to_check.remove::<TxHash>(&tx.tx_hash());
122                                    }
123                                }
124                            }
125                            info!("Stuffing txs to check : {} at block {}", txs_to_check.len(), block_number);
126
127                            let start_time_utc =   chrono::Utc::now();
128
129                            let write_query = WriteQuery::new(Timestamp::from(start_time_utc), "stuffing_waiting").add_field("value", txs_to_check.len() as u64).add_tag("block", block_number);
130                            if let Some(tx) = &influxdb_write_channel_tx
131                                && let Err(e) = tx.send(write_query) {
132                                    error!("Failed to send block latency to influxdb: {:?}", e);
133                            }
134                        }
135                    }
136                    Err(e)=>{
137                        error!("market_event_rx error : {e}")
138                    }
139                }
140            },
141
142            msg = tx_compose_channel_rx.recv() => {
143                let tx_compose_update : Result<MessageTxCompose, RecvError>  = msg;
144                match tx_compose_update {
145                    Ok(tx_compose_msg)=>{
146                        if let TxComposeMessageType::Sign(tx_compose_data) = tx_compose_msg.inner {
147                            for stuffing_tx_hash in tx_compose_data.stuffing_txs_hashes.iter() {
148                                let Some(swap) = & tx_compose_data.swap else {continue};
149
150                                let token_in = swap.get_first_token().map_or(
151                                    Arc::new(Token::new(Address::repeat_byte(0x11))), |x| x.clone()
152                                );
153
154                                let cost = U256::from(tx_compose_data.next_block_base_fee + tx_compose_data.priority_gas_fee) * U256::from(tx_compose_data.gas);
155
156                                let entry = txs_to_check.entry(*stuffing_tx_hash).or_insert(
157                                        TxToCheck{
158                                                block : tx_compose_data.next_block_number,
159                                                token_in : token_in.as_ref().clone(),
160                                                profit : U256::ZERO,
161                                                tips : U256::ZERO,
162                                                swap : swap.clone(),
163                                                cost,
164                                        }
165                                );
166                                let profit = swap.arb_profit();
167                                let profit = token_in.calc_eth_value(profit).unwrap_or_default();
168
169                                if entry.profit < profit {
170                                    entry.token_in = token_in.as_ref().clone();
171                                    entry.profit = profit;
172                                    entry.tips = tx_compose_data.tips.unwrap_or_default();
173
174                                    entry.swap = swap.clone();
175                                }
176                            }
177                        }
178                    }
179                    Err(e)=>{
180                        error!("tx_compose_channel_rx : {e}")
181                    }
182                }
183            }
184        }
185    }
186}
187
188pub struct StuffingTxMonitorActor<P> {
189    client: P,
190
191    tx_compose_channel_rx: Option<broadcast::Sender<MessageTxCompose>>,
192
193    market_events_rx: Option<broadcast::Sender<MarketEvents>>,
194
195    influxdb_write_channel_tx: Option<broadcast::Sender<WriteQuery>>,
196}
197
198impl<P: Provider<Ethereum> + Send + Sync + Clone + 'static> StuffingTxMonitorActor<P> {
199    pub fn new(client: P) -> Self {
200        StuffingTxMonitorActor { client, tx_compose_channel_rx: None, market_events_rx: None, influxdb_write_channel_tx: None }
201    }
202
203    pub fn with_tx_compose_channel(self, tx_compose_channel: broadcast::Sender<MessageTxCompose>) -> Self {
204        Self { tx_compose_channel_rx: Some(tx_compose_channel), ..self }
205    }
206
207    pub fn with_market_events(self, market_events: broadcast::Sender<MarketEvents>) -> Self {
208        Self { market_events_rx: Some(market_events), ..self }
209    }
210
211    pub fn on_bc(self, bc: &Blockchain) -> Self {
212        Self {
213            tx_compose_channel_rx: Some(bc.tx_compose_channel()),
214            market_events_rx: Some(bc.market_events_channel()),
215            influxdb_write_channel_tx: bc.influxdb_write_channel(),
216            ..self
217        }
218    }
219}
220
221impl<P> Component for StuffingTxMonitorActor<P>
222where
223    P: Provider<Ethereum> + Send + Sync + Clone + 'static,
224{
225    fn spawn(self, executor: TaskExecutor) -> Result<()> {
226        let tx_compose_channel_rx = self.tx_compose_channel_rx.clone().ok_or_else(|| eyre!("tx_compose_channel_rx not set"))?;
227        let market_events_rx = self.market_events_rx.clone().ok_or_else(|| eyre!("market_events_rx not set"))?;
228        let influxdb_write_channel_tx = self.influxdb_write_channel_tx.clone();
229
230        executor.spawn(async move {
231            if let Err(e) =
232                stuffing_tx_monitor_worker(self.client.clone(), tx_compose_channel_rx, market_events_rx, influxdb_write_channel_tx).await
233            {
234                tracing::error!("Stuffing tx monitor worker failed: {}", e);
235            }
236        });
237        Ok(())
238    }
239    fn name(&self) -> &'static str {
240        "StuffingTxMonitorActor"
241    }
242}
kabu_defi_health_monitor/stuffing_tx_monitor.rs

kabu_defi_health_monitor/
stuffing_tx_monitor.rs
