Fix aggregator with unaggregated attestation (#6137)

* Use state util to get block root
* Merge branch 'master' of github.com:prysmaticlabs/prysm
* Merge branch 'master' of github.com:prysmaticlabs/prysm
* Merge branch 'master' of github.com:prysmaticlabs/prysm
* Merge branch 'master' of github.com:prysmaticlabs/prysm
* Merge branch 'master' of github.com:prysmaticlabs/prysm
* Merge branch 'master' of github.com:prysmaticlabs/prysm
* Add UnaggregatedAttestationsBySlotIndex
* Tests
* New TestSubmitAggregateAndProof_UnaggregateOk test
* Merge branch 'master' of github.com:prysmaticlabs/prysm into fix-aggregator-broadcast

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

@@ -52,6 +52,22 @@ func (p *AttCaches) UnaggregatedAttestations() []*ethpb.Attestation {
 	return atts
 }
 
+// UnaggregatedAttestationsBySlotIndex returns the unaggregated attestations in cache,
+// filtered by committee index and slot.
+func (p *AttCaches) UnaggregatedAttestationsBySlotIndex(slot uint64, committeeIndex uint64) []*ethpb.Attestation {
+	atts := make([]*ethpb.Attestation, 0)
+
+	p.unAggregateAttLock.RLock()
+	defer p.unAggregateAttLock.RUnlock()
+	for _, a := range p.unAggregatedAtt {
+		if slot == a.Data.Slot && committeeIndex == a.Data.CommitteeIndex {
+			atts = append(atts, a)
+		}
+	}
+
+	return atts
+}
+
 // DeleteUnaggregatedAttestation deletes the unaggregated attestations in cache.
 func (p *AttCaches) DeleteUnaggregatedAttestation(att *ethpb.Attestation) error {
 	if att == nil {

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

@@ -51,3 +51,31 @@ func TestKV_Unaggregated_CanDelete(t *testing.T) {
 		t.Error("Did not receive correct aggregated atts")
 	}
 }
+
+func TestKV_Unaggregated_CanGetByCommitteeAndSlot(t *testing.T) {
+	cache := NewAttCaches()
+
+	att1 := &ethpb.Attestation{Data: &ethpb.AttestationData{Slot: 1, CommitteeIndex: 1}, AggregationBits: bitfield.Bitlist{0b101}}
+	att2 := &ethpb.Attestation{Data: &ethpb.AttestationData{Slot: 1, CommitteeIndex: 2}, AggregationBits: bitfield.Bitlist{0b110}}
+	att3 := &ethpb.Attestation{Data: &ethpb.AttestationData{Slot: 2, CommitteeIndex: 1}, AggregationBits: bitfield.Bitlist{0b110}}
+	atts := []*ethpb.Attestation{att1, att2, att3}
+
+	for _, att := range atts {
+		if err := cache.SaveUnaggregatedAttestation(att); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	returned := cache.UnaggregatedAttestationsBySlotIndex(1, 1)
+	if !reflect.DeepEqual([]*ethpb.Attestation{att1}, returned) {
+		t.Error("Did not receive correct aggregated atts")
+	}
+	returned = cache.UnaggregatedAttestationsBySlotIndex(1, 2)
+	if !reflect.DeepEqual([]*ethpb.Attestation{att2}, returned) {
+		t.Error("Did not receive correct aggregated atts")
+	}
+	returned = cache.UnaggregatedAttestationsBySlotIndex(2, 1)
+	if !reflect.DeepEqual([]*ethpb.Attestation{att3}, returned) {
+		t.Error("Did not receive correct aggregated atts")
+	}
+}

beacon-chain/operations/attestations/pool.go

@@ -23,6 +23,7 @@ type Pool interface {
 	SaveUnaggregatedAttestation(att *ethpb.Attestation) error
 	SaveUnaggregatedAttestations(atts []*ethpb.Attestation) error
 	UnaggregatedAttestations() []*ethpb.Attestation
+	UnaggregatedAttestationsBySlotIndex(slot uint64, committeeIndex uint64) []*ethpb.Attestation
 	DeleteUnaggregatedAttestation(att *ethpb.Attestation) error
 	UnaggregatedAttestationCount() int
 	// For attestations that were included in the block.

beacon-chain/rpc/validator/aggregator.go

@@ -65,7 +65,10 @@ func (as *Server) SubmitAggregateSelectionProof(ctx context.Context, req *ethpb.
 
 	// Filter out the best aggregated attestation (ie. the one with the most aggregated bits).
 	if len(aggregatedAtts) == 0 {
-		return nil, status.Error(codes.Internal, "No aggregated attestation in beacon node")
+		aggregatedAtts = as.AttPool.UnaggregatedAttestationsBySlotIndex(req.Slot, req.CommitteeIndex)
+		if len(aggregatedAtts) == 0 {
+			return nil, status.Errorf(codes.Internal, "Could not find attestation for slot and committee in pool")
+		}
 	}
 	best := aggregatedAtts[0]
 	for _, aggregatedAtt := range aggregatedAtts[1:] {

beacon-chain/rpc/validator/aggregator_test.go

@@ -98,11 +98,55 @@ func TestSubmitAggregateAndProof_IsAggregatorAndNoAtts(t *testing.T) {
 	pubKey := v.PublicKey
 	req := &ethpb.AggregateSelectionRequest{CommitteeIndex: 1, SlotSignature: sig.Marshal(), PublicKey: pubKey}
 
-	if _, err := server.SubmitAggregateSelectionProof(ctx, req); err == nil || !strings.Contains(err.Error(), "No aggregated attestation in beacon node") {
+	if _, err := server.SubmitAggregateSelectionProof(ctx, req); err == nil || !strings.Contains(err.Error(), "Could not find attestation for slot and committee in pool") {
 		t.Error("Did not get wanted error")
 	}
 }
 
+func TestSubmitAggregateAndProof_UnaggregateOk(t *testing.T) {
+	params.SetupTestConfigCleanup(t)
+	c := params.MinimalSpecConfig()
+	c.TargetAggregatorsPerCommittee = 16
+	params.OverrideBeaconConfig(c)
+
+	db := dbutil.SetupDB(t)
+	ctx := context.Background()
+
+	beaconState, privKeys := testutil.DeterministicGenesisState(t, 32)
+	att0, err := generateUnaggregatedAtt(beaconState, 0, privKeys)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = beaconState.SetSlot(beaconState.Slot() + params.BeaconConfig().MinAttestationInclusionDelay)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	aggregatorServer := &Server{
+		HeadFetcher: &mock.ChainService{State: beaconState},
+		SyncChecker: &mockSync.Sync{IsSyncing: false},
+		BeaconDB:    db,
+		AttPool:     attestations.NewPool(),
+		P2P:         &mockp2p.MockBroadcaster{},
+	}
+
+	priv := bls.RandKey()
+	sig := priv.Sign([]byte{'B'})
+	v, err := beaconState.ValidatorAtIndex(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	pubKey := v.PublicKey
+	req := &ethpb.AggregateSelectionRequest{CommitteeIndex: 1, SlotSignature: sig.Marshal(), PublicKey: pubKey}
+
+	if err := aggregatorServer.AttPool.SaveUnaggregatedAttestation(att0); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := aggregatorServer.SubmitAggregateSelectionProof(ctx, req); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestSubmitAggregateAndProof_AggregateOk(t *testing.T) {
 	params.SetupTestConfigCleanup(t)
 	c := params.MinimalSpecConfig()
@@ -196,7 +240,7 @@ func TestSubmitAggregateAndProof_AggregateNotOk(t *testing.T) {
 	pubKey := v.PublicKey
 	req := &ethpb.AggregateSelectionRequest{CommitteeIndex: 1, SlotSignature: sig.Marshal(), PublicKey: pubKey}
 
-	if _, err := aggregatorServer.SubmitAggregateSelectionProof(ctx, req); !strings.Contains(err.Error(), "No aggregated attestation in beacon node") {
+	if _, err := aggregatorServer.SubmitAggregateSelectionProof(ctx, req); !strings.Contains(err.Error(), "Could not find attestation for slot and committee in pool") {
 		t.Error("Did not get wanted error")
 	}
 
@@ -245,3 +289,42 @@ func generateAtt(state *beaconstate.BeaconState, index uint64, privKeys []*bls.S
 
 	return att, nil
 }
+
+func generateUnaggregatedAtt(state *beaconstate.BeaconState, index uint64, privKeys []*bls.SecretKey) (*ethpb.Attestation, error) {
+	aggBits := bitfield.NewBitlist(4)
+	aggBits.SetBitAt(index, true)
+	att := &ethpb.Attestation{
+		Data: &ethpb.AttestationData{
+			CommitteeIndex: 1,
+			Source:         &ethpb.Checkpoint{Epoch: 0, Root: params.BeaconConfig().ZeroHash[:]},
+			Target:         &ethpb.Checkpoint{Epoch: 0},
+		},
+		AggregationBits: aggBits,
+	}
+	committee, err := helpers.BeaconCommitteeFromState(state, att.Data.Slot, att.Data.CommitteeIndex)
+	if err != nil {
+		return nil, err
+	}
+	attestingIndices := attestationutil.AttestingIndices(att.AggregationBits, committee)
+	domain, err := helpers.Domain(state.Fork(), 0, params.BeaconConfig().DomainBeaconAttester, params.BeaconConfig().ZeroHash[:])
+	if err != nil {
+		return nil, err
+	}
+
+	sigs := make([]*bls.Signature, len(attestingIndices))
+	zeroSig := [96]byte{}
+	att.Signature = zeroSig[:]
+
+	for i, indice := range attestingIndices {
+		hashTreeRoot, err := helpers.ComputeSigningRoot(att.Data, domain)
+		if err != nil {
+			return nil, err
+		}
+		sig := privKeys[indice].Sign(hashTreeRoot[:])
+		sigs[i] = sig
+	}
+
+	att.Signature = bls.AggregateSignatures(sigs).Marshal()[:]
+
+	return att, nil
+}