kabu_strategy_backrun/
block_state_change_processor.rs

1use super::affected_pools_state::get_affected_pools_from_state_update;
2use alloy_consensus::BlockHeader;
3use eyre::{Result, eyre};
4use kabu_core_components::Component;
5use kabu_types_blockchain::ChainParameters;
6use kabu_types_entities::BlockHistory;
7use kabu_types_events::{MarketEvents, StateUpdateEvent};
8use kabu_types_market::Market;
9use reth_node_types::NodePrimitives;
10use reth_tasks::TaskExecutor;
11use revm::DatabaseRef;
12use std::sync::Arc;
13use tokio::sync::broadcast::error::RecvError;
14use tokio::sync::{RwLock, broadcast};
15use tracing::error;
16
17pub async fn block_state_change_worker<DB, LDT>(
18    chain_parameters: ChainParameters,
19    market: Arc<RwLock<Market>>,
20    block_history: Arc<RwLock<BlockHistory<DB, LDT>>>,
21    mut market_events_rx: broadcast::Receiver<MarketEvents>,
22    state_updates_broadcaster: broadcast::Sender<StateUpdateEvent<DB, LDT>>,
23) where
24    DB: DatabaseRef + Send + Sync + Clone + 'static,
25    LDT: NodePrimitives,
26    LDT::BlockHeader: BlockHeader,
27{
28    loop {
29        let market_event = match market_events_rx.recv().await {
30            Ok(market_event) => market_event,
31            Err(e) => match e {
32                RecvError::Closed => {
33                    error!("Market events txs channel closed");
34                    error!("MARKET_EVENTS_RX_CLOSED");
35                    break;
36                }
37                RecvError::Lagged(lag) => {
38                    error!("Market events txs channel lagged by {} messages", lag);
39                    continue;
40                }
41            },
42        };
43        let block_hash = match market_event {
44            MarketEvents::BlockStateUpdate { block_hash } => block_hash,
45            _ => continue,
46        };
47
48        let Some(block_history_entry) = block_history.read().await.get_block_history_entry(&block_hash).cloned() else {
49            error!("Block history entry not found in block history: {:?}", block_hash);
50            continue;
51        };
52
53        let Some(block_state_entry) = block_history.read().await.get_block_state(&block_hash).cloned() else {
54            error!("Block state not found in block history: {:?}", block_hash);
55            continue;
56        };
57
58        let Some(state_update) = block_history_entry.state_update.clone() else {
59            error!("Block {:?} has no state update", block_hash);
60            continue;
61        };
62
63        let affected_pools = get_affected_pools_from_state_update(market.clone(), &state_update).await;
64
65        if affected_pools.is_empty() {
66            error!("Could not get affected pools for block {:?}", block_hash);
67            continue;
68        };
69
70        let next_block_number = block_history_entry.header.number() + 1;
71        let next_block_timestamp = block_history_entry.header.timestamp() + 12;
72        let base_fee = block_history_entry.header.base_fee_per_gas().unwrap_or_default();
73        let gas_used = block_history_entry.header.gas_used();
74        let gas_limit = block_history_entry.header.gas_limit();
75        let next_base_fee = chain_parameters.calc_next_block_base_fee(gas_used, gas_limit, base_fee);
76
77        let request = StateUpdateEvent::new(
78            next_block_number,
79            next_block_timestamp,
80            next_base_fee,
81            block_state_entry,
82            state_update,
83            None,
84            affected_pools,
85            Vec::new(),
86            Vec::new(),
87            "block_searcher".to_string(),
88            90_00,
89        );
90        match state_updates_broadcaster.send(request) {
91            Ok(_) => {}
92            Err(error) => tracing::error!(%error, "ERROR_RUNNING_SYNC"),
93        }
94    }
95}
96
97pub struct BlockStateChangeProcessorComponent<DB: Clone + Send + Sync + 'static, LDT: NodePrimitives + 'static> {
98    chain_parameters: ChainParameters,
99    market: Option<Arc<RwLock<Market>>>,
100    block_history: Option<Arc<RwLock<BlockHistory<DB, LDT>>>>,
101    market_events_rx: Option<broadcast::Sender<MarketEvents>>,
102    state_updates_tx: Option<broadcast::Sender<StateUpdateEvent<DB, LDT>>>,
103}
104
105impl<DB: DatabaseRef + Send + Sync + Clone + 'static, LDT: NodePrimitives> BlockStateChangeProcessorComponent<DB, LDT> {
106    pub fn new() -> BlockStateChangeProcessorComponent<DB, LDT> {
107        BlockStateChangeProcessorComponent {
108            chain_parameters: ChainParameters::ethereum(),
109            market: None,
110            block_history: None,
111            market_events_rx: None,
112            state_updates_tx: None,
113        }
114    }
115
116    pub fn with_channels(
117        self,
118        chain_parameters: ChainParameters,
119        market: Arc<RwLock<Market>>,
120        block_history: Arc<RwLock<BlockHistory<DB, LDT>>>,
121        market_events_rx: broadcast::Sender<MarketEvents>,
122        state_updates_tx: broadcast::Sender<StateUpdateEvent<DB, LDT>>,
123    ) -> Self {
124        Self {
125            chain_parameters,
126            market: Some(market),
127            block_history: Some(block_history),
128            market_events_rx: Some(market_events_rx),
129            state_updates_tx: Some(state_updates_tx),
130        }
131    }
132}
133
134impl<DB: DatabaseRef + Send + Sync + Clone + 'static, LDT: NodePrimitives> Default for BlockStateChangeProcessorComponent<DB, LDT> {
135    fn default() -> Self {
136        Self::new()
137    }
138}
139
140impl<DB: DatabaseRef + Send + Sync + Clone + 'static, LDT: NodePrimitives> Component for BlockStateChangeProcessorComponent<DB, LDT> {
141    fn spawn(self, executor: TaskExecutor) -> Result<()> {
142        let name = self.name();
143
144        let market_events_rx = self.market_events_rx.ok_or_else(|| eyre!("market_events_rx not set"))?.subscribe();
145        let state_updates_tx = self.state_updates_tx.ok_or_else(|| eyre!("state_updates_tx not set"))?;
146        let market = self.market.ok_or_else(|| eyre!("market not set"))?;
147        let block_history = self.block_history.ok_or_else(|| eyre!("block_history not set"))?;
148
149        executor.spawn_critical(
150            name,
151            block_state_change_worker(self.chain_parameters, market, block_history, market_events_rx, state_updates_tx),
152        );
153
154        Ok(())
155    }
156
157    fn name(&self) -> &'static str {
158        "BlockStateChangeProcessorComponent"
159    }
160}
kabu_strategy_backrun/block_state_change_processor.rs

kabu_strategy_backrun/
block_state_change_processor.rs
