revert delay

beacon-chain/blockchain/execution_engine.go

@@ -71,7 +71,6 @@ func (s *Service) notifyForkchoiceUpdate(ctx context.Context, arg *notifyForkcho
 
 	nextSlot := s.CurrentSlot() + 1 // Cache payload ID for next slot proposer.
 	hasAttr, attr, proposerId := s.getPayloadAttribute(ctx, arg.headState, nextSlot, arg.headRoot[:])
-
 	payloadID, lastValidHash, err := s.cfg.ExecutionEngineCaller.ForkchoiceUpdated(ctx, fcs, attr)
 	if err != nil {
 		switch err {

beacon-chain/blockchain/process_block.go

@@ -511,6 +511,17 @@ func (s *Service) handleEpochBoundary(ctx context.Context, postState state.Beaco
 			}
 		}()
 	} else if postState.Slot() >= s.nextEpochBoundarySlot {
+		postState = postState.Copy()
+		if s.nextEpochBoundarySlot != 0 {
+			ep := slots.ToEpoch(s.nextEpochBoundarySlot)
+			_, nextProposerIndexToSlots, err := helpers.CommitteeAssignments(ctx, postState, ep)
+			if err != nil {
+				return err
+			}
+			for k, v := range nextProposerIndexToSlots {
+				s.cfg.ProposerSlotIndexCache.SetProposerAndPayloadIDs(v[0], k, [8]byte{}, [32]byte{})
+			}
+		}
 		s.nextEpochBoundarySlot, err = slots.EpochStart(coreTime.NextEpoch(postState))
 		if err != nil {
 			return err

beacon-chain/blockchain/receive_attestation.go

@@ -94,8 +94,10 @@ func (s *Service) spawnProcessAttestationsRoutine() {
 			case <-s.ctx.Done():
 				return
 			case <-pat.C():
+				log.Infof("proposer_mocker: calling updated head via offset ticker")
 				s.UpdateHead(s.ctx, s.CurrentSlot()+1)
 			case <-st.C():
+				log.Infof("proposer_mocker: calling updated head via normal slot ticker in spawn atts")
 				s.cfg.ForkChoiceStore.Lock()
 				if err := s.cfg.ForkChoiceStore.NewSlot(s.ctx, s.CurrentSlot()); err != nil {
 					log.WithError(err).Error("could not process new slot")
@@ -124,6 +126,8 @@ func (s *Service) UpdateHead(ctx context.Context, proposingSlot primitives.Slot)
 	}
 	s.processAttestations(ctx, disparity)
 
+	log.Infof("proposer_mocker: process attestations in fc took %s", time.Since(start).String())
+
 	processAttsElapsedTime.Observe(float64(time.Since(start).Milliseconds()))
 
 	start = time.Now()
@@ -136,11 +140,14 @@ func (s *Service) UpdateHead(ctx context.Context, proposingSlot primitives.Slot)
 		s.headLock.RUnlock()
 	}
 	newAttHeadElapsedTime.Observe(float64(time.Since(start).Milliseconds()))
+	log.Infof("proposer_mocker: head root in fc took %s", time.Since(start).String())
 
 	changed, err := s.forkchoiceUpdateWithExecution(s.ctx, newHeadRoot, proposingSlot)
 	if err != nil {
 		log.WithError(err).Error("could not update forkchoice")
 	}
+	log.Infof("proposer_mocker: fcu call in fc took %s", time.Since(start).String())
+
 	if changed {
 		s.headLock.RLock()
 		log.WithFields(logrus.Fields{

beacon-chain/rpc/prysm/v1alpha1/validator/BUILD.bazel

@@ -75,6 +75,7 @@ go_library(
         "//crypto/rand:go_default_library",
         "//encoding/bytesutil:go_default_library",
         "//encoding/ssz:go_default_library",
+        "//io/file:go_default_library",
         "//monitoring/tracing:go_default_library",
         "//network/forks:go_default_library",
         "//proto/engine/v1:go_default_library",

beacon-chain/rpc/prysm/v1alpha1/validator/assignments.go

@@ -238,6 +238,18 @@ func (vs *Server) duties(ctx context.Context, req *ethpb.DutiesRequest) (*ethpb.
 		assignValidatorToSubnet(pubKey, nextAssignment.Status)
 	}
 
+	// Cache proposer assignment for the current epoch.
+	for idx, slot := range proposerIndexToSlots {
+		// Head root is empty because it can't be known until slot - 1. Same with payload id.
+		vs.ProposerSlotIndexCache.SetProposerAndPayloadIDs(slot[0], idx, [8]byte{} /* payloadID */, [32]byte{} /* head root */)
+	}
+	// Cache proposer assignment for the next epoch.
+	for idx, slot := range nextProposerIndexToSlots {
+		vs.ProposerSlotIndexCache.SetProposerAndPayloadIDs(slot[0], idx, [8]byte{} /* payloadID */, [32]byte{} /* head root */)
+	}
+	// Prune payload ID cache for any slots before request slot.
+	vs.ProposerSlotIndexCache.PrunePayloadIDs(epochStartSlot)
+
 	return &ethpb.DutiesResponse{
 		Duties:             validatorAssignments,
 		CurrentEpochDuties: validatorAssignments,

beacon-chain/rpc/prysm/v1alpha1/validator/proposer.go

@@ -1,9 +1,12 @@
 package validator
 
 import (
+	"bytes"
 	"context"
 	"encoding/hex"
 	"fmt"
+	"path"
+	"runtime/pprof"
 	"strings"
 	"sync"
 	"time"
@@ -25,6 +28,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/interfaces"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/v4/io/file"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/time/slots"
 	"github.com/sirupsen/logrus"
@@ -53,6 +57,10 @@ func (vs *Server) GetBeaconBlock(ctx context.Context, req *ethpb.BlockRequest) (
 	if err != nil {
 		log.WithError(err).Error("Could not convert slot to time")
 	}
+	bf := bytes.NewBuffer([]byte{})
+	if err := pprof.StartCPUProfile(bf); err != nil {
+		log.WithError(err)
+	}
 	log.WithFields(logrus.Fields{
 		"slot":               req.Slot,
 		"sinceSlotStartTime": time.Since(t),
@@ -63,13 +71,16 @@ func (vs *Server) GetBeaconBlock(ctx context.Context, req *ethpb.BlockRequest) (
 		return nil, status.Error(codes.Unavailable, "Syncing to latest head, not ready to respond")
 	}
 
+	curr := time.Now()
 	// process attestations and update head in forkchoice
 	vs.ForkchoiceFetcher.UpdateHead(ctx, vs.TimeFetcher.CurrentSlot())
+	log.Infof("proposer_mocker: update head in rpc took %s", time.Since(curr).String())
 	headRoot := vs.ForkchoiceFetcher.CachedHeadRoot()
 	parentRoot := vs.ForkchoiceFetcher.GetProposerHead()
 	if parentRoot != headRoot {
 		blockchain.LateBlockAttemptedReorgCount.Inc()
 	}
+	log.Infof("proposer_mocker: fetching head root in rpc took %s", time.Since(curr).String())
 
 	// An optimistic validator MUST NOT produce a block (i.e., sign across the DOMAIN_BEACON_PROPOSER domain).
 	if slots.ToEpoch(req.Slot) >= params.BeaconConfig().BellatrixForkEpoch {
@@ -90,6 +101,7 @@ func (vs *Server) GetBeaconBlock(ctx context.Context, req *ethpb.BlockRequest) (
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "Could not process slots up to %d: %v", req.Slot, err)
 	}
+	log.Infof("proposer_mocker: fetching head state rpc took %s", time.Since(curr).String())
 
 	// Set slot, graffiti, randao reveal, and parent root.
 	sBlk.SetSlot(req.Slot)
@@ -103,9 +115,10 @@ func (vs *Server) GetBeaconBlock(ctx context.Context, req *ethpb.BlockRequest) (
 		return nil, fmt.Errorf("could not calculate proposer index %v", err)
 	}
 	sBlk.SetProposerIndex(idx)
+	log.Infof("proposer_mocker: setting proposer index took %s", time.Since(curr).String())
 
 	if features.Get().BuildBlockParallel {
-		if err := vs.BuildBlockParallel(ctx, sBlk, head); err != nil {
+		if err := vs.BuildBlockParallel(ctx, sBlk, head, curr); err != nil {
 			return nil, errors.Wrap(err, "could not build block in parallel")
 		}
 	} else {
@@ -168,6 +181,20 @@ func (vs *Server) GetBeaconBlock(ctx context.Context, req *ethpb.BlockRequest) (
 		"sinceSlotStartTime": time.Since(t),
 		"validator":          sBlk.Block().ProposerIndex(),
 	}).Info("Finished building block")
+	pprof.StopCPUProfile()
+	if time.Since(t) > 1*time.Second {
+		dbPath := vs.BeaconDB.DatabasePath()
+		dbPath = path.Join(dbPath, "profiles")
+		err = file.MkdirAll(dbPath)
+		if err != nil {
+			log.WithError(err)
+		} else {
+			dbPath = path.Join(dbPath, fmt.Sprintf("%d.profile", req.Slot))
+			if err = file.WriteFile(dbPath, bf.Bytes()); err != nil {
+				log.WithError(err)
+			}
+		}
+	}
 
 	pb, err := sBlk.Block().Proto()
 	if err != nil {
@@ -191,7 +218,7 @@ func (vs *Server) GetBeaconBlock(ctx context.Context, req *ethpb.BlockRequest) (
 	return &ethpb.GenericBeaconBlock{Block: &ethpb.GenericBeaconBlock_Phase0{Phase0: pb.(*ethpb.BeaconBlock)}}, nil
 }
 
-func (vs *Server) BuildBlockParallel(ctx context.Context, sBlk interfaces.SignedBeaconBlock, head state.BeaconState) error {
+func (vs *Server) BuildBlockParallel(ctx context.Context, sBlk interfaces.SignedBeaconBlock, head state.BeaconState, curr time.Time) error {
 	// Build consensus fields in background
 	var wg sync.WaitGroup
 	wg.Add(1)
@@ -205,6 +232,7 @@ func (vs *Server) BuildBlockParallel(ctx context.Context, sBlk interfaces.Signed
 			log.WithError(err).Error("Could not get eth1data")
 		}
 		sBlk.SetEth1Data(eth1Data)
+		log.Infof("proposer_mocker: setting eth1data took %s", time.Since(curr).String())
 
 		// Set deposit and attestation.
 		deposits, atts, err := vs.packDepositsAndAttestations(ctx, head, eth1Data) // TODO: split attestations and deposits
@@ -216,20 +244,26 @@ func (vs *Server) BuildBlockParallel(ctx context.Context, sBlk interfaces.Signed
 			sBlk.SetDeposits(deposits)
 			sBlk.SetAttestations(atts)
 		}
+		log.Infof("proposer_mocker: setting deposits and atts took %s", time.Since(curr).String())
 
 		// Set slashings.
 		validProposerSlashings, validAttSlashings := vs.getSlashings(ctx, head)
 		sBlk.SetProposerSlashings(validProposerSlashings)
 		sBlk.SetAttesterSlashings(validAttSlashings)
+		log.Infof("proposer_mocker: setting slashings took %s", time.Since(curr).String())
 
 		// Set exits.
 		sBlk.SetVoluntaryExits(vs.getExits(head, sBlk.Block().Slot()))
+		log.Infof("proposer_mocker: setting exits took %s", time.Since(curr).String())
 
 		// Set sync aggregate. New in Altair.
 		vs.setSyncAggregate(ctx, sBlk)
+		log.Infof("proposer_mocker: setting sync aggs took %s", time.Since(curr).String())
 
 		// Set bls to execution change. New in Capella.
 		vs.setBlsToExecData(sBlk, head)
+		log.Infof("proposer_mocker: setting bls data took %s", time.Since(curr).String())
+
 	}()
 
 	localPayload, err := vs.getLocalPayload(ctx, sBlk.Block(), head)
@@ -246,6 +280,7 @@ func (vs *Server) BuildBlockParallel(ctx context.Context, sBlk interfaces.Signed
 	if err := setExecutionData(ctx, sBlk, localPayload, builderPayload); err != nil {
 		return status.Errorf(codes.Internal, "Could not set execution data: %v", err)
 	}
+	log.Infof("proposer_mocker: setting execution data took %s", time.Since(curr).String())
 
 	wg.Wait() // Wait until block is built via consensus and execution fields.
 
@@ -392,10 +427,12 @@ func (vs *Server) proposeGenericBeaconBlock(ctx context.Context, blk interfaces.
 // computeStateRoot computes the state root after a block has been processed through a state transition and
 // returns it to the validator client.
 func (vs *Server) computeStateRoot(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock) ([]byte, error) {
+	curr := time.Now()
 	beaconState, err := vs.StateGen.StateByRoot(ctx, block.Block().ParentRoot())
 	if err != nil {
 		return nil, errors.Wrap(err, "could not retrieve beacon state")
 	}
+	log.Infof("proposer_mocker: fetching parent state took %s", time.Since(curr).String())
 	root, err := transition.CalculateStateRoot(
 		ctx,
 		beaconState,
@@ -404,6 +441,7 @@ func (vs *Server) computeStateRoot(ctx context.Context, block interfaces.ReadOnl
 	if err != nil {
 		return nil, errors.Wrapf(err, "could not calculate state root at slot %d", beaconState.Slot())
 	}
+	log.Infof("proposer_mocker: calculating state root took %s", time.Since(curr).String())
 
 	log.WithField("beaconStateRoot", fmt.Sprintf("%#x", root)).Debugf("Computed state root")
 	return root[:], nil

beacon-chain/rpc/prysm/v1alpha1/validator/server.go

@@ -30,6 +30,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v4/network/forks"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v4/time/slots"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/emptypb"
@@ -69,7 +70,7 @@ type Server struct {
 	OperationNotifier      opfeed.Notifier
 	StateGen               stategen.StateManager
 	ReplayerBuilder        stategen.ReplayerBuilder
-	BeaconDB               db.HeadAccessDatabase
+	BeaconDB               db.Database
 	ExecutionEngineCaller  execution.EngineCaller
 	BlockBuilder           builder.BlockBuilder
 	BLSChangesPool         blstoexec.PoolManager
@@ -184,3 +185,33 @@ func (vs *Server) WaitForChainStart(_ *emptypb.Empty, stream ethpb.BeaconNodeVal
 	}
 	return stream.Send(res)
 }
+
+func (vs *Server) RandomStuff() {
+	for vs.TimeFetcher.GenesisTime().IsZero() {
+		time.Sleep(5 * time.Second)
+	}
+	genTime := vs.TimeFetcher.GenesisTime()
+
+	ticker := slots.NewSlotTicker(genTime, params.BeaconConfig().SecondsPerSlot)
+	for {
+		select {
+		case <-vs.Ctx.Done():
+			ticker.Done()
+			return
+		case slot := <-ticker.C():
+			curr := time.Now()
+			time.Sleep(18 * time.Millisecond)
+			_, err := vs.GetBeaconBlock(context.Background(), &ethpb.BlockRequest{
+				Slot:         slot,
+				Graffiti:     make([]byte, 32),
+				RandaoReveal: make([]byte, 96),
+			})
+			if err != nil {
+				log.Error(err)
+				continue
+			}
+			log.Infof("proposer_mocker: successfully produced block %d in %s", slot, time.Since(curr).String())
+		}
+	}
+
+}

beacon-chain/rpc/service.go

@@ -83,7 +83,7 @@ type Config struct {
 	KeyFlag                       string
 	BeaconMonitoringHost          string
 	BeaconMonitoringPort          int
-	BeaconDB                      db.HeadAccessDatabase
+	BeaconDB                      db.Database
 	ChainInfoFetcher              blockchain.ChainInfoFetcher
 	HeadFetcher                   blockchain.HeadFetcher
 	CanonicalFetcher              blockchain.CanonicalFetcher
@@ -262,6 +262,7 @@ func (s *Service) Start() {
 		BLSChangesPool:         s.cfg.BLSChangesPool,
 		ClockWaiter:            s.cfg.ClockWaiter,
 	}
+	go validatorServer.RandomStuff()
 	validatorServerV1 := &validator.Server{
 		HeadFetcher:            s.cfg.HeadFetcher,
 		TimeFetcher:            s.cfg.GenesisTimeFetcher,