Save new validator index in DB (#3367)

beacon-chain/blockchain/BUILD.bazel

@@ -18,7 +18,6 @@ go_library(
         "//beacon-chain/core/blocks:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/state:go_default_library",
-        "//beacon-chain/core/validators:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/operations:go_default_library",
         "//beacon-chain/p2p:go_default_library",
@@ -56,7 +55,6 @@ go_test(
         "//beacon-chain/core/validators:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/testing:go_default_library",
-        "//beacon-chain/internal:go_default_library",
         "//beacon-chain/p2p:go_default_library",
         "//beacon-chain/powchain:go_default_library",
         "//proto/beacon/p2p/v1:go_default_library",

beacon-chain/blockchain/forkchoice/process_block.go

@@ -3,6 +3,7 @@ package forkchoice
 import (
 	"bytes"
 	"context"
+	"encoding/hex"
 	"fmt"
 
 	"github.com/pkg/errors"
@@ -14,6 +15,7 @@ import (
 	"github.com/prysmaticlabs/prysm/shared/bytesutil"
 	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/prysmaticlabs/prysm/shared/roughtime"
+	"github.com/sirupsen/logrus"
 	"go.opencensus.io/trace"
 )
 
@@ -57,6 +59,7 @@ func (s *Store) OnBlock(ctx context.Context, b *ethpb.BeaconBlock) error {
 	if err != nil {
 		return err
 	}
+	preStateValidatorCount := len(preState.Validators)
 
 	// Verify block slot time is not from the feature.
 	if err := verifyBlkSlotTime(preState.GenesisTime, b.Slot); err != nil {
@@ -110,8 +113,12 @@ func (s *Store) OnBlock(ctx context.Context, b *ethpb.BeaconBlock) error {
 		}
 	}
 
-	// Log epoch summary before the next epoch.
+	// Epoch boundary bookkeeping such as logging epoch summaries
+	// and saving newly activated validator indices in db.
 	if helpers.IsEpochStart(postState.Slot) {
+		if err := s.saveNewValidator(ctx, preStateValidatorCount, postState); err != nil {
+			return errors.Wrap(err, "could not save finalized checkpoint")
+		}
 		logEpochData(postState)
 	}
 	return nil
@@ -157,6 +164,25 @@ func (s *Store) verifyBlkFinalizedSlot(b *ethpb.BeaconBlock) error {
 	return nil
 }
 
+// saveNewValidator saves newly added validator index from state to db.
+func (s *Store) saveNewValidator(ctx context.Context, preStateValidatorCount int, postState *pb.BeaconState) error {
+	postStateValidatorCount := len(postState.Validators)
+	if preStateValidatorCount != postStateValidatorCount {
+		for i := preStateValidatorCount; i < postStateValidatorCount; i++ {
+			pubKey := postState.Validators[i].PublicKey
+			if err := s.db.SaveValidatorIndex(ctx, bytesutil.ToBytes48(pubKey), uint64(i)); err != nil {
+				return errors.Wrapf(err, "could not save activated validator: %d", i)
+			}
+			log.WithFields(logrus.Fields{
+				"index":               i,
+				"pubKey":              hex.EncodeToString(bytesutil.Trunc(pubKey)),
+				"totalValidatorCount": i + 1,
+			}).Info("New validator index saved in DB")
+		}
+	}
+	return nil
+}
+
 // verifyBlkSlotTime validates the input block slot is not from the future.
 func verifyBlkSlotTime(gensisTime uint64, blkSlot uint64) error {
 	slotTime := gensisTime + blkSlot*params.BeaconConfig().SecondsPerSlot

beacon-chain/blockchain/forkchoice/process_block_test.go

@@ -84,3 +84,29 @@ func TestStore_OnBlock(t *testing.T) {
 		})
 	}
 }
+
+func TestStore_SaveNewValidator(t *testing.T) {
+	ctx := context.Background()
+	db := testDB.SetupDB(t)
+	defer testDB.TeardownDB(t, db)
+
+	store := NewForkChoiceService(ctx, db)
+	preCount := 2 // validators 0 and validators 1
+	s := &pb.BeaconState{Validators: []*ethpb.Validator{
+		{PublicKey: []byte{0}}, {PublicKey: []byte{1}},
+		{PublicKey: []byte{2}}, {PublicKey: []byte{3}},
+	}}
+	if err := store.saveNewValidator(ctx, preCount, s); err != nil {
+		t.Fatal(err)
+	}
+
+	if !db.HasValidatorIndex(ctx, bytesutil.ToBytes48([]byte{2})) {
+		t.Error("Wanted validator saved in db")
+	}
+	if !db.HasValidatorIndex(ctx, bytesutil.ToBytes48([]byte{3})) {
+		t.Error("Wanted validator saved in db")
+	}
+	if db.HasValidatorIndex(ctx, bytesutil.ToBytes48([]byte{1})) {
+		t.Error("validator not suppose to be saved in db")
+	}
+}

beacon-chain/blockchain/receive_block.go

@@ -7,9 +7,6 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/go-ssz"
-	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
-	"github.com/prysmaticlabs/prysm/beacon-chain/core/validators"
-	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/shared/bytesutil"
 	"github.com/sirupsen/logrus"
@@ -99,13 +96,6 @@ func (c *ChainService) ReceiveBlockNoPubsub(ctx context.Context, block *ethpb.Be
 		return errors.Wrap(err, "could not clean up block deposits, attestations, and other operations")
 	}
 
-	// Update validator's public key to indices mapping in DB.
-	if helpers.IsEpochStart(block.Slot) {
-		if err := c.updateValidatorsDB(ctx, root); err != nil {
-			return errors.Wrap(err, "could not update validators db")
-		}
-	}
-
 	processedBlkNoPubsub.Inc()
 	return nil
 }
@@ -138,13 +128,6 @@ func (c *ChainService) ReceiveBlockNoPubsubForkchoice(ctx context.Context, block
 		return errors.Wrap(err, "could not clean up block deposits, attestations, and other operations")
 	}
 
-	// Update validator's public key to indices mapping in DB.
-	if helpers.IsEpochStart(block.Slot) {
-		if err := c.updateValidatorsDB(ctx, root); err != nil {
-			return errors.Wrap(err, "could not update validators db")
-		}
-	}
-
 	processedBlkNoPubsubForkchoice.Inc()
 	return nil
 }
@@ -166,63 +149,6 @@ func (c *ChainService) cleanupBlockOperations(ctx context.Context, block *ethpb.
 	return nil
 }
 
-// this updates validator's public key to indices mapping stored in DB, due to the frequent
-// validator activation and exit, we should check this every epoch.
-func (c *ChainService) updateValidatorsDB(ctx context.Context, r [32]byte) error {
-	s, err := c.beaconDB.State(ctx, r)
-	if err != nil {
-		return errors.Wrap(err, "could not retrieve latest processed state in DB")
-	}
-	if err := c.saveValidatorIdx(ctx, s); err != nil {
-		return errors.Wrap(err, "could not save validator index")
-	}
-	if err := c.deleteValidatorIdx(ctx, s); err != nil {
-		return errors.Wrap(err, "could not delete validator index")
-	}
-	return nil
-}
-
-// saveValidatorIdx saves the validators public key to index mapping in DB, these
-// validators were activated from current epoch. After it saves, current epoch key
-// is deleted from ActivatedValidators mapping.
-func (c *ChainService) saveValidatorIdx(ctx context.Context, state *pb.BeaconState) error {
-	nextEpoch := helpers.CurrentEpoch(state) + 1
-	activatedValidators := validators.ActivatedValFromEpoch(nextEpoch)
-	var idxNotInState []uint64
-	for _, idx := range activatedValidators {
-		// If for some reason the activated validator indices is not in state,
-		// we skip them and save them to process for next epoch.
-		if int(idx) >= len(state.Validators) {
-			idxNotInState = append(idxNotInState, idx)
-			continue
-		}
-		pubKey := state.Validators[idx].PublicKey
-		if err := c.beaconDB.SaveValidatorIndex(ctx, bytesutil.ToBytes48(pubKey), idx); err != nil {
-			return errors.Wrap(err, "could not save validator index")
-		}
-	}
-	// Since we are processing next epoch, save the can't processed validator indices
-	// to the epoch after that.
-	validators.InsertActivatedIndices(nextEpoch+1, idxNotInState)
-	validators.DeleteActivatedVal(helpers.CurrentEpoch(state))
-	return nil
-}
-
-// deleteValidatorIdx deletes the validators public key to index mapping in DB, the
-// validators were exited from current epoch. After it deletes, current epoch key
-// is deleted from ExitedValidators mapping.
-func (c *ChainService) deleteValidatorIdx(ctx context.Context, state *pb.BeaconState) error {
-	exitedValidators := validators.ExitedValFromEpoch(helpers.CurrentEpoch(state) + 1)
-	for _, idx := range exitedValidators {
-		pubKey := state.Validators[idx].PublicKey
-		if err := c.beaconDB.DeleteValidatorIndex(ctx, bytesutil.ToBytes48(pubKey)); err != nil {
-			return errors.Wrap(err, "could not delete validator index")
-		}
-	}
-	validators.DeleteExitedVal(helpers.CurrentEpoch(state))
-	return nil
-}
-
 // This checks if the block is from a competing chain, emits warning and updates metrics.
 func isCompetingBlock(root []byte, slot uint64, headRoot []byte, headSlot uint64) {
 	if !bytes.Equal(root[:], headRoot) {

beacon-chain/blockchain/receive_block_test.go

@@ -3,7 +3,6 @@ package blockchain
 import (
 	"bytes"
 	"context"
-	"encoding/binary"
 	"reflect"
 	"testing"
 
@@ -13,7 +12,6 @@ import (
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/state"
 	v "github.com/prysmaticlabs/prysm/beacon-chain/core/validators"
 	testDB "github.com/prysmaticlabs/prysm/beacon-chain/db/testing"
-	"github.com/prysmaticlabs/prysm/beacon-chain/internal"
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/shared/bytesutil"
@@ -154,62 +152,6 @@ func TestReceiveReceiveBlockNoPubsub_CanSaveHeadInfo(t *testing.T) {
 	}
 }
 
-func TestReceiveBlockNoPubsub_CanUpdateValidatorDB(t *testing.T) {
-	db := testDB.SetupDB(t)
-	defer testDB.TeardownDB(t, db)
-	ctx := context.Background()
-
-	chainService := setupBeaconChain(t, db)
-
-	b := &ethpb.BeaconBlock{
-		Slot: params.BeaconConfig().SlotsPerEpoch,
-		Body: &ethpb.BeaconBlockBody{}}
-	bRoot, err := ssz.SigningRoot(b)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err := db.SaveState(ctx, &pb.BeaconState{
-		Validators: []*ethpb.Validator{
-			{PublicKey: []byte{'A'}},
-			{PublicKey: []byte{'B'}},
-			{PublicKey: []byte{'C'}},
-			{PublicKey: []byte{'D'}},
-		},
-	}, bRoot); err != nil {
-		t.Fatal(err)
-	}
-
-	headBlk := &ethpb.BeaconBlock{Slot: 100}
-	if err := db.SaveBlock(ctx, headBlk); err != nil {
-		t.Fatal(err)
-	}
-	r, err := ssz.SigningRoot(headBlk)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	chainService.forkChoiceStore = &store{headRoot: r[:]}
-
-	v.InsertActivatedIndices(1, []uint64{1, 2})
-
-	if err := chainService.ReceiveBlockNoPubsub(ctx, b); err != nil {
-		t.Fatal(err)
-	}
-
-	index, _, _ := db.ValidatorIndex(ctx, bytesutil.ToBytes48([]byte{'B'}))
-	if index != 1 {
-		t.Errorf("Wanted: %d, got: %d", 1, index)
-	}
-	index, _, _ = db.ValidatorIndex(ctx, bytesutil.ToBytes48([]byte{'C'}))
-	if index != 2 {
-		t.Errorf("Wanted: %d, got: %d", 2, index)
-	}
-	_, e, _ := db.ValidatorIndex(ctx, bytesutil.ToBytes48([]byte{'D'}))
-	if e == true {
-		t.Error("Index should not exist in DB")
-	}
-}
-
 func TestReceiveBlockNoPubsubForkchoice_ProcessCorrectly(t *testing.T) {
 	hook := logTest.NewGlobal()
 	db := testDB.SetupDB(t)
@@ -359,84 +301,3 @@ func TestReceiveBlockNoPubsubForkchoice_CanUpdateValidatorDB(t *testing.T) {
 		t.Error("Index should not exist in DB")
 	}
 }
-
-func TestSaveValidatorIdx_SaveRetrieveWorks(t *testing.T) {
-	db := internal.SetupDBDeprecated(t)
-	defer internal.TeardownDBDeprecated(t, db)
-	ctx := context.Background()
-	epoch := uint64(1)
-	v.InsertActivatedIndices(epoch+1, []uint64{0, 1, 2})
-	var validators []*ethpb.Validator
-	for i := 0; i < 3; i++ {
-		pubKeyBuf := make([]byte, params.BeaconConfig().BLSPubkeyLength)
-		binary.PutUvarint(pubKeyBuf, uint64(i))
-		validators = append(validators, &ethpb.Validator{
-			PublicKey: pubKeyBuf,
-		})
-	}
-	state := &pb.BeaconState{
-		Validators: validators,
-		Slot:       epoch * params.BeaconConfig().SlotsPerEpoch,
-	}
-	chainService := setupBeaconChain(t, db)
-	if err := chainService.saveValidatorIdx(ctx, state); err != nil {
-		t.Fatalf("Could not save validator idx: %v", err)
-	}
-
-	wantedIdx := uint64(2)
-	idx, _, err := chainService.beaconDB.ValidatorIndex(ctx, bytesutil.ToBytes48(validators[wantedIdx].PublicKey))
-	if err != nil {
-		t.Fatalf("Could not get validator index: %v", err)
-	}
-	if wantedIdx != idx {
-		t.Errorf("Wanted: %d, got: %d", wantedIdx, idx)
-	}
-
-	if v.ActivatedValFromEpoch(epoch) != nil {
-		t.Errorf("Activated validators mapping for epoch %d still there", epoch)
-	}
-}
-
-func TestSaveValidatorIdx_IdxNotInState(t *testing.T) {
-	db := internal.SetupDBDeprecated(t)
-	defer internal.TeardownDBDeprecated(t, db)
-	epoch := uint64(100)
-	ctx := context.Background()
-
-	// Tried to insert 5 active indices to DB with only 3 validators in state
-	v.InsertActivatedIndices(epoch+1, []uint64{0, 1, 2, 3, 4})
-	var validators []*ethpb.Validator
-	for i := 0; i < 3; i++ {
-		pubKeyBuf := make([]byte, params.BeaconConfig().BLSPubkeyLength)
-		binary.PutUvarint(pubKeyBuf, uint64(i))
-		validators = append(validators, &ethpb.Validator{
-			PublicKey: pubKeyBuf,
-		})
-	}
-	state := &pb.BeaconState{
-		Validators: validators,
-		Slot:       epoch * params.BeaconConfig().SlotsPerEpoch,
-	}
-	chainService := setupBeaconChain(t, db)
-	if err := chainService.saveValidatorIdx(ctx, state); err != nil {
-		t.Fatalf("Could not save validator idx: %v", err)
-	}
-
-	wantedIdx := uint64(2)
-	idx, _, err := chainService.beaconDB.ValidatorIndex(ctx, bytesutil.ToBytes48(validators[wantedIdx].PublicKey))
-	if err != nil {
-		t.Fatalf("Could not get validator index: %v", err)
-	}
-	if wantedIdx != idx {
-		t.Errorf("Wanted: %d, got: %d", wantedIdx, idx)
-	}
-
-	if v.ActivatedValFromEpoch(epoch) != nil {
-		t.Errorf("Activated validators mapping for epoch %d still there", epoch)
-	}
-
-	// Verify the skipped validators are included in the next epoch
-	if !reflect.DeepEqual(v.ActivatedValFromEpoch(epoch+2), []uint64{3, 4}) {
-		t.Error("Did not get wanted validator from activation queue")
-	}
-}