[#623] Fix using copy numbers during multipart

Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>

api/data/tree.go

@@ -109,7 +109,6 @@ type MultipartInfo struct {
 	Owner         user.ID
 	Created       time.Time
 	Meta          map[string]string
-	CopiesNumbers []uint32
 	Finished      bool
 	CreationEpoch uint64
 }

api/handler/multipart_upload.go

@@ -152,12 +152,6 @@ func (h *handler) CreateMultipartUploadHandler(w http.ResponseWriter, r *http.Re
 		p.Header[api.ContentLanguage] = contentLanguage
 	}
 
-	p.CopiesNumbers, err = h.pickCopiesNumbers(p.Header, reqInfo.Namespace, bktInfo.LocationConstraint)
-	if err != nil {
-		h.logAndSendError(ctx, w, "invalid copies number", reqInfo, err, additional...)
-		return
-	}
-
 	if err = h.obj.CreateMultipartUpload(ctx, p); err != nil {
 		h.logAndSendError(ctx, w, "could create multipart upload", reqInfo, err, additional...)
 		return
@@ -229,6 +223,12 @@ func (h *handler) UploadPartHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	p.CopiesNumbers, err = h.pickCopiesNumbers(parseMetadata(r), reqInfo.Namespace, bktInfo.LocationConstraint)
+	if err != nil {
+		h.logAndSendError(ctx, w, "invalid copies number", reqInfo, err, additional...)
+		return
+	}
+
 	hash, err := h.obj.UploadPart(ctx, p)
 	if err != nil {
 		h.logAndSendError(ctx, w, "could not upload a part", reqInfo, err, additional...)
@@ -354,6 +354,12 @@ func (h *handler) UploadPartCopy(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	p.CopiesNumbers, err = h.pickCopiesNumbers(parseMetadata(r), reqInfo.Namespace, bktInfo.LocationConstraint)
+	if err != nil {
+		h.logAndSendError(ctx, w, "invalid copies number", reqInfo, err, additional...)
+		return
+	}
+
 	info, err := h.obj.UploadPartCopy(ctx, p)
 	if err != nil {
 		h.logAndSendError(ctx, w, "could not upload part copy", reqInfo, err, additional...)
@@ -416,6 +422,12 @@ func (h *handler) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.
 		Parts: reqBody.Parts,
 	}
 
+	c.CopiesNumbers, err = h.pickCopiesNumbers(parseMetadata(r), reqInfo.Namespace, bktInfo.LocationConstraint)
+	if err != nil {
+		h.logAndSendError(ctx, w, "invalid copies number", reqInfo, err, additional...)
+		return
+	}
+
 	// Start complete multipart upload which may take some time to fetch object
 	// and re-upload it part by part.
 	objInfo, err := h.completeMultipartUpload(r, c, bktInfo)

api/handler/multipart_upload_test.go

@@ -81,6 +81,26 @@ func TestDeleteMultipartAllParts(t *testing.T) {
 	require.Empty(t, hc.tp.Objects())
 }
 
+func TestMultipartCopiesNumber(t *testing.T) {
+	hc := prepareHandlerContext(t)
+
+	bktName, objName := "bucket", "object"
+
+	createTestBucket(hc, bktName)
+
+	copies := []uint32{2, 0}
+
+	hc.config.copiesNumbers = map[string][]uint32{"default": copies}
+
+	multipartInfo := createMultipartUpload(hc, bktName, objName, nil)
+	uploadPart(hc, bktName, objName, multipartInfo.UploadID, 1, layer.UploadMinSize)
+
+	objs := hc.tp.Objects()
+	require.Len(t, objs, 1)
+
+	require.EqualValues(t, copies, hc.tp.CopiesNumbers(addrFromObject(objs[0]).EncodeToString()))
+}
+
 func TestSpecialMultipartName(t *testing.T) {
 	hc := prepareHandlerContextWithMinCache(t)
 
@@ -792,3 +812,14 @@ func listPartsBase(hc *handlerContext, bktName, objName string, encrypted bool,
 
 	return listPartsResponse
 }
+
+func addrFromObject(obj *object.Object) oid.Address {
+	var addr oid.Address
+	cnrID, _ := obj.ContainerID()
+	objID, _ := obj.ID()
+
+	addr.SetContainer(cnrID)
+	addr.SetObject(objID)
+
+	return addr
+}

api/layer/frostfs_mock.go

@@ -76,6 +76,7 @@ var _ frostfs.FrostFS = (*TestFrostFS)(nil)
 
 type TestFrostFS struct {
 	objects           map[string]*object.Object
+	copiesNumbers     map[string][]uint32
 	objectErrors      map[string]error
 	objectPutErrors   map[string]error
 	containers        map[string]*container.Container
@@ -88,6 +89,7 @@ type TestFrostFS struct {
 func NewTestFrostFS(key *keys.PrivateKey) *TestFrostFS {
 	return &TestFrostFS{
 		objects:         make(map[string]*object.Object),
+		copiesNumbers:   make(map[string][]uint32),
 		objectErrors:    make(map[string]error),
 		objectPutErrors: make(map[string]error),
 		containers:      make(map[string]*container.Container),
@@ -126,6 +128,10 @@ func (t *TestFrostFS) Objects() []*object.Object {
 	return res
 }
 
+func (t *TestFrostFS) CopiesNumbers(addr string) []uint32 {
+	return t.copiesNumbers[addr]
+}
+
 func (t *TestFrostFS) ObjectExists(objID oid.ID) bool {
 	for _, obj := range t.objects {
 		if id, _ := obj.ID(); id.Equals(objID) {
@@ -346,6 +352,8 @@ func (t *TestFrostFS) CreateObject(ctx context.Context, prm frostfs.PrmObjectCre
 
 	addr := newAddress(cnrID, objID)
 	t.objects[addr.EncodeToString()] = obj
+	t.copiesNumbers[addr.EncodeToString()] = prm.CopiesNumber
+
 	return &frostfs.CreateObjectResult{
 		ObjectID:      objID,
 		CreationEpoch: t.currentEpoch - 1,

api/layer/multipart_upload.go

@@ -58,10 +58,9 @@ type (
 	}
 
 	CreateMultipartParams struct {
-		Info          *UploadInfoParams
-		Header        map[string]string
-		Data          *UploadData
-		CopiesNumbers []uint32
+		Info   *UploadInfoParams
+		Header map[string]string
+		Data   *UploadData
 	}
 
 	UploadData struct {
@@ -75,6 +74,7 @@ type (
 		Reader            io.Reader
 		ContentMD5        string
 		ContentSHA256Hash string
+		CopiesNumbers     []uint32
 	}
 
 	UploadCopyParams struct {
@@ -85,11 +85,13 @@ type (
 		SrcEncryption encryption.Params
 		PartNumber    int
 		Range         *RangeParams
+		CopiesNumbers []uint32
 	}
 
 	CompleteMultipartParams struct {
-		Info  *UploadInfoParams
-		Parts []*CompletedPart
+		Info          *UploadInfoParams
+		Parts         []*CompletedPart
+		CopiesNumbers []uint32
 	}
 
 	CompletedPart struct {
@@ -165,7 +167,6 @@ func (n *Layer) CreateMultipartUpload(ctx context.Context, p *CreateMultipartPar
 		Owner:         n.gateOwner,
 		Created:       TimeNow(ctx),
 		Meta:          make(map[string]string, metaSize),
-		CopiesNumbers: p.CopiesNumbers,
 		CreationEpoch: networkInfo.CurrentEpoch(),
 	}
 
@@ -222,7 +223,7 @@ func (n *Layer) uploadPart(ctx context.Context, multipartInfo *data.MultipartInf
 		Attributes:   make([][2]string, 2),
 		Payload:      p.Reader,
 		CreationTime: TimeNow(ctx),
-		CopiesNumber: multipartInfo.CopiesNumbers,
+		CopiesNumber: p.CopiesNumbers,
 	}
 
 	decSize := p.Size
@@ -372,10 +373,11 @@ func (n *Layer) UploadPartCopy(ctx context.Context, p *UploadCopyParams) (*data.
 	}
 
 	params := &UploadPartParams{
-		Info:       p.Info,
-		PartNumber: p.PartNumber,
-		Size:       size,
-		Reader:     objPayload,
+		Info:          p.Info,
+		PartNumber:    p.PartNumber,
+		Size:          size,
+		Reader:        objPayload,
+		CopiesNumbers: p.CopiesNumbers,
 	}
 
 	return n.uploadPart(ctx, multipartInfo, params)
@@ -472,7 +474,7 @@ func (n *Layer) CompleteMultipartUpload(ctx context.Context, p *CompleteMultipar
 		Header:          initMetadata,
 		Size:            &multipartObjetSize,
 		Encryption:      p.Info.Encryption,
-		CopiesNumbers:   multipartInfo.CopiesNumbers,
+		CopiesNumbers:   p.CopiesNumbers,
 		CompleteMD5Hash: hex.EncodeToString(md5Hash.Sum(nil)) + "-" + strconv.Itoa(len(p.Parts)),
 	})
 	if err != nil {