From 54ee83ac0d4a9b3577a21edb90dd13a433c3f5f4 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Tue, 10 May 2022 16:42:21 -0700 Subject: [PATCH 01/13] Ignore subset aggregates --- .../operations/synccommittee/contribution.go | 30 +++++++++++++++++++ beacon-chain/operations/synccommittee/pool.go | 2 ++ beacon-chain/sync/validate_aggregate_proof.go | 8 +++++ .../sync/validate_sync_contribution_proof.go | 17 +++++++++++ 4 files changed, 57 insertions(+) diff --git a/beacon-chain/operations/synccommittee/contribution.go b/beacon-chain/operations/synccommittee/contribution.go index e5e73ecd7b34..45ffd0038499 100644 --- a/beacon-chain/operations/synccommittee/contribution.go +++ b/beacon-chain/operations/synccommittee/contribution.go @@ -1,6 +1,7 @@ package synccommittee import ( + "bytes" "strconv" "github.com/pkg/errors" @@ -85,6 +86,35 @@ func (s *Store) SyncCommitteeContributions(slot types.Slot) ([]*ethpb.SyncCommit return contributions, nil } +// HasSyncCommitteeContribution returns true if a sync committee contribution exits in the cache. +// Exist means there's a similar contribution where the aggregation bits is the superset of the input bits. +func (s *Store) HasSyncCommitteeContribution(c *ethpb.SyncCommitteeContribution) (bool, error) { + s.contributionLock.RLock() + defer s.contributionLock.RUnlock() + + item := s.contributionCache.RetrieveByKey(syncCommitteeKey(c.Slot)) + if item == nil { + return false, nil + } + contributions, ok := item.Value.([]*ethpb.SyncCommitteeContribution) + if !ok { + return false, errors.New("not typed []ethpb.SyncCommitteeContribution") + } + for _, contribution := range contributions { + if bytes.Equal(c.BlockRoot, contribution.BlockRoot) && c.SubcommitteeIndex == contribution.SubcommitteeIndex { + c, err := contribution.AggregationBits.Contains(c.AggregationBits) + if err != nil { + return false, err + } + if c { + return true, nil + } + } + } + + return false, nil +} + func syncCommitteeKey(slot types.Slot) string { return strconv.FormatUint(uint64(slot), 10) } diff --git a/beacon-chain/operations/synccommittee/pool.go b/beacon-chain/operations/synccommittee/pool.go index 594c3f133b73..b469932a0650 100644 --- a/beacon-chain/operations/synccommittee/pool.go +++ b/beacon-chain/operations/synccommittee/pool.go @@ -19,6 +19,8 @@ type Pool interface { // Methods for Sync Committee Messages. SaveSyncCommitteeMessage(sig *ethpb.SyncCommitteeMessage) error SyncCommitteeMessages(slot types.Slot) ([]*ethpb.SyncCommitteeMessage, error) + + HasSyncCommitteeContribution(c *ethpb.SyncCommitteeContribution) (bool, error) } // NewPool returns the sync committee store fulfilling the pool interface. diff --git a/beacon-chain/sync/validate_aggregate_proof.go b/beacon-chain/sync/validate_aggregate_proof.go index 743fc3c6ff13..8be17d36ff2c 100644 --- a/beacon-chain/sync/validate_aggregate_proof.go +++ b/beacon-chain/sync/validate_aggregate_proof.go @@ -91,6 +91,14 @@ func (s *Service) validateAggregateAndProof(ctx context.Context, pid peer.ID, ms return pubsub.ValidationIgnore, err } + has, err := s.cfg.attPool.HasAggregatedAttestation(m.Message.Aggregate) + if err != nil { + return pubsub.ValidationIgnore, err + } + if has { + return pubsub.ValidationIgnore, err + } + // Verify this is the first aggregate received from the aggregator with index and slot. if s.hasSeenAggregatorIndexEpoch(m.Message.Aggregate.Data.Target.Epoch, m.Message.AggregatorIndex) { return pubsub.ValidationIgnore, nil diff --git a/beacon-chain/sync/validate_sync_contribution_proof.go b/beacon-chain/sync/validate_sync_contribution_proof.go index b41866ed2613..b6d45c2e732b 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof.go +++ b/beacon-chain/sync/validate_sync_contribution_proof.go @@ -148,6 +148,23 @@ func rejectEmptyContribution(m *ethpb.SignedContributionAndProof) validationFn { } func (s *Service) ignoreSeenSyncContribution(m *ethpb.SignedContributionAndProof) validationFn { + return func(ctx context.Context) (pubsub.ValidationResult, error) { + seen := s.hasSeenSyncContributionIndexSlot(m.Message.Contribution.Slot, m.Message.AggregatorIndex, types.CommitteeIndex(m.Message.Contribution.SubcommitteeIndex)) + if seen { + return pubsub.ValidationIgnore, nil + } + seen, err := s.cfg.syncCommsPool.HasSyncCommitteeContribution(m.Message.Contribution) + if err != nil { + return pubsub.ValidationIgnore, err + } + if seen { + return pubsub.ValidationIgnore, nil + } + return pubsub.ValidationAccept, nil + } +} + +func (s *Service) ignoreCached(m *ethpb.SignedContributionAndProof) validationFn { return func(ctx context.Context) (pubsub.ValidationResult, error) { seen := s.hasSeenSyncContributionIndexSlot(m.Message.Contribution.Slot, m.Message.AggregatorIndex, types.CommitteeIndex(m.Message.Contribution.SubcommitteeIndex)) if seen { From 9d8f6d69797c541ecbe3d86d6ba769e92a9eea29 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Tue, 17 May 2022 09:37:16 -0700 Subject: [PATCH 02/13] Add test --- .../synccommittee/contribution_test.go | 119 ++++++++++++++++++ 1 file changed, 119 insertions(+) diff --git a/beacon-chain/operations/synccommittee/contribution_test.go b/beacon-chain/operations/synccommittee/contribution_test.go index 6689efe0a93e..a3222b26e82e 100644 --- a/beacon-chain/operations/synccommittee/contribution_test.go +++ b/beacon-chain/operations/synccommittee/contribution_test.go @@ -3,6 +3,7 @@ package synccommittee import ( "testing" + "github.com/prysmaticlabs/go-bitfield" ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1" "github.com/prysmaticlabs/prysm/testing/require" ) @@ -106,3 +107,121 @@ func TestSyncCommitteeContributionCache_RoundTrip(t *testing.T) { {Slot: 6, SubcommitteeIndex: 1, Signature: []byte{'l'}}, }, conts) } + +func TestSyncCommitteeContributionCache_HasSyncCommitteeContribution(t *testing.T) { + store := NewStore() + + b0 := bitfield.NewBitvector128() + b0.SetBitAt(0, true) + b1 := bitfield.NewBitvector128() + b1.SetBitAt(1, true) + b2 := bitfield.NewBitvector128() + b2.SetBitAt(0, true) + b2.SetBitAt(1, true) + b2.SetBitAt(2, true) + + conts := []*ethpb.SyncCommitteeContribution{ + {Slot: 1, SubcommitteeIndex: 0, AggregationBits: b0}, + {Slot: 1, SubcommitteeIndex: 1, AggregationBits: b1}, + {Slot: 1, SubcommitteeIndex: 1, BlockRoot: []byte{'a'}, AggregationBits: b1}, + {Slot: 1, SubcommitteeIndex: 2, AggregationBits: b2}, + } + + for _, sig := range conts { + require.NoError(t, store.SaveSyncCommitteeContribution(sig)) + } + + type test struct { + name string + input *ethpb.SyncCommitteeContribution + want bool + } + tests := []test{ + { + name: "slot 0, has = false", + input: ðpb.SyncCommitteeContribution{ + AggregationBits: b2, + }, + want: false, + }, + { + name: "slot 1, same root, bit overlaps, has = true", + input: ðpb.SyncCommitteeContribution{ + Slot: 1, + AggregationBits: b0, + }, + want: true, + }, + { + name: "slot 1, same root, bit doesn't overlaps, has = false", + input: ðpb.SyncCommitteeContribution{ + Slot: 1, + AggregationBits: b1, + }, + want: false, + }, + { + name: "slot 1, same root, bit doesn't overlaps, has = false", + input: ðpb.SyncCommitteeContribution{ + Slot: 1, + AggregationBits: b2, + }, + want: false, + }, + { + name: "slot 1, same root, bit overlaps, has = true", + input: ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 1, + AggregationBits: b1, + }, + want: true, + }, + { + name: "slot 1, different root, bit overlaps, has = true", + input: ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 1, + BlockRoot: []byte{'a'}, + AggregationBits: b1, + }, + want: true, + }, + { + name: "slot 1, different root, different committee, bit doesn't overlaps, has = false", + input: ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 3, + BlockRoot: []byte{'a'}, + AggregationBits: b1, + }, + want: false, + }, + { + name: "slot 1, different root, bit doesn't overlaps, has = false", + input: ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 1, + BlockRoot: []byte{'b'}, + AggregationBits: b1, + }, + want: false, + }, + { + name: "slot 1, same root, bit doesn't overlaps, has = true", + input: ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 1, + AggregationBits: b2, + }, + want: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := store.HasSyncCommitteeContribution(tt.input) + require.NoError(t, err) + require.Equal(t, tt.want, got) + }) + } +} From 6fa31aa71dd29e69ff4499eebc16eca6fe9f212b Mon Sep 17 00:00:00 2001 From: terence tsao Date: Tue, 17 May 2022 10:26:41 -0700 Subject: [PATCH 03/13] Update BUILD.bazel --- beacon-chain/operations/synccommittee/BUILD.bazel | 1 + 1 file changed, 1 insertion(+) diff --git a/beacon-chain/operations/synccommittee/BUILD.bazel b/beacon-chain/operations/synccommittee/BUILD.bazel index d66b8b00589d..cb144709c347 100644 --- a/beacon-chain/operations/synccommittee/BUILD.bazel +++ b/beacon-chain/operations/synccommittee/BUILD.bazel @@ -32,5 +32,6 @@ go_test( deps = [ "//proto/prysm/v1alpha1:go_default_library", "//testing/require:go_default_library", + "@com_github_prysmaticlabs_go_bitfield//:go_default_library", ], ) From e1368e90a801ea437cfd0dcd66e3818f41f6d481 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Tue, 17 May 2022 13:33:14 -0700 Subject: [PATCH 04/13] Update validate_sync_contribution_proof_test.go --- beacon-chain/sync/BUILD.bazel | 1 + beacon-chain/sync/validate_sync_contribution_proof_test.go | 3 +++ 2 files changed, 4 insertions(+) diff --git a/beacon-chain/sync/BUILD.bazel b/beacon-chain/sync/BUILD.bazel index e6149f374134..ee35bdb73a15 100644 --- a/beacon-chain/sync/BUILD.bazel +++ b/beacon-chain/sync/BUILD.bazel @@ -181,6 +181,7 @@ go_test( "//beacon-chain/db/testing:go_default_library", "//beacon-chain/operations/attestations:go_default_library", "//beacon-chain/operations/slashings:go_default_library", + "//beacon-chain/operations/synccommittee:go_default_library", "//beacon-chain/p2p:go_default_library", "//beacon-chain/p2p/encoder:go_default_library", "//beacon-chain/p2p/peers:go_default_library", diff --git a/beacon-chain/sync/validate_sync_contribution_proof_test.go b/beacon-chain/sync/validate_sync_contribution_proof_test.go index 4e3c21830472..36ca6d13fbb1 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof_test.go +++ b/beacon-chain/sync/validate_sync_contribution_proof_test.go @@ -21,6 +21,7 @@ import ( "github.com/prysmaticlabs/prysm/beacon-chain/core/transition" "github.com/prysmaticlabs/prysm/beacon-chain/db" testingDB "github.com/prysmaticlabs/prysm/beacon-chain/db/testing" + "github.com/prysmaticlabs/prysm/beacon-chain/operations/synccommittee" "github.com/prysmaticlabs/prysm/beacon-chain/p2p" "github.com/prysmaticlabs/prysm/beacon-chain/p2p/encoder" mockp2p "github.com/prysmaticlabs/prysm/beacon-chain/p2p/testing" @@ -72,6 +73,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -925,6 +927,7 @@ func TestValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ) go s.verifierRoutine() s.cfg.stateGen = stategen.New(db) From 677b2e1101686837e12f462fb9549cd4e3789b14 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Tue, 17 May 2022 13:33:14 -0700 Subject: [PATCH 05/13] Update validate_sync_contribution_proof_test.go --- beacon-chain/sync/BUILD.bazel | 1 + .../sync/validate_sync_contribution_proof_test.go | 10 ++++++++++ 2 files changed, 11 insertions(+) diff --git a/beacon-chain/sync/BUILD.bazel b/beacon-chain/sync/BUILD.bazel index e6149f374134..ee35bdb73a15 100644 --- a/beacon-chain/sync/BUILD.bazel +++ b/beacon-chain/sync/BUILD.bazel @@ -181,6 +181,7 @@ go_test( "//beacon-chain/db/testing:go_default_library", "//beacon-chain/operations/attestations:go_default_library", "//beacon-chain/operations/slashings:go_default_library", + "//beacon-chain/operations/synccommittee:go_default_library", "//beacon-chain/p2p:go_default_library", "//beacon-chain/p2p/encoder:go_default_library", "//beacon-chain/p2p/peers:go_default_library", diff --git a/beacon-chain/sync/validate_sync_contribution_proof_test.go b/beacon-chain/sync/validate_sync_contribution_proof_test.go index 4e3c21830472..38a73b9230fc 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof_test.go +++ b/beacon-chain/sync/validate_sync_contribution_proof_test.go @@ -21,6 +21,7 @@ import ( "github.com/prysmaticlabs/prysm/beacon-chain/core/transition" "github.com/prysmaticlabs/prysm/beacon-chain/db" testingDB "github.com/prysmaticlabs/prysm/beacon-chain/db/testing" + "github.com/prysmaticlabs/prysm/beacon-chain/operations/synccommittee" "github.com/prysmaticlabs/prysm/beacon-chain/p2p" "github.com/prysmaticlabs/prysm/beacon-chain/p2p/encoder" mockp2p "github.com/prysmaticlabs/prysm/beacon-chain/p2p/testing" @@ -72,6 +73,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -265,6 +267,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -309,6 +312,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -371,6 +375,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -436,6 +441,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -516,6 +522,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -682,6 +689,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -778,6 +786,7 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -925,6 +934,7 @@ func TestValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), + WithSyncCommsPool(synccommittee.NewPool()), ) go s.verifierRoutine() s.cfg.stateGen = stategen.New(db) From b1f7eeb534f80282ae8d481cae41d041950d29e1 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Wed, 18 May 2022 09:45:36 -0700 Subject: [PATCH 06/13] Don't utilize pooled objects. Use direct caches --- .../operations/synccommittee/BUILD.bazel | 1 - .../operations/synccommittee/contribution.go | 30 ----- .../synccommittee/contribution_test.go | 119 ------------------ beacon-chain/operations/synccommittee/pool.go | 2 - beacon-chain/sync/BUILD.bazel | 1 - .../sync/pending_attestations_queue.go | 5 +- beacon-chain/sync/service.go | 3 + beacon-chain/sync/validate_aggregate_proof.go | 75 ++++++++--- .../sync/validate_aggregate_proof_test.go | 53 ++++++++ .../sync/validate_sync_contribution_proof.go | 66 +++++++++- .../validate_sync_contribution_proof_test.go | 73 +++++++++-- 11 files changed, 240 insertions(+), 188 deletions(-) diff --git a/beacon-chain/operations/synccommittee/BUILD.bazel b/beacon-chain/operations/synccommittee/BUILD.bazel index cb144709c347..d66b8b00589d 100644 --- a/beacon-chain/operations/synccommittee/BUILD.bazel +++ b/beacon-chain/operations/synccommittee/BUILD.bazel @@ -32,6 +32,5 @@ go_test( deps = [ "//proto/prysm/v1alpha1:go_default_library", "//testing/require:go_default_library", - "@com_github_prysmaticlabs_go_bitfield//:go_default_library", ], ) diff --git a/beacon-chain/operations/synccommittee/contribution.go b/beacon-chain/operations/synccommittee/contribution.go index 45ffd0038499..e5e73ecd7b34 100644 --- a/beacon-chain/operations/synccommittee/contribution.go +++ b/beacon-chain/operations/synccommittee/contribution.go @@ -1,7 +1,6 @@ package synccommittee import ( - "bytes" "strconv" "github.com/pkg/errors" @@ -86,35 +85,6 @@ func (s *Store) SyncCommitteeContributions(slot types.Slot) ([]*ethpb.SyncCommit return contributions, nil } -// HasSyncCommitteeContribution returns true if a sync committee contribution exits in the cache. -// Exist means there's a similar contribution where the aggregation bits is the superset of the input bits. -func (s *Store) HasSyncCommitteeContribution(c *ethpb.SyncCommitteeContribution) (bool, error) { - s.contributionLock.RLock() - defer s.contributionLock.RUnlock() - - item := s.contributionCache.RetrieveByKey(syncCommitteeKey(c.Slot)) - if item == nil { - return false, nil - } - contributions, ok := item.Value.([]*ethpb.SyncCommitteeContribution) - if !ok { - return false, errors.New("not typed []ethpb.SyncCommitteeContribution") - } - for _, contribution := range contributions { - if bytes.Equal(c.BlockRoot, contribution.BlockRoot) && c.SubcommitteeIndex == contribution.SubcommitteeIndex { - c, err := contribution.AggregationBits.Contains(c.AggregationBits) - if err != nil { - return false, err - } - if c { - return true, nil - } - } - } - - return false, nil -} - func syncCommitteeKey(slot types.Slot) string { return strconv.FormatUint(uint64(slot), 10) } diff --git a/beacon-chain/operations/synccommittee/contribution_test.go b/beacon-chain/operations/synccommittee/contribution_test.go index a3222b26e82e..6689efe0a93e 100644 --- a/beacon-chain/operations/synccommittee/contribution_test.go +++ b/beacon-chain/operations/synccommittee/contribution_test.go @@ -3,7 +3,6 @@ package synccommittee import ( "testing" - "github.com/prysmaticlabs/go-bitfield" ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1" "github.com/prysmaticlabs/prysm/testing/require" ) @@ -107,121 +106,3 @@ func TestSyncCommitteeContributionCache_RoundTrip(t *testing.T) { {Slot: 6, SubcommitteeIndex: 1, Signature: []byte{'l'}}, }, conts) } - -func TestSyncCommitteeContributionCache_HasSyncCommitteeContribution(t *testing.T) { - store := NewStore() - - b0 := bitfield.NewBitvector128() - b0.SetBitAt(0, true) - b1 := bitfield.NewBitvector128() - b1.SetBitAt(1, true) - b2 := bitfield.NewBitvector128() - b2.SetBitAt(0, true) - b2.SetBitAt(1, true) - b2.SetBitAt(2, true) - - conts := []*ethpb.SyncCommitteeContribution{ - {Slot: 1, SubcommitteeIndex: 0, AggregationBits: b0}, - {Slot: 1, SubcommitteeIndex: 1, AggregationBits: b1}, - {Slot: 1, SubcommitteeIndex: 1, BlockRoot: []byte{'a'}, AggregationBits: b1}, - {Slot: 1, SubcommitteeIndex: 2, AggregationBits: b2}, - } - - for _, sig := range conts { - require.NoError(t, store.SaveSyncCommitteeContribution(sig)) - } - - type test struct { - name string - input *ethpb.SyncCommitteeContribution - want bool - } - tests := []test{ - { - name: "slot 0, has = false", - input: ðpb.SyncCommitteeContribution{ - AggregationBits: b2, - }, - want: false, - }, - { - name: "slot 1, same root, bit overlaps, has = true", - input: ðpb.SyncCommitteeContribution{ - Slot: 1, - AggregationBits: b0, - }, - want: true, - }, - { - name: "slot 1, same root, bit doesn't overlaps, has = false", - input: ðpb.SyncCommitteeContribution{ - Slot: 1, - AggregationBits: b1, - }, - want: false, - }, - { - name: "slot 1, same root, bit doesn't overlaps, has = false", - input: ðpb.SyncCommitteeContribution{ - Slot: 1, - AggregationBits: b2, - }, - want: false, - }, - { - name: "slot 1, same root, bit overlaps, has = true", - input: ðpb.SyncCommitteeContribution{ - Slot: 1, - SubcommitteeIndex: 1, - AggregationBits: b1, - }, - want: true, - }, - { - name: "slot 1, different root, bit overlaps, has = true", - input: ðpb.SyncCommitteeContribution{ - Slot: 1, - SubcommitteeIndex: 1, - BlockRoot: []byte{'a'}, - AggregationBits: b1, - }, - want: true, - }, - { - name: "slot 1, different root, different committee, bit doesn't overlaps, has = false", - input: ðpb.SyncCommitteeContribution{ - Slot: 1, - SubcommitteeIndex: 3, - BlockRoot: []byte{'a'}, - AggregationBits: b1, - }, - want: false, - }, - { - name: "slot 1, different root, bit doesn't overlaps, has = false", - input: ðpb.SyncCommitteeContribution{ - Slot: 1, - SubcommitteeIndex: 1, - BlockRoot: []byte{'b'}, - AggregationBits: b1, - }, - want: false, - }, - { - name: "slot 1, same root, bit doesn't overlaps, has = true", - input: ðpb.SyncCommitteeContribution{ - Slot: 1, - SubcommitteeIndex: 1, - AggregationBits: b2, - }, - want: false, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got, err := store.HasSyncCommitteeContribution(tt.input) - require.NoError(t, err) - require.Equal(t, tt.want, got) - }) - } -} diff --git a/beacon-chain/operations/synccommittee/pool.go b/beacon-chain/operations/synccommittee/pool.go index b469932a0650..594c3f133b73 100644 --- a/beacon-chain/operations/synccommittee/pool.go +++ b/beacon-chain/operations/synccommittee/pool.go @@ -19,8 +19,6 @@ type Pool interface { // Methods for Sync Committee Messages. SaveSyncCommitteeMessage(sig *ethpb.SyncCommitteeMessage) error SyncCommitteeMessages(slot types.Slot) ([]*ethpb.SyncCommitteeMessage, error) - - HasSyncCommitteeContribution(c *ethpb.SyncCommitteeContribution) (bool, error) } // NewPool returns the sync committee store fulfilling the pool interface. diff --git a/beacon-chain/sync/BUILD.bazel b/beacon-chain/sync/BUILD.bazel index ee35bdb73a15..e6149f374134 100644 --- a/beacon-chain/sync/BUILD.bazel +++ b/beacon-chain/sync/BUILD.bazel @@ -181,7 +181,6 @@ go_test( "//beacon-chain/db/testing:go_default_library", "//beacon-chain/operations/attestations:go_default_library", "//beacon-chain/operations/slashings:go_default_library", - "//beacon-chain/operations/synccommittee:go_default_library", "//beacon-chain/p2p:go_default_library", "//beacon-chain/p2p/encoder:go_default_library", "//beacon-chain/p2p/peers:go_default_library", diff --git a/beacon-chain/sync/pending_attestations_queue.go b/beacon-chain/sync/pending_attestations_queue.go index d3e0789a8461..314537a2a565 100644 --- a/beacon-chain/sync/pending_attestations_queue.go +++ b/beacon-chain/sync/pending_attestations_queue.go @@ -79,7 +79,10 @@ func (s *Service) processPendingAtts(ctx context.Context) error { log.WithError(err).Debug("Could not save aggregate attestation") continue } - s.setAggregatorIndexEpochSeen(att.Aggregate.Data.Target.Epoch, att.AggregatorIndex) + if err := s.setAggregationBits(att.Aggregate.Data, att.Aggregate.AggregationBits); err != nil { + log.WithError(err).Debug("Could not set") + continue + } // Broadcasting the signed attestation again once a node is able to process it. if err := s.cfg.p2p.Broadcast(ctx, signedAtt); err != nil { diff --git a/beacon-chain/sync/service.go b/beacon-chain/sync/service.go index 2ba50617e3eb..c085d86704e9 100644 --- a/beacon-chain/sync/service.go +++ b/beacon-chain/sync/service.go @@ -132,6 +132,8 @@ type Service struct { seenSyncContributionCache *lru.Cache badBlockCache *lru.Cache badBlockLock sync.RWMutex + syncContributionBitsOverlapLock sync.RWMutex + syncContributionBitsOverlapCache *lru.Cache signatureChan chan *signatureVerifier } @@ -221,6 +223,7 @@ func (s *Service) initCaches() { s.seenUnAggregatedAttestationCache = lruwrpr.New(seenUnaggregatedAttSize) s.seenSyncMessageCache = lruwrpr.New(seenSyncMsgSize) s.seenSyncContributionCache = lruwrpr.New(seenSyncContributionSize) + s.syncContributionBitsOverlapCache = lruwrpr.New(seenSyncContributionSize) s.seenExitCache = lruwrpr.New(seenExitSize) s.seenAttesterSlashingCache = make(map[uint64]bool) s.seenProposerSlashingCache = lruwrpr.New(seenProposerSlashingSize) diff --git a/beacon-chain/sync/validate_aggregate_proof.go b/beacon-chain/sync/validate_aggregate_proof.go index 8be17d36ff2c..2dba774c3707 100644 --- a/beacon-chain/sync/validate_aggregate_proof.go +++ b/beacon-chain/sync/validate_aggregate_proof.go @@ -7,6 +7,7 @@ import ( "github.com/libp2p/go-libp2p-core/peer" pubsub "github.com/libp2p/go-libp2p-pubsub" "github.com/pkg/errors" + "github.com/prysmaticlabs/go-bitfield" "github.com/prysmaticlabs/prysm/beacon-chain/core/blocks" "github.com/prysmaticlabs/prysm/beacon-chain/core/feed" "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/operation" @@ -91,16 +92,12 @@ func (s *Service) validateAggregateAndProof(ctx context.Context, pid peer.ID, ms return pubsub.ValidationIgnore, err } - has, err := s.cfg.attPool.HasAggregatedAttestation(m.Message.Aggregate) + // Verify this is the first aggregate received from the aggregator with index and slot. + seen, err := s.hasSeenAggregationBits(m.Message.Aggregate.Data, m.Message.Aggregate.AggregationBits) if err != nil { return pubsub.ValidationIgnore, err } - if has { - return pubsub.ValidationIgnore, err - } - - // Verify this is the first aggregate received from the aggregator with index and slot. - if s.hasSeenAggregatorIndexEpoch(m.Message.Aggregate.Data.Target.Epoch, m.Message.AggregatorIndex) { + if seen { return pubsub.ValidationIgnore, nil } // Check that the block being voted on isn't invalid. @@ -111,7 +108,7 @@ func (s *Service) validateAggregateAndProof(ctx context.Context, pid peer.ID, ms } // Verify aggregate attestation has not already been seen via aggregate gossip, within a block, or through the creation locally. - seen, err := s.cfg.attPool.HasAggregatedAttestation(m.Message.Aggregate) + seen, err = s.cfg.attPool.HasAggregatedAttestation(m.Message.Aggregate) if err != nil { tracing.AnnotateError(span, err) return pubsub.ValidationIgnore, err @@ -128,7 +125,9 @@ func (s *Service) validateAggregateAndProof(ctx context.Context, pid peer.ID, ms return validationRes, err } - s.setAggregatorIndexEpochSeen(m.Message.Aggregate.Data.Target.Epoch, m.Message.AggregatorIndex) + if err := s.setAggregationBits(m.Message.Aggregate.Data, m.Message.Aggregate.AggregationBits); err != nil { + return pubsub.ValidationIgnore, err + } msg.ValidatorData = m @@ -221,21 +220,61 @@ func (s *Service) validateBlockInAttestation(ctx context.Context, satt *ethpb.Si return true } -// Returns true if the node has received aggregate for the aggregator with index and target epoch. -func (s *Service) hasSeenAggregatorIndexEpoch(epoch types.Epoch, aggregatorIndex types.ValidatorIndex) bool { +// Returns true if the node has received aggregate with the same data and superset aggregated bits. +func (s *Service) hasSeenAggregationBits(data *ethpb.AttestationData, bits bitfield.Bitlist) (bool, error) { s.seenAggregatedAttestationLock.RLock() defer s.seenAggregatedAttestationLock.RUnlock() - b := append(bytesutil.Bytes32(uint64(epoch)), bytesutil.Bytes32(uint64(aggregatorIndex))...) - _, seen := s.seenAggregatedAttestationCache.Get(string(b)) - return seen + if len(bits) == 0 { + return false, nil + } + + r, err := data.HashTreeRoot() + if err != nil { + return false, err + } + v, seen := s.seenAggregatedAttestationCache.Get(string(r[:])) + if !seen { + return false, nil + } + bs, ok := v.([]bitfield.Bitlist) + if !ok { + return false, errors.New("could not convert to bitlist") + } + for _, b := range bs { + if b == nil { + return false, errors.New("nil bits") + } + contains, err := b.Contains(bits) + if err != nil { + return false, err + } + if contains { + return true, nil + } + } + return false, nil } -// Set aggregate's aggregator index target epoch as seen. -func (s *Service) setAggregatorIndexEpochSeen(epoch types.Epoch, aggregatorIndex types.ValidatorIndex) { +// Set aggregate's data and aggregated bits. +func (s *Service) setAggregationBits(data *ethpb.AttestationData, bits bitfield.Bitlist) error { s.seenAggregatedAttestationLock.Lock() defer s.seenAggregatedAttestationLock.Unlock() - b := append(bytesutil.Bytes32(uint64(epoch)), bytesutil.Bytes32(uint64(aggregatorIndex))...) - s.seenAggregatedAttestationCache.Add(string(b), true) + + r, err := data.HashTreeRoot() + if err != nil { + return err + } + v, seen := s.seenAggregatedAttestationCache.Get(string(r[:])) + if !seen { + s.seenAggregatedAttestationCache.Add(string(r[:]), []bitfield.Bitlist{bits}) + return nil + } + bs, ok := v.([]bitfield.Bitlist) + if !ok { + return errors.New("could not convert to bitlist") + } + s.seenAggregatedAttestationCache.Add(string(r[:]), append(bs, bits)) + return nil } // This validates the aggregator's index in state is within the beacon committee. diff --git a/beacon-chain/sync/validate_aggregate_proof_test.go b/beacon-chain/sync/validate_aggregate_proof_test.go index 007241bf0385..eeeb8bcebaa7 100644 --- a/beacon-chain/sync/validate_aggregate_proof_test.go +++ b/beacon-chain/sync/validate_aggregate_proof_test.go @@ -9,6 +9,7 @@ import ( pubsub "github.com/libp2p/go-libp2p-pubsub" pubsubpb "github.com/libp2p/go-libp2p-pubsub/pb" + _ "github.com/patrickmn/go-cache" "github.com/prysmaticlabs/go-bitfield" mock "github.com/prysmaticlabs/prysm/beacon-chain/blockchain/testing" "github.com/prysmaticlabs/prysm/beacon-chain/core/helpers" @@ -740,3 +741,55 @@ func TestValidateAggregateAndProof_Optimistic(t *testing.T) { valid := res == pubsub.ValidationIgnore assert.Equal(t, true, valid, "Validation should have ignored the message") } + +func TestService_setAggregationBits(t *testing.T) { + chainService := &mock.ChainService{ + Genesis: time.Now(), + ValidatorsRoot: [32]byte{'A'}, + } + s := NewService(context.Background(), WithP2P(p2ptest.NewTestP2P(t)), WithStateNotifier(chainService.StateNotifier())) + s.initCaches() + + // Empty cache + d := util.HydrateAttestationData(ðpb.AttestationData{}) + has, err := s.hasSeenAggregationBits(d, []byte{}) + require.NoError(t, err) + require.Equal(t, false, has) + + // Cache with entries but same key + require.NoError(t, s.setAggregationBits(d, []byte{5})) + has, err = s.hasSeenAggregationBits(d, []byte{5}) + require.NoError(t, err) + require.Equal(t, true, has) + has, err = s.hasSeenAggregationBits(d, []byte{6}) + require.NoError(t, err) + require.Equal(t, false, has) + has, err = s.hasSeenAggregationBits(d, []byte{7}) + require.NoError(t, err) + require.Equal(t, false, has) + + require.NoError(t, s.setAggregationBits(d, []byte{7})) + has, err = s.hasSeenAggregationBits(d, []byte{6}) + require.NoError(t, err) + require.Equal(t, true, has) + + // Cache with entries but different key + d = util.HydrateAttestationData(ðpb.AttestationData{Slot: 1}) + has, err = s.hasSeenAggregationBits(d, []byte{}) + require.NoError(t, err) + require.Equal(t, false, has) + require.NoError(t, s.setAggregationBits(d, []byte{9})) + has, err = s.hasSeenAggregationBits(d, []byte{9}) + require.NoError(t, err) + require.Equal(t, true, has) + has, err = s.hasSeenAggregationBits(d, []byte{0x0A}) + require.NoError(t, err) + require.Equal(t, false, has) + has, err = s.hasSeenAggregationBits(d, []byte{0x0B}) + require.NoError(t, err) + require.Equal(t, false, has) + require.NoError(t, s.setAggregationBits(d, []byte{0x0B})) + has, err = s.hasSeenAggregationBits(d, []byte{0x0A}) + require.NoError(t, err) + require.Equal(t, true, has) +} diff --git a/beacon-chain/sync/validate_sync_contribution_proof.go b/beacon-chain/sync/validate_sync_contribution_proof.go index b6d45c2e732b..43c7e3d85c87 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof.go +++ b/beacon-chain/sync/validate_sync_contribution_proof.go @@ -6,6 +6,7 @@ import ( "github.com/libp2p/go-libp2p-core/peer" pubsub "github.com/libp2p/go-libp2p-pubsub" + "github.com/prysmaticlabs/go-bitfield" "github.com/prysmaticlabs/prysm/beacon-chain/core/altair" "github.com/prysmaticlabs/prysm/beacon-chain/core/feed" opfeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/operation" @@ -88,7 +89,11 @@ func (s *Service) validateSyncContributionAndProof(ctx context.Context, pid peer return result, err } - s.setSyncContributionIndexSlotSeen(m.Message.Contribution.Slot, m.Message.AggregatorIndex, types.CommitteeIndex(m.Message.Contribution.SubcommitteeIndex)) + con := m.Message.Contribution + if err := s.setSyncContributionBits(con.Slot, con.BlockRoot, con.SubcommitteeIndex, con.AggregationBits); err != nil { + return pubsub.ValidationIgnore, err + } + s.setSyncContributionIndexSlotSeen(con.Slot, m.Message.AggregatorIndex, types.CommitteeIndex(con.SubcommitteeIndex)) msg.ValidatorData = m @@ -149,17 +154,18 @@ func rejectEmptyContribution(m *ethpb.SignedContributionAndProof) validationFn { func (s *Service) ignoreSeenSyncContribution(m *ethpb.SignedContributionAndProof) validationFn { return func(ctx context.Context) (pubsub.ValidationResult, error) { - seen := s.hasSeenSyncContributionIndexSlot(m.Message.Contribution.Slot, m.Message.AggregatorIndex, types.CommitteeIndex(m.Message.Contribution.SubcommitteeIndex)) - if seen { - return pubsub.ValidationIgnore, nil - } - seen, err := s.cfg.syncCommsPool.HasSyncCommitteeContribution(m.Message.Contribution) + c := m.Message.Contribution + seen, err := s.hasSeenSyncContributionBits(c.Slot, c.BlockRoot, c.SubcommitteeIndex, c.AggregationBits) if err != nil { return pubsub.ValidationIgnore, err } if seen { return pubsub.ValidationIgnore, nil } + seen = s.hasSeenSyncContributionIndexSlot(c.Slot, m.Message.AggregatorIndex, types.CommitteeIndex(c.SubcommitteeIndex)) + if seen { + return pubsub.ValidationIgnore, nil + } return pubsub.ValidationAccept, nil } } @@ -335,6 +341,54 @@ func (s *Service) setSyncContributionIndexSlotSeen(slot types.Slot, aggregatorIn s.seenSyncContributionCache.Add(string(b), true) } +// Set sync contribution's slot, root, committee index and bits. +func (s *Service) setSyncContributionBits(slot types.Slot, root []byte, subComIdx uint64, bits bitfield.Bitvector128) error { + s.syncContributionBitsOverlapLock.Lock() + defer s.syncContributionBitsOverlapLock.Unlock() + b := append(root, bytesutil.Bytes32(uint64(slot))...) + b = append(b, bytesutil.Bytes32(subComIdx)...) + v, ok := s.syncContributionBitsOverlapCache.Get(string(b)) + if !ok { + s.syncContributionBitsOverlapCache.Add(string(b), []bitfield.Bitvector128{bits}) + return nil + } + bitsList, ok := v.([]bitfield.Bitvector128) + if !ok { + return errors.New("could not covert cached value to []bitfield.Bitvector128") + } + s.syncContributionBitsOverlapCache.Add(string(b), append(bitsList, bits)) + return nil +} + +// Check sync contribution bits don't have an overlap with one's in cache. +func (s *Service) hasSeenSyncContributionBits(slot types.Slot, root []byte, subComIdx uint64, bits bitfield.Bitvector128) (bool, error) { + s.syncContributionBitsOverlapLock.RLock() + defer s.syncContributionBitsOverlapLock.RUnlock() + b := append(root, bytesutil.Bytes32(uint64(slot))...) + b = append(b, bytesutil.Bytes32(subComIdx)...) + v, ok := s.syncContributionBitsOverlapCache.Get(string(b)) + if !ok { + return false, nil + } + bitsList, ok := v.([]bitfield.Bitvector128) + if !ok { + return false, errors.New("could not covert cached value to []bitfield.Bitvector128") + } + for _, bs := range bitsList { + if bs == nil { + return false, errors.New("nil bitfield") + } + overlaps, err := bs.Overlaps(bits) + if err != nil { + return false, err + } + if overlaps { + return true, nil + } + } + return false, nil +} + // verifySyncSelectionData verifies that the provided sync contribution has a valid // selection proof. func (s *Service) verifySyncSelectionData(ctx context.Context, m *ethpb.ContributionAndProof) error { diff --git a/beacon-chain/sync/validate_sync_contribution_proof_test.go b/beacon-chain/sync/validate_sync_contribution_proof_test.go index 38a73b9230fc..ff92f43b9340 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof_test.go +++ b/beacon-chain/sync/validate_sync_contribution_proof_test.go @@ -21,7 +21,6 @@ import ( "github.com/prysmaticlabs/prysm/beacon-chain/core/transition" "github.com/prysmaticlabs/prysm/beacon-chain/db" testingDB "github.com/prysmaticlabs/prysm/beacon-chain/db/testing" - "github.com/prysmaticlabs/prysm/beacon-chain/operations/synccommittee" "github.com/prysmaticlabs/prysm/beacon-chain/p2p" "github.com/prysmaticlabs/prysm/beacon-chain/p2p/encoder" mockp2p "github.com/prysmaticlabs/prysm/beacon-chain/p2p/testing" @@ -73,7 +72,6 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), - WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -267,7 +265,6 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), - WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -312,7 +309,6 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), - WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -375,7 +371,6 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), - WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -441,7 +436,6 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), - WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -522,7 +516,6 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), - WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -689,7 +682,6 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), - WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -786,7 +778,6 @@ func TestService_ValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), - WithSyncCommsPool(synccommittee.NewPool()), ), setupSvc: func(s *Service, msg *ethpb.SignedContributionAndProof) *Service { s.cfg.stateGen = stategen.New(db) @@ -934,7 +925,6 @@ func TestValidateSyncContributionAndProof(t *testing.T) { WithChainService(chainService), WithStateNotifier(chainService.StateNotifier()), WithOperationNotifier(chainService.OperationNotifier()), - WithSyncCommsPool(synccommittee.NewPool()), ) go s.verifierRoutine() s.cfg.stateGen = stategen.New(db) @@ -1105,3 +1095,66 @@ func syncSelectionProofSigningRoot(st state.BeaconState, slot types.Slot, comIdx selectionData := ðpb.SyncAggregatorSelectionData{Slot: slot, SubcommitteeIndex: uint64(comIdx)} return signing.ComputeSigningRoot(selectionData, dom) } + +func TestService_setSyncContributionIndexSlotSeen(t *testing.T) { + chainService := &mockChain.ChainService{ + Genesis: time.Now(), + ValidatorsRoot: [32]byte{'A'}, + } + s := NewService(context.Background(), WithP2P(mockp2p.NewTestP2P(t)), WithStateNotifier(chainService.StateNotifier())) + s.initCaches() + + // Empty cache + b0 := bitfield.NewBitvector128() + b0.SetBitAt(0, true) + has, err := s.hasSeenSyncContributionBits(0, []byte{}, 0, b0) + require.NoError(t, err) + require.Equal(t, false, has) + + // Cache with entries but same key + require.NoError(t, s.setSyncContributionBits(0, []byte{}, 0, b0)) + has, err = s.hasSeenSyncContributionBits(0, []byte{}, 0, b0) + require.NoError(t, err) + require.Equal(t, true, has) + b1 := bitfield.NewBitvector128() + b1.SetBitAt(1, true) + has, err = s.hasSeenSyncContributionBits(0, []byte{}, 0, b1) + require.NoError(t, err) + require.Equal(t, false, has) + b2 := bitfield.NewBitvector128() + b2.SetBitAt(1, true) + b2.SetBitAt(2, true) + has, err = s.hasSeenSyncContributionBits(0, []byte{}, 0, b2) + require.NoError(t, err) + require.Equal(t, false, has) + b2.SetBitAt(0, true) + has, err = s.hasSeenSyncContributionBits(0, []byte{}, 0, b2) + require.NoError(t, err) + require.Equal(t, true, has) + + // Cache with entries but different key + has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 2, b0) + require.NoError(t, err) + require.Equal(t, false, has) + require.NoError(t, s.setSyncContributionBits(1, []byte{'a'}, 2, b2)) + has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 2, b0) + require.NoError(t, err) + require.Equal(t, true, has) + has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 2, b1) + require.NoError(t, err) + require.Equal(t, true, has) + has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 2, b2) + require.NoError(t, err) + require.Equal(t, true, has) + + // Check invariant with the keys + has, err = s.hasSeenSyncContributionBits(2, []byte{'a'}, 2, b0) + require.NoError(t, err) + require.Equal(t, false, has) + has, err = s.hasSeenSyncContributionBits(1, []byte{'b'}, 2, b0) + require.NoError(t, err) + require.Equal(t, false, has) + has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 3, b0) + require.NoError(t, err) + require.Equal(t, false, has) +} From 417187fcf04d52b752e5f4e9d65d75617352079f Mon Sep 17 00:00:00 2001 From: terence tsao Date: Wed, 18 May 2022 13:37:48 -0700 Subject: [PATCH 07/13] Handle mainnet/minimal better --- .../sync/validate_sync_contribution_proof.go | 34 ++++---- .../validate_sync_contribution_proof_test.go | 80 +++++++++++++++---- .../prysm/v1alpha1/sync_committee_mainnet.go | 4 + .../prysm/v1alpha1/sync_committee_minimal.go | 4 + 4 files changed, 91 insertions(+), 31 deletions(-) diff --git a/beacon-chain/sync/validate_sync_contribution_proof.go b/beacon-chain/sync/validate_sync_contribution_proof.go index 43c7e3d85c87..36c86834d0bd 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof.go +++ b/beacon-chain/sync/validate_sync_contribution_proof.go @@ -6,7 +6,6 @@ import ( "github.com/libp2p/go-libp2p-core/peer" pubsub "github.com/libp2p/go-libp2p-pubsub" - "github.com/prysmaticlabs/go-bitfield" "github.com/prysmaticlabs/prysm/beacon-chain/core/altair" "github.com/prysmaticlabs/prysm/beacon-chain/core/feed" opfeed "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/operation" @@ -90,7 +89,7 @@ func (s *Service) validateSyncContributionAndProof(ctx context.Context, pid peer } con := m.Message.Contribution - if err := s.setSyncContributionBits(con.Slot, con.BlockRoot, con.SubcommitteeIndex, con.AggregationBits); err != nil { + if err := s.setSyncContributionBits(con); err != nil { return pubsub.ValidationIgnore, err } s.setSyncContributionIndexSlotSeen(con.Slot, m.Message.AggregatorIndex, types.CommitteeIndex(con.SubcommitteeIndex)) @@ -155,7 +154,7 @@ func rejectEmptyContribution(m *ethpb.SignedContributionAndProof) validationFn { func (s *Service) ignoreSeenSyncContribution(m *ethpb.SignedContributionAndProof) validationFn { return func(ctx context.Context) (pubsub.ValidationResult, error) { c := m.Message.Contribution - seen, err := s.hasSeenSyncContributionBits(c.Slot, c.BlockRoot, c.SubcommitteeIndex, c.AggregationBits) + seen, err := s.hasSeenSyncContributionBits(c) if err != nil { return pubsub.ValidationIgnore, err } @@ -342,43 +341,44 @@ func (s *Service) setSyncContributionIndexSlotSeen(slot types.Slot, aggregatorIn } // Set sync contribution's slot, root, committee index and bits. -func (s *Service) setSyncContributionBits(slot types.Slot, root []byte, subComIdx uint64, bits bitfield.Bitvector128) error { +func (s *Service) setSyncContributionBits(c *ethpb.SyncCommitteeContribution) error { s.syncContributionBitsOverlapLock.Lock() defer s.syncContributionBitsOverlapLock.Unlock() - b := append(root, bytesutil.Bytes32(uint64(slot))...) - b = append(b, bytesutil.Bytes32(subComIdx)...) + b := append(c.BlockRoot, bytesutil.Bytes32(uint64(c.Slot))...) + b = append(b, bytesutil.Bytes32(c.SubcommitteeIndex)...) v, ok := s.syncContributionBitsOverlapCache.Get(string(b)) if !ok { - s.syncContributionBitsOverlapCache.Add(string(b), []bitfield.Bitvector128{bits}) + s.syncContributionBitsOverlapCache.Add(string(b), [][]byte{c.AggregationBits.Bytes()}) return nil } - bitsList, ok := v.([]bitfield.Bitvector128) + bitsList, ok := v.([][]byte) if !ok { - return errors.New("could not covert cached value to []bitfield.Bitvector128") + return errors.New("could not covert cached value to []bitfield.Bitvector") } - s.syncContributionBitsOverlapCache.Add(string(b), append(bitsList, bits)) + s.syncContributionBitsOverlapCache.Add(string(b), append(bitsList, c.AggregationBits.Bytes())) return nil } // Check sync contribution bits don't have an overlap with one's in cache. -func (s *Service) hasSeenSyncContributionBits(slot types.Slot, root []byte, subComIdx uint64, bits bitfield.Bitvector128) (bool, error) { +func (s *Service) hasSeenSyncContributionBits(c *ethpb.SyncCommitteeContribution) (bool, error) { s.syncContributionBitsOverlapLock.RLock() defer s.syncContributionBitsOverlapLock.RUnlock() - b := append(root, bytesutil.Bytes32(uint64(slot))...) - b = append(b, bytesutil.Bytes32(subComIdx)...) + b := append(c.BlockRoot, bytesutil.Bytes32(uint64(c.Slot))...) + b = append(b, bytesutil.Bytes32(c.SubcommitteeIndex)...) v, ok := s.syncContributionBitsOverlapCache.Get(string(b)) if !ok { return false, nil } - bitsList, ok := v.([]bitfield.Bitvector128) + bitsList, ok := v.([][]byte) if !ok { return false, errors.New("could not covert cached value to []bitfield.Bitvector128") } - for _, bs := range bitsList { - if bs == nil { + for _, b := range bitsList { + if b == nil { return false, errors.New("nil bitfield") } - overlaps, err := bs.Overlaps(bits) + bs := ethpb.ConvertSyncContributionBitVector(b) + overlaps, err := bs.Overlaps(c.AggregationBits) if err != nil { return false, err } diff --git a/beacon-chain/sync/validate_sync_contribution_proof_test.go b/beacon-chain/sync/validate_sync_contribution_proof_test.go index ff92f43b9340..15ab8ae8004d 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof_test.go +++ b/beacon-chain/sync/validate_sync_contribution_proof_test.go @@ -1107,54 +1107,106 @@ func TestService_setSyncContributionIndexSlotSeen(t *testing.T) { // Empty cache b0 := bitfield.NewBitvector128() b0.SetBitAt(0, true) - has, err := s.hasSeenSyncContributionBits(0, []byte{}, 0, b0) + has, err := s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + AggregationBits: b0, + }) require.NoError(t, err) require.Equal(t, false, has) // Cache with entries but same key - require.NoError(t, s.setSyncContributionBits(0, []byte{}, 0, b0)) - has, err = s.hasSeenSyncContributionBits(0, []byte{}, 0, b0) + require.NoError(t, s.setSyncContributionBits(ðpb.SyncCommitteeContribution{ + AggregationBits: b0, + })) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + AggregationBits: b0, + }) require.NoError(t, err) require.Equal(t, true, has) b1 := bitfield.NewBitvector128() b1.SetBitAt(1, true) - has, err = s.hasSeenSyncContributionBits(0, []byte{}, 0, b1) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + AggregationBits: b1, + }) require.NoError(t, err) require.Equal(t, false, has) b2 := bitfield.NewBitvector128() b2.SetBitAt(1, true) b2.SetBitAt(2, true) - has, err = s.hasSeenSyncContributionBits(0, []byte{}, 0, b2) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + AggregationBits: b2, + }) require.NoError(t, err) require.Equal(t, false, has) b2.SetBitAt(0, true) - has, err = s.hasSeenSyncContributionBits(0, []byte{}, 0, b2) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + AggregationBits: b2, + }) require.NoError(t, err) require.Equal(t, true, has) // Cache with entries but different key - has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 2, b0) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 2, + BlockRoot: []byte{'A'}, + AggregationBits: b0, + }) require.NoError(t, err) require.Equal(t, false, has) - require.NoError(t, s.setSyncContributionBits(1, []byte{'a'}, 2, b2)) - has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 2, b0) + require.NoError(t, s.setSyncContributionBits(ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 2, + BlockRoot: []byte{'A'}, + AggregationBits: b2, + })) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 2, + BlockRoot: []byte{'A'}, + AggregationBits: b0, + }) require.NoError(t, err) require.Equal(t, true, has) - has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 2, b1) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 2, + BlockRoot: []byte{'A'}, + AggregationBits: b1, + }) require.NoError(t, err) require.Equal(t, true, has) - has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 2, b2) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 2, + BlockRoot: []byte{'A'}, + AggregationBits: b2, + }) require.NoError(t, err) require.Equal(t, true, has) // Check invariant with the keys - has, err = s.hasSeenSyncContributionBits(2, []byte{'a'}, 2, b0) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + Slot: 2, + SubcommitteeIndex: 2, + BlockRoot: []byte{'A'}, + AggregationBits: b0, + }) require.NoError(t, err) require.Equal(t, false, has) - has, err = s.hasSeenSyncContributionBits(1, []byte{'b'}, 2, b0) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 2, + BlockRoot: []byte{'B'}, + AggregationBits: b0, + }) require.NoError(t, err) require.Equal(t, false, has) - has, err = s.hasSeenSyncContributionBits(1, []byte{'a'}, 3, b0) + has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ + Slot: 1, + SubcommitteeIndex: 3, + BlockRoot: []byte{'A'}, + AggregationBits: b0, + }) require.NoError(t, err) require.Equal(t, false, has) } diff --git a/proto/prysm/v1alpha1/sync_committee_mainnet.go b/proto/prysm/v1alpha1/sync_committee_mainnet.go index 3ef26abe381b..068b6f47b960 100644 --- a/proto/prysm/v1alpha1/sync_committee_mainnet.go +++ b/proto/prysm/v1alpha1/sync_committee_mainnet.go @@ -10,3 +10,7 @@ import ( func NewSyncCommitteeAggregationBits() bitfield.Bitvector128 { return bitfield.NewBitvector128() } + +func ConvertSyncContributionBitVector(b []byte) bitfield.Bitvector128 { + return b +} diff --git a/proto/prysm/v1alpha1/sync_committee_minimal.go b/proto/prysm/v1alpha1/sync_committee_minimal.go index 55ccebe0dcb9..2bc0fac5794e 100644 --- a/proto/prysm/v1alpha1/sync_committee_minimal.go +++ b/proto/prysm/v1alpha1/sync_committee_minimal.go @@ -10,3 +10,7 @@ import ( func NewSyncCommitteeAggregationBits() bitfield.Bitvector8 { return bitfield.NewBitvector8() } + +func ConvertSyncContributionBitVector(b []byte) bitfield.Bitvector8 { + return b +} From 1613754f08ccf6b583a926f07183f9755391fb96 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Thu, 19 May 2022 09:39:41 -0700 Subject: [PATCH 08/13] Revert att changes --- .../sync/pending_attestations_queue.go | 5 +- beacon-chain/sync/validate_aggregate_proof.go | 71 ++++--------------- .../sync/validate_aggregate_proof_test.go | 53 -------------- 3 files changed, 13 insertions(+), 116 deletions(-) diff --git a/beacon-chain/sync/pending_attestations_queue.go b/beacon-chain/sync/pending_attestations_queue.go index 314537a2a565..d3e0789a8461 100644 --- a/beacon-chain/sync/pending_attestations_queue.go +++ b/beacon-chain/sync/pending_attestations_queue.go @@ -79,10 +79,7 @@ func (s *Service) processPendingAtts(ctx context.Context) error { log.WithError(err).Debug("Could not save aggregate attestation") continue } - if err := s.setAggregationBits(att.Aggregate.Data, att.Aggregate.AggregationBits); err != nil { - log.WithError(err).Debug("Could not set") - continue - } + s.setAggregatorIndexEpochSeen(att.Aggregate.Data.Target.Epoch, att.AggregatorIndex) // Broadcasting the signed attestation again once a node is able to process it. if err := s.cfg.p2p.Broadcast(ctx, signedAtt); err != nil { diff --git a/beacon-chain/sync/validate_aggregate_proof.go b/beacon-chain/sync/validate_aggregate_proof.go index 2dba774c3707..743fc3c6ff13 100644 --- a/beacon-chain/sync/validate_aggregate_proof.go +++ b/beacon-chain/sync/validate_aggregate_proof.go @@ -7,7 +7,6 @@ import ( "github.com/libp2p/go-libp2p-core/peer" pubsub "github.com/libp2p/go-libp2p-pubsub" "github.com/pkg/errors" - "github.com/prysmaticlabs/go-bitfield" "github.com/prysmaticlabs/prysm/beacon-chain/core/blocks" "github.com/prysmaticlabs/prysm/beacon-chain/core/feed" "github.com/prysmaticlabs/prysm/beacon-chain/core/feed/operation" @@ -93,11 +92,7 @@ func (s *Service) validateAggregateAndProof(ctx context.Context, pid peer.ID, ms } // Verify this is the first aggregate received from the aggregator with index and slot. - seen, err := s.hasSeenAggregationBits(m.Message.Aggregate.Data, m.Message.Aggregate.AggregationBits) - if err != nil { - return pubsub.ValidationIgnore, err - } - if seen { + if s.hasSeenAggregatorIndexEpoch(m.Message.Aggregate.Data.Target.Epoch, m.Message.AggregatorIndex) { return pubsub.ValidationIgnore, nil } // Check that the block being voted on isn't invalid. @@ -108,7 +103,7 @@ func (s *Service) validateAggregateAndProof(ctx context.Context, pid peer.ID, ms } // Verify aggregate attestation has not already been seen via aggregate gossip, within a block, or through the creation locally. - seen, err = s.cfg.attPool.HasAggregatedAttestation(m.Message.Aggregate) + seen, err := s.cfg.attPool.HasAggregatedAttestation(m.Message.Aggregate) if err != nil { tracing.AnnotateError(span, err) return pubsub.ValidationIgnore, err @@ -125,9 +120,7 @@ func (s *Service) validateAggregateAndProof(ctx context.Context, pid peer.ID, ms return validationRes, err } - if err := s.setAggregationBits(m.Message.Aggregate.Data, m.Message.Aggregate.AggregationBits); err != nil { - return pubsub.ValidationIgnore, err - } + s.setAggregatorIndexEpochSeen(m.Message.Aggregate.Data.Target.Epoch, m.Message.AggregatorIndex) msg.ValidatorData = m @@ -220,61 +213,21 @@ func (s *Service) validateBlockInAttestation(ctx context.Context, satt *ethpb.Si return true } -// Returns true if the node has received aggregate with the same data and superset aggregated bits. -func (s *Service) hasSeenAggregationBits(data *ethpb.AttestationData, bits bitfield.Bitlist) (bool, error) { +// Returns true if the node has received aggregate for the aggregator with index and target epoch. +func (s *Service) hasSeenAggregatorIndexEpoch(epoch types.Epoch, aggregatorIndex types.ValidatorIndex) bool { s.seenAggregatedAttestationLock.RLock() defer s.seenAggregatedAttestationLock.RUnlock() - if len(bits) == 0 { - return false, nil - } - - r, err := data.HashTreeRoot() - if err != nil { - return false, err - } - v, seen := s.seenAggregatedAttestationCache.Get(string(r[:])) - if !seen { - return false, nil - } - bs, ok := v.([]bitfield.Bitlist) - if !ok { - return false, errors.New("could not convert to bitlist") - } - for _, b := range bs { - if b == nil { - return false, errors.New("nil bits") - } - contains, err := b.Contains(bits) - if err != nil { - return false, err - } - if contains { - return true, nil - } - } - return false, nil + b := append(bytesutil.Bytes32(uint64(epoch)), bytesutil.Bytes32(uint64(aggregatorIndex))...) + _, seen := s.seenAggregatedAttestationCache.Get(string(b)) + return seen } -// Set aggregate's data and aggregated bits. -func (s *Service) setAggregationBits(data *ethpb.AttestationData, bits bitfield.Bitlist) error { +// Set aggregate's aggregator index target epoch as seen. +func (s *Service) setAggregatorIndexEpochSeen(epoch types.Epoch, aggregatorIndex types.ValidatorIndex) { s.seenAggregatedAttestationLock.Lock() defer s.seenAggregatedAttestationLock.Unlock() - - r, err := data.HashTreeRoot() - if err != nil { - return err - } - v, seen := s.seenAggregatedAttestationCache.Get(string(r[:])) - if !seen { - s.seenAggregatedAttestationCache.Add(string(r[:]), []bitfield.Bitlist{bits}) - return nil - } - bs, ok := v.([]bitfield.Bitlist) - if !ok { - return errors.New("could not convert to bitlist") - } - s.seenAggregatedAttestationCache.Add(string(r[:]), append(bs, bits)) - return nil + b := append(bytesutil.Bytes32(uint64(epoch)), bytesutil.Bytes32(uint64(aggregatorIndex))...) + s.seenAggregatedAttestationCache.Add(string(b), true) } // This validates the aggregator's index in state is within the beacon committee. diff --git a/beacon-chain/sync/validate_aggregate_proof_test.go b/beacon-chain/sync/validate_aggregate_proof_test.go index eeeb8bcebaa7..007241bf0385 100644 --- a/beacon-chain/sync/validate_aggregate_proof_test.go +++ b/beacon-chain/sync/validate_aggregate_proof_test.go @@ -9,7 +9,6 @@ import ( pubsub "github.com/libp2p/go-libp2p-pubsub" pubsubpb "github.com/libp2p/go-libp2p-pubsub/pb" - _ "github.com/patrickmn/go-cache" "github.com/prysmaticlabs/go-bitfield" mock "github.com/prysmaticlabs/prysm/beacon-chain/blockchain/testing" "github.com/prysmaticlabs/prysm/beacon-chain/core/helpers" @@ -741,55 +740,3 @@ func TestValidateAggregateAndProof_Optimistic(t *testing.T) { valid := res == pubsub.ValidationIgnore assert.Equal(t, true, valid, "Validation should have ignored the message") } - -func TestService_setAggregationBits(t *testing.T) { - chainService := &mock.ChainService{ - Genesis: time.Now(), - ValidatorsRoot: [32]byte{'A'}, - } - s := NewService(context.Background(), WithP2P(p2ptest.NewTestP2P(t)), WithStateNotifier(chainService.StateNotifier())) - s.initCaches() - - // Empty cache - d := util.HydrateAttestationData(ðpb.AttestationData{}) - has, err := s.hasSeenAggregationBits(d, []byte{}) - require.NoError(t, err) - require.Equal(t, false, has) - - // Cache with entries but same key - require.NoError(t, s.setAggregationBits(d, []byte{5})) - has, err = s.hasSeenAggregationBits(d, []byte{5}) - require.NoError(t, err) - require.Equal(t, true, has) - has, err = s.hasSeenAggregationBits(d, []byte{6}) - require.NoError(t, err) - require.Equal(t, false, has) - has, err = s.hasSeenAggregationBits(d, []byte{7}) - require.NoError(t, err) - require.Equal(t, false, has) - - require.NoError(t, s.setAggregationBits(d, []byte{7})) - has, err = s.hasSeenAggregationBits(d, []byte{6}) - require.NoError(t, err) - require.Equal(t, true, has) - - // Cache with entries but different key - d = util.HydrateAttestationData(ðpb.AttestationData{Slot: 1}) - has, err = s.hasSeenAggregationBits(d, []byte{}) - require.NoError(t, err) - require.Equal(t, false, has) - require.NoError(t, s.setAggregationBits(d, []byte{9})) - has, err = s.hasSeenAggregationBits(d, []byte{9}) - require.NoError(t, err) - require.Equal(t, true, has) - has, err = s.hasSeenAggregationBits(d, []byte{0x0A}) - require.NoError(t, err) - require.Equal(t, false, has) - has, err = s.hasSeenAggregationBits(d, []byte{0x0B}) - require.NoError(t, err) - require.Equal(t, false, has) - require.NoError(t, s.setAggregationBits(d, []byte{0x0B})) - has, err = s.hasSeenAggregationBits(d, []byte{0x0A}) - require.NoError(t, err) - require.Equal(t, true, has) -} From ee2c910f6f6307b1e64e2c5141091c678fba9719 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Fri, 20 May 2022 06:53:39 -0700 Subject: [PATCH 09/13] Check overlaps before set --- .../sync/validate_sync_contribution_proof.go | 30 ++++++++++--------- .../validate_sync_contribution_proof_test.go | 3 ++ 2 files changed, 19 insertions(+), 14 deletions(-) diff --git a/beacon-chain/sync/validate_sync_contribution_proof.go b/beacon-chain/sync/validate_sync_contribution_proof.go index 36c86834d0bd..4d3005ecbe63 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof.go +++ b/beacon-chain/sync/validate_sync_contribution_proof.go @@ -169,16 +169,6 @@ func (s *Service) ignoreSeenSyncContribution(m *ethpb.SignedContributionAndProof } } -func (s *Service) ignoreCached(m *ethpb.SignedContributionAndProof) validationFn { - return func(ctx context.Context) (pubsub.ValidationResult, error) { - seen := s.hasSeenSyncContributionIndexSlot(m.Message.Contribution.Slot, m.Message.AggregatorIndex, types.CommitteeIndex(m.Message.Contribution.SubcommitteeIndex)) - if seen { - return pubsub.ValidationIgnore, nil - } - return pubsub.ValidationAccept, nil - } -} - func rejectInvalidAggregator(m *ethpb.SignedContributionAndProof) validationFn { return func(ctx context.Context) (pubsub.ValidationResult, error) { // The `contribution_and_proof.selection_proof` selects the validator as an aggregator for the slot. @@ -355,6 +345,13 @@ func (s *Service) setSyncContributionBits(c *ethpb.SyncCommitteeContribution) er if !ok { return errors.New("could not covert cached value to []bitfield.Bitvector") } + has, err := BitListOverlaps(bitsList, c.AggregationBits) + if err != nil { + return err + } + if has { + return nil + } s.syncContributionBitsOverlapCache.Add(string(b), append(bitsList, c.AggregationBits.Bytes())) return nil } @@ -373,12 +370,17 @@ func (s *Service) hasSeenSyncContributionBits(c *ethpb.SyncCommitteeContribution if !ok { return false, errors.New("could not covert cached value to []bitfield.Bitvector128") } - for _, b := range bitsList { - if b == nil { + return BitListOverlaps(bitsList, c.AggregationBits.Bytes()) +} + +// BitListOverlaps returns true if there's an overlap between two bitlists. +func BitListOverlaps(bitLists [][]byte, b []byte) (bool, error) { + for _, bitList := range bitLists { + if bitList == nil { return false, errors.New("nil bitfield") } - bs := ethpb.ConvertSyncContributionBitVector(b) - overlaps, err := bs.Overlaps(c.AggregationBits) + bl := ethpb.ConvertSyncContributionBitVector(bitList) + overlaps, err := bl.Overlaps(ethpb.ConvertSyncContributionBitVector(b)) if err != nil { return false, err } diff --git a/beacon-chain/sync/validate_sync_contribution_proof_test.go b/beacon-chain/sync/validate_sync_contribution_proof_test.go index 15ab8ae8004d..25e3b26eb6a9 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof_test.go +++ b/beacon-chain/sync/validate_sync_contribution_proof_test.go @@ -1144,6 +1144,9 @@ func TestService_setSyncContributionIndexSlotSeen(t *testing.T) { require.NoError(t, err) require.Equal(t, true, has) + // Make sure set doesn't contain existing overlaps + require.Equal(t, 1, s.syncContributionBitsOverlapCache.Len()) + // Cache with entries but different key has, err = s.hasSeenSyncContributionBits(ðpb.SyncCommitteeContribution{ Slot: 1, From d18079ac022eaa04804dcfd7db0a53e2348625f2 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Tue, 24 May 2022 00:14:54 -0700 Subject: [PATCH 10/13] Feedbacks --- .../sync/validate_sync_contribution_proof.go | 16 +++++++++------- proto/prysm/v1alpha1/sync_committee_mainnet.go | 2 +- proto/prysm/v1alpha1/sync_committee_minimal.go | 2 +- 3 files changed, 11 insertions(+), 9 deletions(-) diff --git a/beacon-chain/sync/validate_sync_contribution_proof.go b/beacon-chain/sync/validate_sync_contribution_proof.go index 4d3005ecbe63..3ed450af95db 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof.go +++ b/beacon-chain/sync/validate_sync_contribution_proof.go @@ -334,7 +334,9 @@ func (s *Service) setSyncContributionIndexSlotSeen(slot types.Slot, aggregatorIn func (s *Service) setSyncContributionBits(c *ethpb.SyncCommitteeContribution) error { s.syncContributionBitsOverlapLock.Lock() defer s.syncContributionBitsOverlapLock.Unlock() - b := append(c.BlockRoot, bytesutil.Bytes32(uint64(c.Slot))...) + var root []byte + copy(root, c.BlockRoot) // Copying due to how pb unmarshalling is carried out, prevent mutation. + b := append(root, bytesutil.Bytes32(uint64(c.Slot))...) b = append(b, bytesutil.Bytes32(c.SubcommitteeIndex)...) v, ok := s.syncContributionBitsOverlapCache.Get(string(b)) if !ok { @@ -345,7 +347,7 @@ func (s *Service) setSyncContributionBits(c *ethpb.SyncCommitteeContribution) er if !ok { return errors.New("could not covert cached value to []bitfield.Bitvector") } - has, err := BitListOverlaps(bitsList, c.AggregationBits) + has, err := bitListOverlaps(bitsList, c.AggregationBits) if err != nil { return err } @@ -370,17 +372,17 @@ func (s *Service) hasSeenSyncContributionBits(c *ethpb.SyncCommitteeContribution if !ok { return false, errors.New("could not covert cached value to []bitfield.Bitvector128") } - return BitListOverlaps(bitsList, c.AggregationBits.Bytes()) + return bitListOverlaps(bitsList, c.AggregationBits.Bytes()) } -// BitListOverlaps returns true if there's an overlap between two bitlists. -func BitListOverlaps(bitLists [][]byte, b []byte) (bool, error) { +// bitListOverlaps returns true if there's an overlap between two bitlists. +func bitListOverlaps(bitLists [][]byte, b []byte) (bool, error) { for _, bitList := range bitLists { if bitList == nil { return false, errors.New("nil bitfield") } - bl := ethpb.ConvertSyncContributionBitVector(bitList) - overlaps, err := bl.Overlaps(ethpb.ConvertSyncContributionBitVector(b)) + bl := ethpb.ConvertToSyncContributionBitVector(bitList) + overlaps, err := bl.Overlaps(ethpb.ConvertToSyncContributionBitVector(b)) if err != nil { return false, err } diff --git a/proto/prysm/v1alpha1/sync_committee_mainnet.go b/proto/prysm/v1alpha1/sync_committee_mainnet.go index 068b6f47b960..56d21af515ba 100644 --- a/proto/prysm/v1alpha1/sync_committee_mainnet.go +++ b/proto/prysm/v1alpha1/sync_committee_mainnet.go @@ -11,6 +11,6 @@ func NewSyncCommitteeAggregationBits() bitfield.Bitvector128 { return bitfield.NewBitvector128() } -func ConvertSyncContributionBitVector(b []byte) bitfield.Bitvector128 { +func ConvertToSyncContributionBitVector(b []byte) bitfield.Bitvector128 { return b } diff --git a/proto/prysm/v1alpha1/sync_committee_minimal.go b/proto/prysm/v1alpha1/sync_committee_minimal.go index 2bc0fac5794e..4d7536900d49 100644 --- a/proto/prysm/v1alpha1/sync_committee_minimal.go +++ b/proto/prysm/v1alpha1/sync_committee_minimal.go @@ -11,6 +11,6 @@ func NewSyncCommitteeAggregationBits() bitfield.Bitvector8 { return bitfield.NewBitvector8() } -func ConvertSyncContributionBitVector(b []byte) bitfield.Bitvector8 { +func ConvertToSyncContributionBitVector(b []byte) bitfield.Bitvector8 { return b } From 592db2b9cf6cbb1a44507a38ca50ab28baa60fb9 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Tue, 24 May 2022 15:56:30 -0700 Subject: [PATCH 11/13] Fixed a copy bug --- beacon-chain/sync/validate_sync_contribution_proof.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/beacon-chain/sync/validate_sync_contribution_proof.go b/beacon-chain/sync/validate_sync_contribution_proof.go index 3ed450af95db..def1d551accc 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof.go +++ b/beacon-chain/sync/validate_sync_contribution_proof.go @@ -334,7 +334,7 @@ func (s *Service) setSyncContributionIndexSlotSeen(slot types.Slot, aggregatorIn func (s *Service) setSyncContributionBits(c *ethpb.SyncCommitteeContribution) error { s.syncContributionBitsOverlapLock.Lock() defer s.syncContributionBitsOverlapLock.Unlock() - var root []byte + root := make([]byte, len(c.BlockRoot)) copy(root, c.BlockRoot) // Copying due to how pb unmarshalling is carried out, prevent mutation. b := append(root, bytesutil.Bytes32(uint64(c.Slot))...) b = append(b, bytesutil.Bytes32(c.SubcommitteeIndex)...) From 78a4a5b675c0d7e044d81eed90b07ad521505b19 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Tue, 24 May 2022 16:28:14 -0700 Subject: [PATCH 12/13] Fixed the same copy bug in committee indices cache --- .../sync/validate_beacon_attestation.go | 2 +- .../sync/validate_beacon_attestation_test.go | 27 +++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/beacon-chain/sync/validate_beacon_attestation.go b/beacon-chain/sync/validate_beacon_attestation.go index af9f658a58ba..356caf7e2fe9 100644 --- a/beacon-chain/sync/validate_beacon_attestation.go +++ b/beacon-chain/sync/validate_beacon_attestation.go @@ -252,7 +252,7 @@ func (s *Service) setSeenCommitteeIndicesSlot(slot types.Slot, committeeID types s.seenUnAggregatedAttestationLock.Lock() defer s.seenUnAggregatedAttestationLock.Unlock() b := append(bytesutil.Bytes32(uint64(slot)), bytesutil.Bytes32(uint64(committeeID))...) - var bits []byte + bits := make([]byte, len(aggregateBits)) copy(bits, aggregateBits) b = append(b, bits...) s.seenUnAggregatedAttestationCache.Add(string(b), true) diff --git a/beacon-chain/sync/validate_beacon_attestation_test.go b/beacon-chain/sync/validate_beacon_attestation_test.go index eb16eac9b53f..a261fe2ceb36 100644 --- a/beacon-chain/sync/validate_beacon_attestation_test.go +++ b/beacon-chain/sync/validate_beacon_attestation_test.go @@ -339,3 +339,30 @@ func TestServiceValidateCommitteeIndexBeaconAttestation_Optimistic(t *testing.T) valid := res == pubsub.ValidationIgnore assert.Equal(t, true, valid, "Should have ignore this message") } + +func TestService_setSeenCommitteeIndicesSlot(t *testing.T) { + chainService := &mockChain.ChainService{ + Genesis: time.Now(), + ValidatorsRoot: [32]byte{'A'}, + } + s := NewService(context.Background(), WithP2P(p2ptest.NewTestP2P(t)), WithStateNotifier(chainService.StateNotifier())) + s.initCaches() + + // Empty cache + b0 := []byte{9} // 1001 + require.Equal(t, false, s.hasSeenCommitteeIndicesSlot(0, 0, b0)) + + // Cache some entries but same key + s.setSeenCommitteeIndicesSlot(0, 0, b0) + require.Equal(t, true, s.hasSeenCommitteeIndicesSlot(0, 0, b0)) + b1 := []byte{14} // 1110 + s.setSeenCommitteeIndicesSlot(0, 0, b1) + require.Equal(t, true, s.hasSeenCommitteeIndicesSlot(0, 0, b0)) + require.Equal(t, true, s.hasSeenCommitteeIndicesSlot(0, 0, b1)) + + // Cache some entries with diff keys + s.setSeenCommitteeIndicesSlot(1, 2, b1) + require.Equal(t, false, s.hasSeenCommitteeIndicesSlot(1, 0, b1)) + require.Equal(t, false, s.hasSeenCommitteeIndicesSlot(0, 2, b1)) + require.Equal(t, true, s.hasSeenCommitteeIndicesSlot(1, 2, b1)) +} From ba756345cd23eb7581b40c23d4f4744ce9900a88 Mon Sep 17 00:00:00 2001 From: terence tsao Date: Tue, 24 May 2022 16:41:26 -0700 Subject: [PATCH 13/13] Use SafeCopyBytes --- beacon-chain/sync/validate_beacon_attestation.go | 4 +--- beacon-chain/sync/validate_sync_contribution_proof.go | 5 ++--- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/beacon-chain/sync/validate_beacon_attestation.go b/beacon-chain/sync/validate_beacon_attestation.go index 356caf7e2fe9..4c98c2c50c85 100644 --- a/beacon-chain/sync/validate_beacon_attestation.go +++ b/beacon-chain/sync/validate_beacon_attestation.go @@ -252,9 +252,7 @@ func (s *Service) setSeenCommitteeIndicesSlot(slot types.Slot, committeeID types s.seenUnAggregatedAttestationLock.Lock() defer s.seenUnAggregatedAttestationLock.Unlock() b := append(bytesutil.Bytes32(uint64(slot)), bytesutil.Bytes32(uint64(committeeID))...) - bits := make([]byte, len(aggregateBits)) - copy(bits, aggregateBits) - b = append(b, bits...) + b = append(b, bytesutil.SafeCopyBytes(aggregateBits)...) s.seenUnAggregatedAttestationCache.Add(string(b), true) } diff --git a/beacon-chain/sync/validate_sync_contribution_proof.go b/beacon-chain/sync/validate_sync_contribution_proof.go index def1d551accc..eda2f75406fe 100644 --- a/beacon-chain/sync/validate_sync_contribution_proof.go +++ b/beacon-chain/sync/validate_sync_contribution_proof.go @@ -334,9 +334,8 @@ func (s *Service) setSyncContributionIndexSlotSeen(slot types.Slot, aggregatorIn func (s *Service) setSyncContributionBits(c *ethpb.SyncCommitteeContribution) error { s.syncContributionBitsOverlapLock.Lock() defer s.syncContributionBitsOverlapLock.Unlock() - root := make([]byte, len(c.BlockRoot)) - copy(root, c.BlockRoot) // Copying due to how pb unmarshalling is carried out, prevent mutation. - b := append(root, bytesutil.Bytes32(uint64(c.Slot))...) + // Copying due to how pb unmarshalling is carried out, prevent mutation. + b := append(bytesutil.SafeCopyBytes(c.BlockRoot), bytesutil.Bytes32(uint64(c.Slot))...) b = append(b, bytesutil.Bytes32(c.SubcommitteeIndex)...) v, ok := s.syncContributionBitsOverlapCache.Get(string(b)) if !ok {