Add LightPeerChainBridgePlugin

Author: cburgdorf

tests/trinity/core/chain-management/test_light_peer_chain.py

@@ -0,0 +1,20 @@
+from trinity.sync.light.service import (
+    LightPeerChain
+)
+from trinity.plugins.builtin.light_peer_chain_bridge import (
+    EventBusLightPeerChain,
+)
+
+
+# These tests may seem obvious but they safe us from runtime errors where
+# changes are made to the `BaseLightPeerChain` that are then forgotton to
+# implement on both derived chains.
+
+def test_can_instantiate_eventbus_light_peer_chain():
+    chain = EventBusLightPeerChain(None)
+    assert chain is not None
+
+
+def test_can_instantiate_light_peer_chain():
+    chain = LightPeerChain(None, None)
+    assert chain is not None

tests/trinity/integration/test_lightchain_integration.py

@@ -200,26 +200,26 @@ async def wait_for_header_sync(block_number):
 
     # https://ropsten.etherscan.io/block/11
     header = headerdb.get_canonical_block_header_by_number(n)
-    body = await peer_chain.get_block_body_by_hash(header.hash)
+    body = await peer_chain.coro_get_block_body_by_hash(header.hash)
     assert len(body['transactions']) == 15
 
-    receipts = await peer_chain.get_receipts(header.hash)
+    receipts = await peer_chain.coro_get_receipts(header.hash)
     assert len(receipts) == 15
     assert encode_hex(keccak(rlp.encode(receipts[0]))) == (
         '0xf709ed2c57efc18a1675e8c740f3294c9e2cb36ba7bb3b89d3ab4c8fef9d8860')
 
     assert len(peer_pool) == 1
     peer = peer_pool.highest_td_peer
-    head = await peer_chain.get_block_header_by_hash(peer.head_hash)
+    head = await peer_chain.coro_get_block_header_by_hash(peer.head_hash)
 
     # In order to answer queries for contract code, geth needs the state trie entry for the block
     # we specify in the query, but because of fast sync we can only assume it has that for recent
     # blocks, so we use the current head to lookup the code for the contract below.
     # https://ropsten.etherscan.io/address/0x95a48dca999c89e4e284930d9b9af973a7481287
     contract_addr = decode_hex('0x8B09D9ac6A4F7778fCb22852e879C7F3B2bEeF81')
-    contract_code = await peer_chain.get_contract_code(head.hash, contract_addr)
+    contract_code = await peer_chain.coro_get_contract_code(head.hash, contract_addr)
     assert encode_hex(contract_code) == '0x600060006000600060006000356000f1'
 
-    account = await peer_chain.get_account(head.hash, contract_addr)
+    account = await peer_chain.coro_get_account(head.hash, contract_addr)
     assert account.code_hash == keccak(contract_code)
     assert account.balance == 0

trinity/chains/light.py

@@ -141,7 +141,7 @@ def get_block_by_hash(self, block_hash: Hash32) -> BaseBlock:
     def get_block_by_header(self, header: BlockHeader) -> BaseBlock:
         # TODO check local cache, before hitting peer
         block_body = self._run_async(
-            self._peer_chain.get_block_body_by_hash(header.hash)
+            self._peer_chain.coro_get_block_body_by_hash(header.hash)
         )
 
         block_class = self.get_vm_class_for_block_number(header.block_number).get_block_class()

trinity/plugins/builtin/light_peer_chain_bridge/__init__.py

@@ -0,0 +1,4 @@
+from .light_peer_chain_bridge import (  # noqa: F401
+    EventBusLightPeerChain,
+    LightPeerChainEventBusHandler,
+)

trinity/plugins/builtin/light_peer_chain_bridge/light_peer_chain_bridge.py

@@ -0,0 +1,257 @@
+from typing import (
+    List,
+    Type,
+    TypeVar,
+)
+
+from cancel_token import (
+    CancelToken,
+)
+
+from eth_typing import (
+    Address,
+    Hash32,
+)
+
+from eth.rlp.accounts import (
+    Account,
+)
+from eth.rlp.headers import (
+    BlockHeader,
+)
+from eth.rlp.receipts import (
+    Receipt,
+)
+
+from lahja import (
+    BaseEvent,
+    BaseRequestResponseEvent,
+    Endpoint,
+)
+
+from p2p.service import (
+    BaseService,
+)
+
+from trinity.utils.async_errors import (
+    await_and_wrap_errors,
+)
+from trinity.rlp.block_body import BlockBody
+from trinity.sync.light.service import (
+    BaseLightPeerChain,
+)
+
+
+class BaseLightPeerChainResponse(BaseEvent):
+
+    def __init__(self, error: Exception) -> None:
+        self.error = error
+
+
+class BlockHeaderResponse(BaseLightPeerChainResponse):
+
+    def __init__(self, block_header: BlockHeader, error: Exception=None) -> None:
+        super().__init__(error)
+        self.block_header = block_header
+
+
+class BlockBodyResponse(BaseLightPeerChainResponse):
+
+    def __init__(self, block_body: BlockBody, error: Exception=None) -> None:
+        super().__init__(error)
+        self.block_body = block_body
+
+
+class ReceiptsResponse(BaseLightPeerChainResponse):
+
+    def __init__(self, receipts: List[Receipt], error: Exception=None) -> None:
+        super().__init__(error)
+        self.receipts = receipts
+
+
+class AccountResponse(BaseLightPeerChainResponse):
+
+    def __init__(self, account: Account, error: Exception=None) -> None:
+        super().__init__(error)
+        self.account = account
+
+
+class BytesResponse(BaseLightPeerChainResponse):
+
+    def __init__(self, bytez: bytes, error: Exception=None) -> None:
+        super().__init__(error)
+        self.bytez = bytez
+
+
+class GetBlockHeaderByHashRequest(BaseRequestResponseEvent[BlockHeaderResponse]):
+
+    def __init__(self, block_hash: Hash32) -> None:
+        self.block_hash = block_hash
+
+    @staticmethod
+    def expected_response_type() -> Type[BlockHeaderResponse]:
+        return BlockHeaderResponse
+
+
+class GetBlockBodyByHashRequest(BaseRequestResponseEvent[BlockBodyResponse]):
+
+    def __init__(self, block_hash: Hash32) -> None:
+        self.block_hash = block_hash
+
+    @staticmethod
+    def expected_response_type() -> Type[BlockBodyResponse]:
+        return BlockBodyResponse
+
+
+class GetReceiptsRequest(BaseRequestResponseEvent[ReceiptsResponse]):
+
+    def __init__(self, block_hash: Hash32) -> None:
+        self.block_hash = block_hash
+
+    @staticmethod
+    def expected_response_type() -> Type[ReceiptsResponse]:
+        return ReceiptsResponse
+
+
+class GetAccountRequest(BaseRequestResponseEvent[AccountResponse]):
+
+    def __init__(self, block_hash: Hash32, address: Address) -> None:
+        self.block_hash = block_hash
+        self.address = address
+
+    @staticmethod
+    def expected_response_type() -> Type[AccountResponse]:
+        return AccountResponse
+
+
+class GetContractCodeRequest(BaseRequestResponseEvent[BytesResponse]):
+
+    def __init__(self, block_hash: Hash32, address: Address) -> None:
+        self.block_hash = block_hash
+        self.address = address
+
+    @staticmethod
+    def expected_response_type() -> Type[BytesResponse]:
+        return BytesResponse
+
+
+class LightPeerChainEventBusHandler(BaseService):
+    """
+    The ``LightPeerChainEventBusHandler`` listens for certain events on the eventbus and
+    delegates them to the ``LightPeerChain`` to get answers. It then propagates responses
+    back to the caller.
+    """
+
+    def __init__(self,
+                 chain: BaseLightPeerChain,
+                 event_bus: Endpoint,
+                 token: CancelToken = None) -> None:
+        super().__init__(token)
+        self.chain = chain
+        self.event_bus = event_bus
+
+    async def _run(self) -> None:
+        self.logger.info("Running LightPeerChainEventBusHandler")
+
+        self.run_daemon_task(self.handle_get_blockheader_by_hash_requests())
+        self.run_daemon_task(self.handle_get_blockbody_by_hash_requests())
+        self.run_daemon_task(self.handle_get_receipts_by_hash_requests())
+        self.run_daemon_task(self.handle_get_account_requests())
+        self.run_daemon_task(self.handle_get_contract_code_requests())
+
+    async def handle_get_blockheader_by_hash_requests(self) -> None:
+        async for event in self.event_bus.stream(GetBlockHeaderByHashRequest):
+
+            val, error = await await_and_wrap_errors(
+                self.chain.coro_get_block_header_by_hash(event.block_hash)
+            )
+
+            self.event_bus.broadcast(
+                event.expected_response_type()(val, error),
+                event.broadcast_config()
+            )
+
+    async def handle_get_blockbody_by_hash_requests(self) -> None:
+        async for event in self.event_bus.stream(GetBlockBodyByHashRequest):
+
+            val, error = await await_and_wrap_errors(
+                self.chain.coro_get_block_body_by_hash(event.block_hash)
+            )
+
+            self.event_bus.broadcast(
+                event.expected_response_type()(val, error),
+                event.broadcast_config()
+            )
+
+    async def handle_get_receipts_by_hash_requests(self) -> None:
+        async for event in self.event_bus.stream(GetReceiptsRequest):
+
+            val, error = await await_and_wrap_errors(self.chain.coro_get_receipts(event.block_hash))
+
+            self.event_bus.broadcast(
+                event.expected_response_type()(val, error),
+                event.broadcast_config()
+            )
+
+    async def handle_get_account_requests(self) -> None:
+        async for event in self.event_bus.stream(GetAccountRequest):
+
+            val, error = await await_and_wrap_errors(
+                self.chain.coro_get_account(event.block_hash, event.address)
+            )
+
+            self.event_bus.broadcast(
+                event.expected_response_type()(val, error),
+                event.broadcast_config()
+            )
+
+    async def handle_get_contract_code_requests(self) -> None:
+
+        async for event in self.event_bus.stream(GetContractCodeRequest):
+
+            val, error = await await_and_wrap_errors(
+                self.chain.coro_get_contract_code(event.block_hash, event.address)
+            )
+
+            self.event_bus.broadcast(
+                event.expected_response_type()(val, error),
+                event.broadcast_config()
+            )
+
+
+class EventBusLightPeerChain(BaseLightPeerChain):
+    """
+    The ``EventBusLightPeerChain`` is an implementation of the ``BaseLightPeerChain`` that can
+    be used from within any process.
+    """
+
+    def __init__(self, event_bus: Endpoint) -> None:
+        self.event_bus = event_bus
+
+    async def coro_get_block_header_by_hash(self, block_hash: Hash32) -> BlockHeader:
+        event = GetBlockHeaderByHashRequest(block_hash)
+        return self._pass_or_raise(await self.event_bus.request(event)).block_header
+
+    async def coro_get_block_body_by_hash(self, block_hash: Hash32) -> BlockBody:
+        event = GetBlockBodyByHashRequest(block_hash)
+        return self._pass_or_raise(await self.event_bus.request(event)).block_body
+
+    async def coro_get_receipts(self, block_hash: Hash32) -> List[Receipt]:
+        event = GetReceiptsRequest(block_hash)
+        return self._pass_or_raise(await self.event_bus.request(event)).receipts
+
+    async def coro_get_account(self, block_hash: Hash32, address: Address) -> Account:
+        event = GetAccountRequest(block_hash, address)
+        return self._pass_or_raise(await self.event_bus.request(event)).account
+
+    async def coro_get_contract_code(self, block_hash: Hash32, address: Address) -> bytes:
+        event = GetContractCodeRequest(block_hash, address)
+        return self._pass_or_raise(await self.event_bus.request(event)).bytez
+
+    TResponse = TypeVar("TResponse", bound=BaseLightPeerChainResponse)
+
+    def _pass_or_raise(self, response: TResponse) -> TResponse:
+        if response.error is not None:
+            raise response.error
+
+        return response

trinity/plugins/builtin/light_peer_chain_bridge/plugin.py

@@ -0,0 +1,55 @@
+import asyncio
+from typing import (
+    cast
+)
+
+from eth.chains.base import (
+    BaseChain
+)
+
+from trinity.constants import (
+    SYNC_LIGHT
+)
+from trinity.extensibility import (
+    BaseEvent,
+    BasePlugin,
+)
+from trinity.chains.light import (
+    LightDispatchChain,
+)
+from trinity.extensibility.events import (
+    ResourceAvailableEvent
+)
+from trinity.plugins.builtin.light_peer_chain_bridge import (
+    LightPeerChainEventBusHandler
+)
+
+
+class LightPeerChainBridgePlugin(BasePlugin):
+    """
+    The ``LightPeerChainBridgePlugin`` runs in the ``networking`` process and acts as a bridge
+    between other processes and the ``LightPeerChain``.
+    It runs only in ``light`` mode.
+    Other plugins can instantiate the ``EventBusLightPeerChain`` from separate processes to
+    interact with the ``LightPeerChain`` indirectly.
+    """
+
+    chain: BaseChain = None
+
+    @property
+    def name(self) -> str:
+        return "LightPeerChain Bridge"
+
+    def should_start(self) -> bool:
+        return self.chain is not None and self.context.chain_config.sync_mode == SYNC_LIGHT
+
+    def handle_event(self, activation_event: BaseEvent) -> None:
+        if isinstance(activation_event, ResourceAvailableEvent):
+            if activation_event.resource_type is BaseChain:
+                self.chain = activation_event.resource
+
+    def start(self) -> None:
+        self.logger.info('LightPeerChain Bridge started')
+        chain = cast(LightDispatchChain, self.chain)
+        handler = LightPeerChainEventBusHandler(chain._peer_chain, self.context.event_bus)
+        asyncio.ensure_future(handler.run())