Merge pull request #129 from tonlabs/check-sync

Check sync

Author: diserere

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Release Notes
 All notable changes to this project will be documented in this file.
 
+## Jun 05, 2020
+### New
+- time sync check while initializing
+
 ## Jun 03, 2020
 ### New
 - local run functions return updated contract state when running with `full_run = true`

ton_client/client/src/setup.rs

@@ -40,6 +40,7 @@ pub(crate) struct SetupParams {
     pub message_processing_timeout_grow_factor: Option<f32>,
     pub wait_for_timeout: Option<u32>,
     pub access_key: Option<String>,
+    pub out_of_sync_threshold: Option<i64>,
 }
 
 impl Into<NodeClientConfig> for SetupParams {
@@ -53,6 +54,7 @@ impl Into<NodeClientConfig> for SetupParams {
                 message_processing_timeout: self.message_processing_timeout.unwrap_or(default.message_processing_timeout),
                 message_processing_timeout_grow_factor: self.message_processing_timeout_grow_factor.unwrap_or(default.message_processing_timeout_grow_factor),
                 wait_for_timeout: self.wait_for_timeout.unwrap_or(default.wait_for_timeout),
+                out_of_sync_threshold: self.out_of_sync_threshold.unwrap_or(default.out_of_sync_threshold),
             }),
             #[cfg(feature = "node_interaction")]
             base_url: self.base_url,
@@ -73,7 +75,7 @@ fn setup(context: &mut ClientContext, config: SetupParams) -> ApiResult<()> {
         .map_err(|err| ApiError::cannot_create_runtime(err))?;
 
     let client = runtime.block_on(ton_sdk::init(config.into()))
-        .map_err(|err|ApiError::config_init_failed(err))?;
+        .map_err(|err| crate::types::apierror_from_sdkerror(&err, ApiError::config_init_failed))?;
 
     context.client = Some(client);
     context.runtime = Some(runtime);

ton_client/client/src/tests/mod.rs

@@ -285,19 +285,20 @@ fn test_wallet_deploy() {
             }),
     ).unwrap();
 
-    let deployed = client.request("contracts.deploy",
+    let deployed = parse_object(client.request("contracts.deploy",
         json!({
                 "abi": abi.clone(),
                 "constructorParams": json!({}),
                 "imageBase64": WALLET_CODE_BASE64,
                 "keyPair": keys,
                 "workchainId": 0,
             }),
-    ).unwrap();
+    ));
 
-    assert_eq!(format!("{{\"address\":\"{}\",\"alreadyDeployed\":false}}", address), deployed);
+    assert_eq!(deployed["address"], address.to_string());
+    assert_eq!(deployed["alreadyDeployed"], false);
 
-    let result = client.request("contracts.run",
+    let result = parse_object(client.request("contracts.run",
         json!({
                 "address": address.to_string(),
                 "abi": abi.clone(),
@@ -307,8 +308,8 @@ fn test_wallet_deploy() {
 				}),
                 "keyPair": keys,
             }),
-    ).unwrap();
-    assert_eq!("{\"output\":{\"value0\":\"0x0\"}}", result);
+    ));
+    assert_eq!(result["output"]["value0"], "0x0");
 }
 
 const GIVER_ADDRESS: &str = "0:841288ed3b55d9cdafa806807f02a0ae0c169aa5edfe88a789a6482429756a94";
@@ -482,3 +483,19 @@ fn test_address_parsing() {
         })).unwrap(),
         base64_url);
 }
+
+
+#[test]
+fn test_out_of_sync() {
+    let client = TestClient::new();
+
+    let result = client.request("setup",
+        json!({
+            "baseUrl": "http://localhost",
+            "outOfSyncThreshold": -1
+        })).unwrap_err();
+
+    let value: Value = serde_json::from_str(&result).unwrap();
+
+   assert_eq!(value["code"], 1013);
+}

ton_client/client/src/types.rs

@@ -188,7 +188,7 @@ impl ApiError {
         let mut error = ApiError::new(
             ApiErrorSource::Node,
             &ApiSdkErrorCode::TransactionsLag,
-            "Existing block transaction not found".to_owned(),
+            "Existing block transaction not found (no transaction appeared for the masterchain block with gen_utime > message expiration time)".to_owned(),
         );
 
         error.data = serde_json::json!({
@@ -270,6 +270,22 @@ impl ApiError {
         error
     }
 
+    pub fn clock_out_of_sync(delta_ms: i64, threshold: i64, expiration_timout: u32) -> Self {
+        let mut error = ApiError::new(
+            ApiErrorSource::Node,
+            &ApiSdkErrorCode::ClockOutOfSync,
+            "Device clock is out of sync with server time".to_owned(),
+        );
+
+        error.data = serde_json::json!({
+            "delta_ms": delta_ms,
+            "threshold_ms": threshold,
+            "expiration_timout_ms": expiration_timout,
+            "tip": "Synchronize your device clock with internet time"
+        });
+        error
+    }
+
     // SDK Cell
 
     pub fn cell_invalid_query<E: Display>(s: E) -> Self {
@@ -816,6 +832,8 @@ where
             ApiError::transactions_lag(msg_id.to_string(), *send_time, block_id.clone(), *timeout),
         Some(SdkError::TransactionWaitTimeout{msg_id, msg: _, send_time, timeout}) =>
             ApiError::transaction_wait_timeout(msg_id.to_string(), *send_time, *timeout),
+        Some(SdkError::ClockOutOfSync{delta_ms, threshold_ms, expiration_timeout}) =>
+            ApiError::clock_out_of_sync(*delta_ms, *threshold_ms, *expiration_timeout),
         _ => default_err(err.to_string())
     }
 }

ton_sdk/src/contract.rs

@@ -612,7 +612,7 @@ impl Contract {
                                         msg_id: message_id.clone(),
                                         send_time: now,
                                         block_id: block["id"].as_str().unwrap_or("").to_owned(),
-                                        timeout: block_timeout
+                                        timeout: 5000
                                     }.into(),
                                 _ => err
                             })?;

ton_sdk/src/error.rs

@@ -145,4 +145,14 @@ pub enum SdkError {
         send_time: u32,
         timeout: u32
     },
+
+    #[fail(display = "Invalid server response: {}", 0)]
+    InvalidServerResponse(String),
+
+    #[fail(display = "Clock out of sync: {}", delta_ms)]
+    ClockOutOfSync {
+        delta_ms: i64,
+        threshold_ms: i64,
+        expiration_timeout: u32
+    },
 }

ton_sdk/src/node_client.rs

@@ -21,6 +21,7 @@ use reqwest::{ClientBuilder};
 use reqwest::StatusCode;
 use reqwest::redirect::Policy;
 use reqwest::header::LOCATION;
+use chrono::prelude::Utc;
 use ton_types::{error, Result};
 
 #[derive(Serialize, Deserialize)]
@@ -43,9 +44,31 @@ pub struct MutationRequest {
     pub body: String
 }
 
+#[allow(dead_code)]
 pub struct NodeClient {
     client: Option<GqlClient>,
-    timeouts: TimeoutsConfig
+    timeouts: TimeoutsConfig,
+    server_info: Option<ServerInfo>,
+}
+
+struct ServerInfo {
+    pub version: u64,
+    pub supports_time: bool
+}
+
+impl ServerInfo {
+    pub fn from_version(version: &str) -> Result<Self> {
+        let mut vec: Vec<&str> = version.split(".").collect();
+        vec.resize(3, "0");
+        let version = u64::from_str_radix(vec[0], 10)? * 1000000
+            + u64::from_str_radix(vec[1], 10)? * 1000
+            + u64::from_str_radix(vec[2], 10)?;
+
+        Ok(ServerInfo {
+            version,
+            supports_time: version >= 26003,
+        })
+    }
 }
 
 impl NodeClient {
@@ -97,25 +120,69 @@ impl NodeClient {
 
         (queries_url, subscriptions_url)
     }
-    
+
+    async fn query_server_info(client: &GqlClient) -> Result<ServerInfo> {
+        let response = client.query("%7Binfo%7Bversion%7D%7D".to_owned()).await?;
+        let version = response["data"]["info"]["version"]
+            .as_str()
+            .ok_or(SdkError::InvalidServerResponse(
+                format!("No version in response: {}", response)))?;
+
+        ServerInfo::from_version(version)
+    }
+
+    async fn get_time_delta(client: &GqlClient) -> Result<i64>{
+        let start = Utc::now().timestamp_millis();
+        let response = client.query("%7Binfo%7Btime%7D%7D".to_owned()).await?;
+        let end = Utc::now().timestamp_millis();
+        let server_time = response["data"]["info"]["time"]
+            .as_i64()
+            .ok_or(SdkError::InvalidServerResponse(
+                format!("No time in response: {}", response)))?;
+
+        Ok(server_time - (start + (end - start) / 2))
+    }
+
+    async fn check_time_delta(client: &GqlClient, timeouts: &TimeoutsConfig) -> Result<()> {
+        let delta = Self::get_time_delta(client).await?;
+        if delta.abs() >= timeouts.out_of_sync_threshold || delta.abs() >= timeouts.message_expiration_timeout as i64 {
+            Err(SdkError::ClockOutOfSync {
+                delta_ms: delta,
+                threshold_ms: timeouts.out_of_sync_threshold,
+                expiration_timeout: timeouts.message_expiration_timeout
+            }.into())
+        } else {
+            Ok(())
+        }
+    }
+
     // Globally initializes client with server address
     pub async fn new(config: NodeClientConfig) -> Result<NodeClient> {
-        let client = if let Some(base_url) = config.base_url {
+        let timeouts = config.timeouts.unwrap_or_default();
+        let (client, server_info) = if let Some(base_url) = config.base_url {
             let (mut queries_server, mut subscriptions_server) = Self::expand_address(base_url);
             if let Some(redirected) = Self::check_redirect(&queries_server).await? {
                 queries_server = redirected.clone();
                 subscriptions_server = redirected
                     .replace("https://", "wss://")
                     .replace("http://", "ws://");
             }
-            Some(GqlClient::new(&queries_server, &subscriptions_server)?)
+
+            let client = GqlClient::new(&queries_server, &subscriptions_server)?;
+            let server_info = Self::query_server_info(&client).await?;
+            if server_info.supports_time {
+                Self::check_time_delta(&client, &timeouts).await?;
+            }
+
+            (Some(client), Some(server_info))
         } else {
-            None
+            (None, None)
         };
 
         Ok(NodeClient {
             client,
-            timeouts: config.timeouts.unwrap_or_default()
+            timeouts: timeouts,
+            server_info
         })
     }
 
@@ -141,9 +208,9 @@ impl NodeClient {
                             let record_value = &value["payload"]["data"][&closure_table];
 
                             if record_value.is_null() {
-                                Err(error!(SdkError::InvalidData {
-                                    msg: format!("Invalid subscription answer: {}", value)
-                                }).into())
+                                Err(error!(SdkError::InvalidServerResponse(
+                                    format!("Invalid subscription answer: {}", value)
+                                )).into())
                             } else {
                                 Ok(record_value.clone())
                             }
@@ -187,7 +254,7 @@ impl NodeClient {
         // try to extract the record value from the answer
         let records_array = &result["data"][&table];
         if records_array.is_null() {
-            Err(SdkError::InvalidData { msg: format!("Invalid query answer: {}", result) }.into())
+            Err(SdkError::InvalidServerResponse(format!("Invalid query answer: {}", result)).into())
         } else {
             Ok(records_array.clone())
         }

ton_sdk/src/types.rs

@@ -27,6 +27,7 @@ pub const DEFAULT_EXPIRATION_TIMEOUT: u32 = 10000;
 pub const DEFAULT_PROCESSING_TIMEOUT: u32 = 40000;
 pub const DEFAULT_TIMEOUT_GROW_FACTOR: f32 = 1.5;
 pub const DEFAULT_WAIT_TIMEOUT: u32 = 40000;
+pub const DEFAULT_OUT_OF_SYNC_THRESHOLD: i64 = 15000;
 
 
 #[derive(Debug, Deserialize, Serialize)]
@@ -38,6 +39,7 @@ pub struct TimeoutsConfig {
     pub message_processing_timeout: u32,
     pub message_processing_timeout_grow_factor: f32,
     pub wait_for_timeout: u32,
+    pub out_of_sync_threshold: i64
 }
 
 impl Default for TimeoutsConfig {
@@ -49,6 +51,7 @@ impl Default for TimeoutsConfig {
             message_processing_timeout: DEFAULT_PROCESSING_TIMEOUT,
             message_processing_timeout_grow_factor: DEFAULT_TIMEOUT_GROW_FACTOR,
             wait_for_timeout: DEFAULT_WAIT_TIMEOUT,
+            out_of_sync_threshold: DEFAULT_OUT_OF_SYNC_THRESHOLD
         }
     }
 }