diff --git a/backend/posix/posix.go b/backend/posix/posix.go index c35f8624..c1f9d9d3 100644 --- a/backend/posix/posix.go +++ b/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 .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 { diff --git a/tests/integration/CompleteMultipartUpload.go b/tests/integration/CompleteMultipartUpload.go index 74f85db8..e2483c95 100644 --- a/tests/integration/CompleteMultipartUpload.go +++ b/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 + }) +} diff --git a/tests/integration/group-tests.go b/tests/integration/group-tests.go index c6035624..b161abe6 100644 --- a/tests/integration/group-tests.go +++ b/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,