fix: serialize concurrent CompleteMultipartUpload calls via rename

When two requests raced to complete the same multipart upload, the first caller to finish would remove the part files and upload directory. The second caller, already past the initial existence check, would then fail mid-flight with confusing errors such as ErrInvalidPart or an I/O error when trying to open a part that no longer exists. Fix this by atomically renaming the upload directory from <uploadID> to <uploadID>.inprogress at the very start of CompleteMultipartUploadWithCopy, before any part data is read. A concurrent caller will now find the original directory absent and receive a clean NoSuchUpload error. A deferred rename restores the original name if the complete does not succeed, allowing the client to retry. ListMultipartUploads is updated to skip any directories whose name ends in .inprogress so in-flight completes do not appear as pending uploads.
2026-09-03 14:46:55 +00:00 · 2026-03-27 16:14:20 -07:00
parent fab9f1f813
commit 13b3dc5267
3 changed files with 144 additions and 6 deletions
@@ -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 {
@@ -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
+	})
+}
@@ -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,