Save unaggregated att in beacon_aggregate_and_proof subscriber (#6109)

* Use state util to get block root
* Merge branch 'master' of github.com:prysmaticlabs/prysm
* Merge branch 'master' of github.com:prysmaticlabs/prysm
* Also save unaggregated att
* Merge refs/heads/master into fix-save-agg
* Handle error
* Merge branch 'fix-save-agg' of github.com:prysmaticlabs/prysm into fix-save-agg
* Test
* Merge refs/heads/master into fix-save-agg

beacon-chain/sync/subscriber_beacon_aggregate_proof.go

@@ -9,6 +9,7 @@ import (
 	ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/feed"
 	"github.com/prysmaticlabs/prysm/beacon-chain/core/feed/operation"
+	"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
 )
 
 // beaconAggregateProofSubscriber forwards the incoming validated aggregated attestation and proof to the
@@ -33,5 +34,10 @@ func (r *Service) beaconAggregateProofSubscriber(ctx context.Context, msg proto.
 		},
 	})
 
+	// An unaggregated attestation can make it here. It’s valid, the aggregator it just itself, although it means poor performance for the subnet.
+	if !helpers.IsAggregated(a.Message.Aggregate) {
+		return r.attPool.SaveUnaggregatedAttestation(a.Message.Aggregate)
+	}
+
 	return r.attPool.SaveAggregatedAttestation(a.Message.Aggregate)
 }

beacon-chain/sync/subscriber_beacon_aggregate_proof_test.go

@@ -12,7 +12,7 @@ import (
 	"github.com/prysmaticlabs/prysm/beacon-chain/operations/attestations"
 )
 
-func TestBeaconAggregateProofSubscriber_CanSave(t *testing.T) {
+func TestBeaconAggregateProofSubscriber_CanSaveAggregatedAttestation(t *testing.T) {
 	c, err := lru.New(10)
 	if err != nil {
 		t.Fatal(err)
@@ -32,3 +32,24 @@ func TestBeaconAggregateProofSubscriber_CanSave(t *testing.T) {
 		t.Error("Did not save aggregated attestation")
 	}
 }
+
+func TestBeaconAggregateProofSubscriber_CanSaveUnaggregatedAttestation(t *testing.T) {
+	c, err := lru.New(10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	r := &Service{
+		attPool:              attestations.NewPool(),
+		seenAttestationCache: c,
+		attestationNotifier:  (&mock.ChainService{}).OperationNotifier(),
+	}
+
+	a := &ethpb.SignedAggregateAttestationAndProof{Message: &ethpb.AggregateAttestationAndProof{Aggregate: &ethpb.Attestation{Data: &ethpb.AttestationData{Target: &ethpb.Checkpoint{}}, AggregationBits: bitfield.Bitlist{0x03}}, AggregatorIndex: 100}}
+	if err := r.beaconAggregateProofSubscriber(context.Background(), a); err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(r.attPool.UnaggregatedAttestations(), []*ethpb.Attestation{a.Message.Aggregate}) {
+		t.Error("Did not save unaggregated attestation")
+	}
+}