Add persist_chain_header API to HeaderDB

Author: cburgdorf

eth/db/chain.py

@@ -34,12 +34,9 @@
 
 from eth.constants import (
     EMPTY_UNCLE_HASH,
-    GENESIS_PARENT_HASH,
 )
 from eth.exceptions import (
-    CanonicalHeadNotFound,
     HeaderNotFound,
-    ParentNotFound,
     TransactionNotFound,
 )
 from eth.db.header import BaseHeaderDB, HeaderDB
@@ -186,63 +183,14 @@ def get_block_uncles(self, uncles_hash: Hash32) -> List[BlockHeader]:
         else:
             return rlp.decode(encoded_uncles, sedes=rlp.sedes.CountableList(BlockHeader))
 
-    # TODO: This method should take a chain of headers as that's the most common use case
-    # and it'd be much faster than inserting each header individually.
-    def persist_header(self,
-                       header: BlockHeader
-                       ) -> Tuple[Tuple[BlockHeader, ...], Tuple[BlockHeader, ...]]:
-        """
-        Returns iterable of headers newly on the canonical chain
-        """
-        is_genesis = header.parent_hash == GENESIS_PARENT_HASH
-        if not is_genesis and not self.header_exists(header.parent_hash):
-            raise ParentNotFound(
-                "Cannot persist block header ({}) with unknown parent ({})".format(
-                    encode_hex(header.hash), encode_hex(header.parent_hash)))
-
-        self.db.set(
-            header.hash,
-            rlp.encode(header),
-        )
-
-        if is_genesis:
-            score = header.difficulty
-        else:
-            score = self.get_score(header.parent_hash) + header.difficulty
-
-        self.db.set(
-            SchemaV1.make_block_hash_to_score_lookup_key(header.hash),
-            rlp.encode(score, sedes=rlp.sedes.big_endian_int),
-        )
-
-        try:
-            head_score = self.get_score(self.get_canonical_head().hash)
-        except CanonicalHeadNotFound:
-            (
-                new_canonical_headers,
-                old_canonical_headers
-            ) = self._set_as_canonical_chain_head(header)
-        else:
-            if score > head_score:
-                (
-                    new_canonical_headers,
-                    old_canonical_headers
-                ) = self._set_as_canonical_chain_head(header)
-            else:
-                new_canonical_headers = tuple()
-                old_canonical_headers = tuple()
-
-        return new_canonical_headers, old_canonical_headers
-
-    # TODO: update this to take a `hash` rather than a full header object.
     def _set_as_canonical_chain_head(self,
-                                     header: BlockHeader
+                                     block_hash: Hash32
                                      ) -> Tuple[Tuple[BlockHeader, ...], Tuple[BlockHeader, ...]]:
         """
         Returns iterable of headers newly on the canonical head
         """
         try:
-            self.get_block_header_by_hash(header.hash)
+            header = self.get_block_header_by_hash(block_hash)
         except HeaderNotFound:
             raise ValueError("Cannot use unknown block hash as canonical head: {}".format(
                 header.hash))

eth/db/header.py

@@ -1,12 +1,18 @@
 from abc import ABC, abstractmethod
 import functools
-from typing import Tuple, Iterable
+from typing import Iterable, Tuple
 
 import rlp
 
+from cytoolz import (
+    first,
+    sliding_window,
+)
+
 from eth_utils import (
     encode_hex,
     to_tuple,
+    ValidationError,
 )
 
 from eth_typing import (
@@ -73,6 +79,12 @@ def persist_header(self,
                        ) -> Tuple[Tuple[BlockHeader, ...], Tuple[BlockHeader, ...]]:
         raise NotImplementedError("ChainDB classes must implement this method")
 
+    @abstractmethod
+    def persist_header_chain(self,
+                             headers: Iterable[BlockHeader]
+                             ) -> Tuple[Tuple[BlockHeader, ...], Tuple[BlockHeader, ...]]:
+        raise NotImplementedError("ChainDB classes must implement this method")
+
 
 class HeaderDB(BaseHeaderDB):
     #
@@ -149,43 +161,66 @@ def header_exists(self, block_hash: Hash32) -> bool:
     def persist_header(self,
                        header: BlockHeader
                        ) -> Tuple[Tuple[BlockHeader, ...], Tuple[BlockHeader, ...]]:
+        return self.persist_header_chain((header,))
+
+    def persist_header_chain(self,
+                             headers: Iterable[BlockHeader]
+                             ) -> Tuple[Tuple[BlockHeader, ...], Tuple[BlockHeader, ...]]:
         """
-        :returns: iterable of headers newly on the canonical chain
+        Return two iterable of headers, the first containing the new canonical headers,
+        the second containing the old canonical headers
         """
-        if header.parent_hash != GENESIS_PARENT_HASH:
-            try:
-                self.get_block_header_by_hash(header.parent_hash)
-            except HeaderNotFound:
+
+        try:
+            first_header = first(headers)
+        except StopIteration:
+            return tuple(), tuple()
+        else:
+
+            for parent, child in sliding_window(2, headers):
+                if parent.hash != child.parent_hash:
+                    raise ValidationError(
+                        "Non-contiguous chain. Expected {} to have {} as parent but was {}".format(
+                            encode_hex(child.hash),
+                            encode_hex(parent.hash),
+                            encode_hex(child.parent_hash),
+                        )
+                    )
+
+            is_genesis = first_header.parent_hash == GENESIS_PARENT_HASH
+            if not is_genesis and not self.header_exists(first_header.parent_hash):
                 raise ParentNotFound(
                     "Cannot persist block header ({}) with unknown parent ({})".format(
-                        encode_hex(header.hash), encode_hex(header.parent_hash)))
+                        encode_hex(first_header.hash), encode_hex(first_header.parent_hash)))
 
-        self.db.set(
-            header.hash,
-            rlp.encode(header),
-        )
+            score = 0 if is_genesis else self.get_score(first_header.parent_hash)
 
-        if header.parent_hash == GENESIS_PARENT_HASH:
-            score = header.difficulty
-        else:
-            score = self.get_score(header.parent_hash) + header.difficulty
+        for header in headers:
+            self.db.set(
+                header.hash,
+                rlp.encode(header),
+            )
 
-        self.db.set(
-            SchemaV1.make_block_hash_to_score_lookup_key(header.hash),
-            rlp.encode(score, sedes=rlp.sedes.big_endian_int),
-        )
+            score += header.difficulty
+
+            self.db.set(
+                SchemaV1.make_block_hash_to_score_lookup_key(header.hash),
+                rlp.encode(score, sedes=rlp.sedes.big_endian_int),
+            )
 
         try:
             head_score = self.get_score(self.get_canonical_head().hash)
         except CanonicalHeadNotFound:
-            new_canonical_headers, old_canonical_headers = self._set_as_canonical_chain_head(
-                header.hash,
-            )
+            (
+                new_canonical_headers,
+                old_canonical_headers
+            ) = self._set_as_canonical_chain_head(header.hash)
         else:
             if score > head_score:
-                new_canonical_headers, old_canonical_headers = self._set_as_canonical_chain_head(
-                    header.hash
-                )
+                (
+                    new_canonical_headers,
+                    old_canonical_headers
+                ) = self._set_as_canonical_chain_head(header.hash)
             else:
                 new_canonical_headers = tuple()
                 old_canonical_headers = tuple()
@@ -297,6 +332,10 @@ async def coro_get_canonical_block_hash(self, block_number: BlockNumber) -> Hash
     async def coro_persist_header(self, header: BlockHeader) -> Tuple[BlockHeader, ...]:
         raise NotImplementedError()
 
+    async def coro_persist_header_chain(self,
+                                        headers: Iterable[BlockHeader]) -> Tuple[BlockHeader, ...]:
+        raise NotImplementedError()
+
 
 # When performing a chain sync (either fast or regular modes), we'll very often need to look
 # up recent block headers to validate the chain, and decoding their RLP representation is

tests/database/test_header_db.py

@@ -8,6 +8,7 @@
 from eth_utils import (
     to_tuple,
     keccak,
+    ValidationError,
 )
 
 from eth.constants import (
@@ -80,8 +81,7 @@ def test_headerdb_get_canonical_head_with_header_chain(headerdb, genesis_header)
 
     headers = mk_header_chain(genesis_header, length=10)
 
-    for header in headers:
-        headerdb.persist_header(header)
+    headerdb.persist_header_chain(headers)
 
     head = headerdb.get_canonical_head()
     assert_headers_eq(head, headers[-1])
@@ -98,6 +98,17 @@ def test_headerdb_persist_header_disallows_unknown_parent(headerdb):
         headerdb.persist_header(header)
 
 
+def test_headerdb_persist_header_chain_disallows_non_contiguous_chain(headerdb, genesis_header):
+    headerdb.persist_header(genesis_header)
+
+    headers = mk_header_chain(genesis_header, length=3)
+
+    non_contiguous_headers = (headers[0], headers[2], headers[1],)
+
+    with pytest.raises(ValidationError, match="Non-contiguous chain"):
+        headerdb.persist_header_chain(non_contiguous_headers)
+
+
 def test_headerdb_persist_header_returns_new_canonical_chain(headerdb, genesis_header):
     gen_result, _ = headerdb.persist_header(genesis_header)
     assert gen_result == (genesis_header,)
@@ -144,8 +155,7 @@ def test_headerdb_get_score_for_non_genesis_headers(headerdb, genesis_header):
     difficulties = tuple(h.difficulty for h in headers)
     scores = tuple(accumulate(operator.add, difficulties, genesis_header.difficulty))
 
-    for header in headers:
-        headerdb.persist_header(header)
+    headerdb.persist_header_chain(headers)
 
     for header, expected_score in zip(headers, scores[1:]):
         actual_score = headerdb.get_score(header.hash)
@@ -212,8 +222,7 @@ def test_headerdb_header_retrieval_by_hash(headerdb, genesis_header):
 
     headers = mk_header_chain(genesis_header, length=10)
 
-    for header in headers:
-        headerdb.persist_header(header)
+    headerdb.persist_header_chain(headers)
 
     # can we get the genesis header by hash
     actual = headerdb.get_block_header_by_hash(genesis_header.hash)
@@ -229,8 +238,7 @@ def test_headerdb_canonical_header_retrieval_by_number(headerdb, genesis_header)
 
     headers = mk_header_chain(genesis_header, length=10)
 
-    for header in headers:
-        headerdb.persist_header(header)
+    headerdb.persist_header_chain(headers)
 
     # can we get the genesis header by hash
     actual = headerdb.get_canonical_block_header_by_number(genesis_header.block_number)

tests/trinity/core/integration_test_helpers.py

@@ -52,6 +52,7 @@ class FakeAsyncHeaderDB(AsyncHeaderDB):
     coro_get_score = async_passthrough('get_score')
     coro_header_exists = async_passthrough('header_exists')
     coro_persist_header = async_passthrough('persist_header')
+    coro_persist_header_chain = async_passthrough('persist_header_chain')
 
 
 class FakeAsyncChainDB(FakeAsyncHeaderDB, AsyncChainDB):

trinity/db/header.py

@@ -4,7 +4,10 @@
 from multiprocessing.managers import (  # type: ignore
     BaseProxy,
 )
-from typing import Tuple
+from typing import (
+    Iterable,
+    Tuple,
+)
 
 from eth_typing import (
     Hash32,
@@ -58,6 +61,11 @@ async def coro_header_exists(self, block_hash: Hash32) -> bool:
     async def coro_persist_header(self, header: BlockHeader) -> Tuple[BlockHeader, ...]:
         raise NotImplementedError("ChainDB classes must implement this method")
 
+    @abstractmethod
+    async def coro_persist_header_chain(self,
+                                        headers: Iterable[BlockHeader]) -> Tuple[BlockHeader, ...]:
+        raise NotImplementedError("ChainDB classes must implement this method")
+
 
 class AsyncHeaderDB(HeaderDB, BaseAsyncHeaderDB):
     async def coro_get_canonical_block_hash(self, block_number: BlockNumber) -> Hash32:
@@ -81,6 +89,10 @@ async def coro_header_exists(self, block_hash: Hash32) -> bool:
     async def coro_persist_header(self, header: BlockHeader) -> Tuple[BlockHeader, ...]:
         raise NotImplementedError("ChainDB classes must implement this method")
 
+    async def coro_persist_header_chain(self,
+                                        headers: Iterable[BlockHeader]) -> Tuple[BlockHeader, ...]:
+        raise NotImplementedError("ChainDB classes must implement this method")
+
 
 class AsyncHeaderDBProxy(BaseProxy, BaseAsyncHeaderDB, BaseHeaderDB):
     coro_get_block_header_by_hash = async_method('get_block_header_by_hash')
@@ -91,6 +103,7 @@ class AsyncHeaderDBProxy(BaseProxy, BaseAsyncHeaderDB, BaseHeaderDB):
     coro_header_exists = async_method('header_exists')
     coro_get_canonical_block_hash = async_method('get_canonical_block_hash')
     coro_persist_header = async_method('persist_header')
+    coro_persist_header_chain = async_method('persist_header_chain')
 
     get_block_header_by_hash = sync_method('get_block_header_by_hash')
     get_canonical_block_hash = sync_method('get_canonical_block_hash')
@@ -100,3 +113,4 @@ class AsyncHeaderDBProxy(BaseProxy, BaseAsyncHeaderDB, BaseHeaderDB):
     header_exists = sync_method('header_exists')
     get_canonical_block_hash = sync_method('get_canonical_block_hash')
     persist_header = sync_method('persist_header')
+    persist_header_chain = sync_method('persist_header_chain')

trinity/sync/light/chain.py

@@ -66,8 +66,7 @@ async def _persist_headers(self) -> None:
             batch_id, headers = await self.wait(self.header_queue.get())
 
             timer = Timer()
-            for header in headers:
-                await self.wait(self.db.coro_persist_header(header))
+            await self.wait(self.db.coro_persist_header_chain(headers))
 
             head = await self.wait(self.db.coro_get_canonical_head())
             self.logger.info(