Do not save duplicated indices (#4118)

* Added a duplication test

* Refactor

* Updated test

* Do not save dups for indices bucket

beacon-chain/db/kv/attestations_test.go

@@ -265,6 +265,74 @@ func TestStore_Attestations_FiltersCorrectly(t *testing.T) {
 	}
 }
 
+func TestStore_DuplicatedAttestations_FiltersCorrectly(t *testing.T) {
+	db := setupDB(t)
+	defer teardownDB(t, db)
+	someRoot := [32]byte{1, 2, 3}
+	att := &ethpb.Attestation{
+		Data: &ethpb.AttestationData{
+			BeaconBlockRoot: someRoot[:],
+			Source: &ethpb.Checkpoint{
+				Root:  someRoot[:],
+				Epoch: 5,
+			},
+			Target: &ethpb.Checkpoint{
+				Root:  someRoot[:],
+				Epoch: 7,
+			},
+		},
+		AggregationBits: bitfield.Bitlist{0b11},
+	}
+	atts := []*ethpb.Attestation{att, att, att}
+	ctx := context.Background()
+	if err := db.SaveAttestations(ctx, atts); err != nil {
+		t.Fatal(err)
+	}
+
+	retrievedAtts, err := db.Attestations(ctx, filters.NewFilter().
+		SetHeadBlockRoot(someRoot[:]))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(retrievedAtts) != 1 {
+		t.Errorf("Expected %d attestations, received %d", 1, len(retrievedAtts))
+	}
+
+	att1 := proto.Clone(att).(*ethpb.Attestation)
+	att1.Data.Source.Epoch = 6
+	atts = []*ethpb.Attestation{att, att, att, att1, att1, att1}
+	if err := db.SaveAttestations(ctx, atts); err != nil {
+		t.Fatal(err)
+	}
+
+	retrievedAtts, err = db.Attestations(ctx, filters.NewFilter().
+		SetHeadBlockRoot(someRoot[:]))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(retrievedAtts) != 2 {
+		t.Errorf("Expected %d attestations, received %d", 1, len(retrievedAtts))
+	}
+
+	retrievedAtts, err = db.Attestations(ctx, filters.NewFilter().
+		SetHeadBlockRoot(someRoot[:]).SetSourceEpoch(5))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(retrievedAtts) != 1 {
+		t.Errorf("Expected %d attestations, received %d", 1, len(retrievedAtts))
+	}
+
+	retrievedAtts, err = db.Attestations(ctx, filters.NewFilter().
+		SetHeadBlockRoot(someRoot[:]).SetSourceEpoch(6))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(retrievedAtts) != 1 {
+		t.Errorf("Expected %d attestations, received %d", 1, len(retrievedAtts))
+	}
+}
+
 func TestStore_Attestations_BitfieldLogic(t *testing.T) {
 	commonData := &ethpb.AttestationData{Slot: 10}
 

beacon-chain/db/kv/utils.go

@@ -39,6 +39,12 @@ func updateValueForIndices(indicesByBucket map[string][]byte, root []byte, tx *b
 				return err
 			}
 		} else {
+			// Do not save duplication in indices bucket
+			for i := 0; i < len(valuesAtIndex); i += 32 {
+				if bytes.Equal(valuesAtIndex[i:i+32], root) {
+					return nil
+				}
+			}
 			if err := bkt.Put(idx, append(valuesAtIndex, root...)); err != nil {
 				return err
 			}

beacon-chain/p2p/options.go

@@ -24,7 +24,7 @@ func buildOptions(cfg *Config, ip net.IP, priKey *ecdsa.PrivateKey) []libp2p.Opt
 		libp2p.EnableRelay(),
 		libp2p.ListenAddrs(listen),
 		whitelistSubnet(cfg.WhitelistCIDR),
-		libp2p.ConnectionManager(connmgr.NewConnManager(int(cfg.MaxPeers), int(cfg.MaxPeers), 1 * time.Second)),
+		libp2p.ConnectionManager(connmgr.NewConnManager(int(cfg.MaxPeers), int(cfg.MaxPeers), 1*time.Second)),
 	}
 	if cfg.EnableUPnP {
 		options = append(options, libp2p.NATPortMap()) //Allow to use UPnP