Fetch mempool transactions for full nodes (#12)

* add storage backends for mempool

This allows unconfirmed transactions to be cached in an sqlite, mysql, or postgres database.
Most recent transactions are cached in memory, so that memory usage remains low.

* clean up cache class

* add SQL transaction caching module

* fix (sqlite) cache database connectivity

* store temporary and finished mempool data in the database

Author: ZenulAbidin

Diff for: .vscode/settings.json

@@ -4,5 +4,8 @@
         "editor.formatOnSave": true
       },
       "editor.wordWrap": "off",
-      "editor.rulers": [80, 120]
+      "editor.rulers": [
+        80,
+        120
+      ]
 }

Diff for: requirements.txt

@@ -17,3 +17,7 @@ aiohttp<4.0
 # Web3 6.14.0 fixes a bug that would make ZPyWallet unit tests fail to setup the testing environment.
 web3>=6.14.0
 websocket-client # automatically pulled by web3, but we also use this package directly.
+
+# For mempool transaction caching inside full nodes.
+psycopg2
+mysql-connector-python

Diff for: zpywallet/address/btc/fullnode.py

@@ -109,7 +109,6 @@ def __init__(self, addresses, transactions=None, **kwargs):
         self.rpc_url = kwargs.get("url")
         self.rpc_user = kwargs.get("user")
         self.rpc_password = kwargs.get("password")
-        self.max_tx_at_once = kwargs.get("max_tx_at_once") or 1000
         self.fast_mode = kwargs.get("fast_mode") or True
         self.min_height = kwargs.get("min_height") or 0
         self.transactions = []

Diff for: zpywallet/address/btctest/fullnode.py

@@ -110,7 +110,6 @@ def __init__(self, addresses, transactions=None, **kwargs):
         self.rpc_url = kwargs.get("url")
         self.rpc_user = kwargs.get("user")
         self.rpc_password = kwargs.get("password")
-        self.max_tx_at_once = kwargs.get("max_tx_at_once") or 1000
         self.min_height = kwargs.get("min_height") or 0
         self.fast_mode = kwargs.get("fast_mode") or True
         self.transactions = []

Diff for: zpywallet/address/dash/fullnode.py

@@ -102,7 +102,6 @@ def __init__(self, addresses, transactions=None, **kwargs):
         self.rpc_url = kwargs.get("url")
         self.rpc_user = kwargs.get("user")
         self.rpc_password = kwargs.get("password")
-        self.max_tx_at_once = kwargs.get("max_tx_at_once") or 1000
         self.min_height = kwargs.get("min_height") or 0
         self.fast_mode = kwargs.get("fast_mode") or True
         self.transactions = []

Diff for: zpywallet/address/dashtest/fullnode.py

@@ -102,7 +102,6 @@ def __init__(self, addresses, transactions=None, **kwargs):
         self.rpc_url = kwargs.get("url")
         self.rpc_user = kwargs.get("user")
         self.rpc_password = kwargs.get("password")
-        self.max_tx_at_once = kwargs.get("max_tx_at_once") or 1000
         self.min_height = kwargs.get("min_height") or 0
         self.fast_mode = kwargs.get("fast_mode") or True
         self.transactions = []

Diff for: zpywallet/address/doge/fullnode.py

@@ -102,7 +102,6 @@ def __init__(self, addresses, transactions=None, **kwargs):
         self.rpc_url = kwargs.get("url")
         self.rpc_user = kwargs.get("user")
         self.rpc_password = kwargs.get("password")
-        self.max_tx_at_once = kwargs.get("max_tx_at_once") or 1000
         self.min_height = kwargs.get("min_height") or 0
         self.fast_mode = kwargs.get("fast_mode") or True
         self.transactions = []

Diff for: zpywallet/address/dogetest/fullnode.py

@@ -102,7 +102,6 @@ def __init__(self, addresses, transactions=None, **kwargs):
         self.rpc_url = kwargs.get("url")
         self.rpc_user = kwargs.get("user")
         self.rpc_password = kwargs.get("password")
-        self.max_tx_at_once = kwargs.get("max_tx_at_once") or 1000
         self.min_height = kwargs.get("min_height") or 0
         self.fast_mode = kwargs.get("fast_mode") or True
         self.transactions = []

Diff for: zpywallet/address/ltc/fullnode.py

@@ -102,7 +102,6 @@ def __init__(self, addresses, transactions=None, **kwargs):
         self.rpc_url = kwargs.get("url")
         self.rpc_user = kwargs.get("user")
         self.rpc_password = kwargs.get("password")
-        self.max_tx_at_once = kwargs.get("max_tx_at_once") or 1000
         self.min_height = kwargs.get("min_height") or 0
         self.fast_mode = kwargs.get("fast_mode") or True
         self.transactions = []

Diff for: zpywallet/address/ltctest/fullnode.py

@@ -102,7 +102,6 @@ def __init__(self, addresses, transactions=None, **kwargs):
         self.rpc_url = kwargs.get("url")
         self.rpc_user = kwargs.get("user")
         self.rpc_password = kwargs.get("password")
-        self.max_tx_at_once = kwargs.get("max_tx_at_once") or 1000
         self.min_height = kwargs.get("min_height") or 0
         self.fast_mode = kwargs.get("fast_mode") or True
         self.transactions = []

Diff for: zpywallet/destination.py

@@ -27,7 +27,8 @@ def __init__(self, address, amount, network, fee_policy=FeePolicy.NONE):
             address (str): The destination address.
             amount (int): The amount of the destination in the lowest denomination.
             network: The network associated with the destination.
-            fee_policy (FeePolicy, optional): The fee policy associated with the destination. Defaults to FeePolicy.NONE.
+            fee_policy (FeePolicy, optional): The fee policy associated with the destination.
+                Defaults to FeePolicy.NONE.
         """
         self._network = network
         self._address = address
@@ -46,7 +47,8 @@ def amount(self, in_standard_units=True):
         Returns the amount of the destination.
 
         Args:
-            in_standard_units (bool, optional): If True, returns the amount in standard units. If False, returns the amount in the lowest denomination. Defaults to True.
+            in_standard_units (bool, optional): If True, returns the amount in standard units.
+                If False, returns the amount in the lowest denomination. Defaults to True.
         """
         if not in_standard_units:
             if self._network.SUPPORTS_EVM:

Diff for: zpywallet/mempool/__init__.py

@@ -1 +1 @@
-__all__ = ["btc"]
+__all__ = ["btclike", "cache"]

Diff for: zpywallet/mempool/btc.py renamed to zpywallet/mempool/btclike.py

@@ -1,10 +1,14 @@
 import json
+import multiprocessing
 import random
 import requests
 from concurrent.futures import ThreadPoolExecutor
 
-from ..errors import NetworkException
-from ..generated import wallet_pb2
+
+from zpywallet.errors import NetworkException
+from zpywallet.generated import wallet_pb2
+
+from zpywallet.mempool.cache import SQLTransactionStorage, DatabaseError
 
 
 def transform_and_sort_transactions(data):
@@ -17,11 +21,12 @@ def transform_and_sort_transactions(data):
     return sorted_data
 
 
-class BitcoinMempool:
-    """Holds the transactions inside the mempool. There should only be one
-    mempool class for each coin running in the entire application.
+class BTCLikeMempool:
+    """Holds the transactions of Bitcoin-like blockchains inside the mempool.
+    There should only be one mempool class for each coin running in the entire
+    application.
 
-    Note: The performance of this class heavily depends on the network speed and CPU
+    The performance of this class heavily depends on the network speed and CPU
     speed of the node as well as the number of threads available, the size of the RPC
     batch work queue specified in the constructor, and the amount of transactions in
     megabytes you are trying to fetch at once.
@@ -59,8 +64,11 @@ def _clean_tx(self, element):
         # Of course this is an unconfirmed transaction.
         # Hence it is also not a coinbase transaction.
         new_element.confirmed = False
+        new_element.txid = element["txid"]
 
         for vin in element["vin"]:
+            if "txid" not in vin.keys():
+                continue
             txinput = new_element.btclike_transaction.inputs.add()
             txinput.txid = vin["txid"]
             txinput.index = vin["vout"]
@@ -82,23 +90,17 @@ def _clean_tx(self, element):
         new_element.btclike_transaction.fee = element["vsize"]
         return new_element
 
-    def _post_clean_tx(self, new_element):
-
+    def _post_clean_tx(self, new_element, sql_transaction_storage):
         for txinput in new_element.btclike_transaction.inputs:
-            fine_rawtx = self.raw_txos.get(txinput.txid)
+            fine_rawtx = sql_transaction_storage.get_rawtx(txinput.txid)
             if fine_rawtx is None:
                 return None  # maybe processing error.
-            txinput.amount = int(fine_rawtx["vout"][txinput.index]["value"] * 1e8)
-            if "address" in fine_rawtx["vout"][txinput.index]["scriptPubKey"].keys():
-                txinput.address = fine_rawtx["vout"][txinput.index]["scriptPubKey"][
-                    "address"
-                ]
-            elif (
-                "addresses" in fine_rawtx["vout"][txinput.index]["scriptPubKey"].keys()
-            ):
-                txinput.address = fine_rawtx["vout"][txinput.index]["scriptPubKey"][
-                    "addresses"
-                ][0]
+            txinput.amount = fine_rawtx.btclike_transaction.outputs[
+                txinput.index
+            ].amount
+            txinput.address = fine_rawtx.btclike_transaction.outputs[
+                txinput.index
+            ].address
 
         # Now we must calculate the total fee
         total_inputs = sum([a.amount for a in new_element.btclike_transaction.inputs])
@@ -124,6 +126,7 @@ def __init__(self, **kwargs):
         self.future_blocks_min = kwargs.get("future_blocks_min") or 0
         self.future_blocks_max = kwargs.get("future_blocks_max") or 1
         self.full_transactions = kwargs.get("full_transactions") or True
+        self.db_connection_parameters = kwargs.get("db_connection_parameters")
         self.transactions = []
         self.in_mempool = []
         self.txos = []
@@ -153,7 +156,7 @@ def _send_rpc_request(self, method, params=None):
         # Certain nodes which are placed behind web servers or Cloudflare will
         # configure rate limits and return some HTML error page if we go over that.
         # Zpywallet is not designed to handle such content so we check for it first.
-        # If you are using the full node facilities, ou are recommended to connect
+        # If you are using the full node facilities, you are recommended to connect
         # to your own node and not to a public one, for this reason.
         try:
             j = response.json()
@@ -179,7 +182,6 @@ def _send_batch_rpc_request(self, reqs):
         try:
             # Requests session is not needed for the full node but we can use it
             # for the other providers in the future.
-
             response = requests.post(
                 self.rpc_url,
                 auth=(
@@ -223,56 +225,48 @@ def _get_block_height(self):
         except Exception as e:
             raise NetworkException(f"Failed to make RPC Call: {str(e)}")
 
-    def _get_raw_mempool(self):
-        height = self._get_block_height()
-        h1, h2, h3, h4, h5 = [
-            a["result"]
-            for a in self._send_batch_rpc_request(
-                [
-                    ("getblockhash", [height]),
-                    ("getblockhash", [height - 1]),
-                    ("getblockhash", [height - 2]),
-                    ("getblockhash", [height - 3]),
-                    ("getblockhash", [height - 4]),
-                ]
-            )
-        ]
-        res = self._send_batch_rpc_request(
-            [
-                ("getblock", [h1, 1]),
-                ("getblock", [h2, 1]),
-                ("getblock", [h3, 1]),
-                ("getblock", [h4, 1]),
-                ("getblock", [h5, 1]),
-            ]
-        )
-        tx_counts = [len(r["result"]["tx"]) for r in res]
-        avg_tx_count = sum(tx_counts) // len(tx_counts)
+    # Internal methods are ran in a separate process which allows the OS
+    # to properly garbage collect the memory, as Python leaves a large footprint
+    # behind.
+    def _internal_mempool_fetch(self):
         res = self._send_rpc_request("getrawmempool", [True])
         sorted_transactions = transform_and_sort_transactions(res)
-        sorted_transactions = sorted_transactions[
-            avg_tx_count
-            * self.future_blocks_min : avg_tx_count
-            * self.future_blocks_max
-        ]
-
-        transaction_batches = [
-            sorted_transactions[i : i + self.max_batch]
-            for i in range(0, len(sorted_transactions), self.max_batch)
-        ]
-
-        for transaction_batch in transaction_batches:
+        return [tx["txid"] for tx in sorted_transactions]
 
-            txids = [tx["txid"] for tx in transaction_batch]
+    def _internal_pass_1(self, transaction_batch):
+        sql_transaction_storage = SQLTransactionStorage(self.db_connection_parameters)
+        sql_transaction_storage.connect()
+        try:
+            # The first pass will be to delete the confirmed transactions inside the DB
+            # if applicable.
+            txids = transaction_batch
+            all_txids = sql_transaction_storage.all_txids()
+            for saved_txid in all_txids:
+                if saved_txid not in txids:
+                    sql_transaction_storage.delete_transaction(saved_txid)
+
+            sql_transaction_storage.wipeout_reftxos()
+            sql_transaction_storage.commit()
+
+            # The second pass will be to create a copy of the mempool transactions
+            # without the ones we already have stored inside the list or DB.
+            new_txids = list(set(txids) - set(all_txids))
+
+            new_txid_batches = [
+                [t for t in new_txids[i : i + self.max_batch]]
+                for i in range(0, len(new_txids), self.max_batch)
+            ]
+            return new_txid_batches
 
-            # We are going to replace the mempool transactions with the ones
-            # yielded by this function.
-            # First we are going to yield the ones that are already stored
-            # to save time.
-            for tx in self.transactions:
-                if tx.txid in txids:
-                    yield tx
+        except Exception as e:
+            sql_transaction_storage.rollback()
+            raise e
 
+    def _internal_pass_2(self, transaction_batch):
+        sql_transaction_storage = SQLTransactionStorage(self.db_connection_parameters)
+        sql_transaction_storage.connect()
+        try:
+            txids = transaction_batch
             # Next we are going to yield new mempool transactions that we don't have
             # Confirmed mempool transactions are dropped by this method and the one above.
             txid_batches = [
@@ -296,8 +290,11 @@ def _get_raw_mempool(self):
             # resolving txins.
             if not self.full_transactions:
                 for tx in temp_transactions:
-                    tx.total_fee = 0
-                    yield (tx)
+                    tx.total_fee = 0  # Because this is actually the (v)size
+                    sql_transaction_storage.store_transaction(tx)
+                    for i in range(len(tx.btclike_transaction.outputs)):
+                        sql_transaction_storage.store_txo0(tx, i)
+                sql_transaction_storage.commit()
                 return
 
             # Otherwise we have to get all of the input txids in one swoop so we can
@@ -306,27 +303,48 @@ def _get_raw_mempool(self):
                 self.txos[i : i + self.max_workers]
                 for i in range(0, len(self.txos), self.max_workers)
             ]
-
             with ThreadPoolExecutor(max_workers=self.rps) as executor:
                 futures = [
                     executor.submit(self._postprocess_transaction, txes)
                     for txes in txid_batches
                 ]
-                self.raw_txos = {}
                 for future in futures:
-                    self.raw_txos.update(future.result())
+                    results = future.result()
+                    for txid, result in results.items():
+                        try:
+                            sql_transaction_storage.store_txo1(txid, result)
+                        except DatabaseError:
+                            pass
 
             self.txos = []
-            new_transactions = []
             for i in range(len(temp_transactions) - 1, -1, -1):
                 temp_transaction = temp_transactions[i]
-                new_transactions.append(self._post_clean_tx(temp_transaction))
+                new_transaction = self._post_clean_tx(
+                    temp_transaction, sql_transaction_storage
+                )
+                if new_transaction is not None:
+                    sql_transaction_storage.store_transaction(new_transaction)
+                    for j in range(len(new_transaction.btclike_transaction.inputs)):
+                        sql_transaction_storage.store_txo0(
+                            new_transaction, j, output=False
+                        )
+                    for j in range(len(new_transaction.btclike_transaction.outputs)):
+                        sql_transaction_storage.store_txo0(new_transaction, j)
                 del temp_transactions[i]
+            sql_transaction_storage.wipeout_reftxos()
+            sql_transaction_storage.commit()
+        except Exception as e:
+            sql_transaction_storage.rollback()
+            raise e
+
+    def _get_raw_mempool(self):
+        with multiprocessing.Pool(1) as pool:
+            transaction_batches = pool.apply(self._internal_mempool_fetch)
+
+        new_txid_batches = self._internal_pass_1(transaction_batches)
 
-            self.raw_txos = {}
-            for tx in new_transactions:
-                if tx is not None:
-                    yield tx
+        for transaction_batch in new_txid_batches:
+            self._internal_pass_2(transaction_batch)
 
     def _postprocess_transaction(self, txes):
         res = self._send_batch_rpc_request(
@@ -336,7 +354,9 @@ def _postprocess_transaction(self, txes):
         input_transactions = {}
         for r in res:
             if type(r) is dict and "result" in r.keys():
-                input_transactions[r["result"]["txid"]] = r["result"]
+                input_transactions[r["result"]["txid"]] = self._clean_tx(
+                    r["result"]
+                ).SerializeToString()
 
         return input_transactions
 
@@ -364,5 +384,5 @@ def _process_transaction(self, txes, txids):
         return temp_transactions
 
     def get_raw_mempool(self):
-        self.transactions = [*self._get_raw_mempool()]
-        return self.transactions
+        self._get_raw_mempool()
+        return []