kabu_strategy_backrun/
arb_actor.rs

1use super::{PendingTxStateChangeProcessorComponent, StateChangeArbSearcherComponent};
2use crate::BackrunConfig;
3use crate::block_state_change_processor::BlockStateChangeProcessorComponent;
4use alloy_network::Network;
5use alloy_provider::Provider;
6use eyre::Result;
7use influxdb::WriteQuery;
8use kabu_core_components::Component;
9use kabu_types_blockchain::ChainParameters;
10use reth_tasks::TaskExecutor;
11use std::sync::Arc;
12use tokio::sync::{RwLock, broadcast};
13
14use kabu_evm_db::KabuDBError;
15use kabu_node_debug_provider::DebugProviderExt;
16use kabu_types_blockchain::Mempool;
17use kabu_types_entities::BlockHistory;
18use kabu_types_events::{MarketEvents, MempoolEvents, MessageHealthEvent, MessageSwapCompose, StateUpdateEvent};
19use kabu_types_market::{Market, MarketState};
20use reth_node_types::NodePrimitives;
21use revm::{Database, DatabaseCommit, DatabaseRef};
22use std::marker::PhantomData;
23use tracing::info;
24
25pub struct StateChangeArbComponent<P, N, DB: Clone + Send + Sync + 'static, LDT: NodePrimitives + 'static> {
26    backrun_config: BackrunConfig,
27    client: P,
28
29    market: Option<Arc<RwLock<Market>>>,
30
31    mempool: Option<Arc<RwLock<Mempool<LDT>>>>,
32
33    chain_parameters: ChainParameters,
34
35    market_state: Option<Arc<RwLock<MarketState<DB>>>>,
36
37    block_history: Option<Arc<RwLock<BlockHistory<DB, LDT>>>>,
38
39    mempool_events_tx: Option<broadcast::Sender<MempoolEvents>>,
40
41    market_events_tx: Option<broadcast::Sender<MarketEvents>>,
42
43    swap_compose_channel_tx: Option<broadcast::Sender<MessageSwapCompose<DB, LDT>>>,
44
45    pool_health_monitor_tx: Option<broadcast::Sender<MessageHealthEvent>>,
46
47    influxdb_tx: Option<broadcast::Sender<WriteQuery>>,
48
49    _n: PhantomData<N>,
50}
51
52impl<P, N, DB, LDT> StateChangeArbComponent<P, N, DB, LDT>
53where
54    N: Network,
55    P: Provider<N> + DebugProviderExt<N> + Send + Sync + Clone + 'static,
56    DB: DatabaseRef + Send + Sync + Clone + 'static,
57    LDT: NodePrimitives + 'static,
58{
59    pub fn new(client: P, backrun_config: BackrunConfig) -> StateChangeArbComponent<P, N, DB, LDT> {
60        StateChangeArbComponent {
61            backrun_config,
62            client,
63            chain_parameters: ChainParameters::ethereum(),
64            market: None,
65            mempool: None,
66            block_history: None,
67            market_state: None,
68            mempool_events_tx: None,
69            market_events_tx: None,
70            swap_compose_channel_tx: None,
71            pool_health_monitor_tx: None,
72            influxdb_tx: None,
73            _n: PhantomData,
74        }
75    }
76
77    pub fn with_market(self, market: Arc<RwLock<Market>>) -> Self {
78        Self { market: Some(market), ..self }
79    }
80
81    pub fn with_mempool(self, mempool: Arc<RwLock<Mempool<LDT>>>) -> Self {
82        Self { mempool: Some(mempool), ..self }
83    }
84
85    pub fn with_market_state(self, market_state: Arc<RwLock<MarketState<DB>>>) -> Self {
86        Self { market_state: Some(market_state), ..self }
87    }
88
89    pub fn with_block_history(self, block_history: Arc<RwLock<BlockHistory<DB, LDT>>>) -> Self {
90        Self { block_history: Some(block_history), ..self }
91    }
92
93    pub fn with_mempool_events_channel(self, mempool_events_tx: broadcast::Sender<MempoolEvents>) -> Self {
94        Self { mempool_events_tx: Some(mempool_events_tx), ..self }
95    }
96
97    pub fn with_market_events_channel(self, market_events_tx: broadcast::Sender<MarketEvents>) -> Self {
98        Self { market_events_tx: Some(market_events_tx), ..self }
99    }
100
101    pub fn with_swap_compose_channel(self, swap_compose_channel_tx: broadcast::Sender<MessageSwapCompose<DB, LDT>>) -> Self {
102        Self { swap_compose_channel_tx: Some(swap_compose_channel_tx), ..self }
103    }
104
105    pub fn with_pool_health_monitor_channel(self, pool_health_monitor_tx: broadcast::Sender<MessageHealthEvent>) -> Self {
106        Self { pool_health_monitor_tx: Some(pool_health_monitor_tx), ..self }
107    }
108
109    pub fn with_influxdb_channel(self, influxdb_tx: broadcast::Sender<WriteQuery>) -> Self {
110        Self { influxdb_tx: Some(influxdb_tx), ..self }
111    }
112}
113
114impl<P, N, DB, LDT> Component for StateChangeArbComponent<P, N, DB, LDT>
115where
116    N: Network<TransactionRequest = alloy_rpc_types::TransactionRequest>,
117    P: Provider<N> + DebugProviderExt<N> + Send + Sync + Clone + 'static,
118    DB: DatabaseRef<Error = KabuDBError> + Database<Error = KabuDBError> + DatabaseCommit + Send + Sync + Clone + Default + 'static,
119    LDT: NodePrimitives + 'static,
120{
121    fn spawn(self, executor: TaskExecutor) -> Result<()> {
122        // Create state update channel for searcher
123        let (searcher_pool_update_channel, _): (broadcast::Sender<StateUpdateEvent<DB, LDT>>, _) = broadcast::channel(100);
124
125        // Spawn state change searcher
126        let searcher = StateChangeArbSearcherComponent::new(self.backrun_config.clone())
127            .with_channels(
128                searcher_pool_update_channel.clone(),
129                self.swap_compose_channel_tx.clone().unwrap(),
130                self.pool_health_monitor_tx.clone().unwrap(),
131                self.influxdb_tx,
132            )
133            .with_market(self.market.clone().unwrap());
134        searcher.spawn(executor.clone())?;
135        info!("State change searcher component started successfully");
136
137        if self.mempool_events_tx.is_some() {
138            let pending_tx_processor = PendingTxStateChangeProcessorComponent::new(self.client.clone())
139                .with_channels(
140                    self.market_events_tx.clone().unwrap(),
141                    self.mempool_events_tx.clone().unwrap(),
142                    searcher_pool_update_channel.clone(),
143                )
144                .with_market(self.market.clone().unwrap())
145                .with_mempool(self.mempool.clone())
146                .with_market_state(self.market_state.clone().unwrap());
147            pending_tx_processor.spawn(executor.clone())?;
148            info!("Pending tx state processor component started successfully");
149        }
150
151        if self.market_events_tx.is_some() {
152            let block_processor = BlockStateChangeProcessorComponent::new().with_channels(
153                self.chain_parameters.clone(),
154                self.market.clone().unwrap(),
155                self.block_history.clone().unwrap(),
156                self.market_events_tx.clone().unwrap(),
157                searcher_pool_update_channel.clone(),
158            );
159            block_processor.spawn(executor.clone())?;
160            info!("Block state change processor component started successfully");
161        }
162
163        Ok(())
164    }
165    fn name(&self) -> &'static str {
166        "StateChangeArbComponent"
167    }
168}
kabu_strategy_backrun/arb_actor.rs

kabu_strategy_backrun/
arb_actor.rs
