Merge pull request #2482 from versity/sis/scope-complete-mpu-retries-to-upload

fix: scope CompleteMultipartUpload retries to the completed upload

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)

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

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,

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 {