kabu_strategy_merger/
diffpath_merger_actor.rs

1use alloy_primitives::{Address, TxHash};
2use eyre::{OptionExt, Result};
3use kabu_core_components::Component;
4use lazy_static::lazy_static;
5use reth_tasks::TaskExecutor;
6use revm::{Database, DatabaseCommit, DatabaseRef};
7use std::collections::HashSet;
8use tokio::sync::{broadcast, broadcast::Receiver, broadcast::error::RecvError};
9use tracing::{debug, error, info};
10
11use kabu_evm_utils::NWETH;
12use kabu_types_events::{MarketEvents, MessageSwapCompose, SwapComposeData, SwapComposeMessage, TxComposeData};
13use kabu_types_market::MarketState;
14use kabu_types_swap::Swap;
15
16lazy_static! {
17    static ref COINBASE: Address = "0x1f9090aaE28b8a3dCeaDf281B0F12828e676c326".parse().unwrap();
18}
19
20fn get_merge_list<'a, DB: Clone + Send + Sync + 'static>(
21    request: &SwapComposeData<DB>,
22    swap_paths: &'a [SwapComposeData<DB>],
23) -> Vec<&'a SwapComposeData<DB>> {
24    let mut ret: Vec<&SwapComposeData<DB>> = Vec::new();
25    let mut pools = request.swap.get_pool_id_vec();
26    for p in swap_paths.iter() {
27        if !p.cross_pools(&pools) {
28            pools.extend(p.swap.get_pool_id_vec());
29            ret.push(p);
30        }
31    }
32    ret
33}
34
35async fn diff_path_merger_worker<DB>(
36    market_events_rx: broadcast::Sender<MarketEvents>,
37    compose_channel_rx: broadcast::Sender<MessageSwapCompose<DB>>,
38    compose_channel_tx: broadcast::Sender<MessageSwapCompose<DB>>,
39) -> Result<()>
40where
41    DB: DatabaseRef + Database + DatabaseCommit + Send + Sync + Clone + 'static,
42{
43    let mut market_events_rx: Receiver<MarketEvents> = market_events_rx.subscribe();
44
45    let mut compose_channel_rx: Receiver<MessageSwapCompose<DB>> = compose_channel_rx.subscribe();
46
47    let mut swap_paths: Vec<SwapComposeData<DB>> = Vec::new();
48    let mut processed_swaps: HashSet<String> = HashSet::new();
49
50    loop {
51        tokio::select! {
52            msg = market_events_rx.recv() => {
53                if let Ok(msg) = msg {
54                    let market_event_msg : MarketEvents = msg;
55                    if let MarketEvents::BlockHeaderUpdate{block_number, block_hash, timestamp, base_fee, next_base_fee} =  market_event_msg {
56                        debug!("Block header update {} {} ts {} base_fee {} next {} ", block_number, block_hash, timestamp, base_fee, next_base_fee);
57                        //cur_block_number = Some( block_number + 1);
58                        //cur_block_time = Some(timestamp + 12 );
59                        //cur_next_base_fee = next_base_fee;
60                        //cur_base_fee = base_fee;
61                        swap_paths = Vec::new();
62                        processed_swaps.clear();
63
64                        // for _counter in 0..5  {
65                        //     if let Ok(msg) = market_events_rx.recv().await {
66                        //         if matches!(msg, MarketEvents::BlockStateUpdate{ block_hash } ) {
67                        //             cur_state_override = latest_block.read().await.node_state_override();
68                        //             debug!("Block state update received {} {}", block_number, block_hash);
69                        //             break;
70                        //         }
71                        //     }
72                        // }
73                    }
74                }
75            }
76
77            msg = compose_channel_rx.recv() => {
78                let msg : Result<MessageSwapCompose<DB>, RecvError> = msg;
79                match msg {
80                    Ok(compose_request)=>{
81                        if let SwapComposeMessage::Ready(sign_request) = compose_request.inner()
82                            && (matches!( sign_request.swap, Swap::BackrunSwapLine(_)) || matches!( sign_request.swap, Swap::BackrunSwapSteps(_))) {
83                                // Create a unique key for this swap to detect duplicates
84                                let swap_key = format!("{:?}", sign_request.swap);
85
86                                // Skip if we've already processed this exact swap
87                                if processed_swaps.contains(&swap_key) {
88                                    debug!("Skipping duplicate swap: {}", swap_key);
89                                    continue;
90                                }
91
92                                let mut merge_list = get_merge_list(sign_request, &swap_paths);
93
94                                if !merge_list.is_empty() {
95                                    let swap_vec : Vec<Swap> = merge_list.iter().map(|x|x.swap.clone()).collect();
96                                    info!("Merging started {:?}", swap_vec );
97
98                                    let mut state = MarketState::new(sign_request.poststate.clone().unwrap().clone());
99
100                                    for dbs in merge_list.iter() {
101                                        state.apply_geth_update_vec( dbs.poststate_update.clone().ok_or_eyre("NO_STATE_UPDATE")?);
102                                    }
103
104                                    merge_list.push(sign_request);
105
106                                    let mut stuffing_txs_hashes : Vec<TxHash> = Vec::new();
107                                    let mut stuffing_txs = Vec::new();
108
109                                    for req in merge_list.iter() {
110                                        for tx in req.tx_compose.stuffing_txs.iter() {
111                                            let tx_hash = *tx.tx_hash();
112                                            if !stuffing_txs_hashes.contains(&tx_hash) {
113                                                stuffing_txs_hashes.push(tx_hash);
114                                                stuffing_txs.push(tx.clone());
115                                            }
116                                        }
117                                    }
118
119                                    let encode_request = MessageSwapCompose::prepare(
120                                        SwapComposeData {
121                                            tx_compose : TxComposeData {
122                                                stuffing_txs_hashes,
123                                                stuffing_txs,
124                                                ..sign_request.tx_compose.clone()
125                                            },
126                                            swap : Swap::Multiple( merge_list.iter().map(|i| i.swap.clone()  ).collect()) ,
127                                            origin : Some("diffpath_merger".to_string()),
128                                            tips_pct : Some(9000),
129                                            poststate : Some(state.state_db),
130                                            ..sign_request.clone()
131                                        }
132                                    );
133                                    info!("+++ Calculation finished. Merge list : {} profit : {}",merge_list.len(), NWETH::to_float(encode_request.inner.swap.arb_profit_eth())  );
134
135                                    if let Err(e) = compose_channel_tx.send(encode_request) {
136                                       error!("{}",e)
137                                    }
138                                }
139
140                                processed_swaps.insert(swap_key);
141                                swap_paths.push(sign_request.clone());
142                                swap_paths.sort_by(|a, b| b.swap.arb_profit_eth().cmp(&a.swap.arb_profit_eth() ) );
143
144                                // Keep only top 20 most profitable swaps to prevent O(n²) comparison explosion
145                                if swap_paths.len() > 20 {
146                                    swap_paths.truncate(20);
147                                }
148                        }
149                    }
150                    Err(e)=>{error!("{e}")}
151                }
152
153            }
154
155        }
156    }
157}
158
159#[derive(Clone)]
160pub struct DiffPathMergerComponent<DB: Clone + Send + Sync + 'static> {
161    market_events: Option<broadcast::Sender<MarketEvents>>,
162
163    compose_channel_rx: Option<broadcast::Sender<MessageSwapCompose<DB>>>,
164
165    compose_channel_tx: Option<broadcast::Sender<MessageSwapCompose<DB>>>,
166
167    _db: std::marker::PhantomData<DB>,
168}
169
170impl<DB> Default for DiffPathMergerComponent<DB>
171where
172    DB: DatabaseRef + Database + DatabaseCommit + Send + Sync + Clone + Default + 'static,
173{
174    fn default() -> Self {
175        Self::new()
176    }
177}
178
179impl<DB> DiffPathMergerComponent<DB>
180where
181    DB: DatabaseRef + Database + DatabaseCommit + Send + Sync + Clone + Default + 'static,
182{
183    pub fn new() -> Self {
184        Self { market_events: None, compose_channel_rx: None, compose_channel_tx: None, _db: std::marker::PhantomData }
185    }
186
187    pub fn with_market_events_channel(self, market_events: broadcast::Sender<MarketEvents>) -> Self {
188        Self { market_events: Some(market_events), ..self }
189    }
190
191    pub fn with_compose_channel(self, compose_channel: broadcast::Sender<MessageSwapCompose<DB>>) -> Self {
192        Self { compose_channel_rx: Some(compose_channel.clone()), compose_channel_tx: Some(compose_channel), ..self }
193    }
194}
195
196impl<DB> Component for DiffPathMergerComponent<DB>
197where
198    DB: DatabaseRef + Database + DatabaseCommit + Send + Sync + Clone + Default + 'static,
199{
200    fn spawn(self, executor: TaskExecutor) -> Result<()> {
201        let name = self.name();
202
203        executor.spawn_critical(name, async move {
204            if let Err(e) =
205                diff_path_merger_worker(self.market_events.unwrap(), self.compose_channel_rx.unwrap(), self.compose_channel_tx.unwrap())
206                    .await
207            {
208                error!("Diff path merger worker failed: {}", e);
209            }
210        });
211
212        Ok(())
213    }
214
215    fn name(&self) -> &'static str {
216        "DiffPathMergerComponent"
217    }
218}
kabu_strategy_merger/diffpath_merger_actor.rs

kabu_strategy_merger/
diffpath_merger_actor.rs
