[blockchain] Add functionality to batch insert blocks during database synchronization

Author: drcpu-github

blockchain/explorer.py

@@ -36,6 +36,7 @@ def __init__(self, log_queue):
         # Get some configuration parameters
         self.config = BlockchainConfig.config
         self.poll_interval = self.config["explorer"]["poll_interval"]
+        self.batch_size = self.config["explorer"]["batch_size"]
         self.addresses_config = self.config["api"]["caching"]["scripts"]["addresses"]
 
         # Set up logger
@@ -117,6 +118,52 @@ def insert_block(self, database, block_hash_hex_str, block, epoch, tapi_periods)
 
         return block_json
 
+    def batch_insert_blocks(self, database, block_hashes, blocks, epochs, tapi_periods):
+        addresses = {}
+        batched_transactions = {}
+        for block_hash, block, epoch in zip(block_hashes, blocks, epochs):
+            # Create block object and parse it to a JSON object
+            block = Block(
+                block=block,
+                block_hash=block_hash,
+                log_queue=self.log_queue,
+                database=self.insert_blocks_database,
+                tapi_periods=tapi_periods,
+                witnet_node=self.insert_blocks_node,
+                transaction_batch=batched_transactions,
+            )
+            block_json = block.process_block("explorer")
+
+            # Insert block
+            database.insert_block(block_json)
+
+            # Insert transactions
+            transactions = self.insert_transactions(database, block_json, epoch)
+            batched_transactions.update(transactions)
+
+            # Insert address data
+            block_addresses = block.process_addresses(as_dict=True)
+            for address, data in block_addresses.items():
+                if address not in addresses:
+                    addresses[address] = [address, epoch] + data
+                else:
+                    addresses[address][1] = epoch
+                    addresses[address][2] += data[0]
+                    addresses[address][3] += data[1]
+                    addresses[address][4] += data[2]
+                    addresses[address][5] += data[3]
+                    addresses[address][6] += data[4]
+                    addresses[address][7] += data[5]
+                    addresses[address][8] += data[6]
+                    addresses[address][9] += data[7]
+                    addresses[address][10] += data[8]
+
+        # Insert all address data
+        database.insert_addresses(list(addresses.values()))
+
+        # Finalize insertions and updates on every block
+        database.finalize(epochs)
+
     def update_cached_views(self, block_json, logger, caching_server):
         epoch = block_json["details"]["epoch"]
 
@@ -263,36 +310,48 @@ def update_cached_views(self, block_json, logger, caching_server):
             self.try_send_request(logger, caching_server, request)
 
     def insert_transactions(self, database, block_json, epoch):
+        transactions = {}
+
         # Insert mint transaction
         database.insert_mint_txn(block_json["transactions"]["mint"], epoch)
+        transactions[block_json["transactions"]["mint"]["hash"]] = block_json["transactions"]["mint"]
 
         # Insert value transfer transactions
         for txn_details in block_json["transactions"]["value_transfer"]:
             database.insert_value_transfer_txn(txn_details, epoch)
+            transactions[txn_details["hash"]] = txn_details
 
         # Insert data request transactions
         for txn_details in block_json["transactions"]["data_request"]:
             database.insert_data_request_txn(txn_details, epoch)
+            transactions[txn_details["hash"]] = txn_details
 
         # Insert commit transactions
         for txn_details in block_json["transactions"]["commit"]:
             database.insert_commit_txn(txn_details, epoch)
+            transactions[txn_details["hash"]] = txn_details
 
         # Insert reveal transactions
         for txn_details in block_json["transactions"]["reveal"]:
             database.insert_reveal_txn(txn_details, epoch)
+            transactions[txn_details["hash"]] = txn_details
 
         # Insert tally transactions
         for txn_details in block_json["transactions"]["tally"]:
             database.insert_tally_txn(txn_details, epoch)
+            transactions[txn_details["hash"]] = txn_details
 
         # Insert tally transactions
         for txn_details in block_json["transactions"]["stake"]:
             database.insert_stake_txn(txn_details, epoch)
+            transactions[txn_details["hash"]] = txn_details
 
         # Insert tally transactions
         for txn_details in block_json["transactions"]["unstake"]:
             database.insert_unstake_txn(txn_details, epoch)
+            transactions[txn_details["hash"]] = txn_details
+
+        return transactions
 
     def insert_blocks_and_transactions(self, log_queue, unconfirmed_blocks_queue):
         # Set up logger
@@ -347,8 +406,9 @@ def insert_blocks_and_transactions(self, log_queue, unconfirmed_blocks_queue):
             else:
                 blockchain = blockchain["result"]
 
+            block_hashes, blocks, epochs = [], [], []
             for epoch, block_hash_hex_str in blockchain:
-                logger.info(f"Inserting data for epoch {epoch}")
+                logger.info(f"Fetching data for epoch {epoch}")
 
                 block = self.insert_blocks_node.get_block(block_hash_hex_str)
                 # The database entries related to this block have not been modified yet
@@ -361,16 +421,33 @@ def insert_blocks_and_transactions(self, log_queue, unconfirmed_blocks_queue):
                 block = block["result"]
 
                 # Insert block
-                block_json = self.insert_block(
-                    self.insert_blocks_database,
-                    block_hash_hex_str,
-                    block,
-                    epoch,
-                    tapi_periods,
-                )
+                if len(blockchain) < self.batch_size:
+                    block_json = self.insert_block(
+                        self.insert_blocks_database,
+                        block_hash_hex_str,
+                        block,
+                        epoch,
+                        tapi_periods,
+                    )
 
-                # Update all cached views
-                self.update_cached_views(block_json, logger, caching_server)
+                    # Update all cached views
+                    self.update_cached_views(block_json, logger, caching_server)
+                # Batch insert block if they older than the batch size
+                else:
+                    block_hashes.append(block_hash_hex_str)
+                    blocks.append(block)
+                    epochs.append(epoch)
+                    if len(block_hashes) == self.batch_size:
+                        self.batch_insert_blocks(
+                            self.insert_blocks_database,
+                            block_hashes,
+                            blocks,
+                            epochs,
+                            tapi_periods,
+                        )
+                        block_hashes = []
+                        blocks = []
+                        epochs = []
 
                 # Check if the block is confirmed and if it isn't track the hash
                 confirmed = block["confirmed"]

blockchain/objects/block.py

@@ -28,6 +28,7 @@ def __init__(
         database=None,
         tapi_periods=None,
         witnet_node=None,
+        transaction_batch=None
     ):
         self.block = block
         self.block_hash = block_hash
@@ -69,6 +70,8 @@ def __init__(
 
         self.tapi_periods = tapi_periods
 
+        self.transaction_batch = transaction_batch
+
     def configure_logging_process(self, queue, label):
         handler = logging.handlers.QueueHandler(queue)
         root = logging.getLogger(label)
@@ -228,6 +231,7 @@ def process_value_transfer_txns(self, call_from):
         if len(self.block["txns_hashes"]["value_transfer"]) > 0:
             value_transfer = ValueTransfer(
                 database=self.database,
+                transaction_batch=self.transaction_batch,
                 logger=self.logger,
                 witnet_node=self.witnet_node,
             )
@@ -251,6 +255,7 @@ def process_data_request_txns(self, call_from):
         if len(self.block["txns_hashes"]["data_request"]) > 0:
             data_request = DataRequest(
                 database=self.database,
+                transaction_batch=self.transaction_batch,
                 logger=self.logger,
                 witnet_node=self.witnet_node,
             )
@@ -274,6 +279,7 @@ def process_commit_txns(self, call_from):
         if len(self.block["txns_hashes"]["commit"]) > 0:
             commit = Commit(
                 database=self.database,
+                transaction_batch=self.transaction_batch,
                 logger=self.logger,
                 witnet_node=self.witnet_node,
             )
@@ -288,6 +294,7 @@ def process_reveal_txns(self, call_from):
         if len(self.block["txns_hashes"]["reveal"]) > 0:
             reveal = Reveal(
                 database=self.database,
+                transaction_batch=self.transaction_batch,
                 logger=self.logger,
                 witnet_node=self.witnet_node,
             )
@@ -319,6 +326,7 @@ def process_stake_txns(self, call_from):
         ):
             stake = Stake(
                 database=self.database,
+                transaction_batch=self.transaction_batch,
                 logger=self.logger,
                 witnet_node=self.witnet_node,
             )

blockchain/transactions/commit.py

@@ -29,6 +29,7 @@ def process_transaction(self, call_from):
         fee = get_commit_and_reveal_fee_for_data_request(
             self.database,
             self.txn_details["data_request"],
+            transaction_batch=self.transaction_batch,
         )
         if "fee" in fee:
             self.txn_details["fee"] = fee["fee"]
@@ -43,6 +44,7 @@ def process_transaction(self, call_from):
         collateral = get_collateral_for_data_request(
             self.database,
             self.txn_details["data_request"],
+            transaction_batch=self.transaction_batch,
         )
         if "collateral" in collateral:
             self.txn_details["collateral"] = collateral["collateral"]

blockchain/transactions/data_request.py

@@ -320,46 +320,54 @@ def calculate_fees(
         return dro_fee, miner_fee
 
 
-def get_commit_and_reveal_fee_for_data_request(database, data_request_hash):
-    sql = """
-        SELECT
-            data_request_txns.commit_and_reveal_fee
-        FROM
-            data_request_txns
-        WHERE
-            data_request_txns.txn_hash=%s
-        LIMIT 1
-    """
-    result = database.sql_return_one(
-        sql,
-        parameters=[bytearray.fromhex(data_request_hash)],
-    )
-
-    if result:
-        return {"fee": result[0]}
+def get_commit_and_reveal_fee_for_data_request(database, data_request_hash, transaction_batch=None):
+    if transaction_batch is not None and data_request_hash in transaction_batch:
+        data_request_txn = transaction_batch[data_request_hash]
+        return {"fee": data_request_txn["commit_and_reveal_fee"]}
     else:
-        return {"error": "transaction not found"}
-
-
-def get_collateral_for_data_request(database, data_request_hash):
-    sql = """
-        SELECT
-            data_request_txns.collateral
-        FROM
-            data_request_txns
-        WHERE
-            data_request_txns.txn_hash=%s
-        LIMIT 1
-    """
-    result = database.sql_return_one(
-        sql,
-        parameters=[bytearray.fromhex(data_request_hash)],
-    )
-
-    if result:
-        return {"collateral": result[0]}
+        sql = """
+            SELECT
+                data_request_txns.commit_and_reveal_fee
+            FROM
+                data_request_txns
+            WHERE
+                data_request_txns.txn_hash=%s
+            LIMIT 1
+        """
+        result = database.sql_return_one(
+            sql,
+            parameters=[bytearray.fromhex(data_request_hash)],
+        )
+
+        if result:
+            return {"fee": result[0]}
+        else:
+            return {"error": "transaction not found"}
+
+
+def get_collateral_for_data_request(database, data_request_hash, transaction_batch=None):
+    if transaction_batch is not None and data_request_hash in transaction_batch:
+        data_request_txn = transaction_batch[data_request_hash]
+        return {"collateral": data_request_txn["collateral"]}
     else:
-        return {"error": "transaction not found"}
+        sql = """
+            SELECT
+                data_request_txns.collateral
+            FROM
+                data_request_txns
+            WHERE
+                data_request_txns.txn_hash=%s
+            LIMIT 1
+        """
+        result = database.sql_return_one(
+            sql,
+            parameters=[bytearray.fromhex(data_request_hash)],
+        )
+
+        if result:
+            return {"collateral": result[0]}
+        else:
+            return {"error": "transaction not found"}
 
 
 def build_retrieval(kinds, urls, all_headers, bodies, scripts):

blockchain/transactions/reveal.py

@@ -33,6 +33,7 @@ def process_transaction(self, call_from):
         fee = get_commit_and_reveal_fee_for_data_request(
             self.database,
             self.txn_details["data_request"],
+            transaction_batch=self.transaction_batch,
         )
         if "fee" in fee:
             self.txn_details["fee"] = fee["fee"]

blockchain/transactions/transaction.py

@@ -19,6 +19,7 @@ class Transaction(object):
     def __init__(
         self,
         database=None,
+        transaction_batch=None,
         logger=None,
         witnet_node=None,
     ):
@@ -41,6 +42,8 @@ def __init__(
                     logger=logger, custom_types=["utxo", "filter"]
                 )
 
+        self.transaction_batch = transaction_batch
+
         # Set up logger
         if logger:
             self.logger = logger
@@ -119,8 +122,21 @@ def get_inputs(self, txn_inputs):
             hash_bytes = bytearray.fromhex(input_hash)
             input_utxos.append((hash_bytes, input_index))
 
-            # Try to find the transaction input value in the database
             outputs = None
+
+            # Try to find the transaction input value in the transactions batch if it exists
+            if self.transaction_batch is not None and input_hash in self.transaction_batch:
+                transaction = self.transaction_batch[input_hash]
+                if "output_value" in transaction:
+                    outputs = transaction["output_value"]
+                    input_values.append(outputs)
+                else:
+                    outputs = transaction["output_values"]
+                    input_values.append(outputs[input_index])
+
+                continue
+
+            # Try to find the transaction input value in the database
             sql = """
                 SELECT
                     type
@@ -176,8 +192,10 @@ def get_inputs(self, txn_inputs):
                     else:
                         input_values.append(outputs[0][input_index])
 
+                    continue
+
             # Fall back: transaction not found in database, fetch it from the node
-            if not outputs:
+            if outputs is None:
                 if self.logger:
                     self.logger.info(
                         f"Could not find input {txn_input['output_pointer']} for transaction {self.txn_hash} in database"