Merge pull request #53 from skalenetwork/filter-out-of-sync-nodes

Filter out of sync nodes

proxy/config.py

@@ -53,3 +53,5 @@
 
 NGINX_CONTAINER_NAME = 'proxy_nginx'
 CONTAINER_RUNNING_STATUS = 'running'
+
+ALLOWED_TIMESTAMP_DIFF = 300

proxy/endpoints.py

@@ -26,10 +26,11 @@
 from Crypto.Hash import keccak
 
 from proxy.node_info import get_node_info
-from proxy.helper import read_json
+from proxy.helper import read_json, make_rpc_call
 from proxy.config import ENDPOINT, SM_ABI_FILEPATH
 from proxy.str_formatters import arguments_list_string
 from proxy.schain_options import parse_schain_options
+from proxy.config import ALLOWED_TIMESTAMP_DIFF
 
 logger = logging.getLogger(__name__)
 
@@ -53,11 +54,22 @@ def __init__(self, schain_name: str, nodes: list):
         self._format_nodes(nodes)
 
     def _format_nodes(self, nodes):
+        for node in nodes:
+            http_endpoint = node['http_endpoint_domain']
+            node['block_ts'] = get_block_ts(http_endpoint)
+
+        max_ts = max(node['block_ts'] for node in nodes)
+        logger.info(f'max_ts: {max_ts}')
+
         for node in nodes:
             http_endpoint = node['http_endpoint_domain']
             if not url_ok(http_endpoint):
                 logger.warning(f'{http_endpoint} is not accesible, removing from the list')
                 continue
+            if not is_node_out_of_sync(node['block_ts'], max_ts):
+                logger.warning(f'{http_endpoint} ts: {node["block_ts"]}, max ts for chain: \
+{max_ts}, allowed timestamp diff: {ALLOWED_TIMESTAMP_DIFF}')
+                continue
             self.http_endpoints.append(http_endpoint.removeprefix(URL_PREFIXES['http']))
             self.ws_endpoints.append(node['ws_endpoint_domain'].removeprefix(URL_PREFIXES['ws']))
             self.fs_endpoints.append(node['domain'])
@@ -80,6 +92,19 @@ def url_ok(url) -> bool:
         return False
 
 
+def is_node_out_of_sync(ts: int, compare_ts: int) -> bool:
+    return abs(compare_ts - ts) < ALLOWED_TIMESTAMP_DIFF
+
+
+def get_block_ts(http_endpoint: str) -> int:
+    res = make_rpc_call(http_endpoint, 'eth_getBlockByNumber', ['latest', False])
+    if res and res.json():
+        res_data = res.json()
+        latest_schain_timestamp_hex = res_data['result']['timestamp']
+        return int(latest_schain_timestamp_hex, 16)
+    return -1
+
+
 def schain_name_to_id(name: str) -> str:
     keccak_hash = keccak.new(data=name.encode("utf8"), digest_bits=256)
     return '0x' + keccak_hash.hexdigest()

proxy/helper.py

@@ -21,6 +21,7 @@
 import json
 import socket
 import logging
+import requests
 from logging import Formatter, StreamHandler
 
 from jinja2 import Environment
@@ -66,3 +67,22 @@ def init_default_logger():
     handlers.append(stream_handler)
 
     logging.basicConfig(level=logging.DEBUG, handlers=handlers)
+
+
+def post_request(url, json, cookies=None):
+    try:
+        return requests.post(
+            url,
+            json=json,
+            cookies=cookies
+        )
+    except requests.exceptions.RequestException:
+        return None
+
+
+def make_rpc_call(http_endpoint, method, params=None):
+    params = params or []
+    return post_request(
+        http_endpoint,
+        json={"jsonrpc": "2.0", "method": method, "params": params, "id": 1}
+    )