Add retries

README.md

@@ -45,6 +45,20 @@ out = account.function_call(contract_id, "counter_set", args)
 print(out)
 ```
 
+## Logging control
+
+```python
+import logging
+
+# To hide NEAR warnings and info messages
+logging.getLogger("near_api").setLevel(logging.ERROR)
+
+# To log all information about RPC requests
+logging.getLogger("urllib3.connectionpool").setLevel(logging.DEBUG).addHandler(handler)
+
+# To log only information about RPC retries
+logging.getLogger("urllib3.connectionpool").setLevel(logging.WARNING).addHandler(handler)
+```
 
 # Contribution
 

near_api/account.py

@@ -1,11 +1,15 @@
 import itertools
 import json
+import logging
+import time
 from typing import Optional
 
 import base58
 
 import near_api
-from near_api import transactions
+from near_api import transactions, providers
+
+log = logging.getLogger(__name__)
 
 # Amount of gas attached by default 1e14.
 DEFAULT_ATTACHED_GAS = 100_000_000_000_000
@@ -25,16 +29,40 @@ def __init__(
             self,
             provider: 'near_api.providers.JsonProvider',
             signer: 'near_api.signer.Signer',
-            account_id: Optional[str] = None
+            account_id: Optional[str] = None,
+            tx_nonce_retry_number: int = 12,
+            tx_nonce_retry_backoff_factor: float = 1.5,
     ):
         self._provider = provider
         self._signer = signer
         self._account_id = account_id or self._signer.account_id
         self._account: dict = provider.get_account(self._account_id)
         self._access_key: dict = provider.get_access_key(self._account_id, self._signer.key_pair.encoded_public_key())
-        # print(account_id, self._account, self._access_key)
+
+        self._tx_nonce_retry_number = tx_nonce_retry_number
+        self._tx_nonce_retry_backoff_factor = tx_nonce_retry_backoff_factor
 
     def _sign_and_submit_tx(self, receiver_id: str, actions: list['transactions.Action']) -> dict:
+        attempt = 0
+        while True:
+            try:
+                return self._sign_and_submit_tx_once(receiver_id, actions)
+            except providers.JsonProviderError as e:
+                if attempt >= self._tx_nonce_retry_number:
+                    raise
+                attempt += 1
+
+                if e.is_invalid_nonce_tx_error():
+                    log.warning("Retrying transaction with new nonce: %s", e)
+                    self.fetch_state()
+                elif e.is_expired_tx_error():
+                    log.warning("Retrying transaction due to expired block hash: %s", e)
+                else:
+                    raise
+
+                time.sleep(self._tx_nonce_retry_backoff_factor ** attempt)
+
+    def _sign_and_submit_tx_once(self, receiver_id: str, actions: list['transactions.Action']) -> dict:
         self._access_key['nonce'] += 1
         block_hash = self._provider.get_status()['sync_info']['latest_block_hash']
         block_hash = base58.b58decode(block_hash.encode('utf8'))
@@ -43,7 +71,7 @@ def _sign_and_submit_tx(self, receiver_id: str, actions: list['transactions.Acti
         result: dict = self._provider.send_tx_and_wait(serialized_tx, 10)
         for outcome in itertools.chain([result['transaction_outcome']], result['receipts_outcome']):
             for log in outcome['outcome']['logs']:
-                print("Log:", log)
+                log.info("Log %s: %s", receiver_id, log)
         if 'Failure' in result['status']:
             raise TransactionError(result['status']['Failure'])
         return result

near_api/providers.py

@@ -1,8 +1,13 @@
 import base64
 import json
-from typing import Union, Tuple, Any
+import logging
+import time
+from typing import Union, Tuple, Any, Optional
 
 import requests
+import urllib3.util
+
+log = logging.getLogger(__name__)
 
 TimeoutType = Union[float, Tuple[float, float]]
 """ The type used as "timeout" argument when sending requests. Quantities are in seconds.
@@ -18,28 +23,104 @@ class FinalityTypes:
 
 
 class JsonProviderError(Exception):
-    pass
+    def get_type(self) -> Optional[str]:
+        try:
+            return self.args[0]["name"]
+        except (IndexError, KeyError):
+            return None
+
+    def get_cause(self) -> Optional[str]:
+        try:
+            return self.args[0]["cause"]["name"]
+        except (IndexError, KeyError):
+            return None
+
+    def is_invalid_nonce_tx_error(self) -> bool:
+        try:
+            return (
+                self.get_type() == "HANDLER_ERROR"
+                and self.get_cause() == "INVALID_TRANSACTION"
+                and "InvalidNonce" in self.args[0]["data"]["TxExecutionError"]["InvalidTxError"]
+            )
+        except (IndexError, KeyError):
+            return False
+
+    def is_expired_tx_error(self) -> bool:
+        try:
+            return (
+                self.get_type() == "HANDLER_ERROR"
+                and self.get_cause() == "INVALID_TRANSACTION"
+                and self.args[0]["data"]["TxExecutionError"]["InvalidTxError"] == "Expired"
+            )
+        except (IndexError, KeyError):
+            return False
 
 
 class JsonProvider(object):
-    def __init__(self, rpc_addr, proxies=None):
+    def __init__(
+            self,
+            rpc_addr,
+            proxies=None,
+            tx_timeout_retry_number: int = 12,
+            tx_timeout_retry_backoff_factor: float = 1.5,
+            http_retry_number: int = 10,
+            http_retry_backoff_factor: float = 1.5,
+            session: Optional[requests.Session] = None,
+    ) -> None:
         if isinstance(rpc_addr, tuple):
             self._rpc_addr = "http://%s:%s" % rpc_addr
         else:
             self._rpc_addr = rpc_addr
         self.proxies = proxies
 
+        if session is None:
+            # Loosely based on https://findwork.dev/blog/advanced-usage-python-requests-timeouts-retries-hooks/
+            adapter = requests.adapters.HTTPAdapter(
+                max_retries=urllib3.util.Retry(
+                    total=http_retry_number,
+                    backoff_factor=http_retry_backoff_factor,
+                    status_forcelist=[502, 503],
+                    allowed_methods=["GET", "POST"],
+                ),
+            )
+
+            session = requests.Session()
+            session.mount("https://", adapter)
+            session.mount("http://", adapter)
+
+        self._session = session
+
+        self._tx_timeout_retry_number = tx_timeout_retry_number
+        self._tx_timeout_retry_backoff_factor = tx_timeout_retry_backoff_factor
+
     def rpc_addr(self) -> str:
         return self._rpc_addr
 
     def json_rpc(self, method: str, params: Union[dict, list, str], timeout: 'TimeoutType' = 2.0) -> dict:
+        attempt = 0
+        while True:
+            try:
+                return self._json_rpc_once(method, params, timeout)
+            except JsonProviderError as e:
+                if attempt >= self._tx_timeout_retry_number:
+                    raise
+                attempt += 1
+
+                if e.get_type() == "HANDLER_ERROR" and e.get_cause() == "TIMEOUT_ERROR":
+                    log.warning("Retrying request to %s as it has timed out: %s", method, e)
+                else:
+                    raise
+
+                time.sleep(self._tx_timeout_retry_backoff_factor ** attempt)
+
+    def _json_rpc_once(self, method: str, params: Union[dict, list, str], timeout: 'TimeoutType' = 2.0) -> dict:
         j = {
             'method': method,
             'params': params,
             'id': "dontcare",
             'jsonrpc': "2.0"
         }
-        r = requests.post(self.rpc_addr(), json=j, timeout=timeout, proxies=self.proxies)
+        r = self._session.post(self.rpc_addr(), json=j, timeout=timeout, proxies=self.proxies)
         r.raise_for_status()
         content = json.loads(r.content)
         if "error" in content:
@@ -56,7 +137,7 @@ def send_tx_and_wait(self, signed_tx: bytes, timeout: 'TimeoutType') -> dict:
                              timeout=timeout)
 
     def get_status(self, timeout: 'TimeoutType' = 2.0) -> dict:
-        r = requests.get("%s/status" % self.rpc_addr(), timeout=timeout)
+        r = self._session.get("%s/status" % self.rpc_addr(), timeout=timeout)
         r.raise_for_status()
         return json.loads(r.content)