Merge pull request #1392 from carver/header-skeleton-sync

Header skeleton sync

Author: carver

eth/chains/base.py

@@ -349,7 +349,16 @@ def validate_chain(
                         child, child.parent_hash, parent.hash))
             should_check_seal = index in indices_to_check_seal
             vm_class = cls.get_vm_class_for_block_number(child.block_number)
-            vm_class.validate_header(child, parent, check_seal=should_check_seal)
+            try:
+                vm_class.validate_header(child, parent, check_seal=should_check_seal)
+            except ValidationError as exc:
+                raise ValidationError(
+                    "%s is not a valid child of %s: %s" % (
+                        child,
+                        parent,
+                        exc,
+                    )
+                ) from exc
 
 
 class Chain(BaseChain):

tests/trinity/core/integration_test_helpers.py

@@ -147,6 +147,7 @@ def load_mining_chain(db):
 
 class DBFixture(Enum):
     twenty_pow_headers = '20pow_headers.ldb'
+    thousand_pow_headers = '1000pow_headers.ldb'
 
 
 def load_fixture_db(db_fixture, db_class=LevelDB):

tests/trinity/core/p2p-proto/test_sync.py

@@ -74,6 +74,46 @@ def finalizer():
     assert head.state_root in chaindb_fresh.db
 
 
+@pytest.mark.asyncio
+async def test_skeleton_syncer(request, event_loop, chaindb_fresh, chaindb_1000):
+    client_peer, server_peer = await get_directly_linked_peers(
+        request, event_loop,
+        alice_headerdb=FakeAsyncHeaderDB(chaindb_fresh.db),
+        bob_headerdb=FakeAsyncHeaderDB(chaindb_1000.db))
+    client_peer_pool = MockPeerPoolWithConnectedPeers([client_peer])
+    client = FastChainSyncer(ByzantiumTestChain(chaindb_fresh.db), chaindb_fresh, client_peer_pool)
+    server_peer_pool = MockPeerPoolWithConnectedPeers([server_peer])
+    server = RegularChainSyncer(
+        ByzantiumTestChain(chaindb_1000.db),
+        chaindb_1000,
+        server_peer_pool,
+    )
+    asyncio.ensure_future(server.run())
+    server_request_handler = ETHRequestServer(FakeAsyncChainDB(chaindb_1000.db), server_peer_pool)
+    asyncio.ensure_future(server_request_handler.run())
+    client_peer.logger.info("%s is serving 1000 blocks", client_peer)
+    server_peer.logger.info("%s is syncing up 1000 blocks", server_peer)
+
+    def finalizer():
+        event_loop.run_until_complete(server.cancel())
+        # Yield control so that server.run() returns, otherwise asyncio will complain.
+        event_loop.run_until_complete(asyncio.sleep(0.1))
+    request.addfinalizer(finalizer)
+
+    # FastChainSyncer.run() will return as soon as it's caught up with the peer.
+    await asyncio.wait_for(client.run(), timeout=20)
+
+    head = chaindb_fresh.get_canonical_head()
+    assert head == chaindb_1000.get_canonical_head()
+
+    # Now download the state for the chain's head.
+    state_downloader = StateDownloader(
+        chaindb_fresh, chaindb_fresh.db, head.state_root, client_peer_pool)
+    await asyncio.wait_for(state_downloader.run(), timeout=20)
+
+    assert head.state_root in chaindb_fresh.db
+
+
 @pytest.mark.asyncio
 async def test_regular_syncer(request, event_loop, chaindb_fresh, chaindb_20):
     client_peer, server_peer = await get_directly_linked_peers(
@@ -108,9 +148,9 @@ def finalizer():
 
     asyncio.ensure_future(client.run())
 
-    await wait_for_head(client.db, server.db.get_canonical_head())
-    head = client.db.get_canonical_head()
-    assert head.state_root in client.db.db
+    await wait_for_head(chaindb_fresh, chaindb_20.get_canonical_head())
+    head = chaindb_fresh.get_canonical_head()
+    assert head.state_root in chaindb_fresh.db
 
 
 @pytest.mark.asyncio
@@ -148,14 +188,26 @@ def finalizer():
 
     asyncio.ensure_future(client.run())
 
-    await wait_for_head(client.db, server.db.get_canonical_head())
+    await wait_for_head(chaindb_fresh, chaindb_20.get_canonical_head())
 
 
 @pytest.fixture
 def leveldb_20():
     yield from load_fixture_db(DBFixture.twenty_pow_headers)
 
 
+@pytest.fixture
+def leveldb_1000():
+    yield from load_fixture_db(DBFixture.thousand_pow_headers)
+
+
+@pytest.fixture
+def chaindb_1000(leveldb_1000):
+    chain = load_mining_chain(FakeAsyncAtomicDB(leveldb_1000))
+    assert chain.chaindb.get_canonical_head().block_number == 1000
+    return chain.chaindb
+
+
 @pytest.fixture
 def chaindb_20(leveldb_20):
     chain = load_mining_chain(FakeAsyncAtomicDB(leveldb_20))

tests/trinity/integration/fixtures/1000pow_headers.ldb.zip

@@ -0,0 +1,7 @@
+from trinity.protocol.eth.monitors import ETHChainTipMonitor
+from trinity.protocol.eth.peer import ETHPeer
+from trinity.sync.common.headers import BaseHeaderChainSyncer
+
+
+class ETHHeaderChainSyncer(BaseHeaderChainSyncer[ETHPeer]):
+    tip_monitor_class = ETHChainTipMonitor

trinity/protocol/eth/sync.py

@@ -0,0 +1,7 @@
+from trinity.protocol.les.monitors import LightChainTipMonitor
+from trinity.protocol.les.peer import LESPeer
+from trinity.sync.common.headers import BaseHeaderChainSyncer
+
+
+class LightHeaderChainSyncer(BaseHeaderChainSyncer[LESPeer]):
+    tip_monitor_class = LightChainTipMonitor

trinity/protocol/les/sync.py

@@ -1,12 +1,6 @@
-from abc import abstractmethod
-from contextlib import contextmanager
-from operator import attrgetter
 from typing import (
     AsyncIterator,
-    Iterator,
-    Optional,
     Tuple,
-    Type,
 )
 
 from cancel_token import (
@@ -43,126 +37,13 @@
 
 from trinity.chains.base import BaseAsyncChain
 from trinity.db.header import AsyncHeaderDB
-from trinity.protocol.common.monitors import BaseChainTipMonitor
-from trinity.protocol.common.peer import BaseChainPeer, BaseChainPeerPool
-from trinity.protocol.eth.peer import ETHPeer
-from trinity.sync.common.events import SyncingRequest, SyncingResponse
-from trinity.utils.datastructures import TaskQueue
+from trinity.protocol.common.peer import (
+    BaseChainPeer,
+)
 
 from .types import SyncProgress
 
 
-class BaseHeaderChainSyncer(BaseService):
-    """
-    Sync with the Ethereum network by fetching/storing block headers.
-
-    Here, the run() method will execute the sync loop until our local head is the same as the one
-    with the highest TD announced by any of our peers.
-    """
-    # We'll only sync if we are connected to at least min_peers_to_sync.
-    min_peers_to_sync = 1
-    # the latest header hash of the peer on the current sync
-    header_queue: TaskQueue[BlockHeader]
-
-    def __init__(self,
-                 chain: BaseAsyncChain,
-                 db: AsyncHeaderDB,
-                 peer_pool: BaseChainPeerPool,
-                 token: CancelToken = None) -> None:
-        super().__init__(token)
-        self.chain = chain
-        self.db = db
-        self.peer_pool = peer_pool
-        self._peer_header_syncer: 'PeerHeaderSyncer' = None
-        self._last_target_header_hash: Hash32 = None
-        self._tip_monitor = self.tip_monitor_class(peer_pool, token=self.cancel_token)
-
-        # pending queue size should be big enough to avoid starving the processing consumers, but
-        # small enough to avoid wasteful over-requests before post-processing can happen
-        max_pending_headers = ETHPeer.max_headers_fetch * 8
-        self.header_queue = TaskQueue(max_pending_headers, attrgetter('block_number'))
-
-    def get_target_header_hash(self) -> Hash32:
-        if self._peer_header_syncer is None and self._last_target_header_hash is None:
-            raise ValidationError("Cannot check the target hash before a sync has run")
-        elif self._peer_header_syncer is not None:
-            return self._peer_header_syncer.get_target_header_hash()
-        else:
-            return self._last_target_header_hash
-
-    @property
-    @abstractmethod
-    def tip_monitor_class(self) -> Type[BaseChainTipMonitor]:
-        pass
-
-    async def _run(self) -> None:
-        self.run_daemon(self._tip_monitor)
-        if self.peer_pool.event_bus is not None:
-            self.run_daemon_task(self.handle_sync_status_requests())
-        try:
-            async for highest_td_peer in self._tip_monitor.wait_tip_info():
-                self.run_task(self.sync(highest_td_peer))
-        except OperationCancelled:
-            # In the case of a fast sync, we return once the sync is completed, and our
-            # caller must then run the StateDownloader.
-            return
-        else:
-            self.logger.debug("chain tip monitor stopped returning tip info to %s", self)
-
-    @property
-    def _syncing(self) -> bool:
-        return self._peer_header_syncer is not None
-
-    @contextmanager
-    def _get_peer_header_syncer(self, peer: BaseChainPeer) -> Iterator['PeerHeaderSyncer']:
-        if self._syncing:
-            raise ValidationError("Cannot sync headers from two peers at the same time")
-
-        self._peer_header_syncer = PeerHeaderSyncer(
-            self.chain,
-            self.db,
-            peer,
-            self.cancel_token,
-        )
-        self.run_child_service(self._peer_header_syncer)
-        try:
-            yield self._peer_header_syncer
-        except OperationCancelled:
-            pass
-        else:
-            self._peer_header_syncer.cancel_nowait()
-        finally:
-            self.logger.info("Header Sync with %s ended", peer)
-            self._last_target_header_hash = self._peer_header_syncer.get_target_header_hash()
-            self._peer_header_syncer = None
-
-    async def sync(self, peer: BaseChainPeer) -> None:
-        if self._syncing:
-            self.logger.debug(
-                "Got a NewBlock or a new peer, but already syncing so doing nothing")
-            return
-        elif len(self.peer_pool) < self.min_peers_to_sync:
-            self.logger.info(
-                "Connected to less peers (%d) than the minimum (%d) required to sync, "
-                "doing nothing", len(self.peer_pool), self.min_peers_to_sync)
-            return
-
-        with self._get_peer_header_syncer(peer) as syncer:
-            async for header_batch in syncer.next_header_batch():
-                new_headers = tuple(h for h in header_batch if h not in self.header_queue)
-                await self.wait(self.header_queue.add(new_headers))
-
-    def get_sync_status(self) -> Tuple[bool, Optional[SyncProgress]]:
-        if not self._syncing:
-            return False, None
-        return True, self._peer_header_syncer.sync_progress
-
-    async def handle_sync_status_requests(self) -> None:
-        async for req in self.peer_pool.event_bus.stream(SyncingRequest):
-            self.peer_pool.event_bus.broadcast(SyncingResponse(*self.get_sync_status()),
-                                               req.broadcast_config())
-
-
 class PeerHeaderSyncer(BaseService):
     """
     Sync as many headers as possible with a given peer.

trinity/sync/common/chain.py

@@ -0,0 +1,6 @@
+# If a peer returns 0 results, wait this many seconds before asking it for anything else
+EMPTY_PEER_RESPONSE_PENALTY = 15.0
+
+# Picked a reorg number that is covered by a single skeleton header request,
+# which covers about 6 days at 15s blocks
+MAX_SKELETON_REORG_DEPTH = 35000