Merge branch 'develop' of https://github.com/prysmaticlabs/prysm into…

… create-log-file

beacon-chain/blockchain/process_block.go

@@ -86,18 +86,6 @@ func (s *Service) postBlockProcess(ctx context.Context, signed interfaces.ReadOn
 			"headRoot":       fmt.Sprintf("%#x", headRoot),
 			"headWeight":     headWeight,
 		}).Debug("Head block is not the received block")
-	} else {
-		// Updating next slot state cache can happen in the background. It shouldn't block rest of the process.
-		go func() {
-			// Use a custom deadline here, since this method runs asynchronously.
-			// We ignore the parent method's context and instead create a new one
-			// with a custom deadline, therefore using the background context instead.
-			slotCtx, cancel := context.WithTimeout(context.Background(), slotDeadline)
-			defer cancel()
-			if err := transition.UpdateNextSlotCache(slotCtx, blockRoot[:], postState); err != nil {
-				log.WithError(err).Debug("could not update next slot state cache")
-			}
-		}()
 	}
 	newBlockHeadElapsedTime.Observe(float64(time.Since(start).Milliseconds()))
 
@@ -126,16 +114,29 @@ func (s *Service) postBlockProcess(ctx context.Context, signed interfaces.ReadOn
 	})
 
 	defer reportAttestationInclusion(b)
-	// Get the current head state (it may be different than the incoming
-	// postState) and update epoch boundary caches. We pass the postState
-	// slot instead of the headState slot below to deal with the case of an
-	// incoming non-canonical block
-	st, err := s.HeadState(ctx)
-	if err != nil {
-		return errors.Wrap(err, "could not get headState")
-	}
-	if err := s.handleEpochBoundary(ctx, postState.Slot(), st, headRoot[:]); err != nil {
-		return errors.Wrap(err, "could not handle epoch boundary")
+	if headRoot == blockRoot {
+		// Updating next slot state cache can happen in the background
+		// except in the epoch boundary in which case we lock to handle
+		// the shuffling and proposer caches updates.
+		// We handle these caches only on canonical
+		// blocks, otherwise this will be handled by lateBlockTasks
+		slot := postState.Slot()
+		if slots.IsEpochEnd(slot) {
+			if err := transition.UpdateNextSlotCache(ctx, blockRoot[:], postState); err != nil {
+				return errors.Wrap(err, "could not update next slot state cache")
+			}
+			if err := s.handleEpochBoundary(ctx, slot, postState, blockRoot[:]); err != nil {
+				return errors.Wrap(err, "could not handle epoch boundary")
+			}
+		} else {
+			go func() {
+				slotCtx, cancel := context.WithTimeout(context.Background(), slotDeadline)
+				defer cancel()
+				if err := transition.UpdateNextSlotCache(slotCtx, blockRoot[:], postState); err != nil {
+					log.WithError(err).Error("could not update next slot state cache")
+				}
+			}()
+		}
 	}
 	onBlockProcessingTime.Observe(float64(time.Since(startTime).Milliseconds()))
 	return nil
@@ -345,7 +346,7 @@ func (s *Service) updateEpochBoundaryCaches(ctx context.Context, st state.Beacon
 		// with a custom deadline, therefore using the background context instead.
 		slotCtx, cancel := context.WithTimeout(context.Background(), slotDeadline)
 		defer cancel()
-		if err := helpers.UpdateCommitteeCache(ctx, st, e+1); err != nil {
+		if err := helpers.UpdateCommitteeCache(slotCtx, st, e+1); err != nil {
 			log.WithError(err).Warn("Could not update committee cache")
 		}
 		if err := helpers.UpdateProposerIndicesInCache(slotCtx, st, e+1); err != nil {
@@ -364,7 +365,7 @@ func (s *Service) handleEpochBoundary(ctx context.Context, slot primitives.Slot,
 	if slot < headState.Slot() {
 		return nil
 	}
-	if (slot+1)%params.BeaconConfig().SlotsPerEpoch != 0 {
+	if !slots.IsEpochEnd(slot) {
 		return nil
 	}
 	copied := headState.Copy()

beacon-chain/blockchain/receive_attestation.go

@@ -92,22 +92,24 @@ func (s *Service) spawnProcessAttestationsRoutine() {
 			return
 		}
 
-		st := slots.NewSlotTicker(s.genesisTime, params.BeaconConfig().SecondsPerSlot)
-		pat := slots.NewSlotTickerWithOffset(s.genesisTime, -reorgLateBlockCountAttestations, params.BeaconConfig().SecondsPerSlot)
+		reorgInterval := time.Second*time.Duration(params.BeaconConfig().SecondsPerSlot) - reorgLateBlockCountAttestations
+		ticker := slots.NewSlotTickerWithIntervals(s.genesisTime, []time.Duration{0, reorgInterval})
 		for {
 			select {
 			case <-s.ctx.Done():
 				return
-			case <-pat.C():
-				s.UpdateHead(s.ctx, s.CurrentSlot()+1)
-			case <-st.C():
-				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")
+			case slotInterval := <-ticker.C():
+				if slotInterval.Interval > 0 {
+					s.UpdateHead(s.ctx, slotInterval.Slot+1)
+				} else {
+					s.cfg.ForkChoiceStore.Lock()
+					if err := s.cfg.ForkChoiceStore.NewSlot(s.ctx, slotInterval.Slot); err != nil {
+						log.WithError(err).Error("could not process new slot")
+					}
+					s.cfg.ForkChoiceStore.Unlock()
+
+					s.UpdateHead(s.ctx, slotInterval.Slot)
 				}
-				s.cfg.ForkChoiceStore.Unlock()
-
-				s.UpdateHead(s.ctx, s.CurrentSlot())
 			}
 		}
 	}()

beacon-chain/core/helpers/shuffle.go

@@ -184,7 +184,7 @@ func innerShuffleList(input []primitives.ValidatorIndex, seed [32]byte, shuffle
 	for {
 		buf[seedSize] = r
 		ph := hashfunc(buf[:pivotViewSize])
-		pivot := bytesutil.FromBytes8(ph[:8]) % listSize
+		pivot := binary.LittleEndian.Uint64(ph[:8]) % listSize
 		mirror := (pivot + 1) >> 1
 		binary.LittleEndian.PutUint32(buf[pivotViewSize:], uint32(pivot>>8))
 		source := hashfunc(buf)

beacon-chain/operations/attestations/prepare_forkchoice.go

@@ -34,14 +34,17 @@ func (s *Service) prepareForkChoiceAtts() {
 	ticker := slots.NewSlotTickerWithIntervals(time.Unix(int64(s.genesisTime), 0), intervals[:])
 	for {
 		select {
-		case <-ticker.C():
+		case slotInterval := <-ticker.C():
 			t := time.Now()
 			if err := s.batchForkChoiceAtts(s.ctx); err != nil {
 				log.WithError(err).Error("Could not prepare attestations for fork choice")
 			}
-			if slots.TimeIntoSlot(s.genesisTime) < intervals[1] {
-				batchForkChoiceAttsT1.Observe(float64(time.Since(t).Milliseconds()))
-			} else if slots.TimeIntoSlot(s.genesisTime) < intervals[2] {
+			switch slotInterval.Interval {
+			case 0:
+				duration := time.Since(t)
+				log.WithField("Duration", duration).Debug("aggregated unaggregated attestations")
+				batchForkChoiceAttsT1.Observe(float64(duration.Milliseconds()))
+			case 1:
 				batchForkChoiceAttsT2.Observe(float64(time.Since(t).Milliseconds()))
 			}
 		case <-s.ctx.Done():

beacon-chain/rpc/apimiddleware/BUILD.bazel

@@ -23,7 +23,7 @@ go_library(
         "//runtime/version:go_default_library",
         "//time/slots:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
-        "@com_github_r3labs_sse//:go_default_library",
+        "@com_github_r3labs_sse_v2//:go_default_library",
     ],
 )
 
@@ -46,6 +46,6 @@ go_test(
         "//testing/require:go_default_library",
         "//time/slots:go_default_library",
         "@com_github_gogo_protobuf//types:go_default_library",
-        "@com_github_r3labs_sse//:go_default_library",
+        "@com_github_r3labs_sse_v2//:go_default_library",
     ],
 )

beacon-chain/rpc/apimiddleware/custom_handlers.go

@@ -17,7 +17,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/api/grpc"
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/rpc/eth/events"
 	"github.com/prysmaticlabs/prysm/v4/runtime/version"
-	"github.com/r3labs/sse"
+	"github.com/r3labs/sse/v2"
 )
 
 // match a number with optional decimals

beacon-chain/rpc/apimiddleware/custom_handlers_test.go

@@ -17,7 +17,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v4/beacon-chain/rpc/eth/events"
 	"github.com/prysmaticlabs/prysm/v4/testing/assert"
 	"github.com/prysmaticlabs/prysm/v4/testing/require"
-	"github.com/r3labs/sse"
+	"github.com/r3labs/sse/v2"
 )
 
 type testSSZResponseJson struct {

beacon-chain/rpc/apimiddleware/custom_hooks.go

@@ -130,28 +130,6 @@ func wrapValidatorIndicesArray(
 	return true, nil
 }
 
-// https://ethereum.github.io/beacon-apis/#/Validator/publishAggregateAndProofs expects posting a top-level array.
-// We make it more proto-friendly by wrapping it in a struct with a 'data' field.
-func wrapSignedAggregateAndProofArray(
-	endpoint *apimiddleware.Endpoint,
-	_ http.ResponseWriter,
-	req *http.Request,
-) (apimiddleware.RunDefault, apimiddleware.ErrorJson) {
-	if _, ok := endpoint.PostRequest.(*SubmitAggregateAndProofsRequestJson); ok {
-		data := make([]*SignedAggregateAttestationAndProofJson, 0)
-		if err := json.NewDecoder(req.Body).Decode(&data); err != nil {
-			return false, apimiddleware.InternalServerErrorWithMessage(err, "could not decode body")
-		}
-		j := &SubmitAggregateAndProofsRequestJson{Data: data}
-		b, err := json.Marshal(j)
-		if err != nil {
-			return false, apimiddleware.InternalServerErrorWithMessage(err, "could not marshal wrapped body")
-		}
-		req.Body = io.NopCloser(bytes.NewReader(b))
-	}
-	return true, nil
-}
-
 // https://ethereum.github.io/beacon-apis/#/Validator/prepareBeaconCommitteeSubnet expects posting a top-level array.
 // We make it more proto-friendly by wrapping it in a struct with a 'data' field.
 func wrapBeaconCommitteeSubscriptionsArray(
@@ -218,28 +196,6 @@ func wrapSyncCommitteeSignaturesArray(
 	return true, nil
 }
 
-// https://ethereum.github.io/beacon-APIs/#/Validator/publishContributionAndProofs expects posting a top-level array.
-// We make it more proto-friendly by wrapping it in a struct with a 'data' field.
-func wrapSignedContributionAndProofsArray(
-	endpoint *apimiddleware.Endpoint,
-	_ http.ResponseWriter,
-	req *http.Request,
-) (apimiddleware.RunDefault, apimiddleware.ErrorJson) {
-	if _, ok := endpoint.PostRequest.(*SubmitContributionAndProofsRequestJson); ok {
-		data := make([]*SignedContributionAndProofJson, 0)
-		if err := json.NewDecoder(req.Body).Decode(&data); err != nil {
-			return false, apimiddleware.InternalServerErrorWithMessage(err, "could not decode body")
-		}
-		j := &SubmitContributionAndProofsRequestJson{Data: data}
-		b, err := json.Marshal(j)
-		if err != nil {
-			return false, apimiddleware.InternalServerErrorWithMessage(err, "could not marshal wrapped body")
-		}
-		req.Body = io.NopCloser(bytes.NewReader(b))
-	}
-	return true, nil
-}
-
 type phase0PublishBlockRequestJson struct {
 	Phase0Block *BeaconBlockJson `json:"phase0_block"`
 	Signature   string           `json:"signature" hex:"true"`

beacon-chain/rpc/apimiddleware/custom_hooks_test.go

@@ -140,46 +140,6 @@ func TestWrapBLSChangesArray(t *testing.T) {
 	})
 }
 
-func TestWrapSignedAggregateAndProofArray(t *testing.T) {
-	t.Run("ok", func(t *testing.T) {
-		endpoint := &apimiddleware.Endpoint{
-			PostRequest: &SubmitAggregateAndProofsRequestJson{},
-		}
-		unwrappedAggs := []*SignedAggregateAttestationAndProofJson{{Signature: "sig"}}
-		unwrappedAggsJson, err := json.Marshal(unwrappedAggs)
-		require.NoError(t, err)
-
-		var body bytes.Buffer
-		_, err = body.Write(unwrappedAggsJson)
-		require.NoError(t, err)
-		request := httptest.NewRequest("POST", "http://foo.example", &body)
-
-		runDefault, errJson := wrapSignedAggregateAndProofArray(endpoint, nil, request)
-		require.Equal(t, true, errJson == nil)
-		assert.Equal(t, apimiddleware.RunDefault(true), runDefault)
-		wrappedAggs := &SubmitAggregateAndProofsRequestJson{}
-		require.NoError(t, json.NewDecoder(request.Body).Decode(wrappedAggs))
-		require.Equal(t, 1, len(wrappedAggs.Data), "wrong number of wrapped items")
-		assert.Equal(t, "sig", wrappedAggs.Data[0].Signature)
-	})
-
-	t.Run("invalid_body", func(t *testing.T) {
-		endpoint := &apimiddleware.Endpoint{
-			PostRequest: &SubmitAggregateAndProofsRequestJson{},
-		}
-		var body bytes.Buffer
-		_, err := body.Write([]byte("invalid"))
-		require.NoError(t, err)
-		request := httptest.NewRequest("POST", "http://foo.example", &body)
-
-		runDefault, errJson := wrapSignedAggregateAndProofArray(endpoint, nil, request)
-		require.Equal(t, false, errJson == nil)
-		assert.Equal(t, apimiddleware.RunDefault(false), runDefault)
-		assert.Equal(t, true, strings.Contains(errJson.Msg(), "could not decode body"))
-		assert.Equal(t, http.StatusInternalServerError, errJson.StatusCode())
-	})
-}
-
 func TestWrapBeaconCommitteeSubscriptionsArray(t *testing.T) {
 	t.Run("ok", func(t *testing.T) {
 		endpoint := &apimiddleware.Endpoint{
@@ -333,75 +293,6 @@ func TestWrapSyncCommitteeSignaturesArray(t *testing.T) {
 	})
 }
 
-func TestWrapSignedContributionAndProofsArray(t *testing.T) {
-	t.Run("ok", func(t *testing.T) {
-		endpoint := &apimiddleware.Endpoint{
-			PostRequest: &SubmitContributionAndProofsRequestJson{},
-		}
-		unwrapped := []*SignedContributionAndProofJson{
-			{
-				Message: &ContributionAndProofJson{
-					AggregatorIndex: "1",
-					Contribution: &SyncCommitteeContributionJson{
-						Slot:              "1",
-						BeaconBlockRoot:   "root",
-						SubcommitteeIndex: "1",
-						AggregationBits:   "bits",
-						Signature:         "sig",
-					},
-					SelectionProof: "proof",
-				},
-				Signature: "sig",
-			},
-			{
-				Message:   &ContributionAndProofJson{},
-				Signature: "sig",
-			},
-		}
-		unwrappedJson, err := json.Marshal(unwrapped)
-		require.NoError(t, err)
-
-		var body bytes.Buffer
-		_, err = body.Write(unwrappedJson)
-		require.NoError(t, err)
-		request := httptest.NewRequest("POST", "http://foo.example", &body)
-
-		runDefault, errJson := wrapSignedContributionAndProofsArray(endpoint, nil, request)
-		require.Equal(t, true, errJson == nil)
-		assert.Equal(t, apimiddleware.RunDefault(true), runDefault)
-		wrapped := &SubmitContributionAndProofsRequestJson{}
-		require.NoError(t, json.NewDecoder(request.Body).Decode(wrapped))
-		require.Equal(t, 2, len(wrapped.Data), "wrong number of wrapped items")
-		assert.Equal(t, "sig", wrapped.Data[0].Signature)
-		require.NotNil(t, wrapped.Data[0].Message)
-		msg := wrapped.Data[0].Message
-		assert.Equal(t, "1", msg.AggregatorIndex)
-		assert.Equal(t, "proof", msg.SelectionProof)
-		require.NotNil(t, msg.Contribution)
-		assert.Equal(t, "1", msg.Contribution.Slot)
-		assert.Equal(t, "root", msg.Contribution.BeaconBlockRoot)
-		assert.Equal(t, "1", msg.Contribution.SubcommitteeIndex)
-		assert.Equal(t, "bits", msg.Contribution.AggregationBits)
-		assert.Equal(t, "sig", msg.Contribution.Signature)
-	})
-
-	t.Run("invalid_body", func(t *testing.T) {
-		endpoint := &apimiddleware.Endpoint{
-			PostRequest: &SubmitContributionAndProofsRequestJson{},
-		}
-		var body bytes.Buffer
-		_, err := body.Write([]byte("invalid"))
-		require.NoError(t, err)
-		request := httptest.NewRequest("POST", "http://foo.example", &body)
-
-		runDefault, errJson := wrapSignedContributionAndProofsArray(endpoint, nil, request)
-		require.Equal(t, false, errJson == nil)
-		assert.Equal(t, apimiddleware.RunDefault(false), runDefault)
-		assert.Equal(t, true, strings.Contains(errJson.Msg(), "could not decode body"))
-		assert.Equal(t, http.StatusInternalServerError, errJson.StatusCode())
-	})
-}
-
 func TestSetInitialPublishBlockPostRequest(t *testing.T) {
 	params.SetupTestConfigCleanup(t)
 	cfg := params.BeaconConfig().Copy()

beacon-chain/rpc/apimiddleware/endpoint_factory.go

@@ -68,9 +68,7 @@ func (_ *BeaconEndpointFactory) Paths() []string {
 		"/eth/v1/validator/attestation_data",
 		"/eth/v1/validator/beacon_committee_subscriptions",
 		"/eth/v1/validator/sync_committee_subscriptions",
-		"/eth/v1/validator/aggregate_and_proofs",
 		"/eth/v1/validator/sync_committee_contribution",
-		"/eth/v1/validator/contribution_and_proofs",
 		"/eth/v1/validator/prepare_beacon_proposer",
 		"/eth/v1/validator/register_validator",
 		"/eth/v1/validator/liveness/{epoch}",
@@ -274,19 +272,9 @@ func (_ *BeaconEndpointFactory) Create(path string) (*apimiddleware.Endpoint, er
 		endpoint.Hooks = apimiddleware.HookCollection{
 			OnPreDeserializeRequestBodyIntoContainer: wrapSyncCommitteeSubscriptionsArray,
 		}
-	case "/eth/v1/validator/aggregate_and_proofs":
-		endpoint.PostRequest = &SubmitAggregateAndProofsRequestJson{}
-		endpoint.Hooks = apimiddleware.HookCollection{
-			OnPreDeserializeRequestBodyIntoContainer: wrapSignedAggregateAndProofArray,
-		}
 	case "/eth/v1/validator/sync_committee_contribution":
 		endpoint.GetResponse = &ProduceSyncCommitteeContributionResponseJson{}
 		endpoint.RequestQueryParams = []apimiddleware.QueryParam{{Name: "slot"}, {Name: "subcommittee_index"}, {Name: "beacon_block_root", Hex: true}}
-	case "/eth/v1/validator/contribution_and_proofs":
-		endpoint.PostRequest = &SubmitContributionAndProofsRequestJson{}
-		endpoint.Hooks = apimiddleware.HookCollection{
-			OnPreDeserializeRequestBodyIntoContainer: wrapSignedContributionAndProofsArray,
-		}
 	case "/eth/v1/validator/prepare_beacon_proposer":
 		endpoint.PostRequest = &FeeRecipientsRequestJSON{}
 		endpoint.Hooks = apimiddleware.HookCollection{