Add blockhash cache and allow user-supplied blockhash (#102)

* allow user to pass recent blockhash themselves

* add blockhash cache

* add pytest-cov dev dependency

* fix dupe stubbed account

* linting

* Bump version: 0.12.1 → 0.13.0

Co-authored-by: kevinheavey <kevinheavey123@gmail.com>

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.12.1
+current_version = 0.13.0
 commit = True
 tag = True
 

Pipfile

@@ -29,6 +29,7 @@ bump2version = "*"
 types-requests = "*"
 notebook = "*"
 pytest-asyncio = "*"
+pytest-cov = "*"
 
 [packages]
 pynacl = "*"
@@ -37,6 +38,8 @@ requests = "*"
 construct = "*"
 typing-extensions = "*"
 httpx = "*"
+cachetools = "*"
+types-cachetools = "*"
 
 [requires]
 python_version = "3.7"

docs/conf.py

@@ -21,7 +21,7 @@
 project = "solana.py"
 copyright = "2020, Michael Huang"
 # *IMPORTANT*: Don't manually change the version here. Use the 'bumpversion' utility.4
-version = "0.12.1"
+version = "0.13.0"
 author = "Michael Huang"
 
 

setup.py

@@ -26,7 +26,7 @@
 setup(
     name="solana",
     # *IMPORTANT*: Don't manually change the version here. Use the 'bumpversion' utility.
-    version="0.12.1",
+    version="0.13.0",
     author="Michael Huang",
     author_mail="michaelhly@gmail.com",
     description="""Solana.py""",

solana/blockhash.py

@@ -5,6 +5,53 @@
 'EETubP5AKHgjPAhzPAFcb8BAY1hMH639CWCFTqi3hq1k'
 """
 from typing import NewType
+from cachetools import TTLCache
 
 Blockhash = NewType("Blockhash", str)
 """Type for blockhash."""
+
+
+class BlockhashCache:
+    """A recent blockhash cache that expires after a given number of seconds."""
+
+    def __init__(self, ttl: int = 60) -> None:
+        """Instantiate the cache (you only need to do this once).
+
+        Args:
+        ----
+            ttl (int): Seconds until cached blockhash expires.
+
+        """
+        maxsize = 300
+        self.unused_blockhashes: TTLCache = TTLCache(maxsize=maxsize, ttl=ttl)
+        self.used_blockhashes: TTLCache = TTLCache(maxsize=maxsize, ttl=ttl)
+
+    def set(self, blockhash: Blockhash, slot: int) -> None:
+        """Update the cache.
+
+        Args:
+        ----
+            blockhash (Blockhash): new Blockhash value.
+            slot (int): the slot which the blockhash came from
+
+        """
+        if slot in self.used_blockhashes or slot in self.unused_blockhashes:
+            return
+        self.unused_blockhashes[slot] = blockhash
+
+    def get(self) -> Blockhash:
+        """Get the cached Blockhash. Raises KeyError if cache has expired.
+
+        Returns
+        -------
+            Blockhash: cached Blockhash.
+
+        """
+        try:
+            slot, blockhash = self.unused_blockhashes.popitem()
+            self.used_blockhashes[slot] = blockhash
+        except KeyError:
+            with self.used_blockhashes.timer:  # type: ignore
+                blockhash = self.used_blockhashes[min(self.used_blockhashes)]
+                # raises ValueError if used_blockhashes is empty
+        return blockhash

solana/rpc/api.py

@@ -6,7 +6,7 @@
 from warnings import warn
 
 from solana.account import Account
-from solana.blockhash import Blockhash
+from solana.blockhash import Blockhash, BlockhashCache
 from solana.publickey import PublicKey
 from solana.rpc import types
 from solana.transaction import Transaction
@@ -34,9 +34,35 @@ def MemcmpOpt(*args, **kwargs) -> types.MemcmpOpts:  # pylint: disable=invalid-n
 class Client(_ClientCore):  # pylint: disable=too-many-public-methods
     """Client class."""
 
-    def __init__(self, endpoint: Optional[str] = None, commitment: Optional[Commitment] = None):
-        """Init API client."""
-        super().__init__(commitment)
+    def __init__(
+        self,
+        endpoint: Optional[str] = None,
+        commitment: Optional[Commitment] = None,
+        blockhash_cache: Union[BlockhashCache, bool] = False,
+    ):
+        """Init API client.
+
+        :param endpoint: URL of the RPC endpoint.
+        :param commitment: Default bank state to query. It can be either "finalized", "confirmed" or "processed".
+        :param blockhash_cache: (Experimental) If True, keep a cache of recent blockhashes to make
+            `send_transaction` calls faster.
+            You can also pass your own BlockhashCache object to customize its parameters.
+
+            The cache works as follows:
+
+            1. Retrieve the oldest unused cached blockhash that is younger than `ttl` seconds,
+                where `ttl` is defined in the BlockhashCache
+                (we prefer unused blockhashes because reusing blockhashes can cause errors in some edge cases,
+                and we prefer slightly older blockhashes because they're more likely to be accepted by every validator).
+            2. If there are no unused blockhashes in the cache, take the oldest used
+                blockhash that is younger than `ttl` seconds.
+            3. Fetch a new recent blockhash *after* sending the transaction. This is to keep the cache up-to-date.
+
+            If you want something tailored to your use case, run your own loop that fetches the recent blockhash,
+            and pass that value in your `.send_transaction` calls.
+
+        """
+        super().__init__(commitment, blockhash_cache)
         self._provider = http.HTTPProvider(endpoint)
 
     def is_connected(self) -> bool:
@@ -930,13 +956,19 @@ def send_raw_transaction(self, txn: Union[bytes, str], opts: types.TxOpts = type
         return self.__post_send_with_confirm(*post_send_args)
 
     def send_transaction(
-        self, txn: Transaction, *signers: Account, opts: types.TxOpts = types.TxOpts()
+        self,
+        txn: Transaction,
+        *signers: Account,
+        opts: types.TxOpts = types.TxOpts(),
+        recent_blockhash: Optional[Blockhash] = None,
     ) -> types.RPCResponse:
         """Send a transaction.
 
         :param txn: Transaction object.
         :param signers: Signers to sign the transaction.
         :param opts: (optional) Transaction options.
+        :param recent_blockhash: (optional) Pass a valid recent blockhash here if you want to
+            skip fetching the recent blockhash or relying on the cache.
 
         >>> from solana.account import Account
         >>> from solana.system_program import TransferParams, transfer
@@ -950,17 +982,24 @@ def send_transaction(
          'result': '236zSA5w4NaVuLXXHK1mqiBuBxkNBu84X6cfLBh1v6zjPrLfyECz4zdedofBaZFhs4gdwzSmij9VkaSo2tR5LTgG',
          'id': 12}
         """
-        try:
-            # TODO: Cache recent blockhash
-            blockhash_resp = self.get_recent_blockhash()
-            if not blockhash_resp["result"]:
-                raise RuntimeError("failed to get recent blockhash")
-            txn.recent_blockhash = Blockhash(blockhash_resp["result"]["value"]["blockhash"])
-        except Exception as err:
-            raise RuntimeError("failed to get recent blockhash") from err
+        if recent_blockhash is None:
+            if self.blockhash_cache:
+                try:
+                    recent_blockhash = self.blockhash_cache.get()
+                except ValueError:
+                    blockhash_resp = self.get_recent_blockhash()
+                    recent_blockhash = self._process_blockhash_resp(blockhash_resp)
+            else:
+                blockhash_resp = self.get_recent_blockhash()
+                recent_blockhash = self.parse_recent_blockhash(blockhash_resp)
+        txn.recent_blockhash = recent_blockhash
 
         txn.sign(*signers)
-        return self.send_raw_transaction(txn.serialize(), opts=opts)
+        txn_resp = self.send_raw_transaction(txn.serialize(), opts=opts)
+        if self.blockhash_cache:
+            blockhash_resp = self.get_recent_blockhash()
+            self._process_blockhash_resp(blockhash_resp)
+        return txn_resp
 
     def simulate_transaction(
         self, txn: Union[bytes, str, Transaction], sig_verify: bool = False, commitment: Optional[Commitment] = None
@@ -1037,7 +1076,7 @@ def __confirm_transaction(self, tx_sig: str, commitment: Optional[Commitment] =
             elapsed_time += sleep_time
 
         if not resp["result"]:
-            raise Exception("Unable to confirm transaction %s" % tx_sig)
+            raise Exception(f"Unable to confirm transaction {tx_sig}")
         err = resp.get("error") or resp["result"].get("meta").get("err")
         if err:
             self._provider.logger.error("Transaction error: %s", err)

solana/rpc/async_api.py

@@ -3,7 +3,7 @@
 from typing import List, Optional, Union
 
 from solana.account import Account
-from solana.blockhash import Blockhash
+from solana.blockhash import Blockhash, BlockhashCache
 from solana.publickey import PublicKey
 from solana.rpc import types
 from solana.transaction import Transaction
@@ -16,9 +16,35 @@
 class AsyncClient(_ClientCore):  # pylint: disable=too-many-public-methods
     """Async client class."""
 
-    def __init__(self, endpoint: Optional[str] = None, commitment: Optional[Commitment] = None) -> None:
-        """Init API client."""
-        super().__init__(commitment)
+    def __init__(
+        self,
+        endpoint: Optional[str] = None,
+        commitment: Optional[Commitment] = None,
+        blockhash_cache: Union[BlockhashCache, bool] = False,
+    ) -> None:
+        """Init API client.
+
+        :param endpoint: URL of the RPC endpoint.
+        :param commitment: Default bank state to query. It can be either "finalized", "confirmed" or "processed".
+        :param blockhash_cache: (Experimental) If True, keep a cache of recent blockhashes to make
+            `send_transaction` calls faster.
+            You can also pass your own BlockhashCache object to customize its parameters.
+
+            The cache works as follows:
+
+            1. Retrieve the oldest unused cached blockhash that is younger than `ttl` seconds,
+                where `ttl` is defined in the BlockhashCache
+                (we prefer unused blockhashes because reusing blockhashes can cause errors in some edge cases,
+                and we prefer slightly older blockhashes because they're more likely to be accepted by every validator).
+            2. If there are no unused blockhashes in the cache, take the oldest used
+                blockhash that is younger than `ttl` seconds.
+            3. Fetch a new recent blockhash *after* sending the transaction. This is to keep the cache up-to-date.
+
+            If you want something tailored to your use case, run your own loop that fetches the recent blockhash,
+            and pass that value in your `.send_transaction` calls.
+
+        """
+        super().__init__(commitment, blockhash_cache)
         self._provider = async_http.AsyncHTTPProvider(endpoint)
 
     async def __aenter__(self) -> "AsyncClient":
@@ -929,13 +955,19 @@ async def send_raw_transaction(
         return await self.__post_send_with_confirm(*post_send_args)
 
     async def send_transaction(
-        self, txn: Transaction, *signers: Account, opts: types.TxOpts = types.TxOpts()
+        self,
+        txn: Transaction,
+        *signers: Account,
+        opts: types.TxOpts = types.TxOpts(),
+        recent_blockhash: Optional[Blockhash] = None,
     ) -> types.RPCResponse:
         """Send a transaction.
 
         :param txn: Transaction object.
         :param signers: Signers to sign the transaction.
         :param opts: (optional) Transaction options.
+        :param recent_blockhash: (optional) Pass a valid recent blockhash here if you want to
+            skip fetching the recent blockhash or relying on the cache.
 
         >>> from solana.account import Account
         >>> from solana.system_program import TransferParams, transfer
@@ -949,17 +981,24 @@ async def send_transaction(
          'result': '236zSA5w4NaVuLXXHK1mqiBuBxkNBu84X6cfLBh1v6zjPrLfyECz4zdedofBaZFhs4gdwzSmij9VkaSo2tR5LTgG',
          'id': 12}
         """
-        try:
-            # TODO: Cache recent blockhash
-            blockhash_resp = await self.get_recent_blockhash()
-            if not blockhash_resp["result"]:
-                raise RuntimeError("failed to get recent blockhash")
-            txn.recent_blockhash = Blockhash(blockhash_resp["result"]["value"]["blockhash"])
-        except Exception as err:
-            raise RuntimeError("failed to get recent blockhash") from err
+        if recent_blockhash is None:
+            if self.blockhash_cache:
+                try:
+                    recent_blockhash = self.blockhash_cache.get()
+                except ValueError:
+                    blockhash_resp = await self.get_recent_blockhash()
+                    recent_blockhash = self._process_blockhash_resp(blockhash_resp)
+            else:
+                blockhash_resp = await self.get_recent_blockhash()
+                recent_blockhash = self.parse_recent_blockhash(blockhash_resp)
+        txn.recent_blockhash = recent_blockhash
 
         txn.sign(*signers)
-        return await self.send_raw_transaction(txn.serialize(), opts=opts)
+        txn_resp = await self.send_raw_transaction(txn.serialize(), opts=opts)
+        if self.blockhash_cache:
+            blockhash_resp = await self.get_recent_blockhash()
+            self._process_blockhash_resp(blockhash_resp)
+        return txn_resp
 
     async def simulate_transaction(
         self, txn: Union[bytes, str, Transaction], sig_verify: bool = False, commitment: Optional[Commitment] = None
@@ -1037,7 +1076,7 @@ async def __confirm_transaction(self, tx_sig: str, commitment: Optional[Commitme
 
         if not resp["result"]:
             print(f"resp: {resp}")
-            raise Exception("Unable to confirm transaction %s" % tx_sig)
+            raise Exception(f"Unable to confirm transaction {tx_sig}")
         err = resp.get("error") or resp["result"].get("meta").get("err")
         if err:
             self._provider.logger.error("Transaction error: %s", err)

solana/rpc/core.py

@@ -2,12 +2,17 @@
 """Helper code for api.py and async_api.py."""
 from base64 import b64encode
 from typing import Any, Dict, List, Optional, Tuple, Union
+
+try:
+    from typing import Literal  # type: ignore
+except ImportError:
+    from typing_extensions import Literal
 from warnings import warn
 
 from base58 import b58decode, b58encode
 
 from solana.account import Account
-from solana.blockhash import Blockhash
+from solana.blockhash import Blockhash, BlockhashCache
 from solana.publickey import PublicKey
 from solana.rpc import types
 from solana.transaction import Transaction
@@ -33,8 +38,11 @@ class _ClientCore:  # pylint: disable=too-few-public-methods
     _get_version = types.RPCMethod("getVersion")
     _validator_exit = types.RPCMethod("validatorExit")
 
-    def __init__(self, commitment: Optional[Commitment] = None):
+    def __init__(self, commitment: Optional[Commitment] = None, blockhash_cache: Union[BlockhashCache, bool] = False):
         self._commitment = commitment or Finalized
+        self.blockhash_cache: Union[BlockhashCache, Literal[False]] = (
+            BlockhashCache() if blockhash_cache is True else blockhash_cache
+        )
 
     def _get_balance_args(
         self, pubkey: Union[PublicKey, str], commitment: Optional[Commitment]
@@ -332,3 +340,17 @@ def _post_send(
         if not resp.get("result"):
             raise Exception("Failed to send transaction")
         return resp
+
+    @staticmethod
+    def parse_recent_blockhash(blockhash_resp: types.RPCResponse) -> Blockhash:
+        """Extract blockhash from JSON RPC result."""
+        if not blockhash_resp["result"]:
+            raise RuntimeError("failed to get recent blockhash")
+        return Blockhash(blockhash_resp["result"]["value"]["blockhash"])
+
+    def _process_blockhash_resp(self, blockhash_resp: types.RPCResponse) -> Blockhash:
+        recent_blockhash = self.parse_recent_blockhash(blockhash_resp)
+        if self.blockhash_cache:
+            slot = blockhash_resp["result"]["context"]["slot"]
+            self.blockhash_cache.set(recent_blockhash, slot)
+        return recent_blockhash