diff --git a/backend/posix/posix.go b/backend/posix/posix.go index a2a39b9e..8d7378f7 100644 --- a/backend/posix/posix.go +++ b/backend/posix/posix.go @@ -2475,58 +2475,65 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C Key: &object, }, "", nil } - // Directory is gone: the concurrent call already completed and cleaned up. - // A directory at the object path is the object of the key with a - // trailing slash, not the completed upload. - if fi, statErr := os.Stat(p.ObjectPath(bucket, object)); statErr == nil && !fi.IsDir() { - etag := multipartClaimToken - if p.dataIntegrityEtag { - etagBytes, etagErr := p.meta.RetrieveAttribute(nil, bucket, object, etagkey) - if etagErr != nil { - return res, "", fmt.Errorf("get object etag: %w", etagErr) - } - etag = string(etagBytes) - } - return s3response.CompleteMultipartUploadResult{ - Bucket: &bucket, - ETag: &etag, - Key: &object, - }, "", nil + // Directory is gone: a completed upload is linked before its + // in-progress directory is removed, so it is the object at this key + // carrying this upload ID. No object, or an object written any other + // way (a PutObject or another upload), means there is no such upload. + // Multipart uploads are never created for keys with a trailing slash, + // and a directory at the object path is the object of such a key. + if strings.HasSuffix(object, "/") { + return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) + } + fi, err := os.Stat(p.ObjectPath(bucket, object)) + if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) { + return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) + } + if isErrNameTooLong(err) { + return res, "", s3err.GetKeyTooLongErr(int64(len(object)), 1024) + } + if err != nil { + return res, "", fmt.Errorf("stat object: %w", err) + } + if fi.IsDir() { + return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) + } + mpMetaBytes, err := p.meta.RetrieveAttribute(nil, bucket, object, mpMetaKey) + if errors.Is(err, meta.ErrNoSuchKey) || errors.Is(err, fs.ErrNotExist) { + return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) + } + if err != nil { + return res, "", fmt.Errorf("get object multipart metadata: %w", err) + } + mpMeta, err := backend.UnmarshalMpUploadMetadata(mpMetaBytes, false) + if err != nil { + return res, "", fmt.Errorf("parse object multipart metadata: %w", err) + } + if mpMeta.UploadID != uploadID { + return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) } - // Last resort: the object stat above may have lost a race with the - // concurrent call's link step. Check the mp-metadata xattr, as this - // multipart upload may have been finalized and the final object has been created - // before or by the racing request - if mpMetaBytes, statErr := p.meta.RetrieveAttribute(nil, bucket, object, mpMetaKey); statErr == nil { - mpMeta, err := backend.UnmarshalMpUploadMetadata(mpMetaBytes, false) - if err != nil { - return res, "", fmt.Errorf("parse object multipart metadata: %w", err) - } - - // The object may have been overwritten by a newer upload or - // it's the result of a completely different multipart upload; only - // treat it as our completion if the upload IDs match. - if mpMeta.UploadID != uploadID { - return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) - } - - etag := multipartClaimToken - if p.dataIntegrityEtag { - etagBytes, etagErr := p.meta.RetrieveAttribute(nil, bucket, object, etagkey) - if etagErr != nil { - return res, "", fmt.Errorf("get object etag: %w", etagErr) - } - etag = string(etagBytes) - } - return s3response.CompleteMultipartUploadResult{ - Bucket: &bucket, - ETag: &etag, - Key: &object, - }, "", nil + etagBytes, err := p.meta.RetrieveAttribute(nil, bucket, object, etagkey) + if err != nil { + return res, "", fmt.Errorf("get object etag: %w", err) + } + etag := string(etagBytes) + // A retry has to list the parts the upload was completed with. The + // data integrity ETag is the object checksum rather than the ETag of + // the parts, so only the upload ID is checked in that mode. + if !p.dataIntegrityEtag && etag != multipartClaimToken { + return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) } - return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload) + versionID, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey) + if err != nil && !errors.Is(err, meta.ErrNoSuchKey) { + return res, "", fmt.Errorf("get object version id: %w", err) + } + + return s3response.CompleteMultipartUploadResult{ + Bucket: &bucket, + ETag: &etag, + Key: &object, + }, string(versionID), nil } if err != nil { return res, "", fmt.Errorf("rename upload to etag dir: %w", err) diff --git a/tests/integration/CompleteMultipartUpload.go b/tests/integration/CompleteMultipartUpload.go index caabbff7..83052c24 100644 --- a/tests/integration/CompleteMultipartUpload.go +++ b/tests/integration/CompleteMultipartUpload.go @@ -2279,6 +2279,289 @@ func CompleteMultipartUpload_already_completed(s *S3Conf) error { }) } +func CompleteMultipartUpload_upload_of_another_bucket(s *S3Conf) error { + testName := "CompleteMultipartUpload_upload_of_another_bucket" + return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error { + obj := "my-obj" + dstBucket := getBucketName() + err := setup(s, dstBucket) + if err != nil { + return err + } + + // the target key exists in the other bucket + dstObjs, err := putObjects(s3client, []string{obj}, dstBucket) + if err != nil { + return err + } + + out, err := createMp(s3client, bucket, obj) + if err != nil { + return err + } + parts, _, err := uploadParts(s3client, 5*1024*1024, 1, bucket, obj, *out.UploadId) + if err != nil { + return err + } + compParts := []types.CompletedPart{} + for _, el := range parts { + compParts = append(compParts, types.CompletedPart{ + ETag: el.ETag, + PartNumber: el.PartNumber, + }) + } + + ctx, cancel := context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: &dstBucket, + Key: &obj, + UploadId: out.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{ + Parts: compParts, + }, + }) + cancel() + if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)); err != nil { + return err + } + + ctx, cancel = context.WithTimeout(context.Background(), shortTimeout) + head, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{ + Bucket: &dstBucket, + Key: &obj, + }) + cancel() + if err != nil { + return err + } + if getString(head.ETag) != getString(dstObjs[0].ETag) { + return fmt.Errorf("expected the target object ETag to stay %s, instead got %s", + getString(dstObjs[0].ETag), getString(head.ETag)) + } + + // the upload can still be completed on its own bucket + ctx, cancel = context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: &bucket, + Key: &obj, + UploadId: out.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{ + Parts: compParts, + }, + }) + cancel() + if err != nil { + return err + } + + return teardown(s, dstBucket) + }) +} + +func CompleteMultipartUpload_upload_of_another_key(s *S3Conf) error { + testName := "CompleteMultipartUpload_upload_of_another_key" + return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error { + obj, target := "my-obj", "other-obj" + _, err := putObjects(s3client, []string{target}, bucket) + if err != nil { + return err + } + + out, err := createMp(s3client, bucket, obj) + if err != nil { + return err + } + parts, _, err := uploadParts(s3client, 5*1024*1024, 1, bucket, obj, *out.UploadId) + if err != nil { + return err + } + compParts := []types.CompletedPart{} + for _, el := range parts { + compParts = append(compParts, types.CompletedPart{ + ETag: el.ETag, + PartNumber: el.PartNumber, + }) + } + + // the target key exists, doesn't exist and is nested under an object + for _, key := range []string{target, "absent-obj", target + "/nested"} { + ctx, cancel := context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: &bucket, + Key: &key, + UploadId: out.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{ + Parts: compParts, + }, + }) + cancel() + if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)); err != nil { + return fmt.Errorf("key %s: %w", key, err) + } + } + + return nil + }) +} + +func CompleteMultipartUpload_non_existing_upload_existing_key(s *S3Conf) error { + testName := "CompleteMultipartUpload_non_existing_upload_existing_key" + return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error { + obj := "my-obj" + objs, err := putObjects(s3client, []string{obj}, bucket) + if err != nil { + return err + } + + ctx, cancel := context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: &bucket, + Key: &obj, + UploadId: getPtr("3d5d2ea6-3cbb-4a4b-9a43-6b7d0e5f2a11"), + MultipartUpload: &types.CompletedMultipartUpload{ + Parts: []types.CompletedPart{ + {ETag: objs[0].ETag, PartNumber: getPtr(int32(1))}, + }, + }, + }) + cancel() + return checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)) + }) +} + +func CompleteMultipartUpload_aborted_upload_existing_key(s *S3Conf) error { + testName := "CompleteMultipartUpload_aborted_upload_existing_key" + return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error { + obj := "my-obj" + out, err := createMp(s3client, bucket, obj) + if err != nil { + return err + } + parts, _, err := uploadParts(s3client, 5*1024*1024, 1, bucket, obj, *out.UploadId) + if err != nil { + return err + } + + _, err = putObjects(s3client, []string{obj}, bucket) + if err != nil { + return err + } + + ctx, cancel := context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{ + Bucket: &bucket, + Key: &obj, + UploadId: out.UploadId, + }) + cancel() + if err != nil { + return err + } + + ctx, cancel = context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: &bucket, + Key: &obj, + UploadId: out.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{ + Parts: []types.CompletedPart{ + {ETag: parts[0].ETag, PartNumber: parts[0].PartNumber}, + }, + }, + }) + cancel() + return checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)) + }) +} + +func CompleteMultipartUpload_already_completed_different_parts(s *S3Conf) error { + testName := "CompleteMultipartUpload_already_completed_different_parts" + return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error { + obj := "my-obj" + out, err := createMp(s3client, bucket, obj) + if err != nil { + return err + } + parts, _, err := uploadParts(s3client, 5*1024*1024, 1, bucket, obj, *out.UploadId) + if err != nil { + return err + } + + ctx, cancel := context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: &bucket, + Key: &obj, + UploadId: out.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{ + Parts: []types.CompletedPart{ + {ETag: parts[0].ETag, PartNumber: parts[0].PartNumber}, + }, + }, + }) + cancel() + if err != nil { + return err + } + + ctx, cancel = context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{ + Bucket: &bucket, + Key: &obj, + UploadId: out.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{ + Parts: []types.CompletedPart{ + {ETag: getPtr(`"00000000000000000000000000000000"`), PartNumber: parts[0].PartNumber}, + }, + }, + }) + cancel() + return checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)) + }) +} + +func CompleteMultipartUpload_already_completed_overwritten(s *S3Conf) error { + testName := "CompleteMultipartUpload_already_completed_overwritten" + return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error { + obj := "my-obj" + out, err := createMp(s3client, bucket, obj) + if err != nil { + return err + } + parts, _, err := uploadParts(s3client, 5*1024*1024, 1, bucket, obj, *out.UploadId) + if err != nil { + return err + } + + completeInput := &s3.CompleteMultipartUploadInput{ + Bucket: &bucket, + Key: &obj, + UploadId: out.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{ + Parts: []types.CompletedPart{ + {ETag: parts[0].ETag, PartNumber: parts[0].PartNumber}, + }, + }, + } + + ctx, cancel := context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.CompleteMultipartUpload(ctx, completeInput) + cancel() + if err != nil { + return err + } + + _, err = putObjects(s3client, []string{obj}, bucket) + if err != nil { + return err + } + + ctx, cancel = context.WithTimeout(context.Background(), shortTimeout) + _, err = s3client.CompleteMultipartUpload(ctx, completeInput) + cancel() + return checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)) + }) +} + // CompleteMultipartUpload_racey_data_integrity creates a single multipart // upload, uploads its parts, then fires multiple concurrent // CompleteMultipartUpload calls for the exact same upload ID. All calls must diff --git a/tests/integration/group-tests.go b/tests/integration/group-tests.go index 328683b1..4c68fb97 100644 --- a/tests/integration/group-tests.go +++ b/tests/integration/group-tests.go @@ -614,6 +614,15 @@ func TestCompleteMultipartUpload(ts *TestState) { } ts.Run(CompleteMultipartUpload_success) ts.Run(CompleteMultipartUpload_already_completed) + ts.Run(CompleteMultipartUpload_upload_of_another_bucket) + ts.Run(CompleteMultipartUpload_upload_of_another_key) + ts.Run(CompleteMultipartUpload_non_existing_upload_existing_key) + ts.Run(CompleteMultipartUpload_aborted_upload_existing_key) + ts.Run(CompleteMultipartUpload_already_completed_overwritten) + // azure keeps no ETag of the completed parts to compare a retry with + if !ts.conf.azureTests { + ts.Run(CompleteMultipartUpload_already_completed_different_parts) + } if !(ts.conf.azureTests || ts.conf.sidecarTests) { ts.Run(CompleteMultipartUpload_racey_success) ts.Run(CompleteMultipartUpload_racey_data_integrity) @@ -2134,6 +2143,7 @@ func TestVersioning(ts *TestState) { ts.Run(Versioning_Multipart_Upload_suspended_overwrite_versioned_object) ts.Run(Versioning_Multipart_Upload_overwrite_keeps_previous_version_metadata) ts.Run(Versioning_Multipart_Upload_conditional_writes_over_delete_marker) + ts.Run(Versioning_Multipart_Upload_already_completed) ts.Run(Versioning_UploadPartCopy_invalid_versionId) ts.Run(Versioning_UploadPartCopy_encoded_versionid_separator_invalid_versionId) ts.Run(Versioning_UploadPartCopy_non_existing_versionId) @@ -3349,6 +3359,12 @@ func GetIntTests() IntTests { "CompleteMultipartUpload_success": CompleteMultipartUpload_success, "CompleteMultipartUpload_data_integrity_etag": CompleteMultipartUpload_data_integrity_etag, "CompleteMultipartUpload_already_completed": CompleteMultipartUpload_already_completed, + "CompleteMultipartUpload_upload_of_another_bucket": CompleteMultipartUpload_upload_of_another_bucket, + "CompleteMultipartUpload_upload_of_another_key": CompleteMultipartUpload_upload_of_another_key, + "CompleteMultipartUpload_non_existing_upload_existing_key": CompleteMultipartUpload_non_existing_upload_existing_key, + "CompleteMultipartUpload_aborted_upload_existing_key": CompleteMultipartUpload_aborted_upload_existing_key, + "CompleteMultipartUpload_already_completed_different_parts": CompleteMultipartUpload_already_completed_different_parts, + "CompleteMultipartUpload_already_completed_overwritten": CompleteMultipartUpload_already_completed_overwritten, "CompleteMultipartUpload_racey_success": CompleteMultipartUpload_racey_success, "CompleteMultipartUpload_racey_data_integrity": CompleteMultipartUpload_racey_data_integrity, "PutBucketAcl_non_existing_bucket": PutBucketAcl_non_existing_bucket, @@ -3767,6 +3783,7 @@ func GetIntTests() IntTests { "Versioning_Multipart_Upload_suspended_overwrite_versioned_object": Versioning_Multipart_Upload_suspended_overwrite_versioned_object, "Versioning_Multipart_Upload_overwrite_keeps_previous_version_metadata": Versioning_Multipart_Upload_overwrite_keeps_previous_version_metadata, "Versioning_Multipart_Upload_conditional_writes_over_delete_marker": Versioning_Multipart_Upload_conditional_writes_over_delete_marker, + "Versioning_Multipart_Upload_already_completed": Versioning_Multipart_Upload_already_completed, "Versioning_UploadPartCopy_invalid_versionId": Versioning_UploadPartCopy_invalid_versionId, "Versioning_UploadPartCopy_encoded_versionid_separator_invalid_versionId": Versioning_UploadPartCopy_encoded_versionid_separator_invalid_versionId, "Versioning_UploadPartCopy_non_existing_versionId": Versioning_UploadPartCopy_non_existing_versionId, diff --git a/tests/integration/versioning.go b/tests/integration/versioning.go index 12494471..13a59bbc 100644 --- a/tests/integration/versioning.go +++ b/tests/integration/versioning.go @@ -4945,6 +4945,60 @@ func Versioning_Multipart_Upload_conditional_writes_over_delete_marker(s *S3Conf }, withVersioning(types.BucketVersioningStatusEnabled)) } +func Versioning_Multipart_Upload_already_completed(s *S3Conf) error { + testName := "Versioning_Multipart_Upload_already_completed" + return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error { + obj := "my-obj" + out, err := createMp(s3client, bucket, obj) + if err != nil { + return err + } + parts, _, err := uploadParts(s3client, 5*1024*1024, 1, bucket, obj, *out.UploadId) + if err != nil { + return err + } + + completeInput := &s3.CompleteMultipartUploadInput{ + Bucket: &bucket, + Key: &obj, + UploadId: out.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{ + Parts: []types.CompletedPart{ + {ETag: parts[0].ETag, PartNumber: parts[0].PartNumber}, + }, + }, + } + + ctx, cancel := context.WithTimeout(context.Background(), shortTimeout) + first, err := s3client.CompleteMultipartUpload(ctx, completeInput) + cancel() + if err != nil { + return err + } + + ctx, cancel = context.WithTimeout(context.Background(), shortTimeout) + second, err := s3client.CompleteMultipartUpload(ctx, completeInput) + cancel() + if err != nil { + return err + } + + if getString(first.VersionId) == "" { + return fmt.Errorf("expected non-empty versionId") + } + if getString(second.VersionId) != getString(first.VersionId) { + return fmt.Errorf("expected the versionId to be %s, instead got %s", + getString(first.VersionId), getString(second.VersionId)) + } + if getString(second.ETag) != getString(first.ETag) { + return fmt.Errorf("expected the ETag to be %s, instead got %s", + getString(first.ETag), getString(second.ETag)) + } + + return nil + }, withVersioning(types.BucketVersioningStatusEnabled)) +} + func Versioning_UploadPartCopy_invalid_versionId(s *S3Conf) error { testName := "Versioning_UploadPartCopy_invalid_versionId" return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {