Backup one block at a time.

Author: Neil Booth

electrumx/server/block_processor.py

@@ -20,7 +20,7 @@
 from electrumx.lib.hash import hash_to_hex_str, HASHX_LEN
 from electrumx.lib.script import is_unspendable_legacy, is_unspendable_genesis
 from electrumx.lib.util import (
-    chunks, class_logger, pack_le_uint32, pack_le_uint64, unpack_le_uint64
+    class_logger, pack_le_uint32, pack_le_uint64, unpack_le_uint64
 )
 from electrumx.server.db import FlushData
 
@@ -209,7 +209,7 @@ def schedule_reorg(self, count):
         self.reorg_count = count
         self.blocks_event.set()
 
-    async def reorg_chain(self, count):
+    async def _reorg_chain(self, count):
         '''Handle a chain reorganisation.
 
         Count is the number of blocks to simulate a reorg, or None for
@@ -220,46 +220,47 @@ async def reorg_chain(self, count):
             self.logger.info(f'faking a reorg of {count:,d} blocks')
         await self.flush(True)
 
-        async def get_raw_blocks(last_height, hex_hashes):
-            heights = range(last_height, last_height - len(hex_hashes), -1)
+        async def get_raw_block(hex_hash, height):
             try:
-                blocks = [self.db.read_raw_block(height) for height in heights]
-                self.logger.info(f'read {len(blocks)} blocks from disk')
-                return blocks
+                block = self.db.read_raw_block(height)
+                self.logger.info(f'read block {hex_hash} at height {height:,d} from disk')
             except FileNotFoundError:
-                return await self.daemon.raw_blocks(hex_hashes)
-
-        _start, last, hashes = await self.reorg_hashes(count)
-        # Reverse and convert to hex strings.
-        hashes = [hash_to_hex_str(hash) for hash in reversed(hashes)]
-        for hex_hashes in chunks(hashes, 50):
-            raw_blocks = await get_raw_blocks(last, hex_hashes)
-            await self.backup_blocks(raw_blocks)
+                block = await self.daemon.raw_blocks([hex_hash])[0]
+                self.logger.info(f'obtained block {hex_hash} at height {height:,d} from daemon')
+            return block
+
+        _start, height, hashes = await self._reorg_hashes(count)
+        hex_hashes = [hash_to_hex_str(block_hash) for block_hash in hashes]
+        for hex_hash in reversed(hex_hashes):
+            raw_block = await get_raw_block(hex_hash, height)
+            await self._backup_block(raw_block)
             # self.touched can include other addresses which is harmless, but remove None.
             self.touched.discard(None)
             self.db.flush_backup(self.flush_data(), self.touched)
-            last -= len(raw_blocks)
+            height -= 1
+
+        self.logger.info('backed up to height {:,d}'.format(self.height))
 
         await self.prefetcher.reset_height(self.height)
         self.backed_up_event.set()
         self.backed_up_event.clear()
 
-    async def reorg_hashes(self, count):
+    async def _reorg_hashes(self, count):
         '''Return a pair (start, last, hashes) of blocks to back up during a
         reorg.
 
         The hashes are returned in order of increasing height.  Start
         is the height of the first hash, last of the last.
         '''
-        start, count = await self.calc_reorg_range(count)
+        start, count = await self._calc_reorg_range(count)
         last = start + count - 1
         s = '' if count == 1 else 's'
         self.logger.info(f'chain was reorganised replacing {count:,d} '
                          f'block{s} at heights {start:,d}-{last:,d}')
 
         return start, last, await self.db.fs_block_hashes(start, count)
 
-    async def calc_reorg_range(self, count):
+    async def _calc_reorg_range(self, count):
         '''Calculate the reorg range'''
 
         def diff_pos(hashes1, hashes2):
@@ -441,38 +442,36 @@ def advance_txs(self, txs, is_unspendable):
 
         return undo_info
 
-    async def backup_blocks(self, raw_blocks):
-        '''Backup the raw blocks and flush.
+    async def _backup_block(self, raw_block):
+        '''Backup the raw block and flush.
 
-        The blocks should be in order of decreasing height, starting at.
-        self.height.  A flush is performed once the blocks are backed up.
+        The blocks should be in order of decreasing height, starting at.  self.height.  A
+        flush is performed once the blocks are backed up.
         '''
         self.db.assert_flushed(self.flush_data())
-        assert self.height >= len(raw_blocks)
+        assert self.height > 0
         genesis_activation = self.coin.GENESIS_ACTIVATION
 
         coin = self.coin
-        for raw_block in raw_blocks:
-            # Check and update self.tip
-            block = coin.block(raw_block)
-            header_hash = coin.header_hash(block.header)
-            if header_hash != self.tip:
-                raise ChainError('backup block {} not tip {} at height {:,d}'
-                                 .format(hash_to_hex_str(header_hash),
-                                         hash_to_hex_str(self.tip),
-                                         self.height))
-            self.tip = coin.header_prevhash(block.header)
-            is_unspendable = (is_unspendable_genesis if self.height >= genesis_activation
-                              else is_unspendable_legacy)
-            self.backup_txs(block.transactions, is_unspendable)
-            self.height -= 1
-            self.db.tx_counts.pop()
-
-            await sleep(0)
 
-        self.logger.info('backed up to height {:,d}'.format(self.height))
+        # Check and update self.tip
+        block = coin.block(raw_block)
+        header_hash = coin.header_hash(block.header)
+        if header_hash != self.tip:
+            raise ChainError('backup block {} not tip {} at height {:,d}'
+                             .format(hash_to_hex_str(header_hash),
+                                     hash_to_hex_str(self.tip),
+                                     self.height))
+        self.tip = coin.header_prevhash(block.header)
+        is_unspendable = (is_unspendable_genesis if self.height >= genesis_activation
+                          else is_unspendable_legacy)
+        self._backup_txs(block.transactions, is_unspendable)
+        self.height -= 1
+        self.db.tx_counts.pop()
+
+        await sleep(0)
 
-    def backup_txs(self, txs, is_unspendable):
+    def _backup_txs(self, txs, is_unspendable):
         # Prevout values, in order down the block (coinbase first if present)
         # undo_info is in reverse block order
         undo_info = self.db.read_undo_info(self.height)
@@ -612,7 +611,7 @@ async def _process_blocks(self):
         async def process_event():
             '''Perform any pending reorg, the process prefetched blocks.'''
             if self.reorg_count is not None:
-                await self.reorg_chain(self.reorg_count)
+                await self._reorg_chain(self.reorg_count)
                 self.reorg_count = None
             blocks = self.prefetcher.get_prefetched_blocks()
             await self._advance_blocks(blocks)