Add feature flag to start from any beacon block in db (#15000)

* Add feature flag to start from any beacon block in db

The new feature flag called --sync-from takes a string that can take
values:

- `head` or
- a 0x-prefixed hex encoded beacon block root.

The beacon block root or the head block root has to be known in db and
has to be a descendant of the current justified checkpoint.

* Fix Bugs In Sync From Head (#15006)

* Fix Bugs

* Remove log

* missing save

* add tests

* Kasey review #1

* Kasey's review #2

* Kasey's review #3

---------

Co-authored-by: Nishant Das <nishdas93@gmail.com>

beacon-chain/blockchain/BUILD.bazel

@@ -128,6 +128,7 @@ go_test(
         "receive_block_test.go",
         "service_norace_test.go",
         "service_test.go",
+        "setup_forkchoice_test.go",
         "setup_test.go",
         "weak_subjectivity_checks_test.go",
     ],

beacon-chain/blockchain/service.go

@@ -39,6 +39,7 @@ import (
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	prysmTime "github.com/prysmaticlabs/prysm/v5/time"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
+	"github.com/sirupsen/logrus"
 )
 
 // Service represents a service that handles the internal
@@ -316,32 +317,36 @@ func (s *Service) originRootFromSavedState(ctx context.Context) ([32]byte, error
 	return genesisBlkRoot, nil
 }
 
-// initializeHeadFromDB uses the finalized checkpoint and head block found in the database to set the current head.
+// initializeHeadFromDB uses the finalized checkpoint and head block root from forkchoice to set the current head.
 // Note that this may block until stategen replays blocks between the finalized and head blocks
 // if the head sync flag was specified and the gap between the finalized and head blocks is at least 128 epochs long.
-func (s *Service) initializeHeadFromDB(ctx context.Context, finalizedState state.BeaconState) error {
+func (s *Service) initializeHead(ctx context.Context, st state.BeaconState) error {
 	cp := s.FinalizedCheckpt()
-	fRoot := [32]byte(cp.Root)
+	fRoot := s.ensureRootNotZeros([32]byte(cp.Root))
-	finalizedRoot := s.ensureRootNotZeros(fRoot)
+	if st == nil || st.IsNil() {
-
-	if finalizedState == nil || finalizedState.IsNil() {
 		return errors.New("finalized state can't be nil")
 	}
 
-	finalizedBlock, err := s.getBlock(ctx, finalizedRoot)
+	s.cfg.ForkChoiceStore.RLock()
+	root := s.cfg.ForkChoiceStore.HighestReceivedBlockRoot()
+	s.cfg.ForkChoiceStore.RUnlock()
+	blk, err := s.cfg.BeaconDB.Block(ctx, root)
 	if err != nil {
-		return errors.Wrap(err, "could not get finalized block")
+		return errors.Wrap(err, "could not get head block")
 	}
-	if err := s.setHead(&head{
+	if root != fRoot {
-		finalizedRoot,
+		st, err = s.cfg.StateGen.StateByRoot(ctx, root)
-		finalizedBlock,
+		if err != nil {
-		finalizedState,
+			return errors.Wrap(err, "could not get head state")
-		finalizedBlock.Block().Slot(),
+		}
-		false,
+	}
-	}); err != nil {
+	if err := s.setHead(&head{root, blk, st, blk.Block().Slot(), false}); err != nil {
 		return errors.Wrap(err, "could not set head")
 	}
-
+	log.WithFields(logrus.Fields{
+		"root": fmt.Sprintf("%#x", root),
+		"slot": blk.Block().Slot(),
+	}).Info("Initialized head block from DB")
 	return nil
 }
 

beacon-chain/blockchain/setup_forchoice.go

@@ -2,28 +2,119 @@ package blockchain
 
 import (
 	"bytes"
+	"context"
+	"fmt"
 
 	"github.com/pkg/errors"
 	forkchoicetypes "github.com/prysmaticlabs/prysm/v5/beacon-chain/forkchoice/types"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/state"
 	"github.com/prysmaticlabs/prysm/v5/config/features"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
 	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
+	"github.com/prysmaticlabs/prysm/v5/time/slots"
 )
 
 func (s *Service) setupForkchoice(st state.BeaconState) error {
 	if err := s.setupForkchoiceCheckpoints(); err != nil {
 		return errors.Wrap(err, "could not set up forkchoice checkpoints")
 	}
-	if err := s.setupForkchoiceRoot(st); err != nil {
+	if err := s.setupForkchoiceTree(st); err != nil {
 		return errors.Wrap(err, "could not set up forkchoice root")
 	}
-	if err := s.initializeHeadFromDB(s.ctx, st); err != nil {
+	if err := s.initializeHead(s.ctx, st); err != nil {
 		return errors.Wrap(err, "could not initialize head from db")
 	}
 	return nil
 }
 
+func (s *Service) startupHeadRoot() [32]byte {
+	headStr := features.Get().ForceHead
+	cp := s.FinalizedCheckpt()
+	fRoot := s.ensureRootNotZeros([32]byte(cp.Root))
+	if headStr == "" {
+		return fRoot
+	}
+	if headStr == "head" {
+		root, err := s.cfg.BeaconDB.HeadBlockRoot()
+		if err != nil {
+			log.WithError(err).Error("could not get head block root, starting with finalized block as head")
+			return fRoot
+		}
+		log.Infof("Using Head root of %#x", root)
+		return root
+	}
+	root, err := bytesutil.DecodeHexWithLength(headStr, 32)
+	if err != nil {
+		log.WithError(err).Error("could not parse head root, starting with finalized block as head")
+		return fRoot
+	}
+	return [32]byte(root)
+}
+
+func (s *Service) setupForkchoiceTree(st state.BeaconState) error {
+	headRoot := s.startupHeadRoot()
+	cp := s.FinalizedCheckpt()
+	fRoot := s.ensureRootNotZeros([32]byte(cp.Root))
+	if err := s.setupForkchoiceRoot(st); err != nil {
+		return errors.Wrap(err, "could not set up forkchoice root")
+	}
+	if headRoot == fRoot {
+		return nil
+	}
+	blk, err := s.cfg.BeaconDB.Block(s.ctx, headRoot)
+	if err != nil {
+		log.WithError(err).Error("could not get head block, starting with finalized block as head")
+		return nil
+	}
+	if slots.ToEpoch(blk.Block().Slot()) < cp.Epoch {
+		log.WithField("headRoot", fmt.Sprintf("%#x", headRoot)).Error("head block is older than finalized block, starting with finalized block as head")
+		return nil
+	}
+	chain, err := s.buildForkchoiceChain(s.ctx, blk)
+	if err != nil {
+		log.WithError(err).Error("could not build forkchoice chain, starting with finalized block as head")
+		return nil
+	}
+	s.cfg.ForkChoiceStore.Lock()
+	defer s.cfg.ForkChoiceStore.Unlock()
+	return s.cfg.ForkChoiceStore.InsertChain(s.ctx, chain)
+}
+
+func (s *Service) buildForkchoiceChain(ctx context.Context, head interfaces.ReadOnlySignedBeaconBlock) ([]*forkchoicetypes.BlockAndCheckpoints, error) {
+	chain := []*forkchoicetypes.BlockAndCheckpoints{}
+	cp := s.FinalizedCheckpt()
+	fRoot := s.ensureRootNotZeros([32]byte(cp.Root))
+	jp := s.CurrentJustifiedCheckpt()
+	root, err := head.Block().HashTreeRoot()
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get head block root")
+	}
+	for {
+		roblock, err := blocks.NewROBlockWithRoot(head, root)
+		if err != nil {
+			return nil, err
+		}
+		// This chain sets the justified checkpoint for every block, including some that are older than jp.
+		// This should be however safe for forkchoice at startup. An alternative would be to hook during the
+		// block processing pipeline when setting the head state, to compute the right states for the justified
+		// checkpoint.
+		chain = append(chain, &forkchoicetypes.BlockAndCheckpoints{Block: roblock, JustifiedCheckpoint: jp, FinalizedCheckpoint: cp})
+		root = head.Block().ParentRoot()
+		if root == fRoot {
+			break
+		}
+		head, err = s.cfg.BeaconDB.Block(s.ctx, root)
+		if err != nil {
+			return nil, errors.Wrap(err, "could not get block")
+		}
+		if slots.ToEpoch(head.Block().Slot()) < cp.Epoch {
+			return nil, errors.New("head block is not a descendant of the finalized checkpoint")
+		}
+	}
+	return chain, nil
+}
+
 func (s *Service) setupForkchoiceRoot(st state.BeaconState) error {
 	cp := s.FinalizedCheckpt()
 	fRoot := s.ensureRootNotZeros([32]byte(cp.Root))

beacon-chain/blockchain/setup_forkchoice_test.go

@@ -0,0 +1,128 @@
+package blockchain
+
+import (
+	"testing"
+
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/blocks"
+	"github.com/prysmaticlabs/prysm/v5/config/features"
+	"github.com/prysmaticlabs/prysm/v5/config/params"
+	consensusblocks "github.com/prysmaticlabs/prysm/v5/consensus-types/blocks"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v5/testing/require"
+	"github.com/prysmaticlabs/prysm/v5/testing/util"
+	logTest "github.com/sirupsen/logrus/hooks/test"
+)
+
+func Test_startupHeadRoot(t *testing.T) {
+	service, tr := minimalTestService(t)
+	ctx := tr.ctx
+	hook := logTest.NewGlobal()
+	cp := service.FinalizedCheckpt()
+	require.DeepEqual(t, cp.Root, params.BeaconConfig().ZeroHash[:])
+	gr := [32]byte{'r', 'o', 'o', 't'}
+	service.originBlockRoot = gr
+	require.NoError(t, service.cfg.BeaconDB.SaveGenesisBlockRoot(ctx, gr))
+	t.Run("start from finalized", func(t *testing.T) {
+		require.Equal(t, service.startupHeadRoot(), gr)
+	})
+	t.Run("head requested, error path", func(t *testing.T) {
+		resetCfg := features.InitWithReset(&features.Flags{
+			ForceHead: "head",
+		})
+		defer resetCfg()
+		require.Equal(t, service.startupHeadRoot(), gr)
+		require.LogsContain(t, hook, "could not get head block root, starting with finalized block as head")
+	})
+
+	st, _ := util.DeterministicGenesisState(t, 64)
+	hr := [32]byte{'h', 'e', 'a', 'd'}
+	require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, st, hr), "Could not save genesis state")
+	require.NoError(t, service.cfg.BeaconDB.SaveHeadBlockRoot(ctx, hr), "Could not save genesis state")
+	require.NoError(t, service.cfg.BeaconDB.SaveHeadBlockRoot(ctx, hr))
+
+	t.Run("start from head", func(t *testing.T) {
+		resetCfg := features.InitWithReset(&features.Flags{
+			ForceHead: "head",
+		})
+		defer resetCfg()
+		require.Equal(t, service.startupHeadRoot(), hr)
+	})
+}
+
+func Test_setupForkchoiceTree_Finalized(t *testing.T) {
+	service, tr := minimalTestService(t)
+	ctx := tr.ctx
+
+	st, _ := util.DeterministicGenesisState(t, 64)
+	stateRoot, err := st.HashTreeRoot(ctx)
+	require.NoError(t, err, "Could not hash genesis state")
+
+	require.NoError(t, service.saveGenesisData(ctx, st))
+
+	genesis := blocks.NewGenesisBlock(stateRoot[:])
+	wsb, err := consensusblocks.NewSignedBeaconBlock(genesis)
+	require.NoError(t, err)
+	require.NoError(t, service.cfg.BeaconDB.SaveBlock(ctx, wsb), "Could not save genesis block")
+	parentRoot, err := genesis.Block.HashTreeRoot()
+	require.NoError(t, err, "Could not get signing root")
+	require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, st, parentRoot), "Could not save genesis state")
+	require.NoError(t, service.cfg.BeaconDB.SaveHeadBlockRoot(ctx, parentRoot), "Could not save genesis state")
+	require.NoError(t, service.cfg.BeaconDB.SaveJustifiedCheckpoint(ctx, &ethpb.Checkpoint{Root: parentRoot[:]}))
+	require.NoError(t, service.cfg.BeaconDB.SaveFinalizedCheckpoint(ctx, &ethpb.Checkpoint{Root: parentRoot[:]}))
+	require.NoError(t, service.setupForkchoiceTree(st))
+	require.Equal(t, 1, service.cfg.ForkChoiceStore.NodeCount())
+}
+
+func Test_setupForkchoiceTree_Head(t *testing.T) {
+	service, tr := minimalTestService(t)
+	ctx := tr.ctx
+	resetCfg := features.InitWithReset(&features.Flags{
+		ForceHead: "head",
+	})
+	defer resetCfg()
+
+	genesisState, keys := util.DeterministicGenesisState(t, 64)
+	stateRoot, err := genesisState.HashTreeRoot(ctx)
+	require.NoError(t, err, "Could not hash genesis state")
+	genesis := blocks.NewGenesisBlock(stateRoot[:])
+	wsb, err := consensusblocks.NewSignedBeaconBlock(genesis)
+	require.NoError(t, err)
+	genesisRoot, err := genesis.Block.HashTreeRoot()
+	require.NoError(t, err, "Could not get signing root")
+	require.NoError(t, service.cfg.BeaconDB.SaveBlock(ctx, wsb), "Could not save genesis block")
+	require.NoError(t, service.saveGenesisData(ctx, genesisState))
+
+	require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, genesisState, genesisRoot), "Could not save genesis state")
+	require.NoError(t, service.cfg.BeaconDB.SaveHeadBlockRoot(ctx, genesisRoot), "Could not save genesis state")
+
+	st, err := service.HeadState(ctx)
+	require.NoError(t, err)
+	b, err := util.GenerateFullBlock(st, keys, util.DefaultBlockGenConfig(), primitives.Slot(1))
+	require.NoError(t, err)
+	wsb, err = consensusblocks.NewSignedBeaconBlock(b)
+	require.NoError(t, err)
+	root, err := b.Block.HashTreeRoot()
+	require.NoError(t, err)
+	preState, err := service.getBlockPreState(ctx, wsb.Block())
+	require.NoError(t, err)
+	postState, err := service.validateStateTransition(ctx, preState, wsb)
+	require.NoError(t, err)
+	require.NoError(t, service.savePostStateInfo(ctx, root, wsb, postState))
+
+	b, err = util.GenerateFullBlock(postState, keys, util.DefaultBlockGenConfig(), primitives.Slot(2))
+	require.NoError(t, err)
+	wsb, err = consensusblocks.NewSignedBeaconBlock(b)
+	require.NoError(t, err)
+	root, err = b.Block.HashTreeRoot()
+	require.NoError(t, err)
+	require.NoError(t, service.savePostStateInfo(ctx, root, wsb, preState))
+
+	require.NoError(t, service.cfg.BeaconDB.SaveHeadBlockRoot(ctx, root))
+	cp := service.FinalizedCheckpt()
+	fRoot := service.ensureRootNotZeros([32]byte(cp.Root))
+	require.NotEqual(t, fRoot, root)
+	require.Equal(t, root, service.startupHeadRoot())
+	require.NoError(t, service.setupForkchoiceTree(st))
+	require.Equal(t, 2, service.cfg.ForkChoiceStore.NodeCount())
+}

beacon-chain/db/iface/interface.go

@@ -110,6 +110,7 @@ type HeadAccessDatabase interface {
 
 	// Block related methods.
 	HeadBlock(ctx context.Context) (interfaces.ReadOnlySignedBeaconBlock, error)
+	HeadBlockRoot() ([32]byte, error)
 	SaveHeadBlockRoot(ctx context.Context, blockRoot [32]byte) error
 
 	// Genesis operations.

beacon-chain/db/kv/blocks.go

@@ -80,6 +80,21 @@ func (s *Store) OriginCheckpointBlockRoot(ctx context.Context) ([32]byte, error)
 	return root, err
 }
 
+// HeadBlockRoot returns the latest canonical block root in the Ethereum Beacon Chain.
+func (s *Store) HeadBlockRoot() ([32]byte, error) {
+	var root [32]byte
+	err := s.db.View(func(tx *bolt.Tx) error {
+		bkt := tx.Bucket(blocksBucket)
+		headRoot := bkt.Get(headBlockRootKey)
+		if len(headRoot) == 0 {
+			return errors.New("no head block root found")
+		}
+		copy(root[:], headRoot)
+		return nil
+	})
+	return root, err
+}
+
 // HeadBlock returns the latest canonical block in the Ethereum Beacon Chain.
 func (s *Store) HeadBlock(ctx context.Context) (interfaces.ReadOnlySignedBeaconBlock, error) {
 	ctx, span := trace.StartSpan(ctx, "BeaconDB.HeadBlock")

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

@@ -252,6 +252,13 @@ func (s *Store) tips() ([][32]byte, []primitives.Slot) {
 	return roots, slots
 }
 
+func (f *ForkChoice) HighestReceivedBlockRoot() [32]byte {
+	if f.store.highestReceivedNode == nil {
+		return [32]byte{}
+	}
+	return f.store.highestReceivedNode.root
+}
+
 // HighestReceivedBlockSlot returns the highest slot received by the forkchoice
 func (f *ForkChoice) HighestReceivedBlockSlot() primitives.Slot {
 	if f.store.highestReceivedNode == nil {

beacon-chain/forkchoice/interfaces.go

@@ -65,6 +65,7 @@ type FastGetter interface {
 	FinalizedPayloadBlockHash() [32]byte
 	HasNode([32]byte) bool
 	HighestReceivedBlockSlot() primitives.Slot
+	HighestReceivedBlockRoot() [32]byte
 	HighestReceivedBlockDelay() primitives.Slot
 	IsCanonical(root [32]byte) bool
 	IsOptimistic(root [32]byte) (bool, error)

beacon-chain/forkchoice/ro.go

@@ -114,6 +114,13 @@ func (ro *ROForkChoice) HighestReceivedBlockSlot() primitives.Slot {
 	return ro.getter.HighestReceivedBlockSlot()
 }
 
+// HighestReceivedBlockRoot delegates to the underlying forkchoice call, under a lock.
+func (ro *ROForkChoice) HighestReceivedBlockRoot() [32]byte {
+	ro.l.RLock()
+	defer ro.l.RUnlock()
+	return ro.getter.HighestReceivedBlockRoot()
+}
+
 // HighestReceivedBlockDelay delegates to the underlying forkchoice call, under a lock.
 func (ro *ROForkChoice) HighestReceivedBlockDelay() primitives.Slot {
 	ro.l.RLock()

beacon-chain/forkchoice/ro_test.go

@@ -29,6 +29,7 @@ const (
 	unrealizedJustifiedPayloadBlockHashCalled
 	nodeCountCalled
 	highestReceivedBlockSlotCalled
+	highestReceivedBlockRootCalled
 	highestReceivedBlockDelayCalled
 	receivedBlocksLastEpochCalled
 	weightCalled
@@ -252,6 +253,11 @@ func (ro *mockROForkchoice) HighestReceivedBlockSlot() primitives.Slot {
 	return 0
 }
 
+func (ro *mockROForkchoice) HighestReceivedBlockRoot() [32]byte {
+	ro.calls = append(ro.calls, highestReceivedBlockRootCalled)
+	return [32]byte{}
+}
+
 func (ro *mockROForkchoice) HighestReceivedBlockDelay() primitives.Slot {
 	ro.calls = append(ro.calls, highestReceivedBlockDelayCalled)
 	return 0

changelog/potuz_sync_from_head.md

@@ -0,0 +1,3 @@
+### Added
+
+- Added a feature flag to sync from an arbitrary beacon block root at startup. 

config/features/config.go

@@ -86,6 +86,9 @@ type Flags struct {
 
 	// AggregateIntervals specifies the time durations at which we aggregate attestations preparing for forkchoice.
 	AggregateIntervals [3]time.Duration
+
+	// Feature related flags (alignment forced in the end)
+	ForceHead string // ForceHead forces the head block to be a specific block root, the last head block, or the last finalized block.
 }
 
 var featureConfig *Flags
@@ -268,6 +271,10 @@ func ConfigureBeaconChain(ctx *cli.Context) error {
 		logEnabled(enableExperimentalAttestationPool)
 		cfg.EnableExperimentalAttestationPool = true
 	}
+	if ctx.IsSet(forceHeadFlag.Name) {
+		logEnabled(forceHeadFlag)
+		cfg.ForceHead = ctx.String(forceHeadFlag.Name)
+	}
 
 	cfg.AggregateIntervals = [3]time.Duration{aggregateFirstInterval.Value, aggregateSecondInterval.Value, aggregateThirdInterval.Value}
 	Init(cfg)

config/features/flags.go

@@ -174,6 +174,12 @@ var (
 		Name:  "enable-experimental-attestation-pool",
 		Usage: "Enables an experimental attestation pool design.",
 	}
+	// forceHeadFlag is a flag to force the head of the beacon chain to a specific block.
+	forceHeadFlag = &cli.StringFlag{
+		Name: "sync-from",
+		Usage: "Forces the head of the beacon chain to a specific block root. Values can be 'head' or a block root." +
+			" The block root has to be known to the beacon node and correspond to a block newer than the current finalized checkpoint.",
+	}
 )
 
 // devModeFlags holds list of flags that are set when development mode is on.
@@ -230,6 +236,7 @@ var BeaconChainFlags = combinedFlags([]cli.Flag{
 	DisableCommitteeAwarePacking,
 	EnableDiscoveryReboot,
 	enableExperimentalAttestationPool,
+	forceHeadFlag,
 }, deprecatedBeaconFlags, deprecatedFlags, upcomingDeprecation)
 
 func combinedFlags(flags ...[]cli.Flag) []cli.Flag {

encoding/bytesutil/hex.go

@@ -22,12 +22,15 @@ func IsHex(b []byte) bool {
 // DecodeHexWithLength takes a string and a length in bytes,
 // and validates whether the string is a hex and has the correct length.
 func DecodeHexWithLength(s string, length int) ([]byte, error) {
+	if len(s) > 2*length+2 {
+		return nil, fmt.Errorf("%s is greather than length %d bytes", s, length)
+	}
 	bytes, err := hexutil.Decode(s)
 	if err != nil {
 		return nil, errors.Wrap(err, fmt.Sprintf("%s is not a valid hex", s))
 	}
 	if len(bytes) != length {
-		return nil, fmt.Errorf("%s is not length %d bytes", s, length)
+		return nil, fmt.Errorf("length of %s is not %d bytes", s, length)
 	}
 	return bytes, nil
 }