Ignore subset aggregates

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)

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))...)
-	var bits []byte
-	copy(bits, aggregateBits)
-	b = append(b, bits...)
+	b = append(b, bytesutil.SafeCopyBytes(aggregateBits)...)
 	s.seenUnAggregatedAttestationCache.Add(string(b), true)
 }
 

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))
+}

beacon-chain/sync/validate_sync_contribution_proof.go

@@ -88,7 +88,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); err != nil {
+		return pubsub.ValidationIgnore, err
+	}
+	s.setSyncContributionIndexSlotSeen(con.Slot, m.Message.AggregatorIndex, types.CommitteeIndex(con.SubcommitteeIndex))
 
 	msg.ValidatorData = m
 
@@ -149,7 +153,15 @@ 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))
+		c := m.Message.Contribution
+		seen, err := s.hasSeenSyncContributionBits(c)
+		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
 		}
@@ -318,6 +330,68 @@ 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(c *ethpb.SyncCommitteeContribution) error {
+	s.syncContributionBitsOverlapLock.Lock()
+	defer s.syncContributionBitsOverlapLock.Unlock()
+	// 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 {
+		s.syncContributionBitsOverlapCache.Add(string(b), [][]byte{c.AggregationBits.Bytes()})
+		return nil
+	}
+	bitsList, ok := v.([][]byte)
+	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
+}
+
+// Check sync contribution bits don't have an overlap with one's in cache.
+func (s *Service) hasSeenSyncContributionBits(c *ethpb.SyncCommitteeContribution) (bool, error) {
+	s.syncContributionBitsOverlapLock.RLock()
+	defer s.syncContributionBitsOverlapLock.RUnlock()
+	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.([][]byte)
+	if !ok {
+		return false, errors.New("could not covert cached value to []bitfield.Bitvector128")
+	}
+	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")
+		}
+		bl := ethpb.ConvertToSyncContributionBitVector(bitList)
+		overlaps, err := bl.Overlaps(ethpb.ConvertToSyncContributionBitVector(b))
+		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 {

beacon-chain/sync/validate_sync_contribution_proof_test.go

@@ -1095,3 +1095,121 @@ func syncSelectionProofSigningRoot(st state.BeaconState, slot types.Slot, comIdx
 	selectionData := &ethpb.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(&ethpb.SyncCommitteeContribution{
+		AggregationBits: b0,
+	})
+	require.NoError(t, err)
+	require.Equal(t, false, has)
+
+	// Cache with entries but same key
+	require.NoError(t, s.setSyncContributionBits(&ethpb.SyncCommitteeContribution{
+		AggregationBits: b0,
+	}))
+	has, err = s.hasSeenSyncContributionBits(&ethpb.SyncCommitteeContribution{
+		AggregationBits: b0,
+	})
+	require.NoError(t, err)
+	require.Equal(t, true, has)
+	b1 := bitfield.NewBitvector128()
+	b1.SetBitAt(1, true)
+	has, err = s.hasSeenSyncContributionBits(&ethpb.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(&ethpb.SyncCommitteeContribution{
+		AggregationBits: b2,
+	})
+	require.NoError(t, err)
+	require.Equal(t, false, has)
+	b2.SetBitAt(0, true)
+	has, err = s.hasSeenSyncContributionBits(&ethpb.SyncCommitteeContribution{
+		AggregationBits: b2,
+	})
+	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(&ethpb.SyncCommitteeContribution{
+		Slot:              1,
+		SubcommitteeIndex: 2,
+		BlockRoot:         []byte{'A'},
+		AggregationBits:   b0,
+	})
+	require.NoError(t, err)
+	require.Equal(t, false, has)
+	require.NoError(t, s.setSyncContributionBits(&ethpb.SyncCommitteeContribution{
+		Slot:              1,
+		SubcommitteeIndex: 2,
+		BlockRoot:         []byte{'A'},
+		AggregationBits:   b2,
+	}))
+	has, err = s.hasSeenSyncContributionBits(&ethpb.SyncCommitteeContribution{
+		Slot:              1,
+		SubcommitteeIndex: 2,
+		BlockRoot:         []byte{'A'},
+		AggregationBits:   b0,
+	})
+	require.NoError(t, err)
+	require.Equal(t, true, has)
+	has, err = s.hasSeenSyncContributionBits(&ethpb.SyncCommitteeContribution{
+		Slot:              1,
+		SubcommitteeIndex: 2,
+		BlockRoot:         []byte{'A'},
+		AggregationBits:   b1,
+	})
+	require.NoError(t, err)
+	require.Equal(t, true, has)
+	has, err = s.hasSeenSyncContributionBits(&ethpb.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(&ethpb.SyncCommitteeContribution{
+		Slot:              2,
+		SubcommitteeIndex: 2,
+		BlockRoot:         []byte{'A'},
+		AggregationBits:   b0,
+	})
+	require.NoError(t, err)
+	require.Equal(t, false, has)
+	has, err = s.hasSeenSyncContributionBits(&ethpb.SyncCommitteeContribution{
+		Slot:              1,
+		SubcommitteeIndex: 2,
+		BlockRoot:         []byte{'B'},
+		AggregationBits:   b0,
+	})
+	require.NoError(t, err)
+	require.Equal(t, false, has)
+	has, err = s.hasSeenSyncContributionBits(&ethpb.SyncCommitteeContribution{
+		Slot:              1,
+		SubcommitteeIndex: 3,
+		BlockRoot:         []byte{'A'},
+		AggregationBits:   b0,
+	})
+	require.NoError(t, err)
+	require.Equal(t, false, has)
+}

proto/prysm/v1alpha1/sync_committee_mainnet.go

@@ -10,3 +10,7 @@ import (
 func NewSyncCommitteeAggregationBits() bitfield.Bitvector128 {
 	return bitfield.NewBitvector128()
 }
+
+func ConvertToSyncContributionBitVector(b []byte) bitfield.Bitvector128 {
+	return b
+}

proto/prysm/v1alpha1/sync_committee_minimal.go

@@ -10,3 +10,7 @@ import (
 func NewSyncCommitteeAggregationBits() bitfield.Bitvector8 {
 	return bitfield.NewBitvector8()
 }
+
+func ConvertToSyncContributionBitVector(b []byte) bitfield.Bitvector8 {
+	return b
+}