MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-01-26 22:23:15 -05:00
Code Issues Packages Projects Releases Wiki Activity

cleanup existing part.N's before renamePart() (#20466)

Browse Source 
this is a safety-net to avoid any unexpected parts to show up.
This commit is contained in:
Harshavardhana 2024-09-24 04:26:41 -07:00 committed by GitHub
parent 0c53d86017
commit f6f0807c86
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 15 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
cmd/erasure-multipart.go
View File

@ -526,6 +526,14 @@ func (er erasureObjects) NewMultipartUpload(ctx context.Context, bucket, object
// renamePart - renames multipart part to its relevant location under uploadID.
func (er erasureObjects) renamePart(ctx context.Context, disks []StorageAPI, srcBucket, srcEntry, dstBucket, dstEntry string, optsMeta []byte, writeQuorum int) ([]StorageAPI, error) {
paths := []string{
dstEntry,
dstEntry + ".meta",
}
// cleanup existing paths first across all drives.
er.cleanupMultipartPath(ctx, paths...)
g := errgroup.WithNErrs(len(disks))
// Rename file on all underlying storage disks.
@ -542,11 +550,6 @@ func (er erasureObjects) renamePart(ctx context.Context, disks []StorageAPI, src
// Wait for all renames to finish.
errs := g.Wait()
paths := []string{
dstEntry,
dstEntry + ".meta",
}
err := reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
if err != nil {
er.cleanupMultipartPath(ctx, paths...)

36
cmd/xl-storage.go
View File

@ -2921,7 +2921,8 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
return res, nil
}
// RenamePart - rename part path to destination path atomically.
// RenamePart - rename part path to destination path atomically, this is meant to be used
// only with multipart API
func (s *xlStorage) RenamePart(ctx context.Context, srcVolume, srcPath, dstVolume, dstPath string, meta []byte) (err error) {
srcVolumeDir, err := s.getVolDir(srcVolume)
if err != nil {
@ -2952,46 +2953,23 @@ func (s *xlStorage) RenamePart(ctx context.Context, srcVolume, srcPath, dstVolum
return err
}
}
srcIsDir := HasSuffix(srcPath, SlashSeparator)
dstIsDir := HasSuffix(dstPath, SlashSeparator)
// Either src and dst have to be directories or files, else return error.
if !(srcIsDir && dstIsDir || !srcIsDir && !dstIsDir) {
// either source or destination is a directory return error.
if srcIsDir || dstIsDir {
return errFileAccessDenied
}
srcFilePath := pathutil.Join(srcVolumeDir, srcPath)
if err = checkPathLength(srcFilePath); err != nil {
return err
}
dstFilePath := pathutil.Join(dstVolumeDir, dstPath)
if err = checkPathLength(dstFilePath); err != nil {
return err
}
if srcIsDir {
// If source is a directory, we expect the destination to be non-existent but we
// we still need to allow overwriting an empty directory since it represents
// an object empty directory.
dirInfo, err := Lstat(dstFilePath)
if isSysErrIO(err) {
return errFaultyDisk
}
if err != nil {
if !osIsNotExist(err) {
return err
}
} else {
if !dirInfo.IsDir() {
return errFileAccessDenied
}
if err = Remove(dstFilePath); err != nil {
if isSysErrNotEmpty(err) || isSysErrNotDir(err) {
return errFileAccessDenied
} else if isSysErrIO(err) {
return errFaultyDisk
}
return err
}
}
}
if err = renameAll(srcFilePath, dstFilePath, dstVolumeDir); err != nil {
if isSysErrNotEmpty(err) || isSysErrNotDir(err) {