VC: new scoring functions.

AllTests-mainnet.md

@@ -599,11 +599,15 @@ OK: 1/1 Fail: 0/1 Skip: 0/1
 + /eth/v1/validator/sync_committee_selections serialization/deserialization test             OK
 + bestSuccess() API timeout test                                                             OK
 + firstSuccessParallel() API timeout test                                                    OK
++ getAggregatedAttestationDataScore() test vectors                                           OK
 + getAttestationDataScore() test vectors                                                     OK
 + getLiveness() response deserialization test                                                OK
++ getSyncCommitteeContributionDataScore() test vectors                                       OK
++ getSyncCommitteeMessageDataScore() test vectors                                            OK
++ getUniqueVotes() test vectors                                                              OK
 + normalizeUri() test vectors                                                                OK
 ```
-OK: 7/7 Fail: 0/7 Skip: 0/7
+OK: 11/11 Fail: 0/11 Skip: 0/11
 ## Validator change pool testing suite
 ```diff
 + addValidatorChangeMessage/getAttesterSlashingMessage                                       OK
@@ -716,4 +720,4 @@ OK: 2/2 Fail: 0/2 Skip: 0/2
 OK: 9/9 Fail: 0/9 Skip: 0/9
 
 ---TOTAL---
-OK: 405/410 Fail: 0/410 Skip: 5/410
+OK: 409/414 Fail: 0/414 Skip: 5/414

beacon_chain/spec/eth2_apis/rest_types.nim

@@ -940,3 +940,27 @@ func toValidatorIndex*(value: RestValidatorIndex): Result[ValidatorIndex,
       err(ValidatorIndexError.TooHighValue)
   else:
     doAssert(false, "ValidatorIndex type size is incorrect")
+
+template withBlck*(x: ProduceBlockResponseV2,
+                   body: untyped): untyped =
+  case x.kind
+  of ConsensusFork.Phase0:
+    const consensusFork {.inject, used.} = ConsensusFork.Phase0
+    template blck: untyped {.inject.} = x.phase0Data
+    body
+  of ConsensusFork.Altair:
+    const consensusFork {.inject, used.} = ConsensusFork.Altair
+    template blck: untyped {.inject.} = x.altairData
+    body
+  of ConsensusFork.Bellatrix:
+    const consensusFork {.inject, used.} = ConsensusFork.Bellatrix
+    template blck: untyped {.inject.} = x.bellatrixData
+    body
+  of ConsensusFork.Capella:
+    const consensusFork {.inject, used.} = ConsensusFork.Capella
+    template blck: untyped {.inject.} = x.capellaData
+    body
+  of ConsensusFork.Deneb:
+    const consensusFork {.inject, used.} = ConsensusFork.Deneb
+    template blck: untyped {.inject.} = x.denebData.blck
+    body

beacon_chain/validator_client/api.nim

@@ -1140,7 +1140,7 @@ proc getHeadBlockRoot*(
   let blockIdent = BlockIdent.init(BlockIdentType.Head)
 
   case strategy
-  of ApiStrategyKind.First, ApiStrategyKind.Best:
+  of ApiStrategyKind.First:
     let res = vc.firstSuccessParallel(RestPlainResponse,
                                       GetBlockRootResponse,
                                       SlotDuration,
@@ -1184,6 +1184,52 @@ proc getHeadBlockRoot*(
       raise (ref ValidatorApiError)(msg: res.error, data: failures)
     return res.get()
 
+  of ApiStrategyKind.Best:
+    let res = vc.bestSuccess(
+      RestPlainResponse,
+      GetBlockRootResponse,
+      SlotDuration,
+      ViableNodeStatus,
+      {BeaconNodeRole.SyncCommitteeData},
+      getBlockRootPlain(it, blockIdent),
+      getSyncCommitteeMessageDataScore(vc, itresponse)):
+      if apiResponse.isErr():
+        handleCommunicationError()
+        ApiResponse[GetBlockRootResponse].err(apiResponse.error)
+      else:
+        let response = apiResponse.get()
+        case response.status
+        of 200:
+          let res = decodeBytes(GetBlockRootResponse, response.data,
+                                response.contentType)
+          if res.isErr():
+            handleUnexpectedData()
+            ApiResponse[GetBlockRootResponse].err($res.error)
+          else:
+            let data = res.get()
+            if data.execution_optimistic.get(false):
+              handleOptimistic()
+              failures.add(failure)
+              ApiResponse[GetBlockRootResponse].err(ResponseECNotInSyncError)
+            else:
+              ApiResponse[GetBlockRootResponse].ok(data)
+        of 400:
+          handle400()
+          ApiResponse[GetBlockRootResponse].err(ResponseInvalidError)
+        of 404:
+          handle404()
+          ApiResponse[GetBlockRootResponse].err(ResponseNotFoundError)
+        of 500:
+          handle500()
+          ApiResponse[GetBlockRootResponse].err(ResponseInternalError)
+        else:
+          handleUnexpectedCode()
+          ApiResponse[GetBlockRootResponse].err(ResponseUnexpectedError)
+
+    if res.isErr():
+      raise (ref ValidatorApiError)(msg: res.error, data: failures)
+    return res.get()
+
   of ApiStrategyKind.Priority:
     vc.firstSuccessSequential(RestPlainResponse, #RestResponse[GetBlockRootResponse],
                               SlotDuration,
@@ -1603,7 +1649,7 @@ proc getAggregatedAttestation*(
   var failures: seq[ApiNodeFailure]
 
   case strategy
-  of ApiStrategyKind.First, ApiStrategyKind.Best:
+  of ApiStrategyKind.First:
     let res = vc.firstSuccessParallel(
       RestPlainResponse,
       GetAggregatedAttestationResponse,
@@ -1642,6 +1688,46 @@ proc getAggregatedAttestation*(
       raise (ref ValidatorApiError)(msg: res.error, data: failures)
     return res.get().data
 
+  of ApiStrategyKind.Best:
+    let res = vc.bestSuccess(
+      RestPlainResponse,
+      GetAggregatedAttestationResponse,
+      OneThirdDuration,
+      ViableNodeStatus,
+      {BeaconNodeRole.AggregatedData},
+      getAggregatedAttestationPlain(it, root, slot),
+      getAggregatedAttestationDataScore(itresponse)):
+      if apiResponse.isErr():
+        handleCommunicationError()
+        ApiResponse[GetAggregatedAttestationResponse].err(apiResponse.error)
+      else:
+        let response = apiResponse.get()
+        case response.status:
+        of 200:
+          let res = decodeBytes(GetAggregatedAttestationResponse, response.data,
+                                response.contentType)
+          if res.isErr():
+            handleUnexpectedData()
+            ApiResponse[GetAggregatedAttestationResponse].err($res.error)
+          else:
+            ApiResponse[GetAggregatedAttestationResponse].ok(res.get())
+        of 400:
+          handle400()
+          ApiResponse[GetAggregatedAttestationResponse].err(
+            ResponseInvalidError)
+        of 500:
+          handle500()
+          ApiResponse[GetAggregatedAttestationResponse].err(
+            ResponseInternalError)
+        else:
+          handleUnexpectedCode()
+          ApiResponse[GetAggregatedAttestationResponse].err(
+            ResponseUnexpectedError)
+
+    if res.isErr():
+      raise (ref ValidatorApiError)(msg: res.error, data: failures)
+    return res.get().data
+
   of ApiStrategyKind.Priority:
     vc.firstSuccessSequential(
       RestPlainResponse,
@@ -1687,7 +1773,7 @@ proc produceSyncCommitteeContribution*(
   var failures: seq[ApiNodeFailure]
 
   case strategy
-  of ApiStrategyKind.First, ApiStrategyKind.Best:
+  of ApiStrategyKind.First:
     let res = vc.firstSuccessParallel(
       RestPlainResponse,
       ProduceSyncCommitteeContributionResponse,
@@ -1728,6 +1814,48 @@ proc produceSyncCommitteeContribution*(
       raise (ref ValidatorApiError)(msg: res.error, data: failures)
     return res.get().data
 
+  of ApiStrategyKind.Best:
+    let res = vc.bestSuccess(
+      RestPlainResponse,
+      ProduceSyncCommitteeContributionResponse,
+      OneThirdDuration,
+      ViableNodeStatus,
+      {BeaconNodeRole.SyncCommitteeData},
+      produceSyncCommitteeContributionPlain(it, slot, subcommitteeIndex, root),
+      getSyncCommitteeContributionDataScore(itresponse)):
+      if apiResponse.isErr():
+        handleCommunicationError()
+        ApiResponse[ProduceSyncCommitteeContributionResponse].err(
+          apiResponse.error)
+      else:
+        let response = apiResponse.get()
+        case response.status:
+        of 200:
+          let res = decodeBytes(ProduceSyncCommitteeContributionResponse,
+                                response.data, response.contentType)
+          if res.isErr():
+            handleUnexpectedData()
+            ApiResponse[ProduceSyncCommitteeContributionResponse].err(
+              $res.error)
+          else:
+            ApiResponse[ProduceSyncCommitteeContributionResponse].ok(res.get())
+        of 400:
+          handle400()
+          ApiResponse[ProduceSyncCommitteeContributionResponse].err(
+            ResponseInvalidError)
+        of 500:
+          handle500()
+          ApiResponse[ProduceSyncCommitteeContributionResponse].err(
+            ResponseInternalError)
+        else:
+          handleUnexpectedCode()
+          ApiResponse[ProduceSyncCommitteeContributionResponse].err(
+            ResponseUnexpectedError)
+
+    if res.isErr():
+      raise (ref ValidatorApiError)(msg: res.error, data: failures)
+    return res.get().data
+
   of ApiStrategyKind.Priority:
     vc.firstSuccessSequential(
       RestPlainResponse,

beacon_chain/validator_client/scoring.nim

@@ -6,15 +6,38 @@
 # at your option. This file may not be copied, modified, or distributed except according to those terms.
 
 import std/strutils
+import ssz_serialization/[types, bitseqs]
+import stew/endians2
+import nimcrypto/hash
 import "."/common
 
 {.push raises: [].}
 
+type
+  CommitteeBitsArray = BitArray[int(MAX_VALIDATORS_PER_COMMITTEE)]
+  CommitteeTable = Table[CommitteeIndex, CommitteeBitsArray]
+
+const
+  DefaultCommitteeTable = default(CommitteeTable)
+  DefaultCommitteeBitsArray = default(CommitteeBitsArray)
+
 func perfectScore*(score: float64): bool =
   score == Inf
 
 proc shortScore*(score: float64): string =
-  if score == Inf: "<perfect>" else: formatFloat(score, ffDecimal, 4)
+  if score == Inf:
+    "<perfect>"
+  elif score == -Inf:
+    "<bad>"
+  else:
+    formatFloat(score, ffDecimal, 4)
+
+func getLexicographicScore(digest: Eth2Digest): float64 =
+  # We calculate score on first 8 bytes of digest.
+  let
+    dvalue = uint64.fromBytesBE(digest.data.toOpenArray(0, sizeof(uint64) - 1))
+    value = float64(dvalue) / float64(high(uint64))
+  value
 
 proc getAttestationDataScore*(rootsSeen: Table[Eth2Digest, Slot],
                               adata: ProduceAttestationDataResponse): float64 =
@@ -47,3 +70,116 @@ proc getAttestationDataScore*(rootsSeen: Table[Eth2Digest, Slot],
 proc getAttestationDataScore*(vc: ValidatorClientRef,
                               adata: ProduceAttestationDataResponse): float64 =
   getAttestationDataScore(vc.rootsSeen, adata)
+
+proc getAggregatedAttestationDataScore*(
+       adata: GetAggregatedAttestationResponse
+     ): float64 =
+  # This procedure returns score value in range [0.0000, 1.0000) and `Inf`.
+  # It returns perfect score when all the bits was set to `1`, but this could
+  # provide wrong expectation for some edge cases (when different attestations
+  # has different committee sizes), but currently this is the only viable way
+  # to return perfect score.
+  const MaxLength = int(MAX_VALIDATORS_PER_COMMITTEE)
+  doAssert(len(adata.data.aggregation_bits) <= MaxLength)
+  let
+    size = len(adata.data.aggregation_bits)
+    ones = countOnes(adata.data.aggregation_bits)
+    res =
+      if ones == size:
+        # We consider score perfect, when all bits was set to 1.
+        Inf
+      else:
+        float64(ones) / float64(size)
+
+  debug "Aggregated attestation score", attestation_data = shortLog(adata.data),
+        block_slot = adata.data.data.slot, committee_size = size,
+        ones_count = ones, score = shortScore(res)
+  res
+
+proc getSyncCommitteeContributionDataScore*(
+       cdata: ProduceSyncCommitteeContributionResponse
+     ): float64 =
+  # This procedure returns score value in range [0.0000, 1.0000) and `Inf`.
+  # It returns perfect score when all the bits was set to `1`, but this could
+  # provide wrong expectation for some edge cases (when different contributions
+  # has different committee sizes), but currently this is the only viable way
+  # to return perfect score.
+  const MaxLength = int(SYNC_SUBCOMMITTEE_SIZE)
+  doAssert(len(cdata.data.aggregation_bits) <= MaxLength)
+  let
+    size = len(cdata.data.aggregation_bits)
+    ones = countOnes(cdata.data.aggregation_bits)
+    res =
+      if ones == size:
+        # We consider score perfect, when all bits was set to 1.
+        Inf
+      else:
+        float64(ones) / float64(size)
+
+  debug "Sync committee contribution score",
+        contribution_data = shortLog(cdata.data), block_slot = cdata.data.slot,
+        committee_size = size, ones_count = ones, score = shortScore(res)
+  res
+
+proc getSyncCommitteeMessageDataScore*(
+       rootsSeen: Table[Eth2Digest, Slot],
+       currentSlot: Slot,
+       cdata: GetBlockRootResponse
+     ): float64 =
+  let
+    slot = rootsSeen.getOrDefault(cdata.data.root, FAR_FUTURE_SLOT)
+    res =
+      if cdata.execution_optimistic.get(true):
+        # Responses from the nodes which are optimistically synced only are
+        # not suitable, score it with minimal possible score.
+        -Inf
+      else:
+        if slot != FAR_FUTURE_SLOT:
+          # When `slot` has been found score value will be in range of
+          # `(1, 2]` or `Inf`.
+          if slot == currentSlot:
+            # Perfect score
+            Inf
+          else:
+            float64(1) +
+              float64(1) / (float64(1) + float64(currentSlot) - float64(slot))
+        else:
+          # Block monitoring is disabled or we missed a block, in this case
+          # score value will be in range of `(0, 1]`
+          getLexicographicScore(cdata.data.root)
+
+  debug "Sync committee message score",
+        head_block_root = shortLog(cdata.data.root), slot = slot,
+        current_slot = currentSlot, score = shortScore(res)
+  res
+
+proc getSyncCommitteeMessageDataScore*(
+       vc: ValidatorClientRef,
+       cdata: GetBlockRootResponse
+     ): float64 =
+  getSyncCommitteeMessageDataScore(
+    vc.rootsSeen, vc.beaconClock.now().slotOrZero(), cdata)
+
+proc processVotes(bits: var CommitteeBitsArray,
+                  attestation: Attestation): int =
+  doAssert(len(attestation.aggregation_bits) <= len(bits))
+  var res = 0
+  for index in 0 ..< len(attestation.aggregation_bits):
+    if attestation.aggregation_bits[index]:
+      if not(bits[index]):
+        inc(res)
+        bits[index] = true
+  res
+
+proc getUniqueVotes*(attestations: openArray[Attestation]): int =
+  var
+    res = 0
+    attested: Table[Slot, CommitteeTable]
+  for attestation in attestations:
+    let count =
+      attested.mgetOrPut(attestation.data.slot, DefaultCommitteeTable).
+        mgetOrPut(CommitteeIndex(attestation.data.index),
+                  DefaultCommitteeBitsArray).
+          processVotes(attestation)
+    res += count
+  res

beacon_chain/validator_client/sync_committee_service.nim

@@ -350,7 +350,7 @@ proc publishSyncMessagesAndContributions(service: SyncCommitteeServiceRef,
   let beaconBlockRoot =
     block:
       try:
-        let res = await vc.getHeadBlockRoot(ApiStrategyKind.First)
+        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