Re-enable prysm aggregation

CHANGELOG.md

@@ -1,4 +1,5 @@
 Development
+  - use go-eth2-client for all beacon node communications
   - beacon block proposal strategy now scales per-node scores based on the distance between the slot and its parent
   - add a default process concurrency for strategies
   - fix race condition in "first" beacon block proposal strategy

go.mod

@@ -6,7 +6,7 @@ require (
 	cloud.google.com/go v0.72.0 // indirect
 	github.com/OneOfOne/xxhash v1.2.5 // indirect
 	github.com/attestantio/dirk v0.9.3
-	github.com/attestantio/go-eth2-client v0.6.10
+	github.com/attestantio/go-eth2-client v0.6.12
 	github.com/aws/aws-sdk-go v1.35.28
 	github.com/ferranbt/fastssz v0.0.0-20201030134205-9b9624098321
 	github.com/goccy/go-yaml v1.8.4 // indirect
@@ -46,5 +46,3 @@ require (
 	google.golang.org/grpc v1.33.2
 	gotest.tools v2.2.0+incompatible
 )
-
-replace github.com/attestantio/go-eth2-client => ../go-eth2-client

go.sum

@@ -73,6 +73,8 @@ github.com/attestantio/dirk v0.9.3/go.mod h1:EfppeT+VjQXnE9Ti5/vxa6ptZJAN2vMXO6K
 github.com/attestantio/go-eth2-client v0.6.9/go.mod h1:ODAZ4yS1YYYew/EsgGsVb/siNEoa505CrGsvlVFdkfo=
 github.com/attestantio/go-eth2-client v0.6.10 h1:PMNBMLk6xfMEUqhaUnsI0/HZRrstZF18Gt6Dm5GelW4=
 github.com/attestantio/go-eth2-client v0.6.10/go.mod h1:ODAZ4yS1YYYew/EsgGsVb/siNEoa505CrGsvlVFdkfo=
+github.com/attestantio/go-eth2-client v0.6.12 h1:7yha5QtDg5xsdTPIc8GWba97+gBkOhmowLVFzRCjdoc=
+github.com/attestantio/go-eth2-client v0.6.12/go.mod h1:ODAZ4yS1YYYew/EsgGsVb/siNEoa505CrGsvlVFdkfo=
 github.com/aws/aws-lambda-go v1.13.3/go.mod h1:4UKl9IzQMoD+QF79YdCuzCwp8VbmG4VAQwij/eHl5CU=
 github.com/aws/aws-sdk-go v1.27.0/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go v1.33.5/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0=

services/attestationaggregator/service.go

@@ -17,6 +17,7 @@ import (
 	"context"
 
 	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/attestantio/vouch/services/accountmanager"
 )
 
 // Duty contains information about an attestation aggregation duty.
@@ -29,6 +30,12 @@ type Duty struct {
 	ValidatorIndex spec.ValidatorIndex
 	// SlotSignature is the signature of the slot by the validator carrying out the aggregation; reuqired for submitting the aggregate.
 	SlotSignature spec.BLSSignature
+	// Attestation is the attestation from the validator that is part of the related to the aggregate.
+	// Required for Prysm non-spec GRPC method.
+	Attestation *spec.Attestation
+	// Account is the account carrying out the aggregation.
+	// Required for Prysm non-spec GRPC method.
+	Account accountmanager.ValidatingAccount
 }
 
 // IsAggregatorProvider provides information about if a validator is an aggregator.

services/attestationaggregator/standard/service.go

@@ -93,9 +93,11 @@ func (s *Service) Aggregate(ctx context.Context, data interface{}) {
 	if s.aggregateAttestationProvider != nil {
 		aggregateAttestation, err = s.aggregateAttestationProvider.AggregateAttestation(ctx, duty.Slot, duty.AttestationDataRoot)
 	} else {
-		// TODO
-		log.Debug().Msg("Not aggregating for non-spec beacon node")
-		return
+		validatorPubKey, err := duty.Account.PubKey(ctx)
+		if err != nil {
+			log.Error().Err(err).Msg("Failed to obtain validator public key")
+		}
+		aggregateAttestation, err = s.prysmAggregateAttestationProvider.PrysmAggregateAttestation(ctx, duty.Attestation, validatorPubKey, duty.SlotSignature)
 	}
 	if err != nil {
 		log.Error().Err(err).Msg("Failed to obtain aggregate attestation")

services/controller/standard/attester.go

@@ -145,6 +145,17 @@ func (s *Service) AttestAndScheduleAggregate(ctx context.Context, data interface
 			continue
 		}
 		if info.IsAggregator {
+			accounts, err := s.validatingAccountsProvider.AccountsByIndex(ctx, []spec.ValidatorIndex{info.Duty.ValidatorIndex})
+			if err != nil {
+				// Don't return here; we want to try to set up as many aggregator jobs as possible.
+				log.Error().Err(err).Msg("Failed to obtain accounts")
+				continue
+			}
+			if len(accounts) == 0 {
+				// Don't return here; we want to try to set up as many aggregator jobs as possible.
+				log.Error().Msg("Failed to obtain account of attester")
+				continue
+			}
 			attestationDataRoot, err := attestation.Data.HashTreeRoot()
 			if err != nil {
 				// Don't return here; we want to try to set up as many aggregator jobs as possible.
@@ -156,6 +167,8 @@ func (s *Service) AttestAndScheduleAggregate(ctx context.Context, data interface
 				AttestationDataRoot: attestationDataRoot,
 				ValidatorIndex:      info.Duty.ValidatorIndex,
 				SlotSignature:       info.Signature,
+				Account:             accounts[0],
+				Attestation:         attestation,
 			}
 			if err := s.scheduler.ScheduleJob(ctx,
 				fmt.Sprintf("Beacon block attestation aggregation for slot %d committee %d", attestation.Data.Slot, attestation.Data.Index),

services/submitter/multinode/submitbeaconblock.go

@@ -15,7 +15,6 @@ package multinode
 
 import (
 	"context"
-	"encoding/json"
 	"sync"
 	"time"
 
@@ -31,6 +30,7 @@ func (s *Service) SubmitBeaconBlock(ctx context.Context, block *spec.SignedBeaco
 		return errors.New("no beacon block supplied")
 	}
 
+	log := log.With().Uint64("slot", uint64(block.Message.Slot)).Logger()
 	sem := semaphore.NewWeighted(s.processConcurrency)
 	var wg sync.WaitGroup
 	for name, submitter := range s.beaconBlockSubmitters {
@@ -42,7 +42,7 @@ func (s *Service) SubmitBeaconBlock(ctx context.Context, block *spec.SignedBeaco
 			submitter eth2client.BeaconBlockSubmitter,
 		) {
 			defer wg.Done()
-			log := log.With().Str("beacon_node_address", name).Uint64("slot", uint64(block.Message.Slot)).Logger()
+			log := log.With().Str("submitter", name).Logger()
 			if err := sem.Acquire(ctx, 1); err != nil {
 				log.Error().Err(err).Msg("Failed to acquire semaphore")
 				return
@@ -64,13 +64,7 @@ func (s *Service) SubmitBeaconBlock(ctx context.Context, block *spec.SignedBeaco
 		}(ctx, sem, &wg, name, submitter)
 	}
 	wg.Wait()
-
-	if e := log.Trace(); e.Enabled() {
-		data, err := json.Marshal(block)
-		if err == nil {
-			e.Str("block", string(data)).Msg("Submitted beacon block")
-		}
-	}
+	log.Trace().Msg("Submitted beacon block")
 
 	return nil
 }