allow driving EL with LC (#3865)

Adds the `--web3-url` launch argument to `nimbus_light_client` to enable
driving the EL with the optimistic head obtained from LC sync protocol.
This will keep issuing `newPayload` / `forkChoiceUpdated` requests for
new blocks, marking them as optimistic. `ZERO_HASH` is reported as the
finalized block for now.

Makefile

@@ -361,6 +361,7 @@ define CONNECT_TO_NETWORK_WITH_LIGHT_CLIENT
 		--network=$(1) \
 		--log-level="$(RUNTIME_LOG_LEVEL)" \
 		--log-file=build/data/shared_$(1)_$(NODE_ID)/nbc_lc_$$(date +"%Y%m%d%H%M%S").log \
+		--data-dir=build/data/shared_$(1)_$(NODE_ID) \
 		--trusted-block-root="$(LC_TRUSTED_BLOCK_ROOT)"
 endef
 

beacon_chain/conf.nim

@@ -150,7 +150,7 @@ type
       name: "wallets-dir" .}: Option[InputDir]
 
     web3Urls* {.
-      desc: "One or more Web3 provider URLs used for obtaining deposit contract data"
+      desc: "One or more execution layer Web3 provider URLs"
       name: "web3-url" .}: seq[string]
 
     web3ForcePolling* {.

beacon_chain/conf_light_client.nim

@@ -39,6 +39,14 @@ type LightClientConf* = object
     desc: "Specifies a path for the written Json log file (deprecated)"
     name: "log-file" .}: Option[OutFile]
 
+  # Storage
+  dataDir* {.
+    desc: "The directory where nimbus will store all blockchain data"
+    defaultValue: config.defaultDataDir()
+    defaultValueDesc: ""
+    abbr: "d"
+    name: "data-dir" .}: OutDir
+
   # Network
   eth2Network* {.
     desc: "The Eth2 network to join"
@@ -116,9 +124,30 @@ type LightClientConf* = object
     desc: "Recent trusted finalized block root to initialize light client from"
     name: "trusted-block-root" .}: Eth2Digest
 
+  # Execution layer
+  web3Urls* {.
+    desc: "One or more execution layer Web3 provider URLs"
+    name: "web3-url" .}: seq[string]
+
+  jwtSecret* {.
+    desc: "A file containing the hex-encoded 256 bit secret key to be used for verifying/generating jwt tokens"
+    name: "jwt-secret" .}: Option[string]
+
+  safeSlotsToImportOptimistically* {.
+    hidden
+    desc: "Modify SAFE_SLOTS_TO_IMPORT_OPTIMISTICALLY"
+    defaultValue: 128
+    name: "safe-slots-to-import-optimistically" .}: uint16
+
   # Testing
   stopAtEpoch* {.
     hidden
     desc: "The wall-time epoch at which to exit the program. (for testing purposes)"
     defaultValue: 0
     name: "stop-at-epoch" .}: uint64
+
+template loadJwtSecret*(
+    rng: var HmacDrbgContext,
+    config: LightClientConf,
+    allowCreate: bool): Option[seq[byte]] =
+  rng.loadJwtSecret(string(config.dataDir), config.jwtSecret, allowCreate)

beacon_chain/consensus_object_pools/consensus_manager.nim

@@ -84,7 +84,7 @@ from web3/engine_api_types import
 
 func `$`(h: BlockHash): string = $h.asEth2Digest
 
-proc runForkchoiceUpdated(
+proc runForkchoiceUpdated*(
     eth1Monitor: Eth1Monitor, headBlockRoot, finalizedBlockRoot: Eth2Digest):
     Future[PayloadExecutionStatus] {.async.} =
   # Allow finalizedBlockRoot to be 0 to avoid sync deadlocks.

beacon_chain/eth1/eth1_monitor.nim

@@ -941,7 +941,11 @@ template getOrDefault[T, E](r: Result[T, E]): T =
 
 proc init*(T: type Eth1Chain, cfg: RuntimeConfig, db: BeaconChainDB): T =
   let
-    finalizedDeposits = db.getEth2FinalizedTo().getOrDefault()
+    finalizedDeposits =
+      if db != nil:
+        db.getEth2FinalizedTo().getOrDefault()
+      else:
+        default(DepositContractSnapshot)
     m = DepositsMerkleizer.init(finalizedDeposits.depositContractState)
 
   T(db: db,

beacon_chain/nimbus_light_client.nim

@@ -10,37 +10,15 @@
 
 import
   std/os,
-  chronicles, chronos,
+  chronicles, chronicles/chronos_tools, chronos,
   eth/keys,
+  ./eth1/eth1_monitor,
   ./spec/beaconstate,
+  ./sync/optimistic_sync_light_client,
   "."/[light_client, nimbus_binary_common, version]
 
-proc onFinalizedHeader(lightClient: LightClient) =
+from ./consensus_object_pools/consensus_manager import runForkchoiceUpdated
-  notice "New LC finalized header",
+from ./gossip_processing/block_processor import newExecutionPayload
-    finalized_header = shortLog(lightClient.finalizedHeader.get)
-
-proc onOptimisticHeader(lightClient: LightClient) =
-  notice "New LC optimistic header",
-    optimistic_header = shortLog(lightClient.optimisticHeader.get)
-
-proc onSecond(
-    lightClient: LightClient,
-    config: LightClientConf,
-    getBeaconTime: GetBeaconTimeFn) =
-  ## This procedure will be called once per second.
-  let wallSlot = getBeaconTime().slotOrZero()
-  if checkIfShouldStopAtEpoch(wallSlot, config.stopAtEpoch):
-    quit(0)
-
-  lightClient.updateGossipStatus(wallSlot + 1)
-
-proc runOnSecondLoop(
-    lightClient: LightClient,
-    config: LightClientConf,
-    getBeaconTime: GetBeaconTimeFn) {.async.} =
-  while true:
-    onSecond(lightClient, config, getBeaconTime)
-    await chronos.sleepAsync(chronos.seconds(1))
 
 programMain:
   var config = makeBannerAndConfig(
@@ -80,6 +58,40 @@ programMain:
       rng, config, netKeys, cfg,
       forkDigests, getBeaconTime, genesis_validators_root)
 
+    eth1Monitor =
+      if config.web3Urls.len > 0:
+        Eth1Monitor.init(
+          cfg, db = nil, getBeaconTime, config.web3Urls,
+          none(DepositContractSnapshot), metadata.eth1Network,
+          forcePolling = false,
+          rng[].loadJwtSecret(config, allowCreate = false))
+      else:
+        nil
+
+    optimisticProcessor = proc(signedBlock: ForkedMsgTrustedSignedBeaconBlock):
+        Future[void] {.async.} =
+      debug "New LC optimistic block",
+        opt = signedBlock.toBlockId(),
+        wallSlot = getBeaconTime().slotOrZero
+      withBlck(signedBlock):
+        when stateFork >= BeaconStateFork.Bellatrix:
+          if blck.message.is_execution_block:
+            await eth1Monitor.ensureDataProvider()
+
+            # engine_newPayloadV1
+            template payload(): auto = blck.message.body.execution_payload
+            discard await eth1Monitor.newExecutionPayload(payload)
+
+            # engine_forkchoiceUpdatedV1
+            discard await eth1Monitor.runForkchoiceUpdated(
+              headBlockRoot = payload.block_hash,
+              finalizedBlockRoot = ZERO_HASH)
+        else: discard
+      return
+    optSync = initLCOptimisticSync(
+      network, getBeaconTime, optimisticProcessor,
+      config.safeSlotsToImportOptimistically)
+
     lightClient = createLightClient(
       network, rng, config, cfg,
       forkDigests, getBeaconTime, genesis_validators_root)
@@ -90,11 +102,73 @@ programMain:
   waitFor network.startListening()
   waitFor network.start()
 
+  proc shouldSyncOptimistically(slot: Slot): bool =
+    const
+      # Maximum age of light client optimistic header to use optimistic sync
+      maxAge = 2 * SLOTS_PER_EPOCH
+
+    if eth1Monitor == nil:
+      false
+    elif getBeaconTime().slotOrZero > slot + maxAge:
+      false
+    else:
+      true
+
+  proc onFinalizedHeader(lightClient: LightClient) =
+    notice "New LC finalized header",
+      finalized_header = shortLog(lightClient.finalizedHeader.get)
+    let optimisticHeader = lightClient.optimisticHeader.valueOr:
+      return
+    if not shouldSyncOptimistically(optimisticHeader.slot):
+      return
+    let finalizedHeader = lightClient.finalizedHeader.valueOr:
+      return
+    optSync.setOptimisticHeader(optimisticHeader)
+    optSync.setFinalizedHeader(finalizedHeader)
+
+  proc onOptimisticHeader(lightClient: LightClient) =
+    notice "New LC optimistic header",
+      optimistic_header = shortLog(lightClient.optimisticHeader.get)
+    let optimisticHeader = lightClient.optimisticHeader.valueOr:
+      return
+    if not shouldSyncOptimistically(optimisticHeader.slot):
+      return
+    optSync.setOptimisticHeader(optimisticHeader)
+
   lightClient.onFinalizedHeader = onFinalizedHeader
   lightClient.onOptimisticHeader = onOptimisticHeader
   lightClient.trustedBlockRoot = some config.trustedBlockRoot
+
+  var nextExchangeTransitionConfTime: Moment
+
+  proc onSecond(time: Moment) =
+    # engine_exchangeTransitionConfigurationV1
+    if time > nextExchangeTransitionConfTime and eth1Monitor != nil:
+      nextExchangeTransitionConfTime = time + chronos.minutes(1)
+      traceAsyncErrors eth1Monitor.exchangeTransitionConfiguration()
+
+    let wallSlot = getBeaconTime().slotOrZero()
+    if checkIfShouldStopAtEpoch(wallSlot, config.stopAtEpoch):
+      quit(0)
+
+    lightClient.updateGossipStatus(wallSlot + 1)
+
+  proc runOnSecondLoop() {.async.} =
+    let sleepTime = chronos.seconds(1)
+    while true:
+      let start = chronos.now(chronos.Moment)
+      await chronos.sleepAsync(sleepTime)
+      let afterSleep = chronos.now(chronos.Moment)
+      let sleepTime = afterSleep - start
+      onSecond(start)
+      let finished = chronos.now(chronos.Moment)
+      let processingTime = finished - afterSleep
+      trace "onSecond task completed", sleepTime, processingTime
+
+  onSecond(Moment.now())
+  optSync.start()
   lightClient.start()
 
-  asyncSpawn runOnSecondLoop(lightClient, config, getBeaconTime)
+  asyncSpawn runOnSecondLoop()
   while true:
     poll()

beacon_chain/sync/optimistic_sync_light_client.nim

@@ -54,10 +54,7 @@ proc reportOptimisticCandidateBlock(optSync: LCOptimisticSync) {.gcsafe.} =
     if finalizedBlock.isOk:
       optSync.finalizedIsExecutionBlock =
         withBlck(finalizedBlock.get):
-          when stateFork >= BeaconStateFork.Bellatrix:
           some blck.message.is_execution_block()
-          else:
-            some false
 
   let
     currentSlot = optSync.lcBlocks.getHeadSlot()

beacon_chain/sync/sync_protocol.nim

@@ -676,7 +676,7 @@ proc useSyncV2*(state: BeaconSyncNetworkState): bool =
   let
     wallTimeSlot = state.getBeaconTime().slotOrZero
 
-  wallTimeSlot.epoch >= state.dag.cfg.ALTAIR_FORK_EPOCH
+  wallTimeSlot.epoch >= state.cfg.ALTAIR_FORK_EPOCH
 
 proc useSyncV2*(peer: Peer): bool =
   peer.networkState(BeaconSync).useSyncV2()

scripts/geth_vars.sh

@@ -8,6 +8,5 @@ GETH_HTTP_BASE_PORT="${GETH_HTTP_BASE_PORT:-8545}"
 GETH_WS_BASE_PORT="${GETH_WS_BASE_PORT:-8546}"
 GETH_AUTH_RPC_PORT_BASE="${GETH_AUTH_RPC_PORT_BASE:-8551}"
 PORT_OFFSET="${PORT_OFFSET:-100}"
-GENESISJSON="${GENESISJSON:-${BASEDIR}/scripts/geth_genesis.json}"
+GENESISJSON="${GENESISJSON:-${BASEDIR}/geth_genesis.json}"
 DISCOVER="--nodiscover"
-

scripts/launch_local_testnet.sh

@@ -428,8 +428,8 @@ download_geth() {
   fi
 }
 
-GETH_NUM_NODES="${NUM_NODES}"
+GETH_NUM_NODES="$(( NUM_NODES + LC_NODES ))"
-NIMBUSEL_NUM_NODES="${NUM_NODES}"
+NIMBUSEL_NUM_NODES="$(( NUM_NODES + LC_NODES ))"
 
 if [[ "${RUN_GETH}" == "1" ]]; then
   if [[ ! -e "${GETH_BINARY}" ]]; then
@@ -802,6 +802,11 @@ for NUM_NODE in $(seq 0 $(( NUM_NODES - 1 ))); do
     fi
   fi
 done
+for NUM_LC in $(seq 0 $(( LC_NODES - 1 ))); do
+  LC_DATA_DIR="${DATA_DIR}/lc${NUM_LC}"
+  rm -rf "${LC_DATA_DIR}"
+  scripts/makedir.sh "${LC_DATA_DIR}" 2>&1
+done
 
 CLI_CONF_FILE="$CONTAINER_DATA_DIR/config.toml"
 
@@ -958,9 +963,19 @@ if [ "$LC_NODES" -ge "1" ]; then
     "${CURL_BINARY}" -s "http://localhost:${BASE_REST_PORT}/eth/v1/beacon/headers/finalized" | \
       "${JQ_BINARY}" -r '.data.root')"
   for NUM_LC in $(seq 0 $(( LC_NODES - 1 ))); do
+    LC_DATA_DIR="${DATA_DIR}/lc${NUM_LC}"
+
+    if [ ${#EL_RPC_PORTS[@]} -eq 0 ]; then # check if the array is empty
+      WEB3_ARG=""
+    else
+      WEB3_ARG="--web3-url=http://127.0.0.1:${EL_RPC_PORTS[$(( NUM_NODES + NUM_LC ))]}"
+    fi
+
+    # TODO re-add --jwt-secret
     ./build/nimbus_light_client \
       --log-level="${LOG_LEVEL}" \
       --log-format="json" \
+      --data-dir="${LC_DATA_DIR}" \
       --network="${CONTAINER_DATA_DIR}" \
       --bootstrap-node="${LC_BOOTSTRAP_NODE}" \
       --tcp-port=$(( BASE_PORT + NUM_NODES + NUM_LC )) \
@@ -968,6 +983,7 @@ if [ "$LC_NODES" -ge "1" ]; then
       --max-peers=$(( NUM_NODES + LC_NODES - 1 )) \
       --nat="extip:127.0.0.1" \
       --trusted-block-root="${LC_TRUSTED_BLOCK_ROOT}" \
+      ${WEB3_ARG} \
       ${STOP_AT_EPOCH_FLAG} \
       &> "${DATA_DIR}/log_lc${NUM_LC}.txt" &
     PIDS="${PIDS},$!"

scripts/test_merge_node.nim

@@ -22,7 +22,6 @@ from web3/engine_api_types import PayloadExecutionStatus
 from ../beacon_chain/networking/network_metadata import Eth1Network
 from ../beacon_chain/spec/datatypes/base import ZERO_HASH
 from ../beacon_chain/spec/presets import Eth1Address, defaultRuntimeConfig
-from ../tests/testdbutil import makeTestDB
 
 # TODO factor this out and have a version with the result of the jwt secret
 # slurp for testing purposes
@@ -54,9 +53,8 @@ proc run() {.async.} =
     echo "args are: web3url jwtsecretfilename"
 
   let
-    db = makeTestDB(64)
     eth1Monitor = Eth1Monitor.init(
-      defaultRuntimeConfig, db, nil, @[paramStr(1)],
+      defaultRuntimeConfig, db = nil, nil, @[paramStr(1)],
       none(DepositContractSnapshot), none(Eth1Network), false,
       some readJwtSecret(paramStr(2)).get)
 

scripts/test_merge_vectors.nim

@@ -19,7 +19,6 @@ from web3/engine_api_types import PayloadExecutionStatus
 from ../beacon_chain/networking/network_metadata import Eth1Network
 from ../beacon_chain/spec/datatypes/base import ZERO_HASH
 from ../beacon_chain/spec/presets import Eth1Address, defaultRuntimeConfig
-from ../tests/testdbutil import makeTestDB
 
 {.push raises: [Defect].}
 
@@ -55,10 +54,9 @@ const
 
 proc run() {.async.} =
   let
-    db = makeTestDB(64)
     jwtSecret = some readJwtSecret("jwt.hex").get
     eth1Monitor = Eth1Monitor.init(
-      defaultRuntimeConfig, db, nil, @[web3Url],
+      defaultRuntimeConfig, db = nil, nil, @[web3Url],
       none(DepositContractSnapshot), none(Eth1Network), false, jwtSecret)
     web3Provider = (await Web3DataProvider.new(
       default(Eth1Address), web3Url, jwtSecret)).get