Optimise aggregate and proof signature verification (#11816)

We are trying to optimise `AggregateAndProofService`. After profiling
the service, I see that most of the CPU time is spent on signature
verifications. From the graph, overall, the function took 6.6% (74
seconds) of all the time (not just execution time. Percentage would be
much higher if we took just cpu time) see the screenshot:

<img width="1470" alt="Screenshot 2024-09-01 at 10 28 38"
src="https://github.com/user-attachments/assets/929ce103-2bf3-43d9-a0fa-ca504e4b58bb">


Now we are trying to aggregate all the signatures and verify them
altogether with `bls.VerifyMultipleSignatures` function in an async way
and run the final functions if verifications succeed. I basically
removed all the code where we verified those three signatures and
instead gathered them for verifying later. After profiling that I see
the following output:

<img width="1468" alt="Screenshot 2024-09-01 at 10 44 31"
src="https://github.com/user-attachments/assets/abb842a3-0b4f-4640-8a88-791a2d0af62b">

Now most of the time, as I see, is spent on public key aggregation when
we are verifying validator aggregated signatures. But I guess there is
no way to optimise that one. As we are spending most of the time on
`NewPublicKeyFromBytes` maybe we could cache constructed keys but I
think bls already does that. So, that's as good as it gets.

---------

Co-authored-by: shota.silagadze <shota.silagadze@taal.com>

cl/beacon/handler/pool.go

@@ -283,28 +283,26 @@ func (a *ApiHandler) PostEthV1ValidatorAggregatesAndProof(w http.ResponseWriter,
 
 	failures := []poolingFailure{}
 	for _, v := range req {
-		if err := a.aggregateAndProofsService.ProcessMessage(r.Context(), nil, v); err != nil && !errors.Is(err, services.ErrIgnore) {
+		encodedSSZ, err := v.EncodeSSZ(nil)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			log.Warn("[Beacon REST] failed to encode aggregate and proof", "err", err)
+			return
+		}
+		gossipData := &sentinel.GossipData{
+			Data: encodedSSZ,
+			Name: gossip.TopicNameBeaconAggregateAndProof,
+		}
+
+		// for this service we are not publishing gossipData as the service does it internally, we just pass that data as a parameter.
+		if err := a.aggregateAndProofsService.ProcessMessage(r.Context(), nil, &cltypes.SignedAggregateAndProofData{
+			SignedAggregateAndProof: v,
+			GossipData:              gossipData,
+		}); err != nil && !errors.Is(err, services.ErrIgnore) {
 			log.Warn("[Beacon REST] failed to process bls-change", "err", err)
 			failures = append(failures, poolingFailure{Index: len(failures), Message: err.Error()})
 			continue
 		}
-		// Broadcast to gossip
-		if a.sentinel != nil {
-			encodedSSZ, err := v.EncodeSSZ(nil)
-			if err != nil {
-				http.Error(w, err.Error(), http.StatusInternalServerError)
-				log.Warn("[Beacon REST] failed to encode aggregate and proof", "err", err)
-				return
-			}
-			if _, err := a.sentinel.PublishGossip(r.Context(), &sentinel.GossipData{
-				Data: encodedSSZ,
-				Name: gossip.TopicNameBeaconAggregateAndProof,
-			}); err != nil {
-				log.Warn("[Beacon REST] failed to publish gossip", "err", err)
-				http.Error(w, err.Error(), http.StatusInternalServerError)
-				return
-			}
-		}
 	}
 }
 

cl/beacon/handler/utils_test.go

@@ -120,8 +120,8 @@ func setupTestingHandler(t *testing.T, v clparams.StateVersion, logger log.Logge
 	syncContributionService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.SignedContributionAndProof) error {
 		return h.syncMessagePool.AddSyncContribution(postState, msg.Message.Contribution)
 	}).AnyTimes()
-	aggregateAndProofsService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.SignedAggregateAndProof) error {
-		opPool.AttestationsPool.Insert(msg.Message.Aggregate.Signature(), msg.Message.Aggregate)
+	aggregateAndProofsService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.SignedAggregateAndProofData) error {
+		opPool.AttestationsPool.Insert(msg.SignedAggregateAndProof.Message.Aggregate.Signature(), msg.SignedAggregateAndProof.Message.Aggregate)
 		return nil
 	}).AnyTimes()
 	voluntaryExitService.EXPECT().ProcessMessage(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, subnetID *uint64, msg *cltypes.SignedVoluntaryExit) error {

cl/cltypes/aggregate.go

@@ -18,6 +18,7 @@ package cltypes
 
 import (
 	libcommon "github.com/erigontech/erigon-lib/common"
+	sentinel "github.com/erigontech/erigon-lib/gointerfaces/sentinelproto"
 	"github.com/erigontech/erigon/cl/cltypes/solid"
 	"github.com/erigontech/erigon/cl/merkle_tree"
 	ssz2 "github.com/erigontech/erigon/cl/ssz"
@@ -54,6 +55,16 @@ func (a *AggregateAndProof) HashSSZ() ([32]byte, error) {
 	return merkle_tree.HashTreeRoot(a.AggregatorIndex, a.Aggregate, a.SelectionProof[:])
 }
 
+// SignedAggregateAndProofData is passed to SignedAggregateAndProof service. The service does the signature verification
+// asynchronously. That's why we cannot wait for its ProcessMessage call to finish to check error. The service
+// will do re-publishing of the gossip or banning the peer in case of invalid signature by itself.
+// that's why we are passing sentinel.SentinelClient and *sentinel.GossipData to enable the service
+// to do all of that by itself.
+type SignedAggregateAndProofData struct {
+	SignedAggregateAndProof *SignedAggregateAndProof
+	GossipData              *sentinel.GossipData
+}
+
 type SignedAggregateAndProof struct {
 	Message   *AggregateAndProof `json:"message"`
 	Signature libcommon.Bytes96  `json:"signature"`

cl/phase1/network/gossip_manager.go

@@ -201,8 +201,12 @@ func (g *GossipManager) routeAndProcess(ctx context.Context, data *sentinel.Goss
 		}
 		return g.blsToExecutionChangeService.ProcessMessage(ctx, data.SubnetId, obj)
 	case gossip.TopicNameBeaconAggregateAndProof:
-		obj := &cltypes.SignedAggregateAndProof{}
-		if err := obj.DecodeSSZ(data.Data, int(version)); err != nil {
+		obj := &cltypes.SignedAggregateAndProofData{
+			GossipData:              data,
+			SignedAggregateAndProof: &cltypes.SignedAggregateAndProof{},
+		}
+
+		if err := obj.SignedAggregateAndProof.DecodeSSZ(data.Data, int(version)); err != nil {
 			return err
 		}
 		return g.aggregateAndProofService.ProcessMessage(ctx, data.SubnetId, obj)