Merge pull request #995 from gsalgado/issue-529

Use persist_block() instead of persist_header() in FastChainSyncer

Author: gsalgado

p2p/chain.py

@@ -400,7 +400,7 @@ async def _calculate_td(self, headers: Tuple[BlockHeader, ...]) -> int:
     async def _process_headers(
             self, peer: HeaderRequestingPeer, headers: Tuple[BlockHeader, ...]) -> int:
         target_td = await self._calculate_td(headers)
-        await self._download_block_parts(
+        bodies = await self._download_block_parts(
             target_td,
             [header for header in headers if not _is_body_empty(header)],
             self.request_bodies,
@@ -423,10 +423,18 @@ async def _process_headers(
             'receipt')
         self.logger.debug("Got block receipts for chain segment")
 
-        # FIXME: Get the bodies returned by self._download_block_parts above and use persit_block
-        # here.
         for header in headers:
-            await self.wait(self.db.coro_persist_header(header))
+            if header.uncles_hash != EMPTY_UNCLE_HASH:
+                body = cast(BlockBody, bodies[_body_key(header)])
+                uncles = body.uncles
+            else:
+                uncles = tuple()
+            vm_class = self.chain.get_vm_class_for_block_number(header.block_number)
+            block_class = vm_class.get_block_class()
+            # We don't need to use our block transactions here because persist_block() doesn't do
+            # anything with them as it expects them to have been persisted already.
+            block = block_class(header, uncles=uncles)
+            await self.wait(self.db.coro_persist_block(block))
 
         head = await self.wait(self.db.coro_get_canonical_head())
         return head.block_number
@@ -438,7 +446,8 @@ async def _download_block_parts(
             request_func: Callable[[int, List[BlockHeader]], int],
             download_queue: 'asyncio.Queue[Tuple[ETHPeer, List[DownloadedBlockPart]]]',
             key_func: Callable[[BlockHeader], Union[bytes, Tuple[bytes, bytes]]],
-            part_name: str) -> 'List[DownloadedBlockPart]':
+            part_name: str
+    ) -> 'Dict[Union[bytes, Tuple[bytes, bytes]], Union[BlockBody, List[Receipt]]]':
         """Download block parts for the given headers, using the given request_func.
 
         Retry timed out parts until we have the parts for all headers.
@@ -483,7 +492,7 @@ async def _download_block_parts(
                 if key_func(header) not in received_keys
             ]
 
-        return parts
+        return dict((part.unique_key, part.part) for part in parts)
 
     def _request_block_parts(
             self,
@@ -577,7 +586,7 @@ async def _handle_block_bodies(self,
 
         # TODO: figure out why mypy is losing the type of the transactions_tries
         # so we can get rid of the ignore
-        for (body, (tx_root, trie_dict_data)) in zip(bodies, transactions_tries):  # type: ignore # noqa: E501
+        for (body, (tx_root, trie_dict_data)) in zip(bodies, transactions_tries):  # type: ignore
             await self.wait(self.db.coro_persist_trie_data_dict(trie_dict_data))
             uncles_hash = await self.wait(self.db.coro_persist_uncles(body.uncles))
             downloaded.append(DownloadedBlockPart(body, (tx_root, uncles_hash)))
@@ -686,7 +695,6 @@ async def _process_headers(
             'body')
         self.logger.info("Got block bodies for chain segment")
 
-        parts_by_key = dict((part.unique_key, part.part) for part in downloaded_parts)
         for header in headers:
             vm_class = self.chain.get_vm_class_for_block_number(header.block_number)
             block_class = vm_class.get_block_class()
@@ -695,7 +703,7 @@ async def _process_headers(
                 transactions: List[BaseTransaction] = []
                 uncles: List[BlockHeader] = []
             else:
-                body = cast(eth.BlockBody, parts_by_key[_body_key(header)])
+                body = cast(eth.BlockBody, downloaded_parts[_body_key(header)])
                 tx_class = block_class.get_transaction_class()
                 transactions = [tx_class.from_base_transaction(tx)
                                 for tx in body.transactions]

tests/p2p/integration_test_helpers.py

@@ -42,6 +42,7 @@ class FakeAsyncHeaderDB(AsyncHeaderDB):
 
 
 class FakeAsyncChainDB(FakeAsyncHeaderDB, AsyncChainDB):
+    coro_persist_block = async_passthrough('persist_block')
     coro_persist_uncles = async_passthrough('persist_uncles')
     coro_persist_trie_data_dict = async_passthrough('persist_trie_data_dict')
     coro_get = async_passthrough('get')

trinity/db/chain.py

@@ -14,6 +14,7 @@
 from eth_typing import Hash32
 
 from evm.db.chain import ChainDB
+from evm.rlp.blocks import BaseBlock
 from evm.rlp.headers import BlockHeader
 from evm.rlp.receipts import Receipt
 from evm.rlp.transactions import BaseTransaction
@@ -29,6 +30,9 @@ class AsyncChainDB(ChainDB, AsyncHeaderDB):
     async def coro_get(self, key: bytes) -> bytes:
         raise NotImplementedError()
 
+    async def coro_persist_block(self, block: BaseBlock) -> None:
+        raise NotImplementedError()
+
     async def coro_persist_uncles(self, uncles: Tuple[BlockHeader]) -> Hash32:
         raise NotImplementedError()
 
@@ -58,6 +62,7 @@ class ChainDBProxy(BaseProxy):
     coro_get_canonical_block_hash = async_method('get_canonical_block_hash')
     coro_get_canonical_block_header_by_number = async_method('get_canonical_block_header_by_number')
     coro_persist_header = async_method('persist_header')
+    coro_persist_block = async_method('persist_block')
     coro_persist_uncles = async_method('persist_uncles')
     coro_persist_trie_data_dict = async_method('persist_trie_data_dict')
     coro_get_block_transactions = async_method('get_block_transactions')