remove `newPayload` from block production flow (#4186)

* remove `newPayload` from block production flow

* refactor block_processor to run `newPayload` as part of `storeBlock`

beacon_chain/gossip_processing/block_processor.nim

@@ -85,6 +85,16 @@ type
 
     verifier: BatchVerifier
 
+  NewPayloadStatus {.pure.} = enum
+    valid
+    notValid
+    invalid
+    noResponse
+
+  BlockProcessingCompleted {.pure.} = enum
+    completed
+    notCompleted
+
 proc addBlock*(
     self: var BlockProcessor, src: MsgSource, blck: ForkedSignedBeaconBlock,
     resfut: Future[Result[void, BlockError]] = nil,
@@ -191,13 +201,108 @@ from ../consensus_object_pools/blockchain_dag import
 from ../consensus_object_pools/block_dag import shortLog
 from ../consensus_object_pools/spec_cache import get_attesting_indices
 from ../spec/datatypes/phase0 import TrustedSignedBeaconBlock
+from ../spec/datatypes/altair import SignedBeaconBlock
+from ../spec/datatypes/bellatrix import SignedBeaconBlock
+
+from eth/async_utils import awaitWithTimeout
+from ../spec/datatypes/bellatrix import ExecutionPayload, SignedBeaconBlock
+
+proc newExecutionPayload*(
+    eth1Monitor: Eth1Monitor, executionPayload: bellatrix.ExecutionPayload):
+    Future[Opt[PayloadExecutionStatus]] {.async.} =
+  if eth1Monitor.isNil:
+    warn "newPayload: attempting to process execution payload without Eth1Monitor. Ensure --web3-url setting is correct and JWT is configured."
+    return Opt.none PayloadExecutionStatus
+
+  debug "newPayload: inserting block into execution engine",
+    parentHash = executionPayload.parent_hash,
+    blockHash = executionPayload.block_hash,
+    stateRoot = shortLog(executionPayload.state_root),
+    receiptsRoot = shortLog(executionPayload.receipts_root),
+    prevRandao = shortLog(executionPayload.prev_randao),
+    blockNumber = executionPayload.block_number,
+    gasLimit = executionPayload.gas_limit,
+    gasUsed = executionPayload.gas_used,
+    timestamp = executionPayload.timestamp,
+    extraDataLen = executionPayload.extra_data.len,
+    baseFeePerGas = $executionPayload.base_fee_per_gas,
+    numTransactions = executionPayload.transactions.len
+
+  # https://github.com/ethereum/execution-apis/blob/v1.0.0-beta.1/src/engine/specification.md#request
+  const NEWPAYLOAD_TIMEOUT = 8.seconds
+
+  try:
+    let
+      payloadResponse =
+        awaitWithTimeout(
+            eth1Monitor.newPayload(
+              executionPayload.asEngineExecutionPayload),
+            NEWPAYLOAD_TIMEOUT):
+          info "newPayload: newPayload timed out"
+          return Opt.none PayloadExecutionStatus
+
+          # Placeholder for type system
+          PayloadStatusV1(status: PayloadExecutionStatus.syncing)
+
+      payloadStatus = payloadResponse.status
+
+    debug "newPayload: succeeded",
+      parentHash = executionPayload.parent_hash,
+      blockHash = executionPayload.block_hash,
+      blockNumber = executionPayload.block_number,
+      payloadStatus
+
+    return Opt.some payloadStatus
+  except CatchableError as err:
+    error "newPayload failed", msg = err.msg
+    return Opt.none PayloadExecutionStatus
+
+proc getExecutionValidity(
+    eth1Monitor: Eth1Monitor,
+    blck: phase0.SignedBeaconBlock | altair.SignedBeaconBlock):
+    Future[NewPayloadStatus] {.async.} =
+  return NewPayloadStatus.valid   # vacuously
+
+proc getExecutionValidity(
+    eth1Monitor: Eth1Monitor, blck: bellatrix.SignedBeaconBlock):
+    Future[NewPayloadStatus] {.async.} =
+  # Eth1 syncing is asynchronous from this
+  # TODO self.consensusManager.eth1Monitor.ttdReached
+  # should gate this when it works more reliably
+  # TODO detect have-TTD-but-not-is_execution_block case, and where
+  # execution payload was non-zero when TTD detection more reliable
+
+  if not blck.message.is_execution_block:
+    return NewPayloadStatus.valid  # vacuously
+
+  try:
+    # Minimize window for Eth1 monitor to shut down connection
+    await eth1Monitor.ensureDataProvider()
+
+    let executionPayloadStatus = await newExecutionPayload(
+      eth1Monitor, blck.message.body.execution_payload)
+    if executionPayloadStatus.isNone:
+      return NewPayloadStatus.noResponse
+
+    case executionPayloadStatus.get
+      of PayloadExecutionStatus.invalid, PayloadExecutionStatus.invalid_block_hash:
+        debug "getExecutionValidity: execution payload invalid",
+          executionPayloadStatus = $executionPayloadStatus.get,
+          blck = shortLog(blck)
+        return NewPayloadStatus.invalid
+      of PayloadExecutionStatus.syncing, PayloadExecutionStatus.accepted:
+        return NewPayloadStatus.notValid
+      of PayloadExecutionStatus.valid:
+        return NewPayloadStatus.valid
+  except CatchableError as err:
+    error "getExecutionValidity: newPayload failed", err = err.msg
+    return NewPayloadStatus.noResponse
 
 proc storeBlock*(
-    self: var BlockProcessor,
-    src: MsgSource, wallTime: BeaconTime,
-    signedBlock: ForkySignedBeaconBlock, payloadValid: bool,
-    queueTick: Moment = Moment.now(),
-    validationDur = Duration()): Result[BlockRef, BlockError] =
+    self: ref BlockProcessor, src: MsgSource, wallTime: BeaconTime,
+    signedBlock: ForkySignedBeaconBlock, queueTick: Moment = Moment.now(),
+    validationDur = Duration()):
+    Future[Result[BlockRef, (BlockError, BlockProcessingCompleted)]] {.async.} =
   ## storeBlock is the main entry point for unvalidated blocks - all untrusted
   ## blocks, regardless of origin, pass through here. When storing a block,
   ## we will add it to the dag and pass it to all block consumers that need
@@ -207,10 +312,25 @@ proc storeBlock*(
     startTick = Moment.now()
     vm = self.validatorMonitor
     dag = self.consensusManager.dag
+    payloadStatus =
+      await self.consensusManager.eth1Monitor.getExecutionValidity(signedBlock)
+    payloadValid = payloadStatus == NewPayloadStatus.valid
 
   # The block is certainly not missing any more
   self.consensusManager.quarantine[].missing.del(signedBlock.root)
 
+  if NewPayloadStatus.invalid == payloadStatus:
+    self.consensusManager.quarantine[].addUnviable(signedBlock.root)
+    return err((BlockError.UnviableFork, BlockProcessingCompleted.completed))
+  elif NewPayloadStatus.noResponse == payloadStatus:
+    # Disallow the `MissingParent` from leaking to the sync/request managers
+    # as it will be descored. However sync and request managers interact via
+    # `processBlock` (indirectly). `validator_duties` does call `storeBlock`
+    # directly, so is exposed to this, but only cares about whether there is
+    # an error or not.
+    return err((
+      BlockError.MissingParent, BlockProcessingCompleted.notCompleted))
+
   # We'll also remove the block as an orphan: it's unlikely the parent is
   # missing if we get this far - should that be the case, the block will
   # be re-added later
@@ -240,7 +360,7 @@ proc storeBlock*(
             trustedBlock.message.slot, trustedBlock.root,
             forkyState.data.current_sync_committee.pubkeys.data[i])
 
-  self.dumpBlock(signedBlock, blck)
+  self[].dumpBlock(signedBlock, blck)
 
   # There can be a scenario where we receive a block we already received.
   # However this block was before the last finalized epoch and so its parent
@@ -254,7 +374,7 @@ proc storeBlock*(
         # this to the appropriate error so that sync etc doesn't retry the block
         self.consensusManager.quarantine[].addUnviable(signedBlock.root)
 
-        return err(BlockError.UnviableFork)
+        return err((BlockError.UnviableFork, BlockProcessingCompleted.completed))
 
       if not self.consensusManager.quarantine[].addOrphan(
           dag.finalizedHead.slot, ForkedSignedBeaconBlock.init(signedBlock)):
@@ -267,7 +387,7 @@ proc storeBlock*(
       self.consensusManager.quarantine[].addUnviable(signedBlock.root)
     else: discard
 
-    return blck
+    return err((blck.error, BlockProcessingCompleted.completed))
 
   let storeBlockTick = Moment.now()
 
@@ -363,9 +483,9 @@ proc storeBlock*(
 
   for quarantined in self.consensusManager.quarantine[].pop(blck.get().root):
     # Process the blocks that had the newly accepted block as parent
-    self.addBlock(MsgSource.gossip, quarantined)
+    self[].addBlock(MsgSource.gossip, quarantined)
 
-  return blck
+  return Result[BlockRef, (BlockError, BlockProcessingCompleted)].ok blck.get
 
 # Enqueue
 # ------------------------------------------------------------------------------
@@ -406,7 +526,7 @@ proc addBlock*(
 # ------------------------------------------------------------------------------
 
 proc processBlock(
-    self: var BlockProcessor, entry: BlockEntry, payloadValid: bool) =
+    self: ref BlockProcessor, entry: BlockEntry) {.async.} =
   logScope:
     blockRoot = shortLog(entry.blck.root)
 
@@ -419,64 +539,27 @@ proc processBlock(
     quit 1
 
   let res = withBlck(entry.blck):
-    self.storeBlock(
-      entry.src, wallTime, blck, payloadValid, entry.queueTick,
-      entry.validationDur)
+    await self.storeBlock(
+      entry.src, wallTime, blck, entry.queueTick, entry.validationDur)
+
+  if res.isErr and res.error[1] == BlockProcessingCompleted.notCompleted:
+    # When an execution engine returns an error or fails to respond to a
+    # payload validity request for some block, a consensus engine:
+    # - MUST NOT optimistically import the block.
+    # - MUST NOT apply the block to the fork choice store.
+    # - MAY queue the block for later processing.
+    # https://github.com/ethereum/consensus-specs/blob/v1.2.0/sync/optimistic.md#execution-engine-errors
+    await sleepAsync(chronos.seconds(1))
+    self[].addBlock(
+      entry.src, entry.blck, entry.resfut, entry.validationDur)
+
+    # To ensure backpressure on the sync manager, do not complete these futures.
+    return
 
   if entry.resfut != nil:
     entry.resfut.complete(
       if res.isOk(): Result[void, BlockError].ok()
-      else: Result[void, BlockError].err(res.error()))
-
-from eth/async_utils import awaitWithTimeout
-from ../spec/datatypes/bellatrix import ExecutionPayload, SignedBeaconBlock
-
-proc newExecutionPayload*(
-    eth1Monitor: Eth1Monitor, executionPayload: bellatrix.ExecutionPayload):
-    Future[Opt[PayloadExecutionStatus]] {.async.} =
-  if eth1Monitor.isNil:
-    warn "newPayload: attempting to process execution payload without Eth1Monitor. Ensure --web3-url setting is correct and JWT is configured."
-    return Opt.none PayloadExecutionStatus
-
-  debug "newPayload: inserting block into execution engine",
-    parentHash = executionPayload.parent_hash,
-    blockHash = executionPayload.block_hash,
-    stateRoot = shortLog(executionPayload.state_root),
-    receiptsRoot = shortLog(executionPayload.receipts_root),
-    prevRandao = shortLog(executionPayload.prev_randao),
-    blockNumber = executionPayload.block_number,
-    gasLimit = executionPayload.gas_limit,
-    gasUsed = executionPayload.gas_used,
-    timestamp = executionPayload.timestamp,
-    extraDataLen = executionPayload.extra_data.len,
-    baseFeePerGas = $executionPayload.base_fee_per_gas,
-    numTransactions = executionPayload.transactions.len
-
-  try:
-    let
-      payloadResponse =
-        awaitWithTimeout(
-            eth1Monitor.newPayload(
-              executionPayload.asEngineExecutionPayload),
-            NEWPAYLOAD_TIMEOUT):
-          info "newPayload: newPayload timed out"
-          return Opt.none PayloadExecutionStatus
-
-          # Placeholder for type system
-          PayloadStatusV1(status: PayloadExecutionStatus.syncing)
-
-      payloadStatus = payloadResponse.status
-
-    debug "newPayload: succeeded",
-      parentHash = executionPayload.parent_hash,
-      blockHash = executionPayload.block_hash,
-      blockNumber = executionPayload.block_number,
-      payloadStatus
-
-    return Opt.some payloadStatus
-  except CatchableError as err:
-    error "newPayload failed", msg = err.msg
-    return Opt.none PayloadExecutionStatus
+      else: Result[void, BlockError].err(res.error()[0]))
 
 proc runQueueProcessingLoop*(self: ref BlockProcessor) {.async.} =
   while true:
@@ -493,72 +576,4 @@ proc runQueueProcessingLoop*(self: ref BlockProcessor) {.async.} =
 
     discard await idleAsync().withTimeout(idleTimeout)
 
-    let
-      blck = await self[].blockQueue.popFirst()
-      hasExecutionPayload =
-        withBlck(blck.blck): blck.message.is_execution_block
-      executionPayloadStatus =
-       if hasExecutionPayload:
-         # Eth1 syncing is asynchronous from this
-         # TODO self.consensusManager.eth1Monitor.ttdReached
-         # should gate this when it works more reliably
-         # TODO detect have-TTD-but-not-is_execution_block case, and where
-         # execution payload was non-zero when TTD detection more reliable
-         when true:
-           # When an execution engine returns an error or fails to respond to a
-           # payload validity request for some block, a consensus engine:
-           # - MUST NOT optimistically import the block.
-           # - MUST NOT apply the block to the fork choice store.
-           # - MAY queue the block for later processing.
-           # https://github.com/ethereum/consensus-specs/blob/v1.2.0/sync/optimistic.md#execution-engine-errors
-           template reEnqueueBlock: untyped =
-             await sleepAsync(chronos.seconds(1))
-             self[].addBlock(
-               blck.src, blck.blck, blck.resfut, blck.validationDur)
-
-           try:
-             # Minimize window for Eth1 monitor to shut down connection
-             await self.consensusManager.eth1Monitor.ensureDataProvider()
-
-             let executionPayload =
-               withBlck(blck.blck):
-                 when stateFork >= BeaconStateFork.Bellatrix:
-                   blck.message.body.execution_payload
-                 else:
-                   doAssert false
-                   default(bellatrix.ExecutionPayload) # satisfy Nim
-
-             let executionPayloadStatus = await newExecutionPayload(
-               self.consensusManager.eth1Monitor, executionPayload)
-             if executionPayloadStatus.isNone:
-               reEnqueueBlock()
-               continue
-
-             executionPayloadStatus.get
-           except CatchableError as err:
-             error "runQueueProcessingLoop: newPayload failed", err = err.msg
-             reEnqueueBlock()
-             continue
-         else:
-           debug "runQueueProcessingLoop: got execution payload before TTD"
-           PayloadExecutionStatus.syncing
-       else:
-         # Vacuously
-         PayloadExecutionStatus.valid
-
-    if executionPayloadStatus in static([
-        PayloadExecutionStatus.invalid,
-        PayloadExecutionStatus.invalid_block_hash]):
-      debug "runQueueProcessingLoop: execution payload invalid",
-        executionPayloadStatus,
-        blck = shortLog(blck.blck)
-      self.consensusManager.quarantine[].addUnviable(blck.blck.root)
-      # Every loop iteration ends with some version of blck.resfut.complete(),
-      # including processBlock(), otherwise the sync manager stalls.
-      if not blck.resfut.isNil:
-        blck.resfut.complete(
-          Result[void, BlockError].err(BlockError.UnviableFork))
-    else:
-      self[].processBlock(
-        blck,
-        payloadValid = executionPayloadStatus == PayloadExecutionStatus.valid)
+    await self.processBlock(await self[].blockQueue.popFirst())

beacon_chain/spec/datatypes/bellatrix.nim

@@ -31,9 +31,6 @@ const
   # https://github.com/ethereum/execution-apis/blob/v1.0.0-beta.1/src/engine/specification.md#request-1
   FORKCHOICEUPDATED_TIMEOUT* = 8.seconds
 
-  # https://github.com/ethereum/execution-apis/blob/v1.0.0-beta.1/src/engine/specification.md#request
-  NEWPAYLOAD_TIMEOUT* = 8.seconds
-
 type
   # https://github.com/ethereum/consensus-specs/blob/v1.2.0/specs/bellatrix/beacon-chain.md#custom-types
   Transaction* = List[byte, Limit MAX_BYTES_PER_TRANSACTION]

beacon_chain/spec/helpers.nim

@@ -207,7 +207,7 @@ template is_sync_committee_update*(update: SomeLightClientUpdate): bool =
   else:
     false
 
-# https://github.com/ethereum/consensus-specs/blob/v1.2.0-rc.3/specs/altair/light-client/sync-protocol.md#is_finality_update
+# https://github.com/ethereum/consensus-specs/blob/v1.2.0/specs/altair/light-client/sync-protocol.md#is_finality_update
 template is_finality_update*(update: SomeLightClientUpdate): bool =
   when update is SomeLightClientUpdateWithFinality:
     not isZeroMemory(update.finality_branch)

beacon_chain/validators/message_router.nim

@@ -121,8 +121,8 @@ proc routeSignedBeaconBlock*(
       signature = shortLog(blck.signature), error = res.error()
 
   let
-    newBlockRef = router[].blockProcessor[].storeBlock(
-      MsgSource.api, sendTime, blck, true)
+    newBlockRef = await router[].blockProcessor.storeBlock(
+      MsgSource.api, sendTime, blck)
 
   # The boolean we return tells the caller whether the block was integrated
   # into the chain

beacon_chain/validators/validator_duties.nim

@@ -424,31 +424,17 @@ proc getExecutionPayload[T](
            latestHead, latestSafe, latestFinalized, timestamp, random,
            feeRecipient, node.consensusManager.eth1Monitor))
       payload = try:
-          awaitWithTimeout(
-            get_execution_payload(payload_id, node.consensusManager.eth1Monitor),
-            GETPAYLOAD_TIMEOUT):
-              beacon_block_payload_errors.inc()
-              warn "Getting execution payload from Engine API timed out", payload_id
-              empty_execution_payload
-        except CatchableError as err:
-          beacon_block_payload_errors.inc()
-          warn "Getting execution payload from Engine API failed",
-                payload_id, err = err.msg
-          empty_execution_payload
-
-      executionPayloadStatus =
         awaitWithTimeout(
-          node.consensusManager.eth1Monitor.newExecutionPayload(payload),
-          NEWPAYLOAD_TIMEOUT):
-            info "getExecutionPayload: newPayload timed out"
-            Opt.none PayloadExecutionStatus
-
-    if executionPayloadStatus.isNone or executionPayloadStatus.get in [
-        PayloadExecutionStatus.invalid,
-        PayloadExecutionStatus.invalid_block_hash]:
-      info "getExecutionPayload: newExecutionPayload invalid",
-        executionPayloadStatus
-      return Opt.none ExecutionPayload
+          get_execution_payload(payload_id, node.consensusManager.eth1Monitor),
+          GETPAYLOAD_TIMEOUT):
+            beacon_block_payload_errors.inc()
+            warn "Getting execution payload from Engine API timed out", payload_id
+            empty_execution_payload
+      except CatchableError as err:
+        beacon_block_payload_errors.inc()
+        warn "Getting execution payload from Engine API failed",
+              payload_id, err = err.msg
+        empty_execution_payload
 
     return Opt.some payload
   except CatchableError as err:

tests/test_block_processor.nim

@@ -21,6 +21,7 @@ import
   ../beacon_chain/eth1/eth1_monitor,
   ./testutil, ./testdbutil, ./testblockutil
 
+from chronos/unittest2/asynctests import asyncTest
 from ../beacon_chain/spec/eth2_apis/dynamic_fee_recipients import
   DynamicFeeRecipientsStore, init
 from ../beacon_chain/validators/action_tracker import ActionTracker
@@ -56,11 +57,10 @@ suite "Block processor" & preset():
         false, "", "", keys.newRng(), taskpool, consensusManager,
         validatorMonitor, getTimeFn)
 
-  test "Reverse order block add & get" & preset():
-    let missing = processor[].storeBlock(
-      MsgSource.gossip, b2.message.slot.start_beacon_time(), b2,
-      payloadValid = true)
-    check: missing.error == BlockError.MissingParent
+  asyncTest "Reverse order block add & get" & preset():
+    let missing = await processor.storeBlock(
+      MsgSource.gossip, b2.message.slot.start_beacon_time(), b2)
+    check: missing.error[0] == BlockError.MissingParent
 
     check:
       not dag.containsForkBlock(b2.root) # Unresolved, shouldn't show up
@@ -68,9 +68,8 @@ suite "Block processor" & preset():
       FetchRecord(root: b1.root) in quarantine[].checkMissing()
 
     let
-      status = processor[].storeBlock(
-        MsgSource.gossip, b2.message.slot.start_beacon_time(), b1,
-        payloadValid = true)
+      status = await processor.storeBlock(
+        MsgSource.gossip, b2.message.slot.start_beacon_time(), b1)
       b1Get = dag.getBlockRef(b1.root)
 
     check: