Attestation API updates for Electra (#6557)

* new V2 endpoint for beacon getBlockAttestations

* nnew GET endpoint version (V2) for getPoolAttestations

* new POST endpoint version (V2) for submitPoolAttestations

* remove premature ncli tests

* review improvements

* review comments and increased test coverage

* small improvements

* documentation typos

---------

Co-authored-by: Pedro Miranda <pedro.miranda@nimbus.team>

beacon_chain/consensus_object_pools/attestation_pool.nim

@@ -570,6 +570,48 @@ iterator attestations*(
         for v in entry.aggregates:
           yield entry.toAttestation(v)
 
+iterator electraAttestations*(
+    pool: AttestationPool, slot: Opt[Slot],
+    committee_index: Opt[CommitteeIndex]): electra.Attestation =
+  let candidateIndices =
+    if slot.isSome():
+      let candidateIdx = pool.candidateIdx(slot.get(), true)
+      if candidateIdx.isSome():
+        candidateIdx.get() .. candidateIdx.get()
+      else:
+        1 .. 0
+    else:
+      0 ..< pool.electraCandidates.len()
+
+  for candidateIndex in candidateIndices:
+    for _, entry in pool.electraCandidates[candidateIndex]:
+      ## data.index field from phase0 is still being used while we have
+      ## 2 attestation pools (pre and post electra). Refer to template addAttToPool
+      ## at addAttestation proc.
+      if committee_index.isNone() or entry.data.index == committee_index.get():
+        var committee_bits: AttestationCommitteeBits
+        committee_bits[int(entry.data.index)] = true
+
+        var singleAttestation = electra.Attestation(
+          aggregation_bits: ElectraCommitteeValidatorsBits.init(entry.committee_len),
+          committee_bits: committee_bits,
+          data: AttestationData(
+            slot: entry.data.slot,
+            index: 0,
+            beacon_block_root: entry.data.beacon_block_root,
+            source: entry.data.source,
+            target: entry.data.target)
+        )
+
+        for index, signature in entry.singles:
+          singleAttestation.aggregation_bits.setBit(index)
+          singleAttestation.signature = signature.toValidatorSig()
+          yield singleAttestation
+          singleAttestation.aggregation_bits.clearBit(index)
+
+        for v in entry.aggregates:
+          yield entry.toElectraAttestation(v)
+
 type
   AttestationCacheKey = (Slot, uint64)
   AttestationCache[CVBType] = Table[AttestationCacheKey, CVBType] ##\

beacon_chain/rpc/rest_beacon_api.nim

@@ -1297,6 +1297,26 @@ proc installBeaconApiHandlers*(router: var RestRouter, node: BeaconNode) =
         node.dag.isFinalized(bid)
       )
 
+  # https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Beacon/getBlockAttestationsV2
+  router.api2(MethodGet,
+             "/eth/v2/beacon/blocks/{block_id}/attestations") do (
+    block_id: BlockIdent) -> RestApiResponse:
+    let
+      blockIdent = block_id.valueOr:
+        return RestApiResponse.jsonError(Http400, InvalidBlockIdValueError,
+                                         $error)
+      bdata = node.getForkedBlock(blockIdent).valueOr:
+        return RestApiResponse.jsonError(Http404, BlockNotFoundError)
+
+    withBlck(bdata):
+      let bid = BlockId(root: forkyBlck.root, slot: forkyBlck.message.slot)
+      RestApiResponse.jsonResponseFinalizedWVersion(
+        forkyBlck.message.body.attestations.asSeq(),
+        node.getBlockOptimistic(bdata),
+        node.dag.isFinalized(bid),
+        consensusFork
+      )
+
   # https://ethereum.github.io/beacon-APIs/#/Beacon/getPoolAttestations
   router.api2(MethodGet, "/eth/v1/beacon/pool/attestations") do (
     slot: Option[Slot],
@@ -1325,6 +1345,45 @@ proc installBeaconApiHandlers*(router: var RestRouter, node: BeaconNode) =
       res.add(item)
     RestApiResponse.jsonResponse(res)
 
+  # https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Beacon/getPoolAttestationsV2
+  router.api2(MethodGet, "/eth/v2/beacon/pool/attestations") do (
+    slot: Option[Slot],
+    committee_index: Option[CommitteeIndex]) -> RestApiResponse:
+    let vindex =
+      if committee_index.isSome():
+        let rindex = committee_index.get()
+        if rindex.isErr():
+          return RestApiResponse.jsonError(Http400,
+                                           InvalidCommitteeIndexValueError,
+                                           $rindex.error)
+        Opt.some(rindex.get())
+      else:
+        Opt.none(CommitteeIndex)
+    let vslot =
+      if slot.isSome():
+        let rslot = slot.get()
+        if rslot.isErr():
+          return RestApiResponse.jsonError(Http400, InvalidSlotValueError,
+                                           $rslot.error)
+        Opt.some(rslot.get())
+      else:
+        Opt.none(Slot)
+
+    let consensusFork =
+      if vslot.isNone():
+        node.dag.cfg.consensusForkAtEpoch(node.currentSlot().epoch())
+      else:
+        node.dag.cfg.consensusForkAtEpoch(vslot.get().epoch)
+
+    if consensusFork < ConsensusFork.Electra:
+      return RestApiResponse.jsonResponseWVersion(
+        toSeq(node.attestationPool[].attestations(vslot, vindex)),
+        consensusFork)
+    else:
+      return RestApiResponse.jsonResponseWVersion(
+        toSeq(node.attestationPool[].electraAttestations(vslot, vindex)),
+        consensusFork)
+
   # https://ethereum.github.io/beacon-APIs/#/Beacon/submitPoolAttestations
   router.api2(MethodPost, "/eth/v1/beacon/pool/attestations") do (
     contentBody: Option[ContentBody]) -> RestApiResponse:
@@ -1342,11 +1401,7 @@ proc installBeaconApiHandlers*(router: var RestRouter, node: BeaconNode) =
     # Since our validation logic supports batch processing, we will submit all
     # attestations for validation.
     let pending =
-      block:
-        var res: seq[Future[SendResult]]
-        for attestation in attestations:
-          res.add(node.router.routeAttestation(attestation))
-        res
+      mapIt(attestations, node.router.routeAttestation(it))
     let failures =
       block:
         var res: seq[RestIndexedErrorMessageItem]
@@ -1372,6 +1427,62 @@ proc installBeaconApiHandlers*(router: var RestRouter, node: BeaconNode) =
     else:
       RestApiResponse.jsonMsgResponse(AttestationValidationSuccess)
 
+  # https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Beacon/submitPoolAttestationsV2
+  router.api2(MethodPost, "/eth/v2/beacon/pool/attestations") do (
+    contentBody: Option[ContentBody]) -> RestApiResponse:
+
+    let
+      headerVersion = request.headers.getString("Eth-Consensus-Version")
+      consensusVersion = ConsensusFork.init(headerVersion)
+    if consensusVersion.isNone():
+      return RestApiResponse.jsonError(Http400, FailedToObtainConsensusForkError)
+
+    if contentBody.isNone():
+          return RestApiResponse.jsonError(Http400, EmptyRequestBodyError)
+
+    var pendingAttestations: seq[Future[SendResult]]
+    template decodeAttestations(AttestationType: untyped) =
+      let dres = decodeBody(seq[AttestationType], contentBody.get())
+      if dres.isErr():
+        return RestApiResponse.jsonError(Http400,
+                                          InvalidAttestationObjectError,
+                                          $dres.error)
+      # Since our validation logic supports batch processing, we will submit all
+      # attestations for validation.
+      for attestation in dres.get():
+        pendingAttestations.add(node.router.routeAttestation(attestation))
+
+    case consensusVersion.get():
+      of ConsensusFork.Phase0 .. ConsensusFork.Deneb:
+        decodeAttestations(phase0.Attestation)
+      of ConsensusFork.Electra:
+        decodeAttestations(electra.Attestation)
+
+    let failures =
+      block:
+        var res: seq[RestIndexedErrorMessageItem]
+        await allFutures(pendingAttestations)
+        for index, future in pendingAttestations:
+          if future.completed():
+            let fres = future.value()
+            if fres.isErr():
+              let failure = RestIndexedErrorMessageItem(index: index,
+                                                        message: $fres.error)
+              res.add(failure)
+          elif future.failed() or future.cancelled():
+            # This is unexpected failure, so we log the error message.
+            let exc = future.error()
+            let failure = RestIndexedErrorMessageItem(index: index,
+                                                      message: $exc.msg)
+            res.add(failure)
+        res
+
+    if len(failures) > 0:
+      RestApiResponse.jsonErrorList(Http400, AttestationValidationError,
+                                    failures)
+    else:
+      RestApiResponse.jsonMsgResponse(AttestationValidationSuccess)
+
   # https://ethereum.github.io/beacon-APIs/#/Beacon/getPoolAttesterSlashings
   router.api2(MethodGet, "/eth/v1/beacon/pool/attester_slashings") do (
     ) -> RestApiResponse:

beacon_chain/spec/eth2_apis/eth2_rest_serialization.nim

@@ -699,6 +699,31 @@ proc jsonResponseFinalized*(t: typedesc[RestApiResponse], data: auto,
   let res = RestApiResponse.prepareJsonResponseFinalized(data, exec, finalized)
   RestApiResponse.response(res, Http200, "application/json")
 
+proc jsonResponseFinalizedWVersion*(t: typedesc[RestApiResponse],
+                            data: auto,
+                            exec: Opt[bool],
+                            finalized: bool,
+                            version: ConsensusFork): RestApiResponse =
+  let
+    headers = [("eth-consensus-version", version.toString())]
+    res =
+      block:
+        var default: seq[byte]
+        try:
+          var stream = memoryOutput()
+          var writer = JsonWriter[RestJson].init(stream)
+          writer.beginRecord()
+          writer.writeField("version", version.toString())
+          if exec.isSome():
+            writer.writeField("execution_optimistic", exec.get())
+          writer.writeField("finalized", finalized)
+          writer.writeField("data", data)
+          writer.endRecord()
+          stream.getOutput(seq[byte])
+        except IOError:
+          default
+  RestApiResponse.response(res, Http200, "application/json", headers = headers)
+
 proc jsonResponseWVersion*(t: typedesc[RestApiResponse], data: auto,
                            version: ConsensusFork): RestApiResponse =
   let
@@ -787,18 +812,16 @@ proc jsonResponseWMeta*(t: typedesc[RestApiResponse],
 proc jsonMsgResponse*(t: typedesc[RestApiResponse],
                       msg: string = ""): RestApiResponse =
   let data =
-    block:
-      var default: seq[byte]
-      try:
-        var stream = memoryOutput()
-        var writer = JsonWriter[RestJson].init(stream)
-        writer.beginRecord()
-        writer.writeField("code", 200)
-        writer.writeField("message", msg)
-        writer.endRecord()
-        stream.getOutput(seq[byte])
-      except IOError:
-        default
+    try:
+      var stream = memoryOutput()
+      var writer = JsonWriter[RestJson].init(stream)
+      writer.beginRecord()
+      writer.writeField("code", 200)
+      writer.writeField("message", msg)
+      writer.endRecord()
+      stream.getOutput(seq[byte])
+    except IOError:
+      default(seq[byte])
   RestApiResponse.response(data, Http200, "application/json")
 
 proc jsonError*(t: typedesc[RestApiResponse], status: HttpCode = Http200,

beacon_chain/spec/eth2_apis/rest_beacon_calls.nim

@@ -313,6 +313,12 @@ proc getBlockAttestations*(block_id: BlockIdent
      meth: MethodGet.}
   ## https://ethereum.github.io/beacon-APIs/#/Beacon/getBlockAttestations
 
+proc getBlockAttestationsV2Plain*(block_id: BlockIdent
+                        ): RestPlainResponse {.
+     rest, endpoint: "/eth/v2/beacon/blocks/{block_id}/attestations",
+     meth: MethodGet.}
+  ## https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Beacon/getBlockAttestationsV2
+
 proc getPoolAttestations*(
     slot: Option[Slot],
     committee_index: Option[CommitteeIndex]
@@ -321,12 +327,27 @@ proc getPoolAttestations*(
      meth: MethodGet.}
   ## https://ethereum.github.io/beacon-APIs/#/Beacon/getPoolAttestations
 
+proc getPoolAttestationsV2Plain*(
+    slot: Option[Slot],
+    committee_index: Option[CommitteeIndex]
+              ): RestPlainResponse {.
+     rest, endpoint: "/eth/v2/beacon/pool/attestations",
+     meth: MethodGet.}
+  ## https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Beacon/getPoolAttestationsV2
+
 proc submitPoolAttestations*(body: seq[phase0.Attestation]):
      RestPlainResponse {.
      rest, endpoint: "/eth/v1/beacon/pool/attestations",
      meth: MethodPost.}
   ## https://ethereum.github.io/beacon-APIs/#/Beacon/submitPoolAttestations
 
+proc submitPoolAttestationsV2*(
+       body: seq[phase0.Attestation] | seq[electra.Attestation]):
+     RestPlainResponse {.
+     rest, endpoint: "/eth/v2/beacon/pool/attestations",
+     meth: MethodPost.}
+  ## https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Beacon/submitPoolAttestationsV2
+
 proc getPoolAttesterSlashings*(): RestResponse[GetPoolAttesterSlashingsResponse] {.
      rest, endpoint: "/eth/v1/beacon/pool/attester_slashings",
      meth: MethodGet.}

beacon_chain/validators/message_router.nim

@@ -235,7 +235,9 @@ proc routeAttestation*(
   return ok()
 
 proc routeAttestation*(
-    router: ref MessageRouter, attestation: phase0.Attestation | electra.Attestation):
+    router: ref MessageRouter,
+    attestation: phase0.Attestation | electra.Attestation,
+    on_chain: static bool = false):
     Future[SendResult] {.async: (raises: [CancelledError]).} =
   # Compute subnet, then route attestation
   let
@@ -252,7 +254,7 @@ proc routeAttestation*(
         attestation = shortLog(attestation)
       return
     committee_index =
-      shufflingRef.get_committee_index(attestation.committee_index()).valueOr:
+      shufflingRef.get_committee_index(attestation.committee_index(on_chain)).valueOr:
         notice "Invalid committee index in attestation",
           attestation = shortLog(attestation)
         return err("Invalid committee index in attestation")