skip superset aggregatation (#12019)

cl/aggregation/pool_impl.go

@@ -112,11 +112,7 @@ func (p *aggregationPoolImpl) AddAttestation(inAtt *solid.Attestation) error {
 func (p *aggregationPoolImpl) GetAggregatationByRoot(root common.Hash) *solid.Attestation {
 	p.aggregatesLock.RLock()
 	defer p.aggregatesLock.RUnlock()
-	att := p.aggregates[root]
-	if att == nil {
-		return nil
-	}
-	return att.Copy()
+	return p.aggregates[root]
 }
 
 func (p *aggregationPoolImpl) sweepStaleAtt(ctx context.Context) {

cl/phase1/network/gossip_manager.go

@@ -259,7 +259,7 @@ func (g *GossipManager) routeAndProcess(ctx context.Context, data *sentinel.Goss
 				return err
 			}
 
-			if g.committeeSub.NeedToAggregate(obj.Attestation.AttestantionData().CommitteeIndex()) {
+			if g.committeeSub.NeedToAggregate(obj.Attestation) {
 				return g.attestationService.ProcessMessage(ctx, data.SubnetId, obj)
 			}
 

cl/phase1/network/services/attestation_service.go

@@ -223,13 +223,17 @@ func (s *attestationService) ProcessMessage(ctx context.Context, subnet *uint64,
 		return fmt.Errorf("invalid finalized checkpoint %w", ErrIgnore)
 	}
 
+	if !s.committeeSubscribe.NeedToAggregate(att.Attestation) {
+		return ErrIgnore
+	}
+
 	aggregateVerificationData := &AggregateVerificationData{
 		Signatures: [][]byte{signature[:]},
 		SignRoots:  [][]byte{signingRoot[:]},
 		Pks:        [][]byte{pubKey[:]},
 		GossipData: att.GossipData,
 		F: func() {
-			err = s.committeeSubscribe.CheckAggregateAttestation(att.Attestation)
+			err = s.committeeSubscribe.AggregateAttestation(att.Attestation)
 			if errors.Is(err, aggregation.ErrIsSuperset) {
 				return
 			}

cl/phase1/network/services/attestation_service_test.go

@@ -330,7 +330,8 @@ func (t *attestationTestSuite) TestAttestationProcessMessage() {
 				t.mockForkChoice.FinalizedCheckpointVal = solid.NewCheckpointFromParameters(
 					mockFinalizedCheckPoint.BlockRoot(),
 					mockFinalizedCheckPoint.Epoch())
-				t.committeeSubscibe.EXPECT().CheckAggregateAttestation(att).Return(nil).Times(1)
+				t.committeeSubscibe.EXPECT().NeedToAggregate(att).Return(true).Times(1)
+				t.committeeSubscibe.EXPECT().AggregateAttestation(att).Return(nil).Times(1)
 			},
 			args: args{
 				ctx:    context.Background(),

cl/validator/committee_subscription/committee_subscription.go

@@ -34,6 +34,7 @@ import (
 	"github.com/erigontech/erigon/cl/gossip"
 	"github.com/erigontech/erigon/cl/phase1/core/state"
 	"github.com/erigontech/erigon/cl/phase1/network/subnets"
+	"github.com/erigontech/erigon/cl/utils"
 	"github.com/erigontech/erigon/cl/utils/eth_clock"
 )
 
@@ -135,7 +136,7 @@ func (c *CommitteeSubscribeMgmt) AddAttestationSubscription(ctx context.Context,
 	return nil
 }
 
-func (c *CommitteeSubscribeMgmt) CheckAggregateAttestation(att *solid.Attestation) error {
+func (c *CommitteeSubscribeMgmt) AggregateAttestation(att *solid.Attestation) error {
 	committeeIndex := att.AttestantionData().CommitteeIndex()
 	c.validatorSubsMutex.RLock()
 	defer c.validatorSubsMutex.RUnlock()
@@ -148,11 +149,25 @@ func (c *CommitteeSubscribeMgmt) CheckAggregateAttestation(att *solid.Attestatio
 	return nil
 }
 
-func (c *CommitteeSubscribeMgmt) NeedToAggregate(committeeIndex uint64) bool {
+func (c *CommitteeSubscribeMgmt) NeedToAggregate(att *solid.Attestation) bool {
+	var (
+		committeeIndex = att.AttestantionData().CommitteeIndex()
+	)
+
 	c.validatorSubsMutex.RLock()
 	defer c.validatorSubsMutex.RUnlock()
-	if sub, ok := c.validatorSubs[committeeIndex]; ok {
-		return sub.aggregate
+	if sub, ok := c.validatorSubs[committeeIndex]; ok && sub.aggregate {
+		root, err := att.AttestantionData().HashSSZ()
+		if err != nil {
+			log.Warn("failed to hash attestation data", "err", err)
+			return false
+		}
+		aggregation := c.aggregationPool.GetAggregatationByRoot(root)
+		if aggregation == nil ||
+			!utils.IsNonStrictSupersetBitlist(aggregation.AggregationBits(), att.AggregationBits()) {
+			// the on bit is not set. need to aggregate
+			return true
+		}
 	}
 	return false
 }

cl/validator/committee_subscription/interface.go

@@ -26,6 +26,6 @@ import (
 //go:generate mockgen -typed=true -destination=./mock_services/committee_subscribe_mock.go -package=mock_services . CommitteeSubscribe
 type CommitteeSubscribe interface {
 	AddAttestationSubscription(ctx context.Context, p *cltypes.BeaconCommitteeSubscription) error
-	CheckAggregateAttestation(att *solid.Attestation) error
-	NeedToAggregate(committeeIndex uint64) bool
+	AggregateAttestation(att *solid.Attestation) error
+	NeedToAggregate(att *solid.Attestation) bool
 }