Implement eth_syncing endpoint (#1421)

* Implement eth_syncing endpoint

* Implement eth_syncing endpoint

* Remove slash based line continuation.

Author: rayrapetyan

tests/trinity/core/json-rpc/test_ipc.py

@@ -22,6 +22,13 @@
     NetworkIdRequest,
     NetworkIdResponse,
 )
+from trinity.sync.common.events import (
+    SyncingRequest,
+    SyncingResponse,
+)
+from trinity.sync.common.types import (
+    SyncProgress
+)
 
 from trinity.utils.version import construct_trinity_client_identifier
 
@@ -447,7 +454,54 @@ async def mock_event_bus_interaction(bus):
     ],
 )
 async def test_peer_pool_over_ipc(
-        monkeypatch,
+        jsonrpc_ipc_pipe_path,
+        request_msg,
+        event_bus_setup_fn,
+        event_bus,
+        expected,
+        event_loop,
+        ipc_server):
+
+    asyncio.ensure_future(event_bus_setup_fn(event_bus))
+
+    result = await get_ipc_response(
+        jsonrpc_ipc_pipe_path,
+        request_msg,
+        event_loop
+    )
+    assert result == expected
+
+
+def mock_syncing(is_syncing, progress=None):
+    async def mock_event_bus_interaction(bus):
+        async for req in bus.stream(SyncingRequest):
+            bus.broadcast(SyncingResponse(is_syncing, progress), req.broadcast_config())
+            break
+
+    return mock_event_bus_interaction
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'request_msg, event_bus_setup_fn, expected',
+    (
+        (
+            build_request('eth_syncing'),
+            mock_syncing(False),
+            {'result': False, 'id': 3, 'jsonrpc': '2.0'},
+        ),
+        (
+            build_request('eth_syncing'),
+            mock_syncing(True, SyncProgress(0, 1, 2)),
+            {'result': {'startingBlock': 0, 'currentBlock': 1, 'highestBlock': 2}, 'id': 3,
+             'jsonrpc': '2.0'},
+        ),
+    ),
+    ids=[
+        'eth_syncing_F', 'eth_syncing_T',
+    ],
+)
+async def test_eth_over_ipc(
         jsonrpc_ipc_pipe_path,
         request_msg,
         event_bus_setup_fn,

trinity/protocol/common/peer.py

@@ -48,6 +48,7 @@ class BaseChainPeer(BasePeer):
 
     head_td: int = None
     head_hash: Hash32 = None
+    head_number: BlockNumber = None
 
     @property
     @abstractmethod

trinity/protocol/les/peer.py

@@ -44,8 +44,6 @@
 
 
 class LESPeer(BaseChainPeer):
-    head_number: BlockNumber = None
-
     max_headers_fetch = MAX_HEADERS_FETCH
 
     _supported_sub_protocols = [LESProtocol, LESProtocolV2]

trinity/rpc/modules/eth.py

@@ -8,6 +8,9 @@
     List,
     Union,
 )
+from mypy_extensions import (
+    TypedDict,
+)
 
 from eth_typing import (
     Address,
@@ -51,6 +54,9 @@
 from trinity.rpc.modules import (
     RPCModule,
 )
+from trinity.sync.common.events import (
+    SyncingRequest,
+)
 from trinity.utils.validation import (
     validate_transaction_call_dict,
     validate_transaction_gas_estimation_dict,
@@ -264,5 +270,17 @@ async def mining(self) -> bool:
     async def protocolVersion(self) -> str:
         return "63"
 
-    async def syncing(self) -> bool:
-        raise NotImplementedError()
+    class SyncProgress(TypedDict):
+        startingBlock: BlockNumber
+        currentBlock: BlockNumber
+        highestBlock: BlockNumber
+
+    async def syncing(self) -> Union[bool, SyncProgress]:
+        res = await self._event_bus.request(SyncingRequest())
+        if res.is_syncing:
+            return {
+                "startingBlock": res.progress.starting_block,
+                "currentBlock": res.progress.current_block,
+                "highestBlock": res.progress.highest_block
+            }
+        return False

trinity/sync/common/chain.py

@@ -4,11 +4,15 @@
 from typing import (
     AsyncIterator,
     Iterator,
+    Optional,
     Tuple,
     Type,
 )
 
-from cancel_token import CancelToken, OperationCancelled
+from cancel_token import (
+    CancelToken,
+    OperationCancelled,
+)
 
 from eth.constants import GENESIS_BLOCK_NUMBER
 from eth.chains import AsyncChain
@@ -23,17 +27,43 @@
     encode_hex,
     ValidationError,
 )
-from eth.rlp.headers import BlockHeader
+from eth.rlp.headers import (
+    BlockHeader,
+)
+
+from p2p.constants import (
+    MAX_REORG_DEPTH,
+    SEAL_CHECK_RANDOM_SAMPLE_RATE,
+)
+from p2p.p2p_proto import (
+    DisconnectReason,
+)
+from p2p.service import (
+    BaseService,
+)
 
-from p2p.constants import MAX_REORG_DEPTH, SEAL_CHECK_RANDOM_SAMPLE_RATE
-from p2p.p2p_proto import DisconnectReason
-from p2p.service import BaseService
+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.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.utils.datastructures import TaskQueue
+from .types import SyncProgress
 
 
 class BaseHeaderChainSyncer(BaseService):
@@ -81,6 +111,8 @@ def tip_monitor_class(self) -> Type[BaseChainTipMonitor]:
 
     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))
@@ -134,6 +166,16 @@ async def sync(self, peer: BaseChainPeer) -> None:
                 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):
     """
@@ -152,6 +194,7 @@ def __init__(self,
         super().__init__(token)
         self.chain = chain
         self.db = db
+        self.sync_progress: SyncProgress = None
         self._peer = peer
         self._target_header_hash = peer.head_hash
 
@@ -183,7 +226,7 @@ async def next_header_batch(self) -> AsyncIterator[Tuple[BlockHeader, ...]]:
             self.logger.debug(
                 "%s announced Head TD %d, which is higher than ours (%d), starting sync",
                 peer, peer.head_td, head_td)
-
+        self.sync_progress = SyncProgress(head.block_number, head.block_number, peer.head_number)
         self.logger.info("Starting sync with %s", peer)
         last_received_header: BlockHeader = None
         # When we start the sync with a peer, we always request up to MAX_REORG_DEPTH extra
@@ -306,6 +349,9 @@ async def next_header_batch(self) -> AsyncIterator[Tuple[BlockHeader, ...]]:
 
             yield headers
             last_received_header = headers[-1]
+            self.sync_progress = self.sync_progress.update_current_block(
+                last_received_header.block_number,
+            )
             start_at = last_received_header.block_number + 1
 
     async def _request_headers(

trinity/sync/common/events.py

@@ -0,0 +1,25 @@
+from typing import (
+    Optional,
+    Type,
+)
+
+from lahja import (
+    BaseEvent,
+    BaseRequestResponseEvent,
+)
+
+from trinity.sync.common.types import (
+    SyncProgress
+)
+
+
+class SyncingResponse(BaseEvent):
+    def __init__(self, is_syncing: bool, progress: Optional[SyncProgress]) -> None:
+        self.is_syncing: bool = is_syncing
+        self.progress: Optional[SyncProgress] = progress
+
+
+class SyncingRequest(BaseRequestResponseEvent[SyncingResponse]):
+    @staticmethod
+    def expected_response_type() -> Type[SyncingResponse]:
+        return SyncingResponse

trinity/sync/common/types.py

@@ -0,0 +1,16 @@
+from typing import (
+    NamedTuple,
+)
+
+from eth_typing import (
+    BlockNumber,
+)
+
+
+class SyncProgress(NamedTuple):
+    starting_block: BlockNumber
+    current_block: BlockNumber
+    highest_block: BlockNumber
+
+    def update_current_block(self, new_current_block: BlockNumber) -> 'SyncProgress':
+        return SyncProgress(self.starting_block, new_current_block, self.highest_block)