test: delay slot0 production by 4s

* Use next slot cache for sync committee

* RWMutex

* change mutex for last cached state

* feat: change mutex

* test: add db

---------

Co-authored-by: Nishant Das <nishdas93@gmail.com>
Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

api/client/beacon/checkpoint.go

@@ -11,6 +11,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/state"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/interfaces"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v4/encoding/ssz/detect"
 	"github.com/prysmaticlabs/prysm/v4/io/file"
 	"github.com/prysmaticlabs/prysm/v4/runtime/version"
@@ -19,6 +20,8 @@ import (
 	"golang.org/x/mod/semver"
 )
 
+var errCheckpointBlockMismatch = errors.New("mismatch between checkpoint sync state and block")
+
 // OriginData represents the BeaconState and ReadOnlySignedBeaconBlock necessary to start an empty Beacon Node
 // using Checkpoint Sync.
 type OriginData struct {
@@ -75,37 +78,40 @@ func DownloadFinalizedData(ctx context.Context, client *Client) (*OriginData, er
 	if err != nil {
 		return nil, errors.Wrap(err, "error unmarshaling finalized state to correct version")
 	}
-	if s.Slot() != s.LatestBlockHeader().Slot {
-		return nil, fmt.Errorf("finalized state slot does not match latest block header slot %d != %d", s.Slot(), s.LatestBlockHeader().Slot)
-	}
 
-	sr, err := s.HashTreeRoot(ctx)
+	slot := s.LatestBlockHeader().Slot
+	bb, err := client.GetBlock(ctx, IdFromSlot(slot))
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to compute htr for finalized state at slot=%d", s.Slot())
-	}
-	header := s.LatestBlockHeader()
-	header.StateRoot = sr[:]
-	br, err := header.HashTreeRoot()
-	if err != nil {
-		return nil, errors.Wrap(err, "error while computing block root using state data")
-	}
-
-	bb, err := client.GetBlock(ctx, IdFromRoot(br))
-	if err != nil {
-		return nil, errors.Wrapf(err, "error requesting block by root = %#x", br)
+		return nil, errors.Wrapf(err, "error requesting block by slot = %d", slot)
 	}
 	b, err := vu.UnmarshalBeaconBlock(bb)
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to unmarshal block to a supported type using the detected fork schedule")
 	}
-	realBlockRoot, err := b.Block().HashTreeRoot()
+	br, err := b.Block().HashTreeRoot()
 	if err != nil {
 		return nil, errors.Wrap(err, "error computing hash_tree_root of retrieved block")
 	}
+	bodyRoot, err := b.Block().Body().HashTreeRoot()
+	if err != nil {
+		return nil, errors.Wrap(err, "error computing hash_tree_root of retrieved block body")
+	}
 
-	log.Printf("BeaconState slot=%d, Block slot=%d", s.Slot(), b.Block().Slot())
-	log.Printf("BeaconState htr=%#x, Block state_root=%#x", sr, b.Block().StateRoot())
-	log.Printf("BeaconState latest_block_header htr=%#x, block htr=%#x", br, realBlockRoot)
+	sbr := bytesutil.ToBytes32(s.LatestBlockHeader().BodyRoot)
+	if sbr != bodyRoot {
+		return nil, errors.Wrapf(errCheckpointBlockMismatch, "state body root = %#x, block body root = %#x", sbr, bodyRoot)
+	}
+	sr, err := s.HashTreeRoot(ctx)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to compute htr for finalized state at slot=%d", s.Slot())
+	}
+
+	log.
+		WithField("block_slot", b.Block().Slot()).
+		WithField("state_slot", s.Slot()).
+		WithField("state_root", sr).
+		WithField("block_root", br).
+		Info("Downloaded checkpoint sync state and block.")
 	return &OriginData{
 		st: s,
 		b:  b,

api/client/beacon/checkpoint_test.go

@@ -440,7 +440,7 @@ func TestDownloadFinalizedData(t *testing.T) {
 		case renderGetStatePath(IdFinalized):
 			res.StatusCode = http.StatusOK
 			res.Body = io.NopCloser(bytes.NewBuffer(ms))
-		case renderGetBlockPath(IdFromRoot(br)):
+		case renderGetBlockPath(IdFromSlot(b.Block().Slot())):
 			res.StatusCode = http.StatusOK
 			res.Body = io.NopCloser(bytes.NewBuffer(mb))
 		default:

beacon-chain/blockchain/head_sync_committee_info.go

@@ -157,7 +157,11 @@ func (s *Service) getSyncCommitteeHeadState(ctx context.Context, slot primitives
 		if headState == nil || headState.IsNil() {
 			return nil, errors.New("nil state")
 		}
-		headState, err = transition.ProcessSlotsIfPossible(ctx, headState, slot)
+		headRoot, err := s.HeadRoot(ctx)
+		if err != nil {
+			return nil, err
+		}
+		headState, err = transition.ProcessSlotsUsingNextSlotCache(ctx, headState, headRoot, slot)
 		if err != nil {
 			return nil, err
 		}

beacon-chain/blockchain/head_sync_committee_info_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/cache"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/signing"
+	dbTest "github.com/prysmaticlabs/prysm/v4/beacon-chain/db/testing"
 	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v4/testing/require"
@@ -15,7 +16,7 @@ import (
 
 func TestService_HeadSyncCommitteeIndices(t *testing.T) {
 	s, _ := util.DeterministicGenesisStateAltair(t, params.BeaconConfig().TargetCommitteeSize)
-	c := &Service{}
+	c := &Service{cfg: &config{BeaconDB: dbTest.SetupDB(t)}}
 	c.head = &head{state: s}
 
 	// Current period
@@ -38,7 +39,7 @@ func TestService_HeadSyncCommitteeIndices(t *testing.T) {
 
 func TestService_headCurrentSyncCommitteeIndices(t *testing.T) {
 	s, _ := util.DeterministicGenesisStateAltair(t, params.BeaconConfig().TargetCommitteeSize)
-	c := &Service{}
+	c := &Service{cfg: &config{BeaconDB: dbTest.SetupDB(t)}}
 	c.head = &head{state: s}
 
 	// Process slot up to `EpochsPerSyncCommitteePeriod` so it can `ProcessSyncCommitteeUpdates`.
@@ -66,7 +67,7 @@ func TestService_headNextSyncCommitteeIndices(t *testing.T) {
 
 func TestService_HeadSyncCommitteePubKeys(t *testing.T) {
 	s, _ := util.DeterministicGenesisStateAltair(t, params.BeaconConfig().TargetCommitteeSize)
-	c := &Service{}
+	c := &Service{cfg: &config{BeaconDB: dbTest.SetupDB(t)}}
 	c.head = &head{state: s}
 
 	// Process slot up to 2 * `EpochsPerSyncCommitteePeriod` so it can run `ProcessSyncCommitteeUpdates` twice.
@@ -81,7 +82,7 @@ func TestService_HeadSyncCommitteePubKeys(t *testing.T) {
 
 func TestService_HeadSyncCommitteeDomain(t *testing.T) {
 	s, _ := util.DeterministicGenesisStateAltair(t, params.BeaconConfig().TargetCommitteeSize)
-	c := &Service{}
+	c := &Service{cfg: &config{BeaconDB: dbTest.SetupDB(t)}}
 	c.head = &head{state: s}
 
 	wanted, err := signing.Domain(s.Fork(), slots.ToEpoch(s.Slot()), params.BeaconConfig().DomainSyncCommittee, s.GenesisValidatorsRoot())

beacon-chain/monitor/process_sync_committee.go

@@ -6,6 +6,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/state"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/interfaces"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
+	"github.com/prysmaticlabs/prysm/v4/runtime/version"
 	"github.com/sirupsen/logrus"
 )
 
@@ -29,6 +30,9 @@ func (s *Service) processSyncAggregate(state state.BeaconState, blk interfaces.R
 	if blk == nil || blk.Body() == nil {
 		return
 	}
+	if blk.Version() == version.Phase0 {
+		return
+	}
 	bits, err := blk.Body().SyncAggregate()
 	if err != nil {
 		log.WithError(err).Error("Could not get SyncAggregate")

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

@@ -25,7 +25,6 @@ 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/encoding/bytesutil"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/time/slots"
 	"github.com/sirupsen/logrus"
@@ -347,11 +346,21 @@ func (vs *Server) GetFeeRecipientByPubKey(ctx context.Context, request *ethpb.Fe
 func (vs *Server) proposeGenericBeaconBlock(ctx context.Context, blk interfaces.SignedBeaconBlock) (*ethpb.ProposeResponse, error) {
 	ctx, span := trace.StartSpan(ctx, "ProposerServer.proposeGenericBeaconBlock")
 	defer span.End()
-	root, err := blk.Block().HashTreeRoot()
-	if err != nil {
-		return nil, fmt.Errorf("could not tree hash block: %v", err)
+
+	if blk.Block().Slot()%params.BeaconConfig().SlotsPerEpoch == 0 {
+		time.Sleep(4 * time.Second)
 	}
 
+	// Do not block proposal critical path with debug logging or block feed updates.
+	defer func() {
+		log.WithField("slot", blk.Block().Slot()).Debugf(
+			"Block proposal received via RPC")
+		vs.BlockNotifier.BlockFeed().Send(&feed.Event{
+			Type: blockfeed.ReceivedBlock,
+			Data: &blockfeed.ReceivedBlockData{SignedBlock: blk},
+		})
+	}()
+
 	unblinder, err := newUnblinder(blk, vs.BlockBuilder)
 	if err != nil {
 		return nil, errors.Wrap(err, "could not create unblinder")
@@ -361,16 +370,6 @@ func (vs *Server) proposeGenericBeaconBlock(ctx context.Context, blk interfaces.
 		return nil, errors.Wrap(err, "could not unblind builder block")
 	}
 
-	// Do not block proposal critical path with debug logging or block feed updates.
-	defer func() {
-		log.WithField("blockRoot", fmt.Sprintf("%#x", bytesutil.Trunc(root[:]))).Debugf(
-			"Block proposal received via RPC")
-		vs.BlockNotifier.BlockFeed().Send(&feed.Event{
-			Type: blockfeed.ReceivedBlock,
-			Data: &blockfeed.ReceivedBlockData{SignedBlock: blk},
-		})
-	}()
-
 	// Broadcast the new block to the network.
 	blkPb, err := blk.Proto()
 	if err != nil {
@@ -379,6 +378,11 @@ func (vs *Server) proposeGenericBeaconBlock(ctx context.Context, blk interfaces.
 	if err := vs.P2P.Broadcast(ctx, blkPb); err != nil {
 		return nil, fmt.Errorf("could not broadcast block: %v", err)
 	}
+	root, err := blk.Block().HashTreeRoot()
+	if err != nil {
+		return nil, fmt.Errorf("could not tree hash block: %v", err)
+	}
+
 	log.WithFields(logrus.Fields{
 		"blockRoot": hex.EncodeToString(root[:]),
 	}).Debug("Broadcasting block")

beacon-chain/sync/validate_beacon_blocks.go

@@ -203,6 +203,7 @@ func (s *Service) validateBeaconBlockPubSub(ctx context.Context, pid peer.ID, ms
 	log.WithFields(logrus.Fields{
 		"blockSlot":          blk.Block().Slot(),
 		"sinceSlotStartTime": receivedTime.Sub(startTime),
+		"validationTime":     prysmTime.Now().Sub(receivedTime),
 		"proposerIndex":      blk.Block().ProposerIndex(),
 		"graffiti":           string(graffiti[:]),
 	}).Debug("Received block")

config/params/testnet_e2e_config.go

@@ -21,7 +21,7 @@ func E2ETestConfig() *BeaconChainConfig {
 	e2eConfig.MaxValidatorsPerWithdrawalsSweep = 128
 
 	// Time parameters.
-	e2eConfig.SecondsPerSlot = 10
+	e2eConfig.SecondsPerSlot = 12
 	e2eConfig.SlotsPerEpoch = 6
 	e2eConfig.SqrRootSlotsPerEpoch = 2
 	e2eConfig.SecondsPerETH1Block = 2

deps.bzl

@@ -2458,6 +2458,8 @@ def prysm_deps():
         ],
         build_file_proto_mode = "disable_global",
         importpath = "github.com/libp2p/go-libp2p",
+        patch_args = ["-p1"],
+        patches = ["//third_party:com_github_libp2p_go_libp2p.patch"],
         sum = "h1:KwA7pXKXpz8hG6Cr1fMA7UkgleogcwQj0sxl5qquWRg=",
         version = "v0.27.5",
     )

third_party/com_github_libp2p_go_libp2p.patch

@@ -0,0 +1,11 @@
+diff --git a/p2p/net/swarm/swarm_conn.go b/p2p/net/swarm/swarm_conn.go
+index 0e79da1b..e770381a 100644
+--- a/p2p/net/swarm/swarm_conn.go
++++ b/p2p/net/swarm/swarm_conn.go
+@@ -130,6 +130,7 @@ func (c *Conn) start() {
+
+ 				// We only get an error here when the swarm is closed or closing.
+ 				if err != nil {
++					scope.Done()
+ 					return
+ 				}