Merge pull request #1400 from NIC619/set_up_eth1_monitor_component

Set up eth1 monitor component

eth2/beacon/scripts/quickstart_state/generate_beacon_genesis.py

@@ -1,12 +1,13 @@
 from pathlib import Path
 import time
 
-from eth_utils import decode_hex
 import ssz
 
-from eth2._utils.hash import hash_eth2
 from eth2.beacon.genesis import initialize_beacon_state_from_eth1
-from eth2.beacon.tools.builder.initializer import create_mock_deposits_and_root
+from eth2.beacon.tools.builder.initializer import (
+    create_keypair_and_mock_withdraw_credentials,
+    create_mock_deposits_and_root,
+)
 from eth2.beacon.tools.fixtures.config_types import Minimal
 from eth2.beacon.tools.fixtures.loading import load_config_at_path, load_yaml_at
 from eth2.beacon.tools.misc.ssz_vector import override_lengths
@@ -30,22 +31,10 @@ def _main():
 
     key_set = load_yaml_at(KEY_DIR / KEY_SET_FILE)
 
-    pubkeys = ()
-    privkeys = ()
-    withdrawal_credentials = ()
-    keymap = {}
-    for key_pair in key_set:
-        pubkey = decode_hex(key_pair["pubkey"])
-        privkey = int.from_bytes(decode_hex(key_pair["privkey"]), "big")
-        withdrawal_credential = (
-            config.BLS_WITHDRAWAL_PREFIX.to_bytes(1, byteorder="big")
-            + hash_eth2(pubkey)[1:]
-        )
-
-        pubkeys += (pubkey,)
-        privkeys += (privkey,)
-        withdrawal_credentials += (withdrawal_credential,)
-        keymap[pubkey] = privkey
+    pubkeys, privkeys, withdrawal_credentials = create_keypair_and_mock_withdraw_credentials(
+        config, key_set
+    )
+    keymap = {pubkey: privkey for pubkey, privkey in zip(pubkeys, privkeys)}
 
     deposits, _ = create_mock_deposits_and_root(
         pubkeys, keymap, config, withdrawal_credentials

eth2/beacon/tools/builder/initializer.py

@@ -1,7 +1,8 @@
-from typing import Dict, Sequence, Tuple, Type
+from typing import Any, Dict, Sequence, Tuple, Type
 
 from eth.constants import ZERO_HASH32
 from eth_typing import BLSPubkey, Hash32
+from eth_utils import decode_hex
 from py_ecc.optimized_bls12_381.optimized_curve import (
     curve_order as BLS12_381_CURVE_ORDER,
 )
@@ -32,6 +33,27 @@ def generate_privkey_from_index(index: int) -> int:
     )
 
 
+def create_keypair_and_mock_withdraw_credentials(
+    config: Eth2Config, key_set: Sequence[Dict[str, Any]]
+) -> Tuple[Tuple[BLSPubkey, ...], Tuple[int, ...], Tuple[Hash32, ...]]:
+    pubkeys: Tuple[BLSPubkey, ...] = ()
+    privkeys: Tuple[int, ...] = ()
+    withdrawal_credentials: Tuple[Hash32, ...] = ()
+    for key_pair in key_set:
+        pubkey = BLSPubkey(decode_hex(key_pair["pubkey"]))
+        privkey = int.from_bytes(decode_hex(key_pair["privkey"]), "big")
+        withdrawal_credential = Hash32(
+            config.BLS_WITHDRAWAL_PREFIX.to_bytes(1, byteorder="big")
+            + hash_eth2(pubkey)[1:]
+        )
+
+        pubkeys += (pubkey,)
+        privkeys += (privkey,)
+        withdrawal_credentials += (withdrawal_credential,)
+
+    return (pubkeys, privkeys, withdrawal_credentials)
+
+
 def create_mock_deposits_and_root(
     pubkeys: Sequence[BLSPubkey],
     keymap: Dict[BLSPubkey, int],

tests-trio/eth1-monitor/conftest.py

@@ -20,6 +20,7 @@
 
 from trinity.components.eth2.eth1_monitor.configs import deposit_contract_json
 from trinity.components.eth2.eth1_monitor.eth1_monitor import Eth1Monitor
+from trinity.components.eth2.eth1_monitor.eth1_data_provider import Web3Eth1DataProvider
 from trinity.components.eth2.eth1_monitor.factories import DepositDataFactory
 from trinity.tools.factories.db import AtomicDBFactory
 
@@ -82,19 +83,25 @@ def func_do_deposit(w3, deposit_contract):
     return functools.partial(deposit, w3=w3, deposit_contract=deposit_contract)
 
 
+@pytest.fixture
+async def eth1_data_provider(w3, deposit_contract):
+    return Web3Eth1DataProvider(
+        w3=w3,
+        deposit_contract_address=deposit_contract.address,
+        deposit_contract_abi=deposit_contract.abi,
+    )
+
+
 @pytest.fixture
 async def eth1_monitor(
-    w3,
-    deposit_contract,
+    eth1_data_provider,
     num_blocks_confirmed,
     polling_period,
     endpoint_server,
     start_block_number,
 ):
     m = Eth1Monitor(
-        w3=w3,
-        deposit_contract_address=deposit_contract.address,
-        deposit_contract_abi=deposit_contract.abi,
+        eth1_data_provider=eth1_data_provider,
         num_blocks_confirmed=num_blocks_confirmed,
         polling_period=polling_period,
         start_block_number=start_block_number,

tests-trio/eth1-monitor/test_eth1_monitor.py

@@ -10,6 +10,7 @@
 from eth2._utils.merkle.common import verify_merkle_branch
 from trinity.components.eth2.eth1_monitor.eth1_monitor import (
     make_deposit_tree_and_root,
+    GetDistanceRequest,
     GetEth1DataRequest,
     GetDepositRequest,
     Eth1Monitor,
@@ -28,8 +29,7 @@
 
 @pytest.mark.trio
 async def test_logs_handling(
-    w3,
-    deposit_contract,
+    eth1_data_provider,
     tester,
     num_blocks_confirmed,
     polling_period,
@@ -40,9 +40,7 @@ async def test_logs_handling(
     amount_0 = func_do_deposit()
     amount_1 = func_do_deposit()
     m = Eth1Monitor(
-        w3=w3,
-        deposit_contract_address=deposit_contract.address,
-        deposit_contract_abi=deposit_contract.abi,
+        eth1_data_provider=eth1_data_provider,
         num_blocks_confirmed=num_blocks_confirmed,
         polling_period=polling_period,
         start_block_number=start_block_number,
@@ -280,6 +278,8 @@ async def test_ipc(
     async def request(request_type, **kwargs):
         await endpoint_client.wait_until_any_endpoint_subscribed_to(request_type)
         resp = await endpoint_client.request(request_type(**kwargs), broadcast_config)
+        if request_type is GetDistanceRequest:
+            return resp
         return resp.to_data()
 
     # Result from IPC should be the same as the direct call with the same args.
@@ -311,3 +311,27 @@ async def request(request_type, **kwargs):
     }
     with pytest.raises(Eth1MonitorValidationError):
         await request(GetEth1DataRequest, **get_eth1_data_kwargs_fails)
+
+    # Test: `get_distance`
+    # Fast forward for a few blocks
+    tester.mine_blocks(5)
+    latest_block = w3.eth.getBlock("latest")
+    latest_confirmed_block_number = latest_block.number - num_blocks_confirmed
+    latest_confirmed_block = w3.eth.getBlock(latest_confirmed_block_number)
+    eth1_voting_period_start_timestamp = latest_confirmed_block["timestamp"]
+    for distance in range(latest_confirmed_block_number):
+        block = w3.eth.getBlock(latest_confirmed_block_number - distance)
+        get_distance_kwargs = {
+            "block_hash": block["hash"],
+            "eth1_voting_period_start_timestamp": eth1_voting_period_start_timestamp,
+        }
+        resp = await request(GetDistanceRequest, **get_distance_kwargs)
+        assert distance == resp.distance
+        assert resp.error is None
+    # Fails
+    get_distance_fail_kwargs = {
+        "block_hash": b'\x12' * 32,
+        "eth1_voting_period_start_timestamp": eth1_voting_period_start_timestamp,
+    }
+    resp = await request(GetDistanceRequest, **get_distance_fail_kwargs)
+    assert resp.error is not None

trinity/components/eth2/beacon/validator.py

@@ -80,6 +80,10 @@
 GetReadyAttestationsFn = Callable[[Slot], Sequence[Attestation]]
 
 
+# FIXME: Read this from validator config
+ETH1_FOLLOW_DISTANCE = 16
+
+
 class Validator(BaseService):
     chain: BaseBeaconChain
     p2p_node: Node

trinity/components/eth2/eth1_monitor/component.py

@@ -0,0 +1,128 @@
+from argparse import (
+    ArgumentParser,
+    _SubParsersAction,
+)
+import json
+from pathlib import Path
+import time
+
+from async_service import Service, TrioManager
+
+from eth_typing import BlockNumber
+
+from lahja import EndpointAPI
+
+# from web3 import Web3
+
+from eth2.beacon.tools.builder.initializer import (
+    create_keypair_and_mock_withdraw_credentials,
+)
+from eth2.beacon.tools.builder.validator import create_mock_deposit_data
+from eth2.beacon.tools.fixtures.loading import load_yaml_at
+from eth2.beacon.typing import Timestamp
+from trinity.boot_info import BootInfo
+from trinity.components.eth2.eth1_monitor.configs import deposit_contract_json
+from trinity.components.eth2.eth1_monitor.eth1_data_provider import FakeEth1DataProvider
+from trinity.config import BeaconAppConfig
+from trinity.db.manager import DBClient
+from trinity.events import ShutdownRequest
+from trinity.extensibility import (
+    TrioIsolatedComponent,
+)
+
+from .eth1_monitor import Eth1Monitor
+
+
+# Fake eth1 monitor config
+# TODO: These configs should be read from a config file, e.g., `eth1_monitor_config.yaml`.
+DEPOSIT_CONTRACT_ABI = json.loads(deposit_contract_json)["abi"]
+DEPOSIT_CONTRACT_ADDRESS = b"\x12" * 20
+NUM_BLOCKS_CONFIRMED = 100
+POLLING_PERIOD = 10
+START_BLOCK_NUMBER = BlockNumber(1000)
+
+# Configs for fake Eth1DataProvider
+NUM_DEPOSITS_PER_BLOCK = 0
+START_BLOCK_TIMESTAMP = Timestamp(int(time.time()) - 2100)  # Around 45 mins ago
+
+
+class Eth1MonitorComponent(TrioIsolatedComponent):
+
+    name = "Eth1 Monitor"
+    endpoint_name = "eth1-monitor"
+
+    @property
+    def is_enabled(self) -> bool:
+        return self._boot_info.trinity_config.has_app_config(BeaconAppConfig)
+
+    @classmethod
+    def configure_parser(cls,
+                         arg_parser: ArgumentParser,
+                         subparser: _SubParsersAction) -> None:
+        # TODO: For now we use fake eth1 monitor.
+        pass
+        # arg_parser.add_argument(
+        #     "--eth1client-rpc",
+        #     help="RPC HTTP endpoint of Eth1 client ",
+        # )
+
+    @classmethod
+    async def do_run(cls, boot_info: BootInfo, event_bus: EndpointAPI) -> None:
+        trinity_config = boot_info.trinity_config
+        beacon_app_config = trinity_config.get_app_config(BeaconAppConfig)
+        chain_config = beacon_app_config.get_chain_config()
+        base_db = DBClient.connect(trinity_config.database_ipc_path)
+
+        # TODO: For now we use fake eth1 monitor.
+        # if boot_info.args.eth1client_rpc:
+        #     w3: Web3 = Web3.HTTPProvider(boot_info.args.eth1client_rpc)
+        # else:
+        #     w3: Web3 = None
+
+        # TODO: For now we use fake eth1 monitor. So we load validators data from
+        # interop setting and hardcode the deposit data into fake eth1 data provider.
+        chain = chain_config.beacon_chain_class(
+            base_db,
+            chain_config.genesis_config,
+        )
+        config = chain.get_state_machine().config
+        key_set = load_yaml_at(
+            Path('eth2/beacon/scripts/quickstart_state/keygen_16_validators.yaml')
+        )
+        pubkeys, privkeys, withdrawal_credentials = create_keypair_and_mock_withdraw_credentials(
+            config,
+            key_set,  # type: ignore
+        )
+        initial_deposits = (
+            create_mock_deposit_data(
+                config=config,
+                pubkey=pubkey,
+                privkey=privkey,
+                withdrawal_credentials=withdrawal_credential,
+            )
+            for pubkey, privkey, withdrawal_credential in zip(
+                pubkeys, privkeys, withdrawal_credentials)
+        )
+
+        with base_db:
+            fake_eth1_data_provider = FakeEth1DataProvider(
+                start_block_number=START_BLOCK_NUMBER,
+                start_block_timestamp=START_BLOCK_TIMESTAMP,
+                num_deposits_per_block=NUM_DEPOSITS_PER_BLOCK,
+                initial_deposits=tuple(initial_deposits),
+            )
+
+            eth1_monitor_service: Service = Eth1Monitor(
+                eth1_data_provider=fake_eth1_data_provider,
+                num_blocks_confirmed=NUM_BLOCKS_CONFIRMED,
+                polling_period=POLLING_PERIOD,
+                start_block_number=START_BLOCK_NUMBER,
+                event_bus=event_bus,
+                base_db=base_db,
+            )
+
+            try:
+                await TrioManager.run_service(eth1_monitor_service)
+            except Exception:
+                await event_bus.broadcast(ShutdownRequest("Eth1 Monitor ended unexpectedly"))
+                raise

trinity/components/eth2/eth1_monitor/db.py

@@ -114,16 +114,17 @@ def __init__(
     ) -> None:
         self.db = db
         self._deposit_count = self._get_deposit_count()
-        # If the parameter `highest_processed_block_number` is given, set it in the database.
+        latest_processed_block_number = self._get_highest_processed_block_number()
         if highest_processed_block_number is not None:
             if highest_processed_block_number < 0:
                 raise DepositDataDBValidationError(
                     "`highest_processed_block_number` should be non-negative: "
                     f"highest_processed_block_number={highest_processed_block_number}"
                 )
-            self._set_highest_processed_block_number(
-                self.db, highest_processed_block_number
-            )
+            elif highest_processed_block_number > latest_processed_block_number:
+                self._set_highest_processed_block_number(
+                    self.db, highest_processed_block_number
+                )
         self._highest_processed_block_number = (
             self._get_highest_processed_block_number()
         )