nimbus-eth2/beacon_chain/validator_client/sync_committee_service.nim

# beacon_chain
# Copyright (c) 2022-2024 Status Research & Development GmbH
# Licensed and distributed under either of
#   * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
#   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed except according to those terms.

{.push raises: [].}

import
  std/sets,
  metrics, chronicles,
  ../spec/datatypes/[phase0, altair, bellatrix],
  ../spec/eth2_apis/rest_types,
  ../validators/activity_metrics,
  "."/[common, api, selection_proofs]

const
  ServiceName = "sync_committee_service"

logScope: service = ServiceName

type
  ContributionItem* = object
    aggregator_index: uint64
    selection_proof: ValidatorSig
    validator: AttachedValidator
    subcommitteeIdx: SyncSubcommitteeIndex

proc serveSyncCommitteeMessage*(
    service: SyncCommitteeServiceRef,
    slot: Slot,
    beaconBlockRoot: Eth2Digest,
    duty: SyncCommitteeDuty
): Future[bool] {.async: (raises: [CancelledError]).} =
  let
    vc = service.client
    startTime = Moment.now()
    fork = vc.forkAtEpoch(slot.epoch)
    genesisValidatorsRoot = vc.beaconGenesis.genesis_validators_root
    vindex = duty.validator_index
    validator = vc.getValidatorForDuties(
      duty.pubkey, slot, slashingSafe = true).valueOr: return false

  logScope:
    validator = validatorLog(validator)
    block_root = shortLog(beaconBlockRoot)
    slot = slot

  let
    message =
      block:
        let res = await getSyncCommitteeMessage(validator, fork,
                                                genesisValidatorsRoot,
                                                slot, beaconBlockRoot)
        if res.isErr():
          warn "Unable to sign committee message using remote signer"
          return
        res.get()

  logScope:
    message = shortLog(message)

  debug "Sending sync committee message",
        delay = vc.getDelay(message.slot.sync_committee_message_deadline())

  let res =
    try:
      await vc.submitPoolSyncCommitteeSignature(message, ApiStrategyKind.First)
    except ValidatorApiError as exc:
      warn "Unable to publish sync committee message",
           reason = exc.getFailureReason()
      return false
    except CancelledError:
      debug "Publish sync committee message request was interrupted"
      return false

  let
    delay = vc.getDelay(message.slot.sync_committee_message_deadline())
    dur = Moment.now() - startTime

  if res:
    beacon_sync_committee_messages_sent.inc()
    beacon_sync_committee_message_sent_delay.observe(delay.toFloatSeconds())
    notice "Sync committee message published",
           validator_index = vindex, delay = delay, duration = dur
  else:
    warn "Sync committee message was not accepted by beacon node",
         validator_index = vindex, delay = delay, duration = dur
  res

proc produceAndPublishSyncCommitteeMessages(
    service: SyncCommitteeServiceRef,
    slot: Slot,
    beaconBlockRoot: Eth2Digest,
    duties: seq[SyncCommitteeDuty]
): Future[void] {.async: (raises: [CancelledError]).} =
  let
    vc = service.client
    startTime = Moment.now()

  let pendingSyncCommitteeMessages =
    block:
      var res: seq[Future[bool]]
      for duty in duties:
        debug "Serving sync message duty", duty, epoch = slot.epoch()
        res.add(service.serveSyncCommitteeMessage(slot,
                                                  beaconBlockRoot,
                                                  duty))
      res

  let statistics =
    block:
      var errored, succeed, failed = 0
      try:
        await allFutures(pendingSyncCommitteeMessages)
      except CancelledError as exc:
        let pending = pendingSyncCommitteeMessages
          .filterIt(not(it.finished())).mapIt(it.cancelAndWait())
        await noCancel allFutures(pending)
        raise exc

      for future in pendingSyncCommitteeMessages:
        if future.completed():
          if future.value:
            inc(succeed)
          else:
            inc(failed)
        else:
          inc(errored)
      (succeed, errored, failed)

  let
    delay = vc.getDelay(slot.attestation_deadline())
    dur = Moment.now() - startTime

  debug "Sync committee message statistics",
        total = len(pendingSyncCommitteeMessages),
        succeed = statistics[0], failed_to_deliver = statistics[1],
        not_accepted = statistics[2], delay = delay, duration = dur,
        slot = slot, duties_count = len(duties)

proc serveContributionAndProof*(
    service: SyncCommitteeServiceRef,
    proof: ContributionAndProof,
    validator: AttachedValidator
): Future[bool] {.async: (raises: [CancelledError]).} =
  ## Signs ConributionAndProof object and sends it to BN.
  let
    vc = service.client
    startTime = Moment.now()
    slot = proof.contribution.slot
    genesisRoot = vc.beaconGenesis.genesis_validators_root
    fork = vc.forkAtEpoch(slot.epoch)

  logScope:
    validator = validatorLog(validator)
    contribution = shortLog(proof.contribution)

  let signature =
    block:
      let res =
        try:
          await validator.getContributionAndProofSignature(
            fork, genesisRoot, proof)
        except CancelledError:
          debug "Sync contribution signing process was interrupted"
          return false

      if res.isErr():
        warn "Unable to sign sync committee contribution using remote signer",
             reason = res.error()
        return false
      res.get()

  debug "Sending sync contribution",
        delay = vc.getDelay(slot.sync_contribution_deadline())

  let restSignedProof = RestSignedContributionAndProof.init(
    proof, signature)

  let res =
    try:
      await vc.publishContributionAndProofs(@[restSignedProof],
                                            ApiStrategyKind.First)
    except ValidatorApiError as exc:
      warn "Unable to publish sync contribution",
           reason = exc.getFailureReason()
      false
    except CancelledError:
      debug "Publication process of sync contribution was interrupted"
      return false

  let dur = Moment.now() - startTime
  if res:
    beacon_sync_committee_contributions_sent.inc()
    notice "Sync contribution published", duration = dur
  else:
    warn "Sync contribution was not accepted by beacon node", duration = dur
  res

proc produceAndPublishContributions(
    service: SyncCommitteeServiceRef,
    slot: Slot,
    beaconBlockRoot: Eth2Digest,
    duties: seq[SyncCommitteeDuty]
) {.async: (raises: [CancelledError]).} =
  let
    vc = service.client
    startTime = Moment.now()

  logScope:
    slot = slot
    block_root = shortLog(beaconBlockRoot)

  var (contributions, pendingFutures, contributionsMap) =
    block:
      var
        resItems: seq[ContributionItem]
        resFutures: seq[FutureBase]
        resMap: array[SYNC_COMMITTEE_SUBNET_COUNT,
                      Future[SyncCommitteeContribution]]
      for duty in duties:
        let validator = vc.getValidatorForDuties(duty.pubkey, slot).valueOr:
          continue
        if validator.index.isNone():
          continue
        for inindex in duty.validator_sync_committee_indices:
          let
            subCommitteeIdx = getSubcommitteeIndex(inindex)
            signature =
              vc.getSyncCommitteeSelectionProof(duty.pubkey,
                                                slot, inindex).valueOr:
                continue

          if is_sync_committee_aggregator(signature):
            resItems.add(ContributionItem(
              aggregator_index: uint64(validator.index.get()),
              selection_proof: signature,
              validator: validator,
              subcommitteeIdx: subCommitteeIdx
            ))
            if isNil(resMap[subCommitteeIdx]):
              let future =
                vc.produceSyncCommitteeContribution(
                  slot, subCommitteeIdx, beaconBlockRoot, ApiStrategyKind.Best)
              resMap[int(subCommitteeIdx)] = future
              resFutures.add(FutureBase(future))
      (resItems, resFutures, resMap)

  if len(contributions) > 0:
    let
      pendingAggregates =
        block:
          var res: seq[Future[bool]]
          while len(pendingFutures) > 0:
            try:
              discard await race(pendingFutures)
            except ValueError:
              raiseAssert "Number of pendingFutures should not be zero"
            except CancelledError as exc:
              let pending = pendingFutures
                .filterIt(not(it.finished())).mapIt(it.cancelAndWait())
              await noCancel allFutures(pending)
              raise exc

            var completed: seq[int]
            for contrib in contributions:
              let future = contributionsMap[contrib.subcommitteeIdx]
              doAssert(not(isNil(future)))
              let index = pendingFutures.find(FutureBase(future))
              if future.finished() and (index >= 0):
                if index notin completed: completed.add(index)
                let aggContribution =
                  try:
                    let tfut =
                      cast[Future[SyncCommitteeContribution].
                        Raising([CancelledError, ValidatorApiError])](future)
                    Opt.some(tfut.read())
                  except FuturePendingError:
                    raiseAssert "Future should be finished"
                  except ValidatorApiError as exc:
                    warn "Unable to get sync message contribution data",
                         reason = exc.getFailureReason()
                    Opt.none(SyncCommitteeContribution)
                  except CancelledError as exc:
                    debug "Request for sync message contribution was " &
                          "interrupted"
                    raise exc

                if aggContribution.isSome():
                  let proof = ContributionAndProof(
                    aggregator_index: contrib.aggregator_index,
                    contribution: aggContribution.get(),
                    selection_proof: contrib.selection_proof
                  )
                  res.add(
                    service.serveContributionAndProof(proof, contrib.validator))

            pendingFutures =
              block:
                var tres: seq[FutureBase]
                for index, value in pendingFutures.pairs():
                  if index notin completed: tres.add(value)
                tres
          res
      statistics =
        block:
          var errored, succeed, failed = 0
          try:
            await allFutures(pendingAggregates)
          except CancelledError as exc:
            let pending = pendingAggregates
              .filterIt(not(it.finished())).mapIt(it.cancelAndWait())
            await noCancel allFutures(pending)
            raise exc

          for future in pendingAggregates:
            if future.completed():
              if future.value:
                inc(succeed)
              else:
                inc(failed)
            else:
              inc(errored)
          (succeed, errored, failed)

    let
      delay = vc.getDelay(slot.aggregate_deadline())
      dur = Moment.now() - startTime

    debug "Sync message contribution statistics",
          total = len(contributions),
          succeed = statistics[0],
          failed_to_create = len(pendingAggregates) - len(contributions),
          failed_to_deliver = statistics[1],
          not_accepted = statistics[2],
          delay = delay, duration = dur

  else:
    debug "No contribution and proofs scheduled for the slot"

proc publishSyncMessagesAndContributions(
    service: SyncCommitteeServiceRef,
    slot: Slot,
    duties: seq[SyncCommitteeDuty]
) {.async: (raises: [CancelledError]).} =
  let vc = service.client

  await vc.waitForBlock(slot, syncCommitteeMessageSlotOffset)

  logScope:
    slot = slot

  block:
    let delay = vc.getDelay(slot.sync_committee_message_deadline())
    debug "Producing sync committee messages", delay = delay,
          duties_count = len(duties)

  let beaconBlockRoot =
    block:
      try:
        let res = await vc.getHeadBlockRoot(ApiStrategyKind.Best)
        if res.execution_optimistic.isNone():
          ## The `execution_optimistic` is missing from the response, we assume
          ## that the BN is unaware optimistic sync, so we consider the BN
          ## to be synchronized with the network.
          ## TODO (cheatfate): This should be removed when VC will be able to
          ## handle getSpec() API call with fork constants.
          res.data.root
        else:
          if res.execution_optimistic.get():
            notice "Execution client not in sync"
            return
          res.data.root
      except ValidatorApiError as exc:
        warn "Unable to retrieve head block's root to sign", reason = exc.msg,
             reason = exc.getFailureReason()
        return
      except CancelledError:
        debug "Block root request was interrupted"
        return

  try:
    await service.produceAndPublishSyncCommitteeMessages(
      slot, beaconBlockRoot, duties)
  except ValidatorApiError as exc:
    warn "Unable to proceed sync committee messages",
         duties_count = len(duties), reason = exc.getFailureReason()
    return
  except CancelledError:
    debug "Sync committee messages production was interrupted"
    return

  let currentTime = vc.beaconClock.now()
  if slot.sync_contribution_deadline() > currentTime:
    let waitDur =
      nanoseconds((slot.sync_contribution_deadline() - currentTime).nanoseconds)
    # Sleeping until `sync_contribution_deadline`.
    debug "Waiting for sync contribution deadline", wait_time = waitDur
    await sleepAsync(waitDur)

  block:
    let delay = vc.getDelay(slot.sync_contribution_deadline())
    debug "Producing contribution and proofs", delay = delay

  try:
    await service.produceAndPublishContributions(slot, beaconBlockRoot, duties)
  except CancelledError:
    debug "Sync committee contributions production was interrupted"
    return

proc processSyncCommitteeTasks(
    service: SyncCommitteeServiceRef,
    slot: Slot
) {.async: (raises: [CancelledError]).} =
  let
    vc = service.client
    duties = vc.getSyncCommitteeDutiesForSlot(slot + 1)
    timeout = vc.beaconClock.durationToNextSlot()

  logScope:
    slot = slot

  try:
    await service.publishSyncMessagesAndContributions(slot,
                                                      duties).wait(timeout)
  except AsyncTimeoutError:
    warn "Unable to publish sync committee messages and contributions in time",
         timeout = timeout
  except CancelledError as exc:
    debug "Sync committee publish task has been interrupted"
    raise exc

proc mainLoop(service: SyncCommitteeServiceRef) {.async: (raises: []).} =
  let vc = service.client
  service.state = ServiceState.Running
  debug "Service started"

  debug "Sync committee processing loop is waiting for initialization"
  try:
    await allFutures(
      vc.preGenesisEvent.wait(),
      vc.genesisEvent.wait(),
      vc.indicesAvailable.wait(),
      vc.forksAvailable.wait()
    )
  except CancelledError:
    debug "Service interrupted"
    return

  doAssert(len(vc.forks) > 0, "Fork schedule must not be empty at this point")

  var currentSlot: Opt[Slot]
  while true:
    # This loop could look much more nicer/better, when
    # https://github.com/nim-lang/Nim/issues/19911 will be fixed, so it could
    # become safe to combine loops, breaks and exception handlers.
    let breakLoop =
      try:
        let
          # We use zero offset here, because we do waiting in
          # waitForBlock(syncCommitteeMessageSlotOffset).
          slot = await vc.checkedWaitForNextSlot(currentSlot, ZeroTimeDiff,
                                                 false)
        if slot.isNone():
          debug "System time adjusted backwards significantly, exiting"
          true
        else:
          currentSlot = slot
          await service.processSyncCommitteeTasks(currentSlot.get())
          false
      except CancelledError:
        debug "Service interrupted"
        true

    if breakLoop:
      break

proc init*(
    t: typedesc[SyncCommitteeServiceRef],
    vc: ValidatorClientRef
): Future[SyncCommitteeServiceRef] {.async: (raises: []).} =
  logScope: service = ServiceName
  let res = SyncCommitteeServiceRef(name: ServiceName, client: vc,
                                    state: ServiceState.Initialized)
  debug "Initializing service"
  res

proc start*(service: SyncCommitteeServiceRef) =
  service.lifeFut = mainLoop(service)