Archive Validator Participation on End of Epoch Event (#3524)

* archive participation begin implementation

* validator participation compute

* comments

* compute participation common func

* full test for archiving data

* gazelle

* complete tests

* gaz

* remove double negative grammar in comment

* use head fetcher and root

* tests pass

beacon-chain/archiver/BUILD.bazel

@@ -7,7 +7,11 @@ go_library(
     visibility = ["//beacon-chain:__subpackages__"],
     deps = [
         "//beacon-chain/blockchain:go_default_library",
+        "//beacon-chain/core/epoch:go_default_library",
+        "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/db:go_default_library",
+        "//proto/beacon/p2p/v1:go_default_library",
+        "@com_github_pkg_errors//:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
     ],
 )
@@ -18,7 +22,15 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//beacon-chain/blockchain/testing:go_default_library",
+        "//beacon-chain/core/helpers:go_default_library",
+        "//beacon-chain/db/testing:go_default_library",
+        "//proto/beacon/p2p/v1:go_default_library",
+        "//proto/eth/v1alpha1:go_default_library",
+        "//shared/params:go_default_library",
         "//shared/testutil:go_default_library",
+        "@com_github_gogo_protobuf//proto:go_default_library",
+        "@com_github_prysmaticlabs_go_bitfield//:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
         "@com_github_sirupsen_logrus//hooks/test:go_default_library",
     ],
 )

beacon-chain/archiver/service.go

@@ -4,8 +4,12 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/epoch"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
 	"github.com/prysmaticlabs/prysm/beacon-chain/db"
+	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	"github.com/sirupsen/logrus"
 )
 
@@ -17,6 +21,7 @@ type Service struct {
 	ctx             context.Context
 	cancel          context.CancelFunc
 	beaconDB        db.Database
+	headFetcher     blockchain.HeadFetcher
 	newHeadNotifier blockchain.NewHeadNotifier
 	newHeadRootChan chan [32]byte
 }
@@ -24,6 +29,7 @@ type Service struct {
 // Config options for the archiver service.
 type Config struct {
 	BeaconDB        db.Database
+	HeadFetcher     blockchain.HeadFetcher
 	NewHeadNotifier blockchain.NewHeadNotifier
 }
 
@@ -34,6 +40,7 @@ func NewArchiverService(ctx context.Context, cfg *Config) *Service {
 		ctx:             ctx,
 		cancel:          cancel,
 		beaconDB:        cfg.BeaconDB,
+		headFetcher:     cfg.HeadFetcher,
 		newHeadNotifier: cfg.NewHeadNotifier,
 		newHeadRootChan: make(chan [32]byte, 1),
 	}
@@ -58,13 +65,34 @@ func (s *Service) Status() error {
 	return nil
 }
 
+// We compute participation metrics by first retrieving the head state and
+// matching validator attestations during the epoch.
+func (s *Service) archiveParticipation(headState *pb.BeaconState) error {
+	participation, err := epoch.ComputeValidatorParticipation(headState)
+	if err != nil {
+		return errors.Wrap(err, "could not compute participation")
+	}
+	return s.beaconDB.SaveArchivedValidatorParticipation(s.ctx, helpers.SlotToEpoch(headState.Slot), participation)
+}
+
 func (s *Service) run() {
 	sub := s.newHeadNotifier.HeadUpdatedFeed().Subscribe(s.newHeadRootChan)
 	defer sub.Unsubscribe()
 	for {
 		select {
-		case h := <-s.newHeadRootChan:
-			log.WithField("headRoot", fmt.Sprintf("%#x", h)).Info("New chain head event")
+		case r := <-s.newHeadRootChan:
+			log.WithField("headRoot", fmt.Sprintf("%#x", r)).Debug("New chain head event")
+			headState := s.headFetcher.HeadState()
+			if !helpers.IsEpochEnd(headState.Slot) {
+				continue
+			}
+			if err := s.archiveParticipation(headState); err != nil {
+				log.WithError(err).Error("Could not archive validator participation")
+			}
+			log.WithField(
+				"epoch",
+				helpers.SlotToEpoch(headState.Slot),
+			).Debug("Successfully archived validator participation during epoch")
 		case <-s.ctx.Done():
 			log.Info("Context closed, exiting goroutine")
 			return

beacon-chain/archiver/service_test.go

@@ -3,14 +3,29 @@ package archiver
 import (
 	"context"
 	"fmt"
+	"io/ioutil"
 	"testing"
 
+	"github.com/gogo/protobuf/proto"
+	"github.com/prysmaticlabs/go-bitfield"
 	mock "github.com/prysmaticlabs/prysm/beacon-chain/blockchain/testing"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
+	dbutil "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
+	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
+	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/prysmaticlabs/prysm/shared/testutil"
+	"github.com/sirupsen/logrus"
 	logTest "github.com/sirupsen/logrus/hooks/test"
 )
 
-func TestArchiverService_ReceivesNewChainHead(t *testing.T) {
+func init() {
+	logrus.SetLevel(logrus.DebugLevel)
+	logrus.SetOutput(ioutil.Discard)
+	params.OverrideBeaconConfig(params.MinimalSpecConfig())
+}
+
+func TestArchiverService_ReceivesNewChainHeadEvent(t *testing.T) {
 	hook := logTest.NewGlobal()
 	ctx, cancel := context.WithCancel(context.Background())
 	svc := &Service{
@@ -19,6 +34,9 @@ func TestArchiverService_ReceivesNewChainHead(t *testing.T) {
 		newHeadRootChan: make(chan [32]byte, 0),
 		newHeadNotifier: &mock.ChainService{},
 	}
+	svc.headFetcher = &mock.ChainService{
+		State: &pb.BeaconState{Slot: 1},
+	}
 	exitRoutine := make(chan bool)
 	go func() {
 		svc.run()
@@ -38,3 +56,129 @@ func TestArchiverService_ReceivesNewChainHead(t *testing.T) {
 	testutil.AssertLogsContain(t, hook, fmt.Sprintf("%#x", [32]byte{1, 2, 3}))
 	testutil.AssertLogsContain(t, hook, "New chain head event")
 }
+
+func TestArchiverService_ComputesAndSavesParticipation(t *testing.T) {
+	hook := logTest.NewGlobal()
+	db := dbutil.SetupDB(t)
+	defer dbutil.TeardownDB(t, db)
+	attestedBalance := uint64(1)
+	validatorCount := uint64(100)
+
+	validators := make([]*ethpb.Validator, validatorCount)
+	balances := make([]uint64, validatorCount)
+	for i := 0; i < len(validators); i++ {
+		validators[i] = &ethpb.Validator{
+			ExitEpoch:        params.BeaconConfig().FarFutureEpoch,
+			EffectiveBalance: params.BeaconConfig().MaxEffectiveBalance,
+		}
+		balances[i] = params.BeaconConfig().MaxEffectiveBalance
+	}
+
+	atts := []*pb.PendingAttestation{{Data: &ethpb.AttestationData{Crosslink: &ethpb.Crosslink{Shard: 0}, Target: &ethpb.Checkpoint{}}}}
+	var crosslinks []*ethpb.Crosslink
+	for i := uint64(0); i < params.BeaconConfig().ShardCount; i++ {
+		crosslinks = append(crosslinks, &ethpb.Crosslink{
+			StartEpoch: 0,
+			DataRoot:   []byte{'A'},
+		})
+	}
+
+	// We initialize a head state that has attestations from participated
+	// validators in a simulated fashion.
+	headState := &pb.BeaconState{
+		Slot:                       (2 * params.BeaconConfig().SlotsPerEpoch) - 1,
+		Validators:                 validators,
+		Balances:                   balances,
+		BlockRoots:                 make([][]byte, 128),
+		Slashings:                  []uint64{0, 1e9, 1e9},
+		RandaoMixes:                make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector),
+		ActiveIndexRoots:           make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector),
+		CompactCommitteesRoots:     make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector),
+		CurrentCrosslinks:          crosslinks,
+		CurrentEpochAttestations:   atts,
+		FinalizedCheckpoint:        &ethpb.Checkpoint{},
+		JustificationBits:          bitfield.Bitvector4{0x00},
+		CurrentJustifiedCheckpoint: &ethpb.Checkpoint{},
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	svc := &Service{
+		beaconDB:        db,
+		ctx:             ctx,
+		cancel:          cancel,
+		newHeadRootChan: make(chan [32]byte, 0),
+		newHeadNotifier: &mock.ChainService{},
+	}
+	svc.headFetcher = &mock.ChainService{
+		State: headState,
+	}
+	exitRoutine := make(chan bool)
+	go func() {
+		svc.run()
+		<-exitRoutine
+	}()
+
+	// Upon receiving a new head.
+	svc.newHeadRootChan <- [32]byte{}
+	if err := svc.Stop(); err != nil {
+		t.Fatal(err)
+	}
+	exitRoutine <- true
+
+	// The context should have been canceled.
+	if svc.ctx.Err() != context.Canceled {
+		t.Error("context was not canceled")
+	}
+
+	wanted := &ethpb.ValidatorParticipation{
+		Epoch:                   helpers.SlotToEpoch(headState.Slot),
+		VotedEther:              attestedBalance,
+		EligibleEther:           validatorCount * params.BeaconConfig().MaxEffectiveBalance,
+		GlobalParticipationRate: float32(attestedBalance) / float32(validatorCount*params.BeaconConfig().MaxEffectiveBalance),
+	}
+
+	retrieved, err := svc.beaconDB.ArchivedValidatorParticipation(ctx, wanted.Epoch)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !proto.Equal(wanted, retrieved) {
+		t.Errorf("Wanted participation for epoch %d %v, retrieved %v", wanted.Epoch, wanted, retrieved)
+	}
+	testutil.AssertLogsContain(t, hook, "archived validator participation")
+}
+
+func TestArchiverService_OnlyArchiveAtEpochEnd(t *testing.T) {
+	hook := logTest.NewGlobal()
+	ctx, cancel := context.WithCancel(context.Background())
+	svc := &Service{
+		ctx:             ctx,
+		cancel:          cancel,
+		newHeadRootChan: make(chan [32]byte, 0),
+		newHeadNotifier: &mock.ChainService{},
+	}
+	exitRoutine := make(chan bool)
+	go func() {
+		svc.run()
+		<-exitRoutine
+	}()
+
+	// The head state is NOT an epoch end.
+	svc.headFetcher = &mock.ChainService{
+		State: &pb.BeaconState{Slot: params.BeaconConfig().SlotsPerEpoch - 3},
+	}
+	svc.newHeadRootChan <- [32]byte{}
+	if err := svc.Stop(); err != nil {
+		t.Fatal(err)
+	}
+	exitRoutine <- true
+
+	// The context should have been canceled.
+	if svc.ctx.Err() != context.Canceled {
+		t.Error("context was not canceled")
+	}
+	testutil.AssertLogsContain(t, hook, "New chain head event")
+	// The service should ONLY log any archival logs if we receive a
+	// head slot that is an epoch end.
+	testutil.AssertLogsDoNotContain(t, hook, "Successfully archived validator participation during epoch")
+}

beacon-chain/blockchain/service.go

@@ -198,7 +198,8 @@ func (s *Service) StateInitializedFeed() *event.Feed {
 	return s.stateInitializedFeed
 }
 
-// HeadUpdatedFeed is a feed that is written to when a new head block is saved to DB.
+// HeadUpdatedFeed is a feed containing the head block root and
+// is written to when a new head block is saved to DB.
 func (s *Service) HeadUpdatedFeed() *event.Feed {
 	return s.headUpdatedFeed
 }

beacon-chain/core/epoch/BUILD.bazel

@@ -2,7 +2,10 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["epoch_processing.go"],
+    srcs = [
+        "epoch_processing.go",
+        "participation.go",
+    ],
     importpath = "github.com/prysmaticlabs/prysm/beacon-chain/core/epoch",
     visibility = ["//beacon-chain:__subpackages__"],
     deps = [
@@ -21,7 +24,10 @@ go_library(
 go_test(
     name = "go_default_test",
     size = "small",
-    srcs = ["epoch_processing_test.go"],
+    srcs = [
+        "epoch_processing_test.go",
+        "participation_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
         "//beacon-chain/core/helpers:go_default_library",

beacon-chain/core/epoch/participation.go

@@ -0,0 +1,35 @@
+package epoch
+
+import (
+	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
+	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
+)
+
+// ComputeValidatorParticipation by matching validator attestations during the epoch,
+// computing the attesting balance, and how much attested compared to the total balances.
+func ComputeValidatorParticipation(state *pb.BeaconState) (*ethpb.ValidatorParticipation, error) {
+	currentEpoch := helpers.SlotToEpoch(state.Slot)
+	finalized := currentEpoch == state.FinalizedCheckpoint.Epoch
+
+	atts, err := MatchAttestations(state, currentEpoch)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not retrieve head attestations")
+	}
+	attestedBalances, err := AttestingBalance(state, atts.Target)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not retrieve attested balances")
+	}
+	totalBalances, err := helpers.TotalActiveBalance(state)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not retrieve total balances")
+	}
+	return &ethpb.ValidatorParticipation{
+		Epoch:                   currentEpoch,
+		Finalized:               finalized,
+		GlobalParticipationRate: float32(attestedBalances) / float32(totalBalances),
+		VotedEther:              attestedBalances,
+		EligibleEther:           totalBalances,
+	}, nil
+}

beacon-chain/core/epoch/participation_test.go

@@ -0,0 +1,69 @@
+package epoch
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/prysmaticlabs/go-bitfield"
+	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
+	"github.com/prysmaticlabs/prysm/shared/params"
+)
+
+func TestComputeValidatorParticipation(t *testing.T) {
+	params.OverrideBeaconConfig(params.MinimalSpecConfig())
+	epoch := uint64(1)
+	attestedBalance := uint64(1)
+	validatorCount := uint64(100)
+
+	validators := make([]*ethpb.Validator, validatorCount)
+	balances := make([]uint64, validatorCount)
+	for i := 0; i < len(validators); i++ {
+		validators[i] = &ethpb.Validator{
+			ExitEpoch:        params.BeaconConfig().FarFutureEpoch,
+			EffectiveBalance: params.BeaconConfig().MaxEffectiveBalance,
+		}
+		balances[i] = params.BeaconConfig().MaxEffectiveBalance
+	}
+
+	atts := []*pb.PendingAttestation{{Data: &ethpb.AttestationData{Crosslink: &ethpb.Crosslink{Shard: 0}, Target: &ethpb.Checkpoint{}}}}
+	var crosslinks []*ethpb.Crosslink
+	for i := uint64(0); i < params.BeaconConfig().ShardCount; i++ {
+		crosslinks = append(crosslinks, &ethpb.Crosslink{
+			StartEpoch: 0,
+			DataRoot:   []byte{'A'},
+		})
+	}
+
+	s := &pb.BeaconState{
+		Slot:                       epoch*params.BeaconConfig().SlotsPerEpoch + 1,
+		Validators:                 validators,
+		Balances:                   balances,
+		BlockRoots:                 make([][]byte, 128),
+		Slashings:                  []uint64{0, 1e9, 1e9},
+		RandaoMixes:                make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector),
+		ActiveIndexRoots:           make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector),
+		CompactCommitteesRoots:     make([][]byte, params.BeaconConfig().EpochsPerHistoricalVector),
+		CurrentCrosslinks:          crosslinks,
+		CurrentEpochAttestations:   atts,
+		FinalizedCheckpoint:        &ethpb.Checkpoint{},
+		JustificationBits:          bitfield.Bitvector4{0x00},
+		CurrentJustifiedCheckpoint: &ethpb.Checkpoint{},
+	}
+
+	res, err := ComputeValidatorParticipation(s)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	wanted := &ethpb.ValidatorParticipation{
+		Epoch:                   epoch,
+		VotedEther:              attestedBalance,
+		EligibleEther:           validatorCount * params.BeaconConfig().MaxEffectiveBalance,
+		GlobalParticipationRate: float32(attestedBalance) / float32(validatorCount*params.BeaconConfig().MaxEffectiveBalance),
+	}
+
+	if !reflect.DeepEqual(res, wanted) {
+		t.Errorf("Incorrect validator participation, wanted %v received %v", wanted, res)
+	}
+}

beacon-chain/core/validators/validator_test.go

@@ -161,5 +161,4 @@ func TestSlashValidator_OK(t *testing.T) {
 		t.Errorf("Did not get expected balance for slashed validator, wanted %d but got %d",
 			state.Validators[slashedIdx].EffectiveBalance/params.BeaconConfig().MinSlashingPenaltyQuotient, state.Balances[slashedIdx])
 	}
-
 }

beacon-chain/rpc/beacon_chain_server.go

@@ -483,30 +483,10 @@ func (bs *BeaconChainServer) ListValidatorAssignments(
 func (bs *BeaconChainServer) GetValidatorParticipation(
 	ctx context.Context, req *ethpb.GetValidatorParticipationRequest,
 ) (*ethpb.ValidatorParticipation, error) {
-
 	headState := bs.headFetcher.HeadState()
-	currentEpoch := helpers.SlotToEpoch(headState.Slot)
-	finalized := currentEpoch == headState.FinalizedCheckpoint.Epoch
-
-	atts, err := epoch.MatchAttestations(headState, currentEpoch)
+	participation, err := epoch.ComputeValidatorParticipation(headState)
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "could not retrieve head attestations: %v", err)
+		return nil, status.Errorf(codes.Internal, "could not compute participation: %v", err)
 	}
-	attestedBalances, err := epoch.AttestingBalance(headState, atts.Target)
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "could not retrieve attested balances: %v", err)
-	}
-
-	totalBalances, err := helpers.TotalActiveBalance(headState)
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "could not retrieve total balances: %v", err)
-	}
-
-	return &ethpb.ValidatorParticipation{
-		Epoch:                   currentEpoch,
-		Finalized:               finalized,
-		GlobalParticipationRate: float32(attestedBalances) / float32(totalBalances),
-		VotedEther:              attestedBalances,
-		EligibleEther:           totalBalances,
-	}, nil
+	return participation, nil
 }