Beacon-api: save atts to pool (#12345)

Co-authored-by: Raul Jordan <raul@prysmaticlabs.com>
Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

beacon-chain/rpc/eth/beacon/pool.go

@@ -83,23 +83,17 @@ func (bs *Server) SubmitAttestations(ctx context.Context, req *ethpbv1.SubmitAtt
 
 		// Broadcast the unaggregated attestation on a feed to notify other services in the beacon node
 		// of a received unaggregated attestation.
-		bs.OperationNotifier.OperationFeed().Send(&feed.Event{
-			Type: operation.UnaggregatedAttReceived,
-			Data: &operation.UnAggregatedAttReceivedData{
-				Attestation: att,
-			},
-		})
+		// Note we can't send for aggregated att because we don't have selection proof.
+		if !corehelpers.IsAggregated(att) {
+			bs.OperationNotifier.OperationFeed().Send(&feed.Event{
+				Type: operation.UnaggregatedAttReceived,
+				Data: &operation.UnAggregatedAttReceivedData{
+					Attestation: att,
+				},
+			})
+		}
 
 		validAttestations = append(validAttestations, att)
-
-		go func() {
-			ctx = trace.NewContext(context.Background(), trace.FromContext(ctx))
-			attCopy := ethpbalpha.CopyAttestation(att)
-			if err := bs.AttestationsPool.SaveUnaggregatedAttestation(attCopy); err != nil {
-				log.WithError(err).Error("Could not handle attestation in operations service")
-				return
-			}
-		}()
 	}
 
 	broadcastFailed := false
@@ -115,6 +109,16 @@ func (bs *Server) SubmitAttestations(ctx context.Context, req *ethpbv1.SubmitAtt
 		if err := bs.Broadcaster.BroadcastAttestation(ctx, subnet, att); err != nil {
 			broadcastFailed = true
 		}
+
+		if corehelpers.IsAggregated(att) {
+			if err := bs.AttestationsPool.SaveAggregatedAttestation(att); err != nil {
+				log.WithError(err).Error("could not save aggregated att")
+			}
+		} else {
+			if err := bs.AttestationsPool.SaveUnaggregatedAttestation(att); err != nil {
+				log.WithError(err).Error("could not save unaggregated att")
+			}
+		}
 	}
 	if broadcastFailed {
 		return nil, status.Errorf(

beacon-chain/rpc/eth/beacon/pool_test.go

@@ -1007,6 +1007,8 @@ func TestServer_SubmitAttestations_Ok(t *testing.T) {
 	for _, r := range [][32]byte{actualAtt1, actualAtt2} {
 		assert.Equal(t, true, reflect.DeepEqual(expectedAtt1, r) || reflect.DeepEqual(expectedAtt2, r))
 	}
+
+	require.Equal(t, 2, s.AttestationsPool.UnaggregatedAttestationCount())
 }
 
 func TestServer_SubmitAttestations_ValidAttestationSubmitted(t *testing.T) {
@@ -1107,6 +1109,8 @@ func TestServer_SubmitAttestations_ValidAttestationSubmitted(t *testing.T) {
 	broadcastRoot, err := broadcaster.BroadcastAttestations[0].HashTreeRoot()
 	require.NoError(t, err)
 	require.DeepEqual(t, expectedAtt, broadcastRoot)
+
+	require.Equal(t, 1, s.AttestationsPool.UnaggregatedAttestationCount())
 }
 
 func TestServer_SubmitAttestations_InvalidAttestationGRPCHeader(t *testing.T) {