Aggregate and pack sync committee messages

beacon-chain/blockchain/testing/mock.go

@@ -37,47 +37,48 @@ var ErrNilState = errors.New("nil state")
 
 // ChainService defines the mock interface for testing
 type ChainService struct {
-	NotFinalized                bool
-	Optimistic                  bool
-	ValidAttestation            bool
-	ValidatorsRoot              [32]byte
-	PublicKey                   [fieldparams.BLSPubkeyLength]byte
-	FinalizedCheckPoint         *ethpb.Checkpoint
-	CurrentJustifiedCheckPoint  *ethpb.Checkpoint
-	PreviousJustifiedCheckPoint *ethpb.Checkpoint
-	Slot                        *primitives.Slot // Pointer because 0 is a useful value, so checking against it can be incorrect.
-	Balance                     *precompute.Balance
-	CanonicalRoots              map[[32]byte]bool
-	Fork                        *ethpb.Fork
-	ETH1Data                    *ethpb.Eth1Data
-	InitSyncBlockRoots          map[[32]byte]bool
-	DB                          db.Database
-	State                       state.BeaconState
-	HeadStateErr                error
-	Block                       interfaces.ReadOnlySignedBeaconBlock
-	VerifyBlkDescendantErr      error
-	stateNotifier               statefeed.Notifier
-	BlocksReceived              []interfaces.ReadOnlySignedBeaconBlock
-	SyncCommitteeIndices        []primitives.CommitteeIndex
-	blockNotifier               blockfeed.Notifier
-	opNotifier                  opfeed.Notifier
-	Root                        []byte
-	SyncCommitteeDomain         []byte
-	SyncSelectionProofDomain    []byte
-	SyncContributionProofDomain []byte
-	SyncCommitteePubkeys        [][]byte
-	Genesis                     time.Time
-	ForkChoiceStore             forkchoice.ForkChoicer
-	ReceiveBlockMockErr         error
-	OptimisticCheckRootReceived [32]byte
-	FinalizedRoots              map[[32]byte]bool
-	OptimisticRoots             map[[32]byte]bool
-	BlockSlot                   primitives.Slot
-	SyncingRoot                 [32]byte
-	Blobs                       []blocks.VerifiedROBlob
-	DataColumns                 []blocks.VerifiedRODataColumn
-	TargetRoot                  [32]byte
-	MockHeadSlot                *primitives.Slot
+	NotFinalized                     bool
+	Optimistic                       bool
+	ValidAttestation                 bool
+	ValidatorsRoot                   [32]byte
+	PublicKey                        [fieldparams.BLSPubkeyLength]byte
+	FinalizedCheckPoint              *ethpb.Checkpoint
+	CurrentJustifiedCheckPoint       *ethpb.Checkpoint
+	PreviousJustifiedCheckPoint      *ethpb.Checkpoint
+	Slot                             *primitives.Slot // Pointer because 0 is a useful value, so checking against it can be incorrect.
+	Balance                          *precompute.Balance
+	CanonicalRoots                   map[[32]byte]bool
+	Fork                             *ethpb.Fork
+	ETH1Data                         *ethpb.Eth1Data
+	InitSyncBlockRoots               map[[32]byte]bool
+	DB                               db.Database
+	State                            state.BeaconState
+	HeadStateErr                     error
+	Block                            interfaces.ReadOnlySignedBeaconBlock
+	VerifyBlkDescendantErr           error
+	stateNotifier                    statefeed.Notifier
+	BlocksReceived                   []interfaces.ReadOnlySignedBeaconBlock
+	SyncCommitteeIndices             []primitives.CommitteeIndex
+	SyncCommitteeIndicesPerValidator map[primitives.ValidatorIndex][]primitives.CommitteeIndex
+	blockNotifier                    blockfeed.Notifier
+	opNotifier                       opfeed.Notifier
+	Root                             []byte
+	SyncCommitteeDomain              []byte
+	SyncSelectionProofDomain         []byte
+	SyncContributionProofDomain      []byte
+	SyncCommitteePubkeys             [][]byte
+	Genesis                          time.Time
+	ForkChoiceStore                  forkchoice.ForkChoicer
+	ReceiveBlockMockErr              error
+	OptimisticCheckRootReceived      [32]byte
+	FinalizedRoots                   map[[32]byte]bool
+	OptimisticRoots                  map[[32]byte]bool
+	BlockSlot                        primitives.Slot
+	SyncingRoot                      [32]byte
+	Blobs                            []blocks.VerifiedROBlob
+	DataColumns                      []blocks.VerifiedRODataColumn
+	TargetRoot                       [32]byte
+	MockHeadSlot                     *primitives.Slot
 }
 
 func (s *ChainService) Ancestor(ctx context.Context, root []byte, slot primitives.Slot) ([]byte, error) {
@@ -510,8 +511,11 @@ func (s *ChainService) HeadValidatorIndexToPublicKey(_ context.Context, _ primit
 }
 
 // HeadSyncCommitteeIndices mocks HeadSyncCommitteeIndices and always return `HeadNextSyncCommitteeIndices`.
-func (s *ChainService) HeadSyncCommitteeIndices(_ context.Context, _ primitives.ValidatorIndex, _ primitives.Slot) ([]primitives.CommitteeIndex, error) {
-	return s.SyncCommitteeIndices, nil
+func (s *ChainService) HeadSyncCommitteeIndices(_ context.Context, index primitives.ValidatorIndex, _ primitives.Slot) ([]primitives.CommitteeIndex, error) {
+	if s.SyncCommitteeIndicesPerValidator == nil {
+		return s.SyncCommitteeIndices, nil
+	}
+	return s.SyncCommitteeIndicesPerValidator[index], nil
 }
 
 // HeadSyncCommitteePubKeys mocks HeadSyncCommitteePubKeys and always return empty nil.

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

@@ -63,8 +63,8 @@ go_library(
         "//beacon-chain/rpc/core:go_default_library",
         "//beacon-chain/startup:go_default_library",
         "//beacon-chain/state:go_default_library",
-        "//beacon-chain/state/stategen:go_default_library",
         "//beacon-chain/state/state-native:go_default_library",
+        "//beacon-chain/state/stategen:go_default_library",
         "//beacon-chain/sync:go_default_library",
         "//config/features:go_default_library",
         "//config/fieldparams:go_default_library",
@@ -81,7 +81,7 @@ go_library(
         "//crypto/rand:go_default_library",
         "//encoding/bytesutil:go_default_library",
         "//encoding/ssz:go_default_library",
-	"//genesis:go_default_library",
+        "//genesis:go_default_library",
         "//math:go_default_library",
         "//monitoring/tracing:go_default_library",
         "//monitoring/tracing/trace:go_default_library",

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

@@ -187,7 +187,7 @@ func (vs *Server) getParentState(ctx context.Context, slot primitives.Slot) (sta
 func (vs *Server) BuildBlockParallel(ctx context.Context, sBlk interfaces.SignedBeaconBlock, head state.BeaconState, skipMevBoost bool, builderBoostFactor primitives.Gwei) (*ethpb.GenericBeaconBlock, error) {
 	// Build consensus fields in background
 	var wg sync.WaitGroup
-	wg.Add(1)
+	wg.Add(2)
 	go func() {
 		defer wg.Done()
 
@@ -224,12 +224,15 @@ func (vs *Server) BuildBlockParallel(ctx context.Context, sBlk interfaces.Signed
 		// Set exits.
 		sBlk.SetVoluntaryExits(vs.getExits(head, sBlk.Block().Slot()))
 
-		// Set sync aggregate. New in Altair.
-		vs.setSyncAggregate(ctx, sBlk)
-
 		// Set bls to execution change. New in Capella.
 		vs.setBlsToExecData(sBlk, head)
 	}()
+	go func() {
+		defer wg.Done()
+
+		// Set sync aggregate. New in Altair.
+		vs.setSyncAggregate(ctx, sBlk)
+	}()
 
 	winningBid := primitives.ZeroWei()
 	var bundle enginev1.BlobsBundler

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

@@ -1,7 +1,9 @@
 package validator
 
 import (
+	"bytes"
 	"context"
+	"sync"
 
 	"github.com/OffchainLabs/prysm/v6/config/params"
 	"github.com/OffchainLabs/prysm/v6/consensus-types/interfaces"
@@ -15,6 +17,7 @@ import (
 	"github.com/OffchainLabs/prysm/v6/runtime/version"
 	"github.com/OffchainLabs/prysm/v6/time/slots"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/go-bitfield"
 )
 
 func (vs *Server) setSyncAggregate(ctx context.Context, blk interfaces.SignedBeaconBlock) {
@@ -51,21 +54,30 @@ func (vs *Server) getSyncAggregate(ctx context.Context, slot primitives.Slot, ro
 	if vs.SyncCommitteePool == nil {
 		return nil, errors.New("sync committee pool is nil")
 	}
-	// Contributions have to match the input root
-	contributions, err := vs.SyncCommitteePool.SyncCommitteeContributions(slot)
+
+	subcommitteeCount := params.BeaconConfig().SyncCommitteeSubnetCount
+
+	poolContributions, err := vs.SyncCommitteePool.SyncCommitteeContributions(slot)
 	if err != nil {
 		return nil, err
 	}
-	proposerContributions := proposerSyncContributions(contributions).filterByBlockRoot(root)
+	// Contributions have to match the input root
+	proposerContributions := proposerSyncContributions(poolContributions).filterByBlockRoot(root)
+
+	aggregatedContributions, err := vs.aggregatedSyncCommitteeMessages(ctx, slot, root)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get aggregated sync committee messages")
+	}
+	proposerContributions = append(proposerContributions, aggregatedContributions...)
 
 	// Each sync subcommittee is 128 bits and the sync committee is 512 bits for mainnet.
 	var bitsHolder [][]byte
-	for i := uint64(0); i < params.BeaconConfig().SyncCommitteeSubnetCount; i++ {
+	for i := uint64(0); i < subcommitteeCount; i++ {
 		bitsHolder = append(bitsHolder, ethpb.NewSyncCommitteeAggregationBits())
 	}
-	sigsHolder := make([]bls.Signature, 0, params.BeaconConfig().SyncCommitteeSize/params.BeaconConfig().SyncCommitteeSubnetCount)
+	sigsHolder := make([]bls.Signature, 0, params.BeaconConfig().SyncCommitteeSize/subcommitteeCount)
 
-	for i := uint64(0); i < params.BeaconConfig().SyncCommitteeSubnetCount; i++ {
+	for i := uint64(0); i < subcommitteeCount; i++ {
 		cs := proposerContributions.filterBySubIndex(i)
 		aggregates, err := synccontribution.Aggregate(cs)
 		if err != nil {
@@ -107,3 +119,89 @@ func (vs *Server) getSyncAggregate(ctx context.Context, slot primitives.Slot, ro
 		SyncCommitteeSignature: syncSigBytes[:],
 	}, nil
 }
+
+func (vs *Server) aggregatedSyncCommitteeMessages(ctx context.Context, slot primitives.Slot, root [32]byte) ([]*ethpb.SyncCommitteeContribution, error) {
+	subcommitteeCount := params.BeaconConfig().SyncCommitteeSubnetCount
+	subcommitteeSize := params.BeaconConfig().SyncCommitteeSize / subcommitteeCount
+	sigsPerSubcommittee := make([][][]byte, subcommitteeCount)
+	bitsPerSubcommittee := make([]bitfield.Bitfield, subcommitteeCount)
+	for i := uint64(0); i < subcommitteeCount; i++ {
+		sigsPerSubcommittee[i] = make([][]byte, 0, subcommitteeSize)
+		bitsPerSubcommittee[i] = ethpb.NewSyncCommitteeAggregationBits()
+	}
+
+	// Get committee position(s) for each message's validator index.
+	scMessages, err := vs.SyncCommitteePool.SyncCommitteeMessages(slot)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get sync committee messages")
+	}
+	messageIndices := make([]primitives.ValidatorIndex, 0, len(scMessages))
+	messageSigs := make([][]byte, 0, len(scMessages))
+	for _, msg := range scMessages {
+		if bytes.Equal(root[:], msg.BlockRoot) {
+			messageIndices = append(messageIndices, msg.ValidatorIndex)
+			messageSigs = append(messageSigs, msg.Signature)
+		}
+	}
+	headSyncCommitteeIndices := make([][]primitives.CommitteeIndex, len(messageIndices))
+	for i, vi := range messageIndices {
+		headSyncCommitteeIndices[i], err = vs.HeadFetcher.HeadSyncCommitteeIndices(ctx, vi, slot)
+		if err != nil {
+			return nil, errors.Wrapf(err, "could not get head sync committee indices for validator index %d", vi)
+		}
+	}
+
+	// Based on committee position(s), set the appropriate subcommittee bit and signature.
+	for i, indices := range headSyncCommitteeIndices {
+		for _, index := range indices {
+			k := uint64(index)
+			subnetIndex := k / subcommitteeSize
+			indexMod := k % subcommitteeSize
+			if !bitsPerSubcommittee[subnetIndex].BitAt(indexMod) {
+				bitsPerSubcommittee[subnetIndex].SetBitAt(indexMod, true)
+				sigsPerSubcommittee[subnetIndex] = append(sigsPerSubcommittee[subnetIndex], messageSigs[i])
+			}
+		}
+	}
+
+	// Aggregate. To improve performance, we aggregate messages for each subcommittee in parallel.
+	// Using a channel is safer than a slice because appending to a slice is not thread-safe.
+	contributionsChan := make(chan *ethpb.SyncCommitteeContribution, subcommitteeCount)
+	var wg sync.WaitGroup
+	wg.Add(int(subcommitteeCount)) // lint:ignore uintcast -- Subcommittee count will never exceed maxint
+	for i := uint64(0); i < subcommitteeCount; i++ {
+		go func() {
+			defer wg.Done()
+
+			aggregatedSig := make([]byte, 96)
+			aggregatedSig[0] = 0xC0
+			if len(sigsPerSubcommittee[i]) != 0 {
+				uncompressedSigs := make([]bls.Signature, len(sigsPerSubcommittee[i]))
+				for j, sig := range sigsPerSubcommittee[i] {
+					uncompressedSigs[j], err = bls.SignatureFromBytesNoValidation(sig)
+					if err != nil {
+						log.WithError(err).Error("Could not create sync committee signature from bytes")
+						// Skip aggregating this subcommittee
+						return
+					}
+				}
+				contributionsChan <- &ethpb.SyncCommitteeContribution{
+					Slot:              slot,
+					BlockRoot:         root[:],
+					SubcommitteeIndex: i,
+					AggregationBits:   bitsPerSubcommittee[i].Bytes(),
+					Signature:         bls.AggregateSignatures(uncompressedSigs).Marshal(),
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	close(contributionsChan)
+
+	result := make([]*ethpb.SyncCommitteeContribution, 0, len(contributionsChan))
+	for c := range contributionsChan {
+		result = append(result, c)
+	}
+
+	return result, nil
+}