Merge pull request #1993 from versity/ben/racing-mp-complete

fix: serialize concurrent CompleteMultipartUpload calls via rename

backend/posix/posix.go

@@ -106,6 +106,7 @@ const (
 	MetaTmpDir          = ".sgwtmp"
 	MetaTmpMultipartDir = MetaTmpDir + "/multipart"
 	onameAttr           = "objname"
+	inProgressSuffix    = ".inprogress"
 	tagHdr              = "X-Amz-Tagging"
 	oldMetaHdr          = "X-Amz-Meta"
 	metadataHdr         = "metadata"
@@ -1670,6 +1671,25 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 		return res, "", err
 	}
 
+	// Atomically rename the upload directory to <uploadID>.inprogress so that
+	// concurrent CompleteMultipartUpload calls for the same upload ID see it as
+	// absent and return ErrNoSuchUpload rather than racing through the rest of
+	// the function.  If this call does not succeed we rename it back (see the
+	// deferred cleanup below).
+	objdirFull := filepath.Join(bucket, MetaTmpMultipartDir, fmt.Sprintf("%x", sum))
+	uploadIDDir := filepath.Join(objdirFull, uploadID)
+	activeUploadName := uploadID + inProgressSuffix
+	uploadIDInProgress := filepath.Join(objdirFull, activeUploadName)
+	if err := os.Rename(uploadIDDir, uploadIDInProgress); err != nil {
+		if errors.Is(err, fs.ErrNotExist) {
+			return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload)
+		}
+		return res, "", fmt.Errorf("mark upload in-progress: %w", err)
+	}
+	// Best-effort rename back on failure so a future retry can still complete.
+	// On success, os.RemoveAll below removes uploadIDInProgress so this is a no-op.
+	defer os.Rename(uploadIDInProgress, uploadIDDir)
+
 	b, err := p.meta.RetrieveAttribute(nil, bucket, object, etagkey)
 	if err == nil || errors.Is(err, fs.ErrNotExist) || errors.Is(err, meta.ErrNoSuchKey) {
 		err = backend.EvaluateObjectPutPreconditions(string(b), input.IfMatch, input.IfNoneMatch, err == nil)
@@ -1680,7 +1700,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 
 	objdir := filepath.Join(MetaTmpMultipartDir, fmt.Sprintf("%x", sum))
 
-	checksums, err := p.retrieveChecksums(nil, bucket, filepath.Join(objdir, uploadID))
+	checksums, err := p.retrieveChecksums(nil, bucket, filepath.Join(objdir, activeUploadName))
 	if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
 		return res, "", fmt.Errorf("get mp checksums: %w", err)
 	}
@@ -1723,7 +1743,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 
 		partNumber = *part.PartNumber
 
-		partObjPath := filepath.Join(objdir, uploadID, fmt.Sprintf("%v", *part.PartNumber))
+		partObjPath := filepath.Join(objdir, activeUploadName, fmt.Sprintf("%v", *part.PartNumber))
 		fullPartPath := filepath.Join(bucket, partObjPath)
 		fi, err := os.Lstat(fullPartPath)
 		if err != nil {
@@ -1784,7 +1804,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 	var composableCsum string
 	var abortOnErrSet bool
 	for i, part := range parts {
-		partObjPath := filepath.Join(objdir, uploadID, fmt.Sprintf("%v", *part.PartNumber))
+		partObjPath := filepath.Join(objdir, activeUploadName, fmt.Sprintf("%v", *part.PartNumber))
 		fullPartPath := filepath.Join(bucket, partObjPath)
 		pf, err := os.Open(fullPartPath)
 		if err != nil {
@@ -1853,7 +1873,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 				if !abortOnErrSet {
 					defer func() {
 						// cleanup tmp dirs
-						os.RemoveAll(filepath.Join(bucket, objdir, uploadID))
+						os.RemoveAll(filepath.Join(bucket, objdir, activeUploadName))
 						// use Remove for objdir in case there are still other
 						// uploads for same object name outstanding, this will
 						// fail if there are any
@@ -1878,7 +1898,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 		}
 	}
 
-	upiddir := filepath.Join(objdir, uploadID)
+	upiddir := filepath.Join(objdir, activeUploadName)
 
 	objMeta := p.loadObjectMetaProperties(nil, bucket, upiddir, nil)
 	err = p.storeObjectMetaProperties(f.File(), bucket, object, objMeta)
@@ -2031,7 +2051,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 	}
 
 	// cleanup tmp dirs
-	os.RemoveAll(filepath.Join(bucket, objdir, uploadID))
+	os.RemoveAll(filepath.Join(bucket, objdir, activeUploadName))
 	// use Remove for objdir in case there are still other uploads
 	// for same object name outstanding, this will fail if there are any
 	os.Remove(filepath.Join(bucket, objdir))
@@ -2467,6 +2487,10 @@ func (p *Posix) ListMultipartUploads(ctx context.Context, mpu *s3.ListMultipartU
 			if !upid.IsDir() {
 				continue
 			}
+			// skip directories that are currently being completed
+			if strings.HasSuffix(upid.Name(), inProgressSuffix) {
+				continue
+			}
 
 			fi, err := upid.Info()
 			if err != nil {

tests/integration/CompleteMultipartUpload.go

@@ -1906,3 +1906,114 @@ func CompleteMultipartUpload_racey_success(s *S3Conf) error {
 			sums, csum)
 	})
 }
+
+// CompleteMultipartUpload_racey_data_integrity creates a single multipart
+// upload, uploads its parts, then fires multiple concurrent
+// CompleteMultipartUpload calls for the exact same upload ID.  Exactly one
+// call must succeed; the rest must fail with NoSuchUpload (the upload was
+// already consumed).  The surviving object must contain exactly the data that
+// was uploaded.  The whole sequence is repeated several times so that
+// intermittent scheduling cannot hide a data-corruption bug.
+func CompleteMultipartUpload_racey_data_integrity(s *S3Conf) error {
+	testName := "CompleteMultipartUpload_racey_data_integrity"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		const (
+			concurrency = 5
+			iterations  = 10
+			partCount   = 3
+		)
+		obj := "my-obj"
+		objSize := int64(15 * 1024 * 1024) // 3 × 5 MiB parts
+		noSuchUpload := s3err.GetAPIError(s3err.ErrNoSuchUpload)
+
+		for iter := range iterations {
+			// Phase 1: create one upload and upload its parts.
+			out, err := createMp(s3client, bucket, obj)
+			if err != nil {
+				return fmt.Errorf("iteration %d: create mp: %w", iter, err)
+			}
+
+			parts, expectedCsum, err := uploadParts(s3client, objSize, partCount, bucket, obj, *out.UploadId)
+			if err != nil {
+				return fmt.Errorf("iteration %d: upload parts: %w", iter, err)
+			}
+
+			compParts := make([]types.CompletedPart, 0, len(parts))
+			for _, el := range parts {
+				compParts = append(compParts, types.CompletedPart{
+					ETag:       el.ETag,
+					PartNumber: el.PartNumber,
+				})
+			}
+
+			// Phase 2: race concurrency complete calls for the same upload ID.
+			// Exactly one must succeed; the others must return NoSuchUpload.
+			var (
+				mu         sync.Mutex
+				successCnt int
+			)
+			eg := errgroup.Group{}
+			for range concurrency {
+				eg.Go(func() error {
+					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 {
+						mu.Lock()
+						successCnt++
+						mu.Unlock()
+						return nil
+					}
+					// Every non-zero error must be NoSuchUpload.
+					return checkApiErr(err, noSuchUpload)
+				})
+			}
+			if err := eg.Wait(); err != nil {
+				return fmt.Errorf("iteration %d: complete phase: %w", iter, err)
+			}
+			if successCnt != 1 {
+				return fmt.Errorf("iteration %d: expected exactly 1 successful complete, got %d",
+					iter, successCnt)
+			}
+
+			// Phase 3: download the object and verify it is complete and
+			// uncorrupted — its checksum must match what was uploaded.
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			resp, err := s3client.GetObject(ctx, &s3.GetObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			if err != nil {
+				cancel()
+				return fmt.Errorf("iteration %d: get object: %w", iter, err)
+			}
+			bdy, err := io.ReadAll(resp.Body)
+			resp.Body.Close()
+			cancel()
+			if err != nil {
+				return fmt.Errorf("iteration %d: read body: %w", iter, err)
+			}
+
+			if resp.ContentLength == nil || *resp.ContentLength != objSize {
+				return fmt.Errorf("iteration %d: expected content-length %d, got %v",
+					iter, objSize, resp.ContentLength)
+			}
+
+			got := sha256.Sum256(bdy)
+			gotHex := hex.EncodeToString(got[:])
+			if gotHex != expectedCsum {
+				return fmt.Errorf("iteration %d: object checksum %q does not match expected %q — data may be corrupted",
+					iter, gotHex, expectedCsum)
+			}
+		}
+
+		return nil
+	})
+}

tests/integration/group-tests.go

@@ -529,6 +529,7 @@ func TestCompleteMultipartUpload(ts *TestState) {
 	ts.Run(CompleteMultipartUpload_success)
 	if !ts.conf.azureTests {
 		ts.Run(CompleteMultipartUpload_racey_success)
+		ts.Run(CompleteMultipartUpload_racey_data_integrity)
 	}
 }
 
@@ -952,6 +953,7 @@ func TestScoutfs(ts *TestState) {
 	ts.Run(CompleteMultipartUpload_should_ignore_the_final_checksum)
 	ts.Run(CompleteMultipartUpload_success)
 	ts.Run(CompleteMultipartUpload_racey_success)
+	ts.Run(CompleteMultipartUpload_racey_data_integrity)
 
 	// posix/scoutfs specific tests
 	ts.Run(PutObject_overwrite_dir_obj)
@@ -1563,6 +1565,7 @@ func GetIntTests() IntTests {
 		"CompleteMultipartUpload_should_succeed_without_final_checksum_type":       CompleteMultipartUpload_should_succeed_without_final_checksum_type,
 		"CompleteMultipartUpload_success":                                          CompleteMultipartUpload_success,
 		"CompleteMultipartUpload_racey_success":                                    CompleteMultipartUpload_racey_success,
+		"CompleteMultipartUpload_racey_data_integrity":                             CompleteMultipartUpload_racey_data_integrity,
 		"PutBucketAcl_non_existing_bucket":                                         PutBucketAcl_non_existing_bucket,
 		"PutBucketAcl_disabled":                                                    PutBucketAcl_disabled,
 		"PutBucketAcl_none_of_the_options_specified":                               PutBucketAcl_none_of_the_options_specified,