Add database methods for optimistic sync (#10357)

* Add database methods for optimistic sync

* Add epoch comparison

* add extra epoch comparison

* Summary instead of block

* fix tests

Co-authored-by: terence tsao <[email protected]>

beacon-chain/blockchain/chain_info.go

@@ -7,6 +7,8 @@ import (
 	types "github.com/prysmaticlabs/eth2-types"
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
 	"github.com/prysmaticlabs/prysm/beacon-chain/forkchoice"
+	doublylinkedtree "github.com/prysmaticlabs/prysm/beacon-chain/forkchoice/doubly-linked-tree"
+	"github.com/prysmaticlabs/prysm/beacon-chain/forkchoice/protoarray"
 	"github.com/prysmaticlabs/prysm/beacon-chain/state"
 	fieldparams "github.com/prysmaticlabs/prysm/config/fieldparams"
 	"github.com/prysmaticlabs/prysm/config/params"
@@ -323,13 +325,44 @@ func (s *Service) IsOptimistic(ctx context.Context) (bool, error) {
 		return false, nil
 	}
 
-	return s.cfg.ForkChoiceStore.IsOptimistic(ctx, s.head.root)
+	return s.IsOptimisticForRoot(ctx, s.head.root)
 }
 
 // IsOptimisticForRoot takes the root and slot as aguments instead of the current head
 // and returns true if it is optimistic.
 func (s *Service) IsOptimisticForRoot(ctx context.Context, root [32]byte) (bool, error) {
-	return s.cfg.ForkChoiceStore.IsOptimistic(ctx, root)
+	optimistic, err := s.cfg.ForkChoiceStore.IsOptimistic(ctx, root)
+	if err == nil {
+		return optimistic, nil
+	}
+	if err != protoarray.ErrUnknownNodeRoot && err != doublylinkedtree.ErrNilNode {
+		return false, err
+	}
+	ss, err := s.cfg.BeaconDB.StateSummary(ctx, root)
+	if err != nil {
+		return false, err
+	}
+	if ss == nil {
+		return false, errInvalidNilSummary
+	}
+
+	validatedCheckpoint, err := s.cfg.BeaconDB.LastValidatedCheckpoint(ctx)
+	if err != nil {
+		return false, err
+	}
+	if slots.ToEpoch(ss.Slot) > validatedCheckpoint.Epoch {
+		return true, nil
+	}
+
+	if slots.ToEpoch(ss.Slot)+1 < validatedCheckpoint.Epoch {
+		return false, nil
+	}
+
+	lastValidated, err := s.cfg.BeaconDB.StateSummary(ctx, bytesutil.ToBytes32(validatedCheckpoint.Root))
+	if err != nil {
+		return false, err
+	}
+	return ss.Slot > lastValidated.Slot, nil
 }
 
 // SetGenesisTime sets the genesis time of beacon chain.

beacon-chain/blockchain/chain_info_test.go

@@ -437,3 +437,79 @@ func TestService_IsOptimisticForRoot_DoublyLinkedTree(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, true, opt)
 }
+
+func TestService_IsOptimisticForRoot_DB_ProtoArray(t *testing.T) {
+	beaconDB := testDB.SetupDB(t)
+	ctx := context.Background()
+	c := &Service{cfg: &config{BeaconDB: beaconDB, ForkChoiceStore: protoarray.New(0, 0, [32]byte{})}, head: &head{slot: 101, root: [32]byte{'b'}}}
+	c.head = &head{root: params.BeaconConfig().ZeroHash}
+	b := util.NewBeaconBlock()
+	b.Block.Slot = 10
+	br, err := b.Block.HashTreeRoot()
+	require.NoError(t, err)
+	require.NoError(t, beaconDB.SaveBlock(context.Background(), wrapper.WrappedPhase0SignedBeaconBlock(b)))
+	require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: br[:], Slot: 10}))
+
+	optimisticBlock := util.NewBeaconBlock()
+	optimisticBlock.Block.Slot = 11
+	optimisticRoot, err := optimisticBlock.Block.HashTreeRoot()
+	require.NoError(t, err)
+	require.NoError(t, beaconDB.SaveBlock(context.Background(), wrapper.WrappedPhase0SignedBeaconBlock(optimisticBlock)))
+
+	validatedBlock := util.NewBeaconBlock()
+	validatedBlock.Block.Slot = 9
+	validatedRoot, err := validatedBlock.Block.HashTreeRoot()
+	require.NoError(t, err)
+	require.NoError(t, beaconDB.SaveBlock(context.Background(), wrapper.WrappedPhase0SignedBeaconBlock(validatedBlock)))
+
+	validatedCheckpoint := &ethpb.Checkpoint{Root: br[:]}
+	require.NoError(t, beaconDB.SaveLastValidatedCheckpoint(ctx, validatedCheckpoint))
+
+	require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: optimisticRoot[:], Slot: 11}))
+	optimistic, err := c.IsOptimisticForRoot(ctx, optimisticRoot)
+	require.NoError(t, err)
+	require.Equal(t, true, optimistic)
+
+	require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: validatedRoot[:], Slot: 9}))
+	validated, err := c.IsOptimisticForRoot(ctx, validatedRoot)
+	require.NoError(t, err)
+	require.Equal(t, false, validated)
+}
+
+func TestService_IsOptimisticForRoot__DB_DoublyLinkedTree(t *testing.T) {
+	beaconDB := testDB.SetupDB(t)
+	ctx := context.Background()
+	c := &Service{cfg: &config{BeaconDB: beaconDB, ForkChoiceStore: doublylinkedtree.New(0, 0)}, head: &head{slot: 101, root: [32]byte{'b'}}}
+	c.head = &head{root: params.BeaconConfig().ZeroHash}
+	b := util.NewBeaconBlock()
+	b.Block.Slot = 10
+	br, err := b.Block.HashTreeRoot()
+	require.NoError(t, err)
+	require.NoError(t, beaconDB.SaveBlock(context.Background(), wrapper.WrappedPhase0SignedBeaconBlock(b)))
+	require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: br[:], Slot: 10}))
+
+	optimisticBlock := util.NewBeaconBlock()
+	optimisticBlock.Block.Slot = 11
+	optimisticRoot, err := optimisticBlock.Block.HashTreeRoot()
+	require.NoError(t, err)
+	require.NoError(t, beaconDB.SaveBlock(context.Background(), wrapper.WrappedPhase0SignedBeaconBlock(optimisticBlock)))
+
+	validatedBlock := util.NewBeaconBlock()
+	validatedBlock.Block.Slot = 9
+	validatedRoot, err := validatedBlock.Block.HashTreeRoot()
+	require.NoError(t, err)
+	require.NoError(t, beaconDB.SaveBlock(context.Background(), wrapper.WrappedPhase0SignedBeaconBlock(validatedBlock)))
+
+	validatedCheckpoint := &ethpb.Checkpoint{Root: br[:]}
+	require.NoError(t, beaconDB.SaveLastValidatedCheckpoint(ctx, validatedCheckpoint))
+
+	require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: optimisticRoot[:], Slot: 11}))
+	optimistic, err := c.IsOptimisticForRoot(ctx, optimisticRoot)
+	require.NoError(t, err)
+	require.Equal(t, true, optimistic)
+
+	require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: validatedRoot[:], Slot: 9}))
+	validated, err := c.IsOptimisticForRoot(ctx, validatedRoot)
+	require.NoError(t, err)
+	require.Equal(t, false, validated)
+}

beacon-chain/blockchain/error.go

@@ -9,6 +9,8 @@ var (
 	errNilBestJustifiedInStore = errors.New("nil best justified checkpoint returned from store")
 	// errNilFinalizedInStore is returned when a nil finalized checkpt is returned from store.
 	errNilFinalizedInStore = errors.New("nil finalized checkpoint returned from store")
+	// errInvalidNilSummary is returned when a nil summary is returned from the DB.
+	errInvalidNilSummary = errors.New("nil summary returned from the DB")
 	// errNilParentInDB is returned when a nil parent block is returned from the DB.
 	errNilParentInDB = errors.New("nil parent block in DB")
 )

beacon-chain/db/iface/interface.go

@@ -45,6 +45,7 @@ type ReadOnlyDatabase interface {
 	HasArchivedPoint(ctx context.Context, slot types.Slot) bool
 	LastArchivedRoot(ctx context.Context) [32]byte
 	LastArchivedSlot(ctx context.Context) (types.Slot, error)
+	LastValidatedCheckpoint(ctx context.Context) (*ethpb.Checkpoint, error)
 	// Deposit contract related handlers.
 	DepositContractAddress(ctx context.Context) ([]byte, error)
 	// Powchain operations.
@@ -74,6 +75,7 @@ type NoHeadAccessDatabase interface {
 	// Checkpoint operations.
 	SaveJustifiedCheckpoint(ctx context.Context, checkpoint *ethpb.Checkpoint) error
 	SaveFinalizedCheckpoint(ctx context.Context, checkpoint *ethpb.Checkpoint) error
+	SaveLastValidatedCheckpoint(ctx context.Context, checkpoint *ethpb.Checkpoint) error
 	// Deposit contract related handlers.
 	SaveDepositContractAddress(ctx context.Context, addr common.Address) error
 	// Powchain operations.

beacon-chain/db/kv/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
         "state_summary.go",
         "state_summary_cache.go",
         "utils.go",
+        "validated_checkpoint.go",
         "wss.go",
     ],
     importpath = "github.com/prysmaticlabs/prysm/beacon-chain/db/kv",
@@ -91,6 +92,7 @@ go_test(
         "state_summary_test.go",
         "state_test.go",
         "utils_test.go",
+        "validated_checkpoint_test.go",
     ],
     data = glob(["testdata/**"]),
     embed = [":go_default_library"],

beacon-chain/db/kv/kv.go

@@ -175,7 +175,6 @@ func NewKVStore(ctx context.Context, dirPath string, config *Config) (*Store, er
 			powchainBucket,
 			stateSummaryBucket,
 			stateValidatorsBucket,
-			validatedTips,
 			// Indices buckets.
 			attestationHeadBlockRootBucket,
 			attestationSourceRootIndicesBucket,

beacon-chain/db/kv/schema.go

@@ -19,7 +19,6 @@ var (
 	powchainBucket          = []byte("powchain")
 	stateValidatorsBucket   = []byte("state-validators")
 	feeRecipientBucket      = []byte("fee-recipient")
-	validatedTips           = []byte("validated-synced-tips")
 
 	// Deprecated: This bucket was migrated in PR 6461. Do not use, except for migrations.
 	slotsHasObjectBucket = []byte("slots-has-objects")
@@ -39,12 +38,13 @@ var (
 	blockRootValidatorHashesBucket      = []byte("block-root-validator-hashes")
 
 	// Specific item keys.
-	headBlockRootKey          = []byte("head-root")
-	genesisBlockRootKey       = []byte("genesis-root")
-	depositContractAddressKey = []byte("deposit-contract")
-	justifiedCheckpointKey    = []byte("justified-checkpoint")
-	finalizedCheckpointKey    = []byte("finalized-checkpoint")
-	powchainDataKey           = []byte("powchain-data")
+	headBlockRootKey           = []byte("head-root")
+	genesisBlockRootKey        = []byte("genesis-root")
+	depositContractAddressKey  = []byte("deposit-contract")
+	justifiedCheckpointKey     = []byte("justified-checkpoint")
+	finalizedCheckpointKey     = []byte("finalized-checkpoint")
+	powchainDataKey            = []byte("powchain-data")
+	lastValidatedCheckpointKey = []byte("last-validated-checkpoint")
 
 	// Below keys are used to identify objects are to be fork compatible.
 	// Objects that are only compatible with specific forks should be prefixed with such keys.

beacon-chain/db/kv/validated_checkpoint.go

@@ -0,0 +1,49 @@
+package kv
+
+import (
+	"context"
+
+	"github.com/prysmaticlabs/prysm/encoding/bytesutil"
+	ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
+	bolt "go.etcd.io/bbolt"
+	"go.opencensus.io/trace"
+)
+
+// LastValidatedCheckpoint returns the latest fully validated checkpoint in beacon chain.
+func (s *Store) LastValidatedCheckpoint(ctx context.Context) (*ethpb.Checkpoint, error) {
+	ctx, span := trace.StartSpan(ctx, "BeaconDB.LastValidatedCheckpoint")
+	defer span.End()
+	var checkpoint *ethpb.Checkpoint
+	err := s.db.View(func(tx *bolt.Tx) error {
+		bkt := tx.Bucket(checkpointBucket)
+		enc := bkt.Get(lastValidatedCheckpointKey)
+		if enc == nil {
+			var finErr error
+			checkpoint, finErr = s.FinalizedCheckpoint(ctx)
+			return finErr
+		}
+		checkpoint = &ethpb.Checkpoint{}
+		return decode(ctx, enc, checkpoint)
+	})
+	return checkpoint, err
+}
+
+// SaveLastValidatedCheckpoint saves the last validated checkpoint in beacon chain.
+func (s *Store) SaveLastValidatedCheckpoint(ctx context.Context, checkpoint *ethpb.Checkpoint) error {
+	ctx, span := trace.StartSpan(ctx, "BeaconDB.SaveLastValidatedCheckpoint")
+	defer span.End()
+
+	enc, err := encode(ctx, checkpoint)
+	if err != nil {
+		return err
+	}
+	return s.db.Update(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket(checkpointBucket)
+		hasStateSummary := s.hasStateSummaryBytes(tx, bytesutil.ToBytes32(checkpoint.Root))
+		hasStateInDB := tx.Bucket(stateBucket).Get(checkpoint.Root) != nil
+		if !(hasStateInDB || hasStateSummary) {
+			return errMissingStateForCheckpoint
+		}
+		return bucket.Put(lastValidatedCheckpointKey, enc)
+	})
+}

beacon-chain/db/kv/validated_checkpoint_test.go

@@ -0,0 +1,67 @@
+package kv
+
+import (
+	"context"
+	"testing"
+
+	"github.com/prysmaticlabs/prysm/encoding/bytesutil"
+	ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1/wrapper"
+	"github.com/prysmaticlabs/prysm/testing/assert"
+	"github.com/prysmaticlabs/prysm/testing/require"
+	"github.com/prysmaticlabs/prysm/testing/util"
+	"google.golang.org/protobuf/proto"
+)
+
+func TestStore_LastValidatedCheckpoint_CanSaveRetrieve(t *testing.T) {
+	db := setupDB(t)
+	ctx := context.Background()
+	root := bytesutil.ToBytes32([]byte{'A'})
+	cp := &ethpb.Checkpoint{
+		Epoch: 10,
+		Root:  root[:],
+	}
+	st, err := util.NewBeaconState()
+	require.NoError(t, err)
+	require.NoError(t, st.SetSlot(1))
+	require.NoError(t, db.SaveState(ctx, st, root))
+	require.NoError(t, db.SaveLastValidatedCheckpoint(ctx, cp))
+
+	retrieved, err := db.LastValidatedCheckpoint(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, true, proto.Equal(cp, retrieved), "Wanted %v, received %v", cp, retrieved)
+}
+
+func TestStore_LastValidatedCheckpoint_DefaultIsFinalized(t *testing.T) {
+	db := setupDB(t)
+	ctx := context.Background()
+
+	genesis := bytesutil.ToBytes32([]byte{'G', 'E', 'N', 'E', 'S', 'I', 'S'})
+	require.NoError(t, db.SaveGenesisBlockRoot(ctx, genesis))
+
+	blk := util.NewBeaconBlock()
+	blk.Block.ParentRoot = genesis[:]
+	blk.Block.Slot = 40
+
+	root, err := blk.Block.HashTreeRoot()
+	require.NoError(t, err)
+
+	cp := &ethpb.Checkpoint{
+		Epoch: 5,
+		Root:  root[:],
+	}
+
+	// a valid chain is required to save finalized checkpoint.
+	require.NoError(t, db.SaveBlock(ctx, wrapper.WrappedPhase0SignedBeaconBlock(blk)))
+	st, err := util.NewBeaconState()
+	require.NoError(t, err)
+	require.NoError(t, st.SetSlot(1))
+	// a state is required to save checkpoint
+	require.NoError(t, db.SaveState(ctx, st, root))
+
+	require.NoError(t, db.SaveFinalizedCheckpoint(ctx, cp))
+
+	retrieved, err := db.LastValidatedCheckpoint(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, true, proto.Equal(cp, retrieved), "Wanted %v, received %v", cp, retrieved)
+}

beacon-chain/forkchoice/doubly-linked-tree/errors.go

@@ -2,7 +2,7 @@ package doublylinkedtree
 
 import "errors"
 
-var errNilNode = errors.New("invalid nil or unknown node")
+var ErrNilNode = errors.New("invalid nil or unknown node")
 var errInvalidBalance = errors.New("invalid node balance")
 var errInvalidProposerBoostRoot = errors.New("invalid proposer boost root")
 var errUnknownFinalizedRoot = errors.New("unknown finalized root")

beacon-chain/forkchoice/doubly-linked-tree/forkchoice.go

@@ -174,7 +174,7 @@ func (f *ForkChoice) IsOptimistic(_ context.Context, root [32]byte) (bool, error
 
 	node, ok := f.store.nodeByRoot[root]
 	if !ok || node == nil {
-		return false, errNilNode
+		return false, ErrNilNode
 	}
 
 	return node.optimistic, nil
@@ -190,7 +190,7 @@ func (f *ForkChoice) AncestorRoot(ctx context.Context, root [32]byte, slot types
 
 	node, ok := f.store.nodeByRoot[root]
 	if !ok || node == nil {
-		return nil, errNilNode
+		return nil, ErrNilNode
 	}
 
 	n := node
@@ -202,7 +202,7 @@ func (f *ForkChoice) AncestorRoot(ctx context.Context, root [32]byte, slot types
 	}
 
 	if n == nil {
-		return nil, errNilNode
+		return nil, ErrNilNode
 	}
 
 	return n.root[:], nil
@@ -237,7 +237,7 @@ func (f *ForkChoice) updateBalances(newBalances []uint64) error {
 			if ok && vote.nextRoot != params.BeaconConfig().ZeroHash {
 				// Protection against nil node
 				if nextNode == nil {
-					return errNilNode
+					return ErrNilNode
 				}
 				nextNode.balance += newBalance
 			}
@@ -246,7 +246,7 @@ func (f *ForkChoice) updateBalances(newBalances []uint64) error {
 			if ok && vote.currentRoot != params.BeaconConfig().ZeroHash {
 				// Protection against nil node
 				if currentNode == nil {
-					return errNilNode
+					return ErrNilNode
 				}
 				if currentNode.balance < oldBalance {
 					return errInvalidBalance
@@ -279,7 +279,7 @@ func (f *ForkChoice) SetOptimisticToValid(ctx context.Context, root [fieldparams
 	defer f.store.nodesLock.Unlock()
 	node, ok := f.store.nodeByRoot[root]
 	if !ok || node == nil {
-		return errNilNode
+		return ErrNilNode
 	}
 	return node.setNodeAndParentValidated(ctx)
 }