nimbus-eth2/beacon_chain/validator_client/duties_service.nim

# beacon_chain
# Copyright (c) 2021-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, sequtils]
import chronicles, metrics
import "."/[common, api, block_service, selection_proofs]

const
  ServiceName = "duties_service"
  SUBSCRIPTION_LOOKAHEAD_EPOCHS* = 4'u64
  AGGREGATION_PRE_COMPUTE_EPOCHS* = 1'u64

logScope: service = ServiceName

type
  DutiesServiceLoop* = enum
    AttesterLoop, ProposerLoop, IndicesLoop, SyncCommitteeLoop,
    ProposerPreparationLoop, ValidatorRegisterLoop, DynamicValidatorsLoop,
    SlashPruningLoop

chronicles.formatIt(DutiesServiceLoop):
  case it
  of AttesterLoop: "attester_loop"
  of ProposerLoop: "proposer_loop"
  of IndicesLoop: "index_loop"
  of SyncCommitteeLoop: "sync_committee_loop"
  of ProposerPreparationLoop: "proposer_prepare_loop"
  of ValidatorRegisterLoop: "validator_register_loop"
  of DynamicValidatorsLoop: "dynamic_validators_loop"
  of SlashPruningLoop: "slashing_pruning_loop"

proc checkDuty(duty: RestAttesterDuty): bool =
  (duty.committee_length <= MAX_VALIDATORS_PER_COMMITTEE) and
  (uint64(duty.committee_index) < MAX_COMMITTEES_PER_SLOT) and
  (uint64(duty.validator_committee_index) < duty.committee_length) and
  (uint64(duty.validator_index) <= VALIDATOR_REGISTRY_LIMIT)

proc checkSyncDuty(duty: RestSyncCommitteeDuty): bool =
  uint64(duty.validator_index) <= VALIDATOR_REGISTRY_LIMIT

proc pollForValidatorIndices*(service: DutiesServiceRef) {.async.} =
  let vc = service.client

  let validatorIdents =
    block:
      var res: seq[ValidatorIdent]
      for validator in vc.attachedValidators[].items():
        if validator.needsUpdate():
          res.add(ValidatorIdent.init(validator.pubkey))
      res

  let start = Moment.now()

  var validators: seq[RestValidator]

  for idents in chunks(validatorIdents, ClientMaximumValidatorIds):
    let res =
      try:
        await vc.getValidators(idents, ApiStrategyKind.First)
      except ValidatorApiError as exc:
        warn "Unable to get head state's validator information",
              reason = exc.getFailureReason()
        return
      except CancelledError as exc:
        debug "Validator's indices processing was interrupted"
        raise exc
      except CatchableError as exc:
        error "Unexpected error occurred while getting validator information",
              err_name = exc.name, err_msg = exc.msg
        return
    for item in res:
      validators.add(item)

  var
    missing: seq[string]
    updated: seq[string]
    list: seq[AttachedValidator]

  for item in validators:
    let validator = vc.attachedValidators[].getValidator(item.validator.pubkey)
    if validator.isNone():
      missing.add(validatorLog(item.validator.pubkey, item.index))
    else:
      validator.get().updateValidator(Opt.some ValidatorAndIndex(
        index: item.index,
        validator: item.validator))
      updated.add(validatorLog(item.validator.pubkey, item.index))
      list.add(validator.get())

  if len(updated) > 0:
    info "Validator indices updated",
      pending = len(validatorIdents) - len(updated),
      missing = len(missing),
      updated = len(updated),
      elapsed_time = (Moment.now() - start)
    trace "Validator indices update dump", missing_validators = missing,
          updated_validators = updated
    vc.indicesAvailable.fire()

proc pollForAttesterDuties*(service: DutiesServiceRef,
                            epoch: Epoch): Future[int] {.async.} =
  var currentRoot: Opt[Eth2Digest]
  let
    vc = service.client
    indices = toSeq(vc.attachedValidators[].indices())
    relevantDuties =
      block:
        var duties: seq[RestAttesterDuty]
        block mainLoop:
          while true:
            block innerLoop:
              for chunk in indices.chunks(DutiesMaximumValidatorIds):
                let res =
                  try:
                    await vc.getAttesterDuties(epoch, chunk,
                                               ApiStrategyKind.First)
                  except ValidatorApiError as exc:
                    warn "Unable to get attester duties", epoch = epoch,
                         reason = exc.getFailureReason()
                    return 0
                  except CancelledError as exc:
                    debug "Attester duties processing was interrupted"
                    raise exc
                  except CatchableError as exc:
                    error "Unexpected error while getting attester duties",
                          epoch = epoch, err_name = exc.name, err_msg = exc.msg
                    return 0
                if currentRoot.isNone():
                  # First request
                  currentRoot = Opt.some(res.dependent_root)
                else:
                  if currentRoot.get() != res.dependent_root:
                    # `dependent_root` must be equal for all requests/response,
                    # if it got changed it means that some reorg was happened in
                    # beacon node and we should re-request all queries again.
                    duties.setLen(0)
                    currentRoot = Opt.none(Eth2Digest)
                    break innerLoop

                for duty in res.data:
                  if checkDuty(duty) and
                     (duty.pubkey in vc.attachedValidators[]):
                    duties.add(duty)
                break mainLoop
        duties

  template checkReorg(a, b: untyped): bool =
    not(a.dependentRoot == b.get())

  let addOrReplaceItems =
    block:
      var alreadyWarned = false
      var res: seq[tuple[epoch: Epoch, duty: RestAttesterDuty]]
      for duty in relevantDuties:
        var dutyFound = false
        vc.attesters.withValue(duty.pubkey, map):
          map[].duties.withValue(epoch, epochDuty):
            dutyFound = true
            if checkReorg(epochDuty[], currentRoot):
              if not(alreadyWarned):
                info "Attester duties re-organization",
                     prior_dependent_root = epochDuty[].dependentRoot,
                     dependent_root = currentRoot.get()
                alreadyWarned = true
              res.add((epoch, duty))
        if not(dutyFound):
          info "Received new attester duty", duty, epoch = epoch,
               dependent_root = currentRoot.get()
          res.add((epoch, duty))
      res

  for item in addOrReplaceItems:
    let dap = DutyAndProof.init(item.epoch, currentRoot.get(), item.duty,
                                Opt.none(ValidatorSig))
    vc.attesters.mgetOrPut(dap.data.pubkey,
                           default(EpochDuties)).duties[dap.epoch] = dap
  return len(addOrReplaceItems)

proc pruneSyncCommitteeDuties*(service: DutiesServiceRef, slot: Slot) =
  let vc = service.client
  if slot.is_sync_committee_period():
    var newSyncCommitteeDuties: SyncCommitteeDutiesMap
    let period = slot.sync_committee_period()
    for key, item in vc.syncCommitteeDuties:
      var currentPeriodDuties = SyncPeriodDuties()
      for periodKey, periodDuty in item.duties:
        if periodKey >= period:
          currentPeriodDuties.duties[periodKey] = periodDuty
      newSyncCommitteeDuties[key] = currentPeriodDuties
    vc.syncCommitteeDuties = newSyncCommitteeDuties

proc pruneSyncCommitteeSelectionProofs*(service: DutiesServiceRef, slot: Slot) =
  let
    vc = service.client
    slotEpoch = slot.epoch()
  var res: seq[Epoch]
  for epoch in vc.syncCommitteeProofs.keys():
    if epoch < slotEpoch: res.add(epoch)
  for epoch in res:
    vc.syncCommitteeProofs.del(epoch)

proc pollForSyncCommitteeDuties*(
       service: DutiesServiceRef,
       period: SyncCommitteePeriod
     ): Future[int] {.async.} =
  let
    vc = service.client
    indices = toSeq(vc.attachedValidators[].indices())
    altairEpoch = vc.runtimeConfig.forkConfig.get().altairEpoch
    epoch = max(period.start_epoch(), altairEpoch)
    relevantDuties =
      block:
        var duties: seq[RestSyncCommitteeDuty]
        # We use `DutiesMaximumValidatorIds` here because validator ids are sent
        # in HTTP request body and NOT in HTTP request headers.
        for chunk in indices.chunks(DutiesMaximumValidatorIds):
          let res =
            try:
              await vc.getSyncCommitteeDuties(epoch, chunk,
                                              ApiStrategyKind.First)
            except ValidatorApiError as exc:
              warn "Unable to get sync committee duties",
                   period = period, epoch = epoch,
                   reason = exc.getFailureReason()
              return 0
            except CancelledError as exc:
              debug "Sync committee duties processing was interrupted",
                    period = period, epoch = epoch
              raise exc
            except CatchableError as exc:
              error "Unexpected error while getting sync committee duties",
                    period = period, epoch = epoch,
                    err_name = exc.name, err_msg = exc.msg
              return 0

          for duty in res.data:
            if checkSyncDuty(duty) and (duty.pubkey in vc.attachedValidators[]):
              duties.add(duty)

        duties

  template checkReorg(a, b: untyped): bool =
    not(compareUnsorted(a.validator_sync_committee_indices,
                        b.validator_sync_committee_indices))

  let addOrReplaceItems =
    block:
      var
        alreadyWarned = false
        res: seq[tuple[period: SyncCommitteePeriod,
                       duty: RestSyncCommitteeDuty]]
      for duty in relevantDuties:
        var dutyFound = false
        vc.syncCommitteeDuties.withValue(duty.pubkey, map):
          map[].duties.withValue(period, periodDuty):
            dutyFound = true
            if checkReorg(periodDuty[], duty):
              if not(alreadyWarned):
                info "Sync committee duties re-organization"
                alreadyWarned = true
              res.add((period, duty))
        if not(dutyFound):
          res.add((period, duty))
          info "Received new sync committee duty", duty, period
      res

  for item in addOrReplaceItems:
    vc.syncCommitteeDuties.mgetOrPut(
      item.duty.pubkey, default(SyncPeriodDuties)).duties[item.period] =
        item.duty
  len(addOrReplaceItems)

proc pruneAttesterDuties(service: DutiesServiceRef, epoch: Epoch) =
  let vc = service.client
  var attesters: AttesterMap
  for key, item in vc.attesters:
    var v = EpochDuties()
    for epochKey, epochDuty in item.duties:
      if (epochKey + HISTORICAL_DUTIES_EPOCHS) >= epoch:
        v.duties[epochKey] = epochDuty
      else:
        debug "Attester duties for the epoch has been pruned",
              pubkey = shortLog(key), epoch = epochKey, loop = AttesterLoop
    attesters[key] = v
  vc.attesters = attesters

proc pollForAttesterDuties*(service: DutiesServiceRef) {.async.} =
  ## Query the beacon node for attestation duties for all known validators.
  ##
  ## This function will perform (in the following order):
  ##
  ## 1. Poll for current-epoch duties and update the local `attesters` map.
  ## 2. Poll for next-epoch duties and update the local `attesters` map.
  ## 3. Push out any attestation subnet subscriptions to the BN.
  let vc = service.client
  let
    currentSlot = vc.getCurrentSlot().get(Slot(0))
    currentEpoch = currentSlot.epoch()
    nextEpoch = currentEpoch + 1'u64

  if vc.attachedValidators[].count() != 0:
    var counts: array[2, tuple[epoch: Epoch, count: int]]
    counts[0] = (currentEpoch,
                 await service.pollForAttesterDuties(currentEpoch))
    counts[1] = (nextEpoch,
                 await service.pollForAttesterDuties(nextEpoch))

    if (counts[0].count == 0) and (counts[1].count == 0):
      debug "No new attester's duties received", slot = currentSlot

    block:
      let
        moment = Moment.now()
        sigres =
          await vc.fillAttestationSelectionProofs(currentSlot,
            currentSlot + Epoch(AGGREGATION_PRE_COMPUTE_EPOCHS))

      if vc.config.distributedEnabled:
        debug "Attestation selection proofs have been received",
              signatures_requested = sigres.signaturesRequested,
              signatures_received = sigres.signaturesReceived,
              selections_requested = sigres.selectionsRequested,
              selections_received = sigres.selectionsReceived,
              selections_processed = sigres.selectionsProcessed,
              total_elapsed_time = (Moment.now() - moment)
      else:
        debug "Attestation selection proofs have been received",
              signatures_requested = sigres.signaturesRequested,
              signatures_received = sigres.signaturesReceived,
              total_elapsed_time = (Moment.now() - moment)

    let subscriptions =
      block:
        var res: seq[RestCommitteeSubscription]
        for item in counts:
          if item.count > 0:
            for duty in vc.attesterDutiesForEpoch(item.epoch):
              if currentSlot + SUBSCRIPTION_BUFFER_SLOTS < duty.data.slot:
                let isAggregator =
                  if duty.slotSig.isSome():
                    is_aggregator(duty.data.committee_length,
                                  duty.slotSig.get())
                  else:
                    false
                let sub = RestCommitteeSubscription(
                  validator_index: duty.data.validator_index,
                  committee_index: duty.data.committee_index,
                  committees_at_slot: duty.data.committees_at_slot,
                  slot: duty.data.slot,
                  is_aggregator: isAggregator
                )
                res.add(sub)
        res

    if len(subscriptions) > 0:
      let res = await vc.prepareBeaconCommitteeSubnet(subscriptions)
      if res == 0:
        warn "Failed to subscribe validators to beacon committee subnets",
             slot = currentSlot, epoch = currentEpoch,
             subscriptions_count = len(subscriptions)

  service.pruneAttesterDuties(currentEpoch)

proc pollForSyncCommitteeDuties*(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  let
    currentSlot = vc.getCurrentSlot().get(Slot(0))
    currentEpoch = currentSlot.epoch()
    altairEpoch =
      if vc.runtimeConfig.forkConfig.isSome():
        vc.runtimeConfig.forkConfig.get().altairEpoch
      else:
        return

  if currentEpoch < altairEpoch:
    # We are not going to poll for sync committee duties until `altairEpoch`.
    return

  let
    currentPeriod = currentEpoch.sync_committee_period()
    nextPeriod = currentPeriod + 1

  if vc.attachedValidators[].count() != 0:
    var counts: array[2, tuple[period: SyncCommitteePeriod, count: int]]
    counts[0] = (currentPeriod,
                 await service.pollForSyncCommitteeDuties(currentPeriod))

    const
      numDelayEpochs = 4  # Chosen empirically
      numLookaheadEpochs =
        max(EPOCHS_PER_SYNC_COMMITTEE_PERIOD, numDelayEpochs) -
        numDelayEpochs + 1
    if (currentEpoch + numLookaheadEpochs) >= nextPeriod.start_epoch:
      counts[1] = (nextPeriod,
                   await service.pollForSyncCommitteeDuties(nextPeriod))
    else:
      # Skip fetching `nextPeriod` until sync committees are likely known,
      # as determined by `numDelayEpochs` from sync committee period start.
      counts[1] = (nextPeriod, 0)

    if (counts[0].count == 0) and (counts[1].count == 0):
      debug "No new sync committee duties received", slot = currentSlot

    block:
      let
        moment = Moment.now()
        sigres =
          await vc.fillSyncCommitteeSelectionProofs(currentSlot,
            currentSlot + Epoch(AGGREGATION_PRE_COMPUTE_EPOCHS))

      if vc.config.distributedEnabled:
        debug "Sync committee selection proofs have been received",
              signatures_requested = sigres.signaturesRequested,
              signatures_received = sigres.signaturesReceived,
              selections_requested = sigres.selectionsRequested,
              selections_received = sigres.selectionsReceived,
              selections_processed = sigres.selectionsProcessed,
              total_elapsed_time = (Moment.now() - moment)
      else:
        debug "Sync committee selection proofs have been received",
              signatures_requested = sigres.signaturesRequested,
              signatures_received = sigres.signaturesReceived,
              total_elapsed_time = (Moment.now() - moment)

    let
      periods =
        block:
          var res: seq[tuple[slot: Slot, period: SyncCommitteePeriod]]
          if service.syncSubscriptionEpoch.get(FAR_FUTURE_EPOCH) !=
             currentEpoch:
            res.add((currentSlot, currentPeriod))
          let
            lookaheadSlot = currentSlot +
                            SUBSCRIPTION_LOOKAHEAD_EPOCHS * SLOTS_PER_EPOCH
            lookaheadPeriod = lookaheadSlot.sync_committee_period()
          if lookaheadPeriod > currentPeriod:
            res.add((lookaheadSlot, lookaheadPeriod))
          res
      subscriptions =
        block:
          var res: seq[RestSyncCommitteeSubscription]
          for item in periods:
            let
              untilEpoch = start_epoch(item.period + 1)
              subscriptionsInfo =
                vc.syncMembersSubscriptionInfoForPeriod(item.period)
            for info in subscriptionsInfo:
              let sub = RestSyncCommitteeSubscription(
                validator_index: info.validator_index,
                sync_committee_indices:
                  info.validator_sync_committee_indices,
                until_epoch: untilEpoch
              )
              res.add(sub)
          res
    if len(subscriptions) > 0:
      let res = await vc.prepareSyncCommitteeSubnets(subscriptions)
      if res == 0:
        warn "Failed to subscribe validators to sync committee subnets",
             slot = currentSlot, epoch = currentPeriod, period = currentPeriod,
             periods = periods, subscriptions_count = len(subscriptions)
      else:
        service.syncSubscriptionEpoch = Opt.some(currentEpoch)

  service.pruneSyncCommitteeDuties(currentSlot)
  service.pruneSyncCommitteeSelectionProofs(currentSlot)

proc pruneBeaconProposers(service: DutiesServiceRef, epoch: Epoch) =
  let vc = service.client

  var proposers: ProposerMap
  for epochKey, data in vc.proposers:
    if (epochKey + HISTORICAL_DUTIES_EPOCHS) >= epoch:
      proposers[epochKey] = data
    else:
      debug "Proposer duty has been pruned", epoch = epochKey,
            loop = ProposerLoop
  vc.proposers = proposers

proc pollForBeaconProposers*(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  let
    currentSlot = vc.getCurrentSlot().get(Slot(0))
    currentEpoch = currentSlot.epoch()

  if vc.attachedValidators[].count() != 0:
    try:
      let res = await vc.getProposerDuties(currentEpoch,
                                           ApiStrategyKind.First)
      let
        dependentRoot = res.dependent_root
        duties = res.data
        relevantDuties = duties.filterIt(it.pubkey in vc.attachedValidators[])

      if len(relevantDuties) > 0:
        vc.addOrReplaceProposers(currentEpoch, dependentRoot, relevantDuties)
      else:
        debug "No relevant proposer duties received", slot = currentSlot,
              duties_count = len(duties)
    except ValidatorApiError as exc:
      notice "Unable to get proposer duties", slot = currentSlot,
             epoch = currentEpoch, reason = exc.getFailureReason()
    except CancelledError as exc:
      debug "Proposer duties processing was interrupted"
      raise exc
    except CatchableError as exc:
      debug "Unexpected error occured while getting proposer duties",
            slot = currentSlot, epoch = currentEpoch, err_name = exc.name,
            err_msg = exc.msg

    service.pruneBeaconProposers(currentEpoch)
    vc.pruneBlocksSeen(currentEpoch)

proc prepareBeaconProposers*(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  let
    currentSlot = vc.getCurrentSlot().get(Slot(0))
    currentEpoch = currentSlot.epoch()
    proposers = vc.prepareProposersList(currentEpoch)

  if len(proposers) > 0:
    let count =
      try:
        await prepareBeaconProposer(vc, proposers)
      except ValidatorApiError as exc:
        warn "Unable to prepare beacon proposers", slot = currentSlot,
              epoch = currentEpoch, err_name = exc.name,
              err_msg = exc.msg, reason = exc.getFailureReason()
        0
      except CancelledError as exc:
        debug "Beacon proposer preparation processing was interrupted"
        raise exc
      except CatchableError as exc:
        error "Unexpected error occured while preparing beacon proposers",
              slot = currentSlot, epoch = currentEpoch, err_name = exc.name,
              err_msg = exc.msg
        0
    debug "Beacon proposers prepared",
          validators_count = vc.attachedValidators[].count(),
          proposers_count = len(proposers),
          prepared_count = count

proc registerValidators*(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  let
    currentSlot = vc.getCurrentSlot().get(Slot(0))
    genesisFork = vc.forks[0]
    registrations =
      try:
        await vc.prepareRegistrationList(getTime(), genesisFork)
      except CancelledError as exc:
        debug "Validator registration preparation was interrupted",
              slot = currentSlot, fork = genesisFork
        raise exc
      except CatchableError as exc:
        var default: seq[SignedValidatorRegistrationV1]
        error "Unexpected error occured while preparing validators " &
              "registration data", slot = currentSlot, fork = genesisFork,
              err_name = exc.name, err_msg = exc.msg
        default

    count =
      if len(registrations) > 0:
        try:
          await registerValidator(vc, registrations)
        except ValidatorApiError as exc:
          warn "Unable to register validators", slot = currentSlot,
                fork = genesisFork, err_name = exc.name,
                err_msg = exc.msg, reason = exc.getFailureReason()
          0
        except CancelledError as exc:
          debug "Validator registration was interrupted", slot = currentSlot,
                fork = genesisFork
          raise exc
        except CatchableError as exc:
          error "Unexpected error occured while registering validators",
                slot = currentSlot, fork = genesisFork, err_name = exc.name,
                err_msg = exc.msg
          0
      else:
        0

  if count > 0:
    debug "Validators registered", slot = currentSlot,
          beacon_nodes_count = count, registrations = len(registrations),
          validators_count = vc.attachedValidators[].count()

proc attesterDutiesLoop(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  debug "Attester duties loop is waiting for initialization"
  await allFutures(
    vc.preGenesisEvent.wait(),
    vc.indicesAvailable.wait(),
    vc.forksAvailable.wait()
  )
  doAssert(len(vc.forks) > 0, "Fork schedule must not be empty at this point")
  while true:
    await service.waitForNextSlot()
    # Cleaning up previous attestation duties task.
    if not(isNil(service.pollingAttesterDutiesTask)) and
       not(service.pollingAttesterDutiesTask.finished()):
      await cancelAndWait(service.pollingAttesterDutiesTask)
    # Spawning new attestation duties task.
    service.pollingAttesterDutiesTask = service.pollForAttesterDuties()

proc proposerDutiesLoop(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  debug "Proposer duties loop is waiting for initialization"
  await allFutures(
    vc.preGenesisEvent.wait(),
    vc.indicesAvailable.wait(),
    vc.forksAvailable.wait()
  )
  doAssert(len(vc.forks) > 0, "Fork schedule must not be empty at this point")
  while true:
    await service.pollForBeaconProposers()
    await service.waitForNextSlot()

proc validatorIndexLoop(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  debug "Validator indices loop is waiting for initialization"
  await vc.preGenesisEvent.wait()
  while true:
    await service.pollForValidatorIndices()
    await service.waitForNextSlot()

proc dynamicValidatorsLoop*(service: DutiesServiceRef,
                            web3signerUrl: Web3SignerUrl,
                            intervalInSeconds: int) {.async.} =
  let vc = service.client
  doAssert(intervalInSeconds > 0)

  proc addValidatorProc(data: KeystoreData) =
    vc.addValidator(data)

  var
    timeout = seconds(intervalInSeconds)
    exitLoop = false

  while not(exitLoop):
    exitLoop =
      try:
        await sleepAsync(timeout)
        timeout =
          block:
            let res = await queryValidatorsSource(web3signerUrl)
            if res.isOk():
              let keystores = res.get()
              debug "Web3Signer has been polled for validators",
                    keystores_found = len(keystores),
                    web3signer_url = web3signerUrl.url
              vc.attachedValidators.updateDynamicValidators(web3signerUrl,
                                                            keystores,
                                                            addValidatorProc)
              seconds(intervalInSeconds)
            else:
              seconds(5)
        false
      except CancelledError:
        true

proc proposerPreparationsLoop(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  debug "Beacon proposer preparation loop is waiting for initialization"
  await allFutures(
    vc.preGenesisEvent.wait(),
    vc.indicesAvailable.wait()
  )
  while true:
    await service.prepareBeaconProposers()
    await service.waitForNextSlot()

proc validatorRegisterLoop(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  doAssert(vc.config.payloadBuilderEnable)
  debug "Validator registration loop is waiting for initialization"
  await allFutures(
    vc.preGenesisEvent.wait(),
    vc.indicesAvailable.wait(),
    vc.forksAvailable.wait()
  )
  doAssert(len(vc.forks) > 0, "Fork schedule must not be empty at this point")
  while true:
    await service.registerValidators()
    await service.waitForNextSlot()

proc syncCommitteeDutiesLoop(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  debug "Sync committee duties loop is waiting for initialization"
  await allFutures(
    vc.preGenesisEvent.wait(),
    vc.indicesAvailable.wait(),
    vc.forksAvailable.wait()
  )
  doAssert(len(vc.forks) > 0, "Fork schedule must not be empty at this point")
  while true:
    await service.waitForNextSlot()
    if not(isNil(service.pollingSyncDutiesTask)) and
       not(service.pollingSyncDutiesTask.finished()):
      await cancelAndWait(service.pollingSyncDutiesTask)
    # Spawning new attestation duties task.
    service.pollingSyncDutiesTask = service.pollForSyncCommitteeDuties()

proc getNextEpochMiddleSlot(vc: ValidatorClientRef): Slot =
  let
    middleSlot = Slot(SLOTS_PER_EPOCH div 2)
    currentSlot = vc.beaconClock.now().slotOrZero()
    slotInEpoch = currentSlot.since_epoch_start()

  if slotInEpoch >= middleSlot:
    (currentSlot.epoch + 1'u64).start_slot() + uint64(middleSlot)
  else:
    currentSlot + (uint64(middleSlot) - uint64(slotInEpoch))

proc pruneSlashingDatabase(service: DutiesServiceRef) {.async.} =
  let
    vc = service.client
    currentSlot = vc.beaconClock.now().slotOrZero()
    startTime = Moment.now()
    blockHeader =
      try:
        await vc.getFinalizedBlockHeader()
      except CancelledError as exc:
        debug "Finalized block header request was interrupted",
              slot = currentSlot
        raise exc
      except CatchableError as exc:
        error "Unexpected error occured while requesting " &
              "finalized block header", slot = currentSlot,
              err_name = exc.name, err_msg = exc.msg
        Opt.none(GetBlockHeaderResponse)
    checkpointTime = Moment.now()
  if blockHeader.isSome():
    let epoch = blockHeader.get().data.header.message.slot.epoch
    vc.finalizedEpoch = Opt.some(epoch)
    if service.lastSlashingEpoch.get(FAR_FUTURE_EPOCH) != epoch:
      vc.attachedValidators[]
        .slashingProtection
        .pruneAfterFinalization(epoch)
      service.lastSlashingEpoch = Opt.some(epoch)
      let finishTime = Moment.now()
      debug "Slashing database has been pruned", slot = currentSlot,
        epoch = currentSlot.epoch(),
        finalized_epoch = epoch,
        elapsed_time = (finishTime - startTime),
        pruning_time = (finishTime - checkpointTime)

proc slashingDatabasePruningLoop(service: DutiesServiceRef) {.async.} =
  let vc = service.client
  debug "Slashing database pruning loop is waiting for initialization"
  await allFutures(
    vc.preGenesisEvent.wait(),
    vc.indicesAvailable.wait(),
    vc.forksAvailable.wait()
  )
  doAssert(len(vc.forks) > 0, "Fork schedule must not be empty at this point")
  while true:
    let slot = await vc.checkedWaitForSlot(vc.getNextEpochMiddleSlot(),
                                           aggregateSlotOffset, false)
    if slot.isNone():
      continue

    if not(isNil(service.pruneSlashingDatabaseTask)) and
       not(service.pruneSlashingDatabaseTask.finished()):
      await cancelAndWait(service.pruneSlashingDatabaseTask)
    service.pruneSlashingDatabaseTask = service.pruneSlashingDatabase()

template checkAndRestart(serviceLoop: DutiesServiceLoop,
                         future: Future[void], body: untyped): untyped =
  if future.finished():
    if future.failed():
      let error = future.readError()
      debug "The loop ended unexpectedly with an error",
            error_name = error.name, error_msg = error.msg, loop = serviceLoop
    elif future.cancelled():
      debug "The loop was interrupted", loop = serviceLoop
    else:
      debug "The loop is finished unexpectedly without an error",
            loop = serviceLoop
    future = body

proc mainLoop(service: DutiesServiceRef) {.async.} =
  let vc = service.client

  service.state = ServiceState.Running
  debug "Service started"

  var
    attestFut = service.attesterDutiesLoop()
    proposeFut = service.proposerDutiesLoop()
    indicesFut = service.validatorIndexLoop()
    syncFut = service.syncCommitteeDutiesLoop()
    prepareFut = service.proposerPreparationsLoop()
    registerFut =
      if vc.config.payloadBuilderEnable:
        service.validatorRegisterLoop()
      else:
        nil
    dynamicFuts =
      if vc.config.web3signerUpdateInterval > 0:
        mapIt(vc.config.web3SignerUrls,
              service.dynamicValidatorsLoop(it, vc.config.web3signerUpdateInterval))
      else:
        debug "Dynamic validators update loop disabled"
        @[]
    slashPruningFut = service.slashingDatabasePruningLoop()
    web3SignerUrls = vc.config.web3SignerUrls

  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:
        var futures = @[
          FutureBase(attestFut),
          FutureBase(proposeFut),
          FutureBase(indicesFut),
          FutureBase(syncFut),
          FutureBase(prepareFut),
          FutureBase(slashPruningFut)
        ]
        for fut in dynamicFuts:
          futures.add fut
        if not(isNil(registerFut)): futures.add(FutureBase(registerFut))
        discard await race(futures)
        checkAndRestart(AttesterLoop, attestFut, service.attesterDutiesLoop())
        checkAndRestart(ProposerLoop, proposeFut, service.proposerDutiesLoop())
        checkAndRestart(IndicesLoop, indicesFut, service.validatorIndexLoop())
        checkAndRestart(SyncCommitteeLoop, syncFut,
                        service.syncCommitteeDutiesLoop())
        checkAndRestart(ProposerPreparationLoop, prepareFut,
                        service.proposerPreparationsLoop())
        if not(isNil(registerFut)):
          checkAndRestart(ValidatorRegisterLoop, registerFut,
                          service.validatorRegisterLoop())
        for i in 0 ..< dynamicFuts.len:
          checkAndRestart(DynamicValidatorsLoop, dynamicFuts[i],
                          service.dynamicValidatorsLoop(
                            web3SignerUrls[i],
                            vc.config.web3signerUpdateInterval))
        checkAndRestart(SlashPruningLoop, slashPruningFut,
                        service.slashingDatabasePruningLoop())
        false
      except CancelledError:
        debug "Service interrupted"
        var pending: seq[Future[void]]
        if not(attestFut.finished()):
          pending.add(attestFut.cancelAndWait())
        if not(proposeFut.finished()):
          pending.add(proposeFut.cancelAndWait())
        if not(indicesFut.finished()):
          pending.add(indicesFut.cancelAndWait())
        if not(syncFut.finished()):
          pending.add(syncFut.cancelAndWait())
        if not(prepareFut.finished()):
          pending.add(prepareFut.cancelAndWait())
        if not(isNil(registerFut)) and not(registerFut.finished()):
          pending.add(registerFut.cancelAndWait())
        for dynamicFut in dynamicFuts:
          if not dynamicFut.finished():
            pending.add(dynamicFut.cancelAndWait())
        if not(isNil(service.pollingAttesterDutiesTask)) and
           not(service.pollingAttesterDutiesTask.finished()):
          pending.add(service.pollingAttesterDutiesTask.cancelAndWait())
        if not(isNil(service.pollingSyncDutiesTask)) and
           not(service.pollingSyncDutiesTask.finished()):
          pending.add(service.pollingSyncDutiesTask.cancelAndWait())
        if not(isNil(service.pruneSlashingDatabaseTask)) and
           not(service.pruneSlashingDatabaseTask.finished()):
          pending.add(service.pruneSlashingDatabaseTask.cancelAndWait())
        await allFutures(pending)
        true
      except CatchableError as exc:
        warn "Service crashed with unexpected error", err_name = exc.name,
             err_msg = exc.msg
        true

    if breakLoop:
      break

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

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