kabu_defi_health_monitor/
pool_health_monitor.rs1use alloy_primitives::Address;
2use eyre::Result;
3use influxdb::{Timestamp, WriteQuery};
4use std::collections::{HashMap, HashSet};
5use std::sync::Arc;
6use std::time::Duration;
7use tokio::sync::broadcast::error::RecvError;
8use tokio::sync::{RwLock, broadcast};
9use tracing::{debug, error, info};
10
11use kabu_core_components::Component;
12use kabu_defi_address_book::TokenAddressEth;
13use kabu_types_events::{HealthEvent, MessageHealthEvent};
14use kabu_types_market::{Market, PoolId, PoolProtocol};
15use lazy_static::lazy_static;
16use reth_tasks::TaskExecutor;
17
18lazy_static! {
19 static ref TRUSTED_TOKENS: HashSet<Address> = HashSet::from_iter(vec![
20 TokenAddressEth::WETH,
21 TokenAddressEth::USDC,
22 TokenAddressEth::USDT,
23 TokenAddressEth::STETH,
24 TokenAddressEth::WSTETH,
25 TokenAddressEth::WBTC,
26 TokenAddressEth::CRV,
27 TokenAddressEth::DAI
28 ]);
29}
30
31pub async fn pool_health_monitor_worker(
32 market: Arc<RwLock<Market>>,
33 mut pool_health_monitor_rx: broadcast::Receiver<MessageHealthEvent>,
34 influx_channel_tx: Option<broadcast::Sender<WriteQuery>>,
35) {
36 let mut pool_errors_map: HashMap<PoolId, u32> = HashMap::new();
37 loop {
40 tokio::select! {
41 msg = pool_health_monitor_rx.recv() => {
42
43 let pool_health_update : Result<MessageHealthEvent, RecvError> = msg;
44 match pool_health_update {
45 Ok(pool_health_message)=>{
46 match pool_health_message.inner {
47 HealthEvent::SwapLineEstimationError(estimate_error) => {
48 debug!("SwapPath health_monitor message update: {} path : {}", estimate_error.msg, estimate_error.swap_path);
49 let start_time=std::time::Instant::now();
55 let mut market_guard = market.write().await;
56 debug!(elapsed = start_time.elapsed().as_micros(), "market_guard market.write acquired");
57 let is_ok = market_guard.set_path_disabled(&estimate_error.swap_path, true);
58 info!("Disabling path : swap_path={} msg={} hash={:#?} ok={}", estimate_error.swap_path, estimate_error.msg, estimate_error.swap_path.get_hash(),is_ok);
59
60
61
62 for (idx, pool) in estimate_error.swap_path.pools.iter().enumerate() {
63 let tokens = [estimate_error.swap_path.tokens[idx].get_address(), estimate_error.swap_path.tokens[idx+1].get_address()];
64 if pool.get_protocol() == PoolProtocol::UniswapV2Like || pool.get_protocol() == PoolProtocol::UniswapV3Like {
65 let pool_id = pool.get_pool_id();
67
68
69
70 market_guard.set_pool_disabled(&pool_id, &tokens[0], &tokens[1], true);
72
73 match market_guard.get_pool(&pool_id) {
74 Some(pool)=>{
75 info!("Disabling pool: protocol={}, pool_id={}, msg=ESTIMATION_FAILED", pool.get_protocol(),pool_id);
76
77 let amount_f64 = -2.0f64;
78 let pool_protocol = pool.get_protocol().to_string();
79 let pool_id = pool.get_pool_id().to_string();
80 let influx_channel_clone = influx_channel_tx.clone();
81
82 if let Err(e) = tokio::time::timeout(
83 Duration::from_secs(1),
84 async move {
85 let start_time_utc = chrono::Utc::now();
86
87 let write_query = WriteQuery::new(Timestamp::from(start_time_utc), "pool_disabled")
88 .add_field("message", "ESTIMATION_FAILED")
89 .add_field("amount", amount_f64)
90 .add_tag("id", pool_id)
91 .add_tag("protocol", pool_protocol)
92 .add_tag("token_from", tokens[0].to_string())
93 .add_tag("token_to", tokens[1].to_string());
94
95 if let Some(tx) = influx_channel_clone
96 && let Err(e) = tx.send(write_query) {
97 error!("Failed to failed pool to influxdb: {:?}", e);
98 }
99 }
100 ).await {
101 error!("Failed to send failed pool info to influxdb: {:?}", e);
102 }
103 }
104 _=>{
105 error!("Disabled pool missing in market: address={}", pool_id);
106 }
107 }
108 }
110
111
112
113 }
114
115
116 drop(market_guard);
117 debug!(elapsed = start_time.elapsed().as_micros(), "market_guard market.write released");
118
119 }
122 HealthEvent::PoolSwapError(swap_error)=>{
123 debug!("Pool health_monitor message update: {:?} {} {} ", swap_error.pool, swap_error.kind, swap_error.amount);
124 let entry = pool_errors_map.entry(swap_error.pool).or_insert(0);
125 *entry += 1;
126 if *entry >= 10 {
127 let start_time=std::time::Instant::now();
128 let mut market_guard = market.write().await;
129 debug!(elapsed = start_time.elapsed().as_micros(), "market_guard market.write acquired");
130
131 market_guard.set_pool_disabled(&swap_error.pool, &swap_error.token_from, &swap_error.token_to, true);
133
134
135 match market_guard.get_pool(&swap_error.pool) {
136 Some(pool)=>{
137 info!("Disabling pool: protocol={}, address={:?}, error={} amount={}", pool.get_protocol(),swap_error.pool, swap_error.kind, swap_error.amount);
138
139 let amount_f64 = if let Some(token_in) = market_guard.get_token(&swap_error.token_from) {
140 token_in.to_float(swap_error.amount)
141 } else {
142 -1.0f64
143 };
144
145 let pool_protocol = pool.get_protocol().to_string();
146 let pool_id = pool.get_pool_id().to_string();
147 let influx_channel_clone = influx_channel_tx.clone();
148
149 if let Err(e) = tokio::time::timeout(
150 Duration::from_secs(1),
151 async move {
152 let start_time_utc = chrono::Utc::now();
153
154 let write_query = WriteQuery::new(Timestamp::from(start_time_utc), "pool_disabled")
155 .add_field("message", swap_error.kind.to_string())
156 .add_field("amount", amount_f64)
157 .add_tag("id", pool_id)
158 .add_tag("protocol", pool_protocol)
159 .add_tag("token_from", swap_error.token_from.to_string())
160 .add_tag("token_to", swap_error.token_to.to_string());
161
162 if let Some(tx) = influx_channel_clone
163 && let Err(e) = tx.send(write_query) {
164 error!("Failed to failed pool to influxdb: {:?}", e);
165 }
166 }
167 ).await {
168 error!("Failed to send failed pool info to influxdb: {:?}", e);
169 }
170
171
172
173 }
174 _=>{
175 error!("Disabled pool missing in market: address={:?}, error={} amount={}", swap_error.pool, swap_error.kind, swap_error.amount);
176 }
177 }
178 drop(market_guard);
181 debug!(elapsed = start_time.elapsed().as_micros(), "market_guard market.write released");
182
183 }
184 }
185 _=>{}
186 }
187 }
188 Err(RecvError::Closed)=>{
189 debug!("pool_health_update channel closed, shutting down");
190 break;
191 }
192 Err(e)=>{
193 error!("pool_health_update error {}", e)
194 }
195 }
196
197 }
198 }
199 }
200}
201
202#[derive(Default)]
203pub struct PoolHealthMonitorComponent {
204 market: Option<Arc<RwLock<Market>>>,
205 pool_health_update_rx: Option<broadcast::Sender<MessageHealthEvent>>,
206 influxdb_tx: Option<broadcast::Sender<WriteQuery>>,
207}
208
209impl PoolHealthMonitorComponent {
210 pub fn new() -> Self {
211 PoolHealthMonitorComponent::default()
212 }
213
214 pub fn with_channels(
215 self,
216 market: Arc<RwLock<Market>>,
217 pool_health_update_rx: broadcast::Sender<MessageHealthEvent>,
218 influxdb_tx: Option<broadcast::Sender<WriteQuery>>,
219 ) -> Self {
220 Self { market: Some(market), pool_health_update_rx: Some(pool_health_update_rx), influxdb_tx }
221 }
222}
223
224impl Component for PoolHealthMonitorComponent {
225 fn spawn(self, executor: TaskExecutor) -> Result<()> {
226 let name = self.name();
227
228 let market = self.market.ok_or_else(|| eyre::eyre!("market not set"))?;
229 let pool_health_update_rx = self.pool_health_update_rx.ok_or_else(|| eyre::eyre!("pool_health_update_rx not set"))?.subscribe();
230
231 executor.spawn_critical(name, pool_health_monitor_worker(market, pool_health_update_rx, self.influxdb_tx));
232
233 Ok(())
234 }
235 fn name(&self) -> &'static str {
236 "PoolHealthMonitorComponent"
237 }
238}