Ignore subset aggregates

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

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.

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

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 {