Avoid block when query (#388)

nz-neooo · web-flow · commit 8968e75a6104 · 2024-03-19T16:32:14.000+13:00
* Avoid block when query

* fmt

* Add timeout to rpc channel

* Add timeout to rpc channel

* fmt
diff --git a/apps/indexer-proxy/proxy/Cargo.toml b/apps/indexer-proxy/proxy/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "subql-indexer-proxy"
-version = "2.0.4"
+version = "2.0.5"
 edition = "2021"
 
 [dependencies]
diff --git a/apps/indexer-proxy/proxy/src/metrics.rs b/apps/indexer-proxy/proxy/src/metrics.rs
@@ -30,6 +30,7 @@ use std::time::Instant;
 use subql_indexer_utils::request::REQUEST_CLIENT;
 use sysinfo::{System, SystemExt};
 use tokio::sync::Mutex;
+use tokio::time::{sleep, Duration};
 
 use crate::cli::COMMAND;
 use crate::primitives::METRICS_LOOP_TIME;
@@ -101,7 +102,7 @@ struct Labels {
 pub fn listen() {
     tokio::spawn(async {
         loop {
-            tokio::time::sleep(std::time::Duration::from_secs(METRICS_LOOP_TIME)).await;
+            sleep(Duration::from_secs(METRICS_LOOP_TIME)).await;
 
             let lock = OWNER_COUNTER.lock().await;
             let current = lock.clone();
diff --git a/apps/indexer-proxy/proxy/src/p2p.rs b/apps/indexer-proxy/proxy/src/p2p.rs
@@ -40,6 +40,7 @@ use tdn::{
     },
 };
 use tokio::sync::{mpsc::Sender, RwLock};
+use tokio::time::{sleep, Duration};
 
 use crate::{
     account::{get_indexer, indexer_healthy},
@@ -57,7 +58,9 @@ pub static P2P_SENDER: Lazy<RwLock<Vec<ChannelRpcSender>>> = Lazy::new(|| RwLock
 pub async fn send(method: &str, params: Vec<RpcParam>, gid: GroupId) {
     let senders = P2P_SENDER.read().await;
     if !senders.is_empty() {
-        senders[0].send(rpc_request(0, method, params, gid)).await;
+        senders[0]
+            .send_timeout(rpc_request(0, method, params, gid), 100)
+            .await;
     }
 }
 
@@ -69,44 +72,53 @@ pub async fn stop_network() {
         debug!("RESTART NEW P2P NETWORK");
         senders[0].send(rpc_request(0, "p2p-stop", vec![], 0)).await;
         drop(senders);
-        tokio::time::sleep(std::time::Duration::from_secs(P2P_RESTART_TIME)).await;
+        sleep(Duration::from_secs(P2P_RESTART_TIME)).await;
     }
 }
 
-pub async fn report_conflict(deployment: &str, channel: &str, conflict: i32, start: i64, end: i64) {
+pub async fn report_conflict(
+    deployment: String,
+    channel: String,
+    conflict: i32,
+    start: i64,
+    end: i64,
+) {
     let senders = P2P_SENDER.read().await;
     if senders.is_empty() {
         warn!("NONE NETWORK WHEN REPORT CONFLICT");
     } else {
         senders[0]
-            .send(rpc_request(
-                0,
-                "payg-report-conflict",
-                vec![
-                    deployment.into(),
-                    channel.into(),
-                    conflict.into(),
-                    start.into(),
-                    end.into(),
-                ],
-                0,
-            ))
+            .send_timeout(
+                rpc_request(
+                    0,
+                    "payg-report-conflict",
+                    vec![
+                        deployment.into(),
+                        channel.into(),
+                        conflict.into(),
+                        start.into(),
+                        end.into(),
+                    ],
+                    0,
+                ),
+                100,
+            )
             .await;
         drop(senders);
     }
 }
 
 async fn check_stable() {
     loop {
-        tokio::time::sleep(std::time::Duration::from_secs(P2P_STABLE_TIME)).await;
+        sleep(Duration::from_secs(P2P_STABLE_TIME)).await;
         let senders = P2P_SENDER.read().await;
         if senders.is_empty() {
             drop(senders);
             continue;
         } else {
             debug!("Check stable connections");
             senders[0]
-                .send(rpc_request(0, "p2p-stable", vec![], 0))
+                .send_timeout(rpc_request(0, "p2p-stable", vec![], 0), 100)
                 .await;
         }
         drop(senders);
@@ -115,7 +127,7 @@ async fn check_stable() {
 
 async fn report_metrics() {
     loop {
-        tokio::time::sleep(std::time::Duration::from_secs(P2P_METRICS_TIME)).await;
+        sleep(Duration::from_secs(P2P_METRICS_TIME)).await;
         let senders = P2P_SENDER.read().await;
         if senders.is_empty() {
             drop(senders);
@@ -132,15 +144,15 @@ async fn report_metrics() {
 
 async fn report_status() {
     loop {
-        tokio::time::sleep(std::time::Duration::from_secs(P2P_METRICS_STATUS_TIME)).await;
+        sleep(Duration::from_secs(P2P_METRICS_STATUS_TIME)).await;
         let senders = P2P_SENDER.read().await;
         if senders.is_empty() {
             drop(senders);
             continue;
         } else {
             debug!("Report projects status");
             senders[0]
-                .send(rpc_request(0, "project-report-status", vec![], 0))
+                .send_timeout(rpc_request(0, "project-report-status", vec![], 0), 100)
                 .await;
         }
         drop(senders);
@@ -149,15 +161,15 @@ async fn report_status() {
 
 async fn broadcast_healthy() {
     loop {
-        tokio::time::sleep(std::time::Duration::from_secs(P2P_BROADCAST_HEALTHY_TIME)).await;
+        sleep(Duration::from_secs(P2P_BROADCAST_HEALTHY_TIME)).await;
         let senders = P2P_SENDER.read().await;
         if senders.is_empty() {
             drop(senders);
             continue;
         } else {
             debug!("Report projects healthy");
             senders[0]
-                .send(rpc_request(0, "project-broadcast-healthy", vec![], 0))
+                .send_timeout(rpc_request(0, "project-broadcast-healthy", vec![], 0), 100)
                 .await;
         }
         drop(senders);
diff --git a/apps/indexer-proxy/proxy/src/payg.rs b/apps/indexer-proxy/proxy/src/payg.rs
@@ -298,6 +298,7 @@ pub async fn query_state(
     state: &Value,
     network_type: MetricsNetwork,
 ) -> Result<(Vec<u8>, String, String)> {
+    debug!("Start handle query channel");
     let project = get_project(project_id).await?;
     let mut state = QueryState::from_json(state)?;
 
@@ -308,6 +309,7 @@ pub async fn query_state(
 
     // check channel state
     let (mut state_cache, keyname) = fetch_channel_cache(state.channel_id).await?;
+    debug!("Got channel cache");
 
     // check signer
     if !state_cache.signer.contains(&signer) {
@@ -351,22 +353,28 @@ pub async fn query_state(
     if local_next > remote_next + price {
         // mark conflict is happend
         let times = ((local_next - remote_next) / price).as_u32() as i32;
-        let now = Utc::now().timestamp();
         if times <= 1 {
-            state_cache.conflict = now;
+            state_cache.conflict = Utc::now().timestamp();
         }
-        let channel = format!("{:#x}", state.channel_id);
-        report_conflict(&project.id, &channel, times, state_cache.conflict, now).await;
     }
 
     if local_next > remote_next + price * conflict {
         warn!(
             "CONFLICT: local_next: {}, remote_next: {}, price: {}, conflict: {}",
             local_next, remote_next, price, conflict
         );
+
+        let project_id = project.id.clone();
+        let channel = format!("{:#x}", state.channel_id);
+        let times = conflict.as_u32() as i32;
+        let start = state_cache.conflict;
+        let end = Utc::now().timestamp();
+        tokio::spawn(report_conflict(project_id, channel, times, start, end));
+
         // overflow the conflict
         return Err(Error::PaygConflict(1050));
     }
+    debug!("Verified channel, start query");
 
     // query the data.
     let (data, signature) = project
@@ -418,7 +426,7 @@ pub async fn query_state(
             .map_err(|e| error!("{:?}", e));
     });
 
-    state.remote = state_cache.spent;
+    state.remote = local_next;
     debug!("Handle query channel success");
     let state_bytes = serde_json::to_vec(&state.to_json()).unwrap_or(vec![]);
     let state_string = general_purpose::STANDARD.encode(&state_bytes);
