Don't submit duplicate aggregated SignedContributionAndProof messages (#15571)

validator/client/sync_committee.go

@@ -130,6 +130,7 @@ func (v *validator) SubmitSignedContributionAndProof(ctx context.Context, slot p
 
 	v.waitToSlotTwoThirds(ctx, slot)
 
+	coveredSubnets := make(map[uint64]bool)
 	for i, comIdx := range indexRes.Indices {
 		isAggregator, err := altair.IsSyncCommitteeAggregator(selectionProofs[i])
 		if err != nil {
@@ -141,6 +142,10 @@ func (v *validator) SubmitSignedContributionAndProof(ctx context.Context, slot p
 		}
 		subCommitteeSize := params.BeaconConfig().SyncCommitteeSize / params.BeaconConfig().SyncCommitteeSubnetCount
 		subnet := uint64(comIdx) / subCommitteeSize
+		if coveredSubnets[subnet] {
+			// Don't submit a message for the same subnet multiple times
+			continue
+		}
 		contribution, err := v.validatorClient.SyncCommitteeContribution(ctx, &ethpb.SyncCommitteeContributionRequest{
 			Slot:      slot,
 			PublicKey: pubKey[:],
@@ -178,6 +183,8 @@ func (v *validator) SubmitSignedContributionAndProof(ctx context.Context, slot p
 			return
 		}
 
+		coveredSubnets[subnet] = true
+
 		contributionSlot := contributionAndProof.Contribution.Slot
 		slotTime, err := slots.StartTime(v.genesisTime, contributionSlot)
 		if err != nil {

validator/client/sync_committee_test.go

@@ -512,3 +512,87 @@ func TestSubmitSignedContributionAndProof_Ok(t *testing.T) {
 		})
 	}
 }
+
+func TestSubmitSignedContributionAndProof_OncePerPubkeyAndSubcommittee(t *testing.T) {
+	// Hardcode secret key in order to have a valid aggregator signature.
+	rawKey, err := hex.DecodeString("659e875e1b062c03f2f2a57332974d475b97df6cfc581d322e79642d39aca8fd")
+	assert.NoError(t, err)
+	validatorKey, err := bls.SecretKeyFromBytes(rawKey)
+	assert.NoError(t, err)
+
+	for _, isSlashingProtectionMinimal := range [...]bool{false, true} {
+		t.Run(fmt.Sprintf("SlashingProtectionMinimal:%v", isSlashingProtectionMinimal), func(t *testing.T) {
+			validator, m, validatorKey, finish := setupWithKey(t, validatorKey, isSlashingProtectionMinimal)
+			validatorIndex := primitives.ValidatorIndex(7)
+			committee := []primitives.ValidatorIndex{0, 3, 4, 2, validatorIndex, 6, 8, 9, 10}
+			validator.duties = &ethpb.ValidatorDutiesContainer{CurrentEpochDuties: []*ethpb.ValidatorDuty{
+				{
+					PublicKey:       validatorKey.PublicKey().Marshal(),
+					CommitteeLength: uint64(len(committee)),
+					ValidatorIndex:  validatorIndex,
+				},
+			}}
+			defer finish()
+
+			// Sync committee aggregator is selected twice in the sync committee
+			aggregatorCommitteeIndices := []primitives.CommitteeIndex{1, 2}
+			var pubKey [fieldparams.BLSPubkeyLength]byte
+			copy(pubKey[:], validatorKey.PublicKey().Marshal())
+			m.validatorClient.EXPECT().SyncSubcommitteeIndex(
+				gomock.Any(), // ctx
+				&ethpb.SyncSubcommitteeIndexRequest{
+					Slot:      1,
+					PublicKey: pubKey[:],
+				},
+			).Return(&ethpb.SyncSubcommitteeIndexResponse{Indices: aggregatorCommitteeIndices}, nil)
+
+			m.validatorClient.EXPECT().
+				DomainData(gomock.Any(), // ctx
+					gomock.Any()). // epoch
+				Times(2).
+				Return(&ethpb.DomainResponse{
+					SignatureDomain: make([]byte, 32),
+				}, nil)
+
+			aggBits := bitfield.NewBitvector128()
+			aggBits.SetBitAt(0, true)
+			m.validatorClient.EXPECT().SyncCommitteeContribution(
+				gomock.Any(), // ctx
+				&ethpb.SyncCommitteeContributionRequest{
+					Slot:      1,
+					PublicKey: pubKey[:],
+					SubnetId:  0,
+				},
+			).Return(&ethpb.SyncCommitteeContribution{
+				BlockRoot:       make([]byte, fieldparams.RootLength),
+				Signature:       make([]byte, 96),
+				AggregationBits: aggBits,
+			}, nil)
+
+			m.validatorClient.EXPECT().
+				DomainData(gomock.Any(), // ctx
+					gomock.Any()). // epoch
+				Return(&ethpb.DomainResponse{
+					SignatureDomain: make([]byte, 32),
+				}, nil)
+
+			m.validatorClient.EXPECT().SubmitSignedContributionAndProof(
+				gomock.Any(), // ctx
+				gomock.AssignableToTypeOf(&ethpb.SignedContributionAndProof{
+					Message: &ethpb.ContributionAndProof{
+						AggregatorIndex: 7,
+						Contribution: &ethpb.SyncCommitteeContribution{
+							BlockRoot:         make([]byte, 32),
+							Signature:         make([]byte, 96),
+							AggregationBits:   bitfield.NewBitvector128(),
+							Slot:              1,
+							SubcommitteeIndex: 1,
+						},
+					},
+				}),
+			).Return(&emptypb.Empty{}, nil)
+
+			validator.SubmitSignedContributionAndProof(t.Context(), 1, pubKey)
+		})
+	}
+}