nimbus-eth2/beacon_chain/validator_client/block_service.nim

import ../spec/datatypes/[phase0, altair]
import common, api
import chronicles

logScope: service = "block_service"

proc publishBlock(vc: ValidatorClientRef, currentSlot, slot: Slot,
                  validator: AttachedValidator) {.async.} =
  let
    genesisRoot = vc.beaconGenesis.genesis_validators_root
    graffiti =
      if vc.config.graffiti.isSome():
        vc.config.graffiti.get()
      else:
        defaultGraffitiBytes()
    fork = vc.fork.get()

  debug "Publishing block", validator = shortLog(validator),
                            delay = vc.getDelay(ZeroDuration),
                            wall_slot = currentSlot,
                            genesis_root = genesisRoot,
                            graffiti = graffiti, fork = fork, slot = slot,
                            wall_slot = currentSlot
  try:
    let randaoReveal = await validator.genRandaoReveal(fork, genesisRoot, slot)
    let beaconBlock =
      try:
        await vc.produceBlock(slot, randaoReveal, graffiti)
      except ValidatorApiError:
        error "Unable to retrieve block data", slot = slot,
              wall_slot = currentSlot, validator = shortLog(validator)
        return
      except CatchableError as exc:
        error "An unexpected error occurred while getting block data",
              err_name = exc.name, err_msg = exc.msg
        return

    let blockRoot = hash_tree_root(beaconBlock)
    var signedBlock = phase0.SignedBeaconBlock(message: beaconBlock,
                                        root: hash_tree_root(beaconBlock))

    # TODO: signing_root is recomputed in signBlockProposal just after
    let signing_root = compute_block_root(fork, genesisRoot, slot,
                                          signedBlock.root)
    let notSlashable = vc.attachedValidators
      .slashingProtection
      .registerBlock(ValidatorIndex(signedBlock.message.proposer_index),
                     validator.pubKey, slot, signing_root)

    if notSlashable.isOk():
      let signature = await validator.signBlockProposal(fork, genesisRoot, slot,
                                                        blockRoot)
      let signedBlock =
        phase0.SignedBeaconBlock(message: beaconBlock, root: blockRoot,
                                          signature: signature)

      debug "Sending block", blck = shortLog(signedBlock.message),
            signature = shortLog(signature), blockRoot = shortLog(blockRoot),
            validator = shortLog(validator)

      let res =
        try:
          await vc.publishBlock(signedBlock)
        except ValidatorApiError:
          error "Unable to publish block", blck = shortLog(signedBlock.message),
                blockRoot = shortLog(blockRoot),
                validator = shortLog(validator),
                validator_index = validator.index.get(),
                wall_slot = currentSlot
          return
        except CatchableError as exc:
          error "An unexpected error occurred while publishing block",
                err_name = exc.name, err_msg = exc.msg
          return
      if res:
        notice "Block published", blck = shortLog(signedBlock.message),
               blockRoot = shortLog(blockRoot), validator = shortLog(validator),
               validator_index = validator.index.get()
      else:
        warn "Block was not accepted by beacon node",
             blck = shortLog(signedBlock.message),
             blockRoot = shortLog(blockRoot),
             validator = shortLog(validator),
             validator_index = validator.index.get(),
             wall_slot = currentSlot
    else:
      warn "Slashing protection activated for block proposal",
           blck = shortLog(beaconBlock), blockRoot = shortLog(blockRoot),
           validator = shortLog(validator),
           validator_index = validator.index.get(),
           wall_slot = currentSlot,
           existingProposal = notSlashable.error
  except CatchableError as exc:
    error "Unexpected error happens while proposing block",
          error_name = exc.name, error_msg = exc.msg

proc proposeBlock(vc: ValidatorClientRef, slot: Slot,
                  proposerKey: ValidatorPubkey) {.async.} =
  let (inFuture, timeToSleep) = vc.beaconClock.fromNow(slot)
  try:
    if inFuture:
      debug "Proposing block", timeIn = timeToSleep,
                               validator = shortLog(proposerKey)
      await sleepAsync(timeToSleep)
    else:
      debug "Proposing block", timeIn = 0.seconds,
                               validator = shortLog(proposerKey)

    let sres = vc.getCurrentSlot()
    if sres.isSome():
      let currentSlot = sres.get()
      let validator =
        block:
          let res = vc.getValidator(proposerKey)
          if res.isNone():
            return
          res.get()
      await vc.publishBlock(currentSlot, slot, validator)

  except CancelledError:
    debug "Proposing task was cancelled", slot = slot,
                                          validator = shortLog(proposerKey)

proc spawnProposalTask(vc: ValidatorClientRef,
                       duty: RestProposerDuty): ProposerTask =
  let future = proposeBlock(vc, duty.slot, duty.pubkey)
  ProposerTask(future: future, duty: duty)

proc contains(data: openArray[RestProposerDuty], task: ProposerTask): bool =
  for item in data:
    if (item.pubkey == task.duty.pubkey) and (item.slot == task.duty.slot):
      return true
  false

proc contains(data: openArray[ProposerTask], duty: RestProposerDuty): bool =
  for item in data:
    if (item.duty.pubkey == duty.pubkey) and (item.duty.slot == duty.slot):
      return true
  false

proc checkDuty(duty: RestProposerDuty, epoch: Epoch, slot: Slot): bool =
  let lastSlot = compute_start_slot_at_epoch(epoch + 1'u64)
  if duty.slot >= slot:
    if duty.slot < lastSlot:
      true
    else:
      warn "Block proposal duty is in the far future, ignoring",
           duty_slot = duty.slot, validator = shortLog(duty.pubkey),
           wall_slot = slot, last_slot_in_epoch = (lastSlot - 1'u64)
      false
  else:
    warn "Block proposal duty is in the past, ignoring", duty_slot = duty.slot,
         validator = shortLog(duty.pubkey), wall_slot = slot
    false

proc addOrReplaceProposers*(vc: ValidatorClientRef, epoch: Epoch,
                            dependentRoot: Eth2Digest,
                            duties: openArray[RestProposerDuty]) =
  let default = ProposedData(epoch: Epoch(0xFFFF_FFFF_FFFF_FFFF'u64))
  let sres = vc.getCurrentSlot()
  if sres.isSome():
    let
      currentSlot = sres.get()
      epochDuties = vc.proposers.getOrDefault(epoch, default)
    if not(epochDuties.isDefault()):
      if epochDuties.dependentRoot != dependentRoot:
        warn "Proposer duties re-organization", duties_count = len(duties),
             wall_slot = currentSlot, epoch = epoch,
             prior_dependent_root = epochDuties.dependentRoot,
             dependent_root = dependentRoot, wall_slot = currentSlot
        let tasks =
          block:
            var res: seq[ProposerTask]
            var hashset = initHashSet[Slot]()

            for task in epochDuties.duties:
              if task notin duties:
                # Task is no more relevant, so cancel it.
                debug "Cancelling running proposal duty task",
                      slot = task.duty.slot,
                      validator = shortLog(task.duty.pubkey)
                task.future.cancel()
              else:
                # If task is already running for proper slot, we keep it alive.
                debug "Keep running previous proposal duty task",
                      slot = task.duty.slot,
                      validator = shortLog(task.duty.pubkey)
                res.add(task)

            for duty in duties:
              if duty notin res:
                debug "New proposal duty received", slot = duty.slot,
                      validator = shortLog(duty.pubkey)
                if checkDuty(duty, epoch, currentSlot):
                  let task = vc.spawnProposalTask(duty)
                  if duty.slot in hashset:
                    error "Multiple block proposers for this slot, " &
                          "producing blocks for all proposers", slot = duty.slot
                  else:
                    hashset.incl(duty.slot)
                  res.add(task)
            res
        vc.proposers[epoch] = ProposedData.init(epoch, dependentRoot, tasks)
    else:
      debug "New block proposal duties received",
            dependent_root = dependentRoot, duties_count = len(duties),
            wall_slot = currentSlot, epoch = epoch
      # Spawn new proposer tasks and modify proposers map.
      let tasks =
        block:
          var hashset = initHashSet[Slot]()
          var res: seq[ProposerTask]
          for duty in duties:
            debug "New proposal duty received", slot = duty.slot,
                  validator = shortLog(duty.pubkey)
            if checkDuty(duty, epoch, currentSlot):
              let task = vc.spawnProposalTask(duty)
              if duty.slot in hashset:
                error "Multiple block proposers for this slot, " &
                      "producing blocks for all proposers", slot = duty.slot
              else:
                hashset.incl(duty.slot)
              res.add(task)
          res
      vc.proposers[epoch] = ProposedData.init(epoch, dependentRoot, tasks)

proc waitForBlockPublished*(vc: ValidatorClientRef, slot: Slot) {.async.} =
  ## This procedure will wait for all the block proposal tasks to be finished at
  ## slot ``slot``
  let pendingTasks =
    block:
      var res: seq[Future[void]]
      let epochDuties = vc.proposers.getOrDefault(slot.epoch())
      for task in epochDuties.duties:
        if task.duty.slot == slot:
          if not(task.future.finished()):
            res.add(task.future)
      res
  if len(pendingTasks) > 0:
    await allFutures(pendingTasks)