added proper scheduled reconnection and matching_orders logic

Author: nazgull08

Cargo.lock

@@ -25,7 +25,7 @@ rustc-hex = "2.1.0"
 schemars = "0.8.0"
 serde = { version = "1.0.198", features = ["derive"] }
 serde_json = "1.0.116"
-spark-market-sdk = "0.6.4" 
+spark-market-sdk = "0.6.5" 
 pangea-client = { git = "https://github.com/nazgull08/pangea-client/"}
 thiserror = "1.0.63"
 tokio = { version = "1.41.0", features = ["rt", "macros", "time"] }

Cargo.toml

@@ -1,4 +1,5 @@
 use crate::indexer::spot_order::{LimitType, OrderStatus, OrderType, SpotOrder};
+use crate::storage::matching_orders::MatchingOrders;
 use crate::storage::order_book::OrderBook;
 use chrono::Utc;
 use log::{error, info};
@@ -27,7 +28,8 @@ pub struct PangeaOrderEvent {
     pub limit_type: Option<String>,
 }
 
-pub async fn handle_order_event(order_book: Arc<OrderBook>, event: PangeaOrderEvent) {
+pub async fn handle_order_event(order_book: Arc<OrderBook>, matching_orders: Arc<MatchingOrders>,
+    event: PangeaOrderEvent) {
     if let Some(event_type) = event.event_type.as_deref() {
         match event_type {
             "Open" => {
@@ -37,13 +39,17 @@ pub async fn handle_order_event(order_book: Arc<OrderBook>, event: PangeaOrderEv
                 }
             }
             "Trade" => {
+                matching_orders.remove(&event.order_id);
+                info!("Order {} removed from matching_orders", &event.order_id);
                 if let Some(match_size) = event.amount {
                     let o_type = event.order_type_to_enum();
                     let l_type = event.limit_type_to_enum();
                     process_trade(&order_book, &event.order_id, match_size, o_type, l_type);
                 }
             }
             "Cancel" => {
+                matching_orders.remove(&event.order_id);
+                info!("Order {} removed from matching_orders", &event.order_id);
                 order_book.remove_order(&event.order_id, event.order_type_to_enum());
                 info!(
                     "Removed order with id: {} due to Cancel event",

src/indexer/order_event_handler.rs

@@ -6,7 +6,7 @@ use pangea_client::{
     futures::StreamExt, provider::FuelProvider, query::Bound, requests::fuel::GetSparkOrderRequest,
     ClientBuilder, Format, WsProvider,
 };
-use tokio::time::sleep;
+use tokio::time::{interval, sleep};
 use std::collections::HashSet;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -16,14 +16,16 @@ use crate::config::env::ev;
 use crate::error::Error;
 use crate::indexer::order_event_handler::handle_order_event;
 use crate::indexer::order_event_handler::PangeaOrderEvent;
+use crate::storage::matching_orders::MatchingOrders;
 use crate::storage::order_book::OrderBook;
 
 pub async fn initialize_pangea_indexer(
     tasks: &mut Vec<tokio::task::JoinHandle<()>>,
     order_book: Arc<OrderBook>,
+    matching_orders: Arc<MatchingOrders>,
 ) -> Result<(), Error> {
     let ws_task_pangea = tokio::spawn(async move {
-        if let Err(e) = start_pangea_indexer(order_book).await {
+        if let Err(e) = start_pangea_indexer(order_book, matching_orders).await {
             eprintln!("Pangea error: {}", e);
         }
     });
@@ -32,22 +34,24 @@ pub async fn initialize_pangea_indexer(
     Ok(())
 }
 
-async fn start_pangea_indexer(order_book: Arc<OrderBook>) -> Result<(), Error> {
+async fn start_pangea_indexer(order_book: Arc<OrderBook>, matching_orders: Arc<MatchingOrders>) -> Result<(), Error> {
     let client = create_pangea_client().await?;
 
     let contract_start_block: i64 = ev("CONTRACT_START_BLOCK")?.parse()?;
     let contract_h256 = H256::from_str(&ev("CONTRACT_ID")?)?;
 
     let mut last_processed_block =
-        fetch_historical_data(&client, &order_book, contract_start_block, contract_h256).await?;
+        fetch_historical_data(&client, &order_book, &matching_orders,
+        contract_start_block, contract_h256).await?;
 
     if last_processed_block == 0 {
         last_processed_block = contract_start_block;
     }
 
     info!("Switching to listening for new orders (deltas)");
 
-    listen_for_new_deltas(&client, &order_book, last_processed_block, contract_h256).await
+    listen_for_new_deltas(&client, &order_book, &matching_orders,
+        last_processed_block, contract_h256).await
 }
 
 async fn create_pangea_client() -> Result<Client<WsProvider>, Error> {
@@ -79,6 +83,7 @@ async fn get_latest_block(chain_id: ChainId) -> Result<i64, Error> {
 async fn fetch_historical_data(
     client: &Client<WsProvider>,
     order_book: &Arc<OrderBook>,
+    matching_orders: &Arc<MatchingOrders>,
     contract_start_block: i64,
     contract_h256: H256,
 ) -> Result<i64, Error> {
@@ -115,7 +120,8 @@ async fn fetch_historical_data(
                 Ok(data) => {
                     let data = String::from_utf8(data)?;
                     let order: PangeaOrderEvent = serde_json::from_str(&data)?;
-                    handle_order_event(order_book.clone(), order).await;
+                    handle_order_event(order_book.clone(), matching_orders.clone(),
+                        order).await;
                 }
                 Err(e) => {
                     error!("Error in the stream of historical orders: {e}");
@@ -134,87 +140,92 @@ async fn fetch_historical_data(
     Ok(last_processed_block)
 }
 
+
 async fn listen_for_new_deltas(
     client: &Client<WsProvider>,
     order_book: &Arc<OrderBook>,
+    matching_orders: &Arc<MatchingOrders>,
     mut last_processed_block: i64,
     contract_h256: H256,
 ) -> Result<(), Error> {
-    let max_retries = 5;
-    let mut retry_count = 0;
-    let retry_delay = Duration::from_secs(5);
+    let mut retry_delay = Duration::from_secs(1);
+    let reconnect_interval = Duration::from_secs(10*60); 
+    let mut reconnect_timer = interval(reconnect_interval);
+    let mut processing = false;
 
     loop {
-        if retry_count > max_retries {
-            error!("Maximum number of retries exceeded. Exiting listen_for_new_deltas.");
-            return Err(Error::MaxRetriesExceeded);
-        }
-
-        let fuel_chain = match ev("CHAIN")?.as_str() {
-            "FUEL" => ChainId::FUEL,
-            _ => ChainId::FUELTESTNET,
-        };
-
-        let request_deltas = GetSparkOrderRequest {
-            from_block: Bound::Exact(last_processed_block + 1),
-            to_block: Bound::Subscribe,
-            market_id__in: HashSet::from([contract_h256]),
-            chains: HashSet::from([fuel_chain]),
-            ..Default::default()
-        };
-
-        let stream_deltas_result = client
-            .get_fuel_spark_orders_by_format(request_deltas, Format::JsonStream, true)
-            .await;
-
-        let stream_deltas = match stream_deltas_result {
-            Ok(stream) => {
-                retry_count = 0;
-                stream
-            }
-            Err(e) => {
-                error!("Failed to initiate stream: {}", e);
-                retry_count += 1;
-                error!("Retrying in {} seconds...", retry_delay.as_secs());
-                sleep(retry_delay).await;
-                continue;
-            }
-        };
-
-        pangea_client::futures::pin_mut!(stream_deltas);
-
-        while let Some(data_result) = stream_deltas.next().await {
-            match data_result {
-                Ok(data) => {
-                    retry_count = 0;
-
-                    if let Err(e) = process_order_data(&data, order_book, &mut last_processed_block).await {
-                        error!("Failed to process order data: {}", e);
+        tokio::select! {
+            _ = reconnect_timer.tick(), if !processing => {
+                info!("Scheduled reconnect to refresh connection...");
+                processing = false; // Сбросим состояние processing
+            },
+            result = async {
+                processing = true;
+                let fuel_chain = match ev("CHAIN")?.as_str() {
+                    "FUEL" => ChainId::FUEL,
+                    _ => ChainId::FUELTESTNET,
+                };
+
+                let request_deltas = GetSparkOrderRequest {
+                    from_block: Bound::Exact(last_processed_block + 1),
+                    to_block: Bound::Subscribe,
+                    market_id__in: HashSet::from([contract_h256]),
+                    chains: HashSet::from([fuel_chain]),
+                    ..Default::default()
+                };
+
+                match client
+                    .get_fuel_spark_orders_by_format(request_deltas, Format::JsonStream, true)
+                    .await
+                {
+                    Ok(stream_deltas) => {
+                        retry_delay = Duration::from_secs(1);
+                        pangea_client::futures::pin_mut!(stream_deltas);
+
+                        while let Some(data_result) = stream_deltas.next().await {
+                            match data_result {
+                                Ok(data) => {
+                                    if let Err(e) = process_order_data(&data, order_book, matching_orders, &mut last_processed_block).await {
+                                        error!("Failed to process order data: {}", e);
+                                    }
+                                }
+                                Err(e) => {
+                                    error!("Error in the stream of new orders (deltas): {}", e);
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        error!("Failed to initiate stream: {}", e);
                     }
                 }
-                Err(e) => {
-                    error!("Error in the stream of new orders (deltas): {}", e);
-                    retry_count += 1;
-                    error!("Retrying connection in {} seconds...", retry_delay.as_secs());
-                    sleep(retry_delay).await;
-                    break;
+
+                info!("Reconnecting to listen for new deltas in {} seconds...", retry_delay.as_secs());
+                sleep(retry_delay).await;
+                retry_delay = (retry_delay * 2).min(Duration::from_secs(60));
+                Ok::<(), Error>(())
+            } => {
+                processing = false;
+                if let Err(e) = result {
+                    error!("Error in listen_for_new_deltas: {:?}", e);
                 }
-            }
+            },
         }
-
-        info!("Reconnecting to listen for new deltas...");
-        sleep(retry_delay).await;
     }
 }
 
 async fn process_order_data(
     data: &[u8],
     order_book: &Arc<OrderBook>,
+    matching_orders: &Arc<MatchingOrders>,
     last_processed_block: &mut i64,
 ) -> Result<(), Error> {
     let data_str = String::from_utf8(data.to_vec())?;
     let order_event: PangeaOrderEvent = serde_json::from_str(&data_str)?;
     *last_processed_block = order_event.block_number;
-    handle_order_event(order_book.clone(), order_event).await;
+    handle_order_event(order_book.clone(),
+        matching_orders.clone(),
+        order_event).await;
     Ok(())
 }

src/indexer/pangea.rs

@@ -28,13 +28,13 @@ async fn main() -> Result<(), Error> {
     let matching_orders = Arc::new(MatchingOrders::new());
     let mut tasks = vec![];
 
-    initialize_pangea_indexer(&mut tasks, Arc::clone(&order_book)).await?;
+    initialize_pangea_indexer(&mut tasks, Arc::clone(&order_book), Arc::clone(&matching_orders)).await?;
 
     let port = ev("SERVER_PORT")?.parse()?;
     let rocket_task = tokio::spawn(run_rocket_server(port, Arc::clone(&order_book)));
     tasks.push(rocket_task);
     let matcher_ws_port = ev("MATCHERS_PORT")?.parse()?;
-    let matcher_websocket = Arc::new(MatcherWebSocket::new(order_book.clone()));
+    let matcher_websocket = Arc::new(MatcherWebSocket::new(order_book.clone(), matching_orders.clone()));
     let matcher_ws_task = tokio::spawn(run_matcher_websocket_server(
         matcher_websocket.clone(),
         matcher_ws_port

src/main.rs

@@ -1,28 +1,29 @@
 use crate::config::env::ev;
 use crate::indexer::spot_order::SpotOrder;
 use crate::matchers::types::{MatcherRequest, MatcherResponse};
+use crate::storage::matching_orders::MatchingOrders;
 use crate::storage::order_book::OrderBook;
 use futures_util::{SinkExt, StreamExt};
 use log::{error, info};
-use std::collections::HashSet;
 use std::sync::Arc;
 use tokio::net::TcpStream;
-use tokio::sync::Mutex;
 use tokio_tungstenite::tungstenite::protocol::Message;
 use tokio_tungstenite::WebSocketStream;
 
 use super::types::{MatcherConnectRequest, MatcherOrderUpdate};
 
 pub struct MatcherWebSocket {
     pub order_book: Arc<OrderBook>,
-    pub matching_orders: Arc<Mutex<HashSet<String>>>,  
+    pub matching_orders: Arc<MatchingOrders>
 }
 
 impl MatcherWebSocket {
-    pub fn new(order_book: Arc<OrderBook>) -> Self {
+    pub fn new(order_book: Arc<OrderBook>,
+        matching_orders: Arc<MatchingOrders>,
+    ) -> Self {
         Self {
             order_book,
-            matching_orders: Arc::new(Mutex::new(HashSet::new())),
+            matching_orders, 
         }
     }
 
@@ -74,33 +75,20 @@ impl MatcherWebSocket {
         if let Err(e) = write.send(Message::Text(response_text)).await {
             error!("Failed to send response to matcher {}: {}", uuid, e);
 
-            
-            let mut matching_orders = self.matching_orders.lock().await;
             for order in available_orders {
-                matching_orders.remove(&order.id);
+                self.matching_orders.remove(&order.id);
             }
         }
     }
 
     async fn handle_order_updates(&self, order_updates: Vec<MatcherOrderUpdate>, _uuid: String) {
         info!("Processing order updates: {:?}", order_updates);
-
-        let mut matching_orders = self.matching_orders.lock().await;
-        for update in order_updates {
-            matching_orders.remove(&update.order_id);
-            info!("Order {} removed from matching_orders", update.order_id);
-        }
     }
 
     async fn get_available_orders(&self, batch_size: usize) -> Vec<SpotOrder> {
         let mut available_orders = Vec::new();
 
-        
-        let matching_order_ids = {
-            let matching_orders = self.matching_orders.lock().await;
-            matching_orders.clone()
-        };
-
+        let matching_order_ids = self.matching_orders.get_all();
 
         let buy_orders = self.order_book.get_buy_orders()
             .values()
@@ -166,11 +154,9 @@ impl MatcherWebSocket {
         }
 
 
-        {
-            let mut matching_orders = self.matching_orders.lock().await;
-            for order_id in new_matching_order_ids {
-                matching_orders.insert(order_id);
-            }
+        for order_id in new_matching_order_ids {
+            self.matching_orders.add(&order_id);
+            info!("----Order {} added to matching_orders", order_id);
         }
 
         available_orders