verifier: request proofs in batches

electrum/network.py

@@ -719,7 +719,6 @@ class Network(PrintError):
             pass
 
     async def get_merkle_for_transaction(self, tx_hash, tx_height):
-        print("getting merkle for transaction", tx_hash, tx_height)
         return await self.interface.session.send_request('blockchain.transaction.get_merkle', [tx_hash, tx_height])
 
     def broadcast_transaction(self, tx):

electrum/verifier.py

@@ -24,6 +24,8 @@
 import asyncio
 from typing import Sequence, Optional
 
+from aiorpcx import TaskGroup
+
 from .util import ThreadJob, bh2u, VerifiedTxInfo, aiosafe
 from .bitcoin import Hash, hash_decode, hash_encode
 from .transaction import Transaction
@@ -49,10 +51,10 @@ class SPV(ThreadJob):
     @aiosafe
     async def main(self):
         while True:
-            await self.run()
+            await self._request_proofs()
             await asyncio.sleep(1)
 
-    async def run(self):
+    async def _request_proofs(self):
         blockchain = self.network.blockchain()
         if not blockchain:
             self.print_error("no blockchain")
@@ -60,33 +62,30 @@ class SPV(ThreadJob):
 
         local_height = self.network.get_local_height()
         unverified = self.wallet.get_unverified_txs()
-        #print("verifier run", len(unverified))
+
-        for tx_hash, tx_height in unverified.items():
+        async with TaskGroup() as group:
-            # do not request merkle branch before headers are available
+            for tx_hash, tx_height in unverified.items():
-            if tx_height <= 0 or tx_height > local_height:
+                # do not request merkle branch before headers are available
-                continue
+                if tx_height <= 0 or tx_height > local_height:
-
+                    continue
-            header = blockchain.read_header(tx_height)
+
-            if header is None:
+                header = blockchain.read_header(tx_height)
-                index = tx_height // 2016
+                if header is None:
-                if index < len(blockchain.checkpoints):
+                    index = tx_height // 2016
-                    await self.network.request_chunk(tx_height, None)
+                    if index < len(blockchain.checkpoints):
-            elif (tx_hash not in self.requested_merkle
+                        await group.spawn(self.network.request_chunk, tx_height, None)
-                    and tx_hash not in self.merkle_roots):
+                elif (tx_hash not in self.requested_merkle
-                self.print_error('requested merkle', tx_hash)
+                        and tx_hash not in self.merkle_roots):
-                self.requested_merkle.add(tx_hash)
+                    self.print_error('requested merkle', tx_hash)
-                self.verify_merkle(tx_hash, await self.network.get_merkle_for_transaction(
+                    self.requested_merkle.add(tx_hash)
-                        tx_hash,
+                    await group.spawn(self._request_and_verify_single_proof, tx_hash, tx_height)
-                        tx_height
-                        ))
 
         if self.network.blockchain() != self.blockchain:
             self.blockchain = self.network.blockchain()
-            self.undo_verifications()
+            self._undo_verifications()
 
-    def verify_merkle(self, tx_hash, merkle):
+    async def _request_and_verify_single_proof(self, tx_hash, tx_height):
-        if self.wallet.verifier is None:
+        merkle = await self.network.get_merkle_for_transaction(tx_hash, tx_height)
-            return  # we have been killed, this was just an orphan callback
         # Verify the hash of the server-provided merkle branch to a
         # transaction matches the merkle root of its block
         tx_height = merkle.get('block_height')
@@ -143,7 +142,7 @@ class SPV(ThreadJob):
         else:
             raise InnerNodeOfSpvProofIsValidTx()
 
-    def undo_verifications(self):
+    def _undo_verifications(self):
         height = self.blockchain.get_forkpoint()
         tx_hashes = self.wallet.undo_verifications(self.blockchain, height)
         for tx_hash in tx_hashes: