From 72f749bbad98c0b9a01485a2b26fb635c5c77701 Mon Sep 17 00:00:00 2001 From: Kewei Date: Thu, 19 Sep 2024 02:25:22 +0800 Subject: [PATCH] skip superset aggregatation (#12019) --- cl/aggregation/pool_impl.go | 6 +--- cl/phase1/network/gossip_manager.go | 2 +- .../network/services/attestation_service.go | 6 +++- .../services/attestation_service_test.go | 3 +- .../committee_subscription.go | 23 +++++++++++--- .../committee_subscription/interface.go | 4 +-- .../mock_services/committee_subscribe_mock.go | 30 +++++++++---------- 7 files changed, 45 insertions(+), 29 deletions(-) diff --git a/cl/aggregation/pool_impl.go b/cl/aggregation/pool_impl.go index 38afe528255..358715ced4c 100644 --- a/cl/aggregation/pool_impl.go +++ b/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) { diff --git a/cl/phase1/network/gossip_manager.go b/cl/phase1/network/gossip_manager.go index 2effacfd1f3..8b9e09e9da5 100644 --- a/cl/phase1/network/gossip_manager.go +++ b/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) } diff --git a/cl/phase1/network/services/attestation_service.go b/cl/phase1/network/services/attestation_service.go index 6638c1ed527..780435a20ce 100644 --- a/cl/phase1/network/services/attestation_service.go +++ b/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 } diff --git a/cl/phase1/network/services/attestation_service_test.go b/cl/phase1/network/services/attestation_service_test.go index 3b8becf459f..400a388982c 100644 --- a/cl/phase1/network/services/attestation_service_test.go +++ b/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(), diff --git a/cl/validator/committee_subscription/committee_subscription.go b/cl/validator/committee_subscription/committee_subscription.go index a462c828607..c1303753fb5 100644 --- a/cl/validator/committee_subscription/committee_subscription.go +++ b/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 } diff --git a/cl/validator/committee_subscription/interface.go b/cl/validator/committee_subscription/interface.go index 0dc47b4f10f..7c1982f28b1 100644 --- a/cl/validator/committee_subscription/interface.go +++ b/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 } diff --git a/cl/validator/committee_subscription/mock_services/committee_subscribe_mock.go b/cl/validator/committee_subscription/mock_services/committee_subscribe_mock.go index a92d667e24e..b809ec83bed 100644 --- a/cl/validator/committee_subscription/mock_services/committee_subscribe_mock.go +++ b/cl/validator/committee_subscription/mock_services/committee_subscribe_mock.go @@ -79,46 +79,46 @@ func (c *MockCommitteeSubscribeAddAttestationSubscriptionCall) DoAndReturn(f fun return c } -// CheckAggregateAttestation mocks base method. -func (m *MockCommitteeSubscribe) CheckAggregateAttestation(arg0 *solid.Attestation) error { +// AggregateAttestation mocks base method. +func (m *MockCommitteeSubscribe) AggregateAttestation(arg0 *solid.Attestation) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "CheckAggregateAttestation", arg0) + ret := m.ctrl.Call(m, "AggregateAttestation", arg0) ret0, _ := ret[0].(error) return ret0 } -// CheckAggregateAttestation indicates an expected call of CheckAggregateAttestation. -func (mr *MockCommitteeSubscribeMockRecorder) CheckAggregateAttestation(arg0 any) *MockCommitteeSubscribeCheckAggregateAttestationCall { +// AggregateAttestation indicates an expected call of AggregateAttestation. +func (mr *MockCommitteeSubscribeMockRecorder) AggregateAttestation(arg0 any) *MockCommitteeSubscribeAggregateAttestationCall { mr.mock.ctrl.T.Helper() - call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CheckAggregateAttestation", reflect.TypeOf((*MockCommitteeSubscribe)(nil).CheckAggregateAttestation), arg0) - return &MockCommitteeSubscribeCheckAggregateAttestationCall{Call: call} + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AggregateAttestation", reflect.TypeOf((*MockCommitteeSubscribe)(nil).AggregateAttestation), arg0) + return &MockCommitteeSubscribeAggregateAttestationCall{Call: call} } -// MockCommitteeSubscribeCheckAggregateAttestationCall wrap *gomock.Call -type MockCommitteeSubscribeCheckAggregateAttestationCall struct { +// MockCommitteeSubscribeAggregateAttestationCall wrap *gomock.Call +type MockCommitteeSubscribeAggregateAttestationCall struct { *gomock.Call } // Return rewrite *gomock.Call.Return -func (c *MockCommitteeSubscribeCheckAggregateAttestationCall) Return(arg0 error) *MockCommitteeSubscribeCheckAggregateAttestationCall { +func (c *MockCommitteeSubscribeAggregateAttestationCall) Return(arg0 error) *MockCommitteeSubscribeAggregateAttestationCall { c.Call = c.Call.Return(arg0) return c } // Do rewrite *gomock.Call.Do -func (c *MockCommitteeSubscribeCheckAggregateAttestationCall) Do(f func(*solid.Attestation) error) *MockCommitteeSubscribeCheckAggregateAttestationCall { +func (c *MockCommitteeSubscribeAggregateAttestationCall) Do(f func(*solid.Attestation) error) *MockCommitteeSubscribeAggregateAttestationCall { c.Call = c.Call.Do(f) return c } // DoAndReturn rewrite *gomock.Call.DoAndReturn -func (c *MockCommitteeSubscribeCheckAggregateAttestationCall) DoAndReturn(f func(*solid.Attestation) error) *MockCommitteeSubscribeCheckAggregateAttestationCall { +func (c *MockCommitteeSubscribeAggregateAttestationCall) DoAndReturn(f func(*solid.Attestation) error) *MockCommitteeSubscribeAggregateAttestationCall { c.Call = c.Call.DoAndReturn(f) return c } // NeedToAggregate mocks base method. -func (m *MockCommitteeSubscribe) NeedToAggregate(arg0 uint64) bool { +func (m *MockCommitteeSubscribe) NeedToAggregate(arg0 *solid.Attestation) bool { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "NeedToAggregate", arg0) ret0, _ := ret[0].(bool) @@ -144,13 +144,13 @@ func (c *MockCommitteeSubscribeNeedToAggregateCall) Return(arg0 bool) *MockCommi } // Do rewrite *gomock.Call.Do -func (c *MockCommitteeSubscribeNeedToAggregateCall) Do(f func(uint64) bool) *MockCommitteeSubscribeNeedToAggregateCall { +func (c *MockCommitteeSubscribeNeedToAggregateCall) Do(f func(*solid.Attestation) bool) *MockCommitteeSubscribeNeedToAggregateCall { c.Call = c.Call.Do(f) return c } // DoAndReturn rewrite *gomock.Call.DoAndReturn -func (c *MockCommitteeSubscribeNeedToAggregateCall) DoAndReturn(f func(uint64) bool) *MockCommitteeSubscribeNeedToAggregateCall { +func (c *MockCommitteeSubscribeNeedToAggregateCall) DoAndReturn(f func(*solid.Attestation) bool) *MockCommitteeSubscribeNeedToAggregateCall { c.Call = c.Call.DoAndReturn(f) return c }