Check before saving attestation to pool (#6912)

* Pool: check before save atts
* Pool: test
* Pool: update metrics
* Merge branch 'master' into check-before-save
* Merge refs/heads/master into check-before-save

beacon-chain/operations/attestations/kv/aggregated.go

@@ -74,6 +74,14 @@ func (p *AttCaches) SaveAggregatedAttestation(att *ethpb.Attestation) error {
 	if !helpers.IsAggregated(att) {
 		return errors.New("attestation is not aggregated")
 	}
+	has, err := p.HasAggregatedAttestation(att)
+	if err != nil {
+		return err
+	}
+	if has {
+		return nil
+	}
+
 	r, err := hashFn(att.Data)
 	if err != nil {
 		return errors.Wrap(err, "could not tree hash attestation")

beacon-chain/operations/attestations/kv/aggregated_test.go

@@ -41,7 +41,7 @@ func TestKV_Aggregated_AggregateUnaggregatedAttestations(t *testing.T) {
 	}
 }
 
-func TestKV_Aggregated_SaveUnaggregatedAttestation(t *testing.T) {
+func TestKV_Aggregated_SaveAggregatedAttestation(t *testing.T) {
 	tests := []struct {
 		name          string
 		att           *ethpb.Attestation
@@ -101,7 +101,47 @@ func TestKV_Aggregated_SaveUnaggregatedAttestation(t *testing.T) {
 				return
 			}
 			if len(cache.aggregatedAtt) != tt.count {
-				t.Errorf("Wrong attestation count, want: %d, got: %d", tt.count, len(cache.unAggregatedAtt))
+				t.Errorf("Wrong attestation count, want: %d, got: %d", tt.count, len(cache.aggregatedAtt))
+			}
+			if cache.AggregatedAttestationCount() != tt.count {
+				t.Errorf("Wrong attestation count, want: %d, got: %d", tt.count, cache.AggregatedAttestationCount())
+			}
+		})
+	}
+}
+
+func TestKV_Aggregated_SaveAggregatedAttestations(t *testing.T) {
+	tests := []struct {
+		name          string
+		atts          []*ethpb.Attestation
+		count         int
+		wantErrString string
+	}{
+		{
+			name: "no duplicates",
+			atts: []*ethpb.Attestation{
+				{Data: &ethpb.AttestationData{Slot: 1},
+					AggregationBits: bitfield.Bitlist{0b1101}},
+				{Data: &ethpb.AttestationData{Slot: 1},
+					AggregationBits: bitfield.Bitlist{0b1101}},
+			},
+			count: 1,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cache := NewAttCaches()
+			if len(cache.aggregatedAtt) != 0 {
+				t.Errorf("Invalid start pool, atts: %d", len(cache.unAggregatedAtt))
+			}
+			err := cache.SaveAggregatedAttestations(tt.atts)
+			if tt.wantErrString == "" && err != nil {
+				t.Error(err)
+				return
+			}
+			if len(cache.aggregatedAtt) != tt.count {
+				t.Errorf("Wrong attestation count, want: %d, got: %d", tt.count, len(cache.aggregatedAtt))
 			}
 			if cache.AggregatedAttestationCount() != tt.count {
 				t.Errorf("Wrong attestation count, want: %d, got: %d", tt.count, cache.AggregatedAttestationCount())

beacon-chain/operations/attestations/prune_expired.go

@@ -14,6 +14,7 @@ func (s *Service) pruneAttsPool() {
 		select {
 		case <-ticker.C:
 			s.pruneExpiredAtts()
+			s.updateMetrics()
 		case <-s.ctx.Done():
 			log.Debug("Context closed, exiting routine")
 			ticker.Stop()